LangGraph 第6章：State 状态管理

State 是 LangGraph 的核心抽象，它连接了所有节点和数据流。本章深入介绍状态管理的各种方式和高级特性。

定义状态的三种方式

1. TypedDict（推荐入门）

使用 Python 的 TypedDict 定义状态结构，简单直观：

from typing import TypedDict, Annotated, List
import operator
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage

class MyState(TypedDict):
    # 简单字段：后一个值覆盖前一个
    input_text: str
    counter: int
    # 带 reducer 的列表字段：每次追加而非覆盖
    steps: Annotated[List[str], operator.add]
    # 消息列表：使用 LangGraph 内置的 add_messages reducer
    messages: Annotated[List[BaseMessage], add_messages]

2. Pydantic BaseModel（推荐生产）

使用 Pydantic 提供更强的类型校验和序列化能力：

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

class PydanticState(BaseModel):
    input_text: str = Field(description="用户输入文本")
    counter: int = Field(default=0, ge=0)
    steps: Annotated[List[str], operator.add] = Field(default_factory=list)
    messages: Annotated[list, add_messages] = Field(default_factory=list)

    class Config:
        arbitrary_types_allowed = True

Pydantic 的优势：

• 自动类型校验：赋值时会检查类型是否正确
• 默认值：通过 Field(default=...) 设置
• 序列化：轻松转为 JSON 或字典
• 文档生成：自动生成 API 文档

3. MessagesState（快捷方式）

对于对话类应用，可以直接使用内置的 MessagesState：

from langgraph.graph import MessagesState
from langchain_core.messages import AnyMessage

# MessagesState 已包含:
# messages: Annotated[List[AnyMessage], add_messages]

class ChatState(MessagesState):
    # 可以添加额外字段
    user_id: str
    session_metadata: dict

三种方式的对比：

State 的读写规则

读取

节点通过函数参数接收状态，直接读取：

def my_node(state: MyState):
    # 直接读取字段
    text = state["input_text"]

    # 使用 .get() 安全读取
    counter = state.get("counter", 0)

    # 读取列表（带 reducer 的字段）
    messages = state.get("messages", [])

写入

节点返回字典，只返回需要更新的字段：

def my_node(state: MyState):
    # 只返回需要更新的字段
    return {
        "counter": state["counter"] + 1,
        "steps": ["node executed"]
    }

重要规则：

1. 返回的字典会与当前状态合并（非覆盖）
2. 对带 reducer 的字段，reducer 决定合并方式
3. 不想修改的字段不需要出现在返回值中

Reducer 机制详解

Reducer 是 LangGraph 状态管理的核心机制，它定义了当多个节点更新同一个字段时，如何处理这些更新。

内置 Reducer

from typing import Annotated
import operator
from langgraph.graph import add_messages

class StateWithReducers(TypedDict):
    # 1. 默认 Reducer：直接覆盖
    simple_value: str

    # 2. operator.add：列表拼接
    collected: Annotated[list, operator.add]

    # 3. add_messages：消息列表的特殊合并
    messages: Annotated[list, add_messages]

add_messages 的工作原理

from langgraph.graph import add_messages
from langchain_core.messages import HumanMessage, AIMessage

# 初始消息列表
existing = [HumanMessage(content="你好")]
# 新消息
new_msg = AIMessage(content="你好！有什么可以帮助你的？")

# add_messages 将新消息追加到列表末尾
result = add_messages(existing, new_msg)
# result: [HumanMessage("你好"), AIMessage("你好！有什么可以帮助你的？")]

对于消息，如果新消息的 id 与已有消息相同，会覆盖而不是追加：

from langchain_core.messages import AIMessage

# id 相同的消息会覆盖
update = AIMessage(content="更新后的回复", id="msg_1")
result = add_messages(
    [AIMessage(content="原始回复", id="msg_1")],
    update
)
# result: [AIMessage("更新后的回复")]

自定义 Reducer

你也可以定义自己的 reducer 函数：

from typing import Annotated

def my_reducer(current_value, new_value):
    """自定义 reducer：取两个值中较大的"""
    if current_value is None:
        return new_value
    return max(current_value, new_value)

class CustomState(TypedDict):
    max_score: Annotated[int, my_reducer]

状态合并示例

# 定义状态
class MergeExample(TypedDict):
    name: Annotated[str, operator.add]
    score: Annotated[int, max_reducer]
    tags: Annotated[list, operator.add]
    messages: Annotated[list, add_messages]

# 初始状态
initial = {
    "name": "Alice",
    "score": 50,
    "tags": ["beginner"],
    "messages": [HumanMessage(content="hi")]
}

# 节点1 返回
node1_output = {"name": " + Bob", "score": 80, "tags": ["student"], "messages": [AIMessage(content="hello")]}

# 合并后状态
# name: "Alice + Bob"    (operator.add 字符串拼接)
# score: 80              (max_reducer 取大值)
# tags: ["beginner", "student"]  (operator.add 列表拼接)
# messages: [Human("hi"), AIMessage("hello")]  (add_messages 追加)

最佳实践

1. 保持状态扁平化：避免深层嵌套的字典，每个字段对应一个独立的概念
2. 最小化状态：只包含节点间需要共享的数据，局部变量放在节点内部
3. 合理使用 Reducer：理解默认（覆盖）、add（追加）、add_messages（消息合并）的区别
4. 生产环境使用 Pydantic：获得更好的类型安全和数据校验

好的状态管理是 LangGraph 应用的基石。下一章我们将深入节点的各种实现方式。