From 0db75f46982a4ddf0132d27ab24e49431923838f Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Thu, 10 Sep 2026 19:57:03 +0000 Subject: [PATCH 1/5] More agno instrumentation improvements --- .../README.rst | 1 + .../instrumentation/genai/agno/patch.py | 458 ++++++++++++++---- .../instrumentation/genai/agno/stream.py | 66 +++ .../instrumentation/genai/agno/utils.py | 20 +- .../tests/test_agent.py | 134 ++++- .../tests/test_knowledge.py | 221 +++++++++ .../tests/test_tools.py | 226 ++++++++- 7 files changed, 1021 insertions(+), 105 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst index 041869854..ac33143fb 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst @@ -34,6 +34,7 @@ The instrumentation automatically traces: * ``Team.run`` and ``Team.arun`` * ``Workflow.run`` and ``Workflow.arun`` * ``FunctionCall.execute`` and ``FunctionCall.aexecute`` +* ``Knowledge.search`` and ``Knowledge.asearch`` Configuration ------------- diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 79c974826..224c50efb 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -3,40 +3,59 @@ """Patching functions for Agno instrumentation.""" +# pylint: disable=import-outside-toplevel + from __future__ import annotations import functools import logging -from collections.abc import AsyncIterator, Awaitable, Callable, Iterator +import sys +from collections.abc import ( + AsyncIterator, + Awaitable, + Callable, + Iterator, + Sequence, +) from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: from agno.agent import RunOutput + from agno.knowledge.document.base import Document + from agno.knowledge.knowledge import Knowledge from agno.run.workflow import WorkflowRunOutput from agno.team import TeamRunOutput + from agno.tools.function import FunctionCall, FunctionExecutionResult AgnoRunOutput = RunOutput | TeamRunOutput | WorkflowRunOutput -from wrapt import wrap_function_wrapper +from wrapt import register_post_import_hook, wrap_function_wrapper from opentelemetry.instrumentation.genai.agno.stream import ( AgnoAgentStreamWrapper, + AgnoToolStreamWrapper, AgnoWorkflowStreamWrapper, AsyncAgnoAgentStreamWrapper, + AsyncAgnoToolStreamWrapper, AsyncAgnoWorkflowStreamWrapper, ) from opentelemetry.instrumentation.genai.agno.utils import ( _get_property_value, format_content, + format_retrieval_document, prepare_tool_definitions, ) from opentelemetry.instrumentation.utils import unwrap +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAI, +) from opentelemetry.semconv._incubating.attributes.error_attributes import ( ErrorTypeValues, ) from opentelemetry.util.genai.handler import TelemetryHandler from opentelemetry.util.genai.invocation import ( AgentInvocation, + RetrievalInvocation, ToolInvocation, WorkflowInvocation, ) @@ -58,6 +77,34 @@ _FUNCTION_CALL_CLASS = "FunctionCall" _AGNO_WORKFLOW_MODULE = "agno.workflow.workflow" _WORKFLOW_CLASS = "Workflow" +_AGNO_KNOWLEDGE_MODULE = "agno.knowledge.knowledge" +_KNOWLEDGE_CLASS = "Knowledge" + + +def _safe_wrap_function( + target_module: str, + target_name: str, + wrapper: Callable[..., Any], +) -> None: + """Safely wrap a method if it exists, deferring if module is not yet imported.""" + + def _apply(mod: Any) -> None: + try: + parts = target_name.split(".") + curr = mod + for part in parts: + curr = getattr(curr, part) + except AttributeError: + # Target class or method may not exist across all supported Agno versions. + return + wrap_function_wrapper(mod, target_name, wrapper) + + if target_module in sys.modules: + _apply(sys.modules[target_module]) + return + + # Defer wrapping to avoid eagerly importing submodules with heavy or optional dependencies. + register_post_import_hook(_apply, target_module) def patch_agent(handler: TelemetryHandler) -> None: @@ -72,78 +119,92 @@ def patch_agent(handler: TelemetryHandler) -> None: f"{_AGENT_CLASS}.arun", _agent_arun(handler), ) - try: - wrap_function_wrapper( - _AGNO_TEAM_MODULE, - f"{_TEAM_CLASS}.run", - _agent_run(handler), - ) - wrap_function_wrapper( - _AGNO_TEAM_MODULE, - f"{_TEAM_CLASS}.arun", - _agent_arun(handler), - ) - except (ImportError, AttributeError): - pass - try: - wrap_function_wrapper( - _AGNO_TOOLS_MODULE, - f"{_FUNCTION_CALL_CLASS}.execute", - _tool_call_execute(handler), - ) - wrap_function_wrapper( - _AGNO_TOOLS_MODULE, - f"{_FUNCTION_CALL_CLASS}.aexecute", - _tool_call_aexecute(handler), - ) - except (ImportError, AttributeError): - pass - try: - wrap_function_wrapper( - _AGNO_WORKFLOW_MODULE, - f"{_WORKFLOW_CLASS}.run", - _workflow_run(handler), - ) - wrap_function_wrapper( - _AGNO_WORKFLOW_MODULE, - f"{_WORKFLOW_CLASS}.arun", - _workflow_arun(handler), - ) - except (ImportError, AttributeError): - pass + _safe_wrap_function( + _AGNO_TEAM_MODULE, + f"{_TEAM_CLASS}.run", + _agent_run(handler), + ) + _safe_wrap_function( + _AGNO_TEAM_MODULE, + f"{_TEAM_CLASS}.arun", + _agent_arun(handler), + ) + _safe_wrap_function( + _AGNO_TOOLS_MODULE, + f"{_FUNCTION_CALL_CLASS}.execute", + _tool_call_execute(handler), + ) + _safe_wrap_function( + _AGNO_TOOLS_MODULE, + f"{_FUNCTION_CALL_CLASS}.aexecute", + _tool_call_aexecute(handler), + ) + _safe_wrap_function( + _AGNO_WORKFLOW_MODULE, + f"{_WORKFLOW_CLASS}.run", + _workflow_run(handler), + ) + _safe_wrap_function( + _AGNO_WORKFLOW_MODULE, + f"{_WORKFLOW_CLASS}.arun", + _workflow_arun(handler), + ) + # Knowledge.retrieve and aretrieve delegate to search and asearch, so wrapping + # search/asearch avoids duplicate spans. + _safe_wrap_function( + _AGNO_KNOWLEDGE_MODULE, + f"{_KNOWLEDGE_CLASS}.search", + _knowledge_search(handler), + ) + _safe_wrap_function( + _AGNO_KNOWLEDGE_MODULE, + f"{_KNOWLEDGE_CLASS}.asearch", + _knowledge_asearch(handler), + ) def unpatch_agent() -> None: """Remove patches from Agno class methods.""" - try: - import agno.agent # pylint: disable=import-outside-toplevel - - unwrap(agno.agent.Agent, "run") - unwrap(agno.agent.Agent, "arun") - except (ImportError, AttributeError): - pass - try: - import agno.team # pylint: disable=import-outside-toplevel - - unwrap(agno.team.Team, "run") - unwrap(agno.team.Team, "arun") - except (ImportError, AttributeError): - pass - try: - import agno.tools.function # pylint: disable=import-outside-toplevel - - unwrap(agno.tools.function.FunctionCall, "execute") - unwrap(agno.tools.function.FunctionCall, "aexecute") - except (ImportError, AttributeError): - pass - # Workflow depends on optional packages (like fastapi), may fail to import. - try: - import agno.workflow.workflow # pylint: disable=import-outside-toplevel - - unwrap(agno.workflow.workflow.Workflow, "run") - unwrap(agno.workflow.workflow.Workflow, "arun") - except (ImportError, AttributeError): - pass + if _AGNO_MODULE in sys.modules: + try: + import agno.agent + + unwrap(agno.agent.Agent, "run") + unwrap(agno.agent.Agent, "arun") + except (ImportError, AttributeError): + pass + if _AGNO_TEAM_MODULE in sys.modules: + try: + import agno.team + + unwrap(agno.team.Team, "run") + unwrap(agno.team.Team, "arun") + except (ImportError, AttributeError): + pass + if _AGNO_TOOLS_MODULE in sys.modules: + try: + import agno.tools.function + + unwrap(agno.tools.function.FunctionCall, "execute") + unwrap(agno.tools.function.FunctionCall, "aexecute") + except (ImportError, AttributeError): + pass + if _AGNO_WORKFLOW_MODULE in sys.modules: + try: + import agno.workflow.workflow + + unwrap(agno.workflow.workflow.Workflow, "run") + unwrap(agno.workflow.workflow.Workflow, "arun") + except (ImportError, AttributeError): + pass + if _AGNO_KNOWLEDGE_MODULE in sys.modules: + try: + import agno.knowledge.knowledge + + unwrap(agno.knowledge.knowledge.Knowledge, "search") + unwrap(agno.knowledge.knowledge.Knowledge, "asearch") + except (ImportError, AttributeError): + pass def _extract_input_content(input_val: Any) -> str: @@ -173,30 +234,36 @@ def _extract_arguments_str(args_val: Any) -> str: def _set_tool_invocation_input( invocation: ToolInvocation, - instance: Any, + instance: FunctionCall, capture_content: bool, ) -> None: if capture_content: - arguments = getattr(instance, "arguments", None) + arguments = instance.arguments if arguments is not None: invocation.arguments = _extract_arguments_str(arguments) +def _fail_tool_invocation( + invocation: ToolInvocation, + result: FunctionExecutionResult, +) -> None: + error = result.error + invocation.fail( + Error( + type=ErrorTypeValues.OTHER.value, + message=str(error) if error else None, + ) + ) + + def _set_tool_invocation_output( invocation: ToolInvocation, - result: Any, + result: FunctionExecutionResult, capture_content: bool, ) -> None: - if getattr(result, "status", None) == "failure": - error = getattr(result, "error", None) - invocation.fail( - Error( - type=ErrorTypeValues.OTHER.value, - message=str(error) if error else None, - ) - ) + if result.status == "failure": return - if capture_content and result is not None: + if capture_content: invocation.tool_result = _extract_output_content(result) @@ -241,6 +308,12 @@ def _set_invocation_output( session_id = getattr(result, "session_id", None) if session_id: invocation.conversation_id = str(session_id) + if isinstance(invocation, AgentInvocation): + if not invocation._request_model: + model = getattr(result, "model", None) + if model: + invocation._request_model = str(model) + invocation.attributes[GenAI.GEN_AI_REQUEST_MODEL] = str(model) def _start_agent_invocation( @@ -251,7 +324,21 @@ def _start_agent_invocation( capture_content: bool, ) -> AgentInvocation: agent_name = getattr(instance, "name", None) - invocation = handler.invoke_local_agent(agent_name=agent_name) + model_obj = getattr(instance, "model", None) + request_model = None + if model_obj is not None: + request_model = getattr(model_obj, "id", None) or ( + model_obj if isinstance(model_obj, str) else None + ) + + invocation = handler.invoke_local_agent( + agent_name=str(agent_name) if agent_name else None, + request_model=str(request_model) if request_model else None, + ) + description = getattr(instance, "description", None) + if description: + invocation.agent_description = str(description) + _set_invocation_input(invocation, instance, args, kwargs, capture_content) invocation.tool_definitions = prepare_tool_definitions( getattr(instance, "tools", None) @@ -261,13 +348,13 @@ def _start_agent_invocation( def _start_tool_invocation( handler: TelemetryHandler, - instance: Any, + instance: FunctionCall, capture_content: bool, ) -> ToolInvocation: - function_obj = getattr(instance, "function", None) + function_obj = instance.function tool_name = getattr(function_obj, "name", None) or "tool" tool_desc = getattr(function_obj, "description", None) - tool_call_id = getattr(instance, "call_id", None) + tool_call_id = instance.call_id invocation = handler.tool( name=str(tool_name), @@ -374,23 +461,57 @@ async def _await_result() -> object: return traced_method +def _handle_tool_result( + invocation: ToolInvocation, + instance: FunctionCall, + result: FunctionExecutionResult, + capture_content: bool, +) -> FunctionExecutionResult: + if result.status == "failure": + _fail_tool_invocation(invocation, result) + return result + + tool_res = result.result + wrapped_stream: Any = None + if isinstance(tool_res, AsyncIterator): + wrapped_stream = AsyncAgnoToolStreamWrapper( + cast(Any, tool_res), invocation, capture_content + ) + elif isinstance(tool_res, Iterator): + wrapped_stream = AgnoToolStreamWrapper( + cast(Any, tool_res), invocation, capture_content + ) + + if wrapped_stream is not None: + result.result = wrapped_stream + instance.result = wrapped_stream + return result + + _set_tool_invocation_output(invocation, result, capture_content) + invocation.stop() + return result + + def _tool_call_execute( handler: TelemetryHandler, ) -> Callable[..., Any]: capture_content = handler.should_capture_content() def traced_method( - wrapped: Callable[..., Any], - instance: Any, + wrapped: Callable[..., FunctionExecutionResult], + instance: FunctionCall, args: tuple[Any, ...], kwargs: dict[str, Any], - ) -> Any: - with _start_tool_invocation( - handler, instance, capture_content - ) as invocation: + ) -> FunctionExecutionResult: + invocation = _start_tool_invocation(handler, instance, capture_content) + try: result = wrapped(*args, **kwargs) - _set_tool_invocation_output(invocation, result, capture_content) - return result + except Exception as error: + invocation.fail(error) + raise + return _handle_tool_result( + invocation, instance, result, capture_content + ) return traced_method @@ -401,17 +522,20 @@ def _tool_call_aexecute( capture_content = handler.should_capture_content() async def traced_method( - wrapped: Callable[..., Awaitable[Any]], - instance: Any, + wrapped: Callable[..., Awaitable[FunctionExecutionResult]], + instance: FunctionCall, args: tuple[Any, ...], kwargs: dict[str, Any], - ) -> Any: - with _start_tool_invocation( - handler, instance, capture_content - ) as invocation: + ) -> FunctionExecutionResult: + invocation = _start_tool_invocation(handler, instance, capture_content) + try: result = await wrapped(*args, **kwargs) - _set_tool_invocation_output(invocation, result, capture_content) - return result + except Exception as error: + invocation.fail(error) + raise + return _handle_tool_result( + invocation, instance, result, capture_content + ) return cast(Callable[..., Any], traced_method) @@ -522,3 +646,133 @@ async def _await_result() -> object: return result return traced_method + + +def _start_retrieval_invocation( + handler: TelemetryHandler, + instance: Knowledge, + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> RetrievalInvocation: + vector_db = instance.vector_db + data_source_id = ( + instance.name + or ( + getattr(vector_db, "name", None) if vector_db is not None else None + ) + or ( + getattr(vector_db, "collection", None) + if vector_db is not None + else None + ) + or ( + getattr(vector_db, "table_name", None) + if vector_db is not None + else None + ) + ) + provider = None + if vector_db is not None: + provider = ( + getattr(vector_db, "provider", None) + or vector_db.__class__.__name__.lower() + ) + + embedder = ( + getattr(vector_db, "embedder", None) if vector_db is not None else None + ) + request_model = None + if embedder is not None: + request_model = getattr(embedder, "id", None) or getattr( + embedder, "model", None + ) + + invocation = handler.retrieval( + data_source_id=str(data_source_id) + if data_source_id is not None + else None, + provider=str(provider).lower() if provider is not None else None, + request_model=str(request_model) + if request_model is not None + else None, + ) + + query = args[0] if args else kwargs.get("query") + if query is not None: + invocation.query_text = str(query) + + max_results = None + if len(args) > 1 and args[1] is not None: + max_results = args[1] + elif kwargs.get("max_results") is not None: + max_results = kwargs.get("max_results") + else: + max_results = instance.max_results + + if max_results is not None: + try: + invocation.top_k = int(max_results) + except (ValueError, TypeError): + pass + + return invocation + + +def _knowledge_search( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + def traced_method( + wrapped: Callable[..., Sequence[Document] | None], + instance: Knowledge, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + invocation = _start_retrieval_invocation( + handler, instance, args, kwargs + ) + try: + result = wrapped(*args, **kwargs) + except Exception as error: + invocation.fail(error) + raise + + if capture_content and result is not None: + invocation.documents = [ + format_retrieval_document(doc) for doc in result + ] + invocation.stop() + return result + + return traced_method + + +def _knowledge_asearch( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + async def traced_method( + wrapped: Callable[..., Awaitable[Sequence[Document] | None]], + instance: Knowledge, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + invocation = _start_retrieval_invocation( + handler, instance, args, kwargs + ) + try: + result = await wrapped(*args, **kwargs) + except Exception as error: + invocation.fail(error) + raise + + if capture_content and result is not None: + invocation.documents = [ + format_retrieval_document(doc) for doc in result + ] + invocation.stop() + return result + + return cast(Callable[..., Any], traced_method) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py index a94177ee5..610a906d0 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py @@ -11,8 +11,12 @@ _get_property_value, format_content, ) +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAI, +) from opentelemetry.util.genai.invocation import ( AgentInvocation, + ToolInvocation, WorkflowInvocation, ) from opentelemetry.util.genai.stream import ( @@ -48,6 +52,14 @@ def _process_chunk(self, chunk: Any) -> None: if session_id and not self._self_agent_invocation.conversation_id: self._self_agent_invocation.conversation_id = str(session_id) + if not self._self_agent_invocation._request_model: + model = getattr(chunk, "model", None) + if model: + self._self_agent_invocation._request_model = str(model) + self._self_agent_invocation.attributes[ + GenAI.GEN_AI_REQUEST_MODEL + ] = str(model) + metrics = getattr(chunk, "metrics", None) if metrics is not None: if getattr(metrics, "input_tokens", None) is not None: @@ -309,3 +321,57 @@ def __init__( self._self_content_parts = [] self._self_completed_content = None self._self_finish_reason = "stop" + + +class AgnoToolStreamWrapper(SyncStreamWrapper[Any]): + """Stream wrapper for synchronous tool executions that return iterators/generators.""" + + def __init__( + self, + stream: Any, + invocation: ToolInvocation, + capture_content: bool, + ) -> None: + super().__init__(stream) + self._self_tool_invocation = invocation + self._self_capture_content = capture_content + self._self_chunks: list[str] = [] + + def _process_chunk(self, chunk: Any) -> None: + if self._self_capture_content: + self._self_chunks.append(format_content(chunk)) + + def _on_stream_end(self) -> None: + if self._self_capture_content: + self._self_tool_invocation.tool_result = "".join(self._self_chunks) + self._self_tool_invocation.stop() + + def _on_stream_error(self, error: BaseException) -> None: + self._self_tool_invocation.fail(error) + + +class AsyncAgnoToolStreamWrapper(AsyncStreamWrapper[Any]): + """Stream wrapper for asynchronous tool executions that return async iterators/generators.""" + + def __init__( + self, + stream: Any, + invocation: ToolInvocation, + capture_content: bool, + ) -> None: + super().__init__(stream) + self._self_tool_invocation = invocation + self._self_capture_content = capture_content + self._self_chunks: list[str] = [] + + def _process_chunk(self, chunk: Any) -> None: + if self._self_capture_content: + self._self_chunks.append(format_content(chunk)) + + def _on_stream_end(self) -> None: + if self._self_capture_content: + self._self_tool_invocation.tool_result = "".join(self._self_chunks) + self._self_tool_invocation.stop() + + def _on_stream_error(self, error: BaseException) -> None: + self._self_tool_invocation.fail(error) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/utils.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/utils.py index d7dc37686..ab082d774 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/utils.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/utils.py @@ -8,7 +8,10 @@ import dataclasses import json from collections.abc import Iterable -from typing import Any, Protocol, cast, runtime_checkable +from typing import TYPE_CHECKING, Any, Protocol, cast, runtime_checkable + +if TYPE_CHECKING: + from agno.knowledge.document.base import Document from opentelemetry.util.genai.types import ( FunctionToolDefinition, @@ -16,6 +19,21 @@ ) +def format_retrieval_document(doc: Document) -> dict[str, Any]: + """Format an Agno Document into a retrieval document dict.""" + doc_dict: dict[str, Any] = {"content": doc.content} + if doc.id is not None: + doc_dict["id"] = str(doc.id) + if doc.reranking_score is not None: + try: + doc_dict["score"] = float(doc.reranking_score) + except (ValueError, TypeError): + pass + if doc.meta_data: + doc_dict["metadata"] = doc.meta_data + return doc_dict + + @runtime_checkable class _ModelDumpJson(Protocol): def model_dump_json(self) -> str: ... diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index 174d45c49..cd803381b 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -18,6 +18,7 @@ from tests.mock_model import MockModel from opentelemetry.instrumentation.genai.agno.patch import ( + _fail_tool_invocation, _set_tool_invocation_output, ) from opentelemetry.semconv._incubating.attributes import ( @@ -224,9 +225,11 @@ def test_failed_tool_result_is_not_captured() -> None: invocation = MagicMock() invocation.tool_result = None + result = SimpleNamespace(status="failure", error="tool failed") + _fail_tool_invocation(invocation, result) _set_tool_invocation_output( invocation, - SimpleNamespace(status="failure", error="tool failed"), + result, capture_content=True, ) @@ -736,3 +739,132 @@ class ToolOutput(BaseModel): "status": "ok", "code": 200, } + + +def test_agent_run_attributes( + instrument_agno, + span_exporter, +) -> None: + """Test that Agent.run extracts description and model attributes.""" + agent = Agent( + id="agent-custom-id", + name="attribute-agent", + description="Custom agent description", + model=MockModel(id="custom-model", provider="custom_provider"), + ) + + def mock_response(*args: Any, **kwargs: Any) -> ModelResponse: + return ModelResponse(content="Response") + + with ( + patch("agno.models.base.Model.response", side_effect=mock_response), + ): + res = agent.run("hello") + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_agent" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_NAME) + == "attribute-agent" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_DESCRIPTION) + == "Custom agent description" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "custom-model" + ) + assert GenAIAttributes.GEN_AI_AGENT_ID not in span.attributes + assert GenAIAttributes.GEN_AI_PROVIDER_NAME not in span.attributes + assert GenAIAttributes.GEN_AI_USAGE_INPUT_TOKENS not in span.attributes + assert GenAIAttributes.GEN_AI_USAGE_OUTPUT_TOKENS not in span.attributes + + +def test_agent_arun_attributes( + instrument_agno, + span_exporter, +) -> None: + """Test that Agent.arun extracts description and model attributes.""" + agent = Agent( + id="async-agent-id", + name="async-attribute-agent", + description="Async description", + model=MockModel(id="custom-async-model", provider="custom_provider"), + ) + + async def mock_aresponse(*args: Any, **kwargs: Any) -> ModelResponse: + return ModelResponse(content="Async response") + + async def _run_async() -> None: + with patch( + "agno.models.base.Model.aresponse", side_effect=mock_aresponse + ): + res = await agent.arun("async hello") + assert res is not None + + asyncio.run(_run_async()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_NAME) + == "async-attribute-agent" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_DESCRIPTION) + == "Async description" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "custom-async-model" + ) + assert GenAIAttributes.GEN_AI_AGENT_ID not in span.attributes + assert GenAIAttributes.GEN_AI_PROVIDER_NAME not in span.attributes + + +def test_agent_run_attributes_from_run_output( + instrument_agno, + span_exporter, +) -> None: + """Test extracting model from RunOutput when not present on Agent.""" + import agno.agent + from agno.agent import RunOutput + + agent = Agent(name="unconfigured-agent", model=MockModel(id=None)) + + mock_run_output = RunOutput( + agent_id="extracted-agent-id", + model="extracted-model", + model_provider="ExtractedProvider", + content="Hello output", + ) + + dispatch_target = ( + "agno.agent._run.run_dispatch" + if hasattr(agno.agent, "_run") + else "agno.agent.agent.Agent._run" + ) + + with ( + patch(dispatch_target, return_value=mock_run_output), + ): + res = agent.run("test") + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "extracted-model" + ) + assert GenAIAttributes.GEN_AI_AGENT_ID not in span.attributes + assert GenAIAttributes.GEN_AI_PROVIDER_NAME not in span.attributes diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py new file mode 100644 index 000000000..20be62650 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py @@ -0,0 +1,221 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for Agno Knowledge instrumentation.""" + +from __future__ import annotations + +import asyncio +import json +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from agno.knowledge.knowledge import Document, Knowledge + +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAIAttributes, +) +from opentelemetry.trace import SpanKind + + +def test_knowledge_search_content_capture( + instrument_agno_content_capture, + span_exporter, +) -> None: + """Test Knowledge.search with content capture enabled.""" + kb = Knowledge(name="test_kb", max_results=5) + kb.vector_db = MagicMock() + kb.vector_db.name = "pgvector_instance" + kb.vector_db.provider = "pgvector" + kb.vector_db.search.return_value = [ + Document( + content="OpenTelemetry is an observability framework.", + id="doc_1", + meta_data={"source": "docs"}, + reranking_score=0.95, + ), + Document( + content="Agno is an agent framework.", + id="doc_2", + reranking_score=0.88, + ), + ] + + docs = kb.search(query="what is OpenTelemetry?", max_results=5) + assert len(docs) == 2 + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "retrieval test_kb" + assert span.kind == SpanKind.CLIENT + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "retrieval" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_DATA_SOURCE_ID) + == "test_kb" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_PROVIDER_NAME) + == "pgvector" + ) + assert span.attributes.get("gen_ai.retrieval.top_k") == 5 + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_RETRIEVAL_QUERY_TEXT) + == "what is OpenTelemetry?" + ) + + raw_docs = span.attributes.get(GenAIAttributes.GEN_AI_RETRIEVAL_DOCUMENTS) + assert isinstance(raw_docs, str) + parsed_docs = json.loads(raw_docs) + assert len(parsed_docs) == 2 + assert ( + parsed_docs[0]["content"] + == "OpenTelemetry is an observability framework." + ) + assert parsed_docs[0]["id"] == "doc_1" + assert parsed_docs[0]["score"] == 0.95 + assert parsed_docs[0]["metadata"] == {"source": "docs"} + assert parsed_docs[1]["content"] == "Agno is an agent framework." + assert parsed_docs[1]["id"] == "doc_2" + assert parsed_docs[1]["score"] == 0.88 + + +def test_knowledge_search_no_content_capture( + instrument_agno, + span_exporter, +) -> None: + """Test Knowledge.search with content capture disabled.""" + kb = Knowledge(name="private_kb", max_results=3) + kb.vector_db = MagicMock() + kb.vector_db.search.return_value = [ + Document(content="Secret content", id="secret_1") + ] + + docs = kb.search(query="secret query") + assert len(docs) == 1 + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "retrieval private_kb" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "retrieval" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_DATA_SOURCE_ID) + == "private_kb" + ) + assert span.attributes.get("gen_ai.retrieval.top_k") == 3 + assert GenAIAttributes.GEN_AI_RETRIEVAL_QUERY_TEXT not in span.attributes + assert GenAIAttributes.GEN_AI_RETRIEVAL_DOCUMENTS not in span.attributes + + +def test_knowledge_asearch_content_capture( + instrument_agno_content_capture, + span_exporter, +) -> None: + """Test Knowledge.asearch with content capture enabled.""" + if not hasattr(Knowledge, "asearch"): + pytest.skip("Knowledge.asearch is not supported in this version of agno") + + kb = Knowledge(name="async_kb") + kb.vector_db = MagicMock() + kb.vector_db.async_search = AsyncMock( + return_value=[ + Document( + content="Async retrieved doc", + id="adoc_1", + reranking_score=0.9, + ) + ] + ) + + search_fn = kb.asearch + + async def _run() -> list[Document]: + return await search_fn(query="async query", max_results=2) + + docs = asyncio.run(_run()) + assert len(docs) == 1 + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "retrieval async_kb" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_DATA_SOURCE_ID) + == "async_kb" + ) + assert span.attributes.get("gen_ai.retrieval.top_k") == 2 + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_RETRIEVAL_QUERY_TEXT) + == "async query" + ) + raw_docs = span.attributes.get(GenAIAttributes.GEN_AI_RETRIEVAL_DOCUMENTS) + assert isinstance(raw_docs, str) + assert "Async retrieved doc" in raw_docs + + +def test_knowledge_retrieve_delegation( + instrument_agno, + span_exporter, +) -> None: + """Test that Knowledge.retrieve delegates to search and creates a retrieval span.""" + if not hasattr(Knowledge, "retrieve"): + pytest.skip("Knowledge.retrieve is not supported in this version of agno") + + kb = Knowledge(name="retrieve_kb") + kb.vector_db = MagicMock() + kb.vector_db.search.return_value = [Document(content="retrieved doc")] + + docs = kb.retrieve("test query") + assert len(docs) == 1 + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "retrieval retrieve_kb" + + +def test_knowledge_aretrieve_delegation( + instrument_agno, + span_exporter, +) -> None: + """Test that Knowledge.aretrieve delegates to asearch and creates a retrieval span.""" + if not hasattr(Knowledge, "aretrieve"): + pytest.skip("Knowledge.aretrieve is not supported in this version of agno") + + kb = Knowledge(name="aretrieve_kb") + kb.vector_db = MagicMock() + kb.vector_db.async_search = AsyncMock( + return_value=[Document(content="async retrieved doc")] + ) + + async def _run() -> list[Document]: + return await kb.aretrieve("async test query") + + docs = asyncio.run(_run()) + assert len(docs) == 1 + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].name == "retrieval aretrieve_kb" + + +def test_knowledge_search_error( + instrument_agno, + span_exporter, +) -> None: + """Test that errors in search record error.type on the span.""" + kb = Knowledge(name="error_kb") + + with pytest.raises(TypeError): + getattr(kb, "search")() + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes.get("error.type") == "TypeError" diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_tools.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_tools.py index 2bed5a8a6..274c9db33 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_tools.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_tools.py @@ -9,10 +9,11 @@ import json from unittest.mock import patch +import pytest from agno.agent import Agent from agno.models.response import ModelResponse from agno.tools import Toolkit -from agno.tools.function import Function +from agno.tools.function import Function, FunctionCall from tests.mock_model import MockModel from opentelemetry.instrumentation.genai.agno.utils import ( @@ -256,3 +257,226 @@ def test_agent_run_without_tools( span = spans[0] assert span.name == "invoke_agent test-no-tools-agent" assert GenAIAttributes.GEN_AI_TOOL_DEFINITIONS not in span.attributes + + +def test_tool_call_execute_sync( + instrument_agno_content_capture, + span_exporter, +) -> None: + def multiply(a: int, b: int) -> int: + """Multiply two numbers.""" + return a * b + + call = FunctionCall( + function=Function.from_callable(multiply), + arguments={"a": 3, "b": 4}, + call_id="call_sync_1", + ) + result = call.execute() + assert result.status == "success" + assert result.result == 12 + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "execute_tool multiply" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "execute_tool" + ) + assert span.attributes.get(GenAIAttributes.GEN_AI_TOOL_NAME) == "multiply" + assert span.attributes.get(GenAIAttributes.GEN_AI_TOOL_TYPE) == "function" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_ID) + == "call_sync_1" + ) + assert span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_RESULT) == "12" + + +def test_tool_call_execute_streaming_success( + instrument_agno_content_capture, + span_exporter, +) -> None: + def stream_gen(prefix: str): + """Yield chunks.""" + yield f"{prefix}_1" + yield f"{prefix}_2" + + call = FunctionCall( + function=Function.from_callable(stream_gen), + arguments={"prefix": "part"}, + call_id="call_stream_1", + ) + result = call.execute() + assert result.status == "success" + + # Span must not be closed yet before draining + assert len(span_exporter.get_finished_spans()) == 0 + + items = list(result.result) + assert items == ["part_1", "part_2"] + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "execute_tool stream_gen" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_ID) + == "call_stream_1" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_RESULT) + == "part_1part_2" + ) + + +def test_tool_call_execute_streaming_error( + instrument_agno_content_capture, + span_exporter, +) -> None: + def failing_stream(): + """Yield then raise.""" + yield "first" + raise ValueError("stream-side error") + + call = FunctionCall( + function=Function.from_callable(failing_stream), + arguments={}, + call_id="call_err_1", + ) + result = call.execute() + assert result.status == "success" + + with pytest.raises(ValueError, match="stream-side error"): + list(result.result) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes.get("error.type") == "ValueError" + + +def test_tool_call_execute_streaming_caller_error( + instrument_agno_content_capture, + span_exporter, +) -> None: + def good_stream(): + """Yield chunks.""" + yield "chunk_a" + yield "chunk_b" + + call = FunctionCall( + function=Function.from_callable(good_stream), + arguments={}, + call_id="call_caller_err", + ) + result = call.execute() + assert result.status == "success" + + with pytest.raises(RuntimeError, match="caller-side failure"): + with result.result as stream: + for item in stream: + if item == "chunk_a": + raise RuntimeError("caller-side failure") + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes.get("error.type") == "RuntimeError" + + +def test_tool_call_aexecute_streaming_success( + instrument_agno_content_capture, + span_exporter, +) -> None: + async def async_stream_gen(prefix: str): + """Yield async chunks.""" + yield f"{prefix}_async_1" + yield f"{prefix}_async_2" + + call = FunctionCall( + function=Function.from_callable(async_stream_gen), + arguments={"prefix": "async_part"}, + call_id="call_astream_1", + ) + + async def _test() -> None: + result = await call.aexecute() + assert result.status == "success" + assert len(span_exporter.get_finished_spans()) == 0 + + chunks = [chunk async for chunk in result.result] + assert chunks == ["async_part_async_1", "async_part_async_2"] + + asyncio.run(_test()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "execute_tool async_stream_gen" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_ID) + == "call_astream_1" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_RESULT) + == "async_part_async_1async_part_async_2" + ) + + +def test_tool_call_aexecute_streaming_error( + instrument_agno_content_capture, + span_exporter, +) -> None: + async def failing_async_stream(): + """Yield then raise.""" + yield "first_async" + raise ValueError("async stream failure") + + call = FunctionCall( + function=Function.from_callable(failing_async_stream), + arguments={}, + call_id="call_aerr_1", + ) + + async def _test() -> None: + result = await call.aexecute() + with pytest.raises(ValueError, match="async stream failure"): + _ = [c async for c in result.result] + + asyncio.run(_test()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes.get("error.type") == "ValueError" + + +def test_tool_call_aexecute_streaming_caller_error( + instrument_agno_content_capture, + span_exporter, +) -> None: + async def good_async_stream(): + yield "a" + yield "b" + + call = FunctionCall( + function=Function.from_callable(good_async_stream), + arguments={}, + call_id="call_acaller_err", + ) + + async def _test() -> None: + result = await call.aexecute() + with pytest.raises(RuntimeError, match="caller async error"): + async with result.result as stream: + async for chunk in stream: + if chunk == "a": + raise RuntimeError("caller async error") + + asyncio.run(_test()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.attributes.get("error.type") == "RuntimeError" From bcbd4ee5f898b1c32196f21156758fd8aad65a2c Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Thu, 10 Sep 2026 20:02:33 +0000 Subject: [PATCH 2/5] Add conformance test --- .../tests/conformance/retrieval.py | 67 +++++++++++++++++++ .../tests/test_conformance.py | 2 + .../tests/test_knowledge.py | 20 +++--- 3 files changed, 81 insertions(+), 8 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/retrieval.py diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/retrieval.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/retrieval.py new file mode 100644 index 000000000..422c7b989 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/retrieval.py @@ -0,0 +1,67 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Conformance scenario: retrieval via Knowledge.search for Agno.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from agno.knowledge.knowledge import Document, Knowledge + +from opentelemetry.instrumentation.genai.agno import AgnoInstrumentor +from opentelemetry.sdk._logs import LoggerProvider +from opentelemetry.sdk.metrics import MeterProvider +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.test_util_genai.conformance import ( + ExpectedViolation, + Scenario, +) +from opentelemetry.test_util_genai.instrumentor import instrument + + +class RetrievalScenario(Scenario): + expected_spans = {"retrieval": 1} + expected_metrics = ("gen_ai.client.operation.duration",) + expected_violations = ( + ExpectedViolation( + advice_id="genai_expected_attribute_missing", + message_substring="server.address", + ), + ) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + with instrument( + AgnoInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + content_capture="SPAN_ONLY", + ): + kb = Knowledge(name="conformance-kb", max_results=2) + kb.vector_db = MagicMock() + kb.vector_db.name = "conformance_vector_db" + kb.vector_db.provider = "pgvector" + kb.vector_db.search.return_value = [ + Document( + content="OpenTelemetry provides observability standards.", + id="doc-1", + meta_data={"source": "docs"}, + reranking_score=0.95, + ), + Document( + content="Agno provides multi-agent framework.", + id="doc-2", + meta_data={"source": "docs"}, + reranking_score=0.88, + ), + ] + kb.search(query="what is OpenTelemetry?", max_results=2) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py index 360cf7179..529586492 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py @@ -21,6 +21,7 @@ ) from .conformance.agent import AgentScenario +from .conformance.retrieval import RetrievalScenario from .conformance.workflow import WorkflowScenario from .conformance.workflow_streaming import WorkflowStreamingScenario @@ -29,6 +30,7 @@ "scenario", [ pytest.param(AgentScenario()), + pytest.param(RetrievalScenario()), pytest.param(WorkflowScenario()), pytest.param(WorkflowStreamingScenario()), ], diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py index 20be62650..ed1eddd5a 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_knowledge.py @@ -7,7 +7,7 @@ import asyncio import json -from unittest.mock import AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock import pytest from agno.knowledge.knowledge import Document, Knowledge @@ -54,12 +54,10 @@ def test_knowledge_search_content_capture( == "retrieval" ) assert ( - span.attributes.get(GenAIAttributes.GEN_AI_DATA_SOURCE_ID) - == "test_kb" + span.attributes.get(GenAIAttributes.GEN_AI_DATA_SOURCE_ID) == "test_kb" ) assert ( - span.attributes.get(GenAIAttributes.GEN_AI_PROVIDER_NAME) - == "pgvector" + span.attributes.get(GenAIAttributes.GEN_AI_PROVIDER_NAME) == "pgvector" ) assert span.attributes.get("gen_ai.retrieval.top_k") == 5 assert ( @@ -120,7 +118,9 @@ def test_knowledge_asearch_content_capture( ) -> None: """Test Knowledge.asearch with content capture enabled.""" if not hasattr(Knowledge, "asearch"): - pytest.skip("Knowledge.asearch is not supported in this version of agno") + pytest.skip( + "Knowledge.asearch is not supported in this version of agno" + ) kb = Knowledge(name="async_kb") kb.vector_db = MagicMock() @@ -166,7 +166,9 @@ def test_knowledge_retrieve_delegation( ) -> None: """Test that Knowledge.retrieve delegates to search and creates a retrieval span.""" if not hasattr(Knowledge, "retrieve"): - pytest.skip("Knowledge.retrieve is not supported in this version of agno") + pytest.skip( + "Knowledge.retrieve is not supported in this version of agno" + ) kb = Knowledge(name="retrieve_kb") kb.vector_db = MagicMock() @@ -186,7 +188,9 @@ def test_knowledge_aretrieve_delegation( ) -> None: """Test that Knowledge.aretrieve delegates to asearch and creates a retrieval span.""" if not hasattr(Knowledge, "aretrieve"): - pytest.skip("Knowledge.aretrieve is not supported in this version of agno") + pytest.skip( + "Knowledge.aretrieve is not supported in this version of agno" + ) kb = Knowledge(name="aretrieve_kb") kb.vector_db = MagicMock() From 7d5d26afff7bac9f615be9470c3f08e07a276d71 Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Thu, 10 Sep 2026 20:06:08 +0000 Subject: [PATCH 3/5] Add changelog entry --- .../.changelog/673.added | 1 + 1 file changed, 1 insertion(+) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/673.added diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/673.added b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/673.added new file mode 100644 index 000000000..1921c8082 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/673.added @@ -0,0 +1 @@ +Add retrieval instrumentation for Knowledge search and streaming support for tool executions. From 4520fcfa6789286f46a4634c3e8601b1791e857d Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Thu, 10 Sep 2026 20:34:04 +0000 Subject: [PATCH 4/5] fix(agno): address review feedback on deferred wrapping and private attributes - Guard deferred post-import hooks with generation tracking and instrumentation state - Prevent stacking wrappers in _safe_wrap_function - Avoid mutating private AgentInvocation._request_model by using public attributes - Add unit tests for deferred uninstrument/re-instrument cycles and chunk model extraction Assisted-by: Antigravity --- .../instrumentation/genai/agno/patch.py | 43 +++++- .../instrumentation/genai/agno/stream.py | 12 +- .../tests/test_instrumentor.py | 124 ++++++++++++++++++ .../tests/test_stream.py | 27 ++++ 4 files changed, 192 insertions(+), 14 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 224c50efb..5515d1c36 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -81,14 +81,21 @@ _KNOWLEDGE_CLASS = "Knowledge" +_instrumentation_generation: int = 0 +_is_instrumented: bool = False + + def _safe_wrap_function( target_module: str, target_name: str, wrapper: Callable[..., Any], + generation: int, ) -> None: """Safely wrap a method if it exists, deferring if module is not yet imported.""" def _apply(mod: Any) -> None: + if not _is_instrumented or _instrumentation_generation != generation: + return try: parts = target_name.split(".") curr = mod @@ -97,6 +104,8 @@ def _apply(mod: Any) -> None: except AttributeError: # Target class or method may not exist across all supported Agno versions. return + if hasattr(curr, "__wrapped__"): + return wrap_function_wrapper(mod, target_name, wrapper) if target_module in sys.modules: @@ -109,45 +118,58 @@ def _apply(mod: Any) -> None: def patch_agent(handler: TelemetryHandler) -> None: """Apply patches to Agno class methods.""" - wrap_function_wrapper( + global _instrumentation_generation, _is_instrumented + _instrumentation_generation += 1 + _is_instrumented = True + current_generation = _instrumentation_generation + + _safe_wrap_function( _AGNO_MODULE, f"{_AGENT_CLASS}.run", _agent_run(handler), + current_generation, ) - wrap_function_wrapper( + _safe_wrap_function( _AGNO_MODULE, f"{_AGENT_CLASS}.arun", _agent_arun(handler), + current_generation, ) _safe_wrap_function( _AGNO_TEAM_MODULE, f"{_TEAM_CLASS}.run", _agent_run(handler), + current_generation, ) _safe_wrap_function( _AGNO_TEAM_MODULE, f"{_TEAM_CLASS}.arun", _agent_arun(handler), + current_generation, ) _safe_wrap_function( _AGNO_TOOLS_MODULE, f"{_FUNCTION_CALL_CLASS}.execute", _tool_call_execute(handler), + current_generation, ) _safe_wrap_function( _AGNO_TOOLS_MODULE, f"{_FUNCTION_CALL_CLASS}.aexecute", _tool_call_aexecute(handler), + current_generation, ) _safe_wrap_function( _AGNO_WORKFLOW_MODULE, f"{_WORKFLOW_CLASS}.run", _workflow_run(handler), + current_generation, ) _safe_wrap_function( _AGNO_WORKFLOW_MODULE, f"{_WORKFLOW_CLASS}.arun", _workflow_arun(handler), + current_generation, ) # Knowledge.retrieve and aretrieve delegate to search and asearch, so wrapping # search/asearch avoids duplicate spans. @@ -155,16 +177,21 @@ def patch_agent(handler: TelemetryHandler) -> None: _AGNO_KNOWLEDGE_MODULE, f"{_KNOWLEDGE_CLASS}.search", _knowledge_search(handler), + current_generation, ) _safe_wrap_function( _AGNO_KNOWLEDGE_MODULE, f"{_KNOWLEDGE_CLASS}.asearch", _knowledge_asearch(handler), + current_generation, ) def unpatch_agent() -> None: """Remove patches from Agno class methods.""" + global _instrumentation_generation, _is_instrumented + _instrumentation_generation += 1 + _is_instrumented = False if _AGNO_MODULE in sys.modules: try: import agno.agent @@ -309,11 +336,11 @@ def _set_invocation_output( if session_id: invocation.conversation_id = str(session_id) if isinstance(invocation, AgentInvocation): - if not invocation._request_model: - model = getattr(result, "model", None) - if model: - invocation._request_model = str(model) - invocation.attributes[GenAI.GEN_AI_REQUEST_MODEL] = str(model) + model = getattr(result, "model", None) + if model: + invocation.attributes.setdefault( + GenAI.GEN_AI_REQUEST_MODEL, str(model) + ) def _start_agent_invocation( @@ -335,6 +362,8 @@ def _start_agent_invocation( agent_name=str(agent_name) if agent_name else None, request_model=str(request_model) if request_model else None, ) + if request_model: + invocation.attributes[GenAI.GEN_AI_REQUEST_MODEL] = str(request_model) description = getattr(instance, "description", None) if description: invocation.agent_description = str(description) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py index 610a906d0..da814d501 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/stream.py @@ -52,13 +52,11 @@ def _process_chunk(self, chunk: Any) -> None: if session_id and not self._self_agent_invocation.conversation_id: self._self_agent_invocation.conversation_id = str(session_id) - if not self._self_agent_invocation._request_model: - model = getattr(chunk, "model", None) - if model: - self._self_agent_invocation._request_model = str(model) - self._self_agent_invocation.attributes[ - GenAI.GEN_AI_REQUEST_MODEL - ] = str(model) + model = getattr(chunk, "model", None) + if model: + self._self_agent_invocation.attributes.setdefault( + GenAI.GEN_AI_REQUEST_MODEL, str(model) + ) metrics = getattr(chunk, "metrics", None) if metrics is not None: diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_instrumentor.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_instrumentor.py index 6c5187820..60324795a 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_instrumentor.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_instrumentor.py @@ -107,3 +107,127 @@ def test_instrumentor_has_required_attributes() -> None: assert callable(instrumentor.instrument) assert callable(instrumentor.uninstrument) assert callable(instrumentor.instrumentation_dependencies) + + +def test_deferred_wrapping_disabled_after_uninstrument( + tracer_provider, logger_provider, meter_provider +) -> None: + """Test that deferred post-import hooks do not patch modules imported after uninstrument.""" + import sys + import types + + from wrapt import notify_module_loaded + + from opentelemetry.instrumentation.genai.agno import patch as patch_module + + fake_mod_name = "agno.test_deferred_uninstrument_module" + if fake_mod_name in sys.modules: + del sys.modules[fake_mod_name] + + instrumentor = AgnoInstrumentor() + instrumentor.instrument( + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + ) + + called: list[bool] = [] + + def dummy_wrapper(wrapped, instance, args, kwargs): + called.append(True) + return wrapped(*args, **kwargs) + + patch_module._safe_wrap_function( + fake_mod_name, + "TargetClass.action", + dummy_wrapper, + patch_module._instrumentation_generation, + ) + + # Uninstrument before module is loaded + instrumentor.uninstrument() + + # Now module is imported + fake_mod = types.ModuleType(fake_mod_name) + + class TargetClass: + def action(self) -> str: + return "original" + + fake_mod.TargetClass = TargetClass + sys.modules[fake_mod_name] = fake_mod + + notify_module_loaded(fake_mod) + + try: + assert not hasattr(TargetClass.action, "__wrapped__") + assert TargetClass().action() == "original" + assert len(called) == 0 + finally: + del sys.modules[fake_mod_name] + + +def test_deferred_wrapping_re_instrument( + tracer_provider, logger_provider, meter_provider +) -> None: + """Test that re-instrumenting patches correctly without stacking wrappers.""" + import sys + import types + + from wrapt import notify_module_loaded + + from opentelemetry.instrumentation.genai.agno import patch as patch_module + + fake_mod_name = "agno.test_deferred_reinstrument_module" + if fake_mod_name in sys.modules: + del sys.modules[fake_mod_name] + + instrumentor = AgnoInstrumentor() + # Cycle 1: instrument then uninstrument + instrumentor.instrument( + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + ) + instrumentor.uninstrument() + + # Cycle 2: re-instrument + instrumentor.instrument( + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + ) + + call_count = 0 + + def dummy_wrapper(wrapped, instance, args, kwargs): + nonlocal call_count + call_count += 1 + return wrapped(*args, **kwargs) + + patch_module._safe_wrap_function( + fake_mod_name, + "TargetClass.action", + dummy_wrapper, + patch_module._instrumentation_generation, + ) + + fake_mod = types.ModuleType(fake_mod_name) + + class TargetClass: + def action(self) -> str: + return "original" + + fake_mod.TargetClass = TargetClass + sys.modules[fake_mod_name] = fake_mod + + notify_module_loaded(fake_mod) + + try: + assert hasattr(TargetClass.action, "__wrapped__") + assert TargetClass().action() == "original" + assert call_count == 1 + finally: + instrumentor.uninstrument() + if fake_mod_name in sys.modules: + del sys.modules[fake_mod_name] diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_stream.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_stream.py index ec6e9db00..a461323d3 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_stream.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_stream.py @@ -79,6 +79,33 @@ def fake_stream(*args, **kwargs): assert "chunk 1 chunk 2" in output_messages +def test_agent_run_stream_model_from_chunk( + instrument_agno, + span_exporter, +) -> None: + """Test extracting model from stream chunks when not present on Agent upfront.""" + from agno.agent import RunOutput + + agent = Agent(name="test-stream-model-agent", model=MockModel(id=None)) + + def fake_stream(*args, **kwargs): + yield RunOutput(content="chunk 1", model="streamed-chunk-model") + yield RunOutput(content="chunk 2", model="streamed-chunk-model") + + with _patch_agent_stream(fake_stream): + stream = agent.run("hello stream", stream=True) + chunks = list(stream) + assert len(chunks) == 2 + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "streamed-chunk-model" + ) + + def test_agent_run_stream_content_capture_disabled( instrument_agno, span_exporter, From 171e8ff769838ceaf4de4aad09b8fe26163163ec Mon Sep 17 00:00:00 2001 From: Dylan Russell Date: Thu, 10 Sep 2026 20:43:07 +0000 Subject: [PATCH 5/5] docs(agno): comment _instrumentation_generation rationale Assisted-by: Antigravity --- .../src/opentelemetry/instrumentation/genai/agno/patch.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py index 5515d1c36..115e2e46f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/src/opentelemetry/instrumentation/genai/agno/patch.py @@ -81,6 +81,8 @@ _KNOWLEDGE_CLASS = "Knowledge" +# wrapt has no unregister API for post-import hooks; monotonic generations +# invalidate deferred hooks registered during prior instrumentation cycles. _instrumentation_generation: int = 0 _is_instrumented: bool = False