内容
**[中文文档](README_CN.md)**
# 多智能体协作协议
> **LLM 是执行者，而不是协调者。 大脑（决策逻辑）和骨架（工作流控制）必须由代码实现。**
一种通用的、与存储无关的协议，使多个 LLM 智能体能够通过共享存储异步协作 —— 没有直接通信、确定性代码驱动的编排和严格的 7 字段数据契约。
---
## 为什么存在该协议
当您将多个 LLM 智能体应用于复杂任务时，两种方法占主导地位：
| 方法 | 问题 |
|----------|---------|
| **让 LLM 自身协调** | 不确定性，易出现幻觉，无法保证一致性。 智能体可能会忘记、矛盾或死锁。 |
| **硬编码刚性流程** | 不灵活。 无法适应任务结构、并行分支或部分故障。 |
该项目走第三条路：**LLM 是执行者，代码是大脑**。 智能体在有界上下文中观察、推理和生成。 Python 编排器（`orchestrator.py`）处理验证、阶段转换、冲突仲裁和状态管理 —— 确定性地，每次。
---
## 核心架构
### 三种角色
```
┌─────────────────────────────────────────────────────────────────────┐
│ 角色三角形 │
├─────────────────────────────────────────────────────────────────────┤
│ │
│ 主节点（执行者） ──────── 输出 ────────→ 验证节点（检查者） │
│ │ │ │
│ │ 验证 ←──────── 反馈 │ │
│ │ │ │
│ └────────── 监控 + 仲裁 ←─────── 管理节点（监督者）│
│ │
│ 主节点生成 → 验证节点验证 → 管理节点聚合 │
│ │
└─────────────────────────────────────────────────────────────────────┘
```
| 角色 | 目的 | 核心动作 |
|------|---------|-------------|
| **主节点** | 执行工作 | 生成结果 → 创建子任务 |
| **验证节点** | 独立验证 | 验证 → 反馈（确认 / 拒绝 / 部分） |
| **管理节点** | 监控和合并 | 监控 → 聚合 → 仲裁 → 合并 |
### 关键设计原则
1. **无直接通信** — 所有交互都通过共享存储。 智能体之间从不直接通信。
2. **代码驱动的编排** — `orchestrator.py` 验证、规范化、路由和仲裁。 LLM 永远不会决定流程控制。
3. **与存储无关** — 相同的协议适用于本地 JSON 文件或分布式 MCP 存储（Redis / MySQL / 等）。
4. **多主节点并行** — 多个主节点可以在流水线、并行或分布式模式下同时工作。
5. **严格的数据契约** — 每个记录都遵循相同的 7 字段模式，并在写入前进行验证。
---
## 7 字段数据契约
写入存储的每个记录都必须符合此契约（在 `schemas/record.schema.json` 中定义）：
```json
{
"id": "TASK_001", // 唯一标识符
"who": "master_pentest_01", // 智能体名称（角色任务节点 ID）
"time": 1719600000, // Unix 时间戳（秒）
"event": "task_completed", // 事件类型（17 种定义类型）
"what": "扫描完成", // 一句话描述
"detail": { // 上下文数据（因事件而异）
"result": "发现 3 个漏洞"
},
"trace": { // 可追溯性链
"parent_id": null, // 父任务 ID（根任务为 null）
"session_id": "sess_1719600000" // 协作会话
}
}
```
### 事件类型
| 类别 | 事件 |
|----------|--------|
| **任务生命周期** | `task_started` · `task_pending` · `task_assigned` · `task_processing` · `task_completed` · `task_failed` · `task_blocked` · `task_retrying` · `task_merged` · `task_agent_lost` |
| **健康检查** | `heartbeat` · `checkpoint` |
| **管理事件** | `summary` · `decision` · `protocol_violation` |
| **常规** | `info` · `error` |
### 四个验证关卡
记录在提交到存储之前会经过四个验证步骤：
```
原始 LLM 输出
│
▼
┌──────────────────┐
│ 1. JSON 修复 │ ← 修复智能引号、尾随逗号、Markdown 分隔符
└──────────────────┘
│
▼
┌──────────────────┐
│ 2. 规范化 │ ← 用安全默认值填充缺失字段
└──────────────────┘
│
▼
┌──────────────────┐
│ 3. 验证 │ ← 检查 7 字段契约 + 事件类型枚举
└──────────────────┘
│
▼
┌──────────────────┐
│ 4. 模式检查 │ ← JSON 模式验证（如果安装了 jsonschema）
└──────────────────┘
│
▼
提交到存储
```
---
## 工作循环：读取 → 思考 → 执行 → 写入 → 循环
每个智能体都遵循相同的 5 步循环，但行为因角色而异：
```
┌─────────────────────────────────────────────────────────────────┐
│ 工作循环 │
│ │
│ ┌─────┐ ┌───────┐ ┌──────┐ ┌──────┐ ┌──────┐ │
│ │ 读取│───→│思考 │───→│ 执行 │───→│写入 │───→│ 循环 │──┐ │
│ └─────┘ └───────┘ └──────┘ └──────┘ └──────┘ │ │
│ ▲ │ │
│ └───────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
```
| 阶段 | 主节点 | 验证节点 | 管理节点 |
|-------|--------|----------|---------|
| **读取** | 任务需求、共享知识、任务队列 | 任务队列、父任务上下文 | 所有记录、心跳、异常 |
| **思考** | 可以并行化吗？需要子任务吗？ | 验证方法？分支任务？ | 异常？需要仲裁吗？需要报告吗？ |
| **执行** | 执行核心业务 | 独立验证或执行分支 | 写入警报、决策、摘要 |
| **写入** | 结果 + 子任务 | 验证裁决 | 摘要 / 决策 / 协议违规 |
| **循环** | 更新检查点 → 继续 | 更新检查点 → 继续 | 更新检查点 → 继续 |
---
## 存储模式
协议与存储后端无关。 只有读写基本操作不同。
### 模式 A：本地存储（独立）
使用 `scripts/orchestrator.py` 和 `state/` 中的 JSON 文件。 无需 MCP，离线即可工作。
```bash
# 初始化
python scripts/orchestrator.py init-state --task-space myproject -o state/myproject.json
# 写入记录
python scripts/orchestrator.py normalize-record raw.json -o normalized.json
python scripts/orchestrator.py validate-record normalized.json
python scripts/orchestrator.py add-record --state state/myproject.json --record normalized.json
# 读取记录（支持增量）
python scripts/orchestrator.py read-records --state state/myproject.json --limit 20
python scripts/orchestrator.py read-records --state state/myproject.json --since 1719600000
# 管理操作
python scripts/orchestrator.py generate-summary --state state/myproject.json -o summary.json
python scripts/orchestrator.py detect-anomalies --state state/myproject.json
python scripts/orchestrator.py arbitrate --state state/myproject.json
python scripts/orchestrator.py merge-outputs --state state/myproject.json -o merge.json
python scripts/orchestrator.py merge-outputs --state state/myproject.json --dry-run # 预览
python scripts/orchestrator.py merge-outputs --state state/myproject.json --force # 强制
```
### 模式 B：MCP 存储（分布式）
使用 MCP 工具（`write_record`、`read_records`、`list_tasks`、`create_task_space`）支持 Redis / MySQL / 等。
```python
# 写入记录（仍然需要先规范化和验证）
write_record(normalized_record)
# 读取记录
read_records(filter_who=None, filter_event=None, limit=20)
# 心跳
write_record({event: "heartbeat", who: "master_proj_01", ...})
# 管理操作
read_records(filter_event="summary") # 聚合
read_records(filter_event="heartbeat") # 健康监测
```
> **⚠️ 性能说明**：MCP `read_records` 不支持服务器端时间过滤。 每次增量检查都执行完整扫描 + 客户端过滤。 这对于中小型工作负载来说是没问题的，但在大规模场景下可能会降低性能。 对于高容量场景，更喜欢本地模式（原生 `--since` 支持）或在 MCP 后端添加服务器端过滤。
---
## 可靠性机制
### 1. 心跳协议
每个智能体在活动工作期间每 **2 分钟** 写入一个 `heartbeat` 事件。 管理节点监控这些事件 —— 如果 **5 分钟** 内没有心跳，它会写入一个 `task_agent_lost` 事件并等待人工干预。
```json
{
"id": "HB_1719600500",
"who": "master_pentest_01",
"time": 1719600500,
"event": "heartbeat",
"what": "心跳：主节点存活",
"detail": {
"current_task": "TASK_001",
"status": "executing",
"progress": "60% complete"
},
"trace": { "parent_id": null, "session_id": "sess_001" }
}
```
> **注意**：LLM 智能体在请求-响应模式下运行，无法自我调度。 心跳依赖于外部触发器（例如 Claude Code 中的 `/loop`、cron 作业或用户交互）。 推荐间隔：主节点/验证节点 5–10 分钟，管理节点 2–5 分钟。
### 2. 重试策略
当任务失败时，智能体会自动重试，然后再升级：
| 尝试次数 | 等待时间 | 操作 |
|---------|------|--------|
| 原始 | — | 执行任务 |
| 重试 1 | ~30 秒 | 写入 `task_retrying` 事件 |
| 重试 2 | ~60 秒 | 写入 `task_retrying` 事件 |
| 最大重试次数超过 | — | 升级到 `task_blocked`，并附带 `requires_human_intervention: true` |
**不重试** 以下情况：模式验证错误、逻辑错误、权限被拒绝或任何确定性失败。
### 3. 增量检查协议
智能体持续一个 `last_check_time` 检查点。 每次唤醒时，它们只读取具有 `time > last_check_time` 的记录：
```
完整扫描： O（总记录数） — 每次都读取所有内容
增量： O（新记录数） — 只读取更改的内容 ← 优先
```
检查点存储为专用的 `checkpoint` 事件（不与 `info` 记录混合），以便清晰地分离关注点。
### 4. 原子写入和文件锁定
- **`atomic_write_text`**：写入临时文件，然后使用 `os.replace()` 进行原子提交 — 防止写入中断导致的 state 损坏。
- **`state_lock`**：带有 PID 跟踪和过时锁定检测（5 分钟超时）的 advisory 文件锁定 — 防止并发读取-修改-写入竞争。
### 5. JSON 修复
LLM 通常会生成格式错误的 JSON。 捆绑的 `json_repair.py` 处理：
| 问题 | 修复 |
|-------|-----|
| 智能引号（`""` `''`） | 替换为 ASCII 引号 |
| Markdown 代码分隔符（`````json ``） | 删除分隔符 |
| 尾随逗号在 `}` / `]` 之前 | 删除 |
| 不平衡的括号 | 提取最内层平衡对象 |
---
## 多主节点架构
多个主节点可以并行工作，每个节点都有自己的私有工作空间：
### 流水线模式（不同阶段）
```
主节点_01：阶段 1 → 主节点_02：阶段 2 → 主节点_03：阶段 3
```
### 并行模式（不同分支）
```
主节点_01：分支 A ─┐
├→ 管理节点：聚合
主节点_02：分支 B ─┘
```
### 分布式模式（不同任务）
```
主节点_01：任务 X → 主节点_02：任务 Y → 管理节点：聚合
```
主节点之间从不直接通信。 它们通过共享存储进行协调，管理节点负责聚合和冲突解决。
---
## 合并与冲突解决
当所有任务完成后，管理节点将主节点的输出合并成最终的可交付成果：
### 合并前检查
- ✅ 所有主任务都有 `task_completed` 事件
- ✅ 没有未解决的 `task_failed` / `task_blocked` 事件
- ✅ 没有任务仍处于 `task_processing` 状态
### 内容级解决
编排器对主节点输出执行三种类型的内容分析：
| 检测 | 条件 | 解决 |
|-----------|-----------|------------|
| **相同** | 相同的 `detail.result` 文本 | 去重；优先选择已验证为 `confirmed` 的结果 |
| **冲突** | 相同任务前缀，不同结果 | 优先选择已验证为 `confirmed` 的结果；如果没有，则选择最新时间戳的结果 |
| **重叠** | 一个结果是另一个结果的子字符串 | 保留更完整（更长）的版本 |
### 合并输出示例
```json
{
"id": "MERGE_1719600600",
"event": "task_merged",
"detail": {
"source_tasks": ["TASK_001", "TASK_002"],
"verification_status": {"TASK_001": "confirmed", "TASK_002": "unverified"},
"overlaps": [
{"tasks": ["TASK_001", "TASK_002"], "type": "identical", "kept": "TASK_001"}
],
"conflicts": [],
"merge_order": ["TASK_001"],
"deduped_from": ["TASK_002"],
"status": "delivered"
}
}
```
---
## 工作空间隔离
| 区域 | 目的 | 主节点 | 验证节点 | 管理节点 |
|------|---------|--------|----------|---------|
| 私有工作空间 | 独家读写 | ✅ 仅限自己 | ✅ 仅限自己 | ❌ |
| 摘要区域 | 聚合报告 | ❌ | ❌ | ✅ 只写 |
| 共享知识 | 输入数据 | ✅ 只读 | ✅ 只读 | ✅ 只读 |
| 任务队列 | 待处理/处理/完成 | ✅ 读写 | ✅ 读取 | ✅ 读取 |
**硬约束**：任何智能体都不得写入另一个智能体的私有工作空间。 违规行为被检测并报告为 `protocol_violation` 事件。
---
## 快速入门
### 先决条件
- Python 3.8+
- （可选）`jsonschema>=4.20` 用于完整的 JSON 模式验证
### 1. 安装依赖项
```bash
pip install -r requirements.txt # 可选：仅 jsonschema
```
### 2. 创建任务空间
```bash
python scripts/orchestrator.py create-task-space myproject_20260702
```
### 3. 初始化状态
```bash
python scripts/orchestrator.py init-state --task-space myproject -o state/myproject.json
```
### 4. 运行自我测试
```bash
python scripts/self_test.py
```
### 5. 验证项目结构
```bash
python scripts/validate_structure.py
```
---
## 项目结构
```
multi-agent-collab/
├── SKILL.md # 协议规范（Claude Code 技能）
├── README.md # 此文件
├── requirements.txt # Python 依赖项
│
├── schemas/
│ └── record.schema.json # 7 字段数据契约（JSON 模式）
│
├── prompts/
│ ├── master-workflow.md # 主节点工作流提示
│ ├── verifier-workflow.md # 验证节点工作流提示
│ └── manager-workflow.md # 管理节点工作流提示
│
├── scripts/
│ ├── orchestrator.py # 确定性编排引擎
│ ├── json_repair.py # LLM JSON 输出修复实用程序
│ ├── self_test.py # 自我测试套件
│ └── validate_structure.py # 项目结构验证
│
└── state/
└── example.json # 示例状态文件
```
---
## 编排器命令参考
| 命令 | 描述 |
|---------|-------------|
| `init-state` | 初始化新的协作状态 |
| `validate-record` | 验证记录是否符合 7 字段契约 + JSON 模式 |
| `normalize-record` | 用安全默认值填充缺失字段 |
| `repair-json` | 修复格式错误的 JSON（智能引号、分隔符、尾随逗号） |
| `add-record` | 将验证的记录添加到状态（带有文件锁定） |
| `read-records` | 读取具有过滤和增量支持的记录 |
| `generate-summary` | 生成全局状态报告 |
| `detect-anomalies` | 检测格式错误、超时、协议违规 |
| `arbitrate` | 解决冲突的记录（最新时间戳获胜） |
| `merge-outputs` | 合并完成的主节点输出并进行重复数据删除和冲突解决 |
| `list-tasks` | 列出可用的任务空间 |
| `create-task-space` | 创建新的任务空间 |
| `check-status` | 检查协作状态（阶段、智能体、摘要） |
---
## 用例
### 渗透测试
- **主节点_01**：网络扫描 → **验证节点_01**：验证发现 → **管理节点**：合并报告
### 代码审查
- **主节点_01**：主要审查 → **验证节点_01**：生成单元测试 → **管理节点**：聚合反馈
### 研究
- **主节点_01**：初步研究 → **主节点_02**：文献综述 → **验证节点_01**：交叉检查声明 → **管理节点**：综合报告
### 任何并行工作流
- 将工作分割到多个主节点，验证节点验证，管理节点聚合
---
## 设计决策
### 为什么使用共享存储而非直接消息传递？
直接的 LLM 到 LLM 通信是不确定性和不可观察的。 共享存储提供：
- **可审计性**：每个交互都是一个具有时间戳的持久记录
- **解耦**：智能体不需要了解彼此
- **可恢复性**：任何智能体都可以重新启动并通过增量读取赶上
### 为什么使用代码驱动的编排而非提示驱动的编排？
提示是建议，而不是保证。 一个 LLM 可能会：
- 跳过验证并写入格式错误的记录
- 忘记在写入前检查冲突
- 发明新的事件类型，从而破坏下游消费者
`orchestrator.py` 机械地强制执行这些。 LLM 专注于在其有界上下文中推理和生成结构化输出。
### 为什么使用 7 字段契约？
具有少量定义明确的字段的固定模式：
- 使验证变得微不足道且快速
- 防止智能体之间的模式漂移
- 启用适用于任何领域的通用工具（聚合、仲裁、合并）
### 为什么将验证节点与主节点分离？
独立验证捕获了自我审查无法发现的错误。 验证节点不应该信任主节点的声明 —— 它应用自己的方法并报告实际发现的内容。
---
连接信息
猜你喜欢
暂无推荐服务器