Agent 项目实战：可审批、可恢复、可评估的巡检任务助手

项目目标

本项目把本阶段的关键能力组合成一个设备巡检助手：读取设备状态、匹配处置规则、暂停等待人工审批、恢复后幂等创建工单，并通过 SSE 暴露结构化事件。固定评估集可以在没有模型和网络的环境中运行。

项目不是要模拟一个“无所不能”的自主 Agent，而是演示决策能力进入真实应用后，控制平面应怎样设计。

架构设计

HTTP / CLI
    │  user_id + thread_id
    ▼
InspectionAgent
    ├── 读取设备状态（只读）
    ├── 匹配处置规则（确定性决策）
    ├── Checkpoint：waiting_approval
    └── 恢复后创建工单（幂等副作用）
            │
            ├── checkpoints/<thread_id>.json
            ├── work_orders.json
            └── SSE 事件轨迹

为了离线可重复，示例用规则完成风险判断。以后接入 LangChain/LangGraph 或平台 Agent 时，可以替换建议生成层，但 Thread 所有权、审批、幂等和评估不应交给模型。

项目结构

完整代码位于 examples/08.15-inspection-agent：

08.15-inspection-agent/
├── agent_core.py
├── app.py
├── data/
│   ├── devices.json
│   ├── eval.jsonl
│   └── rules.json
├── requirements.txt
└── README.md

agent_core.py 只依赖 Python 标准库；app.py 是可选 FastAPI 适配层。运行数据写入自动生成的 runtime/，不纳入版本控制。

核心实现

1. 启动任务并暂停

start() 先检查 thread_id 是否已存在，读取设备和规则。当设备离线或温度达到阈值时，它不会直接创建工单，而是保存如下提案：

{
  "status": "waiting_approval",
  "proposal": {
    "action": "create_work_order",
    "device_id": "AX-3",
    "reason": "设备离线，需要创建人工巡检工单",
    "idempotency_key": "task-001:create_work_order"
  }
}

Checkpoint 使用临时文件、fsync 和同目录原子替换，避免常见的半写入文件。这个实现适合单机教学；多进程生产环境应替换成支持事务和并发控制的持久存储。

2. 审批后恢复

resume() 重新读取 Checkpoint，校验任务所有者、当前状态和决策值。批准后才写入工单，并以 (thread_id, action) 组成幂等键；同一任务重复恢复只会取得已有结果。

编辑原因时会重新校验长度。生产环境还应校验审核角色、设备访问范围、提案版本和乐观锁，防止审核期间任务已变化。

3. 事件与 SSE

每个事件都有稳定 run_id、thread_id 和单调递增 seq。API 层把持久事件映射成标准 SSE：

id: 4
event: approval_required
data: {"event":"approval_required","seq":4,...}

示例接口发送已有事件后结束连接，适合观察状态机。长任务 Runtime 应把执行与订阅解耦，支持按 Last-Event-ID 补发和持续订阅。

运行与验收

进入项目目录后，无需安装依赖即可运行评估：

python agent_core.py evaluate

预期结果为 5 条样例全部通过，覆盖离线待审批、批准创建、拒绝、高温、正常设备和设备不存在等主要分支。

手工走一遍暂停与恢复：

python agent_core.py start task-001 AX-3
python agent_core.py resume task-001 approve
python agent_core.py show task-001

若要启动 HTTP 服务：

python -m pip install -r requirements.txt
uvicorn app:app --reload

请求必须携带 X-User-Id。这只是本地演示身份；真实系统应由认证中间件写入可信主体，并对所有查询、恢复和事件订阅执行同一套所有权检查。

扩展练习

1. 用 LangGraph Checkpointer 替换 JSON Checkpoint，保持对外状态契约不变；
2. 把规则检索替换为第 07 阶段的 RAG，并在提案中保存证据引用；
3. 增加提案版本和乐观锁，测试两个审核人并发恢复；
4. 增加持久事件表与 Last-Event-ID，实现断线补发；
5. 接入真实工单 Sandbox，并用唯一索引验证幂等；
6. 扩充评估集，加入越权、重复恢复、工具超时和恶意编辑。

常见问题与排查

重启后还能恢复，但 SSE 看不到旧事件

确认事件与 Checkpoint 一同持久化，而不是只放在进程内队列。生产系统通常使用事件表或消息系统，并按序号补发。

两个进程创建了重复工单

示例文件存储不提供跨进程事务。生产数据库对幂等键建立唯一约束，并将“检查与创建”放入同一事务。

接入模型后评估变得不稳定

保留确定性控制用例，固定模型和数据版本；将模型质量与权限、审批、状态机测试分层，避免一次失败无法定位。

用户修改请求头就能访问别人任务

本地请求头只是演示。部署时从 JWT、Session 或 API Gateway 获得认证主体，忽略客户端提交的任意 owner 字段。

项目小结

一个能运行的 Agent 项目不仅有模型和工具，还必须有可恢复状态、审批边界、幂等副作用、事件轨迹和回归评估。先用确定性核心把这些工程约束跑通，再替换智能决策层，系统会更容易验证和演进。