目录
- 前言
- 一、什么是 StateGraph：图、节点、边、状态 Schema
- 二、先跑一个最简状态图：键值状态
- 三、Reducer：状态到底怎么合并
- 四、案例：带记忆的多轮对话 Agent
- 五、工程化要点：Schema、Reducer、导入来源、两条写法
- 附：「Deep Agents 等价实现」贴士
- 参考文献
前言
主线：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 的每一步执行，都是「图里的一个节点读写一份共享状态」。
四个基本概念：
一句话收口：积木（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 里已经有「第一问 + 第一答 + 第二问」，模型看的是完整上下文。
拆一下这份「记忆」是怎么来的：
注意：这里的「记忆」是进程内、图状态里的短期记忆。进程一结束，这份状态就没了；想要「重启后还记得」，要靠 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
🎯 更多专栏系列文章可以查看博客主页📑 👍 若文章对你有所触动，恳请点赞 ⭐ 关注 ⭐ 收藏
