Skip to content

fix(langchain): trace nested workflows - #617

Open
1fanwang wants to merge 17 commits into
open-telemetry:mainfrom
1fanwang:1fannnw/langchain-nested-workflow-578
Open

1fanwang wants to merge 17 commits into
open-telemetry:mainfrom
1fanwang:1fannnw/langchain-nested-workflow-578

Conversation

@1fanwang

@1fanwang 1fanwang commented Sep 4, 2026 •

Copy link
Copy Markdown
Contributor

Description

Named LangGraph subgraphs disappear from traces, so their duration and outcome cannot be inspected separately. They now emit workflow spans while agent graphs keep one agent span and ordinary nested chains remain suppressed.

Graph entry points identify compiled graph roots, so metadata inherited from an enclosing graph does not create extra agent spans. A node that opts out of agent spans no longer hides a nested workflow, since that opt-out only says the node is not an agent. Opting out of tracing still applies.

Fixes #578

Type of change

  • Bug fix (non-breaking change which fixes an issue)
  • New feature (non-breaking change which adds functionality) (not applicable)
  • Breaking change (fix or feature that would cause existing functionality to not work as expected) (not applicable)
  • This change requires a documentation update

How has this been tested?

The tests do not assert async parent links. #513 tracks why, and those assertions stay xfailed.

E2E

The reproducer uses real compiled graphs and the OpenTelemetry exporter with local message nodes. It requires no model service or credentials. The historical Before runtime was Python 3.12.9, langchain-core 1.6.9 and langgraph 1.2.14. The current After runs use Python 3.10.21/core 0.3.49/graph 0.3.18 and Python 3.14.7/core 1.7.0/graph 1.2.16. Its third case sets otel_agent_span to False on the node that runs the nested graph, and the nested graph inherits it.

  • Test A: The compiled-graph reproducer shows the missing nested span on upstream main and the emitted spans on this branch.
# Scenario Command Result
1 Before fix Reproducer below The nested workflow is missing; exit code 1.
2 This branch Same reproducer The nested workflow is present; exit code 0.

Save the sources below beside the repository and run from its root with the indicated environments prepared. The historical Before command used its original py312 environment.

Before, at 7a576ee:

