Issue in Stream Execution When Using Command within a Subgraph

[yamato0811]

Checked other resources

• This is a bug, not a usage question. For questions, please use GitHub Discussions.
• I added a clear and detailed title that summarizes the issue.
• I read what a minimal reproducible example is (https://stackoverflow.com/help/minimal-reproducible-example).
• I included a self-contained, minimal example that demonstrates the issue INCLUDING all the relevant imports. The code run AS IS to reproduce the issue.

Example Code

from typing import TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command
from typing_extensions import Literal


class State(TypedDict):
    node_name: str
    foo: str


def subgraph_node_1(state: State) -> Command[Literal["subgraph_node_2"]]:
    return Command(
        goto="subgraph_node_2",
        update={
            "node_name": "subgraph_node_1",
            "foo": "Update at subgraph_node_1!",
        },
    )


def subgraph_node_2(state: State) -> Command:
    return Command(
        goto="node_3",
        update={"node_name": "subgraph_node_2"},
        graph=Command.PARENT,
    )


subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_2", END)
subgraph = subgraph_builder.compile()


# Define main graph
def node_1(state: State) -> Command[Literal["node_2"]]:
    return Command(
        goto="node_2",
        update={"node_name": "node_1"},
    )


def node_3(state: State) -> Command[Literal["__end__"]]:
    return Command(
        goto=END,
        update={"node_name": "node_3"},
    )


main_builder = StateGraph(State)
main_builder.add_node("node_1", node_1)
main_builder.add_node("node_2", subgraph)
main_builder.add_node("node_3", node_3)
main_builder.add_edge(START, "node_1")
main_builder.add_edge("node_2", "node_3")
main_graph = main_builder.compile()


# Build subgraph
with open("graph.md", "w") as file:
    file.write(f"\n{main_graph.get_graph(xray=1).draw_mermaid()}")

initial = {"node_name": ["__start__"]}
for chunk in main_graph.stream(initial, stream_mode="values", subgraphs=True):
    print(chunk)

Error Message and Stack Trace (if applicable)

Description

While using the LangGraph library and attempting stream execution with a Command within a subgraph, I encountered the following two issues:

1. Part of the subgraph's state is not passed to the parent graph.
   • 'foo': 'Update at subgraph_node_1!' is not displayed in the parent graph.
2. The output of the subgraph's last node is not displayed.
   • The execution result at subgraph_node_2 is not shown.

【Output】

((), {'node_name': ['__start__']})
((), {'node_name': 'node_1'})
(('node_2:d41305f4-c2ba-78d9-5d9b-ed1c6b8549da',), {'node_name': 'node_1'})
(('node_2:d41305f4-c2ba-78d9-5d9b-ed1c6b8549da',), {'node_name': 'subgraph_node_1', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'subgraph_node_2'})
((), {'node_name': 'node_3'})

---

When not using Command, the result was as expected.
【Expected Output】

((), {'node_name': ['__start__']})
((), {'node_name': 'node_1'})
(('node_2:cd87a0ec-b602-da30-ffca-48950974937f',), {'node_name': 'node_1'})
(('node_2:cd87a0ec-b602-da30-ffca-48950974937f',), {'node_name': 'subgraph_node_1', 'foo': 'Update at subgraph_node_1!'})
(('node_2:cd87a0ec-b602-da30-ffca-48950974937f',), {'node_name': 'subgraph_node_2', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'subgraph_node_2', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'node_3', 'foo': 'Update at subgraph_node_1!'})

【Expected Output Code】

from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class State(TypedDict):
    node_name: str
    foo: str


def subgraph_node_1(state: State):
    return {
        "node_name": "subgraph_node_1",
        "foo": "Update at subgraph_node_1!",
    }


def subgraph_node_2(state: State):
    return {"node_name": "subgraph_node_2"}


subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph_builder.add_edge("subgraph_node_2", END)
subgraph = subgraph_builder.compile()


# Define main graph
def node_1(state: State):
    return {"node_name": "node_1"}


def node_3(state: State):
    return {"node_name": "node_3"}


main_builder = StateGraph(State)
main_builder.add_node("node_1", node_1)
main_builder.add_node("node_2", subgraph)
main_builder.add_node("node_3", node_3)
main_builder.add_edge(START, "node_1")
main_builder.add_edge("node_1", "node_2")
main_builder.add_edge("node_2", "node_3")
main_builder.add_edge("node_3", END)
main_graph = main_builder.compile()


# Build subgraph
with open("graph.md", "w") as file:
    file.write(f"```mermaid\n{main_graph.get_graph(xray=1).draw_mermaid()}```")

initial = {"node_name": ["__start__"]}
for chunk in main_graph.stream(initial, stream_mode="values", subgraphs=True):
    print(chunk)

System Info

System Information

OS: Linux
OS Version: #54~20.04.1-Ubuntu SMP Fri Oct 6 22:04:33 UTC 2023
Python Version: 3.11.9 | packaged by conda-forge | (main, Apr 19 2024, 18:36:13) [GCC 12.3.0]

Package Information

langchain_core: 0.3.31
langchain: 0.3.14
langsmith: 0.2.3
langchain_aws: 0.2.3
langchain_openai: 0.2.4
langchain_text_splitters: 0.3.5
langgraph_sdk: 0.1.43

Optional packages not installed

langserve

Other Dependencies

aiohttp: 3.11.10
async-timeout: Installed. No version info available.
boto3: 1.35.37
httpx: 0.28.1
jsonpatch: 1.33
langsmith-pyo3: Installed. No version info available.
numpy: 1.26.4
openai: 1.53.0
orjson: 3.10.12
packaging: 23.2
pydantic: 2.10.3
PyYAML: 6.0.1
requests: 2.32.3
requests-toolbelt: 1.0.0
SQLAlchemy: 2.0.36
tenacity: 9.0.0
tiktoken: 0.8.0
typing-extensions: 4.12.2

[vbarda]

@yamato0811 thanks for reporting -- we'll look into this. in the meantime, as you pointed out in the second code snippet, you can just do return {"node_name": "subgraph_node_2"} instead of using Command

[vbarda]

@yamato0811 you should be able to do this now in LangGraph v0.2.69. The only thing to keep in mind is that you would now need to add a reducer to the node_name field. Let me know if you run into any issues.

[ren8k]

@vbarda Thank you for fixing the bug!

The only thing to keep in mind is that you would now need to add a reducer to the node_name field.

Does this mean that a reducer is required for state elements that are shared between ParentGraph and SubGraph, specifically for elements that are updated at the end of the SubGraph?

Considering the code example above, does this mean we need to create a custom reducer and add it to the state element node_name as shown below? Is a custom reducer necessary even just for value replacement?

Also, while I confirmed that the subgraph's state is passed to the parent graph in the code below, I've noticed that the output from the subgraph's final node is still not being displayed. I've reported this issue separately in another Issue.

from typing import Annotated, TypedDict

from typing_extensions import Literal

from langgraph.graph import END, StateGraph
from langgraph.types import Command


def my_reducer(a: str, b: str | None) -> int:
    if b is not None:
        return b
    return a


class State(TypedDict):
    # node_name: Annotated[list[str], add]
    node_name: Annotated[str, my_reducer]
    foo: str


def subgraph_node_1(state: State) -> Command[Literal["subgraph_node_2"]]:
    return Command(
        goto="subgraph_node_2",
        update={
            "node_name": "subgraph_node_1",
            "foo": "Update at subgraph_node_1!",
        },
    )


def subgraph_node_2(state: State) -> Command:
    return Command(
        goto="node_3",
        update={"node_name": "subgraph_node_2"},
        graph=Command.PARENT,
    )


subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.set_entry_point("subgraph_node_1")
subgraph = subgraph_builder.compile()


# Define main graph
def node_1(state: State) -> Command[Literal["node_2"]]:
    return Command(
        goto="node_2",
        update={"node_name": "node_1"},
    )


def node_3(state: State) -> Command[Literal["__end__"]]:
    return Command(
        goto=END,
        update={"node_name": "node_3"},
    )


main_builder = StateGraph(State)
main_builder.add_node("node_1", node_1)
main_builder.add_node("node_2", subgraph)
main_builder.add_node("node_3", node_3)
main_builder.set_entry_point("node_1")
main_graph = main_builder.compile()


# Build subgraph
with open("graph_use_command.md", "w") as file:
    file.write(f"```mermaid\n{main_graph.get_graph(xray=1).draw_mermaid()}```")

initial = {"node_name": "__start__"}
for chunk in main_graph.stream(initial, stream_mode="values", subgraphs=True):
    print(chunk)

Here are the output results:

((), {'node_name': '__start__'})
((), {'node_name': 'node_1'})
(('node_2:c026b2a3-4881-0012-5f08-25720c2d79a9',), {'node_name': 'node_1'})
(('node_2:c026b2a3-4881-0012-5f08-25720c2d79a9',), {'node_name': 'subgraph_node_1', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'subgraph_node_2', 'foo': 'Update at subgraph_node_1!'})
((), {'node_name': 'node_3', 'foo': 'Update at subgraph_node_1!'})

Thank you for reading.