状态图：节点与边

上一章的最小图已经用上了 StateGraph。本章把三个基础零件讲透：State（状态）、节点、边，重点是两个容易困惑的概念——Reducer 与条件边，它们是多轮循环与多步协作的地基。最后给一个打印执行顺序的可跑示例，帮你把"图是怎么跑起来的"看明白。

State：用 TypedDict 描述共享黑板

图里所有节点读写同一份状态。用 TypedDict 声明有哪些字段，字段名即"黑板上的格子"：

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

class State(TypedDict):
    query: str                                     # 普通字段：后写的直接覆盖
    messages: Annotated[list, add_messages]        # 带 Reducer：自动累加

为什么 messages 要加 Annotated 与 add_messages？因为多个节点/多轮都会往历史里追加消息，如果按普通字段处理，后面的返回会把前面覆盖掉；带上 Reducer 后，框架知道"新消息应该追加进列表"。

Reducer：字段冲突时怎么合并

Reducer 就是"当多个更新要写同一个字段时，如何合并"的规则。默认规则是覆盖（后到者胜）；写 Annotated[list, add]（from operator import add）则变成拼接。看一个不用模型也能跑的最小例子：

from operator import add
from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    step: int                           # 覆盖：谁的更新在后面，step 就是谁
    log: Annotated[list, add]           # 累加：两次返回值拼成一条列表

def node_a(state):
    return {"step": 1, "log": ["A 执行了一次"]}

def node_b(state):
    return {"step": 2, "log": ["B 执行了一次"]}

builder = StateGraph(State)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)
graph = builder.compile()

print(graph.invoke({"step": 0, "log": []}))
# 输出：{'step': 2, 'log': ['A 执行了一次', 'B 执行了一次']}

可见 step 最终是 2（覆盖），log 把两段内容都留了下来（累加）。对话历史正是用同类机制累积：add_messages 是专门为消息设计的 Reducer，会按消息 id 去重并支持"同一 id 更新"，比裸 operator.add 更稳。

节点：state -> 部分更新的纯函数

节点遵守两条约定：

读：state 是整个状态的字典，按字段名取值即可。
写：不要原地改 state，返回一个"部分更新"字典，框架负责合并。这样的纯函数便于单独测试：传入构造好的 state，断言返回值。模型调用、工具执行、本地计算都可以包成节点。

边与条件边

builder.add_edge("a", "b")          # 固定边：a 跑完必去 b
builder.add_edge("a", END)          # 去出口

def route(state):
    return "退款" if "退" in state["text"] else "咨询"

# 条件边：route 返回映射里的键，图按映射走向对应节点
builder.add_conditional_edges("a", route, {"退款": "b", "咨询": "c"})

条件边就是"if 分支"的图化：函数只看状态、返回一个去向键。想让某个节点反复执行（形成循环），让它自己的条件边指回自己即可——这正是第 18 章 ReAct 循环的地基。

可跑示例：打印每一步理解执行顺序

用"工单分流"演示：先统一预处理，再按内容条件分流到不同处理组。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class Ticket(TypedDict):
    text: str
    result: str

def preprocess(state):
    print(">>> preprocess 运行")
    return {"result": "预处理完成"}

def route(state):
    return "退款" if "退" in state["text"] else "咨询"

def refund(state):
    print(">>> refund 运行（退款组）")
    return {"result": "已转人工退款专员"}

def consult(state):
    print(">>> consult 运行（咨询组）")
    return {"result": "已自动回复商品介绍"}

builder = StateGraph(Ticket)
builder.add_node("preprocess", preprocess)
builder.add_node("refund", refund)
builder.add_node("consult", consult)
builder.add_edge(START, "preprocess")
builder.add_conditional_edges("preprocess", route, {"退款": "refund", "咨询": "consult"})
builder.add_edge("refund", END)
builder.add_edge("consult", END)
graph = builder.compile()

graph.invoke({"text": "我要退款"})
# 控制台输出：
# >>> preprocess 运行
# >>> refund 运行（退款组）

graph.invoke({"text": "怎么购买"})
# 控制台输出：
# >>> preprocess 运行
# >>> consult 运行（咨询组）

两段运行都先经过 preprocess，随后条件边按文本内容分流，只有命中分支的节点被执行——执行路径是"边"决定的，不是代码书写顺序。

状态字段设计经验

需要"多轮追加"的内容用 Reducer 字段：对话历史用 add_messages，通用列表用 add（operator.add）。
需要"只留最新值"的内容用普通字段：当前提问、最终结论、路由依据等。
布尔或枚举字段常与条件边搭配：一个字段决定走哪个分支，路由一眼可读。
只放会被多个节点共享的字段；一次性中间量尽量在节点内部消化，避免状态膨胀。

常见问题

节点返回空 dict {} 表示什么？什么都不改，只执行副作用（打印、发请求、触发外部动作）。
节点可以是异步函数吗？可以，async def 节点要配合 graph.ainvoke/astream 使用（第 28 章服务端会用到）。
图会无限循环吗？框架默认有递归步数上限，超过会抛错；需要长循环时要么提高上限，要么让条件边具备明确的结束路径。

小结：State 用 TypedDict 定义共享状态，节点是 state->dict 的纯函数，Reducer（add/add_messages）决定字段合并规则，条件边把 if 与循环变成图结构；掌握这四样，就能自由组合出任意多步流程。

笔记加载中…

标签：#Agent实战

返回知识库