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 35320ae55..c9df02257 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 @@ -70,8 +70,12 @@ def on_chain_start( workflow_name_override = ( metadata.get("workflow_name") if metadata else None ) + root_operation_name = self._find_enclosing_root_operation_name( + parent_run_id + ) workflow = self._telemetry_handler.workflow( - name=workflow_name_override or workflow_name + name=workflow_name_override or workflow_name, + root_operation_name=root_operation_name, ) workflow.input_messages = make_input_message(inputs) self._invocation_manager.add_invocation_state( @@ -95,8 +99,14 @@ def on_chain_start( else None ) if suggested_agent_name_lower != agent_invocation_name_lower: + root_operation_name = ( + self._find_enclosing_root_operation_name( + parent_run_id + ) + ) agent = self._telemetry_handler.invoke_local_agent( agent_name=suggested_agent_name, + root_operation_name=root_operation_name, ) agent.input_messages = make_input_message(inputs) @@ -500,3 +510,36 @@ def _find_nearest_agent( return entity current = self._invocation_manager.get_parent_run_id(current) return None + + def _find_enclosing_root_operation_name( + self, run_id: Optional[UUID] + ) -> Optional[str]: + current = run_id + visited: set[UUID] = set() + while current is not None and current not in visited: + visited.add(current) + entity = self._invocation_manager.get_invocation(current) + if isinstance(entity, (AgentInvocation, WorkflowInvocation)): + if entity.root_operation_name: + return entity.root_operation_name + + # If this is the top-most enclosing invocation, derive the + # operation name from invocation fields. + parent = self._invocation_manager.get_parent_run_id(current) + if parent is None: + return self._resolve_operation_name(entity) + + parent = self._invocation_manager.get_parent_run_id(current) + current = parent + return None + + def _resolve_operation_name( + self, entity: AgentInvocation | WorkflowInvocation + ) -> str: + if isinstance(entity, AgentInvocation): + if entity.agent_name: + return f"invoke_agent {entity.agent_name}" + return "invoke_agent" + if entity.name: + return f"invoke_workflow {entity.name}" + return "invoke_workflow" 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 bf72bb0f1..e5a4938a4 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 @@ -107,36 +107,60 @@ def _has_agent_signals(metadata: Optional[dict[str, Any]]) -> bool: ) +def _is_langgraph_graph( + serialized: dict[str, Any], + kwargs: dict[str, Any], +) -> bool: + """Return True if the chain is a LangGraph graph (``Pregel``) invocation. + + LangGraph reports the graph itself under the ``LangGraph`` identifier as the + run name (``kwargs['name']`` at runtime, or ``serialized['name']`` / + ``serialized['graph']['id']`` in the serialized repr). Individual graph + nodes are reported under the node name instead, so this reliably + distinguishes a (sub)graph invocation from a node invocation. + """ + name = kwargs.get("name") + if not name and serialized: + name = serialized.get("name") + if name and LANGGRAPH_IDENTIFIER in str(name): + return True + + if serialized and isinstance(serialized.get("graph"), dict): + graph_id = serialized["graph"].get("id", "") + if LANGGRAPH_IDENTIFIER in str(graph_id): + return True + + return False + + def _looks_like_workflow( serialized: dict[str, Any], metadata: Optional[dict[str, Any]], + kwargs: dict[str, Any], parent_run_id: Optional[UUID], ) -> bool: - """Return True if the chain looks like a top-level workflow/graph.""" - if parent_run_id is not None: - return False + """Return True if the chain looks like a workflow/graph. + Both top-level graphs and nested subgraphs are treated as workflows, so a + multi-graph pipeline produces one ``invoke_workflow`` span per graph. A + nested subgraph is distinguished from a top-level workflow later, when the + span is created, by inspecting the run tree. + """ # 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", "") - graph_id = ( - serialized.get("graph", {}).get("id", "") - if isinstance(serialized.get("graph"), dict) - else "" - ) - return LANGGRAPH_IDENTIFIER in name or LANGGRAPH_IDENTIFIER in graph_id - - # No serialized data to inspect, but this is a top-level chain - # (parent_run_id is None). When we have zero information about a root-level - # chain we prefer to emit a span rather than silently drop it — more data - # is better than missing the outermost invocation entirely. Treat it as a - # workflow so the outermost operation always gets a span even when the - # chain didn't populate its serialized representation. - return True + # A LangGraph graph invocation, whether top-level or a nested subgraph. + if _is_langgraph_graph(serialized, kwargs): + return True + + # A root-level chain with no serialized data to inspect. We have zero + # information about it, but prefer emitting a span for the outermost + # invocation rather than silently dropping it. + if parent_run_id is None and not serialized: + return True + + return False # --------------------------------------------------------------------------- @@ -213,7 +237,7 @@ def classify_chain_run( return OperationName.INVOKE_AGENT # 3. Workflow / orchestration detection. - if _looks_like_workflow(serialized, metadata, parent_run_id): + if _looks_like_workflow(serialized, metadata, kwargs, parent_run_id): return OperationName.INVOKE_WORKFLOW # 4. Default: suppress unclassified chains. diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/cassettes/nested_workflow_conformance.yaml b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/cassettes/nested_workflow_conformance.yaml new file mode 100644 index 000000000..f43f07afa --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/cassettes/nested_workflow_conformance.yaml @@ -0,0 +1,180 @@ +interactions: +- request: + body: |- + { + "messages": [ + { + "content": "You are a research assistant. Provide 2-3 factual sentences.", + "role": "system" + }, + { + "content": "What is the capital of France?", + "role": "user" + } + ], + "model": "gpt-3.5-turbo", + "max_completion_tokens": 200, + "seed": 42, + "stream": false, + "temperature": 0.1 + } + headers: + Accept: + - application/json + Content-Type: + - application/json + Host: + - api.openai.com + authorization: + - Bearer test_openai_api_key + method: POST + uri: https://api.openai.com/v1/chat/completions + response: + body: + string: |- + { + "id": "chatcmpl-nested001research", + "object": "chat.completion", + "created": 1771535300, + "model": "gpt-3.5-turbo-0125", + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": "The capital of France is Paris. Paris sits on the Seine and has been the country's capital since the 10th century.", + "refusal": null, + "annotations": [] + }, + "logprobs": null, + "finish_reason": "stop" + } + ], + "usage": { + "prompt_tokens": 35, + "completion_tokens": 27, + "total_tokens": 62, + "prompt_tokens_details": { + "cached_tokens": 0, + "audio_tokens": 0 + }, + "completion_tokens_details": { + "reasoning_tokens": 0, + "audio_tokens": 0, + "accepted_prediction_tokens": 0, + "rejected_prediction_tokens": 0 + } + }, + "service_tier": "default", + "system_fingerprint": null + } + headers: + Connection: + - keep-alive + Content-Type: + - application/json + Date: + - Wed, 20 May 2026 00:07:12 GMT + Server: + - cloudflare + Set-Cookie: test_set_cookie + content-length: + - '640' + openai-organization: test_openai_org_id + openai-version: + - '2020-10-01' + x-request-id: + - req_nested001research + status: + code: 200 + message: OK +- request: + body: |- + { + "messages": [ + { + "content": "You are an expert summariser. Condense the text below into one clear sentence.", + "role": "system" + }, + { + "content": "The capital of France is Paris. Paris sits on the Seine and has been the country's capital since the 10th century.", + "role": "user" + } + ], + "model": "gpt-3.5-turbo", + "max_completion_tokens": 200, + "seed": 42, + "stream": false, + "temperature": 0.1 + } + headers: + Accept: + - application/json + Content-Type: + - application/json + Host: + - api.openai.com + authorization: + - Bearer test_openai_api_key + method: POST + uri: https://api.openai.com/v1/chat/completions + response: + body: + string: |- + { + "id": "chatcmpl-nested001summary", + "object": "chat.completion", + "created": 1771535301, + "model": "gpt-3.5-turbo-0125", + "choices": [ + { + "index": 0, + "message": { + "role": "assistant", + "content": "Paris, on the Seine, has been France's capital since the 10th century.", + "refusal": null, + "annotations": [] + }, + "logprobs": null, + "finish_reason": "stop" + } + ], + "usage": { + "prompt_tokens": 52, + "completion_tokens": 18, + "total_tokens": 70, + "prompt_tokens_details": { + "cached_tokens": 0, + "audio_tokens": 0 + }, + "completion_tokens_details": { + "reasoning_tokens": 0, + "audio_tokens": 0, + "accepted_prediction_tokens": 0, + "rejected_prediction_tokens": 0 + } + }, + "service_tier": "default", + "system_fingerprint": null + } + headers: + Connection: + - keep-alive + Content-Type: + - application/json + Date: + - Wed, 20 May 2026 00:07:13 GMT + Server: + - cloudflare + Set-Cookie: test_set_cookie + content-length: + - '600' + openai-organization: test_openai_org_id + openai-version: + - '2020-10-01' + x-request-id: + - req_nested001summary + status: + code: 200 + message: OK +version: 1 diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py index 745e272aa..5cf04d325 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py @@ -137,3 +137,131 @@ def validate(self, report: LiveCheckReport) -> None: "Two-node workflow exercises two chat completions " f"(researcher and summariser); saw {operations}" ) + + +class NestedWorkflowScenario(Scenario): + """A LangGraph graph whose node is itself a compiled subgraph. + + The outer graph and inner subgraph are both represented as + ``invoke_workflow`` spans. The inner span should carry + ``gen_ai.root_operation.name`` that points at the outer span name. + """ + + expected_spans = ("invoke_workflow", "invoke_workflow", "chat", "chat") + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.client.token.usage", + ) + expected_violations = ( + # langchain can't populate server.address on chat spans. + 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: + key_override = ( + {} + if os.getenv("OPENAI_API_KEY") + else {"OPENAI_API_KEY": "test_openai_api_key"} + ) + with mock.patch.dict(os.environ, key_override): + with instrument( + LangChainInstrumentor(), + tracer_provider=tracer_provider, + logger_provider=logger_provider, + meter_provider=meter_provider, + semconv="gen_ai_latest_experimental", + content_capture="SPAN_ONLY", + ): + llm = ChatOpenAI( + model="gpt-3.5-turbo", + temperature=0.1, + max_tokens=200, + seed=42, + ) + + def researcher(state: GraphState) -> dict: + response = llm.invoke( + [ + SystemMessage( + content="You are a research assistant. Provide 2-3 factual sentences." + ), + HumanMessage( + content=state["messages"][-1].content + ), + ] + ) + return { + "research": response.content, + "messages": [response], + } + + inner_builder = StateGraph(GraphState) + inner_builder.add_node("researcher", researcher) + inner_builder.add_edge(START, "researcher") + inner_builder.add_edge("researcher", END) + inner_graph = inner_builder.compile() + + def summariser(state: GraphState) -> dict: + response = llm.invoke( + [ + SystemMessage( + content="You are an expert summariser. Condense the text below into one clear sentence." + ), + HumanMessage(content=state["research"]), + ] + ) + return {"messages": [response]} + + outer_builder = StateGraph(GraphState) + outer_builder.add_node("data_gathering", inner_graph) + outer_builder.add_node("summariser", summariser) + outer_builder.add_edge(START, "data_gathering") + outer_builder.add_edge("data_gathering", "summariser") + outer_builder.add_edge("summariser", END) + graph = outer_builder.compile() + + with vcr.use_cassette("nested_workflow_conformance.yaml"): + graph.invoke( + { + "messages": [ + HumanMessage( + content="What is the capital of France?" + ) + ], + "research": "", + } + ) + + def validate(self, report: LiveCheckReport) -> None: + super().validate(report) + workflow_spans = [ + entry["span"] + for entry in report["samples"] + if "span" in entry + and any( + attr["name"] == "gen_ai.operation.name" + and attr["value"] == "invoke_workflow" + for attr in entry["span"]["attributes"] + ) + ] + root_operation_values = [ + attr["value"] + for span in workflow_spans + for attr in span["attributes"] + if attr["name"] == "gen_ai.root_operation.name" + ] + assert len(root_operation_values) == 1, ( + "Exactly one nested workflow span should set " + f"gen_ai.root_operation.name; saw {root_operation_values}" + ) + 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 f077bdd67..4cf3b88f7 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py @@ -48,6 +48,9 @@ def _make_agent_inv_mock() -> mock.MagicMock: # agent_name is an instance attribute set in AgentInvocation.__init__ via the # constructor arg; pre-configure it so spec-restricted attribute access works. agent_inv.agent_name = None + agent_inv.attributes = {} + agent_inv.metric_attributes = {} + agent_inv.root_operation_name = None return agent_inv @@ -57,6 +60,7 @@ def _make_invoke_local_agent_side_effect(inv: mock.MagicMock): def _side_effect(*args, **kwargs): inv.agent_name = kwargs.get("agent_name") + inv.root_operation_name = kwargs.get("root_operation_name") return inv return _side_effect @@ -152,6 +156,108 @@ def test_workflow_registered_in_invocation_manager(self): handler._invocation_manager.get_invocation(run_id) is workflow_inv ) + def test_nested_workflow_sets_root_operation_name(self): + handler, telemetry, _, _ = _make_handler() + parent_run_id = _run_id() + child_run_id = _run_id() + + created_workflows: list[mock.MagicMock] = [] + + def _create_workflow(*_args, **kwargs): + workflow_invocation = mock.MagicMock(spec=WorkflowInvocation) + workflow_invocation.span = mock.MagicMock() + workflow_invocation.span.is_recording.return_value = False + workflow_invocation.name = kwargs.get("name") + workflow_invocation.attributes = {} + workflow_invocation.root_operation_name = kwargs.get( + "root_operation_name" + ) + workflow_invocation.input_messages = [] + workflow_invocation.span.name = ( + f"invoke_workflow {workflow_invocation.name}" + if workflow_invocation.name + else "invoke_workflow" + ) + created_workflows.append(workflow_invocation) + return workflow_invocation + + telemetry.workflow.side_effect = _create_workflow + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=parent_run_id, + parent_run_id=None, + metadata={"workflow_name": "outer_flow"}, + ) + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=child_run_id, + parent_run_id=parent_run_id, + metadata={"workflow_name": "inner_flow"}, + ) + + assert len(created_workflows) == 2 + assert created_workflows[0].root_operation_name is None + assert ( + created_workflows[1].root_operation_name + == "invoke_workflow outer_flow" + ) + assert telemetry.workflow.call_args_list[1].kwargs == { + "name": "inner_flow", + "root_operation_name": "invoke_workflow outer_flow", + } + + def test_nested_workflow_reuses_parent_root_operation_name(self): + handler, telemetry, _, _ = _make_handler() + parent_run_id = _run_id() + child_run_id = _run_id() + + created_workflows: list[mock.MagicMock] = [] + + def _create_workflow(*_args, **kwargs): + workflow_invocation = mock.MagicMock(spec=WorkflowInvocation) + workflow_invocation.span = mock.MagicMock() + workflow_invocation.span.is_recording.return_value = False + workflow_invocation.name = kwargs.get("name") + workflow_invocation.attributes = {} + workflow_invocation.root_operation_name = kwargs.get( + "root_operation_name" + ) + workflow_invocation.input_messages = [] + workflow_invocation.span.name = ( + f"invoke_workflow {workflow_invocation.name}" + if workflow_invocation.name + else "invoke_workflow" + ) + created_workflows.append(workflow_invocation) + return workflow_invocation + + telemetry.workflow.side_effect = _create_workflow + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=parent_run_id, + parent_run_id=None, + metadata={"workflow_name": "outer_flow"}, + ) + created_workflows[0].root_operation_name = "invoke_workflow top_flow" + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=child_run_id, + parent_run_id=parent_run_id, + metadata={"workflow_name": "inner_flow"}, + ) + + assert telemetry.workflow.call_args_list[1].kwargs == { + "name": "inner_flow", + "root_operation_name": "invoke_workflow top_flow", + } + # --------------------------------------------------------------------------- # on_chain_start – INVOKE_AGENT @@ -337,6 +443,87 @@ def test_no_agent_name_registers_none_invocation(self): assert run_id in handler._invocation_manager._invocations assert handler._invocation_manager.get_invocation(run_id) is None + def test_agent_under_workflow_sets_root_operation_name(self): + handler, telemetry, workflow_inv, agent_inv = _make_handler() + workflow_run_id = _run_id() + agent_run_id = _run_id() + workflow_inv.span.name = "invoke_workflow orchestrator" + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=workflow_run_id, + parent_run_id=None, + ) + handler.on_chain_start( + serialized={"name": "math_agent"}, + inputs={}, + run_id=agent_run_id, + parent_run_id=workflow_run_id, + metadata={"agent_name": "math_agent"}, + ) + + assert agent_inv.root_operation_name == "invoke_workflow orchestrator" + assert ( + agent_inv.attributes["gen_ai.root_operation.name"] + == "invoke_workflow orchestrator" + ) + assert ( + agent_inv.metric_attributes["gen_ai.root_operation.name"] + == "invoke_workflow orchestrator" + ) + + def test_nested_agent_uses_outermost_enclosing_operation_name(self): + handler, telemetry, workflow_inv, _ = _make_handler() + workflow_run_id = _run_id() + parent_agent_run_id = _run_id() + child_agent_run_id = _run_id() + workflow_inv.span.name = "invoke_workflow orchestrator" + + created_agents: list[mock.MagicMock] = [] + + def _create_agent(*_args, **kwargs): + agent_invocation = _make_agent_inv_mock() + agent_invocation.agent_name = kwargs.get("agent_name") + agent_invocation.span.name = ( + f"invoke_agent {agent_invocation.agent_name}" + ) + created_agents.append(agent_invocation) + return agent_invocation + + telemetry.invoke_local_agent.side_effect = _create_agent + + handler.on_chain_start( + serialized={"name": "LangGraph"}, + inputs={}, + run_id=workflow_run_id, + parent_run_id=None, + ) + handler.on_chain_start( + serialized={"name": "planner"}, + inputs={}, + run_id=parent_agent_run_id, + parent_run_id=workflow_run_id, + metadata={"agent_name": "planner"}, + ) + handler.on_chain_start( + serialized={"name": "reviewer"}, + inputs={}, + run_id=child_agent_run_id, + parent_run_id=parent_agent_run_id, + metadata={"agent_name": "reviewer"}, + ) + + assert len(created_agents) == 2 + assert ( + created_agents[0].root_operation_name + == "invoke_workflow orchestrator" + ) + assert ( + created_agents[1].root_operation_name + == "invoke_workflow orchestrator" + ) + def test_no_agent_name_child_can_still_find_ancestor_agent(self): """Even when an intermediate node has no agent name, a deeper child must still be able to walk up and find a grandparent AgentInvocation.""" diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_conformance.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_conformance.py index d48fe98ef..25e2d01e5 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_conformance.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_conformance.py @@ -23,7 +23,7 @@ from .conformance.agent import AgentScenario from .conformance.inference import InferenceScenario from .conformance.tool_calling import ToolCallingScenario -from .conformance.workflow import WorkflowScenario +from .conformance.workflow import NestedWorkflowScenario, WorkflowScenario @pytest.mark.parametrize( @@ -33,6 +33,7 @@ AgentScenario(), ToolCallingScenario(), WorkflowScenario(), + NestedWorkflowScenario(), ], ids=lambda s: type(s).__name__, ) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py new file mode 100644 index 000000000..0eb402413 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py @@ -0,0 +1,130 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""End-to-end test: nested LangGraph workflows propagate root operation name. + +An outer graph contains an inner (sub)graph as one of its nodes. Both should be +captured as invoke_workflow spans. The inner workflow should include +`gen_ai.root_operation.name` that points at the outer workflow span name. +""" + +from typing import Annotated, TypedDict + +import pytest +from langchain_core.messages import HumanMessage, SystemMessage +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, StateGraph +from langgraph.graph.message import add_messages + +from opentelemetry.semconv._incubating.attributes import gen_ai_attributes + + +class GraphState(TypedDict): + messages: Annotated[list, add_messages] + research: str + + +def _build_nested_graph(llm: ChatOpenAI): + def researcher(state: GraphState) -> dict: + response = llm.invoke( + [ + SystemMessage( + content="You are a research assistant. Provide 2-3 factual sentences." + ), + HumanMessage(content=state["messages"][-1].content), + ] + ) + return {"research": response.content, "messages": [response]} + + inner_builder = StateGraph(GraphState) + inner_builder.add_node("researcher", researcher) + inner_builder.add_edge(START, "researcher") + inner_builder.add_edge("researcher", END) + inner_graph = inner_builder.compile() + + def summariser(state: GraphState) -> dict: + response = llm.invoke( + [ + SystemMessage( + content="You are an expert summariser. Condense the text below into one clear sentence." + ), + HumanMessage(content=state["research"]), + ] + ) + return {"messages": [response]} + + outer_builder = StateGraph(GraphState) + outer_builder.add_node("data_gathering", inner_graph) + outer_builder.add_node("summariser", summariser) + outer_builder.add_edge(START, "data_gathering") + outer_builder.add_edge("data_gathering", "summariser") + outer_builder.add_edge("summariser", END) + return outer_builder.compile() + + +# span_exporter, start_instrumentation and vcr are fixtures defined in conftest.py +@pytest.mark.vcr() +def test_nested_langgraph_workflow_sets_root_operation_name( + span_exporter, + start_instrumentation, + monkeypatch, + vcr, +): + monkeypatch.setenv( + "OTEL_SEMCONV_STABILITY_OPT_IN", "gen_ai_latest_experimental" + ) + + llm = ChatOpenAI( + model="gpt-3.5-turbo", + temperature=0.1, + max_tokens=200, + seed=42, + ) + graph = _build_nested_graph(llm) + + with vcr.use_cassette("nested_workflow_conformance.yaml"): + graph.invoke( + { + "messages": [ + HumanMessage(content="What is the capital of France?") + ], + "research": "", + } + ) + + spans = span_exporter.get_finished_spans() + + operation = gen_ai_attributes.GEN_AI_OPERATION_NAME + workflow_spans = [ + span + for span in spans + if span.attributes.get(operation) == "invoke_workflow" + ] + chat_spans = [ + span for span in spans if span.attributes.get(operation) == "chat" + ] + + assert len(workflow_spans) == 2, ( + "Expected one workflow span for the outer graph and one for the inner " + f"subgraph; saw {[s.name for s in spans]}" + ) + assert len(chat_spans) == 2, ( + "Expected two chat spans (researcher and summariser); " + f"saw {[s.name for s in spans]}" + ) + + root_operation_values = [ + span.attributes.get("gen_ai.root_operation.name") + for span in workflow_spans + if "gen_ai.root_operation.name" in span.attributes + ] + assert len(root_operation_values) == 1, ( + "Exactly one nested workflow span should carry " + "gen_ai.root_operation.name" + ) + outer_workflow = next( + span + for span in workflow_spans + if "gen_ai.root_operation.name" not in span.attributes + ) + assert root_operation_values[0] == outer_workflow.name 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 6be18dc1e..54100753c 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py @@ -132,16 +132,28 @@ def test_root_chain_with_no_signals_is_workflow(self): ) assert result == OperationName.INVOKE_WORKFLOW - def test_langgraph_name_with_parent_is_not_workflow(self): - # Having a parent disqualifies it from being a top-level workflow. + def test_langgraph_name_with_parent_is_nested_workflow(self): + # A LangGraph (sub)graph with a parent is a nested workflow. Whether it + # is actually nested is decided later from the run tree; classification + # only needs to recognise it as a workflow so a span is emitted. result = classify_chain_run( serialized={"name": "LangGraph"}, metadata=None, kwargs={}, parent_run_id=uuid.uuid4(), ) - # Not a workflow; no agent signals → suppressed - assert result is None + assert result == OperationName.INVOKE_WORKFLOW + + def test_langgraph_kwargs_name_with_parent_is_nested_workflow(self): + # At runtime LangGraph reports the graph identifier via kwargs["name"] + # rather than serialized["name"]. + result = classify_chain_run( + serialized={}, + metadata=None, + kwargs={"name": "LangGraph"}, + parent_run_id=uuid.uuid4(), + ) + assert result == OperationName.INVOKE_WORKFLOW # --- invoke_agent --- diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py index 34de7e514..5ec2ff3a9 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py @@ -24,6 +24,10 @@ ) from opentelemetry.util.types import AttributeValue +# ``gen_ai.root_operation.name`` is not yet in the released semconv package, +# so it is referenced here as a string literal. +_GEN_AI_ROOT_OPERATION_NAME = "gen_ai.root_operation.name" + class AgentInvocation(GenAIInvocation): """Represents a single agent invocation (invoke_agent span). @@ -49,6 +53,7 @@ def __init__( server_address: str | None = None, server_port: int | None = None, agent_name: str | None = None, + root_operation_name: str | None = None, ) -> None: """Use handler.invoke_local_agent() or handler.invoke_remote_agent() instead of calling this directly.""" _operation_name = GenAI.GenAiOperationNameValues.INVOKE_AGENT.value @@ -72,6 +77,7 @@ def __init__( self.agent_id: str | None = None self.agent_description: str | None = None self.agent_version: str | None = None + self.root_operation_name: str | None = root_operation_name self.conversation_id: str | None = None self.data_source_id: str | None = None @@ -106,6 +112,7 @@ def _get_base_attributes(self) -> dict[str, AttributeValue]: (GenAI.GEN_AI_PROVIDER_NAME, self.provider), (GenAI.GEN_AI_REQUEST_MODEL, self.request_model), (GenAI.GEN_AI_AGENT_NAME, self.agent_name), + (_GEN_AI_ROOT_OPERATION_NAME, self.root_operation_name), (server_attributes.SERVER_ADDRESS, self.server_address), (server_attributes.SERVER_PORT, self.server_port), ) @@ -124,6 +131,7 @@ def _get_common_attributes(self) -> dict[str, AttributeValue]: (GenAI.GEN_AI_AGENT_ID, self.agent_id), (GenAI.GEN_AI_AGENT_DESCRIPTION, self.agent_description), (GenAI.GEN_AI_AGENT_VERSION, self.agent_version), + (_GEN_AI_ROOT_OPERATION_NAME, self.root_operation_name), ) return { GenAI.GEN_AI_OPERATION_NAME: self._operation_name, @@ -181,6 +189,7 @@ def _get_metric_attributes(self) -> dict[str, AttributeValue]: (GenAI.GEN_AI_REQUEST_MODEL, self.request_model), (server_attributes.SERVER_ADDRESS, self.server_address), (server_attributes.SERVER_PORT, self.server_port), + (_GEN_AI_ROOT_OPERATION_NAME, self.root_operation_name), ) attrs: dict[str, AttributeValue] = { GenAI.GEN_AI_OPERATION_NAME: self._operation_name, 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 076d060dd..edc081496 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 @@ -23,6 +23,10 @@ ) from opentelemetry.util.types import AttributeValue +# ``gen_ai.root_operation.name`` is not yet in the released semconv package, +# so it is referenced here as a string literal. +_GEN_AI_ROOT_OPERATION_NAME = "gen_ai.root_operation.name" + class WorkflowInvocation(GenAIInvocation): """ @@ -40,6 +44,7 @@ def __init__( logger: Logger, completion_hook: CompletionHook, name: str | None, + root_operation_name: str | None = None, ) -> None: """Use handler.workflow(name) rather than calling this directly.""" _operation_name = "invoke_workflow" @@ -53,6 +58,7 @@ def __init__( span_kind=SpanKind.INTERNAL, ) self.name = name + self.root_operation_name: str | None = root_operation_name self.input_messages: list[InputMessage] = [] self.output_messages: list[OutputMessage] = [] self._start(self._get_base_attributes()) @@ -62,6 +68,8 @@ def _get_base_attributes(self) -> dict[str, AttributeValue]: attrs: dict[str, AttributeValue] = { GenAI.GEN_AI_OPERATION_NAME: self._operation_name, } + if self.root_operation_name: + attrs[_GEN_AI_ROOT_OPERATION_NAME] = self.root_operation_name return attrs def _get_messages_for_span(self) -> dict[str, AttributeValue]: @@ -89,6 +97,12 @@ def _apply_finish(self, error: Error | None = None) -> None: attributes: dict[str, AttributeValue] = { GenAI.GEN_AI_OPERATION_NAME: self._operation_name, } + if self.name: + attributes[GenAI.GEN_AI_WORKFLOW_NAME] = self.name + if self.root_operation_name: + attributes[_GEN_AI_ROOT_OPERATION_NAME] = ( + self.root_operation_name + ) attributes.update(self._get_messages_for_span()) if error is not None: self._apply_error_attributes(error) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/handler.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/handler.py index e28d0e8dd..1ed830b0c 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/handler.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/handler.py @@ -257,6 +257,7 @@ def start_workflow( self, *, name: str | None = None, + root_operation_name: str | None = None, ) -> WorkflowInvocation: """Create and start a workflow invocation. @@ -272,6 +273,7 @@ def start_workflow( self._logger, self._completion_hook, name, + root_operation_name=root_operation_name, ) def stop_llm(self, invocation: LLMInvocation) -> LLMInvocation: # pylint: disable=no-self-use @@ -392,6 +394,7 @@ def start_invoke_local_agent( *, request_model: str | None = None, agent_name: str | None = None, + root_operation_name: str | None = None, ) -> AgentInvocation: """Create and start a local agent invocation (INTERNAL span kind). @@ -411,6 +414,7 @@ def start_invoke_local_agent( span_kind=SpanKind.INTERNAL, request_model=request_model, agent_name=agent_name, + root_operation_name=root_operation_name, ) def start_invoke_remote_agent( @@ -421,6 +425,7 @@ def start_invoke_remote_agent( server_address: str | None = None, server_port: int | None = None, agent_name: str | None = None, + root_operation_name: str | None = None, ) -> AgentInvocation: """Create and start a remote agent invocation (CLIENT span kind). @@ -443,6 +448,7 @@ def start_invoke_remote_agent( agent_name=agent_name, server_address=server_address, server_port=server_port, + root_operation_name=root_operation_name, ) def invoke_local_agent( @@ -450,6 +456,7 @@ def invoke_local_agent( *, request_model: str | None = None, agent_name: str | None = None, + root_operation_name: str | None = None, ) -> AgentInvocation: """Returns an agent invocation (INTERNAL span kind). Starts span when called. @@ -469,6 +476,7 @@ def invoke_local_agent( span_kind=SpanKind.INTERNAL, request_model=request_model, agent_name=agent_name, + root_operation_name=root_operation_name, ) def invoke_remote_agent( @@ -479,6 +487,7 @@ def invoke_remote_agent( server_address: str | None = None, server_port: int | None = None, agent_name: str | None = None, + root_operation_name: str | None = None, ) -> AgentInvocation: """Returns an agent invocation (CLIENT span kind). Starts span when called. @@ -501,11 +510,14 @@ def invoke_remote_agent( agent_name=agent_name, server_address=server_address, server_port=server_port, + root_operation_name=root_operation_name, ) def workflow( self, name: str | None = None, + *, + root_operation_name: str | None = None, ) -> WorkflowInvocation: """Returns a Workflow invocation. Starts a span when called. @@ -521,6 +533,7 @@ def workflow( self._logger, self._completion_hook, name, + root_operation_name=root_operation_name, ) diff --git a/util/opentelemetry-util-genai/tests/test_handler_agent.py b/util/opentelemetry-util-genai/tests/test_handler_agent.py index 338786d1e..1dc88be1c 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_agent.py +++ b/util/opentelemetry-util-genai/tests/test_handler_agent.py @@ -324,6 +324,42 @@ def get_description(self): assert GenAI.GEN_AI_AGENT_NAME not in captured_attributes + def test_root_operation_name_at_construction_available_to_sampler(self): + captured_attributes = {} + + class AttributeCapturingSampler: # pylint: disable=no-self-use + def should_sample( + self, + parent_context, + trace_id, + name, + kind=None, + attributes=None, + links=None, + ): + captured_attributes.update(attributes or {}) + return SamplingResult(Decision.RECORD_AND_SAMPLE, attributes) + + def get_description(self): + return "AttributeCapturingSampler" + + sampler_provider = TracerProvider(sampler=AttributeCapturingSampler()) + sampler_provider.add_span_processor( + SimpleSpanProcessor(self.span_exporter) + ) + handler = TelemetryHandler(tracer_provider=sampler_provider) + + invocation = handler.invoke_local_agent( + agent_name="Sampler Agent", + root_operation_name="invoke_workflow orchestrator", + ) + invocation.stop() + + assert ( + captured_attributes["gen_ai.root_operation.name"] + == "invoke_workflow orchestrator" + ) + class TestAgentInvocationContent(unittest.TestCase): def setUp(self): diff --git a/util/opentelemetry-util-genai/tests/test_handler_workflow.py b/util/opentelemetry-util-genai/tests/test_handler_workflow.py index a304f5b82..769df871c 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_workflow.py +++ b/util/opentelemetry-util-genai/tests/test_handler_workflow.py @@ -221,6 +221,44 @@ def get_description(self): spans = self._get_finished_spans() self.assertEqual(len(spans), 1) + def test_start_workflow_root_operation_name_available_to_sampler( + self, + ) -> None: + captured_attributes = {} + + class AttributeCapturingSampler: # pylint: disable=no-self-use + def should_sample( + self, + parent_context, + trace_id, + name, + kind=None, + attributes=None, + links=None, + ): + captured_attributes.update(attributes or {}) + return SamplingResult(Decision.RECORD_AND_SAMPLE, attributes) + + def get_description(self): + return "AttributeCapturingSampler" + + sampler_provider = TracerProvider(sampler=AttributeCapturingSampler()) + sampler_provider.add_span_processor( + SimpleSpanProcessor(self.span_exporter) + ) + handler = TelemetryHandler(tracer_provider=sampler_provider) + + invocation = handler.workflow( + name="my-workflow", + root_operation_name="invoke_workflow root-workflow", + ) + invocation.stop() + + self.assertEqual( + captured_attributes["gen_ai.root_operation.name"], + "invoke_workflow root-workflow", + ) + def test_workflow_context_manager_sets_attributes_on_span(self) -> None: with self.handler.workflow("wf") as inv: inv.attributes["my.attr"] = "hello"