git checkout --detach 7a576ee88621f939ab079044923a44d286a794b0
.tox/py312-test-instrumentation-genai-langchain-latest/bin/python ../nested_workflow_repro.py; echo "exit=$?"
sync workflow: [('invoke_workflow outer', None)] MISSING
sync agent: [('invoke_agent my_agent', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
sync opt_out: [('invoke_workflow outer', None)] MISSING
async workflow: [('invoke_workflow outer', None)] MISSING
async agent: [('invoke_agent my_agent', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
async opt_out: [('invoke_workflow outer', None)] MISSING
exit=1

After, at bf40a11:

.tox/py310-test-instrumentation-genai-langchain-oldest/bin/python ../nested_workflow_repro.py
sync workflow: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
sync agent: [('invoke_agent my_agent', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
sync opt_out: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
async workflow: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
async agent: [('invoke_agent my_agent', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
async opt_out: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
.tox/py314-test-instrumentation-genai-langchain-latest/bin/python ../nested_workflow_repro.py
sync workflow: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
sync agent: [('invoke_agent my_agent', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
sync opt_out: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
async workflow: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
async agent: [('invoke_agent my_agent', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
async opt_out: [('invoke_workflow named_subgraph', 'invoke_workflow outer'), ('invoke_workflow outer', None)] ok
Reproducer source: nested_workflow_repro.py
import asyncio
from typing import Any, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langgraph.graph import END, START, StateGraph
from opentelemetry.instrumentation.genai.langchain import LangChainInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.test_util_genai.instrumentor import instrument


class State(TypedDict):
    messages: list[BaseMessage]


def respond(_: State) -> State:
    return {"messages": [AIMessage(content="done")]}


def graph(node: Any, name: str, metadata: dict[str, Any] | None = None) -> Any:
    builder = StateGraph(State)
    builder.add_node(node="step", action=node, metadata=metadata)
    builder.add_edge(start_key=START, end_key="step")
    builder.add_edge(start_key="step", end_key=END)
    return builder.compile(name=name)


failed = False
for async_mode in (False, True):
    for case in ("workflow", "agent", "opt_out"):
        exporter = InMemorySpanExporter()
        tracer = TracerProvider()
        tracer.add_span_processor(SimpleSpanProcessor(exporter))
        child = graph(node=respond, name="named_subgraph")
        expected = "invoke_workflow named_subgraph"
        if case == "agent":
            child = child.with_config(metadata={"agent_name": "my_agent"})
            expected = "invoke_agent my_agent"
        # The node that runs the nested graph opts out of agent spans; the
        # nested graph inherits that metadata but never set it itself.
        node_metadata = {"otel_agent_span": False} if case == "opt_out" else None
        outer = graph(node=child, name="outer", metadata=node_metadata)
        with instrument(
            LangChainInstrumentor(), tracer_provider=tracer,
            meter_provider=None, logger_provider=None,
            content_capture="SPAN_ONLY",
        ):
            inputs: State = {"messages": [HumanMessage(content="hello")]}
            if async_mode:
                asyncio.run(outer.ainvoke(input=inputs))
            else:
                outer.invoke(input=inputs)
        spans = exporter.get_finished_spans()
        names = {span.context.span_id: span.name for span in spans}
        seen = [
            (span.name, names[span.parent.span_id] if span.parent else None)
            for span in spans
        ]
        ok = sorted(name for name, _ in seen) == sorted(
            [expected, "invoke_workflow outer"]
        )
        failed = failed or not ok
        print(
            f"{'async' if async_mode else 'sync'} {case}: {seen} "
            f"{'ok' if ok else 'MISSING'}",
            flush=True,
        )
raise SystemExit(1 if failed else 0)

The compatibility observation runs local compiled graphs and prints every emitted span with its conversation ID. Bound IDs fall back to graph metadata, runtime IDs take precedence, bound trace opt-outs hide the child graph, and internal writers emit no agent spans. Both runs use the same current commit above.

.tox/py310-test-instrumentation-genai-langchain-oldest/bin/python ../compatibility_observation.py
sync bound-id: [('invoke_workflow child', 'bound'), ('invoke_workflow outer', None)]
sync runtime-id: [('invoke_workflow child', 'runtime'), ('invoke_workflow outer', None)]
sync runtime-session: [('invoke_workflow child', 'session'), ('invoke_workflow outer', None)]
sync bound-trace: [('invoke_workflow outer', None)]
sync hidden-writer: [('invoke_agent', None)]
async bound-id: [('invoke_workflow child', 'bound'), ('invoke_workflow outer', None)]
async runtime-id: [('invoke_workflow child', 'runtime'), ('invoke_workflow outer', None)]
async runtime-session: [('invoke_workflow child', 'session'), ('invoke_workflow outer', None)]
async bound-trace: [('invoke_workflow outer', None)]
async hidden-writer: [('invoke_agent', None)]
.tox/py314-test-instrumentation-genai-langchain-latest/bin/python ../compatibility_observation.py
sync bound-id: [('invoke_workflow child', 'bound'), ('invoke_workflow outer', None)]
sync runtime-id: [('invoke_workflow child', 'runtime'), ('invoke_workflow outer', None)]
sync runtime-session: [('invoke_workflow child', 'session'), ('invoke_workflow outer', None)]
sync bound-trace: [('invoke_workflow outer', None)]
sync hidden-writer: [('invoke_agent', None)]
async bound-id: [('invoke_workflow child', 'bound'), ('invoke_workflow outer', None)]
async runtime-id: [('invoke_workflow child', 'runtime'), ('invoke_workflow outer', None)]
async runtime-session: [('invoke_workflow child', 'session'), ('invoke_workflow outer', None)]
async bound-trace: [('invoke_workflow outer', None)]
async hidden-writer: [('invoke_agent', None)]
Reproducer source: compatibility_observation.py
from __future__ import annotations

import asyncio
from typing import Any, TypedDict

from langchain_core.messages import AIMessage, BaseMessage, HumanMessage
from langchain_core.runnables import RunnableLambda
from langgraph.graph import END, START, StateGraph
from opentelemetry.instrumentation.genai.langchain import LangChainInstrumentor
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.semconv._incubating.attributes import gen_ai_attributes as GenAI
from opentelemetry.test_util_genai.instrumentor import instrument


class State(TypedDict):
    messages: list[BaseMessage]


def respond(_: State) -> State:
    return {"messages": [AIMessage(content="done")]}


def graph(node: Any, *, name: str) -> Any:
    builder = StateGraph(State)
    builder.add_node(node="step", action=node)
    builder.add_edge(start_key=START, end_key="step")
    builder.add_edge(start_key="step", end_key=END)
    return builder.compile(name=name)


async def observe(*, async_mode: bool, case: str) -> None:
    exporter = InMemorySpanExporter()
    tracer = TracerProvider()
    tracer.add_span_processor(SimpleSpanProcessor(span_exporter=exporter))
    expected_id: str | None = None
    if case in {"bound-id", "runtime-id", "runtime-session"}:
        child = graph(respond, name="child").with_config(
            metadata={"thread_id": "bound"}
        )
        runtime_metadata: dict[str, str | None] = (
            {"thread_id": "runtime"} if case == "runtime-id"
            else {"thread_id": None, "session_id": "session"}
            if case == "runtime-session" else {}
        )

        def run_child(state: State) -> State:
            return child.invoke(input=state, config={"metadata": runtime_metadata})

        async def arun_child(state: State) -> State:
            return await child.ainvoke(
                input=state, config={"metadata": runtime_metadata}
            )

        outer = graph(RunnableLambda(run_child, afunc=arun_child), name="outer")
        expected_id = {
            "bound-id": "bound", "runtime-id": "runtime",
            "runtime-session": "session",
        }[case]
        expected_names = ["invoke_workflow child", "invoke_workflow outer"]
    elif case == "bound-trace":
        child = graph(respond, name="child").with_config(
            metadata={"otel_trace": False}
        )
        outer = graph(child, name="outer")
        expected_names = ["invoke_workflow outer"]
    else:
        outer = graph(respond, name="agent")
        expected_names = ["invoke_agent"]
    with instrument(
        LangChainInstrumentor(), tracer_provider=tracer,
        meter_provider=None, logger_provider=None, content_capture="SPAN_ONLY",
    ):
        inputs: State = {"messages": [HumanMessage(content="hello")]}
        config = {"metadata": {"agent_type": "worker"}} if case == "hidden-writer" else None
        if async_mode:
            await outer.ainvoke(input=inputs, config=config)
        else:
            outer.invoke(input=inputs, config=config)
    spans = exporter.get_finished_spans()
    observed = sorted(
        (span.name, span.attributes.get(GenAI.GEN_AI_CONVERSATION_ID))
        for span in spans
    )
    print(f"{'async' if async_mode else 'sync'} {case}: {observed}", flush=True)
    assert sorted(span.name for span in spans) == expected_names
    if expected_id is not None:
        child_span = next(span for span in spans if span.name == "invoke_workflow child")
        assert child_span.attributes.get(GenAI.GEN_AI_CONVERSATION_ID) == expected_id


for async_mode in (False, True):
    for case in ("bound-id", "runtime-id", "runtime-session", "bound-trace", "hidden-writer"):
        asyncio.run(observe(async_mode=async_mode, case=case))

The local conformance run could not execute because Weaver is not installed. CI runs this target.

Checklist

See CONTRIBUTING.md
for the style guide, changelog guidance, and more.

  • Followed the style guidelines of this project
  • Changelog updated if the change requires an entry
  • Unit tests added
  • Documentation updated

Signed-off-by: 1fanwang <1fannnw@gmail.com>

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟢 Approval recommended

The changes are well-scoped to LangGraph nested workflow span classification and are backed by targeted regression tests (sync and async) covering the reported failure mode.

Pull request overview

This PR fixes LangChain/LangGraph workflow tracing so that named nested subgraphs emit their own invoke_workflow spans, instead of disappearing into the outer workflow span. It extends the “graph announcement” mechanism (introduced for nested agents) to all compiled graphs, while keeping agent graphs classified as invoke_agent and continuing to suppress ordinary nested RunnableSequence chains.

Changes:

  • Extend graph announcements to cover non-agent compiled graphs, enabling nested subgraph workflow spans.
  • Adjust chain-run classification precedence to honor explicit overrides and prefer announced nested workflows over inherited agent metadata.
  • Add regression tests validating that a named nested subgraph produces an additional workflow span (sync + async).
File summaries
File Description
instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_workflow.py New regression tests for nested LangGraph subgraph workflow spans and suppression behavior for non-graph nested chains.
instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_operation_mapping.py Adds unit tests covering workflow override behavior when nested and precedence rules with announcements vs metadata.
instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/operation_mapping.py Updates workflow/agent classification logic to allow nested workflow spans when a compiled-graph announcement is present.
instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/callback_handler.py Uses the generalized graph announcement (claim_graph) and threads announced_workflow into classification; supports workflow naming from announcements.
instrumentation/opentelemetry-instrumentation-genai-langchain/src/opentelemetry/instrumentation/genai/langchain/agent_context.py Generalizes pending context from “agent” to “graph” announcements and announces non-agent graphs during Pregel stream entry.
Review details
  • Files reviewed: 5/5 changed files
  • Comments generated: 0
  • Review effort level: Lite

💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.

@opentelemetry-pr-dashboard

opentelemetry-pr-dashboard Bot commented Sep 4, 2026 •

Copy link
Copy Markdown

Pull request dashboard status

Waiting on the author · refreshed 2026-10-26 20:01 UTC

Respond to 1 review item (e.g. link a commit, explain why not, ask a follow-up):

  • Inline threads: 1
Status above doesn't look right?
  • Just replied or pushed? Anything around or after the refresh time above may not be picked up yet — give it a few minutes.
  • Should this be with reviewers? Comment /dashboard route:reviewers to route it to them.
  • Anything wrong — including the routing? Report it with what you expected; it helps us improve the dashboard.

Signed-off-by: 1fanwang <1fannnw@gmail.com>
Assisted-by: GitHub Copilot CLI (GPT-5.6 Sol)
Signed-off-by: 1fanwang <1fannnw@gmail.com>
Keep graph-owned metadata while using the current local-agent invocation contract.

Signed-off-by: 1fanwang <1fannnw@gmail.com>
Signed-off-by: 1fanwang <1fannnw@gmail.com>
lmolkova and others added 4 commits September 23, 2026 18:37
Signed-off-by: 1fanwang <1fannnw@gmail.com>
Signed-off-by: 1fanwang <1fannnw@gmail.com>
Signed-off-by: 1fanwang <1fannnw@gmail.com>
@opentelemetry-pr-dashboard

This comment has been minimized.

@1fanwang
1fanwang requested review from lmolkova and a lite review from Copilot October 7, 2026 18:43

Copilot AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Changes recommended

Moderate issues affect stream API compatibility and can suppress announced workflow spans under inherited metadata.

1 open finding

🧠 Review effort: Lite

agent_announcement is not None,
ancestor_agent_names,
serialized=serialized,
metadata=metadata,
The only conflict was in on_tool_start in callback_handler.py. Main replaced
the manual tool argument parsing with tool_arguments_to_any_value, and this
branch changed _find_agent_context to return three values. The merged code
keeps both changes.

Assisted-by: GitHub Copilot CLI (claude-sonnet-5.5)
Signed-off-by: 1fanwang <1fannnw@gmail.com>
An enclosing task that sets otel_agent_span to False passes it to every
nested callback, so a nested graph announced as a workflow was dropped
although its own metadata did not set it. Announced workflows now read
that opt-out from their own graph metadata. The otel_trace opt-out and
announced agents are unchanged.

Assisted-by: GitHub Copilot CLI (claude-sonnet-5.5)
Signed-off-by: 1fanwang <1fannnw@gmail.com>

# 3. Workflow / orchestration detection.
if _looks_like_workflow(serialized, metadata, parent_run_id):
if metadata and metadata.get(_META_WORKFLOW_SPAN):

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Callback metadata is inherited by every child run, so without the top-level guard every chain under a run marked otel_workflow_span becomes an invoke_workflow span. E.g. graph.invoke(x, config={"metadata": {"otel_workflow_span": True}}) now reports each node as a workflow too. Chains and nodes will get their own span type (open-telemetry/semantic-conventions-genai#188), which is not workflow.

Could the override apply only on the top-level run (no parent_run_id) as before? Nested graphs don't need it: every Pregel.stream/astream is announced, and classify_chain_run returns invoke_workflow for an announced graph before it looks at the override - test_nested_subgraph_emits_workflow_spans already gets both spans without it. The workflow-override case in test_graph_child_with_local_operation_metadata would go. Test that showcases the problem:

def test_inherited_workflow_override_marks_only_top_level_run(
    instrument_langchain: Any,
    span_exporter: InMemorySpanExporter,
) -> None:
    graph = _graph(_respond, name="outer_graph")

    with instrument_langchain():
        graph.invoke(
            {"messages": [HumanMessage(content="hello")]},
            config={"metadata": {"otel_workflow_span": True}},
        )

    assert [s.name for s in _workflow_spans(span_exporter)] == [
        "invoke_workflow outer_graph"
    ]

Comment on lines +158 to +164
or (
not has_parent_agent
and (
metadata.get(_META_AGENT_TYPE)
or metadata.get(_META_AGENT_SPAN)
)
)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This drops explicitly marked agents nested under an agent, not just inherited signals. A runnable with agent_type called from an agent graph used to get invoke_agent researcher; now it gets no span. Fails on this PR, passes on main:

def test_explicit_agent_nested_under_agent(
    instrument_langchain: Any,
    span_exporter: InMemorySpanExporter,
) -> None:
    researcher = RunnableLambda(_respond).with_config(
        run_name="researcher", metadata={"agent_type": "worker"}
    )
    supervisor = _graph(researcher, name="supervisor").with_config(
        {"metadata": {"agent_name": "supervisor"}}
    )

    with instrument_langchain():
        supervisor.invoke({"messages": [HumanMessage(content="hello")]})

    assert {s.name for s in _agent_spans(span_exporter)} == {
        "invoke_agent supervisor",
        "invoke_agent researcher",
    }

Comment on lines +34 to +35

_pending: ContextVar[tuple[_PendingAgent, ...]] = ContextVar(
"otel_genai_pending_agents", default=()
_PendingAgent = _PendingGraph

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: the alias only exists for test_conversation_id.py; consider switching that test to _PendingGraph and dropping it.

Suggested change
_pending: ContextVar[tuple[_PendingAgent, ...]] = ContextVar(
"otel_genai_pending_agents", default=()
_PendingAgent = _PendingGraph

_attach_to_context=self._attach_to_context,
)
agent.input_messages = make_input_message(inputs)
agent.conversation_id = conversation_id

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: conversation_id is already passed to invoke_local_agent above.

Suggested change
agent.conversation_id = conversation_id

Assisted-by: GitHub Copilot CLI (Auto)
Signed-off-by: 1fanwang <1fannnw@gmail.com>
Assisted-by: GitHub Copilot CLI (Auto)
Signed-off-by: 1fanwang <1fannnw@gmail.com>
@opentelemetry-pr-dashboard

Copy link
Copy Markdown

Hi @1fanwang — just a friendly reminder that this pull request is waiting on you. The dashboard status comment has the open items and is kept current.

  • Replying is enough to hand it off — answer, explain why no change is needed, or ask a follow-up. The dashboard routes it onward once nothing on the list is waiting on you.
  • To hand it back for any other reason, including the dashboard getting this wrong, comment /dashboard route:reviewers.

Assisted-by: GitHub Copilot CLI (Auto)
Signed-off-by: 1fanwang <1fannnw@gmail.com>
Assisted-by: GitHub Copilot CLI (Auto)
Signed-off-by: 1fanwang <1fannnw@gmail.com>
Assisted-by: GitHub Copilot CLI (Auto)
Signed-off-by: 1fanwang <1fannnw@gmail.com>

This branch has not been deployed

No deployments
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Development

Successfully merging this pull request may close these issues.

[langchain] Add nested workflow span on on_chain_start

3 participants