用Python实现Planner-Executor-Reviewer状态机Agent

文章摘要

实现带计划校验、步骤执行、Reviewer反馈、局部重试和最终汇总的状态机Agent。

一、项目目标

  • 结构化计划
  • DAG校验
  • 步骤状态
  • 局部审校
  • 预算与终止

二、总体架构

Goal → Planner → PlanValidator → Executor → Reviewer → Aggregator

架构遵循三条原则：业务接口不直接绑定模型厂商；高风险动作不由模型直接授权；关键步骤必须可追踪、可重放、可审计。

三、项目结构

src/main/java
├── api
├── application
├── domain
├── infrastructure
├── security
└── observability

四、核心数据模型

public record AgentState(
        String requestId,
        String tenantId,
        String status,
        String version,
        Map payload) {}

五、核心服务实现

Plan plan = planner.create(goal);
planValidator.assertAcyclic(plan);
for (PlanStep step : topologicalSorter.sort(plan.steps())) {
    executor.execute(step, state);
    reviewer.review(step, state);
}

六、接口设计

python agent.py --goal "分析三份合同并生成风险报告"

接口应返回稳定的业务错误码，不要把模型SDK异常、数据库异常或第三方Provider响应原样暴露给前端。

七、可靠性设计

  • 使用幂等键保护重复请求
  • 为外部调用设置Deadline和熔断器
  • 状态变化持久化并可恢复
  • 失败采用分类重试而非无限重试
  • 关键配置版本化

八、安全与权限

  • 租户和用户身份来自认证上下文
  • 高风险动作需要业务授权或人工审批
  • 敏感字段脱敏后再记录
  • 最小权限访问数据库与外部Provider

九、可观测性

  • 请求成功率与P95延迟
  • 单任务Token和费用
  • 重试、回退和取消次数
  • 业务质量与失败类型
  • 队列、缓存或索引命中率

十、测试用例

  • 正常流程
  • 权限不足
  • 外部超时与限流
  • 重复请求
  • 并发更新
  • 配置回滚

十一、可继续扩展的能力

  • 管理后台
  • 灰度配置
  • 多Provider支持
  • 离线评测
  • 成本和SLA看板

十二、从Demo进入生产还需要补什么

“用Python实现Planner-Executor-Reviewer状态机Agent”完成核心功能后，还不能直接承担生产流量。至少需要补齐：

  • 数据库迁移与唯一约束；
  • 统一身份认证和租户隔离；
  • 请求幂等、超时、重试和熔断；
  • 配置、Prompt、模型和工具目录版本；
  • 指标、Trace、审计日志和费用结算；
  • 数据备份、恢复和删除策略；
  • 管理后台与人工处理入口。

推荐把一次请求涉及的所有对象通过request_id或task_id关联起来。发生故障时，应能够从入口还原路由、模型调用、工具执行、状态变化和最终结果。

十三、部署与灰度建议

先在测试环境运行固定回归集，再接入脱敏影子流量。确认稳定后，只对内部账号或少量低风险租户开放。灰度期间同时观察质量、延迟、错误率和成本，任何一项明显恶化都应暂停扩量。

如果工具具有真实副作用，灰度环境必须使用沙箱账号、测试订单或只读模式，避免验证过程影响正式业务。

总结

状态机让每一步输入、输出和失败边界显式化，是生产Agent可恢复和可审计的基础。

延伸阅读

如果你正在关注企业级 AI 应用、Spring AI、RAG、Agent 与大模型工程化落地，欢迎访问 智元界：

https://www.zyentor.com/

智元界将持续分享可运行的技术实战、架构设计、问题排查与企业应用案例。