From f63d96806f4c6e0ddf2120ca4bf7b7dc439b9860 Mon Sep 17 00:00:00 2001 From: Liudmila Molkova Date: Wed, 1 Jul 2026 05:29:52 +0000 Subject: [PATCH 1/3] Prototype: add nested workflow detection in langchain --- .../genai/langchain/callback_handler.py | 16 ++ .../genai/langchain/operation_mapping.py | 66 +++++-- .../nested_workflow_conformance.yaml | 180 ++++++++++++++++++ .../tests/conformance/workflow.py | 118 ++++++++++++ .../tests/test_conformance.py | 3 +- .../tests/test_nested_workflow.py | 134 +++++++++++++ .../tests/test_operation_mapping.py | 20 +- .../util/genai/_workflow_invocation.py | 10 + 8 files changed, 521 insertions(+), 26 deletions(-) create mode 100644 instrumentation/opentelemetry-instrumentation-genai-langchain/tests/cassettes/nested_workflow_conformance.yaml create mode 100644 instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py 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..386cf8076 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 @@ -74,6 +74,9 @@ def on_chain_start( name=workflow_name_override or workflow_name ) workflow.input_messages = make_input_message(inputs) + # A workflow whose ancestor is another workflow is nested within it. + if self._find_nearest_workflow(parent_run_id) is not None: + workflow.nested = True self._invocation_manager.add_invocation_state( run_id, parent_run_id, workflow ) @@ -500,3 +503,16 @@ def _find_nearest_agent( return entity current = self._invocation_manager.get_parent_run_id(current) return None + + def _find_nearest_workflow( + self, run_id: Optional[UUID] + ) -> Optional[WorkflowInvocation]: + 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, WorkflowInvocation): + return entity + current = self._invocation_manager.get_parent_run_id(current) + return 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 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..b3ca8d255 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py @@ -137,3 +137,121 @@ 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 is the top-level workflow (no ``gen_ai.workflow.nested``); + the inner subgraph is nested and carries ``gen_ai.workflow.nested = True``. + """ + + expected_spans = ("invoke_workflow", "invoke_workflow", "chat", "chat") + expected_metrics = ( + "gen_ai.client.operation.duration", + "gen_ai.client.token.usage", + ) + # langchain can't populate server.address on chat spans. + expected_violations = ( + ExpectedViolation( + advice_id="genai_expected_attribute_missing", + message_substring="server.address", + ), + ) + + def run( + self, + *, + tracer_provider: TracerProvider, + meter_provider: MeterProvider, + logger_provider: LoggerProvider, + vcr: Any, + ) -> None: + 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) + nested_values = [ + attr["value"] + for entry in report["samples"] + if "span" in entry + for attr in entry["span"]["attributes"] + if attr["name"] == "gen_ai.workflow.nested" + ] + assert nested_values == [True], ( + "Exactly one workflow span (the inner subgraph) must set " + f"gen_ai.workflow.nested=True; saw {nested_values}" + ) + 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..5b93a31f7 --- /dev/null +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py @@ -0,0 +1,134 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +"""End-to-end test: a nested LangGraph workflow emits ``gen_ai.workflow.nested``. + +An outer graph contains an inner (sub)graph as one of its nodes. The outer graph +is the top-level workflow and must NOT carry ``gen_ai.workflow.nested``; the +inner subgraph is nested and must carry ``gen_ai.workflow.nested = True``. +""" + +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 + +_GEN_AI_WORKFLOW_NESTED = "gen_ai.workflow.nested" + + +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_nested_attribute( + 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]}" + ) + + nested_spans = [ + span + for span in workflow_spans + if span.attributes.get(_GEN_AI_WORKFLOW_NESTED) is True + ] + top_level_spans = [ + span + for span in workflow_spans + if _GEN_AI_WORKFLOW_NESTED not in span.attributes + ] + + assert len(nested_spans) == 1, ( + "Exactly one workflow (the inner subgraph) must be marked nested" + ) + assert len(top_level_spans) == 1, ( + "The top-level workflow must not carry gen_ai.workflow.nested" + ) 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/_workflow_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py index 076d060dd..650eb0543 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.workflow.nested`` is not yet in the released semconv package, so it +# is referenced here as a string literal. +_GEN_AI_WORKFLOW_NESTED = "gen_ai.workflow.nested" + class WorkflowInvocation(GenAIInvocation): """ @@ -53,6 +57,8 @@ def __init__( span_kind=SpanKind.INTERNAL, ) self.name = name + self.nested: bool = False + """Whether this workflow is invoked within an enclosing workflow.""" self.input_messages: list[InputMessage] = [] self.output_messages: list[OutputMessage] = [] self._start(self._get_base_attributes()) @@ -89,6 +95,10 @@ 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.nested: + attributes[_GEN_AI_WORKFLOW_NESTED] = True attributes.update(self._get_messages_for_span()) if error is not None: self._apply_error_attributes(error) From b8f502b401d5d43936c26ad4aa2107f8d00b7d2e Mon Sep 17 00:00:00 2001 From: Liudmila Molkova Date: Wed, 1 Jul 2026 05:33:10 +0000 Subject: [PATCH 2/3] up --- .../tests/conformance/workflow.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py index b3ca8d255..3f5ab0e0d 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py @@ -151,12 +151,18 @@ class NestedWorkflowScenario(Scenario): "gen_ai.client.operation.duration", "gen_ai.client.token.usage", ) - # langchain can't populate server.address on chat spans. expected_violations = ( + # langchain can't populate server.address on chat spans. ExpectedViolation( advice_id="genai_expected_attribute_missing", message_substring="server.address", ), + # gen_ai.workflow.nested is a proposed attribute that is not yet part + # of the published semantic-conventions registry. + ExpectedViolation( + advice_id="missing_attribute", + message_substring="gen_ai.workflow.nested", + ), ) def run( From aaa0c287c0882b1d3f18d41e77188a4f14051a30 Mon Sep 17 00:00:00 2001 From: Liudmila Molkova Date: Sat, 11 Jul 2026 12:51:04 -0700 Subject: [PATCH 3/3] update to gen_ai.root_operation.name --- .../genai/langchain/callback_handler.py | 45 ++++- .../tests/conformance/workflow.py | 34 ++-- .../tests/test_callback_handler.py | 187 ++++++++++++++++++ .../tests/test_nested_workflow.py | 34 ++-- .../util/genai/_agent_invocation.py | 9 + .../util/genai/_workflow_invocation.py | 18 +- .../src/opentelemetry/util/genai/handler.py | 13 ++ .../tests/test_handler_agent.py | 36 ++++ .../tests/test_handler_workflow.py | 38 ++++ 9 files changed, 364 insertions(+), 50 deletions(-) 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 386cf8076..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,13 +70,14 @@ 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) - # A workflow whose ancestor is another workflow is nested within it. - if self._find_nearest_workflow(parent_run_id) is not None: - workflow.nested = True self._invocation_manager.add_invocation_state( run_id, parent_run_id, workflow ) @@ -98,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) @@ -504,15 +511,35 @@ def _find_nearest_agent( current = self._invocation_manager.get_parent_run_id(current) return None - def _find_nearest_workflow( + def _find_enclosing_root_operation_name( self, run_id: Optional[UUID] - ) -> Optional[WorkflowInvocation]: + ) -> 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, WorkflowInvocation): - return entity - current = self._invocation_manager.get_parent_run_id(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/tests/conformance/workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py index 3f5ab0e0d..5cf04d325 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/conformance/workflow.py @@ -142,8 +142,9 @@ def validate(self, report: LiveCheckReport) -> None: class NestedWorkflowScenario(Scenario): """A LangGraph graph whose node is itself a compiled subgraph. - The outer graph is the top-level workflow (no ``gen_ai.workflow.nested``); - the inner subgraph is nested and carries ``gen_ai.workflow.nested = True``. + 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") @@ -157,12 +158,6 @@ class NestedWorkflowScenario(Scenario): advice_id="genai_expected_attribute_missing", message_substring="server.address", ), - # gen_ai.workflow.nested is a proposed attribute that is not yet part - # of the published semantic-conventions registry. - ExpectedViolation( - advice_id="missing_attribute", - message_substring="gen_ai.workflow.nested", - ), ) def run( @@ -249,15 +244,24 @@ def summariser(state: GraphState) -> dict: def validate(self, report: LiveCheckReport) -> None: super().validate(report) - nested_values = [ - attr["value"] + workflow_spans = [ + entry["span"] for entry in report["samples"] if "span" in entry - for attr in entry["span"]["attributes"] - if attr["name"] == "gen_ai.workflow.nested" + 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 nested_values == [True], ( - "Exactly one workflow span (the inner subgraph) must set " - f"gen_ai.workflow.nested=True; saw {nested_values}" + 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_nested_workflow.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py index 5b93a31f7..0eb402413 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_nested_workflow.py @@ -1,11 +1,11 @@ # Copyright The OpenTelemetry Authors # SPDX-License-Identifier: Apache-2.0 -"""End-to-end test: a nested LangGraph workflow emits ``gen_ai.workflow.nested``. +"""End-to-end test: nested LangGraph workflows propagate root operation name. -An outer graph contains an inner (sub)graph as one of its nodes. The outer graph -is the top-level workflow and must NOT carry ``gen_ai.workflow.nested``; the -inner subgraph is nested and must carry ``gen_ai.workflow.nested = True``. +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 @@ -18,8 +18,6 @@ from opentelemetry.semconv._incubating.attributes import gen_ai_attributes -_GEN_AI_WORKFLOW_NESTED = "gen_ai.workflow.nested" - class GraphState(TypedDict): messages: Annotated[list, add_messages] @@ -66,7 +64,7 @@ def summariser(state: GraphState) -> dict: # span_exporter, start_instrumentation and vcr are fixtures defined in conftest.py @pytest.mark.vcr() -def test_nested_langgraph_workflow_sets_nested_attribute( +def test_nested_langgraph_workflow_sets_root_operation_name( span_exporter, start_instrumentation, monkeypatch, @@ -115,20 +113,18 @@ def test_nested_langgraph_workflow_sets_nested_attribute( f"saw {[s.name for s in spans]}" ) - nested_spans = [ - span + root_operation_values = [ + span.attributes.get("gen_ai.root_operation.name") for span in workflow_spans - if span.attributes.get(_GEN_AI_WORKFLOW_NESTED) is True + if "gen_ai.root_operation.name" in span.attributes ] - top_level_spans = [ + 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_WORKFLOW_NESTED not in span.attributes - ] - - assert len(nested_spans) == 1, ( - "Exactly one workflow (the inner subgraph) must be marked nested" - ) - assert len(top_level_spans) == 1, ( - "The top-level workflow must not carry gen_ai.workflow.nested" + if "gen_ai.root_operation.name" not in span.attributes ) + assert root_operation_values[0] == outer_workflow.name 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 650eb0543..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,9 +23,9 @@ ) from opentelemetry.util.types import AttributeValue -# ``gen_ai.workflow.nested`` is not yet in the released semconv package, so it -# is referenced here as a string literal. -_GEN_AI_WORKFLOW_NESTED = "gen_ai.workflow.nested" +# ``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): @@ -44,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" @@ -57,8 +58,7 @@ def __init__( span_kind=SpanKind.INTERNAL, ) self.name = name - self.nested: bool = False - """Whether this workflow is invoked within an enclosing workflow.""" + 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()) @@ -68,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]: @@ -97,8 +99,10 @@ def _apply_finish(self, error: Error | None = None) -> None: } if self.name: attributes[GenAI.GEN_AI_WORKFLOW_NAME] = self.name - if self.nested: - attributes[_GEN_AI_WORKFLOW_NESTED] = True + 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"