《实用笔记》：LangGraph中的流式响应处理——每一种场景都适用的3种实用模式。
《实用笔记》操作指南：LangGraph中的流式响应处理——每个团队都应掌握的3种实用模式：合约、校验以及适用于该模式的即插即用代码模板。
本指南将逐步构建从原始材料到可运行系统的完整流程，主题为“LangGraph中的流式响应处理：每位智能体开发者都应掌握的3种实用模式”。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需推测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的功能。
为何流式处理比人们想象中更重要
在处理“为何流式处理更为重要”这一阶段时，首先需列出相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议使用小型、可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的处理流程。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
我们正在研究的示例
在处理“我们分阶段进行的示例”时，首先写下契约：所需的输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，并拒绝默许部分完成的情况。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
import asyncio
import operator
from typing import Annotated, TypedDict
from dotenv import load_dotenv
from langchain_core.messages import BaseMessage, HumanMessage
from langchain_openai import ChatOpenAI
from langgraph.graph import END, START, StateGraph
load_dotenv()
# --- 1. STATE & GRAPH ---
class State(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
llm = ChatOpenAI(model="gpt-4o-mini", streaming=True)
def chatbot_node(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
def dummy_node(state: State) -> State:
return state
builder = StateGraph(State)
builder.add_node("chatbot", chatbot_node)
builder.add_node("dummy", dummy_node)
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", "dummy")
builder.add_edge("dummy", END)
graph = builder.compile()
# --- 2A. stream_mode="updates" — one event per node ---
print("=== Method 1: stream_mode='updates' ===")
for event in graph.stream(
{
"messages": [HumanMessage("List 3 benefits of LangGraph in one line each")],
},
stream_mode="updates",
):
for node_name, output in event.items():
print(f" [{node_name}] {output['messages']}")
# --- 2B. stream_mode="values" — full state after each node ---
print("\n=== Method 2: stream_mode='values' ===")
for snapshot in graph.stream(
{
"messages": [HumanMessage("Say hello in 3 languages")],
},
stream_mode="values",
):
print(f" State has {len(snapshot['messages'])} message(s) now")
print(snapshot["messages"])
# --- 2C. astream_events — token-by-token (async) ---
async def token_stream():
print("\n=== Method 3: Token-by-token streaming ===")
print("🤖 Bot: ", end="", flush=True)
async for event in graph.astream_events(
{
"messages": [HumanMessage("Count from 1 to 5 slowly, one per line")],
},
version="v2",
):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
if chunk:
print(chunk, end="", flush=True)
print()
asyncio.run(token_stream())
步骤1：首先理解状态设计
在完成“第一步：理解需求”阶段时，首先需写下相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前明确成本信息，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。 在完成“第一步：理解需求”阶段时，首先需写下相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。
class State(TypedDict):
messages: Annotated[list[BaseMessage], operator.add]
{"messages": [some_new_message]}
为何这对流媒体服务至关重要
将“为何这对流媒体服务至关重要”视为一个可度量的指标最为有效。在扩大范围之前，先记录一份优秀的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应能指向单一责任主体，而非复杂的流程链。 保持图结构的层次简单且类型明确。嵌套的数据结构会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致无法继续处理。
第二步：图结构本身刻意保持简单
将“第二步：图表”阶段视为可测量的表面来处理效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的、不完整的处理结果。 保持图表的状态简洁且具有明确类型。嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在处理中断后导致无法继续。
chatbot_node
将聊天机器人节点阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想的对话文本、一个故障案例以及回滚说明。 在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 保持图结构简洁且类型明确。嵌套的数据块会掩盖具体是哪个节点修改了哪一字段，还会在进程中断后导致无法继续运行。 将聊天机器人节点阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，需记录一份理想的对话文本、一个故障案例以及回滚说明。 需同时记录正常流程与恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。
def chatbot_node(state: State) -> dict:
return {"messages": [llm.invoke(state["messages"])]}
dummy_node
在 dumbynode 阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。
def dummy_node(state: State) -> State:
return state
START → chatbot → dummy → END
方法 1：stream_mode="updates"
在方法1的流式更新阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
for event in graph.stream(..., stream_mode="updates"):
其功能
在“功能说明”阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本可避免在流程从演示环境转向共享环境时出现意外费用。 对于会消耗资金或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。 在“功能说明”阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续需要补充的内容。
{
"chatbot": {
"messages": [...]
}
}
{
"dummy": {
"messages": [...]
}
}
为何这种模式有用
在处理“为何选择此模式”这一阶段时，首先需明确相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
最佳应用场景
在处理“最佳用例”阶段时，首先需写下相关契约：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，明确成功判定标准，杜绝无声的半完成状态。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
实际案例
在处理实际应用案例阶段时，首先需明确相关规范：所需的输入参数、成功标志，以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在记录功能结果的同时，还需标注执行时间以及Token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。 在处理实际应用案例阶段时，首先需明确相关规范：所需的输入参数、成功标志，以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。
方法2：stream_mode="values"
方法2中的流模式值阶段在被视为可测量的表面时效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一的责任主体，而非复杂的流程链。 保持图表状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致无法继续处理。
for snapshot in graph.stream(..., stream_mode="values"):
其功能
“功能实现”阶段若被视为可度量的对象，效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的、不完整的处理。 保持图结构的状态简洁且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续处理。
print(f" State has {len(snapshot['messages'])} message(s) now")
print(snapshot["messages"])
为何这很重要
“为何这很重要”这一阶段若被视为可度量的对象，效果会最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 在功能结果旁同时记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。 保持图表状态简洁且具有明确类型。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在出现中断后导致流程无法继续。 “为何这很重要”这一阶段若被视为可度量的对象，效果会最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 需同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品的一部分，而非后续需要补充的内容。
最佳应用场景
在“最佳用例”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。相比冗长的脚本，更应优先使用小型、可测试的单元。当某个步骤失败时，故障原因应能指向单一责任点，而非复杂的流程链。对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。
实际案例
在真实场景示例阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
方法 3：使用 astream_events() 进行逐令牌流式处理
对于方法3中的阶段流式事件，在修改代码之前需明确输入参数、该步骤的负责人以及退出条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 在功能结果旁记录执行时间以及令牌或查询成本。提前显示成本可避免在流程从演示环境切换到共享环境时出现意外费用。 当下一步操作为代码编写或工具调用时，应优先使用具有架构验证的结构化输出，而非自由形式的文本。 对于方法3中的阶段流式事件，在修改代码之前需明确输入参数、该步骤的负责人以及退出条件。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。
async for event in graph.astream_events(..., version="v2"):
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
其功能
在处理“其功能”这一阶段时，首先需列出相关约定：所需的输入参数、成功信号以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 建议采用小型、可测试的单元而非庞大的脚本。当某个步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
为何这很重要
在“为何这很重要”这一阶段，首先写下契约：所需的输入、成功信号以及部分失败时会发生什么。这样的清单能确保后续的代码修改保持一致性。 将这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，并拒绝默许部分完成的情况。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。
为何 astream_events() 是异步的
在处理“Why astreamevents 是异步的”这一阶段时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在系统从演示环境切换到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。 在处理“Why astreamevents 是异步的”这一阶段时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程与故障恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。
asyncio.run(token_stream())
理解事件过滤器
将“理解事件过滤器”这一阶段视为可度量的对象来处理，效果最佳。在扩大范围之前，先收集一份典型的成功案例、一个故障实例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤出现故障时，故障应指向单一的责任模块，而非复杂的流程链。 保持图结构的层次清晰且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在流程中断后导致无法继续执行。
if event["event"] == "on_chat_model_stream":
chunk = event["data"]["chunk"].content
用通俗语言解释三种流式处理模式
将“三种流式模式”阶段视为可测量的界面最为有效。在扩大范围之前，先记录一个成功案例、一个失败案例以及回滚说明。 把这一阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许默默地仅完成部分工作。 保持图表状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致无法继续处理。
updates
将更新阶段视为可度量的对象来处理时效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 保持图结构扁平且类型明确。嵌套的数据块会掩盖哪个节点修改了哪个字段，还会在中断后导致流程无法继续。 将更新阶段视为可度量的对象来处理时效果最佳。在扩大范围之前，先记录一份理想的操作日志、一个故障案例以及回滚说明。 需同时记录正常流程与恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的内容。
values
在值处理阶段，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的逻辑连接并不等同于业务功能的完整性。
astream_events
在流式事件阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许的半完成状态。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务上的完整性。
应选择哪种流式模式？
对于“应选择哪种流式处理模式”这一环节，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 在功能结果之外还需记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在流程从演示环境转向共享环境时出现意外费用。 对于那些会产生费用或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。 对于“应选择哪种流式处理模式”这一环节，在修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测其中的隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。
。何时使用 updates：
在处理“何时使用 updates 阶段”时，首先需明确合同规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应指向单一责任点，而非复杂的流程链。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大语言模型接口。
何时使用 values：
在处理“阶段中使用值”时，首先写下契约：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将此阶段视为输入与已验证输出之间的契约。为相关成果命名，定义成功检查标准，并拒绝默许的部分完成。 在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大语言模型。
在以下情况使用 astream_events：
在处理“在阶段中使用astreamevents”时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。 在处理“在阶段中使用astreamevents”时，首先需明确相关规范：所需输入、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 需同时记录正常流程与异常恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续需要补充的内容。
更优的思维模型：面向用户的流式处理与面向开发者的流式处理
将“更优的思维模型”阶段视为可度量的分析对象时效果最佳。在扩大范围之前，先记录一份典型的成功案例、一个失败案例以及回滚说明。
面向用户的流式处理
面向开发者的流式处理
你可能需要的生产级模式
流式传输LangGraph响应时的常见错误
1. 实际需要节点更新却使用令牌流式处理
2>误以为values的行为与令牌流式处理相同
3>忘记astream_events()是异步操作
4>未对事件类型进行过滤
5>构建了可流式传输但无法暴露有意义状态的图结构
高级技巧：在多步骤智能体图中，流式处理的功能更为强大
updates
values
astream_events
一个可做的微小改进
for event in graph.stream(
{
"messages": [HumanMessage("List 3 benefits of LangGraph in one line each")],
},
stream_mode="updates",
):
for node_name, output in event.items():
print(f"\nNode: {node_name}")
for message in output["messages"]:
print(message)