AI Agent团队协作协议设计与实现：关机与审批机制

1. 项目概述

在AI Agent开发中，团队协作协议的设计至关重要。本章将深入探讨如何构建可靠的团队协议机制，重点解析关机协议和计划审批协议的实现原理。通过引入request_id、状态机和追踪器三要素，我们将原本松散的团队沟通升级为结构化交互模式。

1.1 核心需求解析

在之前的版本(s09)中，团队成员间的通信存在两个主要问题：

1. 关机操作过于粗暴：直接标记状态为"shutdown"可能导致数据损坏或状态不一致
2. 高风险操作缺乏审批：团队成员可以直接执行重构等高风险操作，没有审查机制

这两个场景本质上都需要一个请求-响应的交互模式：

• 关机协议：领导发起请求 → 队友审批 → 执行关闭
• 计划审批：队友发起请求 → 领导审批 → 执行计划

2. 协议设计核心要素

2.1 协议三要素

2.1.1 request_id机制

每个请求分配唯一标识符，确保请求和响应能正确关联：

python复制# 生成8位UUID作为request_id
req_id = str(uuid.uuid4())[:8]

为什么需要request_id？

• 解决多请求并发时的响应匹配问题
• 避免"这个响应是对哪个请求的回复"的歧义
• 便于状态追踪和问题排查

2.1.2 有限状态机(FSM)

定义统一的状态流转规则：

code复制[pending] → [approved] 或 [rejected]

状态机优势：

• 防止非法状态转换
• 明确每个状态下的允许操作
• 简化状态管理逻辑

2.1.3 追踪器(Tracker)

使用内存字典记录请求状态：

python复制# 关机请求追踪器
shutdown_requests = {
    "abc12345": {
        "target": "alice",
        "status": "pending"  # 或 approved/rejected
    }
}

# 计划请求追踪器 
plan_requests = {
    "xyz54321": {
        "from": "bob",
        "plan": "重构认证模块...",
        "status": "pending"
    }
}

选择内存存储而非文件的原因：

1. 请求状态是临时数据，进程退出后无需保留
2. 内存访问速度更快，减少IO延迟
3. 简化实现，避免文件锁等复杂问题

2.2 线程安全设计

由于团队成员运行在独立线程中，需要使用锁机制保证数据安全：

python复制_tracker_lock = threading.Lock()

# 更新状态时的标准写法
with _tracker_lock:
    shutdown_requests[req_id]["status"] = "approved"

锁的使用要点：

• 任何对共享数据结构(shutdown_requests/plan_requests)的修改都必须加锁
• 锁的范围应尽量小，只包含必要的操作
• 避免在锁内执行耗时操作(如网络请求)

3. 协议实现细节

3.1 关机协议完整流程

3.1.1 领导发起关机请求

python复制def handle_shutdown_request(teammate: str) -> str:
    req_id = str(uuid.uuid4())[:8]
    
    # 记录到追踪器
    with _tracker_lock:
        shutdown_requests[req_id] = {"target": teammate, "status": "pending"}
    
    # 发送请求消息
    BUS.send(
        "lead",              # 发送者
        teammate,            # 接收者
        "Please shut down gracefully.",
        "shutdown_request",  # 消息类型
        {"request_id": req_id}  # 附加参数
    )
    return f"Shutdown request {req_id} sent"

3.1.2 队友处理关机请求

队友的agent loop中处理流程：

1. 读取收件箱中的shutdown_request消息
2. LLM根据当前工作状态决定是否批准
3. 调用shutdown_response工具回复

python复制def _exec(self, sender: str, tool_name: str, args: dict) -> str:
    if tool_name == "shutdown_response":
        req_id = args["request_id"]
        approve = args["approve"]
        
        # 更新追踪器状态
        with _tracker_lock:
            if req_id in shutdown_requests:
                shutdown_requests[req_id]["status"] = "approved" if approve else "rejected"
        
        # 发送响应
        BUS.send(
            sender,
            "lead",
            args.get("reason", ""),
            "shutdown_response",
            {"request_id": req_id, "approve": approve}
        )
        
        # 如果批准关闭，标记退出标志
        if approve:
            self.should_exit = True
            
        return f"Shutdown {'approved' if approve else 'rejected'}"

3.1.3 优雅退出机制

队友线程不会立即退出，而是完成当前轮次所有操作后再退出：

python复制def _teammate_loop(self):
    should_exit = False
    for _ in range(50):  # 最大循环次数
        # 处理消息和工具调用...
        
        if should_exit:
            break  # 下一轮循环才会真正退出
            
    # 更新最终状态
    member = self._find_member(name)
    if member:
        member["status"] = "shutdown" if should_exit else "idle"

延迟退出的设计考量：

• 确保当前轮次的所有工具调用都能完成
• 避免中断文件写入等关键操作
• 让LLM有机会保存工作状态或发送最后的消息

3.2 计划审批协议实现

3.2.1 队友提交计划

python复制def _exec(self, sender: str, tool_name: str, args: dict) -> str:
    if tool_name == "plan_approval":
        plan_text = args.get("plan", "")
        req_id = str(uuid.uuid4())[:8]
        
        # 记录到追踪器
        with _tracker_lock:
            plan_requests[req_id] = {
                "from": sender,
                "plan": plan_text,
                "status": "pending"
            }
        
        # 发送审批请求
        BUS.send(
            sender,
            "lead",
            plan_text,
            "plan_approval_response",
            {"request_id": req_id, "plan": plan_text}
        )
        return f"Plan submitted (request_id={req_id})"

3.2.2 领导审批计划

python复制def handle_plan_review(request_id: str, approve: bool, feedback: str = "") -> str:
    # 查找对应请求
    with _tracker_lock:
        req = plan_requests.get(request_id)
    if not req:
        return f"Error: Unknown request_id '{request_id}'"
    
    # 更新状态
    with _tracker_lock:
        req["status"] = "approved" if approve else "rejected"
    
    # 发送审批结果
    BUS.send(
        "lead",
        req["from"],
        feedback,
        "plan_approval_response",
        {"request_id": request_id, "approve": approve, "feedback": feedback}
    )
    return f"Plan {req['status']}"

3.2.3 队友处理审批结果

队友在下一轮循环中会收到审批结果，并根据结果决定后续操作：

• 批准：执行计划
• 拒绝：调整计划或放弃

4. 协议对比与工具扩展

4.1 关机协议 vs 计划审批协议

4.2 工具扩展情况

领导工具从9个增加到12个：

• 新增shutdown_request：发起关机请求
• 新增shutdown_response：查询关机状态
• 新增plan_approval：审批计划

队友工具从6个增加到8个：

• 新增shutdown_response：响应关机请求
• 新增plan_approval：提交计划审批

5. 关键实现技巧

5.1 消息总线设计

python复制class MessageBus:
    def __init__(self, inbox_dir: Path):
        self.dir = inbox_dir
        self.dir.mkdir(parents=True, exist_ok=True)
    
    def send(self, sender: str, to: str, content: str, 
             msg_type: str = "message", extra: dict = None) -> str:
        msg = {
            "type": msg_type,
            "from": sender,
            "content": content,
            "timestamp": time.time()
        }
        if extra:
            msg.update(extra)
        # 写入对应队友的收件箱文件
        inbox_path = self.dir / f"{to}.jsonl"
        with open(inbox_path, "a") as f:
            f.write(json.dumps(msg) + "\n")
    
    def read_inbox(self, name: str) -> list:
        inbox_path = self.dir / f"{name}.jsonl"
        if not inbox_path.exists():
            return []
        # 读取后清空文件
        messages = [json.loads(line) for line in inbox_path.read_text().splitlines() if line]
        inbox_path.write_text("")
        return messages

设计要点：

• 每个队友有独立的.jsonl收件箱文件
• 读取后自动清空，实现消息消费语义
• 支持任意类型的附加字段(extra)
• 文件操作是线程安全的(操作系统保证)

5.2 LLM行为引导

通过工具schema和system prompt确保LLM按协议响应：

python复制# shutdown_response工具定义
{
    "name": "shutdown_response",
    "description": "Respond to a shutdown request",
    "input_schema": {
        "type": "object",
        "properties": {
            "request_id": {"type": "string"},
            "approve": {"type": "boolean"},
            "reason": {"type": "string"}
        },
        "required": ["request_id", "approve"]
    }
}

# system prompt片段
"You are '{name}'. Submit plans via plan_approval before major work. Respond to shutdown_request with shutdown_response."

引导策略：

1. 工具schema强制要求request_id和approve参数
2. system prompt明确告知何时使用哪个工具
3. 错误响应会被API拒绝，促使LLM纠正

6. 实践经验与避坑指南

6.1 常见问题排查

1. 请求状态不更新

   • 检查_tracker_lock是否正确使用
   • 确认request_id在请求和响应中一致
   • 验证追踪器字典的键是否存在

2. LLM不使用协议工具响应

   • 检查system prompt是否包含协议说明
   • 验证工具schema的required字段
   • 确保错误响应有明确的反馈

3. 线程阻塞或死锁

   • 避免在锁内执行耗时操作
   • 使用with语句管理锁，确保释放
   • 考虑锁的粒度，必要时拆分大锁

6.2 性能优化建议

1. 请求ID生成：使用uuid4的前8位，在唯一性和长度间取得平衡
2. 状态追踪：定期清理已完成的请求，防止内存膨胀
3. 消息处理：批量读取收件箱，减少IO操作
4. 锁优化：区分读写锁，读多写少时使用RLock

6.3 扩展思路

1. 超时机制：为请求添加超时处理，避免长期pending
2. 优先级系统：区分紧急请求和普通请求
3. 协议版本控制：支持协议升级和兼容
4. 审计日志：记录完整的协议交互过程

7. 完整实现代码

完整代码请参考项目仓库

关键实现要点：

1. 使用Python标准库实现，无额外依赖
2. 线程安全的追踪器和消息总线
3. 清晰的协议状态流转
4. 详细的错误处理和日志

8. 测试与验证

建议测试用例：

python复制# 测试关机协议
1. 创建队友：Spawn alice as a coder
2. 请求关机：Request her shutdown
3. 查看状态：List teammates

# 测试计划审批
1. 创建队友：Spawn bob with a risky task
2. 提交计划：Have him submit a plan
3. 审批拒绝：Review and reject his plan
4. 再次提交：Have him submit revised plan
5. 审批通过：Approve the new plan

验证要点：

• 请求和响应的request_id是否匹配
• 状态是否按预期流转(pending→approved/rejected)
• 线程是否优雅退出
• 高并发下是否有竞态条件