如何从子图流式传输

先决条件

本指南假设您熟悉以下内容：

• 子图
• 聊天模型

如果您创建了一个包含子图的图，您可能希望从这些子图流式传输输出。为此，您可以在父图的.stream()方法中指定subgraphs=True：

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

环境搭建

首先，让我们安装所需的包

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

为LangGraph开发设置LangSmith

注册LangSmith，可以快速发现和解决您的LangGraph项目中的问题，提高项目性能。LangSmith允许您使用跟踪数据来调试、测试和监控使用LangGraph构建的LLM应用程序——更多关于如何开始的信息，请参阅这里。

示例

让我们定义一个简单的例子：

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

我们现在从图中流式传输输出：

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

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

你可以看到我们只从父图节点 (node_1 和 node_2) 发出了更新。要从 子图 节点发出更新，你可以指定 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'}})

好了！流式输出现在包含了来自父图和子图的更新。注意，我们不仅收到了节点的更新，还收到了命名空间信息，这些信息告诉我们正在从哪个图（或子图）进行流式传输。

Comments