values：每步的完整状态#

1
for chunk in customer_service_agent.stream(QUESTION, stream_mode="values"):
2
    rprint(chunk)
3
    print("-" * 50)

图：values 模式的输出——每一片都是”到目前为止的完整状态”，消息列表一片比一片长（[HumanMessage] → [HumanMessage, AIMessage] → … → 最后一条是最终回答）

适合每一步都要完整状态的场景（每片都是”到目前为止的全量状态”，消息列表会越来越长）。

updates：只看变化（默认）#

1
for chunk in customer_service_agent.stream(QUESTION, stream_mode="updates"):
2
    rprint(chunk)
3
    print("-" * 50)

图：updates 模式的输出——每一片只带 model 或 tools 这一个节点的增量（模型决定调工具 / 工具返回了什么），比 values 清爽得多

不传 stream_mode 时就是它。每片只包含这一步新增/变化的内容——想观察”Agent 决定调用哪个工具、工具返回了什么”，用这个最清爽。

messages：打字机效果#

1
for chunk in customer_service_agent.stream(QUESTION, stream_mode="messages"):
2
    # chunk 是元组：(消息片段, 元数据)
3
    print(chunk[0].content, end="", flush=True)

图：messages 模式的输出——一片片 AIMessageChunk，content 就是”我""来""帮""您""查""询”这样一个字一个字冒出来的 token

tasks / debug：监控任务生命周期#

1
for chunk in customer_service_agent.stream(QUESTION, stream_mode="tasks"):
2
    rprint(chunk)
3
    print("-" * 50)

图：tasks 模式的输出——每个 task 都带 id、name（model / tools）、input、error、result，model 和 tools 交替出现，一眼看清任务生命周期

tasks 输出任务的开始/结束时间、结果与错误；debug 在它基础上多出任务步骤、时间戳、task 类型（task / task_result）。排查”Agent 卡在哪一步”时很好用。

图：debug 模式的输出——比 tasks 多出 step（第几步）、timestamp、type（task 开始 / task_result 结束），而且开始和结束是成对出现的

checkpoints：需要先开检查点#

1
from langgraph.checkpoint.memory import InMemorySaver
2

3
checkpointer = InMemorySaver()
4

5
customer_service_agent = create_agent(
6
    model=model,
7
    tools=[query_customer_data, check_order_history, get_current_promotions],
8
    checkpointer=checkpointer,        # ← 启用检查点
9
)
10

11
config = {"configurable": {"thread_id": "session01"}}     # ← 唯一会话 ID
12

13
checkpoint_count = 0
14
for chunk in customer_service_agent.stream(
15
    {"messages": [{"role": "user", "content": "查询客户ID为 CUST123456 的完整信息和可用优惠"}]},
16
    config=config,
17
    stream_mode="checkpoints",
18
):
19
    checkpoint_count += 1
20
    print(f"检查点 #{checkpoint_count}")
21
    print(chunk)

图：checkpoints 模式的输出——检查点 #1 到 #7 依次触发，每片都带 checkpoint_id、parent_config、source（input / loop）、next（下一个节点）、tasks 等状态信息

custom：在工具内部自定义进度#

工具执行过程中想往外面”报进度”，用 get_stream_writer：

1
from langgraph.config import get_stream_writer
2
from langchain.tools import tool
3
import time
4

5
@tool
6
def generate_sales_report() -> str:
7
    """生成销售报告"""
8
    writer = get_stream_writer()
9
    writer({"type": "生成销售报告", "message": "开始生成销售报告"})
10

11
    for i in range(1, 4):                      # 模拟数据处理
12
        time.sleep(0.5)
13
        writer({"type": "生成销售报告", "message": f"生成销售报告进度百分比：{i * 25}%"})
14

15
    writer({"type": "生成销售报告", "message": "报告生成完成"})
16
    return "销售报告：总收入150万元，同比增长12%"
17

18
reporting_agent = create_agent(model=model, tools=[generate_sales_report])
19

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

工具里 writer(...) 发出去的东西，就会从这个流里冒出来——长耗时任务给用户实时进度条就靠它。

1
from langchain.agents import create_agent
2
from langgraph.config import get_stream_writer
3
from langchain.tools import tool
4
import time
5

6
@tool
7
def generate_sales_report() -> str:
8
    """生成销售报告"""
9
    writer = get_stream_writer()
10

11
    writer({"type": "生成销售报告", "message": "开始生成销售报告"})
12

13
    # 模拟数据处理
14
    for i in range(1, 4):
15
        time.sleep(0.5)
16
        writer({"type": "生成销售报告", "message": f"生成销售报告进度百分比：{i * 25}%"})
17

18
    writer({"type": "生成销售报告", "message": "报告生成完成"})
19

20
    return f"销售报告：总收入150万元，同比增长12%"
21

22

23
@tool
24
def generate_inventory_report() -> str:
25
    """生成库存报告"""
26
    writer = get_stream_writer()
27
    writer("开始库存分析...")          # ← 这里直接发字符串，不一定是字典
28
    time.sleep(0.5)
29
    writer("检查当前库存量...")
30
    time.sleep(0.5)
31
    writer("生成库存报告...")
32

33
    return "当前库存量为10000件，库存充足，无异常"
34

35
# 创建报告生成 agent
36
reporting_agent = create_agent(model=model, tools=[generate_sales_report, generate_inventory_report])
37

38
for chunk in reporting_agent.stream(
39
    {"messages": [{"role": "user", "content": "生成销售报告和库存报告"}]},
40
    stream_mode="custom",
41
):
42
    print(chunk)
43
    print("-" * 50)

