LangChain AgentState 状态管理

Agent 在运行过程中需要维护状态，例如对话消息、工具执行结果、学习进度以及结构化输出。

在 LangChain 1.x 中，AgentState 负责定义 Agent 的状态结构，而 LangGraph 负责状态的更新、合并和管理。

本章节介绍 AgentState 的默认字段、Reducer 合并机制、Middleware 流程控制、自定义状态，以及如何通过 Command 和 Checkpointer 管理状态。

一、AgentState 基本结构

AgentState 基于 Python 的 TypedDict，主要包含以下三个字段：

可以通过 Python 查看当前版本的定义：

其中最常用的是 messages，保存用户消息、AI 回复以及工具执行结果。

注意：AgentState 只是状态结构定义，并不意味着数据已经保存到数据库。是否能够跨调用恢复，需要配置 Checkpointer。

二、messages：消息历史与 Reducer

messages 使用 add_messages Reducer，负责合并新旧消息。

它支持三种主要操作：

• 追加：新消息 ID 不存在时，添加到消息列表。
• 替换：新消息 ID 与已有消息相同时，替换对应消息。
• 删除：通过 RemoveMessage 删除指定 ID 的消息。

1、追加消息

输出：

合并前： 2
合并后： 3
1 human 你好
2 ai 你好！
3 human 介绍 Python

2、相同 ID 替换消息

输出：

1 修改后的回答

这里比较的是消息 id，而不是消息内容或消息类型。

3、RemoveMessage 删除消息

输出：

1 你好
3 介绍 Python

注意：删除 Tool Calling 消息时，应同时考虑 AIMessage 和 ToolMessage 的关联关系，避免产生不符合模型 API 要求的消息历史。

三、jump_to：Middleware 流程控制

jump_to 用于让 Middleware 在特定阶段控制 Agent 的下一步执行方向。

jump_to 属于瞬态状态，用于控制执行流程，不适合保存长期业务数据。

实例：提前结束 Agent

下面使用 @before_model 拦截一个演示关键词，展示如何跳过模型调用。

输出：

请求已被演示规则拦截。

这里的中间件在模型执行前检查消息，如果符合条件，就添加一条 AIMessage 并跳转到结束流程。

注意：jump_to="tools" 不能凭空执行任意 Python 函数；工具节点仍然需要有效的 Tool Calling 请求。

四、structured_response：结构化输出

如果通过 create_agent() 的 response_format 参数设置结构化输出，Agent 会将解析结果存入 structured_response。

这个字段通常不需要用户在输入 State 时提供，但可以从执行完成后的返回值中获取。

实例：推荐课程

如果模型正确完成结构化输出，可能得到：

课程： Python 基础教程
理由： Python 语法简洁，适合零基础学习
难度： 入门

返回的是符合定义的 Pydantic 数据结构，而不是必须自行从自然语言中提取字段。

具体的结构化输出机制由模型及 LangChain 集成决定。模型不支持、Schema 不兼容或解析失败时，可能抛出异常。

五、自定义 AgentState

除了默认字段，还可以通过继承 AgentState 添加业务状态。

例如，学习助手可以保存学习等级、已完成主题和当前目标。

其中 NotRequired 表示初始状态可以省略该字段，但不会自动设置默认值。

读取时建议使用：

level = state.get("user_level", "入门")
topics = state.get("completed_topics", [])

创建 Agent 时指定：

agent = create_agent(
    model="deepseek:deepseek-flash",
    tools=[],
    state_schema=LearningAgentState
)

注意，自定义 State 仅负责描述字段，工具需要通过正式的状态更新机制才能修改这些字段。

六、ToolRuntime：读取 Agent 状态

LangChain 1.x 推荐通过 ToolRuntime 访问当前 Agent State。

例如，可以在工具中读取学习等级和已完成主题：

runtime 由 Agent 执行框架注入，模型不需要提供该参数。

因此，不应假设直接执行 view_progress.invoke({}) 就能获得完整 Agent State。

如果需要修改 State，应使用下一节介绍的 Command。

七、Command：更新 AgentState

直接修改 runtime.state 中的字典或列表，并不是可靠的状态更新方式。

推荐使用 Command(update=...) 提交修改。

实例：记录学习主题

这里通过 Command 同时更新两个字段：

• completed_topics：保存新的学习主题列表。
• messages：追加一条与当前调用 ID 关联的工具结果。

在 Agent Tool Calling 场景中，返回 Command 时应保持对应的 ToolMessage，避免破坏消息协议。

注意：这里采用的是列表整体更新。如果多个工具并行修改同一字段，需要额外设计 Reducer 或顺序执行机制。

八、完整实例：学习进度 Agent

下面将自定义 State、ToolRuntime、Command 和 Checkpointer 结合起来，创建可以记录与查询学习进度的 Agent。

文件名：agent_state_demo.py

安装依赖：

pip install -U langchain langchain-deepseek langgraph python-dotenv

配置 .env：

DEEPSEEK_API_KEY=your_deepseek_api_key

如果模型正确调用工具，第一轮状态应包含：

['HTML 基础', 'Python 基础']

第二轮使用相同的 thread_id，可以读取此前保存的状态。

这里的 InMemorySaver 只保存进程内存。程序退出后，状态通常会丢失。

注意：模型是否调用工具并不保证。需要强制记录业务数据时，应由确定性的应用逻辑执行，而不是只依赖模型选择工具。

九、购物车 Agent：状态管理实战

购物车是另一个常见的自定义 State 场景。

例如，需要保存购物车商品列表，并根据商品目录计算总价。

下面使用模拟商品数据，不创建真实订单，也不进行支付。

为了防止模型随意修改价格，商品价格由程序内部目录决定，而不是由模型传入。

如果 Agent 成功执行相关工具，购物车状态为：

['Python 教程']

工具计算出的金额为：

购物车：Python 教程；总价：¥49.90

这里没有让模型提供 price 参数，而是由程序读取预先定义的模拟商品目录。

注意：如果模型并行调用多个修改购物车的工具，可能产生并发状态冲突。生产级购物车应结合数据库事务或专门的状态更新机制管理。

十、自定义 Reducer

AgentState 的自定义字段默认通常采用覆盖式更新。

如果需要合并多个更新，可以通过 Annotated 指定 Reducer。

实例：追加学习日志

当原有状态是：

["学习 HTML"]

新更新为：

["学习 Python"]

Reducer 合并后可以得到：

["学习 HTML", "学习 Python"]

需要注意，operator.add 只是列表拼接，不会自动去重。

如果需要唯一记录、严格顺序或防止重复执行，应设计专门的 Reducer 或在业务层处理。

十一、state_schema 与 Middleware state_schema

除了通过 create_agent(state_schema=...) 定义全局业务状态，Middleware 也可以声明自己的扩展字段。

不能假设 create_agent 中声明的同名字段总能覆盖 Middleware 中的字段定义。两个 Schema 的类型和 Reducer 必须保持兼容。

实例：中间件统计模型调用次数

对于只调用一次模型的简单任务，通常输出：

模型调用次数： 1

如果 Agent 需要多轮模型调用，该计数也会增加。

十二、Checkpointer：保存 AgentState

默认情况下，Agent 每次调用不一定保留上一次的会话状态。

如果需要跨多次调用恢复消息和自定义字段，可以配置 Checkpointer，并通过 thread_id 区分会话。

实例：保存多轮对话

这里使用相同的 thread_id，因此第二轮可以读取此前保存的会话消息。

但 InMemorySaver 不是磁盘持久化方案。

需要在服务重启后继续保留数据时，应使用数据库支持的 Checkpointer。

十三、AgentState 与 Store 的区别

例如，一个学习助手可以这样设计：

AgentState：
  当前对话历史
  当前学习章节
  本轮任务进度

Store：
  用户已完成的全部课程
  长期学习偏好
  跨会话学习记录

对于需要长期保存的真实业务数据，通常应使用持久化 Store 或数据库，而不是仅依赖 AgentState。

十四、常见问题

1、AgentState 是普通字典吗？

AgentState 使用 TypedDict 描述状态，运行时通常以字典形式传递，但实际合并行为还受到 Reducer 和 LangGraph 执行机制控制。

2、add_messages 是否永远只追加消息？

不是。它还可以根据相同 ID 替换消息，以及使用 RemoveMessage 删除消息。

3、jump_to="tools" 能直接执行某个 Python 函数吗？

不能。它只负责流程路由，工具节点仍需要合法的工具调用请求。

4、structured_response 为什么不存在？

通常是没有配置 response_format，或结构化输出没有成功完成。

5、为什么直接修改 runtime.state 没有效果？

因为这不是可靠的图状态更新方式。工具应使用 Command(update=...) 提交状态变更。

6、Command 需要 ToolMessage 吗？

在 Agent 工具调用场景中，应提供与当前 Tool Call ID 对应的 ToolMessage，保持工具请求与工具响应正确关联。

7、自定义 State 可以省略字段吗？

可以使用 NotRequired 定义可选字段，但读取时需要自行处理默认值。

8、如何解决多个工具同时更新状态的冲突？

可以使用合适的 Reducer、限制并行执行或通过确定性的业务流程处理。

9、State 会自动保存到数据库吗？

不会。需要配置 Checkpointer，并选择适合的持久化实现。

10、AgentState 能保存用户长期学习进度吗？

可以配合 Checkpointer 保存线程状态。但对于需要跨多个会话共享的学习进度，更适合使用 Store 或业务数据库。

总结

AgentState 是 LangChain Agent 状态管理的核心。

本章节主要介绍了：

• messages：保存消息历史。
• add_messages：追加、替换和删除消息。
• jump_to：控制 Middleware 流程跳转。
• structured_response：保存结构化响应。
• state_schema：扩展自定义状态字段。
• ToolRuntime：在工具中访问 State。
• Command：提交状态更新。
• Reducer：定义状态合并规则。
• Checkpointer：保存和恢复会话状态。

可以简单记住：

读取状态：runtime.state
更新状态：Command(update={...})
保存会话：checkpointer + thread_id
长期数据：Store