从单兵作战到团队协作

单个Agent的能力受限于单一模型的上下文窗口和推理能力。当任务复杂到需要多种专业能力时，多Agent协作成为必然选择。但如何编排多个Agent高效协作，是一个充满设计权衡的工程问题。

编排模式分类

1. 中心化编排（Orchestrator模式）

一个中心编排器分发任务给多个专业Agent：

class Orchestrator:
    def __init__(self):
        self.agents = {
            "researcher": ResearchAgent(),
            "writer": WriterAgent(),
            "reviewer": ReviewerAgent(),
            "fact_checker": FactCheckAgent(),
        }
        self.task_decomposer = TaskDecomposer()
    
    def execute(self, task):
        # 1. 任务分解
        subtasks = self.task_decomposer.decompose(task)
        
        # 2. 分配给合适的Agent
        results = {}
        for subtask in subtasks:
            agent = self._select_agent(subtask)
            result = agent.execute(subtask, context=results)
            results[subtask.id] = result
        
        # 3. 综合结果
        final = self._synthesize(results)
        return final
    
    def _select_agent(self, subtask):
        """根据子任务类型选择Agent"""
        if subtask.type == "research":
            return self.agents["researcher"]
        elif subtask.type == "writing":
            return self.agents["writer"]
        elif subtask.type == "fact_check":
            return self.agents["fact_checker"]
        # ...

优势：

  • 控制流清晰，易于调试
  • 可以精确控制执行顺序
  • 中心节点维护全局状态

劣势：

  • 中心节点是性能瓶颈
  • 所有通信经过中心，延迟高
  • 中心节点故障则全系统故障

2. 去中心化编排（P2P模式）

Agent之间直接通信，无中心节点：

class P2PAgent:
    def __init__(self, name, capabilities):
        self.name = name
        self.capabilities = capabilities
        self.peers = {}  # 已知的其他Agent
        self.message_queue = asyncio.Queue()
    
    async def run(self):
        """Agent主循环"""
        while True:
            message = await self.message_queue.get()
            
            if message.type == "task":
                # 处理任务
                if self._can_handle(message.task):
                    result = await self._handle(message.task)
                    await self._send(message.sender, "result", result)
                else:
                    # 转发给合适的peer
                    peer = self._find_capable_peer(message.task)
                    await self._send(peer, "task", message.task)
            
            elif message.type == "result":
                self._process_result(message)
    
    async def _send(self, peer_name, msg_type, content):
        """直接发送消息给peer"""
        peer = self.peers[peer_name]
        await peer.message_queue.put({
            "type": msg_type,
            "content": content,
            "sender": self.name
        })

优势：

  • 无单点故障
  • 并行执行效率高
  • 可扩展性好

劣势：

  • 协调复杂，容易混乱
  • 调试困难
  • 一致性保证难

3. 混合编排

结合中心化和去中心化的优点：

class HybridOrchestrator:
    def __init__(self):
        # 中心编排器管理高层流程
        self.workflow = WorkflowEngine()
        
        # Agent分组，组内P2P协作
        self.teams = {
            "research_team": Team([
                WebResearchAgent(),
                DocResearchAgent(),
                DataResearchAgent()
            ]),
            "writing_team": Team([
                OutlineAgent(),
                DraftAgent(),
                EditAgent()
            ])
        }
    
    def execute(self, task):
        # 1. 中心编排：确定高层流程
        plan = self.workflow.plan(task)
        
        # 2. 分阶段执行，每阶段委托给一个团队
        results = {}
        for phase in plan.phases:
            team = self.teams[phase.team]
            
            # 团队内部P2P协作完成阶段任务
            phase_result = team.collaborate(
                task=phase.task,
                context=results
            )
            results[phase.id] = phase_result
        
        return self._finalize(results)

Agent间通信协议

消息格式

class AgentMessage:
    def __init__(self, sender, receiver, msg_type, content):
        self.sender = sender        # 发送者ID
        self.receiver = receiver    # 接收者ID或"broadcast"
        self.type = msg_type        # 消息类型
        self.content = content      # 消息内容
        self.timestamp = datetime.now()
        self.correlation_id = None  # 关联ID（用于追踪对话链）
        self.priority = "normal"    # 优先级
    
    def to_dict(self):
        return {
            "sender": self.sender,
            "receiver": self.receiver,
            "type": self.type,
            "content": self.content,
            "timestamp": self.timestamp.isoformat(),
            "correlation_id": self.correlation_id,
            "priority": self.priority
        }

通信模式

class CommunicationPatterns:
    def request_response(self, sender, receiver, request):
        """请求-响应模式"""
        msg_id = generate_id()
        sender.send(receiver, "request", {
            "id": msg_id,
            "content": request
        })
        # 阻塞等待响应
        response = sender.wait_for_response(msg_id, timeout=30)
        return response
    
    def publish_subscribe(self, publisher, topic, message):
        """发布-订阅模式"""
        subscribers = self.topic_registry.get_subscribers(topic)
        for sub in subscribers:
            publisher.send(sub, "notification", {
                "topic": topic,
                "content": message
            })
    
    def blackboard(self, writer, key, value):
        """黑板模式：共享信息空间"""
        self.blackboard_store.set(key, value, writer=writer)
        # 通知关注该key的Agent
        for watcher in self.blackboard_store.get_watchers(key):
            self.notify(watcher, f"黑板更新: {key}")

任务分配策略

能力匹配

class CapabilityMatcher:
    def __init__(self):
        self.agent_capabilities = {}  # agent_id -> capability set
    
    def register(self, agent_id, capabilities):
        self.agent_capabilities[agent_id] = set(capabilities)
    
    def find_best_agent(self, task_requirements):
        """找到能力最匹配的Agent"""
        required = set(task_requirements)
        
        scores = []
        for agent_id, caps in self.agent_capabilities.items():
            # 覆盖率
            coverage = len(required & caps) / len(required)
            # 负载因子
            load = self._get_load(agent_id)
            # 综合评分
            score = coverage * 0.7 + (1 - load) * 0.3
            
            scores.append((agent_id, score, coverage))
        
        scores.sort(key=lambda x: x[1], reverse=True)
        
        if scores[0][2] < 0.5:  # 覆盖率不足50%
            return None  # 没有合适的Agent
        
        return scores[0][0]

负载均衡

class LoadBalancer:
    def __init__(self):
        self.agent_loads = {}  # agent_id -> current load
    
    def assign(self, task, candidates):
        """考虑负载的分配"""
        # 按负载排序
        sorted_agents = sorted(
            candidates,
            key=lambda a: self.agent_loads.get(a, 0)
        )
        
        # 选择负载最低的
        selected = sorted_agents[0]
        self.agent_loads[selected] = self.agent_loads.get(selected, 0) + 1
        
        return selected
    
    def complete(self, agent_id):
        """任务完成，减少负载"""
        self.agent_loads[agent_id] = max(0, self.agent_loads[agent_id] - 1)

冲突处理

结果冲突

当多个Agent给出不同结果时：

class ConflictResolver:
    def resolve(self, results):
        if len(results) == 1:
            return results[0]
        
        # 1. 投票
        vote_result = self._vote(results)
        if vote_result["confidence"] > 0.8:
            return vote_result["result"]
        
        # 2. 权威Agent裁决
        authority = max(results, key=lambda r: r.agent_authority)
        if authority.agent_authority > 0.9:
            return authority
        
        # 3. 升级到人工
        return self._escalate_to_human(results)
    
    def _vote(self, results):
        """多数投票"""
        votes = {}
        for r in results:
            key = hash(r.content)
            votes[key] = votes.get(key, 0) + r.agent_confidence
        
        best = max(votes.items(), key=lambda x: x[1])
        return {
            "result": best[0],
            "confidence": best[1] / sum(votes.values())
        }

实际案例：内容生产流水线

def build_content_pipeline():
    """构建多Agent内容生产流水线"""
    
    orchestrator = Orchestrator()
    
    # Agent团队
    orchestrator.register_agent("topic_analyst", TopicAnalysisAgent())
    orchestrator.register_agent("researcher", ResearchAgent())
    orchestrator.register_agent("outliner", OutlineAgent())
    orchestrator.register_agent("writer", WriterAgent())
    orchestrator.register_agent("fact_checker", FactCheckAgent())
    orchestrator.register_agent("editor", EditorAgent())
    orchestrator.register_agent("seo_optimizer", SEOAgent())
    
    # 工作流定义
    workflow = [
        {"step": "analyze_topic", "agent": "topic_analyst", "output": "topic_brief"},
        {"step": "research", "agent": "researcher", "input": "topic_brief", "output": "research_data"},
        {"step": "outline", "agent": "outliner", "input": ["topic_brief", "research_data"], "output": "outline"},
        {"step": "write_draft", "agent": "writer", "input": "outline", "output": "draft"},
        {"step": "fact_check", "agent": "fact_checker", "input": "draft", "output": "fact_report"},
        {"step": "edit", "agent": "editor", "input": ["draft", "fact_report"], "output": "final_draft"},
        {"step": "optimize", "agent": "seo_optimizer", "input": "final_draft", "output": "published_content"},
    ]
    
    orchestrator.set_workflow(workflow)
    return orchestrator

监控与可观测性

class AgentSystemMonitor:
    def __init__(self):
        self.metrics = {}
    
    def trace(self, agent_id, action):
        """追踪Agent行为"""
        trace_entry = {
            "agent_id": agent_id,
            "action": action,
            "timestamp": datetime.now(),
            "duration": None
        }
        return trace_entry
    
    def visualize(self):
        """可视化Agent协作流程"""
        return {
            "communication_graph": self._build_comm_graph(),
            "task_flow": self._build_task_flow(),
            "bottleneck_analysis": self._find_bottlenecks(),
            "agent_utilization": self._compute_utilization()
        }

结语

多Agent编排是AI系统从"助手"到"组织"的跃迁。就像人类组织中存在不同协作模式一样，AI Agent的编排也没有万能方案——中心化适合确定性流程，去中心化适合探索性任务，混合模式适合复杂业务。设计多Agent系统时，先明确任务的确定性程度、协作复杂度和容错要求，再选择合适的编排模式。记住：好的编排让1+1>2，差的编排让1+1<1。