1
{'type': '生成销售报告', 'message': '开始生成销售报告'}
2
--------------------------------------------------
3
开始库存分析...
4
--------------------------------------------------
5
{'type': '生成销售报告', 'message': '生成销售报告进度百分比：25%'}
6
--------------------------------------------------
7
检查当前库存量...
8
--------------------------------------------------
9
{'type': '生成销售报告', 'message': '生成销售报告进度百分比：50%'}
10
--------------------------------------------------
11
生成库存报告...
12
--------------------------------------------------
13
{'type': '生成销售报告', 'message': '生成销售报告进度百分比：75%'}
14
--------------------------------------------------
15
{'type': '生成销售报告', 'message': '报告生成完成'}
16
--------------------------------------------------

一、回忆填空（写完再展开对答案）#

1. 流式调用能解决 invoke 的____体验问题；设置方式是 agent.____(stream_mode=...)
2. values 模式每个步骤输出____状态；updates 模式只输出____的内容，它也是____模式
3. messages 模式输出流式 Token，用于实现____效果；它的 chunk 是____元组，chunk[____] 是消息片段
4. messages 模式元数据里的 ____ 字段能告诉你这段输出来自哪个节点（model / tools）
5. tasks 模式输出任务的____时间、结果与错误；____ 模式比它多输出任务步骤、时间戳、task 类型
6. checkpoints 模式需要配合 ____ 参数使用，调用时还要传 config={"configurable": {"thread_id": ...}}
7. custom 模式里，在工具内部用 ____() 拿到 writer，把自定义数据发到流里，适合输出____进度
8. 模式可以组合：传____即可，此时循环要写成 for ____, chunk in ...

二、裸写题#

• 2-1 用 updates 模式看 Agent 的每一步 定义两个工具（查库存、查价格），创建 Agent，用 stream_mode="updates" 问一个需要两个工具的问题，把每一步的 chunk 打印出来，观察哪一步是”模型决定调工具”、哪一步是”工具返回结果”。

  提示（先自己想，实在想不出再点开）

  一级 · 思路：updates 是”只报变化”，所以每片很轻 二级 · 方法：for chunk in agent.stream({...}, stream_mode="updates") 三级 · 骨架：打印时能看到 chunk 里的键（节点名），据此判断是哪一步

• 2-2 实现打字机效果 用 stream_mode="messages" 重跑 2-1 的问题，用 print(chunk[0].content, end="", flush=True) 把回答”一个字一个字”打出来；再顺便打印 chunk[1]["langgraph_node"]，看看 ToolMessage 是从哪个节点来的。

  提示

  一级 · 思路：messages 模式的每片是一小段 token，不是完整消息 二级 · 方法：chunk[0].content + chunk[1] 元数据 三级 · 骨架：end="" 和 flush=True 两个参数别丢，否则看不清”渐进”效果

• 2-3 给工具加进度上报 写一个”生成报表”工具，内部用 get_stream_writer() 每隔一段时间发一条进度（如 25% / 50% / 75%），用 stream_mode="custom" 消费这个流，把进度打印出来。

  提示

  一级 · 思路：进度是”工具自己发的”，不是模型发的 二级 · 方法：from langgraph.config import get_stream_writer，在工具里 writer({"message": ...}) 三级 · 骨架：配合 time.sleep(0.5) 模拟耗时，才能看出”实时”的效果

1
import os
2
import time
3

4
from dotenv import load_dotenv
5
from langchain.agents import create_agent
6
from langchain.chat_models import init_chat_model
7
from langchain.tools import tool
8
from langgraph.config import get_stream_writer
9

10
load_dotenv(override=True)
11

12
model = init_chat_model(
13
    model="deepseek-v4-flash",
14
    model_provider="openai",
15
    api_key=os.getenv("DEEPSEEK_API_KEY"),
16
    base_url=os.getenv("DEEPSEEK_BASE_URL"),
17
)
18

19
# ---------- 2-1 / 2-2 的工具 ----------
20
@tool
21
def check_stock(product: str) -> str:
22
    """查询库存
23

24
    Args:
25
        product: 商品名称
26
    """
27
    return f"{product} 当前库存 120 件"
28

29
@tool
30
def check_price(product: str) -> str:
31
    """查询价格
32

33
    Args:
34
        product: 商品名称
35
    """
36
    return f"{product} 售价 2999 元"
37

38
agent = create_agent(model=model, tools=[check_stock, check_price])
39
question = {"messages": [{"role": "user", "content": "蓝牙耳机的库存和价格分别是多少？"}]}
40

41
# 2-1 updates 模式
42
for chunk in agent.stream(question, stream_mode="updates"):
43
    print(chunk)
44
    print("-" * 50)
45

46
# 2-2 messages 模式（打字机）
47
for chunk in agent.stream(question, stream_mode="messages"):
48
    print(chunk[0].content, end="", flush=True)
49
    # 想看来源节点：
50
    # print(chunk[1].get("langgraph_node"), end=" ")
51

52
# ---------- 2-3 custom 模式 ----------
53
@tool
54
def generate_report() -> str:
55
    """生成销售报表"""
56
    writer = get_stream_writer()
57
    writer("开始生成报表……")
58
    for i in range(1, 4):
59
        time.sleep(0.5)
60
        writer(f"进度：{i * 25}%")
61
    writer("报表生成完成")
62
    return "报表：本月销售额 150 万元，同比 +12%"
63

64
report_agent = create_agent(model=model, tools=[generate_report])
65
for chunk in report_agent.stream(
66
    {"messages": [{"role": "user", "content": "生成一份销售报表"}]},
67
    stream_mode="custom",
68
):
69
    print(chunk)
70
    print("-" * 50)