AI Agent Engineer

AI Agent Engineer

离线协作模拟项目

7 min read

离线协作模拟项目

项目目标

用一个完全确定性的调度器模拟多 Agent 协作，而不调用模型。你将验证：依赖满足后才运行、角色权限受限、失败只有限重试、预算是全局的、相同 key/相同本地排序键 JSON digest 的结果不会重复提交、相同 key/不同结果会冻结为 needs_review、每一步都有状态版本追踪事件。

文件

这些夹具不包含真实会话、凭据或外部数据。

场景结构

角色声明 capabilities；任务声明 owner、requires、capability、max_attempts 和 outcome_plan。outcome_plan 是确定性结果序列，例如 transient_error 后 success，用来测试重试，不代表模型推理。

调度器每轮：

  1. 找出依赖已成功的 pending 任务；
  2. 检查全局 step_budget；
  3. 检查任务所有者是否具有所需 capability；
  4. 产生一次预设结果；
  5. 根据结果和 max_attempts 经 running 进入白名单允许的 succeeded、pending、failed 或 denied 状态；
  6. 将失败/拒绝/冲突的所有未开始依赖后代阻断到不动点，再区分成功、上游失败阻塞和依赖死锁。

结果接收器只接受 running 任务的成功结果。它保存 (task_id, idempotency_key, payload digest)，其中 digest 是本示例固定的 Python 排序键 JSON 序列化；它可重复，但不是跨语言 canonical-JSON 标准。同 key/同 digest 记录 duplicate_result_ignored；同 key/不同 digest 记录 result_conflict_detected，把任务升级为 needs_review，清空公开 result 并标为 result_trust: conflicted。为使离线断言可复核，报告在 result_conflicts 中深拷贝两份合成 payload、digest、到达序号和状态版本，并将所有尚未开始的依赖后代阻断。真实系统应把受保护的证据引用、state_version、lease 和外部 receipt 放进持久存储，且隔离已经开始的下游副作用以供人工查询/补偿。

运行

在 PowerShell 7 中进入本目录后执行：

python -B .\examples\collaboration_simulator.py .\examples\scenario.json
python -B -m unittest discover -s .\examples -p 'test_*.py' -v
python -B -O -m unittest discover -s .\examples -p 'test_*.py'
python -B -W error -m unittest discover -s .\examples -p 'test_*.py'
python -B -O -W error -m unittest discover -s .\examples -p 'test_*.py'

也可把报告写入临时路径：

python -B .\examples\collaboration_simulator.py .\examples\scenario.json --output "$env:TEMP\multi-agent-report.json"

脚本不访问网络，也不在知识库内生成运行产物。

成功返回退出码 0；工作流以 failed、deadlock、needs_review 或 budget_exhausted 等非成功终态结束时返回 1；场景契约错误返回 2。66 项测试覆盖严格 JSON、角色/任务合同、权限、白名单状态迁移、逆序多层依赖、重试、死锁、幂等、结果冲突冻结、预算、事件序列和 CLI。

读懂输出

报告包含 status、steps_used、tasks 和 events。检查：

  • transient_error 之后是否只有允许次数内的下一次 attempt；
  • 缺少 capability 是否进入 denied，而不是换角色绕过；
  • 依赖失败是否让下游 blocked；
  • 同 key、不同结果是否进入 needs_review、清空公开 result，并保留仅供对账的合成冲突证据；
  • event 是否带 task_id、attempt、state_version、from、to 和 reason；
  • steps_used 是否不超过 step_budget。

改造练习

  1. 在场景中新增一个和汇总任务并行的审查任务，观察关键路径。
  2. 把某角色 capability 删除，确认策略拒绝。
  3. 把 step_budget 调小，确认终态为 budget_exhausted。
  4. 制造 A 依赖 B、B 依赖 A，确认报告指出 deadlock。
  5. 为同一 (task_id, idempotency_key) 分别提交字段顺序不同的相同 JSON 与不同 JSON：前者不得多次改变状态，后者必须保留 digest/证据、冻结为 needs_review、清空公开结果，并阻断未开始的下游任务。

掌握检查

  • 我能解释为什么这个项目模拟的是运行时协作，而不是多个“人格”。
  • 我能指出任务合同、状态机、预算和权限分别在哪一层校验。
  • 我能区分 retryable、denied、blocked、deadlock 和 budget_exhausted。
  • 我能区分可安全重放的重复结果与必须人工对账的同 key 冲突，也能说明本示例的 digest 规则不是通用互操作标准。
  • 我能根据事件追踪找到最早失败点。
  • 我能说明真实系统还需模型调用适配器、持久存储、认证、限流和敏感日志治理。

自测题

  1. 如果任务已经 succeeded，迟到的 failed 结果应如何处理？
  2. 为什么全局预算不能由各 Agent 自己统计？
  3. 一个下游任务因依赖 failed 无法运行，它自身应记 failed 还是 blocked？
  4. 怎样把此模拟器改为可恢复而不重复副作用？
  5. 为什么“同 key 但 payload 不同”不能按先到或后到的结果自动继续？

局限

项目没有并发线程、网络传输、真实模型、持久队列或人工审批 UI。它验证的是控制语义，尤其不证明跨进程 lease、原子 compare-and-set、真实身份、受保护证据存储或人类对账恢复。把它迁移到框架前，应保留相同状态、预算和验收测试，并补足这些持久控制面。

参考与下一步

回到 多 Agent 协作目录，再结合 运行监控为真实系统建立追踪。

Backlinks