外观
TIP
对于新应用,我们推荐使用事件流——这是在 LangGraph v1.2 中引入的类型化投影 API。事件流为每个投影(消息、状态、子图、输出)提供独立的迭代器,因此你可以分别消费它们,而无需根据 stream_mode 的数据块进行分支判断。
本页介绍 LangGraph 的 stream-mode(流模式)API。它通过 updates、values、messages、custom、checkpoints、tasks 和 debug 等流模式暴露图执行过程。当你需要直接访问图运行时事件或特定流模式的输出时,可以使用它。
快速开始
基本用法
LangGraph 图通过 stream(同步)和 astream(异步)方法以迭代器形式产出流式输出。传入一个或多个流模式来控制你接收的数据。
python
for chunk in graph.stream(
{"topic": "ice cream"},
stream_mode=["updates", "custom"],
version="v2",
):
if chunk["type"] == "updates":
for node_name, state in chunk["data"].items():
print(f"Node {node_name} updated: {state}")
elif chunk["type"] == "custom":
print(f"Status: {chunk['data']['status']}")bash
Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}完整示例
python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.config import get_stream_writer
class State(TypedDict):
topic: str
joke: str
def generate_joke(state: State):
writer = get_stream_writer()
writer({"status": "thinking of a joke..."})
return {"joke": f"Why did the {state['topic']} go to school? To get a sundae education!"}
graph = (
StateGraph(State)
.add_node(generate_joke)
.add_edge(START, "generate_joke")
.add_edge("generate_joke", END)
.compile()
)
for chunk in graph.stream(
{"topic": "ice cream"},
stream_mode=["updates", "custom"],
version="v2",
):
if chunk["type"] == "updates":
for node_name, state in chunk["data"].items():
print(f"Node {node_name} updated: {state}")
elif chunk["type"] == "custom":
print(f"Status: {chunk['data']['status']}")bash
Status: thinking of a joke...
Node generate_joke updated: {'joke': 'Why did the ice cream go to school? To get a sundae education!'}LangGraph 图通过 stream 方法以迭代器形式产出流式输出。
typescript
for await (const chunk of await graph.stream(inputs, {
streamMode: "updates",
})) {
console.log(chunk);
}流式输出格式(v2)
INFO
需要 LangGraph >= 1.1。本页所有示例均使用 version="v2"。
向 stream() 或 astream() 传入 version="v2" 以获得统一的输出格式。每个数据块都是一个结构一致的 StreamPart 字典——无论流模式、模式数量或子图设置如何:
python
{
"type": "values" | "updates" | "messages" | "custom" | "checkpoints" | "tasks" | "debug",
"ns": (), # 命名空间元组,为子图事件填充
"data": ..., # 实际载荷(类型因流模式而异)
}每种流模式都有对应的 TypedDict,分别为 ValuesStreamPart、UpdatesStreamPart、MessagesStreamPart、CustomStreamPart、CheckpointStreamPart、TasksStreamPart、DebugStreamPart。你可以从 langgraph.types 导入这些类型。联合类型 StreamPart 是基于 part["type"] 的不相交联合,可在编辑器和类型检查器中实现完整的类型收窄。
在 v1(默认)中,输出格式会根据你的流式选项而变化(单一模式返回原始数据,多模式返回 (mode, data) 元组,子图返回 (namespace, data) 元组)。而在 v2 中,格式始终相同:
python
for chunk in graph.stream(inputs, stream_mode="updates", version="v2"):
print(chunk["type"]) # "updates"
print(chunk["ns"]) # ()
print(chunk["data"]) # {"node_name": {"key": "value"}}python
for chunk in graph.stream(inputs, stream_mode="updates"):
print(chunk) # {"node_name": {"key": "value"}}v2 格式还支持类型收窄,这意味着你可以通过 chunk["type"] 过滤数据块并获得正确的载荷类型。每个分支都会将 part["data"] 收窄为该模式对应的特定类型:
python
for part in graph.stream(
{"topic": "ice cream"},
stream_mode=["values", "updates", "messages", "custom"],
version="v2",
):
if part["type"] == "values":
# ValuesStreamPart — 每一步之后的完整状态快照
print(f"State: topic={part['data']['topic']}")
elif part["type"] == "updates":
# UpdatesStreamPart — 每个节点仅更改过的键
for node_name, state in part["data"].items():
print(f"Node `{node_name}` updated: {state}")
elif part["type"] == "messages":
# MessagesStreamPart — 来自 LLM 调用的 (message_chunk, metadata)
msg, metadata = part["data"]
print(msg.content, end="", flush=True)
elif part["type"] == "custom":
# CustomStreamPart — 来自 get_stream_writer() 的任意数据
print(f"Progress: {part['data']['progress']}%")流模式
以列表形式将一个或多个以下流模式传给 stream 或 astream 方法:
| Mode | Type | Description |
|---|---|---|
| values | ValuesStreamPart | 每一步之后的完整状态。 |
| updates | UpdatesStreamPart | 每一步之后的状态更新。同一步骤中的多次更新会分别流式输出。 |
| messages | MessagesStreamPart | 来自 LLM 调用的 (LLM token, metadata) 二元组。 |
| custom | CustomStreamPart | 节点通过 get_stream_writer 发出的自定义数据。 |
| checkpoints | CheckpointStreamPart | 检查点事件(与 get_state() 格式相同)。需要检查点器。 |
| tasks | TasksStreamPart | 任务开始/结束事件,包含结果和错误。需要检查点器。 |
| debug | DebugStreamPart | 所有可用信息——结合了 checkpoints 和 tasks 以及额外的元数据。 |
以列表形式将一个或多个以下流模式传给 stream 方法:
| Mode | Description |
|---|---|
| values | 每一步之后的完整状态。 |
| updates | 每一步之后的状态更新。同一步骤中的多次更新会分别流式输出。 |
| messages | 来自 LLM 调用的 (LLM token, metadata) 二元组。 |
| custom | 节点通过 writer 配置参数发出的自定义数据。 |
| tools | 工具调用生命周期事件(on_tool_start、on_tool_event、on_tool_end、on_tool_error)。 |
| debug | 图执行过程中的所有可用信息。 |
图状态
使用 updates 和 values 流模式来流式输出图执行过程中的状态。
updates流式输出图每一步之后对状态的更新。values流式输出图每一步之后状态的完整值。
python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict):
topic: str
joke: str
def refine_topic(state: State):
return {"topic": state["topic"] + " and cats"}
def generate_joke(state: State):
return {"joke": f"This is a joke about {state['topic']}"}
graph = (
StateGraph(State)
.add_node(refine_topic)
.add_node(generate_joke)
.add_edge(START, "refine_topic")
.add_edge("refine_topic", "generate_joke")
.add_edge("generate_joke", END)
.compile()
)typescript
import { StateGraph, StateSchema, START, END } from "@langchain/langgraph";
import { z } from "zod/v4";
const State = new StateSchema({
topic: z.string(),
joke: z.string(),
});
const graph = new StateGraph(State)
.addNode("refineTopic", (state) => {
return { topic: state.topic + " and cats" };
})
.addNode("generateJoke", (state) => {
return { joke: `This is a joke about ${state.topic}` };
})
.addEdge(START, "refineTopic")
.addEdge("refineTopic", "generateJoke")
.addEdge("generateJoke", END)
.compile();updates
使用此模式仅流式输出每一步之后节点返回的**状态更新**。流式输出包含节点名称以及更新内容。
python
for chunk in graph.stream(
{"topic": "ice cream"},
stream_mode="updates",
version="v2",
):
if chunk["type"] == "updates":
for node_name, state in chunk["data"].items():
print(f"Node `{node_name}` updated: {state}")bash
Node `refine_topic` updated: {'topic': 'ice cream and cats'}
Node `generate_joke` updated: {'joke': 'This is a joke about ice cream and cats'}typescript
for await (const chunk of await graph.stream(
{ topic: "ice cream" },
{ streamMode: "updates" }
)) {
for (const [nodeName, state] of Object.entries(chunk)) {
console.log(`Node ${nodeName} updated:`, state);
}
}values
使用此模式流式输出图每一步之后的**完整状态**。
python
for chunk in graph.stream(
{"topic": "ice cream"},
stream_mode="values",
version="v2",
):
if chunk["type"] == "values":
print(f"topic: {chunk['data']['topic']}, joke: {chunk['data']['joke']}")bash
topic: ice cream, joke:
topic: ice cream and cats, joke:
topic: ice cream and cats, joke: This is a joke about ice cream and catstypescript
for await (const chunk of await graph.stream(
{ topic: "ice cream" },
{ streamMode: "values" }
)) {
console.log(`topic: ${chunk.topic}, joke: ${chunk.joke}`);
}LLM token
使用 messages 流模式可从图中的任何部分(包括节点、工具、子图或任务)逐 token 流式输出大语言模型(LLM)的输出。
messages 模式的流式输出是一个元组 (message_chunk, metadata),其中:
message_chunk:来自 LLM 的 token 或消息片段。metadata:包含图节点和 LLM 调用详情的字典。
如果你的 LLM 没有对应的 LangChain 集成,你可以改用
custom模式流式输出其结果。详情请参阅与任意 LLM 一起使用。
WARNING
Python < 3.11 下的异步代码需要手动配置 在 Python < 3.11 下使用异步代码时,你必须显式地向 ainvoke() 传递 RunnableConfig 以启用正确的流式传输。详情请参阅Python < 3.11 下的异步,或升级到 Python 3.11+。
python
from dataclasses import dataclass
from langchain.chat_models import init_chat_model
from langgraph.graph import StateGraph, START
@dataclass
class MyState:
topic: str
joke: str = ""
model = init_chat_model(model="gpt-5.4-mini")
def call_model(state: MyState):
"""Call the LLM to generate a joke about a topic"""
# 请注意,即使 LLM 是通过 .invoke 而非 .stream 运行的,也会发出消息事件
model_response = model.invoke(
[
{"role": "user", "content": f"Generate a joke about {state.topic}"}
]
)
return {"joke": model_response.content}
graph = (
StateGraph(MyState)
.add_node(call_model)
.add_edge(START, "call_model")
.compile()
)
# "messages" 流模式会连同元数据一起流式输出 LLM token
# 使用 version="v2" 以获得统一的 StreamPart 格式
for chunk in graph.stream(
{"topic": "ice cream"},
stream_mode="messages",
version="v2",
):
if chunk["type"] == "messages":
message_chunk, metadata = chunk["data"]
if message_chunk.content:
print(message_chunk.content, end="|", flush=True)messages 模式的流式输出是一个元组 [message_chunk, metadata],其中:
message_chunk:来自 LLM 的 token 或消息片段。metadata:包含图节点和 LLM 调用详情的字典。
如果你的 LLM 没有对应的 LangChain 集成,你可以改用
custom模式流式输出其结果。详情请参阅与任意 LLM 一起使用。
typescript
import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import * as z from "zod";
const MyState = new StateSchema({
topic: z.string(),
joke: z.string().default(""),
});
const model = new ChatOpenAI({ model: "gpt-5.4-mini" });
const callModel: GraphNode<typeof MyState> = async (state) => {
// 调用 LLM 生成关于某个主题的笑话
// 请注意,即使 LLM 是通过 .invoke 而非 .stream 运行的,也会发出消息事件
const modelResponse = await model.invoke([
{ role: "user", content: `Generate a joke about ${state.topic}` },
]);
return { joke: modelResponse.content };
};
const graph = new StateGraph(MyState)
.addNode("callModel", callModel)
.addEdge(START, "callModel")
.compile();
// "messages" 流模式返回 [messageChunk, metadata] 元组的迭代器
// 其中 messageChunk 是 LLM 流式输出的 token,metadata 是一个字典
// 包含调用 LLM 的图节点信息以及其他信息
for await (const [messageChunk, metadata] of await graph.stream(
{ topic: "ice cream" },
{ streamMode: "messages" }
)) {
if (messageChunk.content) {
console.log(messageChunk.content + "|");
}
}按 LLM 调用进行过滤
你可以将 tags 与 LLM 调用关联起来,按 LLM 调用过滤流式输出的 token。
python
from langchain.chat_models import init_chat_model
# model_1 带有 "joke" 标签
model_1 = init_chat_model(model="gpt-5.4-mini", tags=['joke'])
# model_2 带有 "poem" 标签
model_2 = init_chat_model(model="gpt-5.4-mini", tags=['poem'])
graph = ... # 定义一个使用这些 LLM 的图
# stream_mode 设置为 "messages" 以流式输出 LLM token
# metadata 包含有关 LLM 调用的信息,包括 tags
async for chunk in graph.astream(
{"topic": "cats"},
stream_mode="messages",
version="v2",
):
if chunk["type"] == "messages":
msg, metadata = chunk["data"]
# 根据 metadata 中的 tags 字段过滤流式输出的 token,只包含
# 带有 "joke" 标签的 LLM 调用的 token
if metadata["tags"] == ["joke"]:
print(msg.content, end="|", flush=True)typescript
import { ChatOpenAI } from "@langchain/openai";
// model1 带有 "joke" 标签
const model1 = new ChatOpenAI({
model: "gpt-5.4-mini",
tags: ['joke']
});
// model2 带有 "poem" 标签
const model2 = new ChatOpenAI({
model: "gpt-5.4-mini",
tags: ['poem']
});
const graph = // ... 定义一个使用这些 LLM 的图
// streamMode 设置为 "messages" 以流式输出 LLM token
// metadata 包含有关 LLM 调用的信息,包括 tags
for await (const [msg, metadata] of await graph.stream(
{ topic: "cats" },
{ streamMode: "messages" }
)) {
// 根据 metadata 中的 tags 字段过滤流式输出的 token,只包含
// 带有 "joke" 标签的 LLM 调用的 token
if (metadata.tags?.includes("joke")) {
console.log(msg.content + "|");
}
}扩展示例:按标签过滤
python
from typing import TypedDict
from langchain.chat_models import init_chat_model
from langgraph.graph import START, StateGraph
# joke_model 带有 "joke" 标签
joke_model = init_chat_model(model="gpt-5.4-mini", tags=["joke"])
# poem_model 带有 "poem" 标签
poem_model = init_chat_model(model="gpt-5.4-mini", tags=["poem"])
class State(TypedDict):
topic: str
joke: str
poem: str
async def call_model(state, config):
topic = state["topic"]
print("Writing joke...")
# 注意:对于 python < 3.11,需要显式传递 config
# 因为在 3.11 之前尚不支持上下文变量:https://docs.python.org/3/library/asyncio-task.html#creating-tasks
# 显式传递 config 以确保上下文变量被正确传播
# 在使用异步代码时,Python < 3.11 需要这样做。更多细节请参阅异步部分
joke_response = await joke_model.ainvoke(
[{"role": "user", "content": f"Write a joke about {topic}"}],
config,
)
print("\n\nWriting poem...")
poem_response = await poem_model.ainvoke(
[{"role": "user", "content": f"Write a short poem about {topic}"}],
config,
)
return {"joke": joke_response.content, "poem": poem_response.content}
graph = (
StateGraph(State)
.add_node(call_model)
.add_edge(START, "call_model")
.compile()
)
# stream_mode 设置为 "messages" 以流式输出 LLM token
# metadata 包含有关 LLM 调用的信息,包括 tags
async for chunk in graph.astream(
{"topic": "cats"},
stream_mode="messages",
version="v2",
):
if chunk["type"] == "messages":
msg, metadata = chunk["data"]
if metadata["tags"] == ["joke"]:
print(msg.content, end="|", flush=True)typescript
import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import * as z from "zod";
// jokeModel 带有 "joke" 标签
const jokeModel = new ChatOpenAI({
model: "gpt-5.4-mini",
tags: ["joke"]
});
// poemModel 带有 "poem" 标签
const poemModel = new ChatOpenAI({
model: "gpt-5.4-mini",
tags: ["poem"]
});
const State = new StateSchema({
topic: z.string(),
joke: z.string(),
poem: z.string(),
});
const callModel: GraphNode<typeof State> = async (state) => {
const topic = state.topic;
console.log("Writing joke...");
const jokeResponse = await jokeModel.invoke([
{ role: "user", content: `Write a joke about ${topic}` }
]);
console.log("\n\nWriting poem...");
const poemResponse = await poemModel.invoke([
{ role: "user", content: `Write a short poem about ${topic}` }
]);
return {
joke: jokeResponse.content,
poem: poemResponse.content
};
};
const graph = new StateGraph(State)
.addNode("callModel", callModel)
.addEdge(START, "callModel")
.compile();
// streamMode 设置为 "messages" 以流式输出 LLM token
// metadata 包含有关 LLM 调用的信息,包括 tags
for await (const [msg, metadata] of await graph.stream(
{ topic: "cats" },
{ streamMode: "messages" }
)) {
// 根据 metadata 中的 tags 字段过滤流式输出的 token,只包含
// 带有 "joke" 标签的 LLM 调用的 token
if (metadata.tags?.includes("joke")) {
console.log(msg.content + "|");
}
}从流中省略消息
使用 nostream 标签可完全排除流中的 LLM 输出。带有 nostream 标签的调用仍然会运行并产生输出;只是它们的 token 不会在 messages 模式中发出。
这在以下情况下非常有用:
- 你需要 LLM 输出进行内部处理(例如结构化输出),但不希望将其流式传输到客户端
- 你通过其他通道(例如自定义 UI 消息)流式输出相同的内容,并希望避免
messages流中出现重复输出
python
from typing import Any, TypedDict
from langchain_anthropic import ChatAnthropic
from langgraph.graph import START, StateGraph
stream_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001")
internal_model = ChatAnthropic(model_name="claude-haiku-4-5-20251001").with_config(
{"tags": ["nostream"]}
)
class State(TypedDict):
topic: str
answer: str
notes: str
def answer(state: State) -> dict[str, Any]:
r = stream_model.invoke(
[{"role": "user", "content": f"Reply briefly about {state['topic']}"}]
)
return {"answer": r.content}
def internal_notes(state: State) -> dict[str, Any]:
# Tokens from this model are omitted from stream_mode="messages" because of nostream
r = internal_model.invoke(
[{"role": "user", "content": f"Private notes on {state['topic']}"}]
)
return {"notes": r.content}
graph = (
StateGraph(State)
.add_node("write_answer", answer)
.add_node("internal_notes", internal_notes)
.add_edge(START, "write_answer")
.add_edge("write_answer", "internal_notes")
.compile()
)
initial_state: State = {"topic": "AI", "answer": "", "notes": ""}
stream = graph.stream_events(initial_state, version="v3")ts
import { ChatAnthropic } from "@langchain/anthropic";
import { StateGraph, StateSchema, START } from "@langchain/langgraph";
import * as z from "zod";
const streamModel = new ChatAnthropic({ model: "claude-haiku-4-5-20251001" });
const internalModel = new ChatAnthropic({
model: "claude-haiku-4-5-20251001",
}).withConfig({
tags: ["nostream"],
});
const State = new StateSchema({
topic: z.string(),
answer: z.string().optional(),
notes: z.string().optional(),
});
const contentToText = (content: unknown): string => {
if (typeof content === "string") {
return content;
}
if (Array.isArray(content)) {
return content
.map((block) => {
if (
typeof block === "object" &&
block !== null &&
"text" in block &&
typeof (block as { text?: unknown }).text === "string"
) {
return (block as { text: string }).text;
}
return "";
})
.filter(Boolean)
.join("\n");
}
return "";
};
const writeAnswer = async (state: typeof State.State) => {
const r = await streamModel.invoke([
{ role: "user", content: `Reply briefly about ${state.topic}` },
]);
return { answer: contentToText(r.content) };
};
const internalNotes = async (state: typeof State.State) => {
// Tokens from this model are omitted from streamMode: "messages" because of nostream
const r = await internalModel.invoke([
{ role: "user", content: `Private notes on ${state.topic}` },
]);
return { notes: contentToText(r.content) };
};
const graph = new StateGraph(State)
.addNode("writeAnswer", writeAnswer)
.addNode("internal_notes", internalNotes)
.addEdge(START, "writeAnswer")
.addEdge("writeAnswer", "internal_notes")
.compile();
const stream = await graph.streamEvents(
{ topic: "AI", answer: "", notes: "" },
{ version: "v3" },
);按节点过滤
要仅从特定节点流式输出 token,请使用 stream_mode="messages",并根据流式元数据中的 langgraph_node 字段过滤输出:
python
# "messages" 流模式会连同元数据一起流式输出 LLM token
# 使用 version="v2" 以获得统一的 StreamPart 格式
for chunk in graph.stream(
inputs,
stream_mode="messages",
version="v2",
):
if chunk["type"] == "messages":
msg, metadata = chunk["data"]
# 根据 metadata 中的 langgraph_node 字段过滤流式输出的 token
# 只包含来自指定节点的 token
if msg.content and metadata["langgraph_node"] == "some_node_name":
...typescript
// "messages" 流模式返回 [messageChunk, metadata] 元组
// 其中 messageChunk 是 LLM 流式输出的 token,metadata 是一个字典
// 包含调用 LLM 的图节点信息以及其他信息
for await (const [msg, metadata] of await graph.stream(
inputs,
{ streamMode: "messages" }
)) {
// 根据 metadata 中的 langgraph_node 字段过滤流式输出的 token
// 只包含来自指定节点的 token
if (msg.content && metadata.langgraph_node === "some_node_name") {
// ...
}
}扩展示例:从特定节点流式输出 LLM token
python
from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-5.4-mini")
class State(TypedDict):
topic: str
joke: str
poem: str
def write_joke(state: State):
topic = state["topic"]
joke_response = model.invoke(
[{"role": "user", "content": f"Write a joke about {topic}"}]
)
return {"joke": joke_response.content}
def write_poem(state: State):
topic = state["topic"]
poem_response = model.invoke(
[{"role": "user", "content": f"Write a short poem about {topic}"}]
)
return {"poem": poem_response.content}
graph = (
StateGraph(State)
.add_node(write_joke)
.add_node(write_poem)
# 同时编写笑话和诗歌
.add_edge(START, "write_joke")
.add_edge(START, "write_poem")
.compile()
)
# "messages" 流模式会连同元数据一起流式输出 LLM token
# 使用 version="v2" 以获得统一的 StreamPart 格式
for chunk in graph.stream(
{"topic": "cats"},
stream_mode="messages",
version="v2",
):
if chunk["type"] == "messages":
msg, metadata = chunk["data"]
# 根据 metadata 中的 langgraph_node 字段过滤流式输出的 token
# 只包含来自 write_poem 节点的 token
if msg.content and metadata["langgraph_node"] == "write_poem":
print(msg.content, end="|", flush=True)typescript
import { ChatOpenAI } from "@langchain/openai";
import { StateGraph, StateSchema, GraphNode, START } from "@langchain/langgraph";
import * as z from "zod";
const model = new ChatOpenAI({ model: "gpt-5.4-mini" });
const State = new StateSchema({
topic: z.string(),
joke: z.string(),
poem: z.string(),
});
const writeJoke: GraphNode<typeof State> = async (state) => {
const topic = state.topic;
const jokeResponse = await model.invoke([
{ role: "user", content: `Write a joke about ${topic}` }
]);
return { joke: jokeResponse.content };
};
const writePoem: GraphNode<typeof State> = async (state) => {
const topic = state.topic;
const poemResponse = await model.invoke([
{ role: "user", content: `Write a short poem about ${topic}` }
]);
return { poem: poemResponse.content };
};
const graph = new StateGraph(State)
.addNode("writeJoke", writeJoke)
.addNode("writePoem", writePoem)
// 同时编写笑话和诗歌
.addEdge(START, "writeJoke")
.addEdge(START, "writePoem")
.compile();
// "messages" 流模式返回 [messageChunk, metadata] 元组
// 其中 messageChunk 是 LLM 流式输出的 token,metadata 是一个字典
// 包含调用 LLM 的图节点信息以及其他信息
for await (const [msg, metadata] of await graph.stream(
{ topic: "cats" },
{ streamMode: "messages" }
)) {
// 根据 metadata 中的 langgraph_node 字段过滤流式输出的 token
// 只包含来自 writePoem 节点的 token
if (msg.content && metadata.langgraph_node === "writePoem") {
console.log(msg.content + "|");
}
}自定义数据
要从 LangGraph 节点或工具内部发送自定义的用户定义数据,请按照以下步骤操作:
- 使用
get_stream_writer访问流写入器并发出自定义数据。 - 在调用
.stream()或.astream()时设置stream_mode="custom",以在流中获取自定义数据。你可以组合多个模式(例如["updates", "custom"]),但至少有一个必须是"custom"。
WARNING
Python < 3.11 下的异步代码中无法使用 get_stream_writer 在 Python < 3.11 上运行的异步代码中,get_stream_writer 将无法工作。 请改为在节点或工具中添加 writer 参数并手动传递。 用法示例请参阅Python < 3.11 下的异步。
node
python
from typing import TypedDict
from langgraph.config import get_stream_writer
from langgraph.graph import StateGraph, START
class State(TypedDict):
query: str
answer: str
def node(state: State):
# 获取流写入器以发送自定义数据
writer = get_stream_writer()
# 发送一个自定义键值对(例如进度更新)
writer({"custom_key": "Generating custom data inside node"})
return {"answer": "some data"}
graph = (
StateGraph(State)
.add_node(node)
.add_edge(START, "node")
.compile()
)
inputs = {"query": "example"}
# 设置 stream_mode="custom" 以在流中接收自定义数据
for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
if chunk["type"] == "custom":
print(f"Custom event: {chunk['data']['custom_key']}")tool
python
from langchain.tools import tool
from langgraph.config import get_stream_writer
@tool
def query_database(query: str) -> str:
"""Query the database."""
# 访问流写入器以发送自定义数据
writer = get_stream_writer()
# 发送一个自定义键值对(例如进度更新)
writer({"data": "Retrieved 0/100 records", "type": "progress"})
# 执行查询
# 发送另一个自定义键值对
writer({"data": "Retrieved 100/100 records", "type": "progress"})
return "some-answer"
graph = ... # 定义一个使用此工具的图
# 设置 stream_mode="custom" 以在流中接收自定义数据
for chunk in graph.stream(inputs, stream_mode="custom", version="v2"):
if chunk["type"] == "custom":
print(f"{chunk['data']['type']}: {chunk['data']['data']}")要从 LangGraph 节点或工具内部发送自定义的用户定义数据,请按照以下步骤操作:
- 使用
LangGraphRunnableConfig中的writer参数发出自定义数据。 - 在调用
.stream()时设置streamMode: "custom",以在流中获取自定义数据。你可以组合多个模式(例如["updates", "custom"]),但至少有一个必须是"custom"。
node
typescript
import { StateGraph, StateSchema, GraphNode, START, LangGraphRunnableConfig } from "@langchain/langgraph";
import * as z from "zod";
const State = new StateSchema({
query: z.string(),
answer: z.string(),
});
const node: GraphNode<typeof State> = async (state, config) => {
// 使用 writer 发送一个自定义键值对(例如进度更新)
config.writer({ custom_key: "Generating custom data inside node" });
return { answer: "some data" };
};
const graph = new StateGraph(State)
.addNode("node", node)
.addEdge(START, "node")
.compile();
const inputs = { query: "example" };
// 设置 streamMode: "custom" 以在流中接收自定义数据
for await (const chunk of await graph.stream(inputs, { streamMode: "custom" })) {
console.log(chunk);
}tool
typescript
import { tool } from "@langchain/core/tools";
import { LangGraphRunnableConfig } from "@langchain/langgraph";
import * as z from "zod";
const queryDatabase = tool(
async (input, config: LangGraphRunnableConfig) => {
// 使用 writer 发送一个自定义键值对(例如进度更新)
config.writer({ data: "Retrieved 0/100 records", type: "progress" });
// 执行查询
// 发送另一个自定义键值对
config.writer({ data: "Retrieved 100/100 records", type: "progress" });
return "some-answer";
},
{
name: "query_database",
description: "Query the database.",
schema: z.object({
query: z.string().describe("The query to execute."),
}),
}
);
const graph = // ... 定义一个使用此工具的图
// 设置 streamMode: "custom" 以在流中接收自定义数据
for await (const chunk of await graph.stream(inputs, { streamMode: "custom" })) {
console.log(chunk);
}工具进度
使用 tools 流模式可实时接收工具执行的生命周期事件。这在工具运行期间用于在界面中显示进度指示器、部分结果和错误状态非常有用。
tools 流模式会发出四种事件类型:
| Event | When | Payload |
|---|---|---|
on_tool_start | 工具调用开始 | name, input, toolCallId |
on_tool_event | 工具产出中间数据 | name, data, toolCallId |
on_tool_end | 工具返回最终结果 | name, output, toolCallId |
on_tool_error | 工具抛出错误 | name, error, toolCallId |
定义流式输出进度的工具
要发出 on_tool_event 事件,请将工具函数定义为异步生成器(async function*)。每次 yield 都会向流中发送中间数据,return 值用作工具的最终结果。
typescript
import { tool } from "@langchain/core/tools";
import { z } from "zod/v4";
const searchFlights = tool(
async function* (input) {
const airlines = ["United", "Delta", "American", "JetBlue"];
const completed: string[] = [];
for (let i = 0; i < airlines.length; i++) {
await new Promise((r) => setTimeout(r, 500));
completed.push(airlines[i]);
// 每次 yield 都会向流中发送一个 on_tool_event
yield {
message: `Searching ${airlines[i]}...`,
progress: (i + 1) / airlines.length,
completed,
};
}
// 返回值成为工具结果(ToolMessage.content)
return JSON.stringify({
flights: [
{ airline: "United", price: 450, duration: "5h 30m" },
{ airline: "Delta", price: 520, duration: "5h 15m" },
],
});
},
{
name: "search_flights",
description: "Search for available flights to a destination.",
schema: z.object({
destination: z.string(),
date: z.string(),
}),
}
);INFO
返回 Promise 的现有工具完全兼容。它们会发出 on_tool_start 和 on_tool_end 事件,但不会发出 on_tool_event 事件。
在服务端消费工具事件
向 graph.stream() 传入 streamMode: ["tools"](或与其他模式组合):
typescript
for await (const [mode, chunk] of await graph.stream(
{ messages: [{ role: "user", content: "Find flights to Tokyo" }] },
{ streamMode: ["updates", "tools"] }
)) {
if (mode === "tools") {
switch (chunk.event) {
case "on_tool_start":
console.log(`Tool started: ${chunk.name}`, chunk.input);
break;
case "on_tool_event":
console.log(`Tool progress: ${chunk.name}`, chunk.data);
break;
case "on_tool_end":
console.log(`Tool finished: ${chunk.name}`, chunk.output);
break;
case "on_tool_error":
console.error(`Tool failed: ${chunk.name}`, chunk.error);
break;
}
}
}在 React 中使用 useStream 利用工具进度
当你在流模式中包含 "tools" 时,来自 @langchain/langgraph-sdk/react 的 useStream Hook 会暴露一个 toolProgress 数组。每个条目都是一个跟踪运行中工具当前状态的 ToolProgress 对象:
| Field | Description |
|---|---|
name | 工具名称 |
state | 当前生命周期状态:"starting"、"running"、"completed" 或 "error" |
toolCallId | 来自 LLM 的工具调用 ID |
input | 工具的输入参数 |
data | 来自 on_tool_event 的最新产出数据 |
result | 最终结果,在 on_tool_end 时设置 |
error | 错误信息,在 on_tool_error 时设置 |
typescript
import { useStream } from "@langchain/langgraph-sdk/react";
function Chat() {
const stream = useStream({
assistantId: "my-agent",
streamMode: ["values", "tools"],
});
// 过滤出正在运行的工具
const activeTools = stream.toolProgress.filter(
(t) => t.state === "starting" || t.state === "running"
);
return (
{stream.messages.map((msg) => (
<MessageBubble key={msg.id} message={msg} />
))}
{/* 为运行中的工具显示进度卡片 */}
{activeTools.map((tool) => (
<ToolProgressCard
key={tool.toolCallId ?? tool.name}
name={tool.name}
state={tool.state}
data={tool.data}
/>
))}
);
}扩展示例:带工具进度的旅行规划智能体
此示例展示了一个完整的智能体,其中的异步生成器工具可将搜索进度流式输出到 React UI。
智能体定义:
typescript
import { tool } from "@langchain/core/tools";
import { ChatOpenAI } from "@langchain/openai";
import { createAgent } from "@langchain/langgraph";
import { MemorySaver } from "@langchain/langgraph-checkpoint-memory";
import { z } from "zod/v4";
const searchFlights = tool(
async function* (input) {
const airlines = ["United", "Delta", "American", "JetBlue"];
const completed: string[] = [];
for (let i = 0; i < airlines.length; i++) {
await new Promise((r) => setTimeout(r, 600));
completed.push(`${airlines[i]}: checked`);
yield {
message: `Searching ${airlines[i]}...`,
progress: (i + 1) / airlines.length,
completed,
};
}
return JSON.stringify({
flights: [
{ airline: "United", price: 450, duration: "5h 30m" },
{ airline: "Delta", price: 520, duration: "5h 15m" },
],
});
},
{
name: "search_flights",
description: "Search for available flights.",
schema: z.object({
destination: z.string(),
departure_date: z.string(),
}),
}
);
const checkHotels = tool(
async function* (input) {
const hotels = ["Grand Hyatt", "Marriott", "Hilton"];
const completed: string[] = [];
for (let i = 0; i < hotels.length; i++) {
await new Promise((r) => setTimeout(r, 400));
completed.push(`${hotels[i]}: available`);
yield {
message: `Checking ${hotels[i]}...`,
progress: (i + 1) / hotels.length,
completed,
};
}
return JSON.stringify({
hotels: [
{ name: "Grand Hyatt", price: 250, rating: 4.5 },
{ name: "Marriott", price: 180, rating: 4.2 },
],
});
},
{
name: "check_hotels",
description: "Check hotel availability.",
schema: z.object({
city: z.string(),
check_in: z.string(),
nights: z.number(),
}),
}
);
export const agent = createAgent({
model: new ChatOpenAI({ model: "gpt-5.4-mini" }),
tools: [searchFlights, checkHotels],
checkpointer: new MemorySaver(),
});带进度卡片的 React 组件:
typescript
import { useStream } from "@langchain/langgraph-sdk/react";
function TravelPlanner() {
const stream = useStream<typeof agent>({
assistantId: "travel-agent",
streamMode: ["values", "tools"],
});
const activeTools = stream.toolProgress.filter(
(t) => t.state === "starting" || t.state === "running"
);
return (
{stream.messages.map((msg) => (
{msg.content}
))}
{activeTools.map((tool) => {
const data = tool.data as {
message?: string;
progress?: number;
completed?: string[];
} | undefined;
return (
{tool.name}
{data?.message && {data.message}}
{data?.progress != null && (
<div
style={{
width: `${data.progress * 100}%`,
background: "#4CAF50",
height: 8,
transition: "width 0.3s ease",
}}
/>
)}
{data?.completed?.map((step, i) => (
✓ {step}
))}
);
})}
);
}tools 与 custom 流模式对比
两种流模式都可以呈现工具进度,但它们的用途不同:
tools——自动发出结构化生命周期事件(on_tool_start、on_tool_event、on_tool_end、on_tool_error),除使用async function*外,无需修改工具代码。useStreamHook 开箱即用地提供了响应式的toolProgress数组。custom——使用config.writer()时,你可以完全控制发出哪些数据以及何时发出。当你需要无法映射到工具生命周期的自由格式数据,或想从节点(而不仅仅是工具)流式输出时,可使用此模式。
子图输出
要将子图的输出包含在流式输出中,你可以在父图的 .stream() 方法中设置 subgraphs=True。这样父图和所有子图的输出都会被流式输出。
输出将以元组 (namespace, data) 的形式流式传输,其中 namespace 是包含子图被调用处节点路径的元组,例如 ("parent_node:<task_id>", "child_node:<task_id>")。
v2 (LangGraph >= 1.1)
使用 `version="v2"` 时,子图事件使用相同的 `StreamPart` 格式。`ns` 字段标识来源:
python
for chunk in graph.stream(
{"foo": "foo"},
subgraphs=True,
stream_mode="updates",
version="v2",
):
print(chunk["type"]) # "updates"
print(chunk["ns"]) # () 表示根图,("node_name:<task_id>",) 表示子图
print(chunk["data"]) # {"node_name": {"key": "value"}}v1 (default)
python
for chunk in graph.stream(
{"foo": "foo"},
# 设置 subgraphs=True 以流式输出子图的输出
subgraphs=True,
stream_mode="updates",
):
print(chunk)要将子图的输出包含在流式输出中,你可以在父图的 .stream() 方法中设置 subgraphs: true。这样父图和所有子图的输出都会被流式输出。
输出将以元组 [namespace, data] 的形式流式传输,其中 namespace 是包含子图被调用处节点路径的元组,例如 ["parent_node:<task_id>", "child_node:<task_id>"]。
typescript
for await (const chunk of await graph.stream(
{ foo: "foo" },
{
// 设置 subgraphs: true 以流式输出子图的输出
subgraphs: true,
streamMode: "updates",
}
)) {
console.log(chunk);
}INFO
这适用于每种 stream_mode,包括 "messages"。像 create_agent 这样的智能体构建器返回编译后的图,因此将一个智能体作为节点添加后,它就会变成子图。如果没有 subgraphs=True,父图上的 stream_mode="messages" 将不会发出内部智能体 LLM 调用的 token 数据块。直接调用 agent.stream(...) 则会产生这些 token,这就是为什么这种问题通常只在包装之后才出现。
python
from langchain.agents import create_agent
from langgraph.graph import END, START, StateGraph
graph = (
StateGraph(State)
.add_node("agent", create_agent(model, tools, state_schema=State))
.add_edge(START, "agent")
.add_edge("agent", END)
.compile()
)
for chunk in graph.stream(
{"messages": [{"role": "user", "content": "..."}]},
stream_mode="messages",
subgraphs=True,
version="v2",
):
print(chunk["type"]) # "messages"
print(chunk["ns"]) # () 表示根图,("agent:<task_id>",) 表示子图
print(chunk["data"]) # (token, metadata)INFO
这适用于每种 streamMode,包括 "messages"。createAgent 返回一个 ReactAgent 包装器;将其作为节点添加时,传入 agent.graph,这样父图就会把它当作子图。使用 subgraphs: true 时,消息数据块为 [namespace, [token, metadata]],因此你可以知道是哪个子图发出了它们。
typescript
import { createAgent } from "langchain";
import { END, START, StateGraph } from "@langchain/langgraph";
const agent = createAgent({ model, tools, stateSchema: State });
const graph = new StateGraph(State)
.addNode("agent", agent.graph)
.addEdge(START, "agent")
.addEdge("agent", END)
.compile();
for await (const [ns, data] of await graph.stream(
{ messages: [{ role: "user", content: "..." }] },
{
streamMode: "messages",
subgraphs: true,
}
)) {
const [token, metadata] = data;
console.log(ns, token, metadata);
}扩展示例:从子图流式输出
python
from langgraph.graph import START, StateGraph
from typing import TypedDict
# 定义子图
class SubgraphState(TypedDict):
foo: str # 注意,此键与父图状态共享
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()
# 定义父图
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()
for chunk in graph.stream(
{"foo": "foo"},
stream_mode="updates",
# 设置 subgraphs=True 以流式输出子图的输出
subgraphs=True,
version="v2",
):
if chunk["type"] == "updates":
if chunk["ns"]:
print(f"Subgraph {chunk['ns']}: {chunk['data']}")
else:
print(f"Root: {chunk['data']}")typescript
import { StateGraph, StateSchema, START } from "@langchain/langgraph";
import { z } from "zod/v4";
// 定义子图
const SubgraphState = new StateSchema({
foo: z.string(), // 注意,此键与父图状态共享
bar: z.string(),
});
const subgraphBuilder = new StateGraph(SubgraphState)
.addNode("subgraphNode1", (state) => {
return { bar: "bar" };
})
.addNode("subgraphNode2", (state) => {
return { foo: state.foo + state.bar };
})
.addEdge(START, "subgraphNode1")
.addEdge("subgraphNode1", "subgraphNode2");
const subgraph = subgraphBuilder.compile();
// 定义父图
const ParentState = new StateSchema({
foo: z.string(),
});
const builder = new StateGraph(ParentState)
.addNode("node1", (state) => {
return { foo: "hi! " + state.foo };
})
.addNode("node2", subgraph)
.addEdge(START, "node1")
.addEdge("node1", "node2");
const graph = builder.compile();
for await (const chunk of await graph.stream(
{ foo: "foo" },
{
streamMode: "updates",
// 设置 subgraphs: true 以流式输出子图的输出
subgraphs: true,
}
)) {
console.log(chunk);
}Root: {'node_1': {'foo': 'hi! foo'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_1': {'bar': 'bar'}}
Subgraph ('node_2:dfddc4ba-c3c5-6887-5012-a243b5b377c2',): {'subgraph_node_2': {'foo': 'hi! foobar'}}
Root: {'node_2': {'foo': 'hi! foobar'}}[[], {'node1': {'foo': 'hi! foo'}}]
[['node2:dfddc4ba-c3c5-6887-5012-a243b5b377c2'], {'subgraphNode1': {'bar': 'bar'}}]
[['node2:dfddc4ba-c3c5-6887-5012-a243b5b377c2'], {'subgraphNode2': {'foo': 'hi! foobar'}}]
[[], {'node2': {'foo': 'hi! foobar'}}]请注意,我们收到的不仅是节点更新,还有命名空间,它告诉我们流式输出来自哪个图(或子图)。
检查点
使用 checkpoints 流模式可在图执行过程中接收检查点事件。每个检查点事件与 get_state() 输出的格式相同。需要检查点器。
python
from langgraph.checkpoint.memory import MemorySaver
graph = (
StateGraph(State)
.add_node(refine_topic)
.add_node(generate_joke)
.add_edge(START, "refine_topic")
.add_edge("refine_topic", "generate_joke")
.add_edge("generate_joke", END)
.compile(checkpointer=MemorySaver())
)
config = {"configurable": {"thread_id": "1"}}
for chunk in graph.stream(
{"topic": "ice cream"},
config=config,
stream_mode="checkpoints",
version="v2",
):
if chunk["type"] == "checkpoints":
print(chunk["data"])任务
使用 tasks 流模式可在图执行过程中接收任务开始和结束事件。任务事件包含正在运行的节点、其结果以及任何错误的信息。需要检查点器。
python
from langgraph.checkpoint.memory import MemorySaver
graph = (
StateGraph(State)
.add_node(refine_topic)
.add_node(generate_joke)
.add_edge(START, "refine_topic")
.add_edge("refine_topic", "generate_joke")
.add_edge("generate_joke", END)
.compile(checkpointer=MemorySaver())
)
config = {"configurable": {"thread_id": "1"}}
for chunk in graph.stream(
{"topic": "ice cream"},
config=config,
stream_mode="tasks",
version="v2",
):
if chunk["type"] == "tasks":
print(chunk["data"])调试
使用 debug 流模式可在图执行过程中尽可能多地流式输出信息。流式输出包含节点名称以及完整状态。
python
for chunk in graph.stream(
{"topic": "ice cream"},
stream_mode="debug",
version="v2",
):
if chunk["type"] == "debug":
print(chunk["data"])typescript
for await (const chunk of await graph.stream(
{ topic: "ice cream" },
{ streamMode: "debug" }
)) {
console.log(chunk);
}INFO
debug 模式将 checkpoints 和 tasks 事件与额外的元数据相结合。如果你只需要调试信息的一部分,请直接使用 checkpoints 或 tasks。
同时使用多个模式
你可以将列表作为 stream_mode 参数传入,以同时流式输出多个模式。
使用 version="v2" 时,每个数据块都是一个 StreamPart 字典。使用 chunk["type"] 来区分不同模式:
python
for chunk in graph.stream(inputs, stream_mode=["updates", "custom"], version="v2"):
if chunk["type"] == "updates":
for node_name, state in chunk["data"].items():
print(f"Node `{node_name}` updated: {state}")
elif chunk["type"] == "custom":
print(f"Custom event: {chunk['data']}")python
for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
print(chunk)你可以将数组作为 streamMode 参数传入,以同时流式输出多个模式。
流式输出将是 [mode, chunk] 形式的元组,其中 mode 是流模式的名称,chunk 是该模式流式输出的数据。
typescript
for await (const [mode, chunk] of await graph.stream(inputs, {
streamMode: ["updates", "custom"],
})) {
console.log(chunk);
}高级用法
与任意 LLM 一起使用
你可以使用 stream_mode="custom" 从任何 LLM API 流式输出数据——即使该 API 并未实现 LangChain 对话模型接口。
这使你可以集成提供自有流式接口的原始 LLM 客户端或外部服务,从而让 LangGraph 对自定义设置具有极高的灵活性。
python
from langgraph.config import get_stream_writer
def call_arbitrary_model(state):
"""Example node that calls an arbitrary model and streams the output"""
# 获取流写入器以发送自定义数据
writer = get_stream_writer()
# 假设你有一个会产生数据块的流式客户端
# 使用你的自定义流式客户端生成 LLM token
for chunk in your_custom_streaming_client(state["topic"]):
# 使用 writer 向流中发送自定义数据
writer({"custom_llm_chunk": chunk})
return {"result": "completed"}
graph = (
StateGraph(State)
.add_node(call_arbitrary_model)
# 根据需要添加其他节点和边
.compile()
)
# 设置 stream_mode="custom" 以在流中接收自定义数据
for chunk in graph.stream(
{"topic": "cats"},
stream_mode="custom",
version="v2",
):
if chunk["type"] == "custom":
# chunk 数据将包含从 LLM 流式输出的自定义数据
print(chunk["data"])你可以使用 streamMode: "custom" 从任何 LLM API 流式输出数据——即使该 API 并未实现 LangChain 对话模型接口。
这使你可以集成提供自有流式接口的原始 LLM 客户端或外部服务,从而让 LangGraph 对自定义设置具有极高的灵活性。
typescript
import { StateGraph, GraphNode, StateSchema } from "@langchain/langgraph";
import * as z from "zod";
const State = new StateSchema({ result: z.string() });
const callArbitraryModel: GraphNode<typeof State> = async (state, config) => {
// 调用任意模型并流式输出其结果的示例节点
// 假设你有一个会产生数据块的流式客户端
// 使用你的自定义流式客户端生成 LLM token
for await (const chunk of yourCustomStreamingClient(state.topic)) {
// 使用 writer 向流中发送自定义数据
config.writer({ custom_llm_chunk: chunk });
}
return { result: "completed" };
};
const graph = new StateGraph(State)
.addNode("callArbitraryModel", callArbitraryModel)
// 根据需要添加其他节点和边
.compile();
// 设置 streamMode: "custom" 以在流中接收自定义数据
for await (const chunk of await graph.stream(
{ topic: "cats" },
{ streamMode: "custom" }
)) {
// chunk 将包含从 LLM 流式输出的自定义数据
console.log(chunk);
}扩展示例:流式输出任意对话模型
python
import operator
import json
from typing import TypedDict
from typing_extensions import Annotated
from langgraph.graph import StateGraph, START
from openai import AsyncOpenAI
openai_client = AsyncOpenAI()
model_name = "gpt-5.4-mini"
async def stream_tokens(model_name: str, messages: list[dict]):
response = await openai_client.chat.completions.create(
messages=messages, model=model_name, stream=True
)
role = None
async for chunk in response:
delta = chunk.choices[0].delta
if delta.role is not None:
role = delta.role
if delta.content:
yield {"role": role, "content": delta.content}
# 这是我们的工具
async def get_items(place: str) -> str:
"""Use this tool to list items one might find in a place you're asked about."""
writer = get_stream_writer()
response = ""
async for msg_chunk in stream_tokens(
model_name,
[
{
"role": "user",
"content": (
"Can you tell me what kind of items "
f"i might find in the following place: '{place}'. "
"List at least 3 such items separating them by a comma. "
"And include a brief description of each item."
),
}
],
):
response += msg_chunk["content"]
writer(msg_chunk)
return response
class State(TypedDict):
messages: Annotated[list[dict], operator.add]
# 这是调用工具的图节点
async def call_tool(state: State):
ai_message = state["messages"][-1]
tool_call = ai_message["tool_calls"][-1]
function_name = tool_call["function"]["name"]
if function_name != "get_items":
raise ValueError(f"Tool {function_name} not supported")
function_arguments = tool_call["function"]["arguments"]
arguments = json.loads(function_arguments)
function_response = await get_items(**arguments)
tool_message = {
"tool_call_id": tool_call["id"],
"role": "tool",
"name": function_name,
"content": function_response,
}
return {"messages": [tool_message]}
graph = (
StateGraph(State)
.add_node(call_tool)
.add_edge(START, "call_tool")
.compile()
)让我们用一个包含工具调用的 AIMessage 来调用该图:
python
inputs = {
"messages": [
{
"content": None,
"role": "assistant",
"tool_calls": [
{
"id": "1",
"function": {
"arguments": '{"place":"bedroom"}',
"name": "get_items",
},
"type": "function",
}
],
}
]
}
async for chunk in graph.astream(
inputs,
stream_mode="custom",
version="v2",
):
if chunk["type"] == "custom":
print(chunk["data"]["content"], end="|", flush=True)typescript
import { StateGraph, StateSchema, MessagesValue, GraphNode, START, LangGraphRunnableConfig } from "@langchain/langgraph";
import { tool } from "@langchain/core/tools";
import * as z from "zod";
import OpenAI from "openai";
const openaiClient = new OpenAI();
const modelName = "gpt-5.4-mini";
async function* streamTokens(modelName: string, messages: any[]) {
const response = await openaiClient.chat.completions.create({
messages,
model: modelName,
stream: true,
});
let role: string | null = null;
for await (const chunk of response) {
const delta = chunk.choices[0]?.delta;
if (delta?.role) {
role = delta.role;
}
if (delta?.content) {
yield { role, content: delta.content };
}
}
}
// 这是我们的工具
const getItems = tool(
async (input, config: LangGraphRunnableConfig) => {
let response = "";
for await (const msgChunk of streamTokens(
modelName,
[
{
role: "user",
content: `Can you tell me what kind of items i might find in the following place: '${input.place}'. List at least 3 such items separating them by a comma. And include a brief description of each item.`,
},
]
)) {
response += msgChunk.content;
config.writer?.(msgChunk);
}
return response;
},
{
name: "get_items",
description: "Use this tool to list items one might find in a place you're asked about.",
schema: z.object({
place: z.string().describe("The place to look up items for."),
}),
}
);
const State = new StateSchema({
messages: MessagesValue,
});
const callTool: GraphNode<typeof State> = async (state) => {
const aiMessage = state.messages.at(-1);
const toolCall = aiMessage.tool_calls?.at(-1);
const functionName = toolCall?.function?.name;
if (functionName !== "get_items") {
throw new Error(`Tool ${functionName} not supported`);
}
const functionArguments = toolCall?.function?.arguments;
const args = JSON.parse(functionArguments);
const functionResponse = await getItems.invoke(args);
const toolMessage = {
tool_call_id: toolCall.id,
role: "tool",
name: functionName,
content: functionResponse,
};
return { messages: [toolMessage] };
};
const graph = new StateGraph(State)
// 这是调用工具的图节点
.addNode("callTool", callTool)
.addEdge(START, "callTool")
.compile();让我们用一个包含工具调用的 AIMessage 来调用该图:
typescript
const inputs = {
messages: [
{
content: null,
role: "assistant",
tool_calls: [
{
id: "1",
function: {
arguments: '{"place":"bedroom"}',
name: "get_items",
},
type: "function",
}
],
}
]
};
for await (const chunk of await graph.stream(
inputs,
{ streamMode: "custom" }
)) {
console.log(chunk.content + "|");
}为特定对话模型禁用流式传输
如果你的应用混合使用了支持流式传输和不支持流式传输的模型,你可能需要为不支持流式传输的模型显式禁用流式传输。
在初始化模型时设置 streaming=False。
init_chat_model
python
from langchain.chat_models import init_chat_model
model = init_chat_model(
"claude-sonnet-4-6",
# 设置 streaming=False 以禁用对话模型的流式传输
streaming=False
)Chat model interface
python
from langchain_openai import ChatOpenAI
# 设置 streaming=False 以禁用对话模型的流式传输
model = ChatOpenAI(model="gpt-5.5", streaming=False)在初始化模型时设置 streaming: false。
typescript
import { ChatOpenAI } from "@langchain/openai";
const model = new ChatOpenAI({
model: "gpt-5.5",
// 设置 streaming: false 以禁用对话模型的流式传输
streaming: false,
});INFO
并非所有对话模型集成都支持 streaming 参数。如果你的模型不支持该参数,请改用 disable_streaming=True。该参数通过基类在所有对话模型上都可用。
INFO
并非所有对话模型集成都支持 streaming 参数。如果你的模型不支持该参数,请改用 disableStreaming: true。该参数通过基类在所有对话模型上都可用。
迁移到 v2
v2 流式格式(本页通篇使用)提供了统一的输出格式。以下是主要差异以及如何迁移的总结:
| Scenario | v1 (default) | v2 (version="v2") |
|---|---|---|
| 单一流模式 | 原始数据(dict) | 包含 type、ns、data 的 StreamPart 字典 |
| 多个流模式 | (mode, data) 元组 | 相同的 StreamPart 字典,根据 chunk["type"] 过滤 |
| 子图流式传输 | (namespace, data) 元组 | 相同的 StreamPart 字典,检查 chunk["ns"] |
| 多模式 + 子图 | (namespace, mode, data) 三元组 | 相同的 StreamPart 字典 |
invoke() 返回类型 | 普通字典(状态) | 包含 .value 和 .interrupts 的 GraphOutput |
| 中断位置(流) | 状态字典中的 __interrupt__ 键 | values 流部分上的 interrupts 字段 |
| 中断位置(invoke) | 结果字典中的 __interrupt__ 键 | GraphOutput 上的 .interrupts 属性 |
| Pydantic/数据类输出 | 返回普通字典 | 强制转换为模型/数据类实例 |
v2 invoke 格式
当你向 invoke() 或 ainvoke() 传入 version="v2" 时,它会返回一个带有 .value 和 .interrupts 属性的 GraphOutput 对象:
python
from langgraph.types import GraphOutput
result = graph.invoke(inputs, version="v2")
assert isinstance(result, GraphOutput)
result.value # 你的输出——dict、Pydantic 模型或数据类
result.interrupts # tuple[Interrupt, ...],若未发生则为空对于默认 "values" 以外的任何流模式,invoke(..., stream_mode="updates", version="v2") 会返回 list[StreamPart] 而不是 list[tuple]。
WARNING
在 GraphOutput 上以字典方式访问(result["key"]、"key" in result、result["__interrupt__"])出于向后兼容仍然有效,但已弃用,将在未来版本中移除。请迁移到 result.value 和 result.interrupts。
这会将状态与中断元数据分离。在 v1 中,中断被嵌入返回的字典的 __interrupt__ 键下:
python
config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config, version="v2")
if result.interrupts:
print(result.interrupts[0].value)
graph.invoke(Command(resume=True), config=config, version="v2")python
config = {"configurable": {"thread_id": "thread-1"}}
result = graph.invoke(inputs, config=config)
if "__interrupt__" in result:
print(result["__interrupt__"][0].value)
graph.invoke(Command(resume=True), config=config)Pydantic 与数据类状态强制转换
当你的图状态是 Pydantic 模型或数据类时,v2 的 values 模式会自动将输出强制转换为正确的类型:
python
from pydantic import BaseModel
from typing import Annotated
import operator
class MyState(BaseModel):
value: str
items: Annotated[list[str], operator.add]
# 使用 version="v2" 时,chunk["data"] 是 MyState 实例
for chunk in graph.stream(
{"value": "x", "items": []}, stream_mode="values", version="v2"
):
print(type(chunk["data"])) # <class 'MyState'>Python < 3.11 下的异步
在 Python < 3.11 版本中,asyncio 任务不支持 context 参数。 这限制了 LangGraph 自动传播上下文的能力,并从以下两个关键方面影响 LangGraph 的流式机制:
- 你必须在异步 LLM 调用(例如
ainvoke())中显式传入RunnableConfig,因为回调不会自动传播。 - 你不能在异步节点或工具中使用
get_stream_writer——你必须直接传入writer参数。
扩展示例:手动配置的异步 LLM 调用
python
from typing import TypedDict
from langgraph.graph import START, StateGraph
from langchain.chat_models import init_chat_model
model = init_chat_model(model="gpt-5.4-mini")
class State(TypedDict):
topic: str
joke: str
# 在异步节点函数中接受 config 作为参数
async def call_model(state, config):
topic = state["topic"]
print("Generating joke...")
# 将 config 传给 model.ainvoke() 以确保正确的上下文传播
joke_response = await model.ainvoke(
[{"role": "user", "content": f"Write a joke about {topic}"}],
config,
)
return {"joke": joke_response.content}
graph = (
StateGraph(State)
.add_node(call_model)
.add_edge(START, "call_model")
.compile()
)
# 设置 stream_mode="messages" 以流式输出 LLM token
async for chunk in graph.astream(
{"topic": "ice cream"},
stream_mode="messages",
version="v2",
):
if chunk["type"] == "messages":
message_chunk, metadata = chunk["data"]
if message_chunk.content:
print(message_chunk.content, end="|", flush=True)扩展示例:使用流写入器的异步自定义流式传输
python
from typing import TypedDict
from langgraph.types import StreamWriter
class State(TypedDict):
topic: str
joke: str
# 在异步节点或工具的函数签名中添加 writer 参数
# LangGraph 会自动将流写入器传给该函数
async def generate_joke(state: State, writer: StreamWriter):
writer({"custom_key": "Streaming custom data while generating a joke"})
return {"joke": f"This is a joke about {state['topic']}"}
graph = (
StateGraph(State)
.add_node(generate_joke)
.add_edge(START, "generate_joke")
.compile()
)
# 设置 stream_mode="custom" 以在流中接收自定义数据 #
async for chunk in graph.astream(
{"topic": "ice cream"},
stream_mode="custom",
version="v2",
):
if chunk["type"] == "custom":
print(chunk["data"])