🎯 课程主题

详解 Agent 流式输出的七种 stream_mode 模式，涵盖从基础增量更新到自定义业务输出的完整策略。

📝 核心知识点

1. Agent 流式输出概述

  • 概念说明：与普通模型调用类似，Agent 也支持通过 stream() 方法实现流式输出，实时展示 Agent 运行过程，降低用户等待焦虑感。
  • 关键细节：创建 Agent 后，使用 stream() 替代 invoke()，通过 stream_mode 参数指定输出模式。

2. 七种 stream_mode 详解

模式行为适用场景
values每步骤执行后输出完整状态快照（累积所有历史消息）需要查看每一步完整上下文的场景
updates（默认）每步骤仅输出增量变化，只展示新增加的内容监控 Agent 思考和执行步骤
messages以 Token 级别逐个输出 AIMessage chunk实现打字机效果、实时对话交互（类似 ChatGPT）
tasks输出任务的开始/结束时间、状态和错误信息监控任务生命周期
debug比 tasks 增加时间戳、步骤序号（第1步/第2步）等调试信息开发调试阶段排查问题
checkpoints每当 checkpoint 被创建时触发输出，包含状态快照需要状态持久化的场景（配合 Memory 使用）
custom开发者通过 get_stream_writer() 在工具/节点内部自定义发送数据输出业务逻辑相关的自定义进度信息

3. 实战选择总结

  • 实时对话交互 → messages
  • 观察思考步骤 → updates
  • 查看每步完整状态 → values / tasks / debug
  • 输出自定义业务进度 → custom

🏗️ 架构与工作流

Agent.stream(input, stream_mode="xxx")
    │
    ├── values:    [msg1] → [msg1, msg2] → [msg1, msg2, msg3]  (全量累积)
    ├── updates:   [msg1] → [msg2]        → [msg3]             (仅增量)
    ├── messages:  [token1] → [token2] → ...                    (Token级)
    ├── tasks:     {id, started_at, ended_at, ...}
    ├── debug:     {step:1, timestamp, ...} → {step:2, ...}
    ├── checkpoints: {checkpoint_id, state, ...}
    └── custom:    通过 writer.write() 自定义输出

💻 代码实战

from langchain_deepseek import ChatDeepSeek
from langgraph.prebuilt import create_react_agent

model = ChatDeepSeek(model="deepseek-v4-flash")

# ==================== 定义工具 ====================
def get_customer_info(customer_id: str) -> str:
    """根据客户ID查询客户数据"""
    return f"客户ID {customer_id}: 姓名张三，VIP等级Gold"

def get_order_history(customer_id: str) -> str:
    """根据客户ID查询订单历史"""
    return f"客户ID {customer_id}: 最近订单3笔，总金额￥5800"

def get_promotions() -> str:
    """查询当前促销活动"""
    return "当前活动：满300减50，新用户首单8折"

tools = [get_customer_info, get_order_history, get_promotions]

# ==================== 创建 Agent ====================
agent = create_react_agent(model=model, tools=tools)

user_query = "查询客户ID为1234的用户的完整信息，包括历史订单和可用的优惠"

# === 模式一：values（每步输出完整状态）===
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": user_query}]},
    stream_mode="values"
):
    print(chunk)
    print("-" * 50)

# === 模式二：updates（默认，仅输出增量变化）===
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": user_query}]},
    stream_mode="updates"
):
    print(chunk)
    print("-" * 50)

# === 模式三：messages（Token级别输出，打字机效果）===
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": user_query}]},
    stream_mode="messages"
):
    # chunk 是 (AIMessageChunk, metadata) 元组，取第一个元素的content
    if chunk[0].content:
        print(chunk[0].content, end="", flush=True)

# === 模式四：tasks ===
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": user_query}]},
    stream_mode="tasks"
):
    print(chunk)

# === 模式五：debug ===
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": user_query}]},
    stream_mode="debug"
):
    print(chunk)

# === 模式六：checkpoints（需配合 MemorySaver）===
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()
agent_with_checkpointer = create_react_agent(
    model=model,
    tools=tools,
    checkpointer=checkpointer
)
for chunk in agent_with_checkpointer.stream(
    {"messages": [{"role": "user", "content": user_query}]},
    stream_mode="checkpoints",
    config={"configurable": {"thread_id": "session-1"}}
):
    print(chunk)

# === 模式七：custom（自定义业务输出）===
from langgraph.config import get_stream_writer

def generate_sales_report() -> str:
    """生成销售报告（带自定义流式输出）"""
    writer = get_stream_writer()
    writer.write("开始生成销售报告...")
    writer.write("分析销售数据中... 25%")
    writer.write("分析销售数据中... 50%")
    writer.write("分析销售数据中... 75%")
    writer.write("报告生成完成!")
    return "销售报告：本季度营收￥120万，同比增长15%"

def generate_inventory_report() -> str:
    """生成库存报告（带自定义流式输出）"""
    writer = get_stream_writer()
    writer.write("开始库存分析...")
    writer.write("库存检查中...")
    writer.write("生成库存报告完成!")
    return "库存报告：当前库存充足，周转率正常"

custom_tools = [generate_sales_report, generate_inventory_report]
agent_custom = create_react_agent(model=model, tools=custom_tools)

for chunk in agent_custom.stream(
    {"messages": [{"role": "user", "content": "生成销售报告和库存报告"}]},
    stream_mode="custom"
):
    print(chunk)

⚠️ 常见问题与避坑指南

  • messages 模式输出极其冗长（每个 token 一个 chunk），直接在 Notebook 打印会卡顿，建议只提取 chunk[0].content 做打字机效果。
  • 默认模式是 updates，不指定 stream_mode 时即为此模式。
  • stream_mode 支持传入列表（如 ["updates", "messages"]），可同时享受多种模式的组合效果。
  • checkpoints 模式需要配合 MemorySaver 或持久化存储器使用，否则无法触发检查点事件。
  • custom 模式要求工具内部调用 get_stream_writer() 获取 writer 实例并调用 write() 方法。

💡 个人总结与延伸

七种 stream_mode 覆盖了从调试到生产的全场景需求：日常开发用 updates 追踪步骤，产品上线用 messages 实现打字机效果提升体验，复杂业务用 custom 输出自定义进度。checkpoints 模式是后续实现对话记忆和状态持久化的关键入口，值得深入学习。

最近更新:
Contributors: yangwenguang