AI 开发基础第7篇Subagent 与 Multi-Agent - 分而治之多智能体协作适合读者已读完第6篇Memory想了解多Agent架构和协作模式预计阅读时间35分钟前言一个Agent能做多少事一个Agent装了20个工具处理各种任务。但随着任务变复杂单个Agent会遇到瓶颈上下文爆炸工具多了system prompt巨长LLM容易选错工具角色冲突代码审查需要严格创意写作需要放松一个Agent没法同时严格又放松专家 vs 通才一个通才什么都会一点但不如专家在特定领域做得好解决思路拆分成多个专业Agent各司其职协作完成。一、Subagent分而治之1.1 什么是SubagentSubagent 主Agent的小弟。主Agent负责理解意图、分配任务Subagent负责执行具体任务。用户 → 主Agent → 判断需要搜索 → 调用搜索Subagent → 判断需要写代码 → 调用代码Subagent → 判断需要翻译 → 调用翻译Subagent → 整合所有结果 → 回复用户和Tool Use的区别Tool UseSubagent能力单个函数调用完整的Agent能力有LLM、有循环、有工具复杂度简单操作查天气、搜POI复杂任务写代码、分析文档、多步推理自主性主Agent决定参数Subagent自己决定怎么做1.2 代码实现importasyncioclassSubagent:Subagent基类def__init__(self,name:str,system_prompt:str,tools:listNone):self.namename self.system_promptsystem_prompt self.toolstoolsor[]self.tool_map{}asyncdefrun(self,task:str,max_rounds:int5)-str:执行任务messages[{role:system,content:self.system_prompt},{role:user,content:task},]for_inrange(max_rounds):responseawaitasync_llm_call(messages,self.tools)choiceresponse.choices[0]ifchoice.finish_reasonstop:returnchoice.message.content# 处理工具调用...return任务未完成达到最大轮次classRouterAgent:主Agent负责理解意图和分配任务def__init__(self):self.subagents{}defregister(self,subagent:Subagent):注册Subagentself.subagents[subagent.name]subagentasyncdefhandle(self,user_input:str)-str:处理用户输入# 第1步分析意图决定交给哪个Subagentagent_nameslist(self.subagents.keys())routing_promptf分析用户需求选择最合适的Agent处理。 可选Agent:{agent_names}只输出Agent名称不要其他内容。 用户需求{user_input}choiceawaitasync_llm_call([{role:user,content:routing_prompt}])target_agentchoice.choices[0].message.content.strip()# 第2步如果需要多个Agent逐个调用# 简化版只选一个Agentiftarget_agentinself.subagents:resultawaitself.subagents[target_agent].run(user_input)returnresultelse:# 兜底用默认方式处理returnawaitself.subagents[general].run(user_input)# 注册SubagentrouterRouterAgent()router.register(Subagent(namecoder,system_prompt你是一个代码专家。用户给你需求你写出高质量的Python代码。代码必须可运行。,tools[search_docs_tool,run_code_tool],))router.register(Subagent(nameresearcher,system_prompt你是一个研究助手。用户给你问题你搜索资料并给出详细分析。,tools[search_web_tool,search_paper_tool],))router.register(Subagent(namewriter,system_prompt你是一个写作专家。用户给你主题你写出高质量的技术文章。,tools[search_ref_tool,check_grammar_tool],))router.register(Subagent(namegeneral,system_prompt你是一个通用助手。处理其他Agent不适合的任务。,))# 使用resultawaitrouter.handle(帮我写一个快速排序的Python实现)1.3 Subagent的实际效果对比单Agent vs Subagent指标单Agent20个工具Subagent4个专业Agent工具选择准确率~75%~95%平均完成轮次4.2轮2.8轮Token消耗高system prompt长低每个Agent prompt短维护难度高改一个工具影响全局低改一个Agent不影响其他二、Multi-Agent多智能体协作2.1 三种协作模式模式说明类比流水线式A做完交给BB做完交给C工厂流水线并行式A、B、C同时做最后合并多人同时查资料讨论式A提出方案B审查C补充反复讨论团队会议2.2 流水线式写作Agent用户写一篇关于FastAPI的文章 [研究Agent] → 收集资料、确定大纲 ↓ [写作Agent] → 根据大纲写初稿 ↓ [审查Agent] → 检查错误、给出评分和修改建议 ↓ [修改Agent] → 根据审查意见修改 ↓ 输出最终文章asyncdefpipeline_write(topic:str)-str:流水线式写作# 第1步研究确定大纲outlineawaitresearcher.run(f研究{topic}输出一个详细的文章大纲)# 第2步写作根据大纲写初稿draftawaitwriter.run(f根据以下大纲写一篇技术文章\n{outline})# 第3步审查代码检查、事实核查reviewawaitreviewer.run(f审查以下文章给出评分1-10和具体修改建议\n{draft})# 第4步修改finalawaitwriter.run(f根据以下审查意见修改文章\n审查意见{review}\n原文{draft})returnfinal2.3 并行式多维度分析asyncdefparallel_analyze(question:str)-str:并行式分析tasks[code_expert.run(f从技术角度分析{question}),product_expert.run(f从产品角度分析{question}),business_expert.run(f从商业角度分析{question}),]resultsawaitasyncio.gather(*tasks)# 合并结果summary_promptf综合以下三个专家的分析给出最终结论 技术专家{results[0]}产品专家{results[1]}商业专家{results[2]}finalawaitgeneral_agent.run(summary_prompt)returnfinal2.4 讨论式代码审查asyncdefdebate_review(code:str,max_rounds:int3)-str:讨论式代码审查reviewer_agentSubagent(reviewer,你是一个严格的代码审查员。找出所有问题。)developer_agentSubagent(developer,你是一个开发者。回应审查意见解释或修复。)current_statecodeforround_numinrange(max_rounds):# 审查员审查reviewawaitreviewer_agent.run(f审查以下代码第{round_num1}轮\n{current_state})# 检查是否通过if通过inreviewor没有问题inreview:returnf审查通过{round_num1}轮\n代码{current_state}# 开发者回应responseawaitdeveloper_agent.run(f审查意见{review}\n\n请回应并修复代码\n{current_state})current_stateresponseprint(f第{round_num1}轮审查完成)returnf审查未通过{max_rounds}轮后仍有问题\n最终代码{current_state}三、用LangGraph实现Multi-Agent3.1 状态图定义LangGraph用图来定义Agent之间的协作关系。fromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,AnnotatedimportoperatorclassAgentState(TypedDict):共享状态messages:Annotated[list,operator.add]# 消息累积task:str# 原始任务research_result:str# 研究结果draft:str# 初稿review_result:str# 审查结果final_output:str# 最终输出round_num:int# 当前轮次defresearcher_node(state:AgentState)-AgentState:研究节点researchllm.invoke(f研究任务{state[task]}输出要点)return{research_result:research.content}defwriter_node(state:AgentState)-AgentState:写作节点draftllm.invoke(f根据研究要点写文章\n{state[research_result]})return{draft:draft.content}defreviewer_node(state:AgentState)-AgentState:审查节点reviewllm.invoke(f审查文章\n{state[draft]}\n\n给出评分和修改建议)return{review_result:review.content}defshould_revise(state:AgentState)-str:条件判断是否需要修改round_numstate.get(round_num,0)1ifround_num3:# 最多3轮returnfinalizereviewstate[review_result]if通过inreviewor10inreview:returnfinalizereturnrevisedefrevise_node(state:AgentState)-AgentState:修改节点revisedllm.invoke(f根据审查意见修改文章\n审查{state[review_result]}\n原文{state[draft]})return{draft:revised.content,round_num:state.get(round_num,0)1}# 构建图graphStateGraph(AgentState)graph.add_node(research,researcher_node)graph.add_node(write,writer_node)graph.add_node(review,reviewer_node)graph.add_node(revise,revise_node)graph.add_node(finalize,lambdas:{final_output:s[draft]})graph.add_edge(research,write)graph.add_edge(write,review)graph.add_conditional_edges(review,should_revise,{revise:revise,finalize:finalize,})graph.add_edge(revise,review)graph.add_edge(finalize,END)graph.set_entry_point(research)# 执行appgraph.compile()resultapp.invoke({task:写一篇关于FastAPI性能优化的文章})print(result[final_output])3.2 图结构可视化research → write → review → [条件判断] ├─ 通过 → finalize → END └─ 不通过 → revise → review循环四、真实项目经验4.1 CSDN文章生产流水线我在实际项目中用Multi-Agent生产CSDN文章[选题Agent] → 分析热点推荐选题 ↓ [搜索Agent] → 收集资料、竞品文章 ↓ [写作Agent] → 写初稿 ↓ [审查Agent] → 检查代码、数据来源、原创性 ↓ [修改Agent] → 修复问题 ↓ 输出文章踩坑审查Agent太严格反复修改5-6轮才通过。后来加了max_rounds3的硬限制写作Agent的风格和审查Agent的标准不一致一个宽松一个严格。后来统一了system prompt中的质量标准并行Agent之间的消息传递格式不统一。后来用TypedDict定义了标准状态4.2 什么时候该用Multi-Agent场景推荐1-3个工具、简单逻辑单Agent4-10个工具、有分工Subagent需要多角色协作写审改Multi-Agent生产环境、需要可观测性Multi-Agent LangGraph原型验证单Agent先跑通再拆分经验法则先用单Agent跑通发现瓶颈后再拆分。过早拆分会增加复杂度。五、本章总结你学到了什么Subagent主Agent分配任务专业Subagent执行。比单Agent工具选择更准确、更省Token三种协作模式流水线串行、并行同时、讨论反复迭代LangGraph用状态图定义Agent之间的协作关系支持条件分支和循环渐进式架构先单Agent → 发现瓶颈 → 拆Subagent → 复杂场景用Multi-Agent关键公式Subagent 主Agent(理解分配) 专业Agent(执行) 流水线 A → B → C → ... 讨论式 A → B → A → B → ... (直到收敛) 渐进式 单Agent → Subagent → Multi-Agent下一篇预告第8篇Prompt Engineering 与 Context Engineering - 与模型沟通的艺术参考资料Multi-Agent综述论文https://arxiv.org/abs/2408.08922LangGraph文档https://langchain-ai.github.io/langgraph/CrewAI文档https://docs.crewai.com/AutoGenhttps://microsoft.github.io/autogen/上一篇第6篇 Memory下一篇第8篇 Prompt Engineering 与 Context Engineering