Agent04：多 Agent 协作 | 旋大爷的博客

多 Agent 协作是指多个 Agent 之间相互配合、分工协作来完成复杂任务的能力。单一 Agent 的能力有限，通过多 Agent 协作可以处理更复杂的场景，提高任务完成的质量和效率。

多 Agent 协作的基本原理

为什么需要多 Agent 协作

场景	说明	示例
复杂任务	需要多种专业能力	生成一份完整的周报
多步骤任务	需要多个步骤协同	产品设计 → 开发 → 测试
多角色任务	需要不同角色配合	客服 → 技术支持 → 运维
知识互补	需要不同领域的知识	法律知识 + 技术知识

多 Agent 协作的优势

优势	说明	示例
能力互补	不同 Agent 擅长不同领域	数据分析 Agent + 文案生成 Agent
效率提升	并行处理多个子任务	同时收集数据和撰写文案
质量保证	多个 Agent 交叉验证	审核 Agent 检查生成内容
可扩展性	灵活添加新的 Agent	添加新的专业 Agent

多 Agent 协作的模式

模式	说明	适用场景
流水线模式	按顺序执行，每个 Agent 处理一个阶段	数据收集 → 分析 → 报告生成
并行模式	多个 Agent 同时执行不同任务	同时收集不同数据源
协作模式	Agent 之间相互沟通和协作	需要协商的任务
竞争模式	多个 Agent 竞争完成任务	选择最优方案

分工策略

基于角色的分工

角色	职责	示例
项目经理	任务分配和进度管理	分配任务给各个 Agent
专家 Agent	处理专业领域任务	法律专家、技术专家
执行 Agent	执行具体操作	调用 API、查询数据
审核 Agent	检查结果质量	检查报告准确性

基于能力的分工

class Agent:
    def __init__(self, name, capabilities, tools):
        self.name = name
        self.capabilities = capabilities
        self.tools = tools
    
    def can_handle(self, task):
        """判断是否能处理任务"""
        for capability in self.capabilities:
            if capability in task["required_capabilities"]:
                return True
        return False
    
    def execute(self, task):
        """执行任务"""
        # 调用工具执行任务
        pass

class TaskAllocator:
    def __init__(self, agents):
        self.agents = agents
    
    def allocate(self, task):
        """分配任务给合适的 Agent"""
        # 找到所有能处理该任务的 Agent
        candidates = [agent for agent in self.agents if agent.can_handle(task)]
        
        if not candidates:
            return None
        
        # 选择最合适的 Agent
        best_agent = max(candidates, key=lambda a: sum(
            1 for cap in a.capabilities if cap in task["required_capabilities"]
        ))
        
        return best_agent

分工策略对比

策略	优点	缺点	适用场景
静态分工	简单、高效	不够灵活	任务类型固定
动态分工	灵活、自适应	复杂度高	任务类型多变
混合分工	兼顾效率和灵活性	实现复杂	大多数场景

通信机制

Agent 通信方式

方式	说明	示例
消息传递	通过消息队列传递信息	Agent A 发送结果给 Agent B
共享存储	通过共享存储交换数据	写入和读取共享数据库
直接调用	一个 Agent 直接调用另一个 Agent	Agent A 调用 Agent B 的方法
广播通知	向所有 Agent 发送消息	通知任务状态变化

通信协议设计

class Message:
    def __init__(self, sender, receiver, type, content, timestamp=None):
        self.sender = sender
        self.receiver = receiver
        self.type = type
        self.content = content
        self.timestamp = timestamp or datetime.now().isoformat()
    
    def to_dict(self):
        return {
            "sender": self.sender,
            "receiver": self.receiver,
            "type": self.type,
            "content": self.content,
            "timestamp": self.timestamp
        }

class MessageBus:
    def __init__(self):
        self.topics = {}
    
    def publish(self, topic, message):
        """发布消息到主题"""
        if topic not in self.topics:
            self.topics[topic] = []
        self.topics[topic].append(message)
    
    def subscribe(self, topic, callback):
        """订阅主题"""
        if topic not in self.topics:
            self.topics[topic] = []
        # 简单实现：注册回调函数
        pass
    
    def send_direct(self, sender, receiver, message):
        """直接发送消息"""
        # 将消息发送给指定的 Agent
        pass

通信模式

