AI Agent 工程化开发实战（三）：StateGraph 与状态管理——节点、边、Reducer

---

前言

主线：Agent Runtime → Harness。从本篇起正式进入 Runtime（运行时）层——LangGraph。第 1 篇立了四层认知，第 2 篇用 create_deep_agent 一行跑通、又把它拆开看到内部是 create_agent（薄壳）+ 一套内置中间件栈，最终落到 StateGraph。这一篇，我们亲手把 StateGraph 写一遍，搞懂它最核心的三个词——图、节点、边——以及状态到底是怎么被「合并」的（Reducer）。

一、什么是 StateGraph：图、节点、边、状态 Schema

先建立心智模型。回想第 1 篇那张调用链图：create_deep_agent → create_agent → StateGraph → langchain-core。StateGraph 是 LangGraph（运行时）最核心的类——它是一个「有状态的图」，Agent 的每一步执行，都是「图里的一个节点读写一份共享状态」。

四个基本概念：

1. 图（Graph）：一组节点 + 连接它们的边，定义「先做什么、再做什么」。
2. 节点（Node）：一个函数，入参是当前状态，返回「我要改动的键值」。它只声明自己改了哪几项，不关心全局。
3. 边（Edge）：连接节点。普通边固定走「A → B」；条件边按运行时结果路由到不同节点——工具调用循环就靠它（讲工具调用循环的篇章会展开）。
4. 状态 Schema（State Schema）：状态长什么样的「类型声明」，用 TypedDict 写，决定图里流通的数据结构。

一句话收口：积木（langchain-core）给了「消息 / 工具 / 模型」这些零件，运行时（LangGraph）给了「图」这条流水线——节点是工位，边是传送带，状态就是流水线上流动的工件。 注意分层铁律：StateGraph 依赖积木，它在积木之上、create_agent（薄壳）之下。

二、先跑一个最简状态图：键值状态

别一上来就上「消息列表 + Reducer」——那是最容易劝退新手的坑。先用一个最简键值状态建立手感：状态就一个整数 count，图就一个节点。

from typing import TypedDict

from langgraph.graph import StateGraph, START, END


class CounterState(TypedDict):
    """状态 Schema：用 TypedDict 声明状态长什么样。"""
    count: int


def increment(state: CounterState) -> dict:
    """节点函数：入参是当前状态，返回「我要改动的键」，运行时帮你合并。"""
    return {"count": state["count"] + 1}


graph = StateGraph(CounterState)
graph.add_node("increment", increment)      # 加一个节点
graph.add_edge(START, "increment")          # 起点 → 节点
graph.add_edge("increment", END)            # 节点 → 终点
app = graph.compile()                       # 编译成可执行的图

print("1) 最简键值状态:", app.invoke({"count": 0}))   # {'count': 1}

跑一遍，输出 1) 最简键值状态: {'count': 1}。这里值得停一下，消化三件事：

• StateGraph(CounterState)：把状态 Schema 交给图，图就知道状态里有 count 这一个键。
• 节点只返回它改动的键：increment 返回 {"count": ...}，是一个「部分更新」。运行时负责把它合并进完整状态——你不用手写「读整个状态、改一项、再写回整个状态」。反过来，如果状态有多个键，节点可以只返回其中一部分，其余键保持不变。
• START / END：LangGraph 内置的虚拟节点，标记图的入口和出口。

三、Reducer：状态到底怎么合并

上面 count 例子里，「合并」就是简单地「新值覆盖旧值」。但这是默认行为——一旦同一个键被多个节点写、或者你想「追加而不是覆盖」，就得显式指定合并规则，官方叫 Reducer（一个 (旧值, 新值) -> 结果 的函数）。

先看默认的「覆盖」是什么效果。两个节点都写同一个 text 键：

from typing import TypedDict

from langgraph.graph import StateGraph, START, END


class TextState(TypedDict):
    text: str


def write_a(state: TextState) -> dict:
    return {"text": "A 节点写的"}


def write_b(state: TextState) -> dict:
    return {"text": "B 节点写的"}   # 同一个键，把 A 的结果覆盖掉


g2 = StateGraph(TextState)
g2.add_node("a", write_a)
g2.add_node("b", write_b)
g2.add_edge(START, "a")
g2.add_edge("a", "b")
g2.add_edge("b", END)
app2 = g2.compile()

print("2) 默认覆盖:", app2.invoke({"text": ""}))   # {'text': 'B 节点写的'}

