上个月接了个需求做一个能自动写行业分析报告的AI系统。不是那种三两句话的GPT回答是真的几十页的行业报告——有数据图表、有趋势分析、有竞品对比、有SWOT分析那种。一开始我天真地想找个大模型prompt调调就完事了。结果试了一周单Agent的方案根本搞不定。原因很简单写一份好报告需要好几种完全不同的能力。你得会搜数据、会分析数据、会写文字、会画图表、会组织结构。让一个Agent干所有活就像让一个人同时做厨师、会计、设计师和项目经理——不是不行但质量和效率都拉胯。于是我把目光投向了Multi-Agent架构。为什么单Agent搞不定复杂任务先说说我踩的坑。方案一是用一个大模型gpt-4o给一个超长的system prompt涵盖了数据搜索、分析、写作、配图所有职责。结果prompt长度超过8000 token开头写的规则模型后面全忘了数据分析部分写得还行一到总结部分就变流水账画图表别想了文字描述都画不清楚上下文太长模型越往后回答越糊方案二是把任务拆成多步分多次调用来完成。结果也不理想——上一步的结果下一步用不上上下文衔接总有问题。踩了个大坑之后我才意识到复杂任务需要的是分工协作不是一个人的超长待机。Multi-Agent的核心思路Multi-Agent的思路其实特别简单——符合人的工作习惯。你做个项目不会让一个人同时做调研、写文档、做PPT、给领导汇报对吧你会组个团队有人负责数据调研、有人负责写初稿、有人做设计、有人统筹进度。Multi-Agent就是把这套思路搬到AI系统里。每个Agent有自己擅长的领域和工具它们通过消息机制协作共同完成一个复杂目标。我最后的架构是这样的用户需求: 写一份新能源汽车行业分析报告 ↓ Orchestrator Agent (主控Agent) ├── 解析需求 → 拆解任务 → 分配任务 ├── ├── Research Agent (调研Agent) │ ├── 工具: 搜索引擎、行业数据库、新闻API │ └── 输出: 原始数据 信息来源 │ ├── Analysis Agent (分析Agent) │ ├── 工具: Python数据分析、统计模型 │ └── 输出: 分析结论 数据图表 │ ├── Writing Agent (写作Agent) │ ├── 工具: 文档模板、风格指南 │ └── 输出: 报告正文初稿 │ └── Review Agent (审核Agent) ├── 工具: 一致性检查、事实核查 └── 输出: 修改意见 终稿代码实现从零搭一个Multi-Agent框架说干就干我用Python实现了一套轻量级的Multi-Agent框架。核心代码不复杂贴上来了# multi_agent/core.pyfromtypingimportList,Dict,AnyfromdataclassesimportdataclassfromenumimportEnumimportjsonclassAgentStatus(Enum):IDLEidleWORKINGworkingDONEdoneFAILEDfaileddataclassclassMessage:sender:strrecipient:strcontent:Any msg_type:strtask# task, result, feedback, errorclassBaseAgent:def__init__(self,name:str,description:str,llm_config:dict):self.namename self.descriptiondescription self.llm_configllm_config self.statusAgentStatus.IDLE self.mailbox:List[Message][]defreceive(self,message:Message):self.mailbox.append(message)defprocess(self)-Message:每个Agent实现自己的处理逻辑raiseNotImplementedErrorclassOrchestratorAgent(BaseAgent):def__init__(self,agents:Dict[str,BaseAgent]):super().__init__(Orchestrator,任务分配与编排,{})self.agentsagents self.task_queue[]self.results{}defdecompose_task(self,user_request:str)-List[dict]:把用户请求拆解成子任务promptf你需要将以下任务拆解成可执行的子任务 用户需求{user_request}拆解规则 1. 每个子任务必须有明确的输出物 2. 子任务之间有明确的依赖关系 3. 为每个子任务指定最合适的Agent 以JSON格式输出 [ {{task: 调研新能源汽车市场数据, agent: ResearchAgent, depends_on: []}}, ... ]# 调LLM生成任务分解tasksself._call_llm(prompt)returnjson.loads(tasks)defrun(self,user_request:str)-Dict:tasksself.decompose_task(user_request)# 按依赖关系调度fortaskintasks:agent_nametask[agent]ifagent_nameinself.agents:msgMessage(senderself.name,recipientagent_name,contenttask[task],msg_typetask)self.agents[agent_name].receive(msg)# 收集结果forname,agentinself.agents.items():whileagent.mailbox:resultagent.process()self.results[name]result.contentreturnself.results各Agent的实现Research Agent数据调研这块我踩的坑最多。一开始让Agent直接用搜索引擎搜结果搜出来的全是广告和SEO垃圾。后来加了个可信源优先的策略classResearchAgent(BaseAgent):def__init__(self):super().__init__(Researcher,负责搜索和收集行业数据,{model:gpt-4o,temperature:0.3})self.search_toolsSearchToolset()self.trusted_sources[36kr.com,sohu.com/a/,finance.sina.com.cn,caixin.com,yicai.com,statista.com]defprocess(self)-Message:msgself.mailbox.pop(0)taskmsg.content# 先搜可信源resultself.search_tools.search_with_priority(task,self.trusted_sources)# 没搜到再从全网上找ifnotresult[data]:resultself.search_tools.search_web(task)returnMessage(senderself.name,recipientOrchestrator,contentresult)Analysis Agent分析Agent需要能跑Python代码做数据分析。最稳的做法是让它生成代码在沙箱环境执行classAnalysisAgent(BaseAgent):def__init__(self):super().__init__(Analyst,数据分析与图表生成,{model:gpt-4o,temperature:0.1})self.sandboxPythonSandbox()defprocess(self)-Message:msgself.mailbox.pop(0)datamsg.content# 让LLM生成分析代码codeself._generate_analysis_code(data)# 沙箱执行resultself.sandbox.execute(code)returnMessage(senderself.name,recipientOrchestrator,contentresult)踩坑集锦坑1Agent之间的消息格式要统一刚开始我让每个Agent自由定义返回值格式结果Orchestrator解析各Agent的输出需要写不同的解析器代码很快变得一坨。解决方案统一消息格式所有Agent的输入输出都用{content: ..., metadata: {...}}包装。元数据里放置信度、来源、处理时间等辅助信息。坑2死循环和无限递归有一次Analysis Agent处理完数据后输出格式不对Research Agent收到后以为是新任务又去搜了一遍。搜回来的结果又被Analysis处理又出错…循环了50多次直到API额度用完。解决方案给每个消息加ttlTime To Live字段每传递一次减1减到0就丢弃。同时在Orchestrator里加最大轮数限制。坑3不同Agent对同样的数据理解不一致Research搜回来2026年新能源汽车销量增长30%“Analysis理解成同比增长30%”Writing写成环比增长30%Review没发现这个不一致。解决方案在Review Agent里加专门的一致性检查步骤对关键数据做交叉验证。每个数据点标注计算口径。实际上线后的表现上周刚把整个系统部署到生产环境。说说实际的性能数据指标数值日均处理任务数约30个报告请求单个报告平均耗时4分20秒成功完成率92%失败的8%大多因为外部API超时平均API成本/报告约4.2元用户满意度评分4.3/5分最大的优化点在于并行度。一开始我是串行执行所有子任务后来改成有依赖关系的任务串行、无依赖的并行整体耗时从8分钟降到了4分钟左右。如果要重来我会怎么做回顾整个项目有几件事如果能重新选择我肯定会换种做法1. 一开始就该用现成的框架最开始我非要自己手写全套踩了不少坑后才去看了AutoGen和CrewAI的源码。结果发现人家早就处理了消息路由、错误重试、状态持久化这些核心问题。不要重复造轮子除非你想学习轮子怎么造。2. 慢启动策略开始就给每个Agent配置了完整的工具链结果有的Agent根本用不到。后来改成按需加载——Agent启动时只有基础能力运行过程中动态扩展工具集。3. 日志比我想象的重要得多调试Multi-Agent系统比调试单Agent难一个数量级。Agent之间的消息传递、上下文流转、决策过程如果没有完整的日志出了问题根本不知道从哪查起。后期我加了一套完整的追踪系统每个消息都带trace_id才算是能排查问题了。效果对比拿我那个行业报告需求来说Multi-Agent VS 单Agent的结果维度单AgentMulti-Agent报告质量60分结构乱85分逻辑清晰数据准确性经常编数据有来源可追溯生成速度3分钟5分钟含搜索灵活性换场景要重写prompt只换某个Agent就行成本每次1.5元每次3元说实话成本翻了一倍但质量提升了不止一个档次。对于需要交付给客户看的东西值。写在最后Multi-Agent不是什么新概念以前的机器人学、分布式系统里都有类似的思想。但大模型的出现让智能体协作从理论变成了可以落地的方案。我踩了一圈坑下来核心还是要回归到软件工程的基本原则高内聚、低耦合。每个Agent做好一件事沟通协议标准化错误处理兜好底。如果你也想尝试Multi-Agent建议从小处着手——先做两个Agent的协作比如调研写作跑通了再加其他Agent。一口吃不成胖子先让俩Agent好好配合再说。下期预告Multi-Agent的记忆共享如何让Agent之间心照不宣。参考资源AutoGen - Microsoft的Multi-Agent框架CrewAI - 轻量级Multi-Agent库我的Multi-Agent项目模板
