From 82ddc9af08b205a66f519e5462a267294a3aebae Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 3 Sep 2026 22:53:39 -0400 Subject: [PATCH 1/3] fix(langchain): trace nested workflows Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../genai/langchain/agent_context.py | 67 ++++---- .../genai/langchain/callback_handler.py | 48 ++++-- .../genai/langchain/operation_mapping.py | 35 +++-- .../tests/test_operation_mapping.py | 59 +++++++ .../tests/test_workflow.py | 145 ++++++++++++++++++ 5 files changed, 285 insertions(+), 69 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py index 40eaa25b3..dee89687e 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py @@ -1,17 +1,6 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 -"""Identify ``create_agent`` graph invocations from outside the callback API. - -LangChain callbacks cannot tell a nested ``create_agent`` root apart from any -other named runnable invoked inside a tool: the enclosing agent's ``config`` is -merged over the inner agent's own, so ``lc_agent_name`` and ``ls_integration`` -in the *callback* metadata describe the outer agent. The compiled graph's own -bound config is never shadowed, so this module reads the marker there and has -each graph entry point announce itself on a context stack that the callback -handler consults when the root run starts. -""" - from __future__ import annotations from collections.abc import AsyncIterator, Callable, Iterator, Mapping @@ -27,27 +16,19 @@ @dataclass -class _PendingAgent: - """An agent graph that has started running but whose root run is not seen yet.""" - +class _PendingGraph: name: str | None + is_agent: bool claimed: bool = False -_pending: ContextVar[tuple[_PendingAgent, ...]] = ContextVar( - "otel_genai_pending_agents", default=() +_pending: ContextVar[tuple[_PendingGraph, ...]] = ContextVar( + "otel_genai_pending_graphs", default=() ) -def claim_agent() -> _PendingAgent | None: - """Return the announcement if this run is a create_agent graph root. - - The announcement is made as the graph starts, so the first chain run to see - it is the graph's root. Only the innermost announcement is claimable, and - only once, so internal nodes fall through to metadata-based classification. - The root run's own name is not checked - ``with_config(run_name=...)`` - renames it without making it any less of an agent. - """ +def claim_graph() -> _PendingGraph | None: + """Return the announcement for the next graph root callback.""" pending = _pending.get() if not pending: return None @@ -98,14 +79,18 @@ def _agent_name(graph: Any) -> tuple[bool, str | None]: return False, None -def _push(name: str | None) -> _PendingAgent: - """Announce ``name`` as the innermost running agent.""" - entry = _PendingAgent(name) +def _workflow_name(graph: Any) -> str | None: + name = getattr(graph, "name", None) + return str(name) if name else None + + +def _push(name: str | None, is_agent: bool) -> _PendingGraph: + entry = _PendingGraph(name=name, is_agent=is_agent) _pending.set(_pending.get() + (entry,)) return entry -def _pop(entry: _PendingAgent) -> None: +def _pop(entry: _PendingGraph) -> None: """Withdraw ``entry``, tolerating a stack the caller's context no longer owns.""" pending = _pending.get() if pending and pending[-1] is entry: @@ -118,14 +103,14 @@ def wrap_stream( args: tuple[Any, ...], kwargs: dict[str, Any], ) -> Any: - """Announce an agent graph for the duration of ``Pregel.stream``. + """Announce a graph for the duration of ``Pregel.stream``. ``Pregel.invoke`` runs through ``stream``, so this covers both entry points. """ is_agent, name = _agent_name(instance) if not is_agent: - return wrapped(*args, **kwargs) - return _announce_at_stream_start(wrapped(*args, **kwargs), name) + name = _workflow_name(instance) + return _announce_at_stream_start(wrapped(*args, **kwargs), name, is_agent) def wrap_astream( @@ -134,18 +119,20 @@ def wrap_astream( args: tuple[Any, ...], kwargs: dict[str, Any], ) -> Any: - """Announce an agent graph for the duration of ``Pregel.astream``. + """Announce a graph for the duration of ``Pregel.astream``. ``Pregel.ainvoke`` runs through ``astream``, so this covers both entry points. """ is_agent, name = _agent_name(instance) if not is_agent: - return wrapped(*args, **kwargs) - return _announce_at_astream_start(wrapped(*args, **kwargs), name) + name = _workflow_name(instance) + return _announce_at_astream_start(wrapped(*args, **kwargs), name, is_agent) def _announce_at_stream_start( - stream: Iterator[Any], name: str | None + stream: Iterator[Any], + name: str | None, + is_agent: bool, ) -> Iterator[Any]: """Announce ``name`` for the first step of ``stream`` only. @@ -158,7 +145,7 @@ def _announce_at_stream_start( # A generator body runs in the consumer's context, so the announcement lands # where the callbacks fire - on the first ``next()``, not here. iterator = iter(stream) - entry = _push(name) + entry = _push(name, is_agent) try: first = next(iterator) except StopIteration: @@ -170,11 +157,13 @@ def _announce_at_stream_start( async def _announce_at_astream_start( - stream: AsyncIterator[Any], name: str | None + stream: AsyncIterator[Any], + name: str | None, + is_agent: bool, ) -> AsyncIterator[Any]: """Announce ``name`` for the first step of ``stream`` only.""" iterator = stream.__aiter__() - entry = _push(name) + entry = _push(name, is_agent) try: first = await iterator.__anext__() except StopAsyncIteration: diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py index 2a45e365a..6cb9e97b3 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py @@ -18,7 +18,7 @@ ) from opentelemetry.instrumentation.genai.langchain.agent_context import ( - claim_agent, + claim_graph, ) from opentelemetry.instrumentation.genai.langchain.invocation_manager import ( _InvocationManager, @@ -103,25 +103,41 @@ def on_chain_start( parent_agent, ancestor_agent_names = self._find_agent_context( parent_run_id ) - # A claimed announcement is proof this run is a create_agent root, which - # the callback metadata alone cannot establish for a nested agent. - agent_announcement = claim_agent() + graph_announcement = claim_graph() + announced_agent = ( + graph_announcement is not None and graph_announcement.is_agent + ) + announced_workflow = ( + graph_announcement is not None and not graph_announcement.is_agent + ) declared_agent_name = ( - agent_announcement.name if agent_announcement else None + graph_announcement.name + if graph_announcement and graph_announcement.is_agent + else None + ) + declared_workflow_name = ( + graph_announcement.name + if graph_announcement and not graph_announcement.is_agent + else None ) operation = classify_chain_run( - serialized, - metadata, - kwargs, - parent_run_id, - declared_agent_name, - agent_announcement is not None, - ancestor_agent_names, + serialized=serialized, + metadata=metadata, + kwargs=kwargs, + parent_run_id=parent_run_id, + declared_agent_name=declared_agent_name, + announced_agent=announced_agent, + ancestor_agent_names=ancestor_agent_names, + announced_workflow=announced_workflow, ) conversation_id = _conversation_id(metadata) capture_content = self._telemetry_handler.should_capture_content() if operation == OperationName.INVOKE_WORKFLOW: - workflow_name = kwargs.get("name") or serialized.get("name") + workflow_name = ( + kwargs.get("name") + or serialized.get("name") + or declared_workflow_name + ) workflow_name_override = ( metadata.get("workflow_name") if metadata else None ) @@ -142,7 +158,7 @@ def on_chain_start( kwargs, declared_agent_name, ancestor_agent_names, - agent_announcement is not None, + announced_agent, ) # find if there is an agent already agent_invocation = parent_agent @@ -160,7 +176,7 @@ def on_chain_start( # non-announced runs, suppress a repeated metadata name matching the # enclosing agent - that repetition is inherited config, not a new agent. if ( - agent_announcement is not None + announced_agent or suggested_agent_name_lower != agent_invocation_name_lower ): @@ -185,7 +201,7 @@ def on_chain_start( self._invocation_manager.add_invocation_state( run_id, parent_run_id, None ) - elif agent_announcement is not None: + elif announced_agent: agent = self._telemetry_handler.invoke_local_agent( agent_name=None, ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py index b6c6f09b4..d9adc1dda 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py @@ -153,25 +153,19 @@ def _has_agent_signals( and str(metadata_name).lower() in ancestor_agent_names ) return bool( - metadata.get(_META_AGENT_SPAN) - or (metadata_name and not inherited_name) + (metadata_name and not inherited_name) or metadata.get(_META_AGENT_TYPE) ) def _looks_like_workflow( serialized: dict[str, Any], - metadata: dict[str, Any] | None, parent_run_id: UUID | None, ) -> bool: """Return True if the chain looks like a top-level workflow/graph.""" if parent_run_id is not None: return False - # An explicit workflow override is authoritative. - if metadata and metadata.get(_META_WORKFLOW_SPAN): - return True - # Heuristic: check for LangGraph identifier in the serialized repr. if serialized: name = serialized.get("name", "") @@ -223,6 +217,7 @@ def _should_ignore_chain( metadata.get(_META_AGENT_SPAN) is False and not metadata.get(_META_AGENT_NAME) and not metadata.get(_META_AGENT_TYPE) + and not metadata.get(_META_WORKFLOW_SPAN) ): return True @@ -247,6 +242,7 @@ def classify_chain_run( declared_agent_name: str | None = None, announced_agent: bool = False, ancestor_agent_names: set[str] | None = None, + announced_workflow: bool = False, ) -> str | None: """Classify a ``on_chain_start`` callback into a semconv operation. @@ -255,9 +251,10 @@ def classify_chain_run( Classification order: 1. Check for explicit suppression signals. - 2. Check for agent signals → ``invoke_agent``. - 3. Check for workflow signals → ``invoke_workflow``. - 4. Default: ``None`` (suppress – unclassified chains are not emitted). + 2. Honor explicit agent and workflow overrides. + 3. Prefer nested graph announcements over inherited agent metadata. + 4. Check remaining agent and workflow signals. + 5. Suppress unclassified chains. """ agent_name = resolve_agent_name( serialized, @@ -276,13 +273,23 @@ def classify_chain_run( if ( announced_agent or declared_agent_name - or _has_agent_signals(metadata, ancestor_agent_names) + or (metadata and metadata.get(_META_AGENT_SPAN)) ): return OperationName.INVOKE_AGENT - # 3. Workflow / orchestration detection. - if _looks_like_workflow(serialized, metadata, parent_run_id): + if metadata and metadata.get(_META_WORKFLOW_SPAN): + return OperationName.INVOKE_WORKFLOW + + # 3. A nested graph announcement is stronger than inherited agent metadata. + if announced_workflow and parent_run_id is not None: + return OperationName.INVOKE_WORKFLOW + + if _has_agent_signals(metadata, ancestor_agent_names): + return OperationName.INVOKE_AGENT + + # 4. Workflow / orchestration detection. + if announced_workflow or _looks_like_workflow(serialized, parent_run_id): return OperationName.INVOKE_WORKFLOW - # 4. Default: suppress unclassified chains. + # 5. Default: suppress unclassified chains. return None diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py index 18c63e165..d358adc49 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py @@ -190,6 +190,65 @@ def test_explicit_workflow_override_at_root(self): ) assert result == OperationName.INVOKE_WORKFLOW + def test_explicit_workflow_override_with_parent(self): + result = classify_chain_run( + serialized={"name": "SomeName"}, + metadata={"otel_workflow_span": True}, + kwargs={}, + parent_run_id=uuid.uuid4(), + ) + assert result == OperationName.INVOKE_WORKFLOW + + @pytest.mark.parametrize( + ("metadata", "announced_workflow", "expected"), + [ + pytest.param( + {"agent_type": "outer"}, + True, + OperationName.INVOKE_WORKFLOW, + id="announcement-beats-inherited-agent-type", + ), + pytest.param( + {"otel_agent_span": True}, + True, + OperationName.INVOKE_AGENT, + id="explicit-agent-beats-announcement", + ), + pytest.param( + { + "otel_workflow_span": True, + "agent_type": "inherited", + }, + False, + OperationName.INVOKE_WORKFLOW, + id="explicit-workflow-beats-agent-type", + ), + pytest.param( + { + "otel_workflow_span": True, + "otel_agent_span": False, + }, + False, + OperationName.INVOKE_WORKFLOW, + id="explicit-workflow-beats-agent-suppression", + ), + ], + ) + def test_workflow_override_precedence( + self, + metadata: dict[str, Any], + announced_workflow: bool, + expected: str, + ) -> None: + result = classify_chain_run( + serialized={}, + metadata=metadata, + kwargs={"name": "named_subgraph"}, + parent_run_id=uuid.uuid4(), + announced_workflow=announced_workflow, + ) + assert result == expected + def test_root_chain_with_no_signals_is_workflow(self): # A root chain (no parent) with no special names defaults to workflow. result = classify_chain_run( diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py new file mode 100644 index 000000000..a4d56e23d --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py @@ -0,0 +1,145 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from typing import Any, TypedDict + +import pytest +from langchain_core.messages import AIMessage, BaseMessage, HumanMessage +from langchain_core.runnables import RunnableLambda + +from opentelemetry.instrumentation.genai.langchain import LangChainInstrumentor +from opentelemetry.semconv._incubating.attributes import ( + gen_ai_attributes as GenAI, +) +from opentelemetry.test_util_genai.instrumentor import instrument + +langgraph_graph = pytest.importorskip("langgraph.graph") +END = langgraph_graph.END +START = langgraph_graph.START +StateGraph = langgraph_graph.StateGraph + + +class _State(TypedDict): + messages: list[BaseMessage] + + +def _respond(_: _State) -> _State: + return {"messages": [AIMessage(content="done")]} + + +def _graph(node: Any, *, name: str | None = None) -> Any: + builder = StateGraph(_State) + builder.add_node("step", node) + builder.add_edge(START, "step") + builder.add_edge("step", END) + return builder.compile(name=name) + + +def _nested_graph() -> Any: + return _graph(_graph(_respond, name="named_subgraph")) + + +def _workflow_spans(span_exporter: Any) -> list[Any]: + return [ + span + for span in span_exporter.get_finished_spans() + if span.attributes + and span.attributes.get(GenAI.GEN_AI_OPERATION_NAME) + == "invoke_workflow" + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "async_mode", + [False, True], + ids=["sync", "async"], +) +async def test_nested_graph_emits_workflow_span( + tracer_provider, + meter_provider, + logger_provider, + span_exporter, + async_mode: bool, +) -> None: + with instrument( + LangChainInstrumentor(), + tracer_provider=tracer_provider, + meter_provider=meter_provider, + logger_provider=logger_provider, + content_capture="SPAN_ONLY", + ): + graph = _nested_graph() + inputs = {"messages": [HumanMessage(content="hello")]} + result = ( + await graph.ainvoke(inputs) if async_mode else graph.invoke(inputs) + ) + + assert result == {"messages": [AIMessage(content="done")]} + workflow_spans = _workflow_spans(span_exporter) + assert len(workflow_spans) == 2 + workflow_spans_by_name = { + span.attributes[GenAI.GEN_AI_WORKFLOW_NAME]: span + for span in workflow_spans + } + assert set(workflow_spans_by_name) == {"LangGraph", "named_subgraph"} + inner_span = workflow_spans_by_name["named_subgraph"] + outer_span = workflow_spans_by_name["LangGraph"] + if not async_mode: + assert inner_span.parent.span_id == outer_span.context.span_id + assert all( + GenAI.GEN_AI_INPUT_MESSAGES in span.attributes + for span in workflow_spans + ) + assert all( + GenAI.GEN_AI_OUTPUT_MESSAGES in span.attributes + for span in workflow_spans + ) + + +def test_nested_runnable_sequence_is_not_workflow( + tracer_provider, + meter_provider, + logger_provider, + span_exporter, +) -> None: + sequence = RunnableLambda(_respond) | RunnableLambda(_respond) + + with instrument( + LangChainInstrumentor(), + tracer_provider=tracer_provider, + meter_provider=meter_provider, + logger_provider=logger_provider, + ): + _graph(sequence).invoke({"messages": [HumanMessage(content="hello")]}) + + assert len(_workflow_spans(span_exporter)) == 1 + + +def test_nested_graph_under_agent_metadata_is_workflow( + tracer_provider, + meter_provider, + logger_provider, + span_exporter, +) -> None: + with instrument( + LangChainInstrumentor(), + tracer_provider=tracer_provider, + meter_provider=meter_provider, + logger_provider=logger_provider, + ): + _nested_graph().invoke( + {"messages": [HumanMessage(content="hello")]}, + config={"metadata": {"agent_type": "outer"}}, + ) + + spans = span_exporter.get_finished_spans() + span_names = {span.name for span in spans} + assert "invoke_workflow named_subgraph" in span_names + assert "invoke_agent named_subgraph" not in span_names + nested_span = next( + span for span in spans if span.name == "invoke_workflow named_subgraph" + ) + assert nested_span.parent is not None From b5f44c858a8a61f0ef5e5f6e2aa1b4e307c1080b Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Thu, 3 Sep 2026 23:40:43 -0400 Subject: [PATCH 2/3] docs(langchain): add changelog for #617 Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../.changelog/617.fixed | 1 + 1 file changed, 1 insertion(+) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/617.fixed diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/617.fixed b/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/617.fixed new file mode 100644 index 000000000..c17ad52ab --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/.changelog/617.fixed @@ -0,0 +1 @@ +Emit ``invoke_workflow`` spans for nested LangGraph subgraphs. From 66b7e04a3a4c9ea6dc080fc99fdf3af7867cbdb8 Mon Sep 17 00:00:00 2001 From: 1fanwang <1fannnw@gmail.com> Date: Sat, 5 Sep 2026 11:16:50 -0400 Subject: [PATCH 3/3] fix(langchain): preserve nested graph classification Assisted-by: GitHub Copilot CLI (GPT-5.6 Sol) Signed-off-by: 1fanwang <1fannnw@gmail.com> --- .../genai/langchain/agent_context.py | 94 +++++--- .../genai/langchain/callback_handler.py | 118 ++++++++--- .../genai/langchain/invocation_manager.py | 18 +- .../genai/langchain/operation_mapping.py | 80 +++++-- .../tests/test_callback_handler.py | 5 +- .../tests/test_operation_mapping.py | 14 +- .../tests/test_workflow.py | 200 +++++++++++++++++- 7 files changed, 445 insertions(+), 84 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py index dee89687e..9fd3cc38c 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py @@ -1,6 +1,13 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 +"""Identify graph roots outside the LangChain callback API. + +Callback metadata cannot distinguish a nested graph root from its internal +runnables because parent config is inherited. Graph entry points announce the +root on a context stack so only the matching callback can claim it. +""" + from __future__ import annotations from collections.abc import AsyncIterator, Callable, Iterator, Mapping @@ -19,6 +26,7 @@ class _PendingGraph: name: str | None is_agent: bool + metadata: dict[str, Any] claimed: bool = False @@ -28,7 +36,11 @@ class _PendingGraph: def claim_graph() -> _PendingGraph | None: - """Return the announcement for the next graph root callback.""" + """Claim the innermost graph announcement for its root callback. + + Only the newest announcement is claimable, and only once, so nested graphs + do not leak their classification to internal callbacks. + """ pending = _pending.get() if not pending: return None @@ -58,34 +70,50 @@ def _react_agent_name(graph: Any) -> str | None: return str(name) if name and name != "LangGraph" else "" -def _agent_name(graph: Any) -> tuple[bool, str | None]: +def _bound_metadata(graph: Any) -> dict[str, Any]: + config = getattr(graph, "config", None) + if not isinstance(config, Mapping): + return {} + metadata = cast("Mapping[str, Any]", config).get("metadata") + if not isinstance(metadata, Mapping): + return {} + return dict(cast("Mapping[str, Any]", metadata)) + + +def _agent_name( + graph: Any, + metadata: Mapping[str, Any], +) -> tuple[bool, str | None]: """Return whether ``graph`` is an agent and its application-provided name.""" config = getattr(graph, "config", None) create_agent_name = create_agent_graph_name(config) if create_agent_name: return True, create_agent_name - if isinstance(config, Mapping): - metadata = cast("Mapping[str, Any]", config).get("metadata") - if isinstance(metadata, Mapping): - typed_metadata = cast("Mapping[str, Any]", metadata) - if ( - typed_metadata.get("ls_integration") - == "langchain_create_agent" - ): - return True, None + if metadata.get("ls_integration") == "langchain_create_agent": + return True, None react_name = _react_agent_name(graph) if react_name is not None: return True, react_name or None + if ( + metadata.get("otel_agent_span") + or metadata.get("agent_type") + or metadata.get("agent_name") + ): + name = metadata.get("agent_name") + return True, str(name) if name else None return False, None -def _workflow_name(graph: Any) -> str | None: - name = getattr(graph, "name", None) - return str(name) if name else None - - -def _push(name: str | None, is_agent: bool) -> _PendingGraph: - entry = _PendingGraph(name=name, is_agent=is_agent) +def _push( + name: str | None, + is_agent: bool, + metadata: dict[str, Any], +) -> _PendingGraph: + entry = _PendingGraph( + name=name, + is_agent=is_agent, + metadata=metadata, + ) _pending.set(_pending.get() + (entry,)) return entry @@ -107,10 +135,14 @@ def wrap_stream( ``Pregel.invoke`` runs through ``stream``, so this covers both entry points. """ - is_agent, name = _agent_name(instance) - if not is_agent: - name = _workflow_name(instance) - return _announce_at_stream_start(wrapped(*args, **kwargs), name, is_agent) + metadata = _bound_metadata(instance) + is_agent, name = _agent_name(instance, metadata) + return _announce_at_stream_start( + wrapped(*args, **kwargs), + name, + is_agent, + metadata, + ) def wrap_astream( @@ -123,16 +155,21 @@ def wrap_astream( ``Pregel.ainvoke`` runs through ``astream``, so this covers both entry points. """ - is_agent, name = _agent_name(instance) - if not is_agent: - name = _workflow_name(instance) - return _announce_at_astream_start(wrapped(*args, **kwargs), name, is_agent) + metadata = _bound_metadata(instance) + is_agent, name = _agent_name(instance, metadata) + return _announce_at_astream_start( + wrapped(*args, **kwargs), + name, + is_agent, + metadata, + ) def _announce_at_stream_start( stream: Iterator[Any], name: str | None, is_agent: bool, + metadata: dict[str, Any], ) -> Iterator[Any]: """Announce ``name`` for the first step of ``stream`` only. @@ -145,7 +182,7 @@ def _announce_at_stream_start( # A generator body runs in the consumer's context, so the announcement lands # where the callbacks fire - on the first ``next()``, not here. iterator = iter(stream) - entry = _push(name, is_agent) + entry = _push(name, is_agent, metadata) try: first = next(iterator) except StopIteration: @@ -160,10 +197,11 @@ async def _announce_at_astream_start( stream: AsyncIterator[Any], name: str | None, is_agent: bool, + metadata: dict[str, Any], ) -> AsyncIterator[Any]: """Announce ``name`` for the first step of ``stream`` only.""" iterator = stream.__aiter__() - entry = _push(name, is_agent) + entry = _push(name, is_agent, metadata) try: first = await iterator.__anext__() except StopAsyncIteration: diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py index 6cb9e97b3..450e18fa5 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py @@ -26,7 +26,9 @@ from opentelemetry.instrumentation.genai.langchain.operation_mapping import ( OperationName, classify_chain_run, + operation_metadata, resolve_agent_name, + without_inherited_operation_metadata, ) from opentelemetry.instrumentation.genai.langchain.utils import ( _legacy_function_call_request, @@ -100,10 +102,33 @@ def on_chain_start( metadata: dict[str, Any] | None = None, **kwargs: Any, ) -> Any: - parent_agent, ancestor_agent_names = self._find_agent_context( - parent_run_id - ) + ( + parent_agent, + ancestor_agent_names, + inherited_operation_metadata, + ) = self._find_agent_context(parent_run_id) graph_announcement = claim_graph() + effective_metadata = without_inherited_operation_metadata( + metadata, + inherited_operation_metadata, + ) + if graph_announcement and graph_announcement.metadata: + effective_metadata = { + **(effective_metadata or {}), + **graph_announcement.metadata, + } + if graph_announcement is not None: + graph_metadata = tuple( + item + for item in ( + *(inherited_operation_metadata or ()), + operation_metadata(metadata), + operation_metadata(graph_announcement.metadata), + ) + if item + ) + else: + graph_metadata = None announced_agent = ( graph_announcement is not None and graph_announcement.is_agent ) @@ -115,14 +140,9 @@ def on_chain_start( if graph_announcement and graph_announcement.is_agent else None ) - declared_workflow_name = ( - graph_announcement.name - if graph_announcement and not graph_announcement.is_agent - else None - ) operation = classify_chain_run( serialized=serialized, - metadata=metadata, + metadata=effective_metadata, kwargs=kwargs, parent_run_id=parent_run_id, declared_agent_name=declared_agent_name, @@ -133,13 +153,11 @@ def on_chain_start( conversation_id = _conversation_id(metadata) capture_content = self._telemetry_handler.should_capture_content() if operation == OperationName.INVOKE_WORKFLOW: - workflow_name = ( - kwargs.get("name") - or serialized.get("name") - or declared_workflow_name - ) + workflow_name = kwargs.get("name") workflow_name_override = ( - metadata.get("workflow_name") if metadata else None + effective_metadata.get("workflow_name") + if effective_metadata + else None ) workflow = self._telemetry_handler.workflow( name=workflow_name_override or workflow_name @@ -148,13 +166,16 @@ def on_chain_start( if capture_content: workflow.input_messages = make_input_message(inputs) self._invocation_manager.add_invocation_state( - run_id, parent_run_id, workflow + run_id, + parent_run_id, + workflow, + graph_metadata=graph_metadata, ) elif operation == OperationName.INVOKE_AGENT: # agent name passed by the user suggested_agent_name = resolve_agent_name( serialized, - metadata, + effective_metadata, kwargs, declared_agent_name, ancestor_agent_names, @@ -187,39 +208,61 @@ def on_chain_start( if capture_content: agent.input_messages = make_input_message(inputs) - if metadata: - agent.agent_id = metadata.get("agent_id") - agent.agent_description = metadata.get( + if effective_metadata: + agent.agent_id = effective_metadata.get("agent_id") + agent.agent_description = effective_metadata.get( "agent_description" ) self._invocation_manager.add_invocation_state( - run_id, parent_run_id, agent + run_id, + parent_run_id, + agent, + graph_metadata=graph_metadata, ) else: # We create invoke_agent span for the initial chain for agent. All follow-up chains invoked for agent invocation will not create agent span. self._invocation_manager.add_invocation_state( - run_id, parent_run_id, None + run_id, + parent_run_id, + None, + graph_metadata=graph_metadata, ) elif announced_agent: agent = self._telemetry_handler.invoke_local_agent( agent_name=None, ) - agent.input_messages = make_input_message(inputs) + agent.conversation_id = conversation_id + if capture_content: + agent.input_messages = make_input_message(inputs) + if effective_metadata: + agent.agent_id = effective_metadata.get("agent_id") + agent.agent_description = effective_metadata.get( + "agent_description" + ) self._invocation_manager.add_invocation_state( - run_id, parent_run_id, agent + run_id, + parent_run_id, + agent, + graph_metadata=graph_metadata, ) else: # No agent name could be resolved; still register the run_id so that # parent-child traversal through _find_agent_context is not broken for # any children of this node. self._invocation_manager.add_invocation_state( - run_id, parent_run_id, None + run_id, + parent_run_id, + None, + graph_metadata=graph_metadata, ) else: # For unclassified chains, we still want to track them in the invocation manager to maintain the parent-child relationships, even though we won't create spans for them. self._invocation_manager.add_invocation_state( - run_id, parent_run_id, None + run_id, + parent_run_id, + None, + graph_metadata=graph_metadata, ) def on_chain_end( @@ -638,7 +681,7 @@ def on_tool_start( arguments = json.loads(input_str) except (json.JSONDecodeError, ValueError): arguments = input_str - nearest_agent, _ = self._find_agent_context(parent_run_id) + nearest_agent, _, _ = self._find_agent_context(parent_run_id) agent_name = nearest_agent.agent_name if nearest_agent else None tool_invocation = self._telemetry_handler.tool( @@ -759,13 +802,26 @@ def on_retriever_error( def _find_agent_context( self, run_id: UUID | None - ) -> tuple[AgentInvocation | None, set[str]]: + ) -> tuple[ + AgentInvocation | None, + set[str], + tuple[dict[str, Any], ...] | None, + ]: current = run_id visited: set[UUID] = set() nearest_agent: AgentInvocation | None = None ancestor_agent_names: set[str] = set() + inherited_operation_metadata: tuple[dict[str, Any], ...] | None = None while current is not None and current not in visited: visited.add(current) + graph_metadata = self._invocation_manager.get_graph_metadata( + current + ) + if ( + inherited_operation_metadata is None + and graph_metadata is not None + ): + inherited_operation_metadata = graph_metadata entity = self._invocation_manager.get_invocation(current) if isinstance(entity, AgentInvocation): if nearest_agent is None: @@ -773,4 +829,8 @@ def _find_agent_context( if entity.agent_name: ancestor_agent_names.add(entity.agent_name.lower()) current = self._invocation_manager.get_parent_run_id(current) - return nearest_agent, ancestor_agent_names + return ( + nearest_agent, + ancestor_agent_names, + inherited_operation_metadata, + ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/invocation_manager.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/invocation_manager.py index b85309ce4..1eaa74de4 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/invocation_manager.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/invocation_manager.py @@ -2,6 +2,7 @@ # SPDX-License-Identifier: Apache-2.0 from dataclasses import dataclass, field +from typing import Any from uuid import UUID from opentelemetry.util.genai.types import GenAIInvocation @@ -14,6 +15,7 @@ class _InvocationState: invocation: GenAIInvocation | None children: list[UUID] = field(default_factory=lambda: list()) parent_run_id: UUID | None = None + graph_metadata: tuple[dict[str, Any], ...] | None = None ended: bool = False @@ -30,10 +32,14 @@ def add_invocation_state( run_id: UUID, parent_run_id: UUID | None, invocation: GenAIInvocation | None, + *, + graph_metadata: tuple[dict[str, Any], ...] | None = None, ) -> None: - invocation_state = _InvocationState(invocation=invocation) - - invocation_state.parent_run_id = parent_run_id + invocation_state = _InvocationState( + invocation=invocation, + parent_run_id=parent_run_id, + graph_metadata=graph_metadata, + ) if parent_run_id is not None and parent_run_id in self._invocations: parent_invocation_state = self._invocations[parent_run_id] parent_invocation_state.children.append(run_id) @@ -48,6 +54,12 @@ def get_parent_run_id(self, run_id: UUID) -> UUID | None: invocation_state = self._invocations.get(run_id) return invocation_state.parent_run_id if invocation_state else None + def get_graph_metadata( + self, run_id: UUID + ) -> tuple[dict[str, Any], ...] | None: + invocation_state = self._invocations.get(run_id) + return invocation_state.graph_metadata if invocation_state else None + def delete_invocation_state(self, run_id: UUID) -> None: invocation_state = self._invocations.get(run_id) if not invocation_state: diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py index d9adc1dda..04ad512d8 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py @@ -55,10 +55,22 @@ class OperationName: _META_WORKFLOW_SPAN = "otel_workflow_span" _META_AGENT_NAME = "agent_name" _META_AGENT_TYPE = "agent_type" +_META_AGENT_ID = "agent_id" +_META_AGENT_DESCRIPTION = "agent_description" +_META_WORKFLOW_NAME = "workflow_name" _META_LANGCHAIN_AGENT_NAME = "lc_agent_name" _META_LANGCHAIN_INTEGRATION = "ls_integration" _META_OTEL_TRACE = "otel_trace" _LANGCHAIN_CREATE_AGENT = "langchain_create_agent" +_OPERATION_METADATA_KEYS = ( + _META_AGENT_SPAN, + _META_WORKFLOW_SPAN, + _META_AGENT_NAME, + _META_AGENT_TYPE, + _META_AGENT_ID, + _META_AGENT_DESCRIPTION, + _META_WORKFLOW_NAME, +) # --------------------------------------------------------------------------- @@ -87,6 +99,33 @@ def create_agent_graph_name(config: Any) -> str | None: return str(name) if name else None +def operation_metadata( + metadata: Mapping[str, Any] | None, +) -> dict[str, Any]: + """Return operation markers inherited by graph child callbacks.""" + if not metadata: + return {} + return { + key: metadata[key] + for key in _OPERATION_METADATA_KEYS + if key in metadata + } + + +def without_inherited_operation_metadata( + metadata: dict[str, Any] | None, + inherited_operation_metadata: tuple[Mapping[str, Any], ...] | None, +) -> dict[str, Any] | None: + if not metadata or not inherited_operation_metadata: + return metadata + filtered_metadata = dict(metadata) + for inherited_metadata in inherited_operation_metadata: + for key, value in inherited_metadata.items(): + if key in filtered_metadata and filtered_metadata[key] == value: + filtered_metadata.pop(key) + return filtered_metadata + + def resolve_agent_name( serialized: dict[str, Any], metadata: dict[str, Any] | None, @@ -243,6 +282,7 @@ def classify_chain_run( announced_agent: bool = False, ancestor_agent_names: set[str] | None = None, announced_workflow: bool = False, + inherited_operation_metadata: tuple[Mapping[str, Any], ...] | None = None, ) -> str | None: """Classify a ``on_chain_start`` callback into a semconv operation. @@ -251,14 +291,18 @@ def classify_chain_run( Classification order: 1. Check for explicit suppression signals. - 2. Honor explicit agent and workflow overrides. - 3. Prefer nested graph announcements over inherited agent metadata. + 2. Honor graph announcements. + 3. Honor explicit agent and workflow overrides. 4. Check remaining agent and workflow signals. 5. Suppress unclassified chains. """ + effective_metadata = without_inherited_operation_metadata( + metadata, + inherited_operation_metadata, + ) agent_name = resolve_agent_name( serialized, - metadata, + effective_metadata, kwargs, declared_agent_name, ancestor_agent_names, @@ -266,29 +310,33 @@ def classify_chain_run( ) # 1. Suppress known noise. - if _should_ignore_chain(metadata, agent_name, kwargs, declared_agent_name): + if _should_ignore_chain( + effective_metadata, + agent_name, + kwargs, + declared_agent_name, + ): return None - # 2. Agent detection. - if ( - announced_agent - or declared_agent_name - or (metadata and metadata.get(_META_AGENT_SPAN)) - ): + # 2. Graph announcements come from the graph's own bound config. + if announced_agent or declared_agent_name: return OperationName.INVOKE_AGENT - if metadata and metadata.get(_META_WORKFLOW_SPAN): + if announced_workflow: return OperationName.INVOKE_WORKFLOW - # 3. A nested graph announcement is stronger than inherited agent metadata. - if announced_workflow and parent_run_id is not None: + # 3. Explicit callback metadata. + if effective_metadata and effective_metadata.get(_META_AGENT_SPAN): + return OperationName.INVOKE_AGENT + + if effective_metadata and effective_metadata.get(_META_WORKFLOW_SPAN): return OperationName.INVOKE_WORKFLOW - if _has_agent_signals(metadata, ancestor_agent_names): + # 4. Remaining callback signals. + if _has_agent_signals(effective_metadata, ancestor_agent_names): return OperationName.INVOKE_AGENT - # 4. Workflow / orchestration detection. - if announced_workflow or _looks_like_workflow(serialized, parent_run_id): + if _looks_like_workflow(serialized, parent_run_id): return OperationName.INVOKE_WORKFLOW # 5. Default: suppress unclassified chains. diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py index 4bea441ec..baf6c1cad 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py @@ -158,15 +158,16 @@ def test_workflow_span_created(self): handler._invocation_manager.get_invocation(run_id) is workflow_inv ) - def test_workflow_name_from_serialized(self): + def test_workflow_name_from_callback(self): handler, telemetry, _, _ = _make_handler() run_id = _run_id() handler.on_chain_start( - serialized={"name": "MyLangGraph"}, + serialized={}, inputs={}, run_id=run_id, parent_run_id=None, + name="MyLangGraph", ) telemetry.workflow.assert_called_once_with(name="MyLangGraph") diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py index d358adc49..bd56586f2 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py @@ -211,8 +211,8 @@ def test_explicit_workflow_override_with_parent(self): pytest.param( {"otel_agent_span": True}, True, - OperationName.INVOKE_AGENT, - id="explicit-agent-beats-announcement", + OperationName.INVOKE_WORKFLOW, + id="announcement-beats-inherited-agent-override", ), pytest.param( { @@ -249,6 +249,16 @@ def test_workflow_override_precedence( ) assert result == expected + def test_root_graph_announcement_beats_inherited_agent_metadata(self): + result = classify_chain_run( + serialized={}, + metadata={"agent_type": "outer"}, + kwargs={"name": "LangGraph"}, + parent_run_id=None, + announced_workflow=True, + ) + assert result == OperationName.INVOKE_WORKFLOW + def test_root_chain_with_no_signals_is_workflow(self): # A root chain (no parent) with no special names defaults to workflow. result = classify_chain_run( diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py index a4d56e23d..30efc253c 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py @@ -87,8 +87,6 @@ async def test_nested_graph_emits_workflow_span( assert set(workflow_spans_by_name) == {"LangGraph", "named_subgraph"} inner_span = workflow_spans_by_name["named_subgraph"] outer_span = workflow_spans_by_name["LangGraph"] - if not async_mode: - assert inner_span.parent.span_id == outer_span.context.span_id assert all( GenAI.GEN_AI_INPUT_MESSAGES in span.attributes for span in workflow_spans @@ -97,6 +95,186 @@ async def test_nested_graph_emits_workflow_span( GenAI.GEN_AI_OUTPUT_MESSAGES in span.attributes for span in workflow_spans ) + if async_mode: + pytest.xfail( + "nested spans are not parented in async: " + "https://github.com/open-telemetry/" + "opentelemetry-python-genai/issues/513" + ) + assert inner_span.parent.span_id == outer_span.context.span_id + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "async_mode", + [False, True], + ids=["sync", "async"], +) +@pytest.mark.parametrize( + ("agent_metadata", "agent_span_name"), + [ + pytest.param( + {"agent_name": "my_agent"}, + "invoke_agent my_agent", + id="agent-name", + ), + pytest.param( + {"agent_type": "custom"}, + "invoke_agent", + id="agent-type", + ), + pytest.param( + {"otel_agent_span": True}, + "invoke_agent", + id="agent-override", + ), + ], +) +async def test_nested_graph_with_bound_agent_metadata_is_agent( + tracer_provider, + meter_provider, + logger_provider, + span_exporter, + async_mode: bool, + agent_metadata: dict[str, Any], + agent_span_name: str, +) -> None: + subgraph = _graph(_respond, name="my_agent").with_config( + { + "metadata": { + **agent_metadata, + "thread_id": "thread-1", + "agent_id": "agent-1", + "agent_description": "test agent", + } + } + ) + graph = _graph(subgraph) + + with instrument( + LangChainInstrumentor(), + tracer_provider=tracer_provider, + meter_provider=meter_provider, + logger_provider=logger_provider, + content_capture="SPAN_ONLY", + ): + inputs = {"messages": [HumanMessage(content="hello")]} + outer_config: dict[str, Any] = { + "metadata": { + "agent_name": "outer", + "agent_id": "outer-id", + "agent_description": "outer agent", + } + } + result = ( + await graph.ainvoke(inputs, config=outer_config) + if async_mode + else graph.invoke(inputs, config=outer_config) + ) + + assert result == {"messages": [AIMessage(content="done")]} + operation_spans = [ + span + for span in span_exporter.get_finished_spans() + if span.attributes + and span.attributes.get(GenAI.GEN_AI_OPERATION_NAME) + in {"invoke_agent", "invoke_workflow"} + ] + agent_spans = [ + span + for span in operation_spans + if span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "invoke_agent" + ] + workflow_spans = [ + span + for span in operation_spans + if span.attributes[GenAI.GEN_AI_OPERATION_NAME] == "invoke_workflow" + ] + assert len(agent_spans) == 1 + assert [span.name for span in workflow_spans] == [ + "invoke_workflow LangGraph" + ] + assert agent_spans[0].name == agent_span_name + assert ( + agent_spans[0].attributes[GenAI.GEN_AI_CONVERSATION_ID] == "thread-1" + ) + assert agent_spans[0].attributes[GenAI.GEN_AI_AGENT_ID] == "agent-1" + assert ( + agent_spans[0].attributes[GenAI.GEN_AI_AGENT_DESCRIPTION] + == "test agent" + ) + assert GenAI.GEN_AI_INPUT_MESSAGES in agent_spans[0].attributes + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "async_mode", + [False, True], + ids=["sync", "async"], +) +@pytest.mark.parametrize( + ("operation_metadata", "operation_span_name"), + [ + pytest.param( + {"agent_name": "inner_agent"}, + "invoke_agent inner_agent", + id="agent-name", + ), + pytest.param( + {"agent_type": "custom"}, + "invoke_agent marked_node", + id="agent-type", + ), + pytest.param( + {"otel_agent_span": True}, + "invoke_agent marked_node", + id="agent-override", + ), + pytest.param( + {"otel_workflow_span": True}, + "invoke_workflow marked_node", + id="workflow-override", + ), + ], +) +async def test_graph_child_with_local_operation_metadata( + tracer_provider, + meter_provider, + logger_provider, + span_exporter, + async_mode: bool, + operation_metadata: dict[str, Any], + operation_span_name: str, +) -> None: + node = RunnableLambda(_respond).with_config( + run_name="marked_node", + metadata=operation_metadata, + ) + graph = _graph(node) + + with instrument( + LangChainInstrumentor(), + tracer_provider=tracer_provider, + meter_provider=meter_provider, + logger_provider=logger_provider, + ): + inputs = {"messages": [HumanMessage(content="hello")]} + result = ( + await graph.ainvoke(inputs) if async_mode else graph.invoke(inputs) + ) + + assert result == {"messages": [AIMessage(content="done")]} + operation_spans = [ + span + for span in span_exporter.get_finished_spans() + if span.attributes + and span.attributes.get(GenAI.GEN_AI_OPERATION_NAME) + in {"invoke_agent", "invoke_workflow"} + ] + assert {span.name for span in operation_spans} == { + operation_span_name, + "invoke_workflow LangGraph", + } def test_nested_runnable_sequence_is_not_workflow( @@ -118,11 +296,19 @@ def test_nested_runnable_sequence_is_not_workflow( assert len(_workflow_spans(span_exporter)) == 1 -def test_nested_graph_under_agent_metadata_is_workflow( +@pytest.mark.parametrize( + "agent_metadata", + [ + pytest.param({"agent_type": "outer"}, id="agent-type"), + pytest.param({"otel_agent_span": True}, id="agent-override"), + ], +) +def test_nested_graph_under_inherited_agent_metadata_is_workflow( tracer_provider, meter_provider, logger_provider, span_exporter, + agent_metadata: dict[str, Any], ) -> None: with instrument( LangChainInstrumentor(), @@ -132,13 +318,19 @@ def test_nested_graph_under_agent_metadata_is_workflow( ): _nested_graph().invoke( {"messages": [HumanMessage(content="hello")]}, - config={"metadata": {"agent_type": "outer"}}, + config={"metadata": agent_metadata}, ) spans = span_exporter.get_finished_spans() span_names = {span.name for span in spans} assert "invoke_workflow named_subgraph" in span_names assert "invoke_agent named_subgraph" not in span_names + assert not [ + span + for span in spans + if span.attributes + and span.attributes.get(GenAI.GEN_AI_OPERATION_NAME) == "invoke_agent" + ] nested_span = next( span for span in spans if span.name == "invoke_workflow named_subgraph" )