多agent协作规范

多Agent协作规范

多 Agent 协作的模式、通信与约束

---

协作模式

规则：多 Agent 协作必须选择明确的协作模式，不同模式适用于不同场景，禁止模式混用导致职责混乱。

[Requirement] 多 Agent 系统必须有且仅有一个”编排者（Orchestrator）“，负责任务分发和结果汇总。

[Standard] 根据任务特征选择协作模式：简单并行用”分工模式”，复杂推理用”层级模式”，质量敏感用”评审模式”。

[Default] 3 个及以下 Agent 使用”分工模式”，超过 3 个或任务复杂度高时使用”层级模式”。

三种核心协作模式

模式	结构	适用场景	优点	缺点
分工模式 (Pipeline)	Agent 串行接力，前一个输出是后一个输入	线性流程任务	简单可控	瓶颈在最慢的 Agent
层级模式 (Hierarchical)	主 Agent 调度子 Agent，子 Agent 并行执行	复杂任务、多模块	可扩展、并行度高	编排复杂
评审模式 (Reviewer)	生成 Agent + 评审 Agent，评审不过则返工	质量敏感任务	质量有保障	迭代次数多

模式结构图

分工模式 (Pipeline)：
Agent A → Agent B → Agent C → 最终输出

层级模式 (Hierarchical)：
        ┌→ Agent A
Orchestrator ─→ Agent B → 汇总 → 最终输出
        └→ Agent C

评审模式 (Reviewer)：
Agent A (生成) → Agent B (评审) → 通过 → 最终输出
                     │
                     └─ 不通过 → 反馈给 A → 重新生成

示例：

场景：实现一个完整的用户注册功能

层级模式：
Orchestrator（主 Agent）
├─ 子 Agent A：生成 schema 和数据模型
├─ 子 Agent B：生成核心业务逻辑
├─ 子 Agent C：生成 API 路由
├─ 子 Agent D：生成单元测试
└─ 主 Agent 汇总，检查一致性

反模式：

• 多个 Agent 之间没有编排者，各自为政
• 同时使用分工和层级模式，职责不清
• 简单任务用复杂的层级模式，过度设计

---

通信机制

规则：Agent 之间的通信必须通过明确的消息协议，禁止直接共享内部状态，所有交互通过消息传递完成。

[Requirement] Agent 之间通信必须使用结构化消息（JSON），包含：发送者、接收者、消息类型、内容、时间戳。

[Requirement] Agent 不得直接读写其他 Agent 的内部状态，只能通过消息请求获取。

[Standard] 消息分为 4 种类型：任务消息、结果消息、反馈消息、控制消息。

[Default] 消息通道使用共享内存/消息队列，不使用全局变量。

消息结构

{
  "message_id": "msg_20250101_001",
  "from": "orchestrator",
  "to": "agent_code",
  "type": "task",
  "content": {
    "task_id": "T1",
    "description": "实现用户注册接口",
    "context": {...},
    "constraints": [...]
  },
  "timestamp": "2025-01-01T10:00:00Z"
}

消息类型

类型	说明	发送方 → 接收方
task	任务分配	编排者 → 执行者
result	任务结果	执行者 → 编排者
feedback	评审反馈	评审者 → 执行者
control	控制指令（停止/重试/状态查询）	编排者 → 任意

通信约束

约束	说明
单向依赖	子 Agent 只与编排者通信，不互相通信
消息幂等	同一消息重复投递不产生副作用
超时机制	每条消息有超时时间，超时则重试或报错
消息队列	消息持久化，避免丢失

示例：

// 任务消息
{
  "from": "orchestrator",
  "to": "agent_schema",
  "type": "task",
  "content": {
    "task_id": "T1",
    "description": "定义 UserCreate schema",
    "constraints": ["邮箱格式校验", "密码长度 8-64"]
  }
}
 
// 结果消息
{
  "from": "agent_schema",
  "to": "orchestrator",
  "type": "result",
  "content": {
    "task_id": "T1",
    "status": "success",
    "output": "class UserCreate(BaseModel): ..."
  }
}

反模式：

• Agent 之间通过全局变量传递数据
• 消息没有结构化，纯文本沟通
• 子 Agent 之间直接通信，绕过编排者
• 消息没有超时机制，死等不返回

---

角色划分

规则：每个 Agent 必须有明确的角色定义和职责边界，禁止角色重叠或职责不清。

[Requirement] 每个 Agent 必须定义：角色名称、职责范围、输入、输出、不可做的事。

[Requirement] 同一系统中不得有两个职责完全相同的 Agent（避免冗余和冲突）。

[Standard] 角色按职责分为：编排者、执行者、评审者、工具代理四类。

[Default] 每个 Agent 的职责范围不超过 3 个核心任务，超过则拆分。

标准角色定义

角色	职责	输入	输出	不可做
编排者 (Orchestrator)	任务拆分、分发、汇总	原始需求	最终交付物	不直接编写代码
执行者 (Executor)	按任务生成代码/文档	子任务描述	代码/文档	不做架构决策
评审者 (Reviewer)	审查产出物质量	待评审内容	评审意见	不直接修改代码
工具代理 (Tool Agent)	执行具体工具调用	工具调用请求	工具执行结果	不做逻辑判断

Agent 角色卡模板

# agents/coder_agent.yaml
name: coder_agent
role: 代码执行者
responsibilities:
  - 按任务描述生成代码
  - 生成配套的单元测试
  - 运行 lint 和测试并报告结果
inputs:
  - task_description: 任务描述
  - context: 相关代码和规范
outputs:
  - code: 生成的代码
  - tests: 单元测试
  - report: 自检报告
forbidden:
  - 不得修改架构设计
  - 不得引入未审批的第三方依赖
  - 不得硬编码密钥
model: gpt-4o
max_tokens: 8000

示例：

4 Agent 协作实现注册功能：

1. Orchestrator（编排者）
   - 拆分任务为 T1-T5
   - 分发给对应 Agent
   - 汇总结果

2. Schema Agent（执行者）
   - 生成 UserCreate、UserResponse 模型
   - 输出：schema 代码

3. Coder Agent（执行者）
   - 生成 register_user 业务逻辑
   - 输出：service 代码

4. Reviewer Agent（评审者）
   - 审查 schema 和 service 代码
   - 输出：评审意见，不通过则打回

反模式：

• 所有 Agent 都是”全能型”，什么都做
• 两个 Agent 负责同一块代码，互相覆盖
• Agent 没有”不可做”的约束，越权操作

---

约束规则

规则：多 Agent 协作必须遵守一系列约束，防止冲突、死锁、资源浪费和质量失控。

[Requirement] 同一文件同一时间只能被一个 Agent 修改，其他 Agent 如需修改必须等待。

[Requirement] Agent 执行任务有最大轮次限制，超过则终止并人工介入，防止死循环。

[Requirement] Agent 不得自行启动新的 Agent，Agent 的创建和调度由编排者统一管理。

[Standard] Agent 的输出必须经过校验（格式/内容/约束）才能传递给下一个环节。

[Standard] 所有 Agent 共享同一份规范上下文（System Prompt），不得各自为政。

约束清单

编号	约束	级别	说明
C1	文件锁	Requirement	同一文件同时只能一个 Agent 写
C2	最大轮次	Requirement	单个 Agent 最多执行 10 轮
C3	统一规范	Standard	所有 Agent 使用同一套编码规范
C4	输出校验	Standard	Agent 输出必须格式正确、约束满足
C5	禁止自启	Requirement	Agent 不得自行创建子 Agent
C6	超时机制	Standard	每个任务有超时，超时则终止
C7	变更范围	Standard	单个 Agent 单次变更 ≤ 400 行
C8	禁止越权	Requirement	Agent 不得做角色定义外的事

并发控制示例

# 文件锁：防止多个 Agent 同时修改同一文件
import fcntl
 
class FileLock:
    def __init__(self, file_path):
        self.file_path = file_path
        self.lock_file = file_path + ".lock"
 
    def acquire(self):
        self.fd = open(self.lock_file, "w")
        fcntl.flock(self.fd, fcntl.LOCK_EX)
 
    def release(self):
        fcntl.flock(self.fd, fcntl.LOCK_UN)
        self.fd.close()

示例：

违反约束 C1 的场景：
Agent A 正在修改 user.py 的 register_user 函数
Agent B 同时修改 user.py 的 delete_user 函数
→ 合并时产生冲突，或一方覆盖另一方

正确做法：
Agent A 获取 user.py 的文件锁，完成后释放
Agent B 等待锁释放后再修改

反模式：

• 多个 Agent 同时改同一个文件
• Agent 陷入死循环，无限重试
• Agent 自己又启动了新的 Agent，导致 Agent 爆炸
• 不同 Agent 使用不同的编码规范，产出风格不一致

---

调试

规则：多 Agent 系统的调试必须可观测、可追溯、可复现，禁止”黑盒运行、出问题靠猜”。

[Requirement] 每个 Agent 的每次调用必须记录：输入、输出、耗时、Token 用量、是否成功。

[Requirement] Agent 之间的消息通信必须全部记录，支持按任务 ID 回溯完整链路。

[Standard] 提供”单步执行”模式，可逐 Agent 调试，暂停在任意环节。

[Standard] 关键节点（任务分发、结果汇总、评审反馈）必须有人类可介入的检查点。

[Suggestion] 使用可视化工具展示 Agent 协作拓扑和消息流转。

日志记录要求

记录项	说明	示例
trace_id	全局追踪 ID，贯穿整个协作流程	trace_abc123
agent_id	Agent 标识	coder_agent
task_id	任务 ID	T1
input	Agent 输入（脱敏后）	任务描述、上下文摘要
output	Agent 输出（脱敏后）	生成的代码摘要
duration	执行耗时（毫秒）	12500
tokens	Token 用量	prompt: 2000, completion: 500
status	成功/失败	success
error	失败原因（如有）	timeout

调试模式

模式	说明	适用场景
正常模式	全自动执行	稳定的任务
单步模式	每步暂停，人工确认后继续	调试、首次运行
干跑模式 (Dry-run)	只生成计划，不实际执行	验证任务拆分
回放模式	按记录重新执行，对比结果	复现问题

示例：

# 启动单步调试模式
python run_agents.py --mode step --trace-id trace_abc123
 
# 输出示例
[Step 1/5] Orchestrator → agent_schema: task T1
  Input: 定义 UserCreate schema
  Output: class UserCreate(BaseModel): ...
  Duration: 8.2s, Tokens: 1500
  [PRESS ENTER TO CONTINUE]
 
[Step 2/5] Orchestrator → agent_coder: task T2
  Input: 实现 register_user
  ...

反模式：

• 多 Agent 运行后没有任何日志，出问题无法定位
• 调试时直接全量运行，不支持单步
• Token 用量不记录，成本失控
• 消息通信不记录，无法回溯 Agent 之间的交互