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 边
什么时候结束条件边返回 ENDD03 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)。
对话图的形状:一个会拐弯的环 START model tools END 有 tool_calls 回 model 无 tool_calls 条件边让 model 有两个去向;tools→model 形成循环
图注:这就是 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_callsshould_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 时就不会晕。
控制流:节点靠通道互相唤醒(不是直接调用) model messages 通道 branch:to:tools tools 触发 写msg 写msg model 与 tools 从不"直接调用"对方,只是读写共享通道
图注:节点之间解耦——加/删节点只影响通道订阅关系,这就是图能灵活重连的根本。
L06

阶段 1 五天串讲(一张知识地图)

一句话核心源码
D01LangGraph 分成 langgraph/checkpoint/prebuilt/cli/sdk 几个包,核心是 langgraphREADME / pyproject
D02最小图:定义 State → add_node → add_edge → compile → invokeStateGraph 用法
D03StateGraph 是建造器,7 个内部容器存图纸,compile 才翻译成引擎state.py:201-207
D04节点 = 规格单(StateNodeSpec)→运行单元(PregelNode);边 = 通道写/订阅_node.py / state.py:1537
D05State 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_limitGraphRecursionError(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 挖累加通道打地基。
← Day 05 · State 与 TypedDict Day 07 · reducers 入门 →