多Agent协作项目的工程复盘：任务分配不均与通信风暴的解决方案

星辰AI

537人浏览 · 2026-07-24 14:01:14

星辰AI · 2026-07-24 14:01:14 发布

多Agent协作项目的工程复盘：任务分配不均与通信风暴的解决方案

一、从单Agent到多Agent的"复杂度跳跃"

单Agent很好理解——一个LLM调工具，循环直到任务完成。但当把"写一份市场分析报告"拆成"搜索Agent + 分析Agent + 写作Agent"三个协作时，问题指数级增长：

搜索Agent找到了300条相关信息，全部传给分析Agent——token爆炸+信息过载
分析Agent的处理时间太长（3分钟），写作Agent一直在空转等待
三个Agent之间的通信消息在5轮对话后累积到87条——通信开销超过了任务执行本身

单Agent到多Agent不是加法，是乘法——任务分配、负载均衡、通信协议、超时处理，每个环节都可能成为瓶颈。

二、三个核心问题的工程解法

问题一：任务分配不均导致的"木桶效应"

搜索Agent 30秒完成，分析Agent需要3分钟——整个流水线的速度由最慢的Agent决定。

解法：动态任务拆分。不是把"所有搜索结果→一个分析Agent"，而是：

class TaskOrchestrator:
    async def execute(self, task: Task) -> Result:
        # 1. 拆分任务
        subtasks = await self.decompose(task)
        
        # 2. 估算每个子任务的复杂度
        estimated_times = [self.estimate_time(st) for st in subtasks]
        
        # 3. 动态分配——计算密集型任务分给多个Agent并行
        assignments = self.balanced_assign(subtasks, self.available_agents)
        
        # 4. 并行执行 + 超时控制
        results = await asyncio.gather(
            *[agent.execute(assign) for agent, assign in assignments.items()],
            return_exceptions=True
        )
        
        return self.merge_results(results, task)
    
    def balanced_assign(self, subtasks, agents):
        """将总工作量均分给所有可用Agent"""
        # 按估计时间排序，大任务优先分配
        sorted_tasks = sorted(subtasks, key=lambda t: self.estimate_time(t), reverse=True)
        
        assignments = {agent: [] for agent in agents}
        loads = {agent: 0 for agent in agents}
        
        for task in sorted_tasks:
            # 分配给当前负载最小的Agent
            best_agent = min(loads, key=loads.get)
            assignments[best_agent].append(task)
            loads[best_agent] += self.estimate_time(task)
        
        return assignments

问题二：Agent间的通信风暴

3个Agent在5轮对话中产生了87条消息。大多数消息是冗余的中间状态更新。Agent A说"我在搜索"，Agent B问"找到了多少"，Agent A回"120条，正在筛选"——这些对最终任务没有价值。

解法：分层通信——区分"控制消息"和"数据消息"。

# 通信协议精简
class AgentMessage:
    type: str  # "data" | "status" | "error" | "done"
    priority: int  # 0=低(调试), 1=正常, 2=高(阻塞)
    payload: dict

class CommunicationManager:
    def __init__(self):
        self.data_channel = asyncio.Queue()    # 数据通道——仅结果
        self.control_channel = asyncio.Queue() # 控制通道——状态/错误
    
    async def send_data(self, agent_id: str, data: dict):
        # 数据仅发给需要它的Agent，不全网广播
        target = self.get_downstream_agent(agent_id)
        await self.data_channel.put({
            "target": target,
            "payload": data,
        })
    
    async def send_control(self, message: AgentMessage):
        # 仅阻塞性状态变更才广播
        if message.priority >= 2:
            await self.control_channel.put(message)

效果：消息量从87条降到19条（78%减少），通信延迟从总时间的35%降到8%。

问题三：Agent失败时的级联恢复

如果分析Agent中途失败（API超时/LLM返回格式错误），下游的写作Agent会收到空数据。原来的方案是整个任务重来——浪费时间。

解法：检查点机制——每个Agent完成后保存中间结果。失败时从最近检查点恢复，而非从头开始。

class CheckpointManager:
    async def save(self, stage: str, data: dict):
        key = f"checkpoint:{self.task_id}:{stage}"
        await redis.set(key, json.dumps(data), ex=3600)
    
    async def load(self, stage: str) -> dict | None:
        key = f"checkpoint:{self.task_id}:{stage}"
        data = await redis.get(key)
        return json.loads(data) if data else None

class ResilientOrchestrator:
    async def execute_with_retry(self, task: Task):
        # 尝试从最近检查点恢复
        last_stage = await self.checkpoint.detect_last_stage(task.id)
        
        if last_stage:
            saved_data = await self.checkpoint.load(last_stage)
            # 从失败点继续，而非从头
            return await self.continue_from(last_stage, saved_data, task)
        
        return await self.execute_from_scratch(task)