13 状态、记忆与检查点:Agent 如何保存可恢复上下文
结论先说:记忆系统最重要的不是“存更多”,而是“分层存对” #
Agent 一旦开始跨多步执行,你马上会碰到三个问题:
- 当前这轮状态怎么传递
- 进程中断后怎么恢复
- 历史知识和当前上下文怎么分离
如果你把这些都叫“memory”,系统很快就会变得不可控。
一、先把四层分开 #
1.1 短期状态(State) #
只服务当前执行轮次
1.2 工作记忆(Working Memory) #
保存当前阶段最重要的发现、焦点和局部结论
1.3 Checkpoint / 会话恢复 #
保存足够让系统继续执行的结构化状态
1.4 RAG / 长期知识层 #
保存长期可检索证据,不直接参与当前状态机字段更新
State -> 当前节点之间怎么传
Working -> 当前任务阶段还要记什么
Checkpoint -> 中断后怎么接着跑
RAG -> 长期知道什么事实
二、为什么 State 不能等于“所有上下文” #
一个常见错误是把所有东西都塞进 AgentState:
- 全量 trace
- 所有历史结果
- 所有检索证据
- 所有日志
这会导致三种问题:
- state 膨胀
- checkpoint 变慢
- 下一轮 prompt 污染
所以合理的 AgentState 应该是“当前控制循环必须知道的最小信息集合”。
class AgentState(TypedDict):
current_input: str
plan: Optional[list[Action]]
steps: list[StepResult]
evaluation: Optional[dict]
error_count: int
final_output: Optional[str]
状态的价值在于控制流程,不是充当数据垃圾桶。
三、Working Memory 为什么不是数据库 #
工作记忆保存的是“当前任务最值得保留的局部事实”,比如:
- 当前怀疑的攻击路径
- 前一轮结论摘要
- 当前搜索焦点
- 需要避免重复尝试的策略
它和数据库、RAG 的区别是:
- 生命周期更短
- 信息粒度更摘要化
- 目标是辅助下一步决策,而不是长期留档
如果把工作记忆直接做成“原始结果缓存”,它很快就会变成第二个日志系统。
四、Checkpoint 的核心不是“完整保存”,而是“最小可恢复” #
Checkpoint 最关键的一点是:
它不应该保存所有东西,它应该只保存“恢复执行所需的最小状态”。
# 伪代码:checkpoint 保存的应该是结构,不是全量痕迹
checkpoint = {
"thread_id": thread_id,
"plan": state["plan"],
"step_index": state["step_index"],
"results_summary": summarize(state["results"]),
"evaluation": state["evaluation"],
}
为什么不能一把 pickle(state) 全存 #
因为这样会:
- 放大 checkpoint 体积
- 把无关 trace 一起塞进去
- 让 schema 变更后的恢复更脆弱
所以正确的恢复思路应该是:
恢复当前执行语义
而不是
恢复所有原始细节
五、SQLite checkpoint 为什么适合这个阶段 #
很多人一看到 checkpoint 就想上 Redis、Postgres、甚至对象存储。
但 SecurityClaw 这种阶段,SQLite 是一个很合理的选择:
优点 #
- 本地可用
- 无额外服务依赖
- 事务和持久化都够用
- 调试简单
限制 #
- 并发扩展弱
- schema 演进时需要小心迁移
- 大规模 trace 持久化不合适
这就是典型的工程取舍:
先用足够可靠、足够轻的 checkpoint 方案,把恢复语义做对;再考虑扩展性。
六、RAG 为什么不能混进 checkpoint #
这是最容易踩坑的边界问题之一。
如果你把最近检索到的所有文本都写进 checkpoint,会发生什么?
- 恢复成本暴涨
- 下次启动就带着一堆陈旧证据
- prompt 预算被历史检索垃圾吞掉
- 真正的当前状态被淹没
所以正确做法是:
- checkpoint 保存“恢复需要的结构状态”
- RAG 继续保存在独立知识层
- 恢复后如果需要,再重新检索
这会让系统慢一点点,但会让语义边界干净得多。
七、有界性才是记忆系统第一原则 #
一个真实可运行的 Agent,记忆系统必须有界。
至少要限制这些东西 #
- trace 条数
- 单条 trace 长度
- checkpoint 频率
- working memory 总大小
def limit_trace(trace: list[dict], max_items=50):
if len(trace) <= max_items:
return trace
return trace[:1] + trace[-(max_items - 1):]
这个设计背后的思想很朴素:
- 保留起点
- 保留最近上下文
- 丢掉中间噪音
如果没有这种限制,记忆系统最后一定变成“系统负担”。
八、这一篇真正要记住的框架 #
State / Memory / Checkpoint / RAG 的关系,可以压成这张表:
| 层 | 作用 | 生命周期 | 关注点 |
|---|---|---|---|
| State | 控制当前循环 | 单轮执行 | 节点间传递 |
| Working Memory | 保留当前焦点 | 多步任务内 | 摘要与压缩 |
| Checkpoint | 中断恢复 | 跨请求/重启 | 最小可恢复 |
| RAG | 长期知识 | 长期 | 检索与注入 |
如果你把这四层讲不清,Agent 的“记忆系统”大概率只是一个名字好听的混合大对象。
九、验证命令 #
pytest tests/ -k "memory or checkpoint or state" -v
如果要看 SQLite checkpoint:
sqlite3 data/conversations.db "SELECT * FROM checkpoints ORDER BY created_at DESC LIMIT 5;"