模式	说明	适用场景
请求-响应	发送请求，等待响应	调用工具
发布-订阅	发布消息，多个订阅者接收	状态更新
广播	向所有 Agent 发送消息	全局通知
点对点	两个 Agent 直接通信	协作任务

仲裁机制

仲裁的必要性

场景	说明	示例
结果冲突	多个 Agent 返回不同结果	两个 Agent 对同一问题有不同答案
资源竞争	多个 Agent 竞争同一资源	同时调用同一个 API
任务冲突	多个 Agent 的任务相互冲突	修改同一数据
优先级冲突	多个任务的优先级冲突	紧急任务和普通任务

仲裁策略

class Arbiter:
    def __init__(self):
        self.rules = []
    
    def add_rule(self, rule):
        """添加仲裁规则"""
        self.rules.append(rule)
    
    def arbitrate(self, conflicts):
        """仲裁冲突"""
        resolutions = []
        
        for conflict in conflicts:
            resolved = False
            
            for rule in self.rules:
                resolution = rule.apply(conflict)
                if resolution:
                    resolutions.append(resolution)
                    resolved = True
                    break
            
            if not resolved:
                # 默认仲裁策略
                resolutions.append(self.default_arbitration(conflict))
        
        return resolutions
    
    def default_arbitration(self, conflict):
        """默认仲裁策略"""
        if conflict["type"] == "result_conflict":
            # 选择置信度最高的结果
            return max(conflict["results"], key=lambda x: x["confidence"])
        elif conflict["type"] == "resource_conflict":
            # 按优先级分配
            return sorted(conflict["requests"], key=lambda x: x["priority"], reverse=True)[0]
        elif conflict["type"] == "task_conflict":
            # 取消后提交的任务
            return conflict["tasks"][0]

仲裁规则示例

规则	说明	适用场景
置信度优先	选择置信度最高的结果	结果冲突
优先级优先	按优先级处理请求	资源竞争
时间优先	先到先得	任务冲突
权威优先	选择更权威的来源	信息冲突

冲突处理

冲突类型

类型	说明	示例
数据冲突	数据不一致	两个 Agent 对同一数据有不同值
目标冲突	目标不一致	一个 Agent 要保存，另一个要删除
资源冲突	资源不足	多个 Agent 需要同一资源
通信冲突	通信失败	Agent 之间无法正常通信

冲突处理策略

class ConflictHandler:
    def __init__(self, arbiter):
        self.arbiter = arbiter
    
    def detect_conflicts(self, agents):
        """检测冲突"""
        conflicts = []
        
        # 检查数据冲突
        data_conflicts = self.detect_data_conflicts(agents)
        conflicts.extend(data_conflicts)
        
        # 检查目标冲突
        goal_conflicts = self.detect_goal_conflicts(agents)
        conflicts.extend(goal_conflicts)
        
        # 检查资源冲突
        resource_conflicts = self.detect_resource_conflicts(agents)
        conflicts.extend(resource_conflicts)
        
        return conflicts
    
    def resolve_conflicts(self, conflicts):
        """解决冲突"""
        return self.arbiter.arbitrate(conflicts)
    
    def detect_data_conflicts(self, agents):
        """检测数据冲突"""
        conflicts = []
        # 检查各个 Agent 的数据是否一致
        return conflicts
    
    def detect_goal_conflicts(self, agents):
        """检测目标冲突"""
        conflicts = []
        # 检查各个 Agent 的目标是否冲突
        return conflicts
    
    def detect_resource_conflicts(self, agents):
        """检测资源冲突"""
        conflicts = []
        # 检查资源使用情况
        return conflicts

冲突解决模式

模式	说明	适用场景
协商	Agent 之间协商解决	可调和的冲突
投票	投票决定	多个选项可选
仲裁	由仲裁者决定	需要权威决策
回避	暂时回避冲突	非关键冲突

结果合并

结果合并的挑战

挑战	说明	示例
格式不一致	不同 Agent 返回不同格式	一个返回 JSON，一个返回文本
内容重叠	多个 Agent 返回相同内容	重复信息
内容冲突	多个 Agent 返回冲突内容	矛盾的结论
信息不全	需要整合多个 Agent 的信息	每个 Agent 只返回部分信息

结果合并策略