输出 2) 默认覆盖: {'text': 'B 节点写的'}——后写的 B 覆盖了先写的 A。这就是默认语义：同一个键，后写覆盖前写。

但对「消息列表」这种场景，覆盖是要出事的。多轮对话里，状态里的 messages 是一串消息，模型节点每轮「追加」一条新回复——如果用默认覆盖，每次模型回复都会把之前的历史整个冲掉。所以消息列表要用专门的 Reducer：add_messages。

from typing import Annotated, TypedDict

from langchain_core.messages import AIMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class ChatState(TypedDict):
    # 给 messages 指定 add_messages 作 reducer：追加而非覆盖
    messages: Annotated[list, add_messages]


def append_ai(state: ChatState) -> dict:
    return {"messages": [AIMessage(content="这是 AI 追加的一条")]}


g3 = StateGraph(ChatState)
g3.add_node("append_ai", append_ai)
g3.add_edge(START, "append_ai")
g3.add_edge("append_ai", END)
app3 = g3.compile()

result = app3.invoke({"messages": [HumanMessage(content="用户的第一句")]})
print("3) add_messages 追加:", [(type(m).__name__, m.content) for m in result["messages"]])
# [('HumanMessage', '用户的第一句'), ('AIMessage', '这是 AI 追加的一条')]

看输出：用户的 HumanMessage 没被冲掉，AI 的 AIMessage 被追加到了后面——这就是 add_messages 干的事。

Annotated[list, add_messages] 这行的读法：messages 这个键的类型是 list，合并时用 add_messages 这个函数。Annotated[X, Y] 是 Python 的类型注解语法——第一个参数是类型，第二个参数是「元数据」（这里就是 Reducer）。add_messages 比「简单 append」更聪明：它按消息 ID 去重，同 ID 的新消息会替换旧消息——这正是「工具调用被结果替换」所需的语义（讲工具调用循环的篇章会用上）。

导入来源：add_messages 从 langgraph.graph.message 导入；StateGraph / START / END 从 langgraph.graph 导入。

以上三个例子合起来就是 code/03-StateGraph与状态管理/basics.py（真源，可直接运行）；文章为讲解方便按例拆开贴出、各自带 import，与真源逻辑一致。

四、案例：带记忆的多轮对话 Agent

把 add_messages 用起来，做一个真正「带记忆」的多轮对话 Agent。这个案例没有工具（工具是下一篇章的活），只有「模型节点 + 消息状态」，重点看状态如何跨轮累积。模型仍用 DeepSeek（换模型的成本见第 2 篇：同厂商改一行、换厂商改 import + model 两行）。

运行前先装依赖并配 key：pip install langgraph langchain-deepseek，再设置 DEEPSEEK_API_KEY（若已按第 2 篇装过 deepagents，langgraph 已随依赖装上，可跳过装包）。

import sys
from typing import Annotated, TypedDict

from langchain_core.messages import HumanMessage, SystemMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

# Windows 控制台默认 GBK，模型回复若带 emoji，print 会抛 UnicodeEncodeError；强制 UTF-8 输出
sys.stdout.reconfigure(encoding="utf-8")


class ChatState(TypedDict):
    """带记忆的对话状态：messages 用 add_messages 追加，而非默认覆盖。"""
    messages: Annotated[list, add_messages]


model = ChatDeepSeek(model="deepseek-chat")


def call_model(state: ChatState) -> dict:
    """模型节点：读入当前消息，返回模型的回复。"""
    return {"messages": [model.invoke(state["messages"])]}


graph = StateGraph(ChatState)
graph.add_node("model", call_model)
graph.add_edge(START, "model")
graph.add_edge("model", END)
app = graph.compile()

# 手动把历史喂回去，演示「带记忆」——讲持久化与断点恢复（Checkpointer）的篇章会自动持久化，这里先手动
messages = [SystemMessage(content="你是一个简洁的助手，回答尽量简短，不超过两句话。")]
for question in ["你好，我叫小明，帮我记一下。", "我刚才说我叫什么名字？"]:
    messages.append(HumanMessage(content=question))
    result = app.invoke({"messages": messages})
    messages = result["messages"]          # 状态里累积了历史，下一轮带着它走
    print(f"你：{question}")
    print(f"AI：{result['messages'][-1].content}")
    print("---")

跑一遍，输出大概是（每次措辞可能不同，关键看第二问是否出现「小明」）：

你：你好，我叫小明，帮我记一下。
AI：你好，小明！很高兴认识你。已记住你的名字，有什么需要帮忙的吗？
---
你：我刚才说我叫什么名字？
AI：你叫小明。有什么我可以帮你的吗？
---

