1. 多Agent协作到底在解决什么问题1.1 从单兵作战到团队配合的必然演进我最早接触大模型应用开发的时候几乎所有需求都是围绕“一个模型搞定所有事”来设计的。你给它一段提示词它给你一个回答简单直接。但很快问题就来了当你需要模型完成一个涉及多个步骤、多个知识领域、需要反复校验的复杂任务时单模型的表现就开始拉胯。比如让它写一份完整的技术方案它可能在技术选型部分表现不错但到了成本估算环节就开始胡编乱造再到风险评估部分又完全忽略了关键依赖项。这不是模型能力不行而是单次推理的上下文窗口和注意力分配机制天然存在瓶颈。一个模型在一次对话中要同时扮演架构师、程序员、测试工程师、项目经理它的“注意力”被稀释了每个角色都只能做到及格线附近。多Agent协作的核心思路就是把一个大而全的任务拆解成多个小而专的角色让每个Agent在自己擅长的领域内独立工作再通过一套协作机制把结果拼装起来。这就像一个小型创业团队有人负责产品设计有人负责技术实现有人负责质量把关有人负责统筹协调。每个人不需要是全才但组合起来能完成远超个人能力的复杂工程。1.2 多Agent系统的典型应用场景从我这几年实际落地的项目来看多Agent协作在以下几类场景中价值最明显复杂代码工程一个Agent负责需求分析和架构设计一个负责编码实现一个负责代码审查和测试用例生成还有一个负责文档撰写。这种分工在大型项目重构或者从零搭建系统时效率提升非常显著。科研论文辅助研究定位、文献综述、数据分析、论文撰写、质量校准这几个环节对能力的要求差异很大用不同的Agent分别处理再通过调度器串联比让一个模型从头写到尾质量高出一大截。多步骤业务流程自动化比如电商场景下的商品上架流程涉及图片处理、文案生成、价格策略计算、库存同步等多个环节每个环节用专门的Agent处理通过任务调度器编排执行顺序和依赖关系。复杂决策支持需要从多个维度分析同一个问题时可以让不同Agent分别从技术可行性、成本效益、风险控制等角度给出评估最后由一个汇总Agent综合各方意见形成决策建议。这些场景的共同特征是任务可分解、子任务之间有依赖关系、每个子任务对专业能力的要求不同。如果你的任务本身很简单一句话就能说清楚那单模型完全够用上多Agent反而是过度设计。1.3 协作架构与任务调度的核心挑战多Agent系统听起来很美好但真正落地时会遇到几个硬骨头第一个是通信问题。Agent之间怎么传递信息用自然语言还是结构化数据传递过程中信息会不会失真我见过太多项目因为Agent之间“各说各话”导致最终结果驴唇不对马嘴。第二个是调度问题。哪个Agent先执行哪些可以并行某个Agent执行失败了怎么办需不需要重试重试几次这些调度逻辑如果设计不好整个系统要么卡死要么陷入无限循环。第三个是状态管理问题。多Agent协作过程中会产生大量的中间状态——某个Agent的输出、某个任务的执行进度、全局的上下文信息。这些状态怎么存储、怎么共享、怎么保证一致性都是需要仔细设计的。第四个是质量控制问题。多个Agent各自输出结果后怎么判断哪个结果可信怎么发现和纠正错误需不需要引入一个“裁判”角色来做质量把关这些问题在单Agent场景下根本不存在但在多Agent系统里每一个都可能成为项目失败的导火索。接下来我会逐一拆解这些问题的解决思路和实操方案。2. 协作架构设计的核心思路与方案选型2.1 三种主流协作架构的对比与选择在实际项目中多Agent协作架构大致可以归为三类每种都有各自的适用场景和取舍。第一种是中心化编排架构。有一个“协调者”Agent作为中枢所有任务的分发、结果的汇总、异常的处理都由它来负责。其他Agent只负责执行具体任务不关心全局状态。这种架构的好处是逻辑清晰、易于调试出问题了直接看协调者的日志就能定位。缺点是协调者容易成为性能瓶颈而且一旦协调者本身判断失误整个系统就会跑偏。第二种是去中心化对等架构。Agent之间直接通信没有统一的协调者。每个Agent根据自己的能力和当前状态决定是否接受某个任务。这种架构灵活性高、扩展性好但调试难度极大而且容易出现“三个和尚没水喝”的局面——谁都觉得该别人做最后任务没人处理。第三种是混合架构。顶层有一个轻量级的调度器负责宏观任务分配底层Agent之间可以点对点通信处理细节协调。这是我目前最推荐的方案兼顾了可控性和灵活性。架构类型适用场景优势劣势中心化编排任务流程固定、步骤明确逻辑清晰、易调试、状态可控协调者瓶颈、单点故障去中心化对等任务动态性强、Agent能力对等灵活、扩展性好调试困难、容易死锁混合架构大多数实际生产场景兼顾可控性与灵活性设计复杂度较高选型的时候我的经验是如果你的任务流程可以在纸上画出一个清晰的流程图那就用中心化编排如果任务边界模糊、需要Agent自主判断那就考虑混合架构纯去中心化的方案我建议只在实验性项目中使用生产环境慎用。2.2 Agent角色划分的原则与实操角色划分是多Agent系统设计的第一步也是最容易出问题的一步。我见过太多项目把角色划得太细结果Agent之间通信开销比实际执行时间还长也见过划得太粗跟单Agent没什么区别。我的经验是遵循**“高内聚、低耦合”**的原则高内聚一个Agent负责的任务应该在逻辑上紧密相关共享相似的上下文和知识背景。比如“代码生成”和“代码审查”虽然相关但前者需要创造性思维后者需要批判性思维放在一个Agent里容易互相干扰应该分开。低耦合Agent之间的依赖关系要尽可能少接口要尽可能清晰。如果Agent A的输出格式稍微变一下Agent B就完全没法工作那这个设计就有问题。具体操作上我通常按以下步骤来划分角色列出所有需要完成的任务节点把整个复杂任务拆解成原子级别的步骤每个步骤用一句话描述清楚输入和输出。按能力需求聚类把需要相似能力、相似知识背景的任务节点归为一组。为每组定义一个Agent角色明确这个Agent的职责边界、输入输出格式、可用的工具和知识库。检查角色间的依赖关系画出依赖图看看有没有循环依赖有没有可以合并的角色。注意角色数量不是越多越好。根据我的实测大多数场景下3到5个Agent是比较合理的范围。超过7个Agent后通信开销和协调复杂度会急剧上升整体效率反而下降。2.3 通信协议与数据格式设计Agent之间怎么“说话”直接决定了系统的稳定性和可维护性。我试过纯自然语言通信也试过完全结构化的JSON通信最后发现混合方案最实用。具体来说Agent之间的通信消息包含两部分结构化头部包含发送者、接收者、消息类型、任务ID、时间戳、优先级等元信息。这部分用严格的JSON Schema约束确保机器可解析。自然语言正文包含具体的任务描述、执行结果、问题反馈等内容。这部分允许自然语言表达但需要遵循一定的模板。{ header: { msg_id: task-001-step-03, sender: architect_agent, receiver: coder_agent, msg_type: task_assignment, priority: high, timestamp: 2025-01-15T10:30:00Z, parent_task: task-001 }, body: { task_description: 根据架构设计文档实现用户认证模块, input_artifacts: [architecture_doc_v2.md, api_spec.json], expected_output: 可运行的Python代码包含单元测试, constraints: [使用FastAPI框架, JWT认证, 密码加密存储], deadline: 2025-01-15T11:00:00Z } }这种设计的核心考虑是结构化头部让调度器可以自动路由和追踪自然语言正文让Agent可以灵活表达复杂信息。两者结合既保证了系统的可观测性又保留了足够的灵活性。2.4 状态管理与上下文传递策略多Agent协作中最容易踩的坑就是状态管理。我早期做过一个项目三个Agent协作写一份报告结果因为上下文传递没设计好第三个Agent拿到的信息是第一个Agent的原始输出完全忽略了第二个Agent的修改意见最后出来的东西前后矛盾。后来我总结了一套分层状态管理的方案全局状态整个任务的生命周期内共享的信息比如任务目标、总体约束、最终交付物要求。这部分存在一个中心化的状态存储中所有Agent都可以读取。会话状态单个Agent在执行某个子任务时的上下文包括它接收到的输入、中间推理过程、产生的输出。这部分只在该Agent内部维护不对外暴露。交接状态Agent之间传递任务时附带的信息包包含上游Agent的输出、对下游Agent的期望、需要注意的事项。这部分随着任务流转而传递。实操中我建议用事件溯源的方式来管理全局状态。每次Agent完成一个动作就往事件日志里追加一条记录而不是直接修改状态。这样任何时候都可以通过重放事件日志来恢复系统状态调试的时候特别有用。3. 任务调度机制的核心实现3.1 任务分解与依赖图构建任务调度第一步是把大任务拆成小任务并且理清它们之间的依赖关系。这个过程我通常分两步走第一步是粗粒度分解。把整个项目按阶段划分比如“需求分析→架构设计→编码实现→测试验证→文档撰写”。这一步主要靠人工经验因为不同领域的任务分解方式差异很大。第二步是细粒度分解。对每个阶段内的任务进一步拆分直到每个任务都可以由一个Agent在一次执行中完成。这一步可以借助大模型来自动完成但需要人工审核。分解完成后用有向无环图DAG来表示任务依赖关系。每个节点是一个任务每条边表示依赖关系。比如“编码实现”依赖于“架构设计”的输出“测试验证”依赖于“编码实现”的输出。# 任务依赖图的简化表示 task_graph { task_001: { name: 需求分析, agent: analyst_agent, dependencies: [], status: pending }, task_002: { name: 架构设计, agent: architect_agent, dependencies: [task_001], status: pending }, task_003: { name: 编码实现, agent: coder_agent, dependencies: [task_002], status: pending }, task_004: { name: 测试验证, agent: tester_agent, dependencies: [task_003], status: pending }, task_005: { name: 文档撰写, agent: writer_agent, dependencies: [task_002, task_003], status: pending } }构建依赖图的时候有几个关键点需要注意避免循环依赖A等B的输出B等A的输出这种死锁一旦出现整个系统就卡死了。构建完图之后一定要做一次拓扑排序检测。识别可并行任务没有依赖关系的任务可以并行执行这是提升效率的关键。比如“文档撰写”和“测试验证”如果都只依赖“编码实现”那它们就可以同时进行。粒度适中任务拆得太细调度开销大拆得太粗并行度不够。我的经验是每个任务的预期执行时间在30秒到5分钟之间比较合适。3.2 调度器的工作流程与核心逻辑调度器是整个多Agent系统的心脏。它的核心工作就是一个循环检查就绪任务→分配Agent→监控执行→处理结果→更新状态→重复。具体流程如下初始化加载任务依赖图将所有没有依赖的任务标记为“就绪”。调度循环从就绪队列中取出优先级最高的任务。检查该任务对应的Agent是否空闲。如果空闲分配任务如果忙碌将任务放回队列等待。Agent执行任务期间调度器持续监控执行状态。任务完成后更新任务状态为“已完成”并检查哪些下游任务的依赖已经全部满足将它们标记为“就绪”。如果任务失败根据重试策略决定是重试、跳过还是终止整个流程。终止条件所有任务完成或者遇到不可恢复的错误。import asyncio from collections import deque class TaskScheduler: def __init__(self, task_graph, agents): self.task_graph task_graph self.agents agents self.ready_queue deque() self.running_tasks {} self.completed_tasks set() self.failed_tasks set() def initialize(self): 初始化找出所有无依赖的任务 for task_id, task_info in self.task_graph.items(): if not task_info[dependencies]: self.ready_queue.append(task_id) task_info[status] ready def update_ready_tasks(self, completed_task_id): 当一个任务完成后更新下游任务的状态 for task_id, task_info in self.task_graph.items(): if task_info[status] ! pending: continue deps task_info[dependencies] if all(dep in self.completed_tasks for dep in deps): task_info[status] ready self.ready_queue.append(task_id) async def run(self): 调度主循环 self.initialize() while self.ready_queue or self.running_tasks: # 分配就绪任务 while self.ready_queue: task_id self.ready_queue.popleft() task_info self.task_graph[task_id] agent self.agents.get(task_info[agent]) if agent and agent.is_idle(): task_info[status] running future asyncio.create_task( agent.execute(task_info) ) self.running_tasks[task_id] future if not self.running_tasks: await asyncio.sleep(0.5) continue # 等待任意一个任务完成 done, _ await asyncio.wait( self.running_tasks.values(), return_whenasyncio.FIRST_COMPLETED ) for future in done: task_id self._find_task_by_future(future) try: result future.result() self.completed_tasks.add(task_id) self.task_graph[task_id][status] completed self.task_graph[task_id][result] result self.update_ready_tasks(task_id) except Exception as e: self.failed_tasks.add(task_id) self.task_graph[task_id][status] failed self.task_graph[task_id][error] str(e) # 根据重试策略处理 self._handle_failure(task_id) del self.running_tasks[task_id]这段代码展示了一个最简化的调度器实现。实际生产环境中还需要考虑更多细节比如优先级队列、超时处理、资源限制、优雅降级等。3.3 并行执行与资源竞争的处理多Agent系统的一个核心优势就是可以并行执行任务但并行也带来了资源竞争的问题。我遇到过最典型的情况是两个Agent同时调用同一个外部API结果触发了速率限制两个都失败了。处理资源竞争我通常采用以下几种策略第一种是资源池化。把所有可用的外部资源API配额、数据库连接、计算资源统一管理Agent需要使用时先申请用完归还。调度器根据资源可用情况决定是否分配任务。第二种是优先级抢占。高优先级的任务可以抢占低优先级任务占用的资源。这在紧急任务场景下很有用但需要小心处理被抢占任务的恢复。第三种是排队等待。当资源不足时任务进入等待队列调度器定期检查资源释放情况。这是最简单的方案但可能导致某些任务长时间得不到执行。class ResourcePool: def __init__(self, resources): self.resources resources self.locks {r: asyncio.Lock() for r in resources} self.usage {r: 0 for r in resources} async def acquire(self, resource_name, timeout30): 申请资源支持超时 lock self.locks.get(resource_name) if not lock: raise ValueError(f未知资源: {resource_name}) try: await asyncio.wait_for(lock.acquire(), timeouttimeout) self.usage[resource_name] 1 return True except asyncio.TimeoutError: return False def release(self, resource_name): 释放资源 lock self.locks.get(resource_name) if lock and lock.locked(): self.usage[resource_name] - 1 lock.release()实操心得资源竞争问题在开发环境往往不明显因为并发量小。但一到生产环境各种问题就会集中爆发。我的建议是在开发阶段就用压力测试工具模拟高并发场景提前暴露问题。3.4 失败重试与异常恢复机制多Agent系统里失败是常态而不是异常。网络抖动、API限流、模型输出格式错误、Agent判断失误这些都会导致任务失败。关键是怎么优雅地处理失败。我的重试策略通常分三个层次第一层是即时重试。对于网络超时、临时限流这类瞬时故障立即重试往往就能成功。但要注意设置最大重试次数和退避策略避免雪崩。第二层是降级重试。如果即时重试多次都失败就换一种方式执行。比如原来用高精度模型现在换成快速模型原来调用外部API现在用本地缓存数据。第三层是人工介入。如果降级重试也失败就把任务标记为“需要人工处理”暂停整个流程通知相关人员介入。async def execute_with_retry(agent, task, max_retries3, backoff_factor2): 带重试的任务执行 last_error None for attempt in range(max_retries): try: result await agent.execute(task) return result except TransientError as e: # 瞬时错误可以重试 last_error e wait_time backoff_factor ** attempt await asyncio.sleep(wait_time) except PermanentError as e: # 永久错误重试没有意义 raise e except Exception as e: # 未知错误记录后重试 last_error e await asyncio.sleep(1) raise MaxRetriesExceeded( f任务 {task[id]} 重试 {max_retries} 次后仍然失败: {last_error} )异常恢复还有一个重要方面是状态回滚。如果一个任务执行到一半失败了它可能已经产生了一些副作用比如修改了数据库、发送了消息。恢复的时候需要把这些副作用清理掉或者设计成幂等操作重复执行不会产生额外影响。4. 完整实操从零搭建一个多Agent协同系统4.1 环境准备与基础框架搭建这一节我带你从零开始搭建一个可运行的多Agent协同系统。为了让大家都能跟着做我选择Python作为开发语言用最少的依赖来实现核心功能。首先安装必要的依赖pip install asyncio aiohttp pydantic loguru如果你打算接入实际的大模型API还需要安装对应的SDK。这里我用一个抽象的LLM接口来演示你可以根据自己使用的模型平台替换具体实现。项目目录结构建议这样组织multi_agent_system/ ├── core/ │ ├── agent.py # Agent基类 │ ├── scheduler.py # 调度器 │ ├── message.py # 消息定义 │ └── state.py # 状态管理 ├── agents/ │ ├── analyst.py # 分析Agent │ ├── architect.py # 架构Agent │ ├── coder.py # 编码Agent │ └── reviewer.py # 审查Agent ├── config/ │ └── settings.py # 配置文件 └── main.py # 入口先定义消息格式和Agent基类from pydantic import BaseModel, Field from typing import Optional, Any from datetime import datetime import uuid class MessageHeader(BaseModel): msg_id: str Field(default_factorylambda: str(uuid.uuid4())) sender: str receiver: str msg_type: str # task_assignment, task_result, error, query priority: str normal timestamp: str Field(default_factorylambda: datetime.now().isoformat()) parent_task: Optional[str] None class MessageBody(BaseModel): content: str artifacts: list[str] [] metadata: dict[str, Any] {} class Message(BaseModel): header: MessageHeader body: MessageBodyAgent基类的设计要考虑几个关键点每个Agent有唯一的名称和角色描述、有独立的执行方法、能够接收和发送消息、维护自己的内部状态。from abc import ABC, abstractmethod class BaseAgent(ABC): def __init__(self, name: str, role: str, llm_client): self.name name self.role role self.llm llm_client self.state idle self.inbox [] self.memory [] # 短期记忆 def is_idle(self): return self.state idle def receive_message(self, message: Message): self.inbox.append(message) abstractmethod async def execute(self, task: dict) - dict: 执行任务子类必须实现 pass def build_prompt(self, task: dict) - str: 构建提示词子类可以覆盖 return f 你是{self.role}。 当前任务{task.get(description, )} 输入材料{task.get(input_artifacts, [])} 约束条件{task.get(constraints, [])} 请完成你的任务并输出结果。 4.2 定义具体Agent角色与提示词工程有了基类之后我们来定义几个具体的Agent。每个Agent的核心差异在于系统提示词和输出格式要求。分析Agent负责需求拆解和任务规划class AnalystAgent(BaseAgent): def __init__(self, llm_client): super().__init__( nameanalyst, role需求分析师擅长将复杂需求拆解为可执行的任务列表, llm_clientllm_client ) async def execute(self, task: dict) - dict: self.state working prompt f 你是一名资深需求分析师。请对以下需求进行深度分析 原始需求{task[description]} 请输出 1. 核心目标一句话概括 2. 关键约束条件列出所有限制 3. 任务拆解将需求拆分为3-7个子任务每个子任务包含名称、描述、预期输出、依赖关系 4. 风险点识别可能出问题的地方 输出格式要求使用JSON格式包含 goals, constraints, subtasks, risks 四个字段。 response await self.llm.generate(prompt) result self._parse_response(response) self.memory.append({task: task[id], result: result}) self.state idle return result架构Agent负责技术方案设计class ArchitectAgent(BaseAgent): def __init__(self, llm_client): super().__init__( namearchitect, role系统架构师擅长技术选型和系统设计, llm_clientllm_client ) async def execute(self, task: dict) - dict: self.state working prompt f 你是一名资深系统架构师。基于以下需求分析结果设计技术方案 需求分析{task.get(input_artifacts, [])} 请输出 1. 技术栈选型每项选择都要说明理由 2. 系统架构图描述用文字描述模块划分和交互关系 3. 关键接口定义输入输出格式 4. 数据模型设计 5. 部署方案建议 输出格式Markdown格式的技术设计文档。 response await self.llm.generate(prompt) self.memory.append({task: task[id], result: response}) self.state idle return {design_doc: response, format: markdown}编码Agent和审查Agent的逻辑类似这里不再赘述。关键是要让每个Agent的提示词聚焦在自己的专业领域不要试图让一个Agent解决所有问题。提示提示词工程在多Agent系统里比单Agent更重要。因为Agent之间会互相传递信息一个Agent的输出格式如果不稳定下游Agent就会遭殃。我的经验是在提示词里明确指定输出格式并且要求Agent在输出中包含“置信度”字段方便下游判断是否可以直接使用。4.3 调度器与Agent的集成运行把Agent和调度器组装起来形成一个完整的运行系统class MultiAgentSystem: def __init__(self, llm_client): self.llm llm_client self.agents {} self.scheduler None self.event_log [] def register_agent(self, agent: BaseAgent): self.agents[agent.name] agent def log_event(self, event_type: str, data: dict): self.event_log.append({ type: event_type, timestamp: datetime.now().isoformat(), data: data }) async def run_task(self, task_description: str): # 第一步分析需求 analyst self.agents[analyst] analysis_result await analyst.execute({ id: task-001, description: task_description }) self.log_event(analysis_complete, analysis_result) # 第二步根据分析结果构建任务图 task_graph self._build_task_graph(analysis_result) # 第三步启动调度器 self.scheduler TaskScheduler(task_graph, self.agents) await self.scheduler.run() # 第四步汇总结果 final_result self._aggregate_results() return final_result def _build_task_graph(self, analysis_result: dict) - dict: 将分析结果转换为任务依赖图 graph {} subtasks analysis_result.get(subtasks, []) for i, subtask in enumerate(subtasks): task_id ftask-{i2:03d} graph[task_id] { name: subtask[name], description: subtask[description], agent: self._match_agent(subtask), dependencies: subtask.get(dependencies, []), status: pending, expected_output: subtask.get(expected_output, ) } return graph def _match_agent(self, subtask: dict) - str: 根据子任务描述匹配最合适的Agent # 简化版基于关键词匹配 description subtask[description].lower() if any(kw in description for kw in [设计, 架构, 选型]): return architect elif any(kw in description for kw in [编码, 实现, 开发]): return coder elif any(kw in description for kw in [测试, 审查, 校验]): return reviewer else: return analyst运行入口async def main(): # 初始化LLM客户端这里用伪代码表示 llm_client YourLLMClient(api_keyyour-key) # 创建系统 system MultiAgentSystem(llm_client) # 注册Agent system.register_agent(AnalystAgent(llm_client)) system.register_agent(ArchitectAgent(llm_client)) system.register_agent(CoderAgent(llm_client)) system.register_agent(ReviewerAgent(llm_client)) # 执行任务 result await system.run_task( 设计并实现一个用户登录注册系统包含前端页面和后端API ) print(最终结果, result) if __name__ __main__: asyncio.run(main())4.4 运行效果验证与调优系统跑起来之后你需要验证它是否真的在工作。我通常从以下几个维度来评估第一个维度是任务完成率。所有子任务中成功完成的比例是多少如果低于80%说明要么任务分解有问题要么Agent能力不够。第二个维度是结果质量。最终输出是否满足原始需求有没有明显的错误或遗漏这个需要人工评估但可以设计一些自动化检查规则来辅助。第三个维度是执行效率。整个流程花了多长时间哪些环节是瓶颈并行执行的比例是多少第四个维度是资源消耗。总共调用了多少次大模型消耗了多少token成本是否在可接受范围内根据评估结果进行调优常见的调优方向包括调整任务粒度如果某个Agent总是超时可能是任务太大需要进一步拆分。优化提示词如果某个Agent的输出格式总是不对需要在提示词里更明确地约束。增加缓存如果某些任务重复执行可以缓存结果避免重复计算。调整调度策略如果并行度不够检查依赖关系是否设置得太严格。5. 常见问题与排查技巧实录5.1 Agent之间信息传递失真的排查这是多Agent系统最高频的问题。表现是上游Agent明明输出了正确的结果但下游Agent理解错了导致最终结果偏离预期。排查思路分三步第一步检查消息内容。把Agent之间传递的原始消息打印出来看看上游到底发了什么。我遇到过很多次是因为上游Agent的输出格式不稳定有时候是JSON有时候是Markdown下游Agent解析失败。第二步检查提示词。下游Agent的提示词里有没有明确说明“你会收到什么格式的输入”如果没有它就只能靠猜。第三步检查上下文长度。如果上游Agent的输出特别长下游Agent的上下文窗口可能装不下导致信息被截断。解决方案我通常用结构化输出格式校验的组合拳。要求上游Agent必须输出符合特定Schema的JSON下游Agent在接收时先做格式校验校验不通过就触发重试。from pydantic import BaseModel, ValidationError class TaskOutput(BaseModel): summary: str details: str artifacts: list[str] confidence: float next_steps: list[str] def validate_agent_output(raw_output: str) - TaskOutput: 校验Agent输出格式 try: # 尝试提取JSON import json data json.loads(raw_output) return TaskOutput(**data) except (json.JSONDecodeError, ValidationError) as e: raise FormatError(fAgent输出格式错误: {e})5.2 任务死锁与循环依赖的处理死锁的表现是整个系统卡住不动所有Agent都在等待但没有任务在执行。根本原因通常是循环依赖A等B的输出B等C的输出C等A的输出。或者资源死锁Agent A占用了资源X等待资源YAgent B占用了资源Y等待资源X。排查方法检查任务依赖图是否有环。用拓扑排序算法检测如果有环排序会失败。检查资源申请顺序。如果所有Agent都按相同的顺序申请资源就不会死锁。加超时机制。任何等待操作都要设置超时时间超时后释放资源并报错。def detect_cycle(task_graph: dict) - list[str]: 检测任务图中的循环依赖 visited set() rec_stack set() cycle_path [] def dfs(node, path): visited.add(node) rec_stack.add(node) path.append(node) for dep in task_graph[node].get(dependencies, []): if dep not in visited: if dfs(dep, path): return True elif dep in rec_stack: # 找到环 cycle_start path.index(dep) cycle_path.extend(path[cycle_start:]) return True path.pop() rec_stack.remove(node) return False for node in task_graph: if node not in visited: if dfs(node, []): return cycle_path return []5.3 模型输出不稳定的应对策略大模型的输出天然具有随机性同样的输入可能得到不同的输出。这在多Agent系统里会被放大因为一个Agent的输出不稳定会级联影响到下游所有Agent。我的应对策略包括设置温度参数对于需要稳定输出的Agent比如代码生成把temperature设低0.1-0.3对于需要创造性的Agent比如方案设计可以适当调高0.7-0.9。多次采样投票对于关键决策让同一个Agent执行3次取多数一致的结果。输出格式强约束在提示词里明确要求输出JSON并给出Schema。大多数模型在强约束下输出会稳定很多。后处理校验对Agent的输出做自动化校验不符合要求的直接打回重做。问题类型表现排查方法解决方案信息失真下游理解偏离上游意图打印原始消息对比结构化输出格式校验任务死锁系统卡住无进展检测依赖图环路拓扑排序超时机制输出不稳定同样输入不同输出多次执行对比降低温度格式约束资源竞争并发任务互相干扰监控资源使用资源池化排队机制性能瓶颈整体执行时间过长分析各环节耗时增加并行度缓存5.4 性能瓶颈定位与优化经验多Agent系统跑得慢原因可能有很多。我通常用分段计时的方法来定位瓶颈。在每个Agent执行前后打时间戳记录每个任务的耗时。然后分析是模型推理慢还是Agent之间通信慢是某个特定Agent慢还是所有Agent都慢是串行执行导致的慢还是并行度不够根据我的经验最常见的性能瓶颈是模型推理时间占总耗时的70%以上。优化方向包括使用更快的模型处理简单任务复杂任务才用大模型。对重复性任务的结果进行缓存。把可以并行的任务真正并行起来。减少不必要的Agent间通信轮次。还有一个容易被忽视的优化点是提示词长度。提示词越长模型推理越慢。我见过一个项目每个Agent的提示词都塞了几千字的背景信息结果光推理就花了十几秒。后来把背景信息精简到几百字速度提升了三倍多。5.5 多Agent系统的安全边界与风险控制多Agent系统因为涉及多个自主决策的实体风险控制比单Agent更重要。我通常从以下几个层面来设置安全边界输入层面对用户输入做过滤和校验防止恶意输入导致Agent执行危险操作。决策层面给Agent设置明确的权限边界比如编码Agent只能写代码不能执行代码审查Agent只能读不能改。输出层面对Agent的输出做审核特别是涉及敏感操作删除文件、发送请求、修改配置时必须经过人工确认。监控层面记录所有Agent的操作日志定期审计异常行为。注意多Agent系统里一个Agent的失误可能被其他Agent放大。比如分析Agent错误地把某个任务标记为“低优先级”调度器就会把它排到最后导致整个项目延期。所以关键决策节点一定要有校验机制不能完全信任单个Agent的判断。5.6 从实验到生产的环境适配要点实验室里跑通的多Agent系统直接搬到生产环境往往会出问题。我总结了几条环境适配的经验第一网络环境不同。实验室通常是内网延迟低、带宽大。生产环境可能跨区域调用延迟高、不稳定。需要增加重试和超时机制。第二并发量不同。实验室可能同时跑一两个任务生产环境可能几十上百个任务并发。需要做压力测试确保调度器和资源池能扛住。第三数据规模不同。实验室用的小数据集生产环境可能是海量数据。需要考虑分页、流式处理、增量更新等策略。第四容错要求不同。实验室挂了重启就行生产环境挂了可能造成业务损失。需要设计优雅降级方案比如某个Agent不可用时自动切换到备用方案。第五监控要求不同。实验室看看日志就行生产环境需要完整的监控告警体系。关键指标包括任务成功率、平均执行时间、资源利用率、错误率等。我在实际项目中的做法是先在预发布环境用生产级别的数据量和并发量跑一轮观察系统表现根据结果调整配置然后再正式上线。上线后前一周保持高频监控发现问题及时处理。这套流程虽然麻烦但能避免很多生产事故。
