前言

在构建 AI Agent 时,流式输出(Streaming)是提升用户体验的关键。它能让用户在模型思考或调用工具时实时看到进展,而不是盯着空白屏幕等待最终结果。

在 LangChain/LangGraph 的 agent.stream 方法中,最常用的两种模式是 valuesmessages。它们的应用场景和数据结构大不相同,让我们通过实战代码来拆解。

我们代码整理还是使用上一集中的代码,本集只放关键改动代码。

模式一

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 包含两个部分:

  1. AIMessageChunk: 包含当前的文本片段(如 content="I'll")。

  2. 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%...”)。

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