diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added new file mode 100644 index 000000000..8ac19baa0 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/.changelog/350.added @@ -0,0 +1 @@ +Add instrumentation for Agno Team, Workflow, and Model response methods. diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst index 13c67100b..0201df280 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/README.rst @@ -25,6 +25,17 @@ Usage # Instrument Agno AgnoInstrumentor().instrument() +Supported Operations +-------------------- + +The instrumentation automatically traces: + +* ``Agent.run`` and ``Agent.arun`` +* ``Team.run`` and ``Team.arun`` +* ``Workflow.run`` and ``Workflow.arun`` +* ``Model.response`` and ``Model.aresponse`` +* ``FunctionCall.execute`` and ``FunctionCall.aexecute`` + Configuration ------------- diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml b/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml index a4b53d926..9fed56251 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/pyproject.toml @@ -28,7 +28,7 @@ dependencies = [ "opentelemetry-api ~= 1.43", "opentelemetry-instrumentation >= 0.64b0, <1", "opentelemetry-semantic-conventions >= 0.64b0, <1", - "opentelemetry-util-genai >= 1.0b0, <2", + "opentelemetry-util-genai >= 1.1b0.dev, <2", ] [project.optional-dependencies] 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 1c89fa155..7d53d610e 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 @@ -19,7 +19,9 @@ from opentelemetry.util.genai.handler import TelemetryHandler from opentelemetry.util.genai.invocation import ( AgentInvocation, + InferenceInvocation, ToolInvocation, + WorkflowInvocation, ) from opentelemetry.util.genai.types import ( InputMessage, @@ -31,12 +33,18 @@ _AGNO_MODULE = "agno.agent" _AGENT_CLASS = "Agent" +_AGNO_TEAM_MODULE = "agno.team" +_TEAM_CLASS = "Team" _AGNO_TOOLS_MODULE = "agno.tools.function" _FUNCTION_CALL_CLASS = "FunctionCall" +_AGNO_WORKFLOW_MODULE = "agno.workflow.workflow" +_WORKFLOW_CLASS = "Workflow" +_AGNO_MODELS_MODULE = "agno.models.base" +_MODEL_CLASS = "Model" def patch_agent(handler: TelemetryHandler) -> None: - """Apply patches to agno.agent.Agent and agno.tools.function.FunctionCall class methods.""" + """Apply patches to Agno class methods.""" wrap_function_wrapper( _AGNO_MODULE, f"{_AGENT_CLASS}.run", @@ -47,6 +55,19 @@ 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, @@ -60,10 +81,36 @@ def patch_agent(handler: TelemetryHandler) -> None: ) 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 + try: + wrap_function_wrapper( + _AGNO_MODELS_MODULE, + f"{_MODEL_CLASS}.response", + _model_response(handler), + ) + wrap_function_wrapper( + _AGNO_MODELS_MODULE, + f"{_MODEL_CLASS}.aresponse", + _model_aresponse(handler), + ) + except (ImportError, AttributeError): + pass def unpatch_agent() -> None: - """Remove patches from agno.agent.Agent class methods.""" + """Remove patches from Agno class methods.""" try: import agno.agent # pylint: disable=import-outside-toplevel @@ -71,6 +118,13 @@ def unpatch_agent() -> None: 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 @@ -78,6 +132,21 @@ def unpatch_agent() -> None: 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 + try: + import agno.models.base # pylint: disable=import-outside-toplevel + + unwrap(agno.models.base.Model, "response") + unwrap(agno.models.base.Model, "aresponse") + except (ImportError, AttributeError): + pass def _extract_input_content(input_val: Any) -> str: @@ -136,16 +205,12 @@ def _set_tool_invocation_output( def _set_invocation_input( - invocation: AgentInvocation, + invocation: AgentInvocation | WorkflowInvocation, instance: Any, args: tuple[Any, ...], kwargs: dict[str, Any], capture_content: bool, ) -> None: - agent_id = getattr(instance, "agent_id", None) - if agent_id: - invocation.agent_id = str(agent_id) - if capture_content and (args or "input" in kwargs): input_val = args[0] if args else kwargs.get("input") if input_val is not None: @@ -156,7 +221,7 @@ def _set_invocation_input( def _set_invocation_output( - invocation: Any, + invocation: AgentInvocation | WorkflowInvocation, result: Any, capture_content: bool, ) -> None: @@ -164,12 +229,16 @@ def _set_invocation_output( output_str = _extract_output_content(result) invocation.output_messages = [ OutputMessage( - role="assistant", + role=str(getattr(result, "role", "assistant")), parts=[Text(content=output_str)], - finish_reason="stop", + finish_reason=str(getattr(result, "finish_reason", "stop")), ) ] - if hasattr(result, "session_id") and getattr(result, "session_id"): + if ( + isinstance(invocation, AgentInvocation) + and hasattr(result, "session_id") + and getattr(result, "session_id") + ): invocation.conversation_id = str(getattr(result, "session_id")) @@ -180,7 +249,7 @@ def _start_agent_invocation( kwargs: dict[str, Any], capture_content: bool, ) -> AgentInvocation: - agent_name = getattr(instance, "name", None) or "Agent" + agent_name = getattr(instance, "name", None) invocation = handler.invoke_local_agent(agent_name=agent_name) _set_invocation_input(invocation, instance, args, kwargs, capture_content) invocation.tool_definitions = prepare_tool_definitions( @@ -291,3 +360,155 @@ async def traced_method( return result return cast(Callable[..., Any], traced_method) + + +def _start_workflow_invocation( + handler: TelemetryHandler, + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + capture_content: bool, +) -> WorkflowInvocation: + workflow_name = getattr(instance, "name", None) + invocation = handler.workflow(name=workflow_name) + _set_invocation_input(invocation, instance, args, kwargs, capture_content) + return invocation + + +def _start_model_invocation( + handler: TelemetryHandler, + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + capture_content: bool, +) -> InferenceInvocation: + provider_name = ( + getattr(instance, "provider", None) + or instance.__class__.__name__ + or "agno" + ) + request_model = getattr(instance, "id", None) + invocation = handler.inference( + provider=str(provider_name), + request_model=str(request_model) if request_model else None, + ) + if capture_content and (args or "messages" in kwargs): + messages_val: Any = args[0] if args else kwargs.get("messages") + if messages_val and isinstance(messages_val, (list, tuple)): + input_msgs: list[InputMessage] = [] + for msg in cast(list[Any], messages_val): + role = str(getattr(msg, "role", "user")) + content: Any = getattr(msg, "content", "") + content_str = str(content) if content is not None else "" + input_msgs.append( + InputMessage(role=role, parts=[Text(content=content_str)]) + ) + if input_msgs: + invocation.input_messages = input_msgs + return invocation + + +def _set_model_invocation_output( + invocation: Any, + result: Any, + capture_content: bool, +) -> None: + if capture_content and result is not None: + content = getattr(result, "content", None) + if content is not None: + output_str = ( + str(content) if not isinstance(content, str) else content + ) + invocation.output_messages = [ + OutputMessage( + role=str(getattr(result, "role", "assistant")), + parts=[Text(content=output_str)], + finish_reason=str( + getattr(result, "finish_reason", "stop") + ), + ) + ] + + +def _workflow_run( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + def traced_method( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_workflow_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = wrapped(*args, **kwargs) + _set_invocation_output(invocation, result, capture_content) + return result + + return traced_method + + +def _workflow_arun( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + async def traced_method( + wrapped: Callable[..., Awaitable[Any]], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_workflow_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = await wrapped(*args, **kwargs) + _set_invocation_output(invocation, result, capture_content) + return result + + return cast(Callable[..., Any], traced_method) + + +def _model_response( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + def traced_method( + wrapped: Callable[..., Any], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_model_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = wrapped(*args, **kwargs) + _set_model_invocation_output(invocation, result, capture_content) + return result + + return traced_method + + +def _model_aresponse( + handler: TelemetryHandler, +) -> Callable[..., Any]: + capture_content = handler.should_capture_content() + + async def traced_method( + wrapped: Callable[..., Awaitable[Any]], + instance: Any, + args: tuple[Any, ...], + kwargs: dict[str, Any], + ) -> Any: + with _start_model_invocation( + handler, instance, args, kwargs, capture_content + ) as invocation: + result = await wrapped(*args, **kwargs) + _set_model_invocation_output(invocation, result, capture_content) + return result + + return cast(Callable[..., Any], traced_method) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py new file mode 100644 index 000000000..1925e0174 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/conformance/workflow.py @@ -0,0 +1,49 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""Conformance scenario: basic workflow run for Agno.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +pytest.importorskip("fastapi") +pytest.importorskip("agno.workflow.workflow") + +from agno.workflow.workflow import Workflow + +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 Scenario +from opentelemetry.test_util_genai.instrumentor import instrument + + +class WorkflowScenario(Scenario): + expected_spans = {"invoke_workflow": 1} + expected_metrics = ("gen_ai.invoke_workflow.duration",) + + 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", + ): + workflow = Workflow( + name="test-conformance-workflow", + steps=[], + session_id="session-workflow", + ) + workflow.run("hello workflow conformance") diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt index 4706b0aeb..56455ff9f 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.latest.txt @@ -39,6 +39,7 @@ agno wrapt>=2.2.2 +fastapi # test with the latest version of opentelemetry-api, sdk, semantic conventions, and instrumentation -e util/opentelemetry-util-genai diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt index 8eecc5b0c..37d900aae 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/requirements.oldest.txt @@ -21,5 +21,6 @@ # OpenTelemetry SDK and test utilities come transitively from opentelemetry-test-util-genai, which # every oldest env installs. Pin here only test-only deps that nothing else already provides. # -# There is currently nothing to pin: agno has no test-only dependency that isn't already -# provided by a declared bound or by the shared test fixtures. +# Drop this once opentelemetry-util-genai 1.1b0 is published. +-e util/opentelemetry-util-genai +fastapi >= 0.100.0 diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py index 7b418cd08..c6f516867 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_agent.py @@ -8,8 +8,11 @@ import asyncio from unittest.mock import patch +import pytest from agno.agent import Agent +from agno.models.message import Message from agno.models.response import ModelResponse +from agno.team import Team from agno.tools.function import Function, FunctionCall from tests.mock_model import MockModel @@ -150,3 +153,168 @@ async def _run_async() -> None: assert ( span.attributes.get(GenAIAttributes.GEN_AI_TOOL_CALL_ID) == "call-456" ) + + +def test_team_run_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Team.run emits an invoke_agent span.""" + member = Agent(name="member-agent", model=MockModel(id="mock-model")) + team = Team( + name="test-sync-team", + members=[member], + model=MockModel(id="mock-model"), + ) + mock_output = ModelResponse(content="Hello back from team!") + + with patch("agno.models.base.Model.response", return_value=mock_output): + res = team.run("hello team world") + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "invoke_agent test-sync-team" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_agent" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_NAME) + == "test-sync-team" + ) + + +def test_team_arun_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Team.arun emits an invoke_agent span.""" + member = Agent(name="member-agent", model=MockModel(id="mock-model")) + team = Team( + name="test-async-team", + members=[member], + model=MockModel(id="mock-model"), + ) + mock_output = ModelResponse(content="Async hello back from team!") + + async def _run_async() -> None: + with patch( + "agno.models.base.Model.aresponse", return_value=mock_output + ): + res = await team.arun("hello async team world") + assert res is not None + + asyncio.run(_run_async()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "invoke_agent test-async-team" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_agent" + ) + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_AGENT_NAME) + == "test-async-team" + ) + + +def test_model_response_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Model.response emits a chat span.""" + model = MockModel(id="test-mock-model") + mock_output = ModelResponse(content="Hello from model!", role="assistant") + + with patch.object(MockModel, "invoke", return_value=mock_output): + res = model.response( + messages=[Message(role="user", content="hello model")] + ) + assert res is not None + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "chat test-mock-model" + assert span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) == "chat" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "test-mock-model" + ) + + +def test_model_aresponse_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Model.aresponse emits a chat span.""" + model = MockModel(id="test-mock-model-async") + mock_output = ModelResponse( + content="Async hello from model!", role="assistant" + ) + + async def _run_async() -> None: + with patch.object(MockModel, "ainvoke", return_value=mock_output): + res = await model.aresponse( + messages=[Message(role="user", content="hello async model")] + ) + assert res is not None + + asyncio.run(_run_async()) + + spans = span_exporter.get_finished_spans() + assert len(spans) == 1 + span = spans[0] + assert span.name == "chat test-mock-model-async" + assert span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) == "chat" + assert ( + span.attributes.get(GenAIAttributes.GEN_AI_REQUEST_MODEL) + == "test-mock-model-async" + ) + + +def test_workflow_run_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Workflow.run emits an invoke_workflow span.""" + pytest.importorskip("fastapi") + pytest.importorskip("agno.workflow.workflow") + from agno.workflow.workflow import Workflow + + workflow = Workflow(name="test-workflow", steps=[]) + with patch.object(Workflow, "run", wraps=workflow.run): + workflow.run("test input") + + spans = span_exporter.get_finished_spans() + assert any( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_workflow" + for span in spans + ) + + +@pytest.mark.asyncio +async def test_workflow_arun_spans( + instrument_agno, + span_exporter, +) -> None: + """Test that Workflow.arun emits an invoke_workflow span.""" + pytest.importorskip("fastapi") + pytest.importorskip("agno.workflow.workflow") + from agno.workflow.workflow import Workflow + + workflow = Workflow(name="test-workflow-async", steps=[]) + with patch.object(Workflow, "arun", wraps=workflow.arun): + await workflow.arun("test input") + + spans = span_exporter.get_finished_spans() + assert any( + span.attributes.get(GenAIAttributes.GEN_AI_OPERATION_NAME) + == "invoke_workflow" + for span in spans + ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py index 86e887d4c..63d8d2ab6 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-agno/tests/test_conformance.py @@ -21,12 +21,14 @@ ) from .conformance.agent import AgentScenario +from .conformance.workflow import WorkflowScenario @pytest.mark.parametrize( "scenario", [ pytest.param(AgentScenario()), + pytest.param(WorkflowScenario()), ], ids=lambda s: type(s).__name__, ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py index b2bbeb045..848b4e1b3 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py @@ -35,6 +35,7 @@ class WorkflowScenario(Scenario): expected_metrics = ( "gen_ai.client.operation.duration", "gen_ai.client.token.usage", + "gen_ai.invoke_workflow.duration", ) # langchain can't populate server.address on chat spans. expected_violations = ( diff --git a/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py b/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py index 85c074cb4..bc6d94f7d 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py +++ b/instrumentation/opentelemetry-instrumentation-genai-openai-agents/tests/conformance/orchestration.py @@ -75,7 +75,10 @@ class OrchestrationScenario(Scenario): "invoke_agent": 2, "execute_tool": 1, } - expected_metrics = ("gen_ai.client.operation.duration",) + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.invoke_workflow.duration", + ) expected_violations = ( # `FunctionSpanData` in the openai-agents library doesn't expose # `tool_call_id`, so our `execute_tool` spans can't set diff --git a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py index d22801d96..efa5dadbf 100644 --- a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py +++ b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/_setup_weaver.py @@ -207,25 +207,22 @@ def _provision_genai_root() -> Path: ) upstream_pins = _load_version_pins(genai_target / "versions.env") - try: - upstream_version = upstream_pins["SEMCONV_VERSION"] - except KeyError as missing: - raise RuntimeError( - f"genai repo's versions.env is missing {missing!s}" - ) from missing + upstream_version = upstream_pins.get("SEMCONV_VERSION") + if upstream_version is not None: + upstream_target = cache_root / f"upstream-{upstream_version}" + if not (upstream_target / "model").is_dir(): + upstream_archive_url = ( + "https://github.com/open-telemetry/semantic-conventions/" + f"archive/refs/tags/{upstream_version}.tar.gz" + ) + _download_and_extract( + upstream_archive_url, upstream_target, label="upstream-semconv" + ) - upstream_target = cache_root / f"upstream-{upstream_version}" - if not (upstream_target / "model").is_dir(): - upstream_archive_url = ( - "https://github.com/open-telemetry/semantic-conventions/" - f"archive/refs/tags/{upstream_version}.tar.gz" + filtered = _materialize_filtered_upstream( + genai_target, upstream_target ) - _download_and_extract( - upstream_archive_url, upstream_target, label="upstream-semconv" - ) - - filtered = _materialize_filtered_upstream(genai_target, upstream_target) - _rewrite_manifest_dependency(genai_target, filtered) + _rewrite_manifest_dependency(genai_target, filtered) stamp.touch() return genai_target diff --git a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py index 516d99a72..e30cfd947 100644 --- a/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py +++ b/util/opentelemetry-test-util-genai/src/opentelemetry/test_util_genai/conformance.py @@ -72,9 +72,7 @@ class ExpectedViolation: def matches(self, violation: dict[str, Any]) -> bool: return violation.get( "id" - ) == self.advice_id and self.message_substring in str( - violation.get("message", "") - ) + ) == self.advice_id and self.message_substring in str(violation) class Scenario(ABC): diff --git a/util/opentelemetry-util-genai/.changelog/350.fixed b/util/opentelemetry-util-genai/.changelog/350.fixed new file mode 100644 index 000000000..52f754115 --- /dev/null +++ b/util/opentelemetry-util-genai/.changelog/350.fixed @@ -0,0 +1 @@ +Record `gen_ai.invoke_workflow.duration` metric on workflow invocations (`WorkflowInvocation`). diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py index 8c48dba51..52e0a65b3 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py @@ -12,6 +12,8 @@ from types import TracebackType from typing import TYPE_CHECKING, Any, TypeAlias +from typing_extensions import Self + from opentelemetry._logs import Logger, LogRecord from opentelemetry.context import Context, attach, detach from opentelemetry.semconv._incubating.attributes import ( @@ -211,7 +213,7 @@ def fail(self, error: Error | BaseException) -> None: error = Error.from_exception(error, self._error_type_resolver) self._finish(error) - def __enter__(self) -> GenAIInvocation: + def __enter__(self) -> Self: return self def __exit__( diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py index f06d72dd3..042c59c72 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py @@ -87,6 +87,15 @@ def _get_messages_for_span(self) -> dict[str, AttributeValue]: key: value for key, value in optional_attrs if value is not None } + def _get_metric_attributes(self) -> dict[str, AttributeValue]: + attrs: dict[str, AttributeValue] = { + GenAI.GEN_AI_OPERATION_NAME: self._operation_name, + } + if self._name is not None: + attrs[GenAI.GEN_AI_WORKFLOW_NAME] = self._name + attrs.update(self.metric_attributes) + return attrs + def _apply_finish(self, error: Error | None = None) -> None: attributes: dict[str, AttributeValue] = self._get_messages_for_span() if error is not None: @@ -97,4 +106,4 @@ def _apply_finish(self, error: Error | None = None) -> None: inputs=self.input_messages, outputs=self.output_messages, ) - # TODO: Add workflow metrics when supported + self._metrics_recorder.record_workflow(self) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py index c697e89d2..445ce86d6 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/instruments.py @@ -48,6 +48,20 @@ def create_duration_histogram(meter: Meter) -> Histogram: ) +def create_workflow_duration_histogram(meter: Meter) -> Histogram: + name = getattr( + gen_ai_metrics, + "GEN_AI_INVOKE_WORKFLOW_DURATION", + "gen_ai.invoke_workflow.duration", + ) + return meter.create_histogram( + name=name, + description="Measures the duration of a workflow execution.", + unit="s", + explicit_bucket_boundaries_advisory=_GEN_AI_CLIENT_OPERATION_DURATION_BUCKETS, + ) + + def create_token_histogram(meter: Meter) -> Histogram: return meter.create_histogram( name=gen_ai_metrics.GEN_AI_CLIENT_TOKEN_USAGE, diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py index 57796c77d..be45f2105 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/metrics.py @@ -17,6 +17,7 @@ create_time_per_output_chunk_histogram, create_time_to_first_chunk_histogram, create_token_histogram, + create_workflow_duration_histogram, ) from opentelemetry.util.types import Attributes @@ -28,6 +29,9 @@ class InvocationMetricsRecorder: def __init__(self, meter: Meter): self._duration_histogram: Histogram = create_duration_histogram(meter) + self._workflow_duration_histogram: Histogram = ( + create_workflow_duration_histogram(meter) + ) self._token_histogram: Histogram = create_token_histogram(meter) self._time_to_first_chunk_histogram: Histogram = ( create_time_to_first_chunk_histogram(meter) @@ -58,6 +62,19 @@ def record(self, invocation: GenAIInvocation) -> None: context=invocation._span_context, ) + def record_workflow(self, invocation: GenAIInvocation) -> None: + """Record duration metric for a workflow invocation.""" + attributes = invocation._get_metric_attributes() + duration_seconds = max( + timeit.default_timer() - invocation._monotonic_start_s, + 0.0, + ) + self._workflow_duration_histogram.record( + duration_seconds, + attributes=attributes, + context=invocation._span_context, + ) + def record_time_to_first_chunk( self, ttfc_seconds: float, diff --git a/util/opentelemetry-util-genai/tests/test_handler_workflow.py b/util/opentelemetry-util-genai/tests/test_handler_workflow.py index cf45f90cb..5ad8131c2 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_workflow.py +++ b/util/opentelemetry-util-genai/tests/test_handler_workflow.py @@ -17,6 +17,7 @@ from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAI, ) +from opentelemetry.test.test_base import TestBase from opentelemetry.trace import INVALID_SPAN, SpanKind from opentelemetry.trace.status import StatusCode from opentelemetry.util.genai.handler import TelemetryHandler @@ -287,3 +288,39 @@ def test_workflow_context_manager_with_messages(self) -> None: spans[0].attributes[GenAI.GEN_AI_OPERATION_NAME], "invoke_workflow", ) + + +class TestWorkflowInvocationMetrics(TestBase): + def _harvest_metrics(self): + metrics = self.get_sorted_metrics() + metrics_by_name = {} + for metric in metrics or []: + points = metric.data.data_points or [] + metrics_by_name.setdefault(metric.name, []).extend(points) + return metrics_by_name + + def test_workflow_records_duration(self) -> None: + handler = TelemetryHandler( + tracer_provider=self.tracer_provider, + meter_provider=self.meter_provider, + ) + with patch("timeit.default_timer", return_value=1000.0): + invocation = handler.workflow(name="test-workflow") + + with patch("timeit.default_timer", return_value=1002.0): + invocation.stop() + + metrics = self._harvest_metrics() + self.assertIn("gen_ai.invoke_workflow.duration", metrics) + duration_points = metrics["gen_ai.invoke_workflow.duration"] + self.assertEqual(len(duration_points), 1) + duration_point = duration_points[0] + self.assertEqual( + duration_point.attributes[GenAI.GEN_AI_OPERATION_NAME], + "invoke_workflow", + ) + self.assertEqual( + duration_point.attributes[GenAI.GEN_AI_WORKFLOW_NAME], + "test-workflow", + ) + self.assertAlmostEqual(duration_point.sum, 2.0, places=3) diff --git a/versions.env b/versions.env index efe17e853..7114c53a7 100644 --- a/versions.env +++ b/versions.env @@ -6,4 +6,4 @@ WEAVER_VERSION=v0.25.1 # The genai semconv registry has no tagged releases yet, so we pin a SHA on `main`. # renovate: datasource=git-refs depName=open-telemetry/semantic-conventions-genai packageName=https://github.com/open-telemetry/semantic-conventions-genai.git versioning=git -SEMCONV_GENAI_REF=f77b9235f2ad49fe95b61e9809ca82bb08ef9d47 +SEMCONV_GENAI_REF=4a39b6ef1363fab57bc40e18c82abb32cc89ebcd