W5-D5 · State 升级：从便利贴到安检门

1
2
3
4
5

用户输入
  ↓
LLM 决策 ──有 tool_calls──→ 执行工具 ──→ 结果回填 ──┐
  │                                                  │
  └──────────── 无 tool_calls → 输出最终答案 ←───────┘

1
2

resp = client.chat.completions.create(model=..., messages=messages)
answer = resp.choices[0].message.content   # 一次问答，结束

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

while True:                                    # ← 循环 = 图里"再决策"的部分
    resp = client.chat.completions.create(
        model=..., messages=messages, tools=tools)
    msg = resp.choices[0].message
    messages.append(msg)

    if not msg.tool_calls:                     # ← 路由：if tool_calls
        break                                  #    没有 → 结束
    for tc in msg.tool_calls:                  # ← 执行工具
        result = run_tool(tc.function.name, json.loads(tc.function.arguments))
        messages.append({                      # ← 手动回填
            "role": "tool", "tool_call_id": tc.id, "content": str(result)})

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

while True:                                    # ← ⚠️ 循环还得自己写！
    try:
        resp = llm_with_tools.invoke(messages) # ← 统一了厂商/schema/解析
    except Exception:
        resp = llm_without_tools.invoke(messages)  # ← 降级自愈
    messages.append(resp)
    if not resp.tool_calls:
        break
    for tc in resp.tool_calls:
        out = run_tool(tc["name"], tc["args"])
        messages.append(ToolMessage(content=str(out), tool_call_id=tc["id"]))

1
2
3
4
5
6

builder.add_node("agent", agent_node)                    # LLM 决策
builder.add_node("tools", ToolNode(tools))               # 执行 + 回填（自动）
builder.add_conditional_edges("agent", tools_condition)  # ← if tool_calls 变成路由
builder.add_edge("tools", "agent")                       # ← while True 变成回边
graph = builder.compile()
graph.invoke(...)                                        # 框架替你跑循环

1
2
3
4

  手写 fc_loop  解决：模型要调工具，谁来执行、谁来回填 → 你写 while + if
  LangChain     解决：调 LLM 的样板（厂商 / schema / 消息解析）→ 但循环还得自己写
  LangGraph     解决：控制流不可见、难扩展
                → 循环变回边、判断变路由、状态变 reducer、工具变 ToolNode

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

from typing import TypedDict
from pydantic import BaseModel, Field

# —— TypedDict：只在写代码时给你提示，运行时完全不检查 ——
class TD(TypedDict):
    count: int

td: TD = {"count": "不是整数"}   # 运行时零报错！字符串被当作 int 用下去

# —— Pydantic：运行时真校验 ——
class PD(BaseModel):
    count: int = Field(ge=0, description="必须是非负整数")

PD(count="不是整数")   # ❌ ValidationError：类型错误被拦下
PD(count=-5)           # ❌ ValidationError：ge=0 拦住负数
PD(count=5).model_dump()   # ✅ {'count': 5}

1
2
3
4
5
6
7

from typing import Annotated
from pydantic import BaseModel

class RichState(BaseModel):
    messages: Annotated[list, add_messages]   # 字段类型 T + 元数据（reducer）
    #         └───────┬───────┘  └─────┬─────┘
    #           字段类型 list      第二个参数：reducer 函数

1
2
3
4
5
6
7
8
9

  节点返回的 dict 里，每个 key 触发对应字段的一次更新：

  ① 有 Annotated 的字段（带 reducer）→ 走 reducer 合并
     messages: Annotated[list, add_messages]
     → 新消息被"追加"进现有列表

  ② 没有 Annotated 的字段 → 整体覆盖（last write wins）
     counter: int
     → 谁最后写，就整个盖掉前面的

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

from typing import Annotated
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class DemoState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)  # 有 reducer：追加
    counter: int = 0                                       # 无 reducer：覆盖

def n1(state: DemoState) -> dict:
    return {"messages": [{"role": "user", "content": "第一条"}], "counter": 1}

def n2(state: DemoState) -> dict:
    return {"messages": [{"role": "assistant", "content": "第二条"}], "counter": 2}

b = StateGraph(DemoState)
b.add_node("n1", n1); b.add_node("n2", n2)
b.add_edge(START, "n1"); b.add_edge("n1", "n2"); b.add_edge("n2", END)
g = b.compile()

final = g.invoke({})
print(len(final["messages"]))   # 2  →  reducer：n1、n2 的两条都留下了（追加）
print(final["counter"])         # 2  →  无 reducer：n2 覆盖了 n1 的 1

1
2
3
4
5
6
7

# 计数器：每个节点返回 {"visits": 1}，就会在现值上 +1
def increment(current: int, new: int) -> int:
    return current + new

# 日志：字符串列表拼接
def add_log(current: list[str], new: list[str]) -> list[str]:
    return current + new

 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

from typing import Annotated
from pydantic import BaseModel, Field
from langgraph.graph import StateGraph, START, END

class CounterState(BaseModel):
    visits: Annotated[int, increment] = 0                     # 有 reducer → 累加
    log: Annotated[list[str], add_log] = Field(default_factory=list)   # 有 reducer → 拼接
    plain: int = 0                                            # 无 reducer → 覆盖（对照组）

def step1(state: CounterState) -> dict:
    return {"visits": 1, "log": ["step1 执行了"], "plain": 1}

def step2(state: CounterState) -> dict:
    return {"visits": 1, "log": ["step2 执行了"], "plain": 2}

b = StateGraph(CounterState)
b.add_node("step1", step1)
b.add_node("step2", step2)
b.add_edge(START, "step1"); b.add_edge("step1", "step2"); b.add_edge("step2", END)
g = b.compile()

final = g.invoke({})
print("visits:", final["visits"])   # 2    ← step1 +1，step2 再 +1（increment 累加）
print("log:", final["log"])          # ['step1 执行了', 'step2 执行了']（add_log 拼接）
print("plain:", final["plain"])      # 2    ← 无 reducer，step2 覆盖了 step1 的 1

1
2
3

  visits: 2     increment（自定义 reducer）→ 每个节点写 1 都累加 → 计步器
  log: 两条      add_log（自定义 reducer）  → 每次都追加 → 聊天记录本
  plain: 2      普通字段（无 reducer）      → 后写覆盖 → 便利贴

1
2
3
4
5

from langgraph.graph import MessagesState

class MyState(MessagesState):    # 继承扩展：messages 字段免费送
    summary: str = ""            # 再自己加业务字段
    memory_path: str = ""

 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

"""W5-D5: Pydantic state + reducer 机制——为 D6 MemoryManager 打底"""
import os
from typing import Annotated
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition

# ── 1. Pydantic state：messages 追加 / summary 覆盖 / 计数 ──
class RichState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    summary: str = Field(default="", description="对话摘要")     # 普通字段：后写覆盖
    tool_use_count: int = 0                                     # 普通字段：见第八章为何是 0

# ── 2. 工具 + LLM ──
@tool
def get_weather(city: str) -> str:
    """查询城市天气(模拟)"""
    return f"{city}:晴,25°C"

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"),
)

# ── 3. 节点：Pydantic state 支持属性访问 state.messages ──
def agent_node(state: RichState) -> dict:
    return {"messages": [llm.bind_tools([get_weather]).invoke(state.messages)]}

def tag_node(state: RichState) -> dict:
    """收尾节点：普通字段每次覆盖式写入"""
    return {"summary": "本轮对话结束", "tool_use_count": state.tool_use_count + 1}

# ── 4. 构图：agent 分叉只能走条件边，END 写进 path_map ──
builder = StateGraph(RichState)
builder.add_node("agent", agent_node)
builder.add_node("tools", ToolNode([get_weather]))
builder.add_node("tag", tag_node)

builder.add_edge(START, "agent")
builder.add_edge("tools", "agent")       # 回边：工具结果喂回 agent
builder.add_edge("tag", END)

# agent 之后是"二选一"：有 tool_calls → tools；没有 → 收尾 tag
builder.add_conditional_edges("agent", tools_condition, {
    "tools": "tools",   # 有 tool_calls → 工具节点
    END: "tag",         # 无 tool_calls → 收尾节点（再让 tag 连 END）
})
graph = builder.compile()

# ── 5. 跑通验证 ──
result = graph.invoke({"messages": [("user", "北京天气怎么样")]})
print("=== 完整链路 ===")
for m in result["messages"]:
    m.pretty_print()
print("=== state 最终状态 ===")
print("summary:", result.get("summary"))
print("tool_use_count:", result.get("tool_use_count"))

1
2
3

def tag_node(state: RichState) -> dict:
    print(f"[tag] 执行了！进入时 tool_use_count = {state.tool_use_count}")
    ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

def increment(current: int, new: int) -> int:
    return current + new

class RichState(BaseModel):
    messages: Annotated[list, add_messages] = Field(default_factory=list)
    tool_use_count: Annotated[int, increment] = 0   # ← reducer：谁写都累加

# 然后让每个节点都返回 {"tool_use_count": 1}：
#   agent 决策 1 次 → +1
#   每次工具执行 → +1  → tool_use_count 变成"真实调用次数"

1
2

  普通 int 字段：后写覆盖      —— 适合"只写一次的最终值"（summary）
  Annotated[int, increment]：每写一次累加 —— 适合"路过就计数"（调用次数、步数）

 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

# ===== state 两种写法 =====
class TD(TypedDict):          # 运行时零校验（只给静态提示）
    count: int

class PD(BaseModel):          # 运行时真校验 + Field 约束 + model_dump
    count: int = Field(ge=0)

# ===== 字段合并规则 =====
# 有 Annotated（带 reducer）→ 合并（追加 / 累加）
# 无 Annotated → 后写覆盖（last write wins）

# ===== 三种 reducer =====
messages: Annotated[list, add_messages]        # 官方：消息追加
visits:   Annotated[int, increment]            # 自定义：累加
def increment(current: int, new: int) -> int:
    return current + new

# ===== 官方预置 =====
from langgraph.graph import MessagesState
class MyState(MessagesState):      # 底层 = messages: Annotated[list, add_messages]
    summary: str = ""

# ===== Pydantic state 三规则 =====
# 1. 可变默认必须 Field(default_factory=list)
# 2. 分叉节点只用 add_conditional_edges，收尾走 {END: "节点名"} 再连 END
# 3. state 保持扁平；嵌套 + reducer 会整体覆盖

# ===== 调试心法 =====
# 值不对 → 先确认代码跑没跑（加 print），再谈为什么不对
# 计数器要"路过就 +1" → 用 reducer；只写一次 → 普通字段

1
2
3

  手写 fc_loop   解决"谁来执行工具"      —— while + if + 手动回填
  LangChain      解决"调 LLM 的样板"     —— 但循环还得自己写
  LangGraph      解决"控制流不可见"      —— 循环 → 回边，state → reducer

1
2
3
4

  TypedDict：便利贴 —— 写着"这是整数"，没人检查
  Pydantic：安检门 —— 不是整数就拦下，早失败早发现
  普通字段：后写覆盖（last write wins）
  Annotated + reducer：合并（追加 / 累加）—— 计数器用它