S21-X08 · Agent 状态该存什么：AgentState 设计

Agent 跑不稳，十次里有七次不是模型的问题，是状态设计的问题：要么把整篇文档塞进 state 导致每轮 prompt 雪崩，要么该记的没记、重放时从第一步重头再来，要么两个 Agent 同时写同一个字段互相覆盖。

状态设计的核心不是"存多少"，而是一条判据：这个东西重算一遍贵不贵？贵就存，便宜就不存。

1. 必须进状态的五类东西

类别	例子	不存的后果
消息历史	用户输入、模型回复、工具调用记录	模型失忆，用户得重复说一遍
当前位置	当前步骤、已完成的阶段、等待中的审批	恢复时不知道从哪继续
工具结果	查询返回的结构化数据、文件 id	每一步都要重新调用外部系统
错误与重试计数	最后一次失败原因、已重试次数	无限重试，或失败后静默放弃
预算与计数	token 用量、调用次数、步数	没法设终止条件，跑飞了才发现

这五类里有三类是"外部世界不可重复取得"的：用户的话、工具返回的当时数据、已经花掉的钱。它们必须留。

2. 不该进状态的三类东西

类别	为什么不该进	替代做法
大文件正文	每次 checkpoint 都要序列化，token 成本线性增长	存路径或 URL，用时读
原始日志	体积大、90% 与决策无关	存聚合指标 + 关键行号
可重算的中间结果	从已有状态能推出来的，等于冗余	用函数/属性即时计算

一个常见的坏设计是把 PDF 原文、HTML 源码、整个检索返回塞进 state。表现是：前两轮正常，第五轮开始每轮多烧几千 token，checkpoint 文件涨到几十兆。

# 反例
state = {"doc": open("manual.pdf", "rb").read(), "pages": [...all 200 pages...]}

# 正例
state = {"doc_id": "manual-v3", "pages_needed": [12, 13], "excerpt_hash": "9f3c"}

3. 状态与记忆：两个不同的时间尺度

维度	State（会话内）	Memory（跨会话）
生命周期	一次会话 / 一个 thread	跨会话、跨用户
读写者	图里的节点	检索层 + 写入层
存储	checkpoint 后端	向量库 / 关系库 / KV
内容	这次任务的完整过程	沉淀下来的偏好、事实、结论
出错代价	重跑一次	污染后续所有会话

判断某条信息该放哪："下一次会话还需要它吗？" 需要就写记忆，不需要就留在 state 里随会话结束消失。把会话临时数据写进长期记忆是污染的主要来源——比如把"这次用户输入的临时路径"当成用户偏好存下来。

def after_task(state):
    facts = extract_stable_facts(state["messages"])   # 只挑跨会话成立的
    if facts:
        memory.add(facts, user_id=state["user_id"], ttl_days=180)
    return {}                                          # 不往 state 里塞

4. 压缩策略：四选一，别兼用

策略	做法	省下的	风险
滑动窗口	只保留最近 N 轮	简单可靠	早期关键约束被丢掉
摘要替换	把旧消息交给模型压成一段	压缩率高	摘要写歪，后续全歪
结构化事实	抽成 `{key: value}` 存字段	体积最小、可检索	需要额外抽取步骤，可能漏
分层保留	最近 N 轮原文 + 更早的摘要 + 事实表	平衡最好	实现复杂，要三处一致

生产上推荐分层保留：最近 6–10 轮保持原文（模型需要逐字复现最近的约束），更早的压成摘要，稳定不变的信息（用户角色、系统约束、任务目标）单独放结构化字段。

MAX_TURNS = 8

def compress(state):
    msgs = state["messages"]
    if len(msgs) <= MAX_TURNS:
        return {}
    old, recent = msgs[:-MAX_TURNS], msgs[-MAX_TURNS:]
    return {
        "summary": summarize(old, prev=state.get("summary")),  # 增量摘要
        "messages": recent,                                    # 用覆盖语义截断
        "facts": state.get("facts", {}) | extract_facts(old),
    }

注意 messages 在这里是覆盖而不是追加——所以 messages 字段的 reducer 与"截断"这个动作必须显式设计，别让 add_messages 一边追加一边截断。

5. 序列化与恢复

后端	落点	恢复粒度	适合
进程内内存	进程内存	进程级	开发
本地文件 / SQLite	单机磁盘	线程级	单机服务
Postgres / Redis	网络存储	线程级 + 多实例	生产
对象存储	冷备	按需加载	归档与事后复盘

三条硬规则：

只放可序列化的东西：dict/list/str/int/bool/None。放进去一个 datetime 或数据库连接，第一次恢复就炸。
大结果与状态分离：工具返回 2MB 的表格时，状态里存 result_id，正文落对象存储。
checkpoint 要有版本号：状态结构会改，重新部署后旧 checkpoint 反序列化失败是常态。加 schema_version 字段，读旧数据时走迁移。

state = {
    "schema_version": 3,
    "thread_id": "user-42",
    "messages": [...],
    "budget": {"tokens": 18342, "steps": 5, "usd": 0.21},
    "pending": None,
}

6. 多 Agent 共享状态：所有权先定义

冲突形态	表现	解决
两个 Agent 写同一字段	后写的覆盖先写的，结果不可复现	按字段划分所有权，一人一字段
共享列表无 reducer	顺序执行看起来正常，一并行就丢数据	加 `operator.add` / `add_messages`
谁都能改"当前步骤"	流程控制失效，跳步或回退	步骤字段只由编排器写
事后分不清是谁改的	出问题无法归因	每条写入带 `author` 与时间戳

最省心的做法是显式所有权表：在代码里写明每个字段的写入者，而不是靠约定。

字段	唯一写入者	其他 Agent
`messages`	所有节点（reducer 合并）	只追加，不截断
`plan`	规划节点	只读
`evidence`	检索节点	只读
`approval`	人工/审批节点	只读
`budget`	编排器	通过返回值申请

7. 幂等与重放

Agent 的"重试"经常等于"把同一个有副作用的动作做第二遍"——发了两封邮件，建了两张工单。幂等的做法是给每个副作用动作一个稳定的幂等键，而不是靠"应该不会重放吧"。

import hashlib

def idem_key(state, action, args) -> str:
    raw = f'{state["thread_id"]}|{state["step"]}|{action}|{json.dumps(args, sort_keys=True)}'
    return hashlib.sha256(raw.encode()).hexdigest()[:16]

def send_ticket(state):
    key = idem_key(state, "create_ticket", {"title": state["title"]})
    if key in state.get("done_keys", []):          # 做过，直接复用结果
        return {"ticket_id": state["ticket_ids"][key]}
    tid = api.create(title=state["title"], idempotency_key=key)
    # 只返回增量：done_keys 用 operator.add 合并，ticket_ids 按 key 覆盖合并
    return {"ticket_ids": {**state.get("ticket_ids", {}), key: tid},
            "done_keys": [key]}

动作类型	幂等键怎么取	备注
读操作	不需要	天然可重放
创建类写操作	`thread_id + step + 参数hash`	重放复用首次结果
发送类（邮件/消息）	同上 + 接收人	键里必须含接收人
外部已有幂等支持	直接用对方的 key	键要能跨进程稳定生成

注意上面这段：节点不修改传入的 state，只返回自己写的那几个字段，done_keys 交给 operator.add 合并、ticket_ids 按 key 合并。这是幂等能配合并发重放的前提。

thread_id 进入幂等键是有意的：同一会话内的重放应该复用，不同会话里的相同请求应该各自执行。

8. 排错

现象	原因	处理
第 N 轮突然变慢变贵	历史消息无限追加	分层压缩，滑动窗口
恢复后从头开始跑	状态没落盘，或 `thread_id` 变了	用持久化 checkpointer，固定线程 id
反序列化报错	状态里有不可序列化对象	只放基础类型，禁止自定义对象
并行时结果少了一块	共享列表没有 reducer	加 `operator.add`
同一个副作用执行两次	没有幂等键	加键并落盘已执行集合
用户偏好丢失	该写记忆的写进了 state	会话结束前落一份到记忆层
checkpoint 越来越大	大对象入了状态	状态只留 id，正文外置

9. 边界

场景	建议	理由
单轮问答型 Agent	状态只留消息历史	加字段都是负担
长流程自动化	完整状态 + 数据库 checkpoint	需要断点续跑
多 Agent 协作	显式所有权 + reducer	冲突必须在设计时消除
强合规场景	状态 + 动作审计日志分离	状态可丢，审计不能丢
高频短会话	状态尽量小，记忆尽量薄	每会话的存储成本会被放大

10. 自测

"该不该进状态"的判据是什么？举一个"贵所以要存"的例子。
State 与 Memory 的时间尺度差异是什么，判断某条信息归属时该问哪个问题？
分层压缩里，为什么最近几轮要保留原文而不是全部摘要？
恢复执行失败的三大常见原因是什么？
并行写同一个没有 reducer 的字段会出现什么症状？
幂等键里为什么必须包含 thread_id？