How to stream from subgraphs

Prerequisites

This guide assumes familiarity with the following:

• Subgraphs
• Chat Models

If you have created a graph with subgraphs, you may wish to stream outputs from those subgraphs. To do so, you can specify subgraphs=True in parent graph's .stream() method:

for chunk in parent_graph.stream(
    {"foo": "foo"},
    subgraphs=True
):
    print(chunk)

Setup

First let's install the required packages

%%capture --no-stderr
%pip install -U langgraph

Set up LangSmith for LangGraph development

Sign up for LangSmith to quickly spot issues and improve the performance of your LangGraph projects. LangSmith lets you use trace data to debug, test, and monitor your LLM apps built with LangGraph — read more about how to get started here.

Example

Let's define a simple example:

from langgraph.graph import START, StateGraph
from typing import TypedDict


# Define subgraph
class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str


def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}


def subgraph_node_2(state: SubgraphState):
    return {"foo": state["foo"] + state["bar"]}


subgraph_builder = StateGraph(SubgraphState)
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 = subgraph_builder.compile()


# Define parent graph
class ParentState(TypedDict):
    foo: str


def node_1(state: ParentState):
    return {"foo": "hi! " + state["foo"]}


builder = StateGraph(ParentState)
builder.add_node("node_1", node_1)
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
graph = builder.compile()

API Reference: START | StateGraph

Let's now stream the outputs from the graph:

for chunk in graph.stream({"foo": "foo"}, stream_mode="updates"):
    print(chunk)

{'node_1': {'foo': 'hi! foo'}}
{'node_2': {'foo': 'hi! foobar'}}

You can see that we're only emitting the updates from the parent graph nodes (node_1 and node_2). To emit the updates from the subgraph nodes you can specify subgraphs=True:

for chunk in graph.stream(
    {"foo": "foo"},
    stream_mode="updates",
    subgraphs=True,
):
    print(chunk)

((), {'node_1': {'foo': 'hi! foo'}})
(('node_2:b692b345-cfb3-b709-628c-f0ba9608f72e',), {'subgraph_node_1': {'bar': 'bar'}})
(('node_2:b692b345-cfb3-b709-628c-f0ba9608f72e',), {'subgraph_node_2': {'foo': 'hi! foobar'}})
((), {'node_2': {'foo': 'hi! foobar'}})

Voila! The streamed outputs now contain updates from both the parent graph and the subgraph. Note that we are receiving not just the node updates, but we also the namespaces which tell us what graph (or subgraph) we are streaming from.

Comments