【由浅入深探究langchain】第十集-Agent的流式输出
前言
在构建 AI Agent 时,流式输出(Streaming)是提升用户体验的关键。它能让用户在模型思考或调用工具时实时看到进展,而不是盯着空白屏幕等待最终结果。
在 LangChain/LangGraph 的 agent.stream 方法中,最常用的两种模式是 values 和 messages。它们的应用场景和数据结构大不相同,让我们通过实战代码来拆解。
我们代码整理还是使用上一集中的代码,本集只放关键改动代码。
模式一
stream_mode="values" —— 获取状态全量快照
values 模式可以理解为 “状态快照”。每当 Agent 完成一个步骤(比如 LLM 决定调用工具、工具返回结果、LLM 生成最终回答),它都会返回当前状态中所有的消息列表。
for event in agent.stream(
{"messages": [{"role": "user", "content": "What's the price of gold now?"}]},
stream_mode="values"
):
messages = event["messages"]
print(f"--- 当前历史消息数量: {len(messages)} 条 ---")
# 打印当前步骤产生的新消息
for message in messages:
message.pretty_print()
运行结果:

输出逻辑拆解
当我们询问金价时,控制台会分多次打印:
-
Step 1: 返回包含
Human Message的列表(长度 1)。 -
Step 2: 返回包含
Human Message+AI Message (Tool Call)的列表(长度 2)。 -
Step 3: 返回包含前两条 +
Tool Message (结果)的列表(长度 3)。 -
Step 4: 返回包含所有消息 +
AI Message (最终回答)的列表(长度 4)。
小技巧: 在 values 模式下,由于每次迭代都会返回整个消息历史,如果我们遍历整个列表打印,会导致前面的内容被重复打印。因此,通常我们只打印 messages[-1],即当前步骤最新产生的消息。如下所示:
# for message in messages:
# message.pretty_print()
messages[-1].pretty_print()
模式二
stream_mode="messages" —— 获取 Token 级实时流
如果你想实现像 ChatGPT 那样“一个字一个字”蹦出来的效果,messages 模式是你的首选。它返回的是 消息碎片(Chunks),能够实时获取模型生成的每一个 Token。
for chunk in agent.stream(
{"messages":[{"role":"user","content":"What's the price of gold now?"}]},
stream_mode="messages" #token by token
):
print(chunk)
结果展示:

在此模式下,每一个 chunk 包含两个部分:
-
AIMessageChunk: 包含当前的文本片段(如
content="I'll")。 -
Metadata: 包含 LangGraph 的节点信息(如当前处于
model节点还是tools节点)。
根据结构,我们修改打印语句
# print(chunk)
print(chunk[0].content,end="")
运行打印结果如下:

可以看到这句话里面,I'll check the current price of gold for you.是Ai Message的,gold's price is $2000是Tool Message,The current price of gold is $2000.是Ai Message的。
说明它不仅流式输出 AI 的话语,连工具返回的结果和最终总结都会以 Token 或碎片的形式实时推送到前端。
总结
| 特性 | stream_mode="values" | stream_mode="messages" |
| 返回内容 | 完整的状态/消息列表 | 单个消息碎片 (Chunks/Tokens) |
| 颗粒度 | 步骤级 (Node level) | 字符级 (Token level) |
| 典型用途 | 监控 Agent 运行步骤、调试逻辑 | 实现打字机效果、实时 UI 更新 |
| 打印策略 | 打印 messages[-1] 防止重复 |
使用 end="" 实时拼接输出 |
-
如果你需要追踪 Agent 走到了哪一步,请使用
values。 -
如果你需要极致的响应速度体验,请使用
messages。
除了这两种常见模式,还有其他的模式做一个简单介绍
1. stream_mode="updates" —— 局部增量模式
这是开发中最常用的调试模式。
-
核心逻辑:它不返回整个消息列表,也不返回 Token,而是只返回当前节点(Node)执行完后产生的增量变化。
-
适用场景:你想知道 Agent 内部每一步具体改动了什么,或者想监控特定的 State 变量。
2. stream_mode="debug" —— 全量调试模式
这是最“话痨”的模式,适合你在代码跑不通、找不到 Bug 时开启。
-
核心逻辑:它会把每一层级的详细信息都吐出来,包括输入、输出、配置、Checkpoint 信息等。
-
适用场景:深入排查 LangGraph 内部执行逻辑、检查状态回滚(Time Travel)等高级功能。
3. stream_mode="custom" —— 自定义触发模式
如果你在编写自定义 Node 时,使用了 list(StreamWriter(...)),这个模式就能排上用场。
-
核心逻辑:你可以手动控制什么时候发送流数据,发送什么内容。
-
适用场景:在长耗时的逻辑内部,手动向前端推送进度条(例如:“正在查询数据库 20%...”)。
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。
更多推荐


所有评论(0)