定义与组织状态

这是 LangGraph 的核心第一课。状态（State）是整个图运转时所有节点共享的"黑板"，设计得好不好，直接决定你的工作流是否清晰、能否扩展。

一、状态到底是什么

在 LangGraph 里，状态就是一个普通的 Python 字典，但用 TypedDict 或 Pydantic 模型描述了它的字段和类型。图执行过程中，每个节点都会收到一份当前状态，处理后返回一个部分更新，框架会把它合并回主状态。

flowchart LR
    S1[初始状态] --> N1[节点A]
    N1 -->|返回更新| S2[合并后状态]
    S2 --> N2[节点B]
    N2 -->|返回更新| S3[最终状态]

如果你有 Java/LangGraph4j 背景，可以把 State 理解为一个带类型的 Map<String, Object>，但比 Java 更灵活：你可以用 Annotated 给字段挂一个 reducer（归约函数），定义"如何合并新旧值"。

二、用 TypedDict 定义多字段状态

最常见、最推荐的写法是 TypedDict：

from typing import TypedDict, Annotated
from langgraph.graph import add_messages  # 内置的消息追加 reducer

class RAGState(TypedDict):
    # 输入区：用户原始问题
    user_input: str
    # 中间区：检索到的文档片段
    retrieved_docs: list[str]
    # 消息历史（带 reducer，自动追加而非覆盖）
    messages: Annotated[list, add_messages]
    # 迭代次数：用于限制重试
    iteration_count: int
    # 输出区：最终答案
    answer: str

字段说明

字段	类型	作用	合并方式
user_input	str	用户问题	直接覆盖
retrieved_docs	list[str]	检索结果	直接覆盖
messages	Annotated[list, add_messages]	对话历史	追加（去重/同 id 更新）
iteration_count	int	重试计数	直接覆盖
answer	str	最终输出	直接覆盖

三、什么时候用 Annotated + reducer

默认情况下，节点返回 {"key": value} 会直接覆盖旧值。但有些字段你需要"追加"语义，典型场景：

• 消息历史：每轮回复要 append，而不是覆盖前面的对话。
• 并行汇聚的列表：多个并行节点都要往同一个 list 里塞结果，必须用 reducer 合并，否则会冲突报错。

写法：

from operator import add
from typing import Annotated

class State(TypedDict):
    # 自定义 reducer：用 operator.add，list 就追加，int 就累加
    docs: Annotated[list[str], add]        # 追加而不是覆盖
    score: Annotated[int, add]            # 累加而不是覆盖
    messages: Annotated[list, add_messages]  # 消息专用 reducer

规则记牢：凡是有多节点要"往里加"的字段，就必须挂 reducer，否则并行执行时会报 InvalidUpdateError。详见 并行与 MapReduce。

四、用 Pydantic v2 做状态（带校验）

当你想对状态做类型校验、默认值、字段约束时，Pydantic 是更好的选择：

from pydantic import BaseModel, Field
from typing import Annotated
from langgraph.graph.message import add_messages

class RAGState(BaseModel):
    user_input: str = Field(..., description="用户原始问题")
    retrieved_docs: list[str] = Field(default_factory=list)
    messages: Annotated[list, add_messages] = []
    iteration_count: int = Field(default=0, ge=0)  # 必须 >=0
    answer: str = ""

Pydantic 状态的好处：节点收到状态时类型是确定的，传错类型会立即抛错，而不是等到运行到一半才崩。新手推荐先用 TypedDict（更轻量），项目复杂后再切 Pydantic。

五、状态字段命名与分层建议

为了让工作流可读、可维护，建议把状态字段分三个区：

┌─────────────────────────────────────┐
│ 输入区   user_input / question / file │  ← 用户/外部进来
├─────────────────────────────────────┤
│ 中间区   retrieved_docs / draft /     │  ← 节点间传递
│          scores / iteration_count     │
├─────────────────────────────────────┤
│ 输出区   answer / summary / result    │  ← 最终产出
└─────────────────────────────────────┘

命名建议：

• 用名词，不要用动词（retrieved_docs 而不是 retrieve）。
• 计数类字段统一后缀 _count 或加 iteration_ 前缀。
• 列表字段用复数（docs、messages）。

六、完整示例：检索-生成状态

下面是一个完整可运行的"检索-生成"状态定义与使用：

from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. 定义状态
class RAGState(TypedDict):
    user_input: str
    retrieved_docs: list[str]
    messages: Annotated[list, add_messages]
    answer: str

# 2. 定义节点（节点只读写状态字段）
def retrieve(state: RAGState) -> dict:
    query = state["user_input"]
    # 这里假装去检索，实际接你的向量库
    docs = [f"关于「{query}」的文档片段1", f"关于「{query}」的文档片段2"]
    return {"retrieved_docs": docs}   # 只返回要更新的字段

def generate(state: RAGState) -> dict:
    docs = state["retrieved_docs"]
    answer = f"基于 {len(docs)} 个片段生成的答案。"
    return {"answer": answer}

# 3. 连接图
graph = StateGraph(RAGState)
graph.add_node("retrieve", retrieve)
graph.add_node("generate", generate)
graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "generate")
graph.add_edge("generate", END)

app = graph.compile()

# 4. 运行
result = app.invoke({"user_input": "LangGraph 是什么？"})
print(result["answer"])  # 基于 2 个片段生成的答案。

运行：

python rag_state.py

七、常见踩坑

踩坑 1：状态塞了不可序列化的对象

把文件句柄、数据库连接、lambda、自定义类的实例塞进状态，会导致 MemorySaver / SqliteSaver 检查点保存失败。

# ❌ 错误：状态里放了连接对象
class BadState(TypedDict):
    conn: object  # 永远不要这么做

# ✅ 正确：状态只放可序列化数据，连接在节点内部用完即关
def node(state):
    conn = create_connection()  # 局部变量
    ...
    return {"data": rows}      # 只返回数据

踩坑 2：状态过大导致检查点变慢

把整个语料库或大段文本都塞进状态，每一步都要序列化存检查点，会很慢。只放索引或指针，需要时再去外部存储取。

踩坑 3：返回了 State 里不存在的 key

节点返回 {"summary": "..."} 但 State 里没有 summary 字段，LangGraph 会直接忽略（TypedDict 模式）或报错（严格模式）。返回前对照 State 定义检查一遍。

踩坑 4：把 messages 当普通 list 却不挂 reducer

messages 如果不写 Annotated[list, add_messages]，每次节点返回 {"messages": [...]} 会覆盖历史，多轮对话就"失忆"了。这是新手最高频的 bug。

八、小结

• 状态是图的共享黑板，用 TypedDict（轻量）或 Pydantic（带校验）定义。
• 需要"追加"语义的字段（消息、并行汇聚列表）必须挂 Annotated[field, reducer]。
• 字段按"输入区/中间区/输出区"分层，命名用名词、复数表示列表。
• 状态只放可序列化的轻量数据，重资源在节点内局部使用。

下一篇 创建与连接节点 讲如何往这些状态上挂节点。