1.3 核心概念：图、节点、边与状态

1.3 核心概念：图、节点、边与状态

四个词撑起整套框架

本节给出全册的词汇表：图（Graph）是节点与边的集合；节点（Node）是接收状态、返回增量的函数；边（Edge）分普通边与条件边；状态（State）是跨节点共享且按通道归约的数据。理解四者关系，后面所有代码都是这套词汇的拼写。我们发现初学者最容易混淆的是「节点返回全量还是增量」，这一点必须在一开始钉死。

概念拆解：归约器是状态的关键

节点返回的不是全量状态，而是「要合并的增量」。合并规则由归约器决定：operator.add 用于累加列表，lambda 用于覆盖或自定义合并。消息类状态常用 add_messages，它会按消息 id 去重并追加。边的路由函数返回下一个节点名或 END。START/END 是两个特殊端点，代表图的入口与出口。类比生物：状态像细胞外基质，节点像细胞器，边像信号通路，归约器像受体的合并规则。归约器配错，是后面所有「状态没累积」「消息翻倍」类 bug 的根因。

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import add_messages

class S(TypedDict):
    # 消息通道用 add_messages 归约：按 id 去重追加
    msgs: Annotated[list, add_messages]
    step: Annotated[int, operator.add]

def node_a(s: S):
    return {"msgs": [("user", "你好")], "step": 1}

def node_b(s: S):
    return {"msgs": [("ai", "收到")], "step": 1}

b = StateGraph(S)
b.add_node("a", node_a)
b.add_node("b", node_b)
b.add_edge(START, "a")
b.add_edge("a", "b")
b.add_edge("b", END)
g = b.compile()
print(g.invoke({"msgs": [], "step": 0}))
​

# 这一行归约器的差别，决定了跨节点计数是否存在
​

案例：消息去重机制

背景：多轮对话中同一消息不应被重复追加，否则上下文越滚越大且模型会重复看到旧指令。

操作：用 add_messages 作为归约器，返回增量消息列表，框架按 id 自动去重。

结果：重复 invoke 不会让消息翻倍，上下文长度可控。

解读：归约器把「状态如何合并」集中在一处，节点只需关心增量，职责清晰。

变式：自定义归约器可实现「只保留最近 N 条」的滑动窗口，避免长对话把上下文撑爆。

工程清单

• 节点返回的是增量不是全量，合并规则由归约器决定，这个约定是所有状态正确性的根基。

• add_messages 按消息 id 去重追加，避免多轮对话消息翻倍，是消息类状态的事实标准做法。

• START/END 是特殊端点，条件边路由函数返回节点名或 END，路由键要和编译期映射一一对应。

常见误区

把状态字段默认当成覆盖式，结果循环里的前一轮数据被冲掉；需要历史就用 operator.add 或 add_messages，归约器配错是状态类 bug 的头号根因。

归约器对照与选择规则

四种归约语义对应不同的状态字段类型，选错会在循环或并行时暴露问题。下表是选择时的速查表：

选择规则一句话：字段描述「当前焦点」用覆盖，「发生过什么」用累加或消息合并。拿不准时先写成累加，因为覆盖会丢弃历史、几乎无法挽回，而累加最多浪费一点存储。注意 operator.add 对列表是「连接」而非「数字相加」之外的第二个语义，一个字段类型对应一套行为，别混用。

节点返回值与状态通道的对应

节点返回的字典里每个 key 必须落在状态 schema 里，值会按该通道的归约器合并。这条对应关系有三个易错点：

1. 返回了 schema 里不存在的 key，LangGraph 不会报错，只是静默忽略，结果「我以为写了、其实没写」。
2. 返回的 key 名与 schema 差一个字符（user_msg 对 user_message），同样静默丢失。
3. 多个节点并行写同一通道时，没有归约器的字段会直接抛冲突错误——这是设计，逼你显式声明合并语义。

class S(TypedDict):
    msgs: Annotated[list, add_messages]
    final: str

def node_x(s: S):
    # typo：schema 里叫 final，这里写成了 finial，该值被静默丢弃
    return {"msgs": [("ai", "ok")], "finial": "结果"}
​

这类错误编译期拦不住，只能靠「读完一次运行的状态」来发现。建议在联调阶段每次跑完打印一次最终状态，对照预期核对每个字段。

练习：把顺序脚本改写成图

下面是一段命令式脚本，请你先读懂流程，再对照右侧改写成 StateGraph。这是后面所有章节的起点练习。

# 命令式：if/else 决定流程
def run(user_input):
    msgs = [("user", user_input)]
    reply = model.invoke(msgs)
    msgs.append(("ai", reply))
    if "工具" in reply:
        tool_res = call_tool(reply)
        msgs.append(("tool", tool_res))
        msgs = msgs + [("ai", model.invoke(msgs))]
    return msgs
​

# 图式：每一步是一个节点，分支由条件边表达，工具结果回环
b = StateGraph(S)
b.add_node("call_model", call_model)   # 生成回复
b.add_node("use_tool", use_tool)       # 调用工具
b.add_edge(START, "call_model")
b.add_conditional_edges("call_model", need_tool, {"yes": "use_tool", "no": END})
b.add_edge("use_tool", "call_model")   # 工具结果回到模型，形成回边
g = b.compile()
​

对照两张图会发现：命令式版本把流程逻辑藏进 if，图式版本把同样的判断搬进条件边路由函数。改写后每个环节能单独测试、单独可视化，还能在 use_tool 后插入审计或人工确认节点而不动其它部分，这正是三原语组合的价值。