串行接力式多智能体协作系统：完整设计方案
串行接力式多智能体协作系统：完整设计方案
> 继《串行接力式多智能体协作规范》之后的深度设计与系统实现
一、设计理念：从"对话"到"接力"
1.1 现有系统的局限
当前的多智能体系统（如 AutoGen、LangChain Agents、Claude Code）大多采用并联模式：
用户消息 ──┬──> Agent A ──> 响应
├──> Agent B ──> 响应
├──> Agent C ──> 响应
└──> Agent D ──> 响应
特点：
- 不适合：复杂任务流程、质量管控
1.2 串行接力的本质
串行接力式协作模拟的是人类团队的分工协作：
需求 ──> 架构设计 ──> 开发实现 ──> 代码评审 ──> 测试验证 ──> 交付
↓ ↓ ↓ ↓
[阻塞] [阻塞] [阻塞] [阻塞]
[继续] [继续] [继续] [继续]
核心理念：
- 单一信道：每个 Agent 有且只有一个信箱文件
- 谁读谁销毁：读取即处理，处理完才传递
- 绝对可观测：看文件状态就知道系统状态
二、核心机制：文件接力
2.1 通信协议：单一信箱制
原则：
- 每个 Agent 有且只有一个"信箱文件"
- 谁读取文件，谁负责处理
- 处理完毕后，必须：
1. 归档日志（复制到 history_logs/）
2. 清空信箱（表示"消息已送达"）
3. 写入下一个信箱（传递接力棒）
2.2 目录结构设计
/project_root/
├── skills/ # 角色技能定义
│ ├── architect.md # 架构师
│ ├── developer.md # 开发工程师
│ ├── reviewer.md # 评审专家
│ └── tester.md # 测试工程师
│
├── channels/ # 通信信道（核心！）
│ ├── to_architect.md # 人类 -> 架构师
│ ├── to_developer.md # 架构师/评审/测试 -> 开发
│ ├── to_reviewer.md # 开发工程师 -> 评审
│ ├── to_tester.md # 评审 -> 测试工程师
│ └── to_human.md # 架构师/测试 -> 人类（汇报）
│
├── history_logs/ # 历史日志归档
│ ├── 2026-03-22-093000_to_architect.md
│ ├── 2026-03-22-093015_to_developer.md
│ └── ...
│
└── workspace/ # 实际代码工作区
├── src/
├── tests/
└── docs/
2.3 轮询与处理逻辑
这是所有 Agent 通用的底层运行脚本：
#!/usr/bin/env python3
"""
Agent 通用轮询脚本
每个 Agent 运行一个独立的进程
"""
import os
import time
import shutil
from datetime import datetime
class AgentRunner:
def __init__(self, role_name, inbox_file, outbox_files):
self.role_name = role_name
self.inbox = inbox_file
self.outboxes = outbox_files # dict: {recipient: file_path}
def get_timestamp(self):
return datetime.now().strftime("%Y-%m-%d-%H%M%S")
def should_process(self):
"""检测条件：文件存在 + 大小非0 + 有更新"""
if not os.path.exists(self.inbox):
return False
if os.path.getsize(self.inbox) == 0:
return False
return True
def archive_and_clear(self):
"""归档日志 + 清空信箱"""
timestamp = self.get_timestamp()
basename = os.path.basename(self.inbox)
# 归档
archive_path = f"history_logs/{timestamp}_{basename}"
shutil.copy2(self.inbox, archive_path)
# 清空
with open(self.inbox, 'w') as f:
f.write("")
print(f"📂 已归档: {archive_path}")
print(f"🗑️ 已清空: {self.inbox}")
def deliver(self, recipient, content):
"""传递接力棒"""
outbox = self.outboxes.get(recipient)
if not outbox:
print(f"⚠️ 未知接收者: {recipient}")
return False
with open(outbox, 'w') as f:
f.write(content)
print(f"📤 已传递给: {recipient} -> {outbox}")
return True
def process(self, content):
"""核心工作：调用 LLM"""
# 这里调用实际的 LLM API
# 返回结构化结果：
# {
# "action": "deliver",
# "recipient": "developer",
# "content": "..."
# }
pass
def run_loop(self):
"""主循环"""
print(f"🚀 {self.role_name} 已启动")
print(f"📥 监听: {self.inbox}")
while True:
if self.should_process():
print(f"\n📬 收到新消息...")
# 1. 读取任务
with open(self.inbox, 'r') as f:
content = f.read()
# 2. 核心工作
result = self.process(content)
# 3. 归档日志
self.archive_and_clear()
# 4. 传递接力棒
if result.get('action') == 'deliver':
self.deliver(
result['recipient'],
result['content']
)
elif result.get('action') == 'respond_to_human':
self.deliver('human', result['content'])
time.sleep(5) # 轮询间隔
# 使用示例
if __name__ == "__main__":
runner = AgentRunner(
role_name="Architect",
inbox_file="channels/to_architect.md",
outboxes={
"human": "channels/to_human.md",
"developer": "channels/to_developer.md"
}
)
runner.run_loop()
三、角色定义与协作流程
3.1 角色关系图
┌─────────────────────────────────────┐
│ 人类 (Human) │
│ - 下发需求 │
│ - 接收汇报 │
└──────────────┬──────────────────────┘
│
to_architect │ to_human
▼
┌──────────────────────────────────────┐
│ 架构师 (Architect) │
│ - 需求分析 │
│ - 任务拆解 │
│ - 技术选型 │
└───────────────┬──────────────────────┘
│
to_developer │ to_architect
▼
┌──────────────────────────────────────┐
│ 开发工程师 (Developer) │
│ - 代码实现 │
│ - 技术攻关 │
└───────────────┬──────────────────────┘
│
to_reviewer │ to_developer
▼
┌──────────────────────────────────────┐
│ 评审专家 (Reviewer) │
│ - 代码审查 │
│ - 质量把关 │
└───────────────┬──────────────────────┘
│
to_tester │ to_developer
▼
┌──────────────────────────────────────┐
│ 测试工程师 (Tester) │
│ - 测试验证 │
│ - Bug 反馈 │
└──────────────────────────────────────┘
3.2 详细交互流程
#### 阶段一：需求澄清 (Human ↔ Architect)
步骤 1: 人类下发需求
┌─────────────────────────────────────────┐
│ to_architect.md │
├─────────────────────────────────────────┤
│ FROM: human │
│ TIME: 2026-03-22 09:30:00 │
│ │
│ 需求：实现一个用户登录系统 │
│ - 支持用户名密码登录 │
│ - 支持微信扫码登录 │
│ - 需要记住密码功能 │
└─────────────────────────────────────────┘
步骤 2: 架构师分析需求
┌─────────────────────────────────────────┐
│ 架构师思考： │
│ 1. 需求清晰度：80% │
│ 2. 模糊点： │
│ - 微信扫码需要回调地址 │
│ - "记住密码"是多久？ │
│ 3. 技术选型： │
│ - MySQL + Redis │
│ - JWT Token │
└─────────────────────────────────────────┘
步骤 3: 架构师提问（写入 to_human.md）
┌─────────────────────────────────────────┐
│ to_human.md │
├─────────────────────────────────────────┤
│ 架构师提问： │
│ │
│ Q1: 微信扫码登录的回跳地址是什么？ │
│ (例如: https://xxx.com/auth/callback)│
│ │
│ Q2: "记住密码"是指多长时间？ │
│ (7天？30天？永久？) │
│ │
│ Q3: 是否需要支持手机号验证码登录？ │
└─────────────────────────────────────────┘
步骤 4: 人类回答
┌─────────────────────────────────────────┐
│ to_architect.md │
├─────────────────────────────────────────┤
│ 人类回答： │
│ │
│ A1: 回跳地址: https://somingbai.com/ │
│ auth/callback │
│ │
│ A2: 记住密码 = 30天 │
│ │
│ A3: 先不做手机号登录，后续扩展 │
└─────────────────────────────────────────┘
#### 阶段二：任务拆解 (Architect → Developer)
步骤 5: 架构师拆解任务
┌─────────────────────────────────────────┐
│ to_developer.md │
├─────────────────────────────────────────┤
│ TASK START │
│ ID: 001 │
│ DESC: 实现用户注册接口 │
│ - 用户名密码注册 │
│ - 密码 bcrypt 加密存储 │
│ - 返回 JWT Token │
│ ---------------------------------------│
│ ID: 002 │
│ DESC: 实现用户登录接口 │
│ - 用户名密码校验 │
│ - 生成 JWT Token │
│ - 30天有效期 │
│ ---------------------------------------│
│ ID: 003 │
│ DESC: 实现微信扫码登录 │
│ - 生成二维码 │
│ - 轮询扫码状态 │
│ - 回调处理 │
│ ---------------------------------------│
│ ID: 004 │
│ DESC: 实现"记住密码"功能 │
│ - 登录页记住我复选框 │
│ - 记住密码时 30天免登录 │
└─────────────────────────────────────────┘
#### 阶段三：开发执行 (Developer)
步骤 6: 开发工程师处理任务 001
┌─────────────────────────────────────────┐
│ 开发者思考： │
│ - 这是一个注册接口 │
│ - 需要： │
│ 1. 用户名格式验证 │
│ 2. 密码强度验证 │
│ 3. bcrypt 加密 │
│ 4. JWT 生成 │
│ - 技术栈：Python + Flask │
└─────────────────────────────────────────┘
步骤 7: 遇到阻塞，提问架构师
┌─────────────────────────────────────────┐
│ to_architect.md │
├─────────────────────────────────────────┤
│ 开发者提问： │
│ │
│ Q: 数据库用户表结构还没定， │
│ 请确认字段定义 │
│ │
│ 建议的表结构： │
│ id, username, password_hash, │
│ wechat_openid, created_at │
└─────────────────────────────────────────┘
步骤 8: 架构师解答
┌─────────────────────────────────────────┐
│ to_developer.md │
├─────────────────────────────────────────┤
│ 架构师解答： │
│ │
│ 表名: users │
│ 字段: │
│ - id: INT PRIMARY KEY AUTO_INCREMENT │
│ - username: VARCHAR(50) UNIQUE │
│ - password_hash: VARCHAR(255) │
│ - wechat_openid: VARCHAR(100) │
│ - created_at: TIMESTAMP │
│ - updated_at: TIMESTAMP │
└─────────────────────────────────────────┘
#### 阶段四：代码评审 (Developer → Reviewer)
步骤 9: 开发完成，提交评审
┌─────────────────────────────────────────┐
│ to_reviewer.md │
├─────────────────────────────────────────┤
│ FROM: developer │
│ TASK_ID: 001 │
│ │
│ 实现：用户注册接口 │
│ │
│ 代码摘要： │
│ @app.route('/api/register', methods=['POST'])
│ def register(): │
│ data = request.get_json() │
│ username = data.get('username') │
│ password = data.get('password') │
│ │
│ # 密码加密 │
│ password_hash = bcrypt.hash(...) │
│ │
│ # 生成 JWT │
│ token = jwt.encode(...) │
│ │
│ return {'token': token} │
└─────────────────────────────────────────┘
步骤 10: 评审专家审查
┌─────────────────────────────────────────┐
│ 评审意见： │
│ │
│ ⚠️ 问题 1: SQL 注入风险 │
│ 当前直接拼接 SQL │
│ │
│ ⚠️ 问题 2: 密码强度未验证 │
│ 应该检查密码复杂度 │
│ │
│ ✅ 代码结构清晰 │
│ ✅ 错误处理完善 │
└─────────────────────────────────────────┘
步骤 11: 评审驳回
┌─────────────────────────────────────────┐
│ to_developer.md │
├─────────────────────────────────────────┤
│ 评审结果: 驳回 │
│ │
│ ISSUE 1: SQL 注入风险 │
│ - 当前: 直接拼接 SQL │
│ - 修复: 使用参数化查询 │
│ │
│ ISSUE 2: 密码强度未验证 │
│ - 当前: 无验证 │
│ - 修复: 添加密码强度检查 │
│ (至少8位，包含数字和字母) │
└─────────────────────────────────────────┘
步骤 12: 开发修复
┌─────────────────────────────────────────┐
│ 开发者修复： │
│ │
│ 1. 使用 SQLAlchemy ORM │
│ 2. 添加密码强度验证 │
│ │
│ 重新提交评审... │
└─────────────────────────────────────────┘
步骤 13: 评审通过
┌─────────────────────────────────────────┐
│ to_tester.md │
├─────────────────────────────────────────┤
│ 评审结果: 通过 │
│ │
│ TASK_ID: 001 │
│ 开发者: 已修复所有问题 │
│ 可以进行测试 │
└─────────────────────────────────────────┘
#### 阶段五：测试验证
步骤 14: 测试工程师执行测试
┌─────────────────────────────────────────┐
│ 测试用例： │
│ │
│ 1. 注册成功 │
│ 输入: {username: "test", │
│ password: "Test1234"} │
│ 期望: 返回 token │
│ │
│ 2. 用户名已存在 │
│ 输入: {username: "exists", │
│ password: "Test1234"} │
│ 期望: 400 错误 │
│ │
│ 3. 密码强度不足 │
│ 输入: {username: "new", │
│ password: "123"} │
│ 期望: 400 错误 │
└─────────────────────────────────────────┘
步骤 15: 测试通过，生成报告
┌─────────────────────────────────────────┐
│ to_human.md │
├─────────────────────────────────────────┤
│ 任务完成报告 │
│ │
│ TASK_ID: 001 │
│ 状态: ✅ 已完成 │
│ │
│ 测试结果: │
│ - 用例总数: 3 │
│ - 通过: 3 │
│ - 失败: 0 │
│ │
│ 下一任务: 002（用户登录接口） │
└─────────────────────────────────────────┘
四、SKILL.md 指令优化
4.1 Architect SKILL.md
# Role: 架构师 (Architect)
你是项目协作的枢纽，负责需求分析、任务拆解和技术决策。
## 监听文件
- `channels/to_architect.md`
## 输出文件
- `channels/to_human.md` - 向人类提问/汇报
- `channels/to_developer.md` - 下发任务/解答问题
## 工作流程
### 1. 处理需求（来自人类）
当收到人类需求时：
- 分析需求完整性
- 识别模糊点，写入 `to_human.md` 提问
- 需求明确后，拆解任务写入 `to_developer.md`
### 2. 处理问题（来自开发者）
当收到开发者提问时：
- 分析技术问题
- 给出解答，写入 `to_developer.md`
## 任务拆解原则
每个任务必须包含：
- **ID**: 001, 002, 003...
- **DESC**: 完整描述，确保开发者能独立完成
- **技术要求**: 关键技术点
输出格式：
--- TASK START ---
ID: 001
DESC: 实现用户注册接口
技术要求:
--- TASK END ---
## 关键规则
1. **一次只下发一个任务**：等上一个任务完成后再下发下一个
2. **任务必须可独立完成**：避免任务间强依赖
3. **必须记录任务进度**：便于人类了解整体状态
4.2 Developer SKILL.md
# Role: 开发工程师 (Developer)
你是代码实现的核心，负责根据任务描述编写代码。
## 监听文件
- `channels/to_developer.md`
## 输出文件
- `channels/to_architect.md` - 提问
- `channels/to_reviewer.md` - 提交评审
- `channels/to_developer.md` - 继续（如果收到的是解答）
## 工作流程
### 1. 接收任务
当收到新任务时：
- 理解任务描述
- 分析技术难点
- 如有阻塞，写入 `to_architect.md` 提问
- 如无阻塞，开始编写代码
### 2. 接收解答
当收到架构师解答时：
- 继续之前的任务
- 完成后提交评审
### 3. 接收驳回
当收到评审驳回时：
- 修复问题
- 重新提交 `to_reviewer.md`
## 代码规范
1. **一次只处理一个任务**
2. **必须包含测试用例**
3. **提交评审前自我审查**
## 输出格式：提交评审
--- REVIEW REQUEST ---
TASK_ID: 001
SUMMARY: 实现用户注册接口
CHANGES:
---
## 关键规则
- 遇到问题先尝试解决，解决不了再提问
- 提问必须包含上下文和已尝试的方案
4.3 Reviewer SKILL.md
# Role: 评审专家 (Reviewer)
你是代码质量的守护者，负责审查代码并提出改进意见。
## 监听文件
- `channels/to_reviewer.md`
## 输出文件
- `to_developer.md` - 驳回/通过
- `to_tester.md` - 请求测试
## 工作流程
### 1. 代码审查
收到评审请求时：
- 阅读代码
- 检查以下方面：
- 安全性（SQL注入、XSS等）
- 性能（数据库索引、缓存等）
- 可维护性（代码结构、注释等）
- 测试覆盖
### 2. 决策
- 如果有问题：写入 `to_developer.md` 驳回
- 如果通过：写入 `to_tester.md` 请求测试
## 输出格式：驳回
--- REJECT ---
TASK_ID: 001
ISSUES:
位置: routes/register.py:15
当前: 直接拼接 SQL
修复: 使用参数化查询
位置: models/user.py
当前: 无验证
修复: 添加正则检查
---
## 输出格式：通过
--- APPROVE ---
TASK_ID: 001
SUMMARY: 代码质量良好，可以测试
---
4.4 Tester SKILL.md
# Role: 测试工程师 (Tester)
你是质量的最终把关者，负责验证功能是否符合要求。
## 监听文件
- `channels/to_tester.md`
## 输出文件
- `to_developer.md` - Bug反馈
- `to_human.md` - 任务完成报告
## 工作流程
### 1. 测试执行
收到测试请求时：
- 根据任务描述设计测试用例
- 执行测试
- 记录结果
### 2. 决策
- 如果有Bug：写入 `to_developer.md` 反馈
- 如果通过：写入 `to_human.md` 完成报告
## 输出格式：Bug反馈
--- BUG REPORT ---
TASK_ID: 001
BUGS:
重现步骤:
1. POST /api/register
2. username="test", password="Test1234"
期望: 返回 200 + token
实际: 返回 500
原因: 数据库连接未释放
---
## 输出格式：完成报告
--- COMPLETE ---
TASK_ID: 001
STATUS: ✅ 已完成
TEST RESULTS:
---
五、与现有系统的整合
5.1 模式切换
现有 agents 系统是并联模式，串行系统作为补充：
# agents.py 扩展
class AgentSystem:
def __init__(self):
self.mode = "parallel" # 默认并联
def set_mode(self, mode):
"""切换工作模式"""
if mode == "serial":
self.start_serial_workflow()
else:
self.start_parallel_mode()
def handle_message(self, message):
if self.mode == "serial":
return self.handle_serial(message)
else:
return self.handle_parallel(message)
5.2 适用场景
| 场景 | 推荐模式 | 原因 |
|------|----------|------|
| 快速问答 | 并联 | 多角度即时响应 |
| 头脑风暴 | 并联 | 发散思维 |
| 需求分析 | 串行 | 严谨流程 |
| 功能开发 | 串行 | 质量可控 |
| 代码审查 | 串行 | 多人协作 |
| 复杂项目 | 串行 | 阶段把关 |
5.3 启动串行模式
用户可以通过以下方式启动串行流程：
# 方式一：关键词触发
用户输入: "/workflow 开发一个用户登录系统"
# 方式二：UI 按钮
[并联模式] [串行模式]
# 方式三：菜单选择
> 请选择协作模式：
> 1. 快速问答（并联）
> 2. 项目开发（串行）
六、系统的独特优势
6.1 绝对可观测性
# 任何时刻，只需查看 channels/ 目录
$ ls -la channels/
-rw-r--r-- 1 root root 120 Mar 22 09:35 to_developer.md # 👈 接力棒在开发者手中
-rw-r--r-- 1 root root 0 Mar 22 09:30 to_architect.md
-rw-r--r-- 1 root root 0 Mar 22 09:15 to_reviewer.md
-rw-r--r-- 1 root root 0 Mar 22 09:10 to_tester.md
状态判断：
6.2 天然串行化
由于文件"读即清空"的机制：
不可能出现：
┌─────────────────────────────────────┐
│ Developer 正在处理 │
│ Reviewer 同时也在处理 ❌ 并发冲突 │
└─────────────────────────────────────┘
只能是这样：
┌─────────────────────────────────────┐
│ Developer 处理完成 → 清空文件 │
│ Reviewer 检测到更新 → 开始处理 │
│ ✅ 严格串行，无竞争 │
└─────────────────────────────────────┘
6.3 极简实现
不需要：
只需要：
6.4 完整的审计日志
$ ls history_logs/
2026-03-22-093000_to_architect.md # 人类需求
2026-03-22-093015_to_developer.md # 架构师任务
2026-03-22-093100_to_architect.md # 开发者提问
2026-03-22-093105_to_developer.md # 架构师解答
2026-03-22-093200_to_reviewer.md # 提交评审
2026-03-22-093210_to_developer.md # 评审驳回
2026-03-22-093300_to_reviewer.md # 重新提交
2026-03-22-093310_to_tester.md # 评审通过
2026-03-22-093400_to_human.md # 测试完成
每个步骤都有完整记录，可追溯、可回溯。
七、基于Claude Code规范的正确目录结构
> 本章节参考 Claude Code 官方规范设计
7.1 推荐目录结构
project_root/
├── .claude/
│ ├── skills/
│ │ ├── architect/
│ │ │ └── SKILL.md
│ │ ├── developer/
│ │ │ └── SKILL.md
│ │ ├── reviewer/
│ │ │ └── SKILL.md
│ │ └── tester/
│ │ └── SKILL.md
│ └── scripts/
│ ├── workflow_utils.py
│ ├── check_inbox.sh
│ ├── archive_message.sh
│ └── send_message.sh
├── channels/
│ ├── to_architect.md
│ ├── to_developer.md
│ ├── to_reviewer.md
│ ├── to_tester.md
│ └── to_human.md
├── history_logs/
│ └── .gitkeep
└── workspace/
└── .gitkeep
7.2 核心脚本实现
#### workflow_utils.py
#!/usr/bin/env python3
import os
import shutil
from datetime import datetime
from pathlib import Path
class WorkflowUtils:
def __init__(self, base_dir="."):
self.base_dir = Path(base_dir)
self.channels_dir = self.base_dir / "channels"
self.history_dir = self.base_dir / "history_logs"
def get_inbox_path(self, role):
return self.channels_dir / f"to_{role}.md"
def read_message(self, role):
inbox = self.get_inbox_path(role)
if not inbox.exists() or inbox.stat().st_size == 0:
return None
return inbox.read_text(encoding="utf-8")
def clear_inbox(self, role):
inbox = self.get_inbox_path(role)
inbox.write_text("", encoding="utf-8")
def archive_message(self, role, content):
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
archive_path = self.history_dir / f"{timestamp}_to_{role}.md"
archive_path.write_text(content, encoding="utf-8")
return archive_path
def send_message(self, to_role, content):
outbox = self.get_inbox_path(to_role)
outbox.write_text(content, encoding="utf-8")
def process_and_forward(self, from_role, to_role, processor_fn):
content = self.read_message(from_role)
if not content:
return False
new_content = processor_fn(content)
self.archive_message(from_role, content)
self.clear_inbox(from_role)
if new_content:
self.send_message(to_role, new_content)
return True
#### check_inbox.sh
#!/bin/bash
ROLE=$1
INBOX="channels/to_${ROLE}.md"
if [ -f "$INBOX" ] && [ -s "$INBOX" ]; then
echo "有新消息"
exit 0
else
echo "无新消息"
exit 1
fi
#### archive_message.sh
#!/bin/bash
ROLE=$1
TIMESTAMP=$(date +%Y%m%d_%H%M%S)
INBOX="channels/to_${ROLE}.md"
ARCHIVE="history_logs/${TIMESTAMP}_to_${ROLE}.md"
if [ -f "$INBOX" ] && [ -s "$INBOX" ]; then
cp "$INBOX" "$ARCHIVE"
> "$INBOX"
echo "已归档: $ARCHIVE"
fi
#### send_message.sh
#!/bin/bash
ROLE=$1
MESSAGE_FILE=$2
OUTBOX="channels/to_${ROLE}.md"
if [ -f "$MESSAGE_FILE" ]; then
cp "$MESSAGE_FILE" "$OUTBOX"
echo "已发送给 $ROLE"
else
echo "错误: 消息文件不存在"
exit 1
fi
7.3 角色SKILL.md的正确位置
.claude/skills/architect/SKILL.md
.claude/skills/developer/SKILL.md
.claude/skills/reviewer/SKILL.md
.claude/skills/tester/SKILL.md
---
八、局限性与改进方向
8.1 当前局限
1. 轮询延迟：5秒轮询间隔，最长5秒延迟
2. 无并行任务：同时只能处理一个任务
3. 文件锁：多进程需要文件锁防止冲突
4. 无断点续传：进程重启后状态丢失
7.2 改进方向
1. 事件驱动：使用 inotify/FSEvents 替代轮询
2. 任务队列：支持任务优先级
3. 状态持久化：记录任务状态到数据库
4. 超时机制：任务超时自动唤醒
5. 并行 lanes：支持多条并行任务线
7.3 演进路线
v1.0 (当前)
- 基础串行流程
- 文件接力机制
- 基本角色定义
v2.0
- 事件驱动替代轮询
- 任务状态持久化
- 超时和告警机制
v3.0
- 多 lane 并行
- 任务优先级
- 可视化监控面板
八、总结
串行接力式多智能体协作系统，是对人类团队协作方式的数字化模拟：
- 架构师 = 项目经理 + 技术负责人
- 开发者 = 执行者
- 评审专家 = 代码审查者
- 测试工程师 = 质量保障
这套系统的核心价值：
1. 可观测：一眼看清整个流程状态
2. 可追溯：每个决策都有完整记录
3. 可控制：人类可以随时介入、调整
4. 可扩展：新增角色只需添加信箱文件
在 AI 时代，我们需要可控的 AI 协作流程，而非完全自治的黑盒系统。这套"文件接力"机制，正是对这一理念的实践。
---
九、业界最佳实践参考
> 本章节深度参考 Microsoft Azure、OpenAI Agents SDK、CrewAI 等业界领先实践
9.1 五大核心编排模式（Microsoft Azure）
根据 [Microsoft Azure Architecture Center](https://learn.microsoft.com/en-us/azure/architecture/ai-ml/guide/ai-agent-design-patterns) 的权威定义，多智能体系统有五大核心编排模式：
| 模式 | 描述 | 适用场景 | 我们的对应 |
|------|------|----------|------------|
| Sequential（串行） | Agent 按预定义线性顺序执行 | 多阶段流程、渐进优化 | ✅ 核心模式 |
| Concurrent（并发） | 多Agent同时处理同一任务 | 多视角分析、结果聚合 | 扩展模式 |
| Group Chat（群聊） | 多Agent讨论，投票/选举决策 | 协商共识、多方评审 | 评审阶段 |
| Handoff（交接） | Agent之间传递控制权 | 任务路由、专业分工 | ✅ 接力机制 |
| Magentic（磁力） | 动态邀请相关Agent参与 | 开放任务、自适应 | 扩展方向 |
关键洞察（Microsoft原文）：
> "The choice of which agent gets invoked next is deterministically defined as part of the workflow and isn't a choice given to agents in the process."
这正是我们"文件接力"的设计哲学：下一个Agent是谁，由工作流定义，而非Agent自主决定。
9.2 LLM驱动 vs 代码驱动（OpenAI Agents SDK）
[OpenAI Agents SDK](https://openai.github.io/openai-agents-python/multi_agent/) 区分了两种编排方式：
| 方式 | 特点 | 适用场景 |
|------|------|----------|
| LLM-driven | Agent自主决策下一步 | 开放性任务、探索性工作 |
| Code-driven | 代码控制流程 | 确定性流程、可预测结果 |
我们的选择：Code-driven
理由：
9.3 Task 抽象（CrewAI）
[CrewAI](https://docs.crewai.com/en/concepts/tasks) 提出了优秀的 Task 抽象：
Task(
description="任务描述",
expected_output="期望输出格式",
agent=responsible_agent,
guardrails=[validation_function],
human_input=True
)
关键概念值得借鉴：
1. Expected Output（期望输出） - 每个任务必须明确"完成标准"
2. Guardrails（护栏） - 任务输出验证函数，不合格则阻止流程继续
3. Human Input（人类介入） - 关键节点可要求人类审核
9.4 Human-in-the-Loop（人类介入模式）
参考 [Zapier](https://zapier.com/blog/human-in-the-loop/) 和 [Cloudflare Agents](https://developers.cloudflare.com/agents/concepts/human-in-the-loop/) 的最佳实践：
| 阶段 | 场景 | 介入方式 |
|------|------|----------|
| 需求澄清 | 需求模糊或冲突 | 提问 |
| 技术选型 | 多种方案各有优劣 | 选择确认 |
| 评审驳回 | 严重问题需人工确认 | 审核 |
| 最终交付 | 交付前最终确认 | 批准 |
9.5 渐进式复杂度原则
Microsoft Azure 提出的复杂度阶梯：
Level 1: Direct model call（直接调用模型）
↓ 如果Prompt工程解决不了
Level 2: Single agent with tools（单Agent+工具）
↓ 如果单Agent无法可靠处理
Level 3: Multi-agent orchestration（多Agent编排）
核心原则：始终使用**最低复杂度**满足需求。
9.6 与业界方案的对比
| 维度 | 我们的方案 | Microsoft Azure | OpenAI SDK | CrewAI |
|------|------------|----------------|------------|--------|
| 通信机制 | 文件系统 | API调用 | SDK原语 | Task对象 |
| 状态管理 | 文件状态 | 内存/数据库 | 内存 | Task状态 |
| 可观测性 | `ls channels/` | 日志系统 | SDK监控 | Dashboard |
| 复杂度 | 极简 | 中等 | 中等 | 较高 |
| 适用场景 | 确定性流程 | 企业级 | 快速开发 | 复杂协作 |
---
十、总结与演进
10.1 方案优势
1. 极简实现：无需消息队列、无需复杂状态管理
2. 绝对可观测：文件状态即系统状态
3. 自然模拟人类团队：架构师→开发→评审→测试
4. 符合业界最佳实践：借鉴Microsoft、OpenAI、CrewAI
10.2 适用场景
| 场景 | 推荐模式 |
|------|----------|
| 快速问答 | 并联 |
| 头脑风暴 | 并发 |
| 需求开发 | **串行（本文）** |
| 复杂项目 | 串行 + 并发子流程 |
10.3 演进路线
v1.0 (当前)
✅ 文件接力机制
✅ 基本角色定义
✅ 串行流程
v2.0 (规划)
🔄 事件驱动（inotify）替代轮询
🔄 Guardrail 验证机制
🔄 Human-in-the-Loop 审核点
v3.0 (愿景)
📋 可视化流程设计器
📋 多lane并行任务
📋 统计分析面板
---
参考资料
1. Microsoft Azure - [AI Agent Orchestration Patterns](https://learn.microsoft.com/en-us/azure/architecture/ai-ml/guide/ai-agent-design-patterns)
2. OpenAI - [Agents SDK Multi-Agent](https://openai.github.io/openai-agents-python/multi_agent/)
3. CrewAI - [Tasks Documentation](https://docs.crewai.com/en/concepts/tasks)
4. Zapier - [Human-in-the-Loop in AI Workflows](https://zapier.com/blog/human-in-the-loop/)
5. Cloudflare - [Human in the Loop](https://developers.cloudflare.com/agents/concepts/human-in-the-loop/)
---
*本文是《串行接力式多智能体协作规范》的深度扩展，参考了Microsoft、OpenAI、CrewAI等业界最佳实践。*