class ResultMerger:
    def __init__(self):
        self.strategies = {
            "concatenate": self.concatenate_strategy,
            "merge": self.merge_strategy,
            "select": self.select_strategy,
            "synthesize": self.synthesize_strategy
        }
    
    def merge(self, results, strategy="synthesize"):
        """合并多个 Agent 的结果"""
        if strategy in self.strategies:
            return self.strategies[strategy](results)
        return self.default_strategy(results)
    
    def concatenate_strategy(self, results):
        """简单拼接"""
        return "\n\n".join([r["content"] for r in results])
    
    def merge_strategy(self, results):
        """合并去重"""
        merged = {}
        for result in results:
            for key, value in result["content"].items():
                if key not in merged or result["confidence"] > merged[key]["confidence"]:
                    merged[key] = {"value": value, "confidence": result["confidence"]}
        return merged
    
    def select_strategy(self, results):
        """选择最优结果"""
        return max(results, key=lambda x: x["confidence"])["content"]
    
    def synthesize_strategy(self, results):
        """综合生成新结果"""
        # 使用模型综合所有结果
        prompt = f"""
        请综合以下多个来源的信息，生成一个完整、一致的回答：
        
        {json.dumps(results, indent=2)}
        
        要求：
        1. 整合所有有用信息
        2. 解决信息冲突
        3. 去除重复内容
        4. 保持逻辑清晰
        """
        
        response = openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content

结果合并模式

模式	说明	适用场景
拼接	简单拼接所有结果	信息互补且不冲突
合并	合并相同字段，去重	结构化数据
选择	选择最优结果	结果冲突时
综合	生成新的综合结果	需要深度整合

多 Agent 协作的常见问题

问题1：协作效率低

表现：多个 Agent 之间协调开销大

解决方案：

优化通信机制
减少不必要的通信
使用异步通信

问题2：结果不一致

表现：不同 Agent 返回的结果不一致

解决方案：

添加结果验证机制
使用仲裁机制
实现结果合并

问题3：死锁

表现：Agent 之间相互等待

解决方案：

设置超时机制
使用锁管理
避免循环依赖

问题4：扩展性差

表现：添加新 Agent 困难

解决方案：

使用标准化接口
实现插件化架构
提供 Agent 注册机制

多 Agent 协作的最佳实践

协作架构设计

┌─────────────────────────────────────────────────────────┐
│                    协作管理层                            │
├─────────────────────────────────────────────────────────┤
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐     │
│  │  任务分配器  │  │  仲裁器      │  │  结果合并器  │     │
│  │ Task        │  │ Arbiter     │  │ Result      │     │
│  │ Allocator   │  │             │  │ Merger      │     │
│  └──────┬──────┘  └──────┬──────┘  └──────┬──────┘     │
└─────────┼────────────────┼────────────────┼─────────────┘
          │                │                │
          ▼                ▼                ▼
┌─────────────────────────────────────────────────────────┐
│                    Agent 层                             │
├─────────────────────────────────────────────────────────┤
│  ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐         │
│  │Agent1│ │Agent2│ │Agent3│ │Agent4│ │Agent5│         │
│  │(专家)│ │(执行)│ │(审核)│ │(分析)│ │(生成)│         │
│  └──────┘ └──────┘ └──────┘ └──────┘ └──────┘         │
└─────────────────────────────────────────────────────────┘

协作流程

1. 任务接收
   - 接收用户任务
   - 分析任务需求
   - 确定所需能力

2. 任务分配
   - 根据 Agent 能力分配任务
   - 确定协作模式（流水线/并行/协作）
   - 设置任务优先级

3. Agent 执行
   - 各个 Agent 执行子任务
   - Agent 之间通信协作
   - 处理执行过程中的冲突

4. 结果收集
   - 收集各个 Agent 的结果
   - 检测结果冲突
   - 进行结果验证

5. 结果合并
   - 合并多个 Agent 的结果
   - 解决信息冲突
   - 生成最终报告

6. 结果反馈
   - 返回结果给用户
   - 收集用户反馈
   - 更新 Agent 知识

评估指标

指标	定义	目标值
协作效率	完成任务的时间	根据任务类型
结果质量	结果的准确性和完整性	> 90%
冲突解决率	成功解决冲突的比例	> 95%
扩展性	添加新 Agent 的难度	低

项目判断清单

需要多种专业能力 → 使用多 Agent 协作
需要并行处理 → 使用并行模式
需要审核环节 → 添加审核 Agent
结果可能冲突 → 添加仲裁机制
需要整合信息 → 实现结果合并
协作效率低 → 优化通信机制
扩展性差 → 使用插件化架构
需要监控 → 添加协作日志和指标