状态管理与类型系统

状态（State）是 LangGraph 的核心——所有节点通过共享状态来通信，边的条件路由也依赖状态。理解状态管理是掌握 LangGraph 的关键。

状态定义方式

LangGraph 支持三种状态定义方式，适用于不同场景：

1. TypedDict（推荐，轻量）

最常用的方式，纯类型注解，无运行时开销：

from typing import TypedDict

class AgentState(TypedDict):
    user_input: str
    messages: list
    next_action: str
    iteration_count: int

优点：轻量、无额外依赖、IDE 类型提示友好 缺点：无运行时验证，传入错误类型不会报错

2. Pydantic BaseModel（带验证）

需要运行时数据验证时使用：

from pydantic import BaseModel, Field

class AgentState(BaseModel):
    user_input: str = Field(description="用户输入")
    messages: list = Field(default_factory=list)
    next_action: str = Field(default="think")
    iteration_count: int = Field(default=0, ge=0)

    class Config:
        arbitrary_types_allowed = True

优点：运行时验证、默认值、序列化支持 缺点：有 Pydantic 依赖、创建实例稍慢

3. Dataclass（Python 原生）

介于 TypedDict 和 Pydantic 之间：

from dataclasses import dataclass, field

@dataclass
class AgentState:
    user_input: str = ""
    messages: list = field(default_factory=list)
    next_action: str = "think"
    iteration_count: int = 0

优点：Python 原生、有基本的类型提示 缺点：无运行时验证、不如 Pydantic 灵活

三种方式的对比

特性	TypedDict	Pydantic	Dataclass
运行时验证	❌	✅	❌
默认值	❌	✅	✅
序列化	手动	内置	手动
类型提示	✅	✅	✅
性能	最快	较慢	快
适用场景	大多数场景	需要验证时	简单场景

建议：大多数情况下使用 TypedDict，需要数据验证（如接收外部输入）时使用 Pydantic。

Annotated 与 Reducer

这是 LangGraph 状态管理中最重要也最容易混淆的概念。

问题：状态更新的冲突

假设状态中有一个消息列表，两个节点都要往里追加消息：

class State(TypedDict):
    messages: list

def node_a(state: State) -> dict:
    return {"messages": ["消息A"]}

def node_b(state: State) -> dict:
    return {"messages": ["消息B"]}

如果两个节点并行执行，它们的返回值 {"messages": ["消息A"]} 和 {"messages": ["消息B"]} 会互相覆盖，而不是追加。这就是需要 Reducer 的原因。

Reducer：定义合并策略

Reducer 是一个函数，告诉 LangGraph 如何合并两个同名字段的更新值。使用 typing.Annotated 来指定：

from typing import Annotated

def add_messages(existing: list, new: list) -> list:
    """将新消息追加到已有消息列表"""
    return existing + new

class State(TypedDict):
    messages: Annotated[list, add_messages]  # 使用 Reducer
    current_step: str  # 普通字段，直接覆盖

现在两个节点同时更新 messages 时，结果会是 ["消息A", "消息B"]，而不是覆盖。

内置 Reducer：add_messages

LangGraph 提供了一个常用的内置 Reducer——add_messages，专门用于消息列表的追加：

from typing import Annotated
from langgraph.graph.message import add_messages
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    # 等价于自定义的 add_messages 函数

add_messages 的特殊之处在于，它不只是简单追加，还会根据消息的 id 进行去重和替换：

from langchain_core.messages import HumanMessage, AIMessage

# 初始消息
existing = [HumanMessage(content="你好", id="msg-1")]

# 新消息（有相同 id 的会被替换）
new = [AIMessage(content="你好！", id="msg-2")]

# add_messages 结果：
# [HumanMessage(content="你好", id="msg-1"), AIMessage(content="你好！", id="msg-2")]

自定义 Reducer

你可以定义任意的合并逻辑：

import operator
from typing import Annotated

# 使用 operator 模块
class State(TypedDict):
    # 数值累加
    total_score: Annotated[int, operator.add]
    # 列表追加
    items: Annotated[list, operator.add]
    # 集合合并
    tags: Annotated[set, operator.or_]

# 自定义 Reducer
def keep_max(existing: int, new: int) -> int:
    """保留较大值"""
    return max(existing, new)

class State2(TypedDict):
    max_score: Annotated[int, keep_max]

Reducer 的执行时机

Reducer 在每个节点执行之后被调用，用于合并节点的返回值和当前状态：

当前状态: {"messages": [msg1, msg2]}
    ↓
节点返回: {"messages": [msg3]}
    ↓
Reducer 调用: add_messages([msg1, msg2], [msg3])
    ↓
新状态: {"messages": [msg1, msg2, msg3]}

消息管理

LangGraph 与 LangChain 的消息系统深度集成。理解消息类型对构建 Agent 至关重要。

消息类型

from langchain_core.messages import (
    SystemMessage,    # 系统提示
    HumanMessage,     # 用户消息
    AIMessage,        # AI 回复
    ToolMessage,      # 工具调用结果
)

# 典型的消息序列
messages = [
    SystemMessage(content="你是一个有用的助手"),
    HumanMessage(content="今天天气怎么样？"),
    AIMessage(content="我需要查询天气工具", tool_calls=[...]),
    ToolMessage(content="晴天，25°C", tool_call_id="..."),
    AIMessage(content="今天天气晴朗，温度25°C。"),
]

消息与状态的集成

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

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    tool_results: Annotated[list, add_messages]

def agent_node(state: AgentState) -> dict:
    """调用 LLM，可能产生 tool_calls"""
    response = llm.bind_tools(tools).invoke(state["messages"])
    return {"messages": [response]}

def tool_node(state: AgentState) -> dict:
    """执行工具调用"""
    last_message = state["messages"][-1]
    results = []
    for tool_call in last_message.tool_calls:
        result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
        results.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"],
        ))
    return {"messages": results}

消息修剪（Trimming）

长对话会导致 token 超限。LangGraph 提供了内置的消息修剪功能：

from langgraph.graph.message import trim_messages

# 保留最近的 10 条消息
trimmed = trim_messages(
    state["messages"],
    max_tokens=4000,
    strategy="last",       # 保留最后 N 条
    token_counter=len,     # 简单计数（实际用 tiktoken）
)

复杂状态示例

一个完整的多步骤 Agent 可能需要这样的状态：

from typing import Annotated, Optional
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
from pydantic import BaseModel

# 工具执行结果
class ToolResult(BaseModel):
    tool_name: str
    result: str
    success: bool

# 图的状态
class AgentState(TypedDict):
    # 核心：消息列表，使用 add_messages 自动追加
    messages: Annotated[list, add_messages]

    # 当前执行步骤
    current_step: str

    # 迭代计数（用于防止无限循环）
    iteration_count: int

    # 工具调用历史
    tool_history: Annotated[list[dict], lambda a, b: a + b]

    # 最终输出
    final_answer: Optional[str]

    # 错误信息
    error: Optional[str]

状态的不可变性

LangGraph 中状态遵循函数式更新原则：节点不直接修改状态，而是返回一个包含新值的字典，框架负责合并。

# ❌ 错误：直接修改状态
def bad_node(state: AgentState) -> dict:
    state["messages"].append("新消息")  # 直接修改！
    return {}

# ✅ 正确：返回更新字典
def good_node(state: AgentState) -> dict:
    return {"messages": ["新消息"]}  # 通过 Reducer 合并

小结

概念	说明
TypedDict	轻量状态定义，推荐大多数场景
Pydantic	带验证的状态定义
Annotated	为字段附加 Reducer
Reducer	定义状态字段的合并策略
add_messages	内置 Reducer，消息追加+去重
消息类型	SystemMessage / HumanMessage / AIMessage / ToolMessage

下一篇

→ 节点与边的高级用法：条件路由、Send API、Map-Reduce 模式、子图嵌套。