关键在第二问：模型能答出「小明」，不是因为它有什么魔法，而是因为状态里的 messages 把第一轮的历史累积下来了——第二轮 invoke 时，messages 里已经有「第一问 + 第一答 + 第二问」，模型看的是完整上下文。

拆一下这份「记忆」是怎么来的：

1. messages: Annotated[list, add_messages] 让 call_model 返回的 [AIMessage] 是「追加」而非「覆盖」——否则每轮模型回复会把历史冲掉，第二轮就只剩一条 AI 回复、没有上下文了。
2. messages = result["messages"] 把累积后的完整状态取出来，作为下一轮的输入喂回去。

注意：这里的「记忆」是进程内、图状态里的短期记忆。进程一结束，这份状态就没了；想要「重启后还记得」，要靠 Checkpointer 把状态持久化——那是「持久化与断点恢复」篇章的活，本篇先把「状态累积」这件事立住。

（本文代码实测于 langgraph 1.2.14 / langchain-core 1.6.9 / langchain-deepseek 1.1.1；完整真源见 code/03-StateGraph与状态管理/conversation_agent.py。）

五、工程化要点：Schema、Reducer、导入来源、两条写法

① 状态 Schema 设计：用 TypedDict，别用裸 dict。 TypedDict 给你 IDE 补全和类型检查，键写错了编辑器会红——状态键一多，这是保命的东西。节点返回的 dict 是「部分更新」，只写自己改的键即可。

② Reducer 的选择。 默认是「覆盖」，够用就别加戏；只有三件事需要显式指定：

• 要追加（消息、日志）→ 用 add_messages 或自定义 reducer；
• 要按 ID 合并 / 去重（消息）→ add_messages 现成；
• 要累加 / 取最新 / 并集等业务语义 → 写一个 (旧, 新) -> 结果 的函数，用 Annotated[类型, 你的函数] 挂上。

另一个坑：如果同一个键被并行分支同时写（fan-in）而它又没挂 Reducer，LangGraph 会抛 InvalidUpdateError——运行时不知道两个并行的值怎么合并。所以只要一个键可能被多个节点「同时」写，就必须显式挂 Reducer。

③ 导入来源。 add_messages 从 langgraph.graph.message 导入；StateGraph / START / END 从 langgraph.graph 导入。别照旧教程写 MessageGraph——它在 LangGraph 1.0 已弃用，统一用 StateGraph + messages 键（配合 add_messages）。

④ StateGraph vs Functional API。 LangGraph 有两条写法：你刚学的 StateGraph（显式声明节点/边），和 Functional API（用 @entrypoint / @task 装饰器，把每个节点写成一个函数、靠调用关系自动建图）。二者跑在同一个运行时上，是同一枚硬币的两面。本专栏主线用 StateGraph——因为它更贴近 create_agent 内部的真实实现，你能「看清图长什么样」；Functional API 是它的函数式等价写法，适合图结构简单、更想用「顺序代码」表达的场景，后面在合适处点到为止。

附：「Deep Agents 等价实现」贴士

Deep Agents 等价实现
本文手写的「StateGraph + add_messages 消息状态」，在薄壳 / 套件层的等价写法是一行 create_agent / create_deep_agent——它们内部正是用这张图搭的。例如最普通的「模型 + 消息」循环：

from langchain.agents import create_agent

agent = create_agent(model=model, tools=[], system_prompt="你是简洁的助手")
result = agent.invoke({"messages": [{"role": "user", "content": "你好"}]})

需要自己掌控状态结构（自定义 Reducer、增删节点）时，再回到本文的底层写法。

参考文献

[1] LangGraph 官方文档 · Overview — https://docs.langchain.com/oss/python/langgraph/overview
[2] LangGraph 官方文档 · Graph API（节点 / 边 / Reducer）— https://docs.langchain.com/oss/python/langgraph/graph-api
[3] LangGraph API 参考 · StateGraph — https://reference.langchain.com/python/langgraph/graph/state/StateGraph
[4] LangGraph API 参考 · add_messages — https://reference.langchain.com/python/langgraph/graph/message/add_messages

---

🎯 更多专栏系列文章可以查看博客主页📑 👍 若文章对你有所触动，恳请点赞 ⭐ 关注 ⭐ 收藏

转载自 CSDN-专业IT技术社区

原文链接：https://blog.csdn.net/xiaobing259/article/details/168774285