Agent 状态持久化与人审手把手

怎么用

这篇的核心是把 Agent 从“一次性脚本”升级为“可靠工作流”。每一步都要回答一个生产问题：断了怎么办、谁审批、怎么回放、怎么证明没乱执行。

Step 1 · 定义状态模型

先写清 state，而不是先写节点：

from typing import TypedDict, Literal
 
class AgentState(TypedDict):
    task_id: str
    user_id: str
    goal: str
    plan: list[str]
    status: Literal["running", "waiting_approval", "done", "failed"]
    tool_result: dict
    audit_log: list[dict]

验收：每个节点只读写必要字段，不把所有东西塞进一个字符串。

Step 2 · 设计节点

最小工作流：

plan -> risk_check -> human_approval? -> execute_tool -> summarize

风险检查节点只负责判断，不执行工具。执行节点只接受已经批准的计划。

Step 3 · 加 checkpoint

Checkpoint 的目标是中断后能恢复，不是简单保存聊天记录。要保存：

• state
• 当前节点
• thread_id / task_id
• 时间戳
• 版本号

验收：手动中断进程后，重新启动能从等待审批或下一节点继续。

Step 4 · 加人审 interrupt

当风险为 high 时进入等待审批：

{
  "status": "waiting_approval",
  "approval_request": {
    "action": "send_email",
    "reason": "外部发送动作",
    "preview": "将发送给 3 个客户"
  }
}

前端或 CLI 展示 preview，审批通过后 resume。

Step 5 · 审批结果写入审计

不要只存 approved=true。要存：

approval_id, approver, decision, reason, approved_at, diff, trace_id

拒绝时也要写入，并让 Agent 生成替代方案。

Step 6 · Time Travel 与回放

保留关键 checkpoint，可以对比：

• 原计划
• 审批前 state
• 工具执行后 state
• 最终总结

验收：能拿一个 task_id 讲出完整过程，而不是只看最终回答。

Step 7 · 失败重试

工具失败不要直接让模型“再试”。先分类：

• 网络超时：可重试
• 参数错误：回到 plan 修正
• 权限不足：转人工或拒绝
• 幂等冲突：查询已有结果

Step 8 · 面试讲述

用这个结构回答：

1. Agent 的风险不在“会不会调用工具”，而在“能否控制副作用”。
2. 我用 state + checkpoint 保证可恢复。
3. 高风险节点 interrupt 到人审。
4. 审批和工具结果进入 audit log，支持回放和追责。