状态管理在 Agent 系统中的设计模式
难度：中级 分类：Memory & State
简短回答
Agent 系统的状态管理决定了"Agent 在任意时刻知道什么、做过什么、下一步该做什么"。核心设计模式包括：共享状态图（LangGraph 模式）——用 TypedDict 定义全局状态，通过 Reducer 函数合并并发更新，内置 Checkpointing 实现持久化和恢复；事件溯源（Event Sourcing）——记录所有状态变更事件而非最终状态，支持完整回放和审计；有限状态机（FSM）——用明确的状态和转移规则管理流程；黑板模式——共享空间让多 Agent 自主读写。LangGraph 的 State + Reducer + Checkpoint 模式已成为业界主流，被 Klarna、Replit 等企业用于生产。
详细解析
为什么 Agent 需要状态管理？
代码块
Plain Text
无状态 Agent：
每次 LLM 调用都是独立的 → 无法做多步任务
工具调用后结果丢失 → 无法基于前序结果决策
中断后无法恢复 → 长任务不可靠
有状态 Agent：
状态跨步骤持久化 → 支持复杂多步工作流
可以暂停和恢复 → 支持 Human-in-the-Loop
状态可检查 → 支持调试和审计
模式 1：共享状态图（LangGraph 模式）
LangGraph 的核心——状态是流经图中每个节点的共享内存对象：
代码块
Python
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
import operator
# 1. 定义状态 Schema
class AgentState(TypedDict):
messages: Annotated[list, operator.add] # 追加模式
research_data: str # 替换模式
draft: str
revision_count: int
status: str
# 2. 节点读取状态、处理、返回更新
def research_node(state: AgentState) -> dict:
query = state["messages"][-1]["content"]
data = search(query)
return {
"research_data": data,
"status": "research_complete"
}
# LangGraph 自动用 Reducer 合并更新到全局状态
def write_node(state: AgentState) -> dict:
Reducer 的关键作用：
代码块
Python
# Reducer 决定状态如何更新
class State(TypedDict):
messages: Annotated[list, operator.add] # 追加：新消息加到列表末尾
count: Annotated[int, operator.add] # 累加：数字相加
result: str # 替换：新值直接覆盖旧值
模式 2：Checkpointing（持久化与恢复）
代码块
Python
from langgraph.checkpoint.sqlite import SqliteSaver
# SqliteSaver.from_conn_string 是 context manager，必须用 with ... as 形式打开
# （直接赋值给变量会得到一个未进入的 contextmanager 对象，运行时报错）
Checkpointing 使得以下场景成为可能：
•
2 小时的 Agent 任务中途 Pod 重启 → 从检查点恢复
•
Human-in-the-Loop：暂停等待审批 → 审批后从暂停点继续
•
时间旅行调试：回到第 N 步查看当时的完整状态
模式 3：事件溯源（Event Sourcing）
代码块
Python
class EventSourcedState:
"""记录所有状态变更事件，支持回放"""
def __init__(self):
self.events = [] # 事件日志
self.current_state = {} # 当前状态（由事件推导）
def apply_event(self, event: dict):
self.events.append({
**event,
"timestamp": datetime.now(),
"sequence": len(self.events)
})
self.current_state = self.rebuild()
def rebuild(self, up_to: int = None):
"""从事件日志重建状态"""
模式 4：有限状态机（FSM）
代码块
Python
模式 5：Human-in-the-Loop 状态管理
代码块
Python
# LangGraph 的中断与恢复
from langgraph.types import interrupt
def sensitive_action_node(state):
"""需要人工审批的敏感操作"""
action = plan_action(state)
# 暂停执行，等待人工审批
approval = interrupt({
"action": action,
"reason": "此操作将修改生产数据库",
"options": ["approve", "reject", "modify"]
})
模式 6：外置状态 + 环境审计（LongHorizon-Harness，2026-08）
前面五个模式回答的是"状态存在哪、怎么合并、怎么恢复"。LongHorizon-Harness 提出的是另一个维度：谁有权写状态。
它把长程执行重构成一个任务状态管理问题，核心约束只有一条——状态只接受环境已验证的事实，模型的自我判断不算数：
状态只接受环境验证过的事实：Manager 派发、Executor 零上下文执行、Auditor 只读核验，围绕唯一任务状态闭环。
对比常见做法：让单个 Agent 在不断膨胀的上下文里自我评估"我刚才做成了吗"，一次误判会留在上下文里，后续每一步都在这条错误结论上继续推理，误差持续传播。外置状态 + 独立审计切断这条链路——Executor 的上下文每次归零，错误无处驻留；状态里的每条事实都被环境验证过。
效果（论文数据）：WeaveBench 51.8% → 80.7%，OSWorld 2.0 2.8% → 8.3%，增益可迁移到 Claude Opus 4.7。
设计要点：
•
状态是唯一的跨步骤载体，子任务之间不共享上下文，只共享状态
•
写状态需要凭据：来源必须是环境的可观测结果（文件确实存在、测试确实通过），不是模型说"我做完了"
•
审计只读：Auditor 不修改环境，避免它自己成为新的错误来源
设计原则
模式选择指南
常见误区 / 面试追问
误区："全局变量就是状态管理" — Agent 的状态需要持久化、可恢复、可审计。用全局变量管理状态在 Pod 重启后丢失、无法调试、不支持并发。需要 Checkpointer 或事件溯源等正式机制。
误区："状态越详细越好" — 状态越大，序列化/反序列化越慢，Checkpoint 占用越多存储。只保留决策所需的最小信息集。中间推理过程应该在 Trace 中记录而非状态中保存。
追问："LangGraph 的 Reducer 和 Redux 的 Reducer 有什么关系？" — 概念相同——都是纯函数，接收当前状态和更新，返回新状态。LangGraph 用 Python 的 Annotated 类型标注来声明 Reducer 规则（如 operator.add 表示追加）。
追问："Checkpointing 对性能的影响？" — 每个 super-step 后保存检查点会增加延迟。优化方案：(1) 使用异步写入；(2) 只在关键步骤做 Checkpoint；(3) 用 Redis 替代 SQLite 做高性能 Checkpoint。