W5-D4 · 观察一张图：四种流式看法

1
2

for event in graph.stream(TEST_INPUT, stream_mode="updates"):
    print(event)

1
2
3
4
5

{'agent': {'messages': [AIMessage(tool_calls=[add(15, 27)])]}}     # agent 决策要调 add
{'tools': {'messages': [ToolMessage(content='42')]}}               # tools 执行并回填
{'agent': {'messages': [AIMessage(tool_calls=[multiply(42, 2)])]}} # agent 看到 42，再调 multiply
{'tools': {'messages': [ToolMessage(content='84')]}}               # tools 再回填
{'agent': {'messages': [AIMessage(content='(15+27)×2 = 84')]}}     # agent 给最终答案

1
2

for i, snap in enumerate(graph.stream(TEST_INPUT, stream_mode="values")):
    print(f"第{i}步 消息数={len(snap['messages'])}  最后一条={snap['messages'][-1].__class__.__name__}")

1
2
3
4
5
6

第0步 消息数=1  最后一条=HumanMessage      # 初始：只有用户问题
第1步 消息数=2  最后一条=AIMessage         # agent 决策（带 tool_calls）
第2步 消息数=3  最后一条=ToolMessage       # tools 回填
第3步 消息数=4  最后一条=AIMessage         # agent 再决策
第4步 消息数=5  最后一条=ToolMessage       # tools 再回填
第5步 消息数=6  最后一条=AIMessage         # 最终答案

1
2
3
4

for chunk, metadata in graph.stream(TEST_INPUT, stream_mode="messages"):
    if chunk.content:                       # 过滤空 chunk
        print(chunk.content, end="", flush=True)   # flush=True：立即刷到屏幕
print()

1
2
3
4
5

  模型在干什么            流出的内容            chunk.content
  ─────────────────      ──────────────       ───────────
  决策要不要调工具         只有角色标记            空
  输出工具参数            {"a":15,"b":27}       空（参数不进 content）
  生成正式回答            "(15+27)×2 = 84"      有字 ✅

1
2
3

for chunk, metadata in graph.stream(TEST_INPUT, stream_mode="messages"):
    if chunk.content:
        print(f"[{metadata['langgraph_node']}]", chunk.content, end="")

1
2
3
4
5
6
7

def stream_answer(graph, user_input: str):
    """打字机函数：逐字 yield 模型的回答。W10 接 FastAPI SSE 直接用"""
    for chunk, metadata in graph.stream(
        {"messages": [("user", user_input)]}, stream_mode="messages"
    ):
        if chunk.content:
            yield chunk.content

1
2
3

# 用法（生成器逐字消费）：
for word in stream_answer(graph, "讲个笑话"):
    print(word, end="", flush=True)

1
2
3
4
5

import time

print("准备调模型")
time.sleep(3)          # 假装等 API 3 秒 —— 这 3 秒里程序啥也干不了
print("拿到结果")

1
2
3
4
5
6
7
8
9

import asyncio

async def make_coffee():            # ① async def：定义一个"协程"
    print("开始冲泡")
    await asyncio.sleep(1)          # ② await：挂起，把控制权交出去
    print("咖啡好了")
    return "☕"

asyncio.run(make_coffee())          # ③ asyncio.run()：唯一的启动入口

1
2
3
4
5
6

async def hello():
    return "hi"

f = hello()          # ⚠️ 注意：调用不执行！返回的是一个"协程对象"
print(f)             # <coroutine object hello at 0x...>
# 协程对象要被执行，得放进事件循环（asyncio.run 干的事）

1
2
3
4
5
6

  事件循环 = 那个不停巡视的咖啡师/前台
    │
    ├─ 任务 A 说："我要等网络响应"  → 好，你去等着，我记下你
    ├─ 任务 B 说："我要算点东西"    → 好，你先算（CPU 不空转）
    ├─ 任务 A 的响应到了！           → 好，叫你回来继续
    └─ 所有任务都完成了             → 事件循环收工

1
2
3
4
5
6
7

# 同步 for：每取一个元素，都要"排队等"
for item in normal_iterator:
    ...

# 异步 for：取元素的过程可以"等"，等待时不阻塞别人
async for item in async_iterator:
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

import asyncio

async def make_coffee():
    print("[咖啡] 开始冲泡")
    await asyncio.sleep(2)              # 模拟：泡咖啡要等 2 秒
    print("[咖啡] 完成")
    return "咖啡好了"

async def main():
    task = asyncio.create_task(make_coffee())   # 把协程"派给"事件循环，不等它
    print("[主] 咖啡开始煮了，我先去干点别的")
    await asyncio.sleep(1)              # 主任务也等 1 秒（但它不阻塞咖啡）
    print("[主] 干完了，回来等咖啡")
    result = await task                 # 等咖啡任务结束，拿结果
    print("[主] 拿到：", result)

asyncio.run(main())

1
2
3
4
5

[咖啡] 开始冲泡
[主] 咖啡开始煮了，我先去干点别的
[主] 干完了，回来等咖啡
[咖啡] 完成
[主] 拿到： 咖啡好了

1
2
3
4

# 1. async def：定义一个"可暂停"的协程
# 2. await：在协程里挂起，等结果的同时让事件循环干别的
# 3. async for：异步迭代——等下一个元素时挂起不阻塞
# 4. asyncio.run(coro)：普通脚本启动异步的"唯一入口"

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

async def run_events():
    async for ev in graph.astream_events(TEST_INPUT, version="v2"):
        if ev["event"] == "on_chat_model_stream":       # 模型吐 token
            data = ev["data"]["chunk"].content
            if data:
                print(data, end="", flush=True)
        elif ev["event"] == "on_tool_start":            # 工具开始执行
            print(f"\n[tool开始] {ev['name']}")
        elif ev["event"] == "on_tool_end":              # 工具执行完
            print(f"[tool结束] {ev['name']} → {ev['data'].get('output')}")

asyncio.run(run_events())

1
2
3
4
5
6
7

[token] 让我帮你算
[tool开始] add
[tool结束] add → 42
[token] 然后乘以2
[tool开始] multiply
[tool结束] multiply → 84
[token] 所以结果是 84

1
2
3
4
5

  messages 模式：只要"模型说的话" → 打字机 → 够用，别用 events
  astream_events：要"过程全记录" → 埋点/计时/监控/trace → 用它

  ⚠️ events 是杀鸡用的牛刀——它多一层事件解析，
     打字机场景用 messages 更简单直接。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

"""W5-D4: 一张 ReAct 图，四种看法（updates / values / messages / events）"""
import os, asyncio
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# ── 1. 复用 D3 的 ReAct 图 ──
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

@tool
def add(a: int, b: int) -> int:
    """计算两个整数之和"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """计算两个整数之积"""
    return a * b

tools = [add, multiply]

llm = ChatOpenAI(
    model="deepseek-v4-flash",
    base_url="https://api.deepseek.com",
    api_key=os.getenv("DEEPSEEK_API_KEY") or os.getenv("OPENAI_API_KEY"),
)
llm_with_tools = llm.bind_tools(tools)

def agent_node(state: AgentState) -> dict:
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

builder = StateGraph(AgentState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
graph = builder.compile()

TEST_INPUT = {"messages": [("user", "帮我算 (15 + 27) 乘以 2")]}

# ── 看法①：updates（看结构）──
print("=== ① updates：谁改了什么 ===")
for event in graph.stream(TEST_INPUT, stream_mode="updates"):
    print("  ", event)

# ── 看法②：values（看全貌）──
print("\n=== ② values：每一步完整 state ===")
for i, snap in enumerate(graph.stream(TEST_INPUT, stream_mode="values")):
    print(f"  第{i}步 消息数={len(snap['messages'])}  "
          f"最后一条={snap['messages'][-1].__class__.__name__}")

# ── 看法③：messages（打字机）──
print("\n=== ③ messages：token 级打字机 ===")
for chunk, metadata in graph.stream(TEST_INPUT, stream_mode="messages"):
    if chunk.content:                 # 过滤空 chunk（工具调用阶段没文本）
        print(chunk.content, end="", flush=True)
print("\n")

# ── 看法④：astream_events（最细粒度，异步）──
print("=== ④ astream_events：token + 工具事件 ===")
async def run_events():
    async for ev in graph.astream_events(TEST_INPUT, version="v2"):
        if ev["event"] == "on_chat_model_stream":
            data = ev["data"]["chunk"].content
            if data:
                print(data, end="", flush=True)
        elif ev["event"] == "on_tool_start":
            print(f"\n  [tool开始] {ev['name']}")
        elif ev["event"] == "on_tool_end":
            print(f"  [tool结束] {ev['name']} → {ev['data'].get('output')}")

asyncio.run(run_events())             # 🎯 异步代码的唯一启动入口
print("\n=== 完成 ===")

1
2
3
4
5

async def run_events():                       # ① async def：协程
    async for ev in graph.astream_events(...):# ③ async for：异步迭代收事件
        ...
asyncio.run(run_events())                    # ④ asyncio.run：启动入口
# （② await 被藏在 async for 内部——等下一个事件时自动挂起）

1
2
3
4
5
6

for mode, data in graph.stream(TEST_INPUT, stream_mode=["updates", "messages"]):
    if mode == "updates":
        print("[结构]", data)          # 节点增量
    else:
        # mode == "messages"：token 流（这里不打印，太乱）
        pass

1
2

# stream_mode="custom"：节点里手动吐自定义数据（进度条/自定义事件）——🟡 了解
# subgraphs=True：嵌套图（W6 子图）时把子图事件也吐出来——W6 见

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

# ===== 四种看法 =====
for event in graph.stream(TEST_INPUT, stream_mode="updates"):     # ① 谁改了什么
    print(event)                                                   # {节点: 更新}

for i, snap in enumerate(graph.stream(TEST_INPUT, stream_mode="values")):
    print(f"第{i}步 消息数={len(snap['messages'])}")              # ② 完整 state

for chunk, metadata in graph.stream(TEST_INPUT, stream_mode="messages"):
    if chunk.content:                                              # ③ token 级
        print(chunk.content, end="", flush=True)                   #    打字机

async def run_events():                                            # ④ 事件级（异步）
    async for ev in graph.astream_events(TEST_INPUT, version="v2"):
        if ev["event"] == "on_chat_model_stream" and ev["data"]["chunk"].content:
            print(ev["data"]["chunk"].content, end="", flush=True)
        elif ev["event"] == "on_tool_start":
            print(f"\n[tool开始] {ev['name']}")
asyncio.run(run_events())

# ===== 选型口诀 =====
# 调结构用 updates，看全貌用 values，做产品用 messages，做监控用 events

# ===== 打字机封装（W10 接 SSE）=====
def stream_answer(graph, user_input: str):
    for chunk, metadata in graph.stream(
        {"messages": [("user", user_input)]}, stream_mode="messages"
    ):
        if chunk.content:
            yield chunk.content          # yield 出去，W10 用 FastAPI SSE 推送

# ===== 异步四板斧 =====
# async def  定义可暂停的协程（调用不执行！）
# await      挂起点：等的时候把控制权交给事件循环
# async for  异步迭代：等下一个元素时不阻塞
# asyncio.run(coro)  普通脚本唯一的启动入口

1
2
3
4
5
6
7

  同步 = 银行柜台：一个人办事，全队等他
  异步 = 咖啡馆叫号：等咖啡的时候，咖啡师在冲别人的

  async def（可暂停的函数）
  await（等的时候让出控制权）
  async for（异步迭代：等下一个事件不阻塞）
  asyncio.run()（启动入口）