Day 06 / 共 60 天 · 阶段 1 入门与心智
第一个对话图 + 阶段 1 心智小结
阶段 1 收官日。今天不引入新源码机制,而是把 D01-05 学的 StateGraph、节点、边、State schema、通道,串成一张真能跑的对话图:用现成的 MessagesState、一个模型节点、一条条件边形成"想回答就回答、想调工具就循环"的结构。然后用"通道触发"心智模型把它从头到尾走一遍——让入门阶段的知识合上闭环。
📍 阶段 1「入门与心智」(D01-06) · 你在第 6 天(收官)
D01 全景分包→
D02 跑通第一个图→
D03 StateGraph→
D04 节点与边→
D05 State/TypedDict→
D06 对话图+小结
💡 一句话锚点
一个最小对话 Agent = MessagesState(自带消息累加)+ 一个调模型的节点 + 一个执行工具的节点 + 一条条件边(模型想调工具就去工具节点、否则结束)。这四件事全是你前五天学过的零件,今天只是拼起来。拼完再用"谁写通道→谁被触发"的视角复盘一遍,你对 LangGraph 的核心心智就成型了。
L01
一个对话图需要哪些零件
🤔 痛点"对话 Agent"听起来复杂,但拆开无非要回答四个问题:① 对话历史存哪、怎么越攒越多?② 谁去调大模型?③ 模型说"我要查天气"(工具调用)时谁去执行?④ 什么时候该停、什么时候要再问模型一轮?前三天学的机制正好一一对应。
| 对话图的需求 | 用哪个学过的机制 | 来自 |
|---|---|---|
| 历史消息越攒越多、不能互相覆盖 | 带 add_messages reducer 的字段 | D05 reducer 嗅探 |
| 调模型 / 执行工具 | 两个普通节点(函数) | D04 节点 |
| "模型→工具→再回模型"的循环 | 条件边 + 普通边 | D03/D04 边 |
| 什么时候结束 | 条件边返回 END | D03 START/END |
大白话对话 Agent 不是"新东西",而是把"累加字段 + 两个节点 + 一条会拐弯的边"组合出的一个经典形状。这个形状后面阶段 9 的
create_react_agent 就是它的工业化封装。L02
MessagesState:白得一个消息累加字段
你不用手写 Annotated[list, add_messages],LangGraph 备好了一个基类:
libs/langgraph/langgraph/graph/message.py:372-373
class MessagesState(TypedDict):
messages: Annotated[list[AnyMessage], add_messages]
messages 字段就是对话历史。类型 list[AnyMessage](人类/AI/工具消息都算)。Annotated[..., add_messages]★关键:给这个字段配了 add_messages reducer(D05 学的"嗅探"会把它变成累加通道)。所以节点返回的新消息是追加不是覆盖,还能按 ID 去重/更新(D08 深入)。继承它加自己字段写 class State(MessagesState): 额外字段... 就白得一个配好的 messages。这是"约定优于配置"。💡 为什么消息字段一定要 reducer?回忆 D05 的边界:模型节点可能和工具节点在不同超步都往
messages 写,且历史必须累积而非覆盖。若用默认 LastValue,新消息会把整段历史顶掉。add_messages 正是为此存在——它让 messages 成为一个"只进不出、按 ID 智能合并"的通道。那条件边(Agent 的分叉点)在源码里长什么样?看 add_conditional_edges 的签名:
libs/langgraph/langgraph/graph/state.py:969-976(节选)
def add_conditional_edges(
self,
source: str, # 从哪个节点出发(本例 "model")
path: Callable[..., Hashable | Sequence[Hashable]] | Runnable[...], # 路由函数
path_map: dict[Hashable, str] | list[str] | None = None, # 可选:返回值→节点名 映射
) -> Self:
三个参数:
source 是分叉起点;path 就是你的 should_continue(跑完 source 后调它决定去哪);path_map 可省——省了就直接用 path 的返回值当目标节点名(本例返回 "tools" 或 END)。D13-14 会深挖它怎么解析和执行。L03
搭一个对话循环(可运行)
把零件拼起来。为了聚焦"图的结构",模型和工具用假的替身(真实换成 ChatOpenAI 即可):
from langgraph.graph import StateGraph, START, END, MessagesState
from langchain_core.messages import AIMessage, ToolMessage
def call_model(state: MessagesState) -> dict:
last = state["messages"][-1].content
if "天气" in last and not any(getattr(m, "tool_calls", None) for m in state["messages"]):
# 模型决定调工具:返回一条带 tool_calls 的 AI 消息
return {"messages": [AIMessage(content="", tool_calls=[
{"name": "get_weather", "args": {"city": "北京"}, "id": "t1"}])]}
return {"messages": [AIMessage(content="今天北京晴,25℃。")]} # 直接回答
def call_tool(state: MessagesState) -> dict:
return {"messages": [ToolMessage(content="晴 25℃", tool_call_id="t1")]}
def should_continue(state: MessagesState) -> str:
last = state["messages"][-1]
return "tools" if getattr(last, "tool_calls", None) else END # ★路由决策
g = StateGraph(MessagesState)
g.add_node("model", call_model)
g.add_node("tools", call_tool)
g.add_edge(START, "model") # 入口
g.add_conditional_edges("model", should_continue) # 模型后:要么去 tools,要么 END
g.add_edge("tools", "model") # 工具后:回模型再想一轮(循环!)
app = g.compile()
print(app.invoke({"messages": [("user", "北京天气怎么样")]}))
add_edge(START, "model")入口:图一开始就把输入喂给 model 节点。(等价 set_entry_point("model"),D18 讲 START)add_conditional_edges("model", should_continue)★条件边:model 跑完调 should_continue 看最后一条消息有没有 tool_calls——有就去 "tools",没有就 END。这是 Agent 的"大脑分叉点"(D13-14 深入)。add_edge("tools", "model")★循环的关键:工具执行完回到 model,让模型基于工具结果再想。这条边让图变成一个可以转好几圈的环(D17 讲循环上限如何防止转不停)。invoke({"messages": [("user", ...)]})输入一条用户消息元组,add_messages 会把它归一成 HumanMessage(D08)。图注:这就是 ReAct Agent 的骨架。阶段 9 的 create_react_agent 只是把它做全(并行工具、错误处理等)。
L04
一次 invoke 走了哪些步
输入 "北京天气怎么样",这张图实际转了三个"超步"(superstep,阶段 4 精讲):
超步 0:START → model用户消息进 messages 通道。model 被触发,判断"含天气且还没调过工具" → 返回带 tool_calls 的 AI 消息。add_messages 把它追加进历史。条件边判定should_continue 看最后一条消息有 tool_calls → 返回 "tools"。于是"通往 tools 的通道"被写入。超步 1:model → toolstools 被触发,执行 get_weather,返回 ToolMessage("晴 25℃"),追加进历史。超步 2:tools → model固定边把控制权还给 model。model 再判断"已经有工具结果了" → 返回最终答复 "今天北京晴,25℃。"。条件边判定 → END最后一条消息无 tool_calls → should_continue 返回 END,图停止。输出整段 messages。📝 最终 messages(4 条)
看:历史一路累加没被覆盖(
[HumanMessage("北京天气怎么样"), AIMessage(tool_calls=[get_weather]), ToolMessage("晴 25℃"), AIMessage("今天北京晴,25℃。")]看:历史一路累加没被覆盖(
add_messages 的功劳),控制权在 model↔tools 之间循环了一圈(条件边+回边的功劳)。
L05
心智模型:一切都是"写通道 / 被通道触发"
D04 埋下的种子在这里收:上面那些"超步""触发""路由",落到源码全是通道读写。回顾编译后节点的样子:
libs/langgraph/langgraph/graph/state.py:1518-1531(attach_node 造 PregelNode,节选)
self.nodes[key] = PregelNode(
triggers=[branch_channel], # 被"通往我的通道"触发才跑
channels=(... input_channels), # 读哪些状态字段
writers=[ChannelWrite(...)], # 结果写回哪些通道(状态字段 + 通往下游的通道)
bound=node.runnable, # 你的 call_model / call_tool 函数
)
💡 把整个 LangGraph 压缩成一句话
"节点 = 被某通道触发 → 读若干通道 → 跑你的函数 → 把结果写回若干通道;边 = 让上游写'通往下游的通道';State 字段 = 一个个决定'怎么合并写入'的通道。" 你今天这张对话图,在引擎眼里就是:
messages 通道 + 若干 branch:to:* 触发通道,节点轮流被唤醒、读写它们。记住这一句,阶段 4 学 Pregel 时就不会晕。图注:节点之间解耦——加/删节点只影响通道订阅关系,这就是图能灵活重连的根本。
L06
阶段 1 五天串讲(一张知识地图)
| 天 | 一句话 | 核心源码 |
|---|---|---|
| D01 | LangGraph 分成 langgraph/checkpoint/prebuilt/cli/sdk 几个包,核心是 langgraph | README / pyproject |
| D02 | 最小图:定义 State → add_node → add_edge → compile → invoke | StateGraph 用法 |
| D03 | StateGraph 是建造器,7 个内部容器存图纸,compile 才翻译成引擎 | state.py:201-207 |
| D04 | 节点 = 规格单(StateNodeSpec)→运行单元(PregelNode);边 = 通道写/订阅 | _node.py / state.py:1537 |
| D05 | State schema 被 _get_channels 逐字段翻译成通道,reducer 靠数参数嗅出 | state.py:1801-1908 |
| D06 | 把上面拼成对话图,用"写通道/被触发"心智模型复盘 | 本页 |
🍼 阶段 1 结束你应该有的直觉看到任何一张 LangGraph 图,你能在脑子里翻译成:"这些字段各是什么通道(覆盖还是累加)?哪些节点订阅哪些触发通道?入口在哪、什么条件下停?"能做到这三问,入门就通了。
L07
两处设计取舍 + 一处边界
🎨 设计取舍①:为什么用"条件边 + 回边"表达循环,而不是内置 while?
朴素做法:给框架一个
while_loop(condition, body) 原语。LangGraph 选择用普通的条件边 + 一条回到上游的边自然形成环。好处:循环、分支、并行、扇出全都用同一套"边"表达,没有特殊语法;图结构是纯声明式的数据(D03 的那些 dict/set),可被可视化、被检查、被序列化。代价:新手容易搭出"停不下来的环"(忘了给 END 出口)——所以框架必须有 recursion_limit 兜底(正是 D17 的主题)。🎨 设计取舍②:为什么提供 MessagesState 这种"约定基类"?
本可以让每个人自己写
Annotated[list, add_messages]。官方额外固化了 MessagesState。好处:99% 的对话图第一行就对了,不会有人误用 LastValue 把历史写没;且给后续 create_react_agent 一个统一的状态契约。代价:多一个要记的名字、且它"藏"起了 reducer 细节——新手可能不知道 messages 为什么会自动累加。本教程 D05/D08 补上这层认知正是为此。⚠️ 边界:条件边函数忘记返回 END,图会转到递归上限报错如果
should_continue 在任何情况下都返回节点名、永远不返回 END,或者你把 tools→model 的边搭成了 tools→tools,图就会无限循环,最终撞上 recursion_limit 抛 GraphRecursionError(D17)。排查口诀:每个可能进入循环的条件边,都要问"它有没有一条通往 END 的出路,且出路的条件真的会被满足"。真实 Agent 里"模型不再产生 tool_calls"就是那条出路。L08
今日小结 + 动手 + 明日预告
🧠 今天你应该能回答
- 最小对话 Agent 由哪四个零件组成?(MessagesState + model 节点 + tools 节点 + 条件边)
- MessagesState 帮你做了什么?(自带
Annotated[list, add_messages]的 messages 字段) - 循环靠什么表达?(条件边分叉 + 一条 tools→model 的回边)
- 图什么时候停?(条件边返回 END——本例是"模型不再产生 tool_calls")
- 用一句话概括 LangGraph 的运行模型?(节点被通道触发、读写通道;边就是通道写/订阅)
- 为什么消息字段必须用 add_messages 而不是默认覆盖?(要累积历史、支持并行/按ID合并)
- 没有 END 出口的循环会怎样?(撞 recursion_limit 抛 GraphRecursionError)
✋ 10 分钟动手
# 1. 看 MessagesState 定义
sed -n '372,373p' libs/langgraph/langgraph/graph/message.py
# 2. 把 L03 的代码存成 chat.py 跑起来,打印每条消息
python3 chat.py
# 3. 改造实验:把 should_continue 永远返回 "tools"(去掉 END 出口),
# 观察它撞 recursion_limit 报 GraphRecursionError(预告 D17)
# 4. 用 stream 看每一步(体会"超步")
python3 -c "
# 复用 chat.py 里的 g
from chat import app
for chunk in app.stream({'messages':[('user','北京天气怎么样')]}):
print(chunk)
"
💡 明日预告 · Day 07(进入阶段 2)今天多次用到
add_messages、reducer。明天正式进入阶段 2「状态与数据流」:D07 从头讲 reducer 入门——Annotated[x, reducer] 在 graph/state.py 里到底怎么被读出、怎么变成"合并两个值"的通道,为 D08 深挖 add_messages、D09 挖累加通道打地基。