1. 从单体 Agent 到 Multi-Agent 的必然演进1.1 单体 Agent 到底能扛多少事先说结论单体 Agent 不是不能用而是它的能力边界比大多数人想象的要窄得多。我过去一年里用 Python 写过不下二十个 Agent 项目从最简单的 ReAct 循环到带工具调用的复杂编排踩过的坑基本都指向同一个方向——当任务复杂度超过某个阈值单体 Agent 的表现会断崖式下跌而不是线性退化。什么叫单体 Agent就是一个 LLM 实例配一套系统提示词挂几个工具跑一个 ReAct 或者 Plan-and-Execute 的循环。它自己决定调什么工具、什么时候停、怎么组织答案。这种架构在任务步骤少于五步、上下文窗口占用低于 50% 的时候表现相当不错。比如“帮我查一下今天北京的天气然后推荐穿什么衣服”这种任务单体 Agent 处理得又快又好。但问题在于真实业务里的任务很少这么干净。你让它做一个竞品分析它需要先搜索、再筛选、再对比、再总结、再生成报告。每一步的输出都会塞进上下文到了第四五步的时候上下文已经塞满了前面所有中间结果模型开始“忘事”、开始重复、开始胡编。这不是模型不行这是架构的天花板。1.2 三个硬约束把单体 Agent 逼到墙角第一个约束是上下文窗口的物理限制。不管你用的是 128K 还是 1M token 的模型上下文都不是无限大的。而且更关键的是上下文越长模型的注意力越分散中间信息被“淹没”的概率越高。我实测过一个 200K 上下文的任务到第 15 轮工具调用的时候模型已经记不住第 3 轮查到的关键数据了。这不是模型的问题是 Transformer 架构本身的注意力衰减特性决定的。第二个约束是工具集的组合爆炸。单体 Agent 如果挂 20 个工具它在每一步选择工具时的决策空间就是 20 选 1再加上参数组合实际决策空间大得离谱。我试过给一个单体 Agent 挂 15 个工具结果它在简单任务上开始乱调工具明明只需要查数据库它非要去调邮件发送接口。工具越多单体 Agent 的决策质量越差这是实测结论。第三个约束是错误累积与不可恢复性。单体 Agent 跑一个长链条任务中间任何一步出错后面全崩。而且它没有“回滚”机制也没有“换个人来试试”的机制。它只能硬着头皮往下走越走越偏。我见过最离谱的一次一个单体 Agent 在第三步把日期搞错了后面七步全部基于错误日期在跑最后输出了一份完全错误的报告但它自己“觉得”没问题。这三个约束叠加在一起就形成了一个清晰的天花板单体 Agent 适合短链条、少工具、低容错要求的任务一旦任务变复杂它必然走向 Multi-Agent。1.3 Multi-Agent 不是银弹但它是目前最合理的解法Multi-Agent 的核心思路很简单把一个大任务拆成若干子任务每个子任务交给一个专门的 Agent 去处理Agent 之间通过消息传递来协作。这样做的好处是每个 Agent 的上下文窗口只需要装自己那部分信息工具集也可以按需裁剪错误可以被隔离在单个 Agent 内部不会污染全局。但我要泼一盆冷水Multi-Agent 不是没有代价的。它引入了通信开销、协调复杂度、状态一致性问题。我见过太多团队一上来就搞 Multi-Agent结果发现比单体还难调。所以关键不是“要不要 Multi-Agent”而是“什么时候该切、怎么切、切完之后怎么管”。下面这张表是我在实际项目中总结的切换判断依据你可以直接拿去用判断维度单体 Agent 适用Multi-Agent 适用任务步骤数少于 5 步超过 8 步工具数量少于 5 个超过 8 个上下文占用低于 50% 窗口持续超过 70% 窗口错误容忍度低错了重跑就行高错了不能全崩子任务独立性强耦合可解耦并行需求无有这张表不是绝对的但如果你发现自己的单体 Agent 同时踩中了右边三列以上那就该考虑拆了。2. 拆解 Multi-Agent 的核心架构模式2.1 三种主流编排模式及其适用场景Multi-Agent 的架构模式目前主流的有三种流水线模式、主管模式、去中心化模式。这三种不是互斥的实际项目里经常混用但你需要知道每种模式的核心逻辑和适用边界。流水线模式最简单就是把任务拆成顺序执行的阶段每个阶段一个 Agent前一个的输出是后一个的输入。比如“搜索 Agent → 筛选 Agent → 分析 Agent → 报告 Agent”。这种模式的好处是逻辑清晰、调试容易每个 Agent 的职责非常明确。坏处是它是串行的没有并行能力而且如果中间某个 Agent 输出质量差后面全受影响。我一般用这种模式处理“步骤固定、顺序明确”的任务比如数据清洗管道。主管模式是我用得最多的。它有一个“主管 Agent”负责拆解任务、分配子任务、汇总结果下面挂若干个“工人 Agent”各自处理具体子任务。主管 Agent 不干具体活它只做调度和决策。这种模式的好处是灵活主管可以根据任务情况动态调整分配策略。坏处是主管 Agent 本身可能成为瓶颈如果主管的上下文塞满了所有子任务的返回结果它也会崩。我的经验是主管 Agent 的上下文里只保留子任务的“摘要”而不是“全文”全文存在外部存储里需要的时候再取。去中心化模式最复杂Agent 之间可以互相通信、协商、投票。这种模式理论上最灵活但实际工程里极难调试。我试过一次用去中心化模式做多源信息聚合结果两个 Agent 陷入了无限互相询问的循环烧了几百万 token 才被我手动掐断。除非你有非常明确的协商协议和终止条件否则不建议一上来就去中心化。2.2 Agent 之间的通信协议怎么设计Multi-Agent 最容易出问题的地方就是通信。我见过太多项目Agent 之间传的消息格式不统一导致解析失败、信息丢失、重复处理。通信协议的设计要抓住三个要点结构化、可追溯、有边界。结构化是指消息必须是固定格式的不能是自由文本。我一般用 JSON 作为消息载体包含这几个字段sender、receiver、task_id、status、payload、timestamp。payload里再根据具体任务定义子结构。这样做的好处是每个 Agent 收到消息后可以程序化解析不需要用 LLM 去“理解”消息格式。可追溯是指每条消息都要有唯一的task_id和parent_task_id这样才能追踪一个任务从拆解到完成的完整链路。我吃过这个亏有一次一个子任务失败了但我找不到它是从哪个父任务拆出来的排查了整整一个下午。后来我强制要求所有消息必须带parent_task_id排查效率直接翻倍。有边界是指消息不能无限大。我设定了一个硬规则单条消息的payload不超过 2000 token超过的部分必须存到外部存储比如 Redis 或者本地文件消息里只放引用 ID。这个规则救了我很多次因为 Agent 之间传大段文本是上下文爆炸的主要元凶之一。# 一个典型的 Agent 消息结构 message { sender: search_agent, receiver: analysis_agent, task_id: task_20260415_001, parent_task_id: task_20260415_root, status: completed, payload: { summary: 找到 5 篇相关文章核心观点已提取, data_ref: redis://task_20260415_001_result, confidence: 0.85 }, timestamp: 2026-04-15T10:30:00Z }2.3 状态管理与共享内存的取舍Multi-Agent 系统里状态管理是个绕不开的问题。每个 Agent 都有自己的局部状态但有些状态需要共享比如全局的任务进度、已经确认的事实、用户的偏好设置。怎么管这些共享状态直接决定了系统的稳定性和可扩展性。我试过三种方案共享内存、消息传递、外部存储。共享内存最快但并发写的时候容易出竞态条件而且 Agent 多了之后内存占用飙升。消息传递最安全但延迟高而且状态同步逻辑复杂。外部存储比如 Redis是我目前最推荐的它兼顾了速度和安全性而且天然支持多 Agent 并发访问。具体做法是每个 Agent 在需要读取共享状态时从 Redis 里拉取在需要更新共享状态时用 Redis 的原子操作比如HSET、LPUSH写入。关键是要给每个状态字段设计好版本号或者时间戳避免旧数据覆盖新数据。我一般用updated_at字段来做乐观锁写入前先检查版本版本不对就重试。注意共享状态里不要放太大的数据。我见过有人把整个文档塞进 Redis 的共享状态里结果每次读取都要序列化反序列化性能直接崩了。大文件放对象存储Redis 里只放引用。2.4 错误隔离与重试机制的设计Multi-Agent 相比单体 Agent 最大的优势之一就是错误隔离。单体 Agent 一步错步步错Multi-Agent 可以把错误限制在单个 Agent 内部不影响其他 Agent。但前提是你得设计好错误隔离和重试机制。我的做法是给每个 Agent 配一个“健康检查”和“重试策略”。健康检查是指 Agent 在返回结果前自己先做一次质量校验比如检查输出是否为空、是否包含关键字段、是否满足格式要求。如果校验不通过Agent 自己先重试一次。重试策略是指如果 Agent 连续失败 N 次主管 Agent 会收到一个“失败信号”然后决定是换一个 Agent 重试、还是跳过这个子任务、还是终止整个流程。这里有个坑重试不能无限重试。我设定了一个硬上限单个子任务最多重试 3 次超过 3 次就上报给主管。主管根据子任务的重要性决定下一步。如果是关键子任务整个流程暂停等待人工介入如果是非关键子任务标记为“降级完成”继续往下走。# 一个简单的重试装饰器 def retry_on_failure(max_retries3, delay1): def decorator(func): def wrapper(*args, **kwargs): for attempt in range(max_retries): try: result func(*args, **kwargs) if validate_result(result): return result else: print(f结果校验失败第 {attempt 1} 次重试) except Exception as e: print(f执行异常{e}第 {attempt 1} 次重试) time.sleep(delay * (attempt 1)) raise MaxRetriesExceeded(f重试 {max_retries} 次后仍然失败) return wrapper return decorator这个装饰器我用了很久简单但有效。关键是validate_result这个函数要根据具体任务来写不能通用。比如搜索 Agent 的校验是“结果不为空且包含至少一个 URL”分析 Agent 的校验是“输出包含结论段落且字数超过 100”。3. 用 Python 从零搭建一个 Multi-Agent 系统3.1 环境准备与依赖选型动手之前先把环境理清楚。Python 版本我建议 3.10 以上因为要用到一些新的类型注解语法。核心依赖就几个openai或者anthropic的 SDK 用来调模型redis用来做共享状态pydantic用来做消息校验asyncio用来做并发。其他的按需加。我不建议一上来就用 LangChain 或者 AutoGen 这种重框架。原因很简单这些框架封装太厚出了问题你根本不知道是框架的锅还是你的锅。我自己的做法是先用原生 Python 把核心逻辑跑通理解每个环节在干什么然后再考虑要不要用框架来提效。如果你时间紧用框架没问题但一定要读一遍框架的源码至少知道它的 Agent 循环是怎么实现的。# 创建虚拟环境 python -m venv multi_agent_env source multi_agent_env/bin/activate # Linux/Mac # multi_agent_env\Scripts\activate # Windows # 安装核心依赖 pip install openai redis pydantic asyncio提示如果你在国内pip 安装可能会慢可以换源。但我不建议用任何来路不明的镜像直接用官方源加--timeout参数就行。3.2 定义 Agent 基类与消息总线先定义一个 Agent 基类把所有 Agent 共有的逻辑抽出来接收消息、处理消息、发送消息、状态管理。然后定义一个消息总线负责 Agent 之间的消息路由。import json import time import uuid from abc import ABC, abstractmethod from typing import Any, Dict, Optional from pydantic import BaseModel, Field class AgentMessage(BaseModel): sender: str receiver: str task_id: str Field(default_factorylambda: str(uuid.uuid4())) parent_task_id: Optional[str] None status: str pending payload: Dict[str, Any] {} timestamp: float Field(default_factorytime.time) class BaseAgent(ABC): def __init__(self, name: str, bus: MessageBus): self.name name self.bus bus self.context: list [] self.max_context_tokens 8000 abstractmethod async def process(self, message: AgentMessage) - AgentMessage: pass async def receive(self, message: AgentMessage): response await self.process(message) await self.bus.publish(response) def add_to_context(self, content: str): self.context.append(content) self._trim_context() def _trim_context(self): # 简单的上下文裁剪保留最近 N 条 if len(self.context) 20: self.context self.context[-20:] class MessageBus: def __init__(self): self.agents: Dict[str, BaseAgent] {} self.history: list [] def register(self, agent: BaseAgent): self.agents[agent.name] agent async def publish(self, message: AgentMessage): self.history.append(message) receiver self.agents.get(message.receiver) if receiver: await receiver.receive(message) else: print(f未找到接收者{message.receiver})这个基类很简陋但够用。关键点是_trim_context方法它决定了 Agent 的上下文怎么裁剪。我用的策略是“保留最近 20 条”但实际项目里你可能需要更复杂的策略比如“保留最近 10 条 所有关键结论”。3.3 实现一个搜索 Agent 和一个分析 Agent有了基类实现具体 Agent 就简单了。搜索 Agent 负责调搜索工具分析 Agent 负责对搜索结果做分析。class SearchAgent(BaseAgent): async def process(self, message: AgentMessage) - AgentMessage: query message.payload.get(query, ) # 这里模拟搜索实际项目里换成真实的搜索 API results await self._search(query) self.add_to_context(f搜索{query}结果{results[:200]}) return AgentMessage( senderself.name, receivermessage.payload.get(next, analysis_agent), parent_task_idmessage.task_id, statuscompleted, payload{ summary: f找到 {len(results)} 条结果, results: results, query: query } ) async def _search(self, query: str) - list: # 模拟搜索延迟 await asyncio.sleep(0.5) return [ {title: f关于 {query} 的文章 1, url: https://example.com/1}, {title: f关于 {query} 的文章 2, url: https://example.com/2}, ] class AnalysisAgent(BaseAgent): async def process(self, message: AgentMessage) - AgentMessage: results message.payload.get(results, []) query message.payload.get(query, ) # 这里调 LLM 做分析 analysis await self._analyze(query, results) self.add_to_context(f分析{query}结论{analysis[:200]}) return AgentMessage( senderself.name, receivermessage.payload.get(next, report_agent), parent_task_idmessage.task_id, statuscompleted, payload{ analysis: analysis, source_count: len(results) } ) async def _analyze(self, query: str, results: list) - str: # 模拟 LLM 分析 await asyncio.sleep(1) return f针对 {query} 的分析基于 {len(results)} 条结果核心观点是……这两个 Agent 的逻辑很直白但实际项目里你需要处理更多边界情况搜索结果为空怎么办、分析结果置信度低怎么办、LLM 调用超时怎么办。这些我后面会讲。3.4 主管 Agent 的调度逻辑实现主管 Agent 是整个系统的核心它负责拆解任务、分配子任务、汇总结果。我一般把主管 Agent 的逻辑写成状态机每个状态对应一个处理阶段。class SupervisorAgent(BaseAgent): def __init__(self, name: str, bus: MessageBus): super().__init__(name, bus) self.task_queue: list [] self.completed_tasks: dict {} async def process(self, message: AgentMessage) - AgentMessage: if message.status completed: # 子任务完成记录结果 self.completed_tasks[message.task_id] message.payload return await self._check_completion(message) # 新任务拆解并分配 subtasks await self._decompose(message.payload.get(task, )) for subtask in subtasks: self.task_queue.append(subtask) await self.bus.publish(AgentMessage( senderself.name, receiversubtask[agent], parent_task_idmessage.task_id, payloadsubtask[payload] )) return AgentMessage( senderself.name, receiversystem, statusdispatched, payload{subtask_count: len(subtasks)} ) async def _decompose(self, task: str) - list: # 这里调 LLM 做任务拆解 # 实际项目里需要更复杂的拆解逻辑 return [ {agent: search_agent, payload: {query: task, next: analysis_agent}}, ] async def _check_completion(self, message: AgentMessage) - AgentMessage: # 检查是否所有子任务都完成了 if len(self.completed_tasks) len(self.task_queue): return AgentMessage( senderself.name, receiverreport_agent, statusready_for_report, payload{results: self.completed_tasks} ) return AgentMessage( senderself.name, receiversystem, statuswaiting, payload{completed: len(self.completed_tasks), total: len(self.task_queue)} )这个主管 Agent 的逻辑是简化版实际项目里你需要处理子任务之间的依赖关系、并行执行、超时控制。但核心思路是一样的拆解、分配、监控、汇总。3.5 跑通第一个 Multi-Agent 流程把上面的代码拼起来跑一个完整的流程async def main(): bus MessageBus() supervisor SupervisorAgent(supervisor, bus) search_agent SearchAgent(search_agent, bus) analysis_agent AnalysisAgent(analysis_agent, bus) bus.register(supervisor) bus.register(search_agent) bus.register(analysis_agent) # 发起一个任务 await bus.publish(AgentMessage( senderuser, receiversupervisor, payload{task: 分析 2026 年 AI Agent 的发展趋势} )) # 等待流程完成 await asyncio.sleep(5) print(流程完成历史消息数, len(bus.history)) if __name__ __main__: asyncio.run(main())这个流程跑通之后你会看到消息在 Agent 之间流转。但别高兴太早这只是最理想的情况。实际项目里你会遇到各种问题消息丢失、Agent 卡死、上下文爆炸、LLM 调用失败。下面我就来讲这些坑怎么填。4. 实战中踩过的坑与排查技巧4.1 上下文爆炸的三种典型场景与解法上下文爆炸是 Multi-Agent 系统里最常见的问题没有之一。我总结了三类典型场景每一类都有对应的解法。第一类是“消息累积型爆炸”。Agent 每收到一条消息就往上下文里塞塞着塞着就满了。解法很简单给每个 Agent 设定上下文上限超过上限就触发裁剪。裁剪策略我一般用“保留最近 N 条 所有标记为关键的消息”。关键消息的标记由 Agent 自己决定比如分析 Agent 会把“最终结论”标记为关键中间过程不标记。第二类是“大 payload 型爆炸”。某个 Agent 返回了一个巨大的 JSON直接塞进下一个 Agent 的上下文瞬间爆掉。解法是强制 payload 大小限制超过 2000 token 的部分必须存外部存储消息里只放引用。我一般用 Redis 的SETEX命令存设置 1 小时过期避免垃圾数据堆积。第三类是“循环引用型爆炸”。两个 Agent 互相调用A 调 BB 又调 A无限循环。解法是给每个任务链设置最大深度超过深度就强制终止。我一般设最大深度为 10超过就抛异常让主管 Agent 决定怎么处理。爆炸类型典型表现解法预防措施消息累积型上下文逐渐变长响应变慢定期裁剪保留关键消息设定上下文上限自动触发裁剪大 payload 型单条消息巨大解析失败外部存储 引用 ID强制 payload 大小限制循环引用型两个 Agent 无限互调设置最大调用深度调用链追踪检测循环4.2 Agent 卡死与超时处理Agent 卡死比上下文爆炸更隐蔽因为它不报错就是不动了。我遇到过几次都是因为 LLM 调用没有设超时或者工具调用卡在了某个网络请求上。解法是给所有外部调用设超时LLM 调用设 30 秒工具调用设 10 秒超时就抛异常让重试机制接管。还有一个隐蔽的卡死场景Agent 在等一个永远不会来的消息。比如主管 Agent 分配了子任务但子任务 Agent 因为某种原因没有返回结果主管 Agent 就一直等。解法是给每个子任务设超时超过时间没有完成就标记为失败主管 Agent 根据失败策略决定下一步。async def call_with_timeout(coro, timeout30): try: return await asyncio.wait_for(coro, timeouttimeout) except asyncio.TimeoutError: raise TimeoutError(f调用超时超过 {timeout} 秒)这个call_with_timeout函数我每个项目都会用简单但救命。4.3 消息丢失与重复处理的排查消息丢失在异步系统里很常见尤其是用消息队列的时候。我排查消息丢失一般分三步先看消息总线的日志确认消息有没有发出去再看接收方的日志确认消息有没有收到最后看处理逻辑确认消息有没有被正确处理。重复处理是另一个坑。因为网络抖动或者重试机制同一条消息可能被处理两次。解法是给每条消息加唯一 ID接收方维护一个“已处理消息 ID 集合”收到消息先查集合如果已经处理过就直接跳过。这个集合我一般用 Redis 的 Set 来存设置 24 小时过期。注意已处理消息 ID 集合不能无限增长一定要设过期时间。我见过有人不设过期跑了三个月之后 Redis 里堆了几百万个 ID内存直接爆了。4.4 常见问题速查表问题现象可能原因排查步骤解决方案Agent 不响应LLM 调用超时检查 LLM 调用日志设超时 重试上下文溢出消息累积过多检查上下文长度裁剪 外部存储消息丢失总线故障检查总线日志加确认机制 重试重复处理重试导致检查消息 ID去重集合流程卡死子任务未返回检查子任务状态超时 失败策略输出质量差上下文污染检查上下文内容隔离 裁剪成本飙升无限循环检查调用链最大深度限制5. 从单体到多体的迁移策略与成本控制5.1 渐进式迁移先拆一个 Agent 试试如果你现在有一个跑得还行的单体 Agent不要一上来就全拆成 Multi-Agent。我的建议是渐进式迁移先找一个最痛的环节拆出去跑通了再拆下一个。比如你的单体 Agent 在“搜索”这一步经常出问题那就先把搜索逻辑拆成一个独立的 Search Agent其他部分不变。这样你只需要处理一个 Agent 的通信和状态管理复杂度可控。跑通之后再把“分析”拆出去以此类推。我自己的项目就是这么迁的从单体到三个 Agent 花了大概两周中间没有一天是全部推倒重来的。渐进式迁移的好处是每一步都可回滚出了问题不至于全盘崩溃。5.2 Token 成本与延迟的平衡Multi-Agent 的 Token 成本通常比单体高因为 Agent 之间的通信本身就要消耗 Token。我实测过一个任务单体 Agent 消耗 5000 token拆成三个 Agent 之后消耗 12000 token翻了一倍多。但换来的好处是任务成功率从 60% 提升到了 90%这个 trade-off 是值得的。控制成本的关键是减少不必要的通信。我一般会做这几件事Agent 之间只传摘要不传全文、主管 Agent 的上下文只保留关键决策信息、非关键子任务的输出不进入主管上下文。这几招下来Token 成本能压到原来的 1.5 倍左右而不是 2 倍以上。延迟方面Multi-Agent 因为多了通信环节端到端延迟通常比单体高 30% 到 50%。如果你的场景对延迟敏感可以考虑并行执行无依赖的子任务。比如搜索和分析可以并行搜索 Agent 在跑的时候分析 Agent 可以先处理其他数据。5.3 什么情况下不该用 Multi-Agent最后说说什么情况下不该用 Multi-Agent。如果你的任务步骤少于 5 步、工具少于 5 个、上下文占用低于 50%那单体 Agent 完全够用没必要引入 Multi-Agent 的复杂度。我见过一个团队为了一个“查天气 推荐穿衣”的任务搞了三个 Agent结果调试成本比单体高了十倍效果还差不多。还有一种情况是任务高度依赖全局上下文。比如“根据整篇文档回答一个问题”这种任务拆成多个 Agent 反而会丢失全局信息不如单体 Agent 一次性处理。Multi-Agent 适合的是“可解耦、有明确子任务边界”的场景不是所有场景都适合。我在实际项目里的体会是Multi-Agent 是一种架构选择不是一种信仰。该用的时候用不该用的时候别硬上。判断标准就一条单体 Agent 是不是已经撞到天花板了如果是那就拆如果不是那就继续优化单体。