W5-D3 · 手搭 ReAct 图：把 W2 的 fc_loop 画成一张带环的图

1
2
3
4
5
6

                    ┌─────────────────────────────┐
                    │                             │
                    ▼                             │
START → agent ──(有 tool_calls)──→ tools ────────┘
            │
            └──(无 tool_calls)──→ END

1
2
3

  while True:              →  tools → agent 回边（圈）
  if tool_calls:           →  tools_condition 条件路由
  execute + append 结果    →  ToolNode 自动执行并回填

1
2
3
4

  第 1 轮  messages = [Human, AI]              记着"我叫小明"
  第 2 轮  return {"messages": [AI]}           覆盖！
           messages = [AI]                     ← 上一轮的对话没了
  第 3 轮  模型看到只有一条消息，根本不知道你叫什么

1
2
3
4
5

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]   # 追加，不是覆盖

1
2
3
4

# add_messages 的"更新"语义：同 id 的消息会被替换而不是追加
# 这让"覆盖某条历史消息""删除某条消息"成为可能
from langchain_core.messages import RemoveMessage
return {"messages": [RemoveMessage(id=msg_id)]}

1
2
3

  ① 读 LLM 最后一条消息里的 tool_calls
  ② 逐个调用对应的工具函数
  ③ 把结果包装成 ToolMessage，追加回 state 的 messages

1

builder.add_node("tools", ToolNode(tools))

1
2
3
4
5

  W4 的痛：                              W5 的解：
  手动 for 循环执行                   →   ToolNode 自动执行
  失败要 try/except 手动回填          →   ToolNode 自动回填（含失败）
  绑了工具却没有执行节点 → 空 content  →   两者成对出现才完整
  并行调用怕回填不齐 → 只能强制串行    →   并行也能逐条正确回填

1
2
3
4

tools_condition
# 等价于：
#   最后一条 AI 消息有 tool_calls → 返回 "tools"
#   没有                          → 返回 END

1

builder.add_conditional_edges("agent", tools_condition)

1
2

finish_reason: 'tool_calls'   → 模型说"我要调工具，话没说完"
finish_reason: 'stop'         → 模型说"我说完了"

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

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.tools import tool

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]

1
2
3
4
5
6
7

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)   # 🎯 关键：绑了工具的 llm 才会返回 tool_calls

1
2
3
4
5

def agent_node(state: AgentState) -> dict:
    if len(state["messages"]) > 10:
        forced = {"role": "user", "content": "请基于以上工具结果直接给出最终答案，不要再调用工具。"}
        return {"messages": [llm_with_tools.invoke(state["messages"] + [forced])]}
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

1
2
3
4
5
6
7

# ⚠️ 1. forced 必须包成列表：list + dict 会 TypeError
state["messages"] + forced      # ❌ TypeError: can only concatenate list (not "dict") to list
state["messages"] + [forced]    # ✅

# ⚠️ 2. 收口指令是发给模型的，错字会改变模型行为
"请基于以上共计结果……不要在调用工具。"     # ❌ "共计"可能被引申为"再求和"
"请基于以上工具结果……不要再调用工具。"     # ✅

1
2
3
4
5
6
7
8
9

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)   # 有 tool_calls → tools，否则 → END
builder.add_edge("tools", "agent")              # 🎯 回边！

graph = builder.compile()

1
2
3
4
5
6
7

  ✅ 有回边：  agent → tools → agent → END
              工具结果回到 agent，模型看到结果，生成最终答案

  ❌ 没回边：  agent → tools → END
              工具执行完就结束，没人把结果变成答案
              → 你拿到的最后一条消息是 ToolMessage(content='60')，
                而不是 "答案是 60"

 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

"""W5-D3: 手搭 ReAct 图 —— agent ↔ tools 循环"""
import os
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. State：messages 用 reducer 自动累加 ──
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# ── 2. 工具（@tool 从签名 + docstring 自动生成 schema，docstring 是给模型看的）──
@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]

# ── 3. LLM（多厂商只换 base_url / model）──
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)   # 关键：绑了工具的 llm 才会返回 tool_calls

# ── 4. agent 节点（含防死循环：消息超 10 条强制收口）──
def agent_node(state: AgentState) -> dict:
    if len(state["messages"]) > 10:
        forced = {"role": "user", "content": "请基于以上工具结果直接给出最终答案，不要再调用工具。"}
        return {"messages": [llm_with_tools.invoke(state["messages"] + [forced])]}
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# ── 5. 构图：ReAct 经典拓扑 ──
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)   # 有 tool_calls → tools，否则 END
builder.add_edge("tools", "agent")               # 🔴 回边！没有它工具跑完就结束，永远没有最终答案

graph = builder.compile()

# ── 6. 链式任务验收：需要"先 add 再 multiply"两次工具调用 ──
result = graph.invoke(
    {"messages": [("user", "帮我算一下 (12 + 8) 乘以 3")]},
    config={"recursion_limit": 25},              # 框架级保险丝
)
print("=== 完整链路（肉眼确认）===")
for m in result["messages"]:
    m.pretty_print()

# ── 7. stream 看 agent / tools 交替 ──
print("\n=== stream 看节点流转 ===")
for event in graph.stream({"messages": [("user", "3 加 5 等于几")]}, stream_mode="updates"):
    print(event)

# ── 8. prebuilt 对照验证 ──
from langgraph.prebuilt import create_react_agent

prebuilt_agent = create_react_agent(llm_with_tools, tools)
result2 = prebuilt_agent.invoke({"messages": [("user", "(12+8)*3 等于几")]})

print("=== prebuilt 版输出 ===")
for m in result2["messages"]:
    m.pretty_print()

1
2
3
4

result = graph.invoke({"messages": [("user", "帮我算一下 (12 + 8) 乘以 3")]})
print("=== 完整链路（肉眼确认）===")
for m in result["messages"]:
    m.pretty_print()

1
2
3
4
5
6

HumanMessage: 帮我算一下 (12 + 8) 乘以 3
AIMessage:    tool_calls=[add(12, 8)]      ← agent 第一次决策
ToolMessage:  20                            ← tools 执行 add 并回填
AIMessage:    tool_calls=[multiply(20, 3)]  ← agent 看到 20，再调 multiply
ToolMessage:  60                            ← tools 执行 multiply
AIMessage:    (12+8)×3 = 60                 ← agent 无 tool_calls → END

1
2
3

  它先调 add → 看到结果 20 → 再调 multiply → 看到 60 → 才作答
              ↑                        ↑
         每一次"看到结果再决定"，都是一次 agent→tools→agent 的转圈

1
2

for event in graph.stream({"messages": [("user", "3 加 5 等于几")]}, stream_mode="updates"):
    print(event)

1
2
3

{'agent': ...tool_calls=[add(3, 5)]...}   ← agent 决策要调工具
{'tools': ...ToolMessage(content='8')...} ← tools 执行并回填
{'agent': ...tool_calls=[]...}            ← agent 拿到 8，不再调 → 结束

1
2
3
4

graph.invoke(
    {"messages": [...]},
    config={"recursion_limit": 25}    # 默认值就是 25
)

1
2
3

if len(state["messages"]) > 10:
    forced = {"role": "user", "content": "请基于以上工具结果直接给出最终答案，不要再调用工具。"}
    return {"messages": [llm_with_tools.invoke(state["messages"] + [forced])]}

1
2
3
4
5
6
7

  每转一圈，messages 增加几条？
    简单情况：+1（AI）+1（Tool）= +2 条
    并行调 2 个工具：+1（AI）+2（Tool）= +3 条
    并行调 3 个工具：+1（AI）+3（Tool）= +4 条

  所以 "> 10 条" 究竟等于几轮？
    → 3 轮 ~ 5 轮不等，取决于模型有没有并行调用

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

import operator
from typing import Annotated

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    rounds: Annotated[int, operator.add]      # reducer 自动累加

def agent_node(state: AgentState) -> dict:
    if state.get("rounds", 0) >= 3:           # 🎯 精确的 3 轮上限
        forced = {"role": "user", "content": "请基于以上工具结果直接给出最终答案，不要再调用工具。"}
        return {"rounds": 1, "messages": [llm_with_tools.invoke(state["messages"] + [forced])]}
    return {"rounds": 1, "messages": [llm_with_tools.invoke(state["messages"])]}

1
2
3
4
5
6

Human: 帮我算一下 (12+8)×3
AI:    tool_calls=[add(12, 8)]      ← 决策1：先算括号
Tool:  20                           ← add 执行，回填
AI:    tool_calls=[multiply(20, 3)] ← 决策2：看到 20，再算乘
Tool:  60                           ← multiply 执行，回填
AI:    (12+8)×3 = 60                ← 无 tool_calls → 结束

1
2
3

{'agent': ...tool_calls=[add(3, 5)]...}   ← agent 决策要调工具
{'tools': ...ToolMessage(content='8')...} ← tools 执行并回填
{'agent': ...tool_calls=[]...}            ← agent 拿到 8，不再调 → 结束

1
2
3
4
5
6
7
8

from langgraph.prebuilt import create_react_agent

prebuilt_agent = create_react_agent(llm_with_tools, tools)
result2 = prebuilt_agent.invoke({"messages": [("user", "(12+8)*3 等于几")]})

print("=== prebuilt 版输出 ===")
for m in result2["messages"]:
    m.pretty_print()

1
2
3

  第 1 次调用： messages = [Human]                 → 全量计费
  第 2 次调用： messages = [Human, AI, Tool]       → 前 384 token 命中缓存，省钱
  第 3 次调用： messages = [Human, AI, Tool, AI...] → 命中更多

1
2

state["messages"] + forced      # ❌ TypeError: can only concatenate list (not "dict") to list
state["messages"] + [forced]    # ✅

 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

# ===== ReAct 最小图 =====
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

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]     # 🎯 追加，不是覆盖

@tool
def add(a: int, b: int) -> int:
    """计算两个整数之和"""                       # 🎯 docstring = 给模型的说明书
    return a + b

llm_with_tools = ChatOpenAI(...).bind_tools(tools)   # 🎯 用返回值

def agent_node(state: AgentState) -> dict:
    if len(state["messages"]) > 10:             # 业务级软收口
        forced = {"role": "user", "content": "请基于以上工具结果直接给出最终答案，不要再调用工具。"}
        return {"messages": [llm_with_tools.invoke(state["messages"] + [forced])]}  # ⚠️ [forced]
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

b = StateGraph(AgentState)
b.add_node("agent", agent_node)
b.add_node("tools", ToolNode(tools))
b.add_edge(START, "agent")
b.add_conditional_edges("agent", tools_condition)    # 🎯 挂 agent 后面
b.add_edge("tools", "agent")                         # 🔴 回边，缺了没有最终答案
graph = b.compile()

# 跑 + 确认链路
result = graph.invoke({"messages": [("user", "帮我算一下 (12 + 8) 乘以 3")]},
                      config={"recursion_limit": 25})    # 框架级保险丝
for m in result["messages"]:
    m.pretty_print()

# 对照验证
from langgraph.prebuilt import create_react_agent
prebuilt = create_react_agent(llm_with_tools, tools)

1
2
3
4

  while True              →  tools → agent 回边
  if tool_calls           →  tools_condition
  execute + append        →  ToolNode（还顺手治好了 W4 的 insufficient 坑）
  MAX_ROUNDS              →  框架不包业务级收口，得自己加

1
2

  预期（串行）：  AI(add) → Tool(20) → AI(multiply) → Tool(60) → AI(答案)   5 步
  实际（并行）：  AI(add, multiply) → Tool(20) → Tool(60) → AI(答案)        4 步

1
2
3

  7! = 7×6×5×4×3×2×1 = 5040，再 ÷ 2 = 2520
  模型要么连续调 6 次 multiply（考验回边），
  要么一次并行调多个（考验 ToolNode 的多回填）