17.1 AgentState 是 LangGraph 中所有节点间流转的公共数据容器

骑士雄师

94人浏览 · 2026-06-09 11:52:56

骑士雄师 · 2026-06-09 11:52:56 发布

AgentState 是 LangGraph 中所有节点间流转的公共数据容器，它的设计和更新机制是高效 Agent 与复杂工作流的关键。

🧠 1. AgentState 概念速记

角色与作用：在 LangGraph 的StateGraph中，State（特指AgentState类）是贯穿所有节点的核心共享数据对象，充当一块"公共黑板"。每个节点（Node）都接收当前 State 作为输入，并返回对该 State 的更新。它为多步、多节点工作流提供了结构化的数据传递和共享方式。
关系辨析：AgentState是 Agent 当前正在使用的可变容器，存储着对话历史和中间计算结果。Checkpoint则是该 State 在某一时刻的不可变快照，用于实现故障恢复与时间旅行。你可以把它们的关系理解为：AgentState像一张草稿纸，而Checkpoint是草稿纸的留影照片。

🧱 2. AgentState 的三种主要形态

🧱 形态一：最简单的 `State`（TypedDict）

这是最基础的自定义方式，只定义一个结构而不指定归约器。比如，你的AgentState中就只定义了需要包含哪些字段。如果不用Annotated指定归约器，节点返回的字段更新默认会直接覆盖旧值。这种方式适用于只需固定值的简单场景。

# 简单状态：每次更新都会覆盖原有值
from typing import TypedDict

class SimpleState(TypedDict):
    counter: int
    user_name: str

🧱 形态二：预置的 `MessagesState` (带 `add_messages`)

LangGraph 为你预置了MessagesState，它是最常见、最基础的状态设计之一。它的核心是messages字段，并配置了add_messages函数作为归约器（Reducer）。这个归约器确保节点返回的新消息会被追加到现有消息列表中，而不是覆盖，从而自然地构建对话历史。这对于构建聊天机器人或 Agent 是至关重要的。

from langgraph.graph import MessagesState

class MyAgentState(MessagesState):
    pass # 也可以扩展其他字段，但继承了 messages 的追加行为

🧱 形态三：自定义 + 归约器（用于复杂场景）

在你的实际业务中，Agent 往往需要记住更多业务数据，比如用户的首选项（温度单位、搜索偏好）、流程状态（是否已搜索、重试次数）、业务字段（订单号、购物车）等。
为此，你会继承TypedDict，并通过Annotated对特定字段指定归约器（Reducer）。归约器定义了节点返回的状态更新如何与现有状态合并。这是官方推荐的最佳实践，因为它能让你明确控制每个字段的更新逻辑。

⚙️ 3. AgentState 状态更新机制详解

这是掌握 AgentState 最核心的部分。当你的节点函数返回一个字典时，LangGraph 不是简单地替换旧状态，而是会根据每个字段的归约规则，将节点的"更新"合并到现有状态中。

(1). 默认归约器（Default Reducer）

如果一个字段没有使用Annotated指定归约器，LangGraph 会使用默认行为：用新值完全覆盖旧值。要覆盖一个字段，节点函数返回该字段的新值即可。

# 状态定义
from typing import TypedDict
class MyState(TypedDict):
    user_input: str      # 默认会覆盖
    step_count: int      # 默认会覆盖

# 节点函数会这样更新状态
def some_node(state: MyState) -> dict:
    return {"user_input": "新的输入", "step_count": state["step_count"] + 1}

(2). 预置归约器：`add_messages`

这是专门用于messages列表的归约器。它的逻辑是：将节点返回的新消息追加到现有消息列表的末尾。

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage

class MyState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages] # 使用 add_messages 归约器
    other_field: str                                    # 默认覆盖

当一个节点return {"messages": [AIMessage(content="...")]}时，该 AIMessage 会被追加，而非替换原有列表。

(3). 自定义归约器（`Annotated` 语法）

官方推荐使用 Annotated[FieldType, ReducerFunction] 语法来为字段绑定归约器，这也是类型注解的最佳实践。

from typing import Annotated, TypedDict

def my_reducer(current: int | None, new: int) -> int:
    """一个自定义归约器：返回当前值和新增值的和"""
    if current is None:
        return new
    return current + new

class MyCustomReducerState(TypedDict):
    accumulated: Annotated[int, my_reducer]

(4). 自定义归约器（旧版 `operator.add` 方式）

对于简单的列表追加，也可以使用operator.add，但可读性不如add_messages。当前主流最佳实践是add_messages。

import operator
from typing import Annotated, TypedDict

class MyCustomState(TypedDict):
    results: Annotated[list[str], operator.add] # 使用 operator.add 实现追加

(5). 更底层的 `StateSchema` 类（补充认知）

作为知识补充，LangGraph 提供了更底层的 StateSchema 类来定义状态，它提供了统一的 API，可与 Zod 等标准 schema 库集成。但在日常使用中，TypedDict + Annotated 的语法更常见。

(6). 归约器函数签名总结

归约器函数的标准签名是 reducer(current: Value, new_value: UpdateValue) -> Value，你编写的逻辑需要基于这个模式来合并新旧数据。

💡 4. 最佳实践与面试高频点

为 List 字段使用归约器：如果想让节点能追加数据到列表字段，务必为其绑定归约器。否则默认是覆盖，可能会丢失信息。
为新增字段绑定归约器：在自定义状态时，根据业务逻辑决定归约策略，是覆盖还是累积。
合理拆分状态：将业务数据、流程状态、对话消息分层存放，结构更清晰。
节点设计原则：节点职责应聚焦于核心业务逻辑，状态管理则交给框架和归约器处理，这符合单一职责原则。
类型与安全：推荐使用TypedDict配合Annotated进行强类型约束，这能提升代码健壮性和可维护性。
文档与可视化：为条件边的路由函数提供Literal类型注解能让代码自文档化，也便于 LangGraph Studio 可视化。

✨ 5. 完整示例代码

以下是一个综合案例，演示如何在一个 LangGraph 工作流中定义和使用包含多种归约器的自定义 State。

import os
from typing import TypedDict, Annotated, Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, END, START
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage

# 1. 加载 .env
load_dotenv()

# 2. 创建模型实例
llm = ChatOpenAI(
    model="qwen-plus",
    temperature=0.7,
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    base_url=os.getenv("DASHSCOPE_BASE_URL", "https://dashscope.aliyuncs.com/compatible-mode/v1"),
    model_kwargs={"extra_body": {"enable_thinking": False}}
)

# 3. 定义自定义 AgentState，展示多种归约器
class AgentState(TypedDict):
    # 使用 LangGraph 内置的 add_messages，用于对话历史追加
    messages: Annotated[list, add_messages]
    
    # 默认覆盖：记录当前用户的温度单位偏好
    temperature_unit: str  
    
    # 自定义归约器：记录一段对话中的模型调用次数
    model_call_count: Annotated[int, lambda x, y: x + y if x is not None else y] 

# 4. 节点函数 1：处理用户请求，更新偏好
def process_request(state: AgentState) -> dict:
    user_msgs = [msg for msg in state["messages"] if isinstance(msg, HumanMessage)]
    if user_msgs:
        content = user_msgs[-1].content
        if "华氏度" in content:
            return {"temperature_unit": "华氏度"}
        elif "摄氏度" in content:
            return {"temperature_unit": "摄氏度"}
    return {}

# 5. 节点函数 2：调用 LLM 并记录调用次数，展示多种归约器的作用
def call_model(state: AgentState) -> dict:
    current_count = state.get("model_call_count", 0)
    # 这个返回体现了 add_messages 和 model_call_count 归约器的同时生效
    response = llm.invoke(state["messages"])
    return {
        "messages": [response],  
        "model_call_count": 1     
    }

# 6. 路由函数：决定后续流程
def should_continue(state: AgentState) -> Literal["continue", "end"]:
    # 连续调用次数不超过 2 次才继续
    if state.get("model_call_count", 0) < 2:
        return "continue"
    return "end"

# 7. 构建图
workflow = StateGraph(AgentState)

workflow.add_node("process_preference", process_request)
workflow.add_node("process_model", call_model)

workflow.set_entry_point("process_preference")
workflow.add_edge("process_preference", "process_model")

# 添加条件循环
workflow.add_conditional_edges(
    "process_model",
    should_continue,
    {
        "continue": "process_model",   # 循环回自身继续
        "end": END
    }
)

app = workflow.compile()

# 8. 运行测试
if __name__ == "__main__":
    # 模拟多轮对话
    initial_state = {
        "messages": [HumanMessage(content="帮我查一下上海现在的温度，我用摄氏度。")],
        "temperature_unit": "摄氏度",
        "model_call_count": 0,
    }
    print("初始状态:", initial_state)
    
    final_state = app.invoke(initial_state)
    
    print("\n=== 最终状态 ===")
    print(f"Temperature Unit: {final_state['temperature_unit']}")
    print(f"Model Call Count: {final_state['model_call_count']}")
    print(f"Messages: {final_state['messages']}")

📝 总结

AgentState是 LangGraph 工作流的"记忆载体"，其通过归约器（Reducer） 机制实现了灵活可控的状态更新。我们可以使用TypedDict定义状态结构，用Annotated为特定字段绑定归约器。在编写 Agent 时，合理设计和运用AgentState（包括消息与业务字段）是保证任务连贯性、正确性和可调试性的基石。