跳到主要内容

状态管理

核对日期：2026-05-09。

1. 定义与边界​

状态管理是记录和更新 Agent 执行过程中必要运行数据的机制，包括任务状态、步骤状态、工具结果、计划、审批结果、错误、成本和 trace 关联信息。它不同于长期记忆；长期记忆面向跨任务知识沉淀，状态管理面向一次或一组任务的正确执行。

2. 为什么重要​

Agent 执行不是单次函数调用。它可能经历模型推理、工具调用、等待用户审批、队列重试、异步回调和恢复执行。没有显式状态，系统只能依赖内存变量，一旦进程重启、任务超时或 worker 被杀，就无法恢复，也无法解释线上行为。

3. 核心机制​

一个生产状态对象应包含：

{
  "run_id": "run_01",
  "session_id": "sess_01",
  "agent_version": "support-agent@1.8.0",
  "prompt_version": "2026-05-09.3",
  "status": "waiting_tool",
  "step_index": 4,
  "plan": [{"id": "s1", "status": "done"}],
  "messages_ref": "object://trace/run_01/messages.json",
  "tool_calls": [{"id": "tool_9", "name": "crm.lookup", "status": "ok"}],
  "approvals": [],
  "cost": {"input_tokens": 12000, "output_tokens": 900},
  "last_error": null,
  "created_at": "2026-05-09T10:00:00Z",
  "updated_at": "2026-05-09T10:01:10Z"
}

状态更新建议采用 append-only 事件 + 当前快照：

4. 架构模式​

模式适用场景注意点
内存状态本地 demo、单步调用不能用于长任务和生产恢复。
数据库快照常规 API/worker简单直接，但回放能力弱。
事件溯源高审计、高回放要求实现复杂，需要事件版本治理。
Workflow checkpointLangGraph/Temporal 等图或工作流要设计 checkpoint schema 迁移。

5. 工程实现​

状态更新必须原子化：

def apply_event(run_id, expected_version, event):
    state = store.get_for_update(run_id)
    if state.version != expected_version:
        raise ConflictError("state changed")
    next_state = reducer(state, event)
    store.append_event(run_id, event)
    store.save_snapshot(run_id, next_state, version=state.version + 1)
    return next_state

状态字段分级：

  • 必存：run_id、status、step_index、模型/Prompt/工具版本、错误、成本、trace id。
  • 可外链：完整消息、工具大结果、检索文档、文件。
  • 禁止明文：密钥、完整支付信息、未脱敏个人敏感信息。

6. 生产实践​

  • 用状态机限制非法跃迁，例如 completed 不能回到 running。
  • 状态写入与副作用调用之间要有幂等键，避免“状态失败但工具成功”造成重复执行。
  • 对长上下文不要全部塞进状态快照，使用对象存储引用和摘要。
  • 定义状态 TTL 与归档策略，满足审计和隐私删除要求。
  • 状态 schema 变更要兼容旧 run 的恢复。

7. 常见反模式​

  • 只保存最终回答，不保存中间计划和工具调用。
  • 让模型自由生成状态 JSON，再直接覆盖数据库。
  • 不记录 Prompt 和工具版本，导致回放不可解释。
  • 把会话历史、长期记忆和任务运行状态混成一个字段。
  • 失败后只改 status=failed，没有错误类型、可重试标记和补偿信息。

8. 评测方法​

  • 恢复测试：在每个关键步骤强制 kill worker，验证能从状态继续或安全失败。
  • 并发测试：同一 run 多 worker 竞争时不会重复执行副作用。
  • 回放测试：从事件日志重建快照，与当前快照一致。
  • 数据治理测试：敏感字段不会进入普通日志和低权限状态表。

9. 安全与治理​

  • 状态读写按租户和用户隔离，不能只靠 run_id 猜测权限。
  • 外部工具返回内容进入状态前要做敏感信息分类和大小限制。
  • 审批状态不可由模型直接写入，应由审批服务写入。
  • 高风险状态变更写审计日志，例如权限升级、付款、删除数据。

10. 权威资料​

11. 二次精修：Agent 状态 schema​

状态管理的核心不是“保存聊天记录”，而是保存 Agent 执行到哪里、依据什么、下一步允许做什么。

{
  "run_id": "run_123",
  "tenant_id": "t_123",
  "user_id": "u_456",
  "session_id": "ses_789",
  "status": "waiting_approval",
  "step_index": 4,
  "plan": [{"id": "s1", "status": "done"}, {"id": "s2", "status": "pending"}],
  "tool_results": [{"tool": "search_policy", "result_ref": "blob://tool/r1"}],
  "pending_approvals": [{"action": "refund", "amount": 199}],
  "budgets": {"max_tokens": 20000, "max_cost_usd": 0.5, "deadline_at": "2026-05-09T11:00:00Z"},
  "version": 17,
  "updated_at": "2026-05-09T10:15:00Z"
}
状态字段写入时机观测指标安全要求
status每个关键阶段卡住时长、失败率不允许跳过审批态
step_index每次计划推进循环次数、最大步数超限触发熔断
tool_results工具返回后工具错误率、耗时参数和结果脱敏
pending_approvals高危动作前审批等待时长审批人身份审计
budgetsrun 创建时token/cost 消耗超预算必须停止
version乐观锁更新写冲突次数防止并发覆盖

12. 状态流与并发控制​

def update_state(store, run_id, expected_version, patch):
    current = store.get(run_id)
    if current["version"] != expected_version:
        raise ConflictError("state version changed")
    next_state = apply_patch(current, patch)
    next_state["version"] += 1
    store.put(run_id, next_state)
    return next_state

13. 灾难恢复与验收​

  • 每个长任务至少保存 checkpoint，恢复时从最后一个无副作用边界继续。
  • 状态 schema 变更要有迁移脚本和回滚脚本，旧 trace 至少在保留期内可读取。
  • 关键状态写入要有 audit log，尤其是审批、预算、工具副作用和人工覆盖。
  • 验收指标：状态丢失率、写冲突率、恢复成功率、悬挂任务数量、平均恢复时间。
  • 反模式：把模型上下文当唯一状态；把所有状态放在前端 session；失败后只能整单重跑。

14. 补充权威资料​