多智能体协作实战：框架选型vs自研，企业到底怎么选？

多智能体协作实战框架选型 vs 自研企业到底怎么选1. 先说结论没有最好的框架只有最贴合业务的选择。选框架还是自研取决于三个问题你的业务流程是固定流程还是需要动态协商你的团队有没有精力维护底层状态机你的业务对可控性的要求有多高2. 多智能体协作的四种模式在选框架之前先搞清楚你的业务属于哪种协作模式。模式一流水线PipelineAgent A → Agent B → Agent C → 输出特点顺序执行每个Agent只做一件事输出传给下一个。适合场景内容生成流水线 research → draft → review → publish数据处理流水线extract → transform → load代码审查流水线lint → type-check → security-scan → summary框架选择LangGraph有状态图或自研简单队列即可模式二并行分工ParallelAgent A ┐ Agent B ├──→ 汇总Agent → 输出 Agent C ┘特点多个Agent并行处理不同子任务最后汇总。适合场景多维度数据分析销售Agent 库存Agent 客户Agent并行多语言内容生成中文Agent 英文Agent 日文Agent并行多模型投票GPT-4 Claude Gemini分别回答投票决定框架选择CrewAI原生支持并行Task或 asyncioPython原生模式三辩论协商DebateAgent A ←→ Agent B ←→ Agent C → 裁判Agent → 输出特点多个Agent持不同观点通过辩论收敛到最优解。适合场景代码方案评审方案A Agent vs 方案B Agent评审Agent裁决投资决策看多Agent vs 看空Agent分析师Agent综合安全策略制定攻击视角Agent vs 防御视角Agent框架选择AutoGen对话驱动或自研消息队列 辩论轮次控制模式四层级指挥Hierarchical指挥官Agent ├── 执行Agent A ├── 执行Agent B └── 执行Agent C特点一个指挥官Agent分解任务、分配子任务、汇总结果。适合场景复杂项目管理PM Agent分解需求分配给开发Agent、测试Agent、文档Agent客服路由路由Agent判断意图分配给退款Agent、咨询Agent、投诉AgentRAG问答系统路由Agent选择知识库检索Agent执行检索框架选择CrewAIHierarchical Process或LangGraphSupervisor节点3. 三大框架的核心差异不只是API不同维度AutoGenLangGraphCrewAI协作方式对话驱动Agent间发消息图结构状态流转角色驱动Crew Task状态管理弱靠消息历史强TypedDict State中Task输出传递灵活性高自由对话高任意图结构中固定流程可控性低Agent可能跑偏高每步都可干预中流程可配置学习曲线陡概念多陡需要理解图状态平缓直观生产就绪中高中Human-in-the-Loop原生支持原生支持需自行实现适合场景快速原型、研究生产级系统、复杂流程内容流水线、快速交付4. 决策框架选框架还是自研4.1 用框架如果你……✅ 业务流程相对标准流水线或并行分工✅ 团队AI工程经验有限✅ 需要快速出Demo或MVP✅ 不需要深度定制底层通信机制推荐路径内容生成/报告生成 →CrewAI复杂状态流转/需要人工审批 →LangGraph快速验证多Agent对话效果 →AutoGen4.2 自研如果你……✅ 业务流程高度定制化行业特有流程✅ 对延迟/成本有极致要求框架开销不可接受✅ 需要与现有系统深度集成ERP、CRM、自研平台✅ 对可控性要求极高金融/医疗/安全场景✅ 团队有扎实的Python异步编程经验自研不是重造轮子而是只造你需要的轮子。5. 自研多智能体系统最小可行方案如果你决定自研下面给你一个最小可行架构约200行Python代码覆盖80%的企业场景。5.1 核心抽象fromabcimportABC,abstractmethodfromdataclassesimportdataclass,fieldfromtypingimportAny,Dict,List,Optional,CallablefromenumimportEnumimportasyncioimportloggingimporttimeimportuuid loggerlogging.getLogger(__name__)classAgentRole(Enum):WORKERworker# 执行者COORDINATORcoordinator# 协调者JUDGEjudge# 裁决者dataclassclassAgentMessage:Agent间通信消息msg_id:strfield(default_factorylambda:str(uuid.uuid4()))from_agent:strto_agent:str# 表示广播msg_type:strtask# task / result / error / broadcastpayload:AnyNonetimestamp:floatfield(default_factorytime.time)correlation_id:str# 关联ID用于追踪同一个任务的消息classBaseAgent(ABC):所有Agent的基类def__init__(self,name:str,role:AgentRoleAgentRole.WORKER):self.namename self.rolerole self.message_queue:asyncio.Queueasyncio.Queue()self.context:Dict[str,Any]{}# 共享上下文self._runningFalseabstractmethodasyncdefhandle(self,message:AgentMessage)-Optional[AgentMessage]:处理收到的消息返回响应消息或Nonepassasyncdefsend(self,to_agent:str,msg_type:str,payload:Any,correlation_id:str)-AgentMessage:发送消息给另一个AgentmsgAgentMessage(from_agentself.name,to_agentto_agent,msg_typemsg_type,payloadpayload,correlation_idcorrelation_idorstr(uuid.uuid4()),)# 实际生产中用Redis Pub/Sub或消息队列logger.info(f[{self.name}] → [{to_agent}]:{msg_type})returnmsgasyncdefrun(self,coordinator:Coordinator):Agent主循环self._runningTruewhileself._running:try:msgawaitasyncio.wait_for(self.message_queue.get(),timeout1.0)responseawaitself.handle(msg)ifresponse:awaitcoordinator.dispatch(response)exceptasyncio.TimeoutError:continueexceptExceptionase:logger.error(f[{self.name}] 处理消息失败:{e})error_msgawaitself.send(to_agentmsg.from_agentifmsgindir()else,msg_typeerror,payloadstr(e),correlation_idmsg.correlation_idifmsgindir()else,)awaitcoordinator.dispatch(error_msg)5.2 协调者消息路由classCoordinator:协调者负责消息路由和Agent生命周期管理def__init__(self):self.agents:Dict[str,BaseAgent]{}self.history:List[AgentMessage][]# 消息历史审计用defregister(self,agent:BaseAgent):注册Agentself.agents[agent.name]agent logger.info(f注册Agent:{agent.name}({agent.role.value}))asyncdefdispatch(self,msg:AgentMessage):路由消息self.history.append(msg)ifmsg.to_agent:# 广播消息forname,agentinself.agents.items():ifname!msg.from_agent:awaitagent.message_queue.put(msg)elifmsg.to_agentinself.agents:awaitself.agents[msg.to_agent].message_queue.put(msg)else:logger.warning(f目标Agent不存在:{msg.to_agent})asyncdefbroadcast(self,from_agent:str,payload:Any,correlation_id:str)-str:广播消息给所有Agentcorr_idcorrelation_idorstr(uuid.uuid4())msgAgentMessage(from_agentfrom_agent,to_agent,msg_typebroadcast,payloadpayload,correlation_idcorr_id,)awaitself.dispatch(msg)returncorr_idasyncdefrun_all(self):启动所有Agenttasks[asyncio.create_task(agent.run(self))foragentinself.agents.values()]awaitasyncio.gather(*tasks)defstop_all(self):停止所有Agentforagentinself.agents.values():agent._runningFalse5.3 实战三个Worker 一个协调者importopenaiclassResearchAgent(BaseAgent):研究Agent搜集资料def__init__(self):super().__init__(researcher,AgentRole.WORKER)asyncdefhandle(self,message:AgentMessage)-Optional[AgentMessage]:ifmessage.msg_typetask:topicmessage.payload.get(topic,)logger.info(f[researcher] 开始研究:{topic})# 调用LLM做研究实际生产中用搜索APIresponseawaitself._call_llm(f研究主题{topic}列出5个关键点)returnawaitself.send(to_agentmessage.from_agent,msg_typeresult,payload{topic:topic,findings:response},correlation_idmessage.correlation_id,)returnNoneasyncdef_call_llm(self,prompt:str)-str:# 简化示例实际用openai/anthropic等awaitasyncio.sleep(1)# 模拟LLM调用returnf关于{prompt}的研究结果...classWriterAgent(BaseAgent):写作Agent根据研究结果写文章def__init__(self):super().__init__(writer,AgentRole.WORKER)asyncdefhandle(self,message:AgentMessage)-Optional[AgentMessage]:ifmessage.msg_typeresultandmessage.from_agentresearcher:findingsmessage.payload.get(findings,)logger.info(f[writer] 开始写作基于研究结果)responseawaitself._call_llm(f根据以下研究结果写一篇文章{findings})returnawaitself.send(to_agentcoordinator,msg_typeresult,payload{article:response},correlation_idmessage.correlation_id,)returnNoneasyncdef_call_llm(self,prompt:str)-str:awaitasyncio.sleep(2)returnf基于研究结果生成的文章...\n\n{prompt[:50]}...classCoordinatorAgent(BaseAgent):协调者Agent分解任务、汇总结果def__init__(self):super().__init__(coordinator,AgentRole.COORDINATOR)self.results{}asyncdefhandle(self,message:AgentMessage)-Optional[AgentMessage]:ifmessage.msg_typestart:# 收到启动信号分配任务topicmessage.payload.get(topic)logger.info(f[coordinator] 收到任务开始协调:{topic})# 发给researcherreturnawaitself.send(to_agentresearcher,msg_typetask,payload{topic:topic},correlation_idmessage.correlation_id,)elifmessage.msg_typeresultandmessage.from_agentwriter:# 收到最终文章articlemessage.payload.get(article)logger.info(f[coordinator] 收到最终文章任务完成)self.results[message.correlation_id]article# 在实际生产中这里可以回调业务系统或写入数据库print(f\n✅ 任务完成文章\n{article}\n)returnNone# 运行示例asyncdefmain():# 创建协调者coordinatorCoordinator()# 创建并注册Agentcoord_agentCoordinatorAgent()researcherResearchAgent()writerWriterAgent()coordinator.register(coord_agent)coordinator.register(researcher)coordinator.register(writer)# 启动所有Agent异步asyncio.create_task(coordinator.run_all())# 发送启动任务corr_idawaitcoordinator.broadcast(from_agentuser,payload{topic:多智能体协作框架选型},)# 等待结果简化实际用asyncio.Eventawaitasyncio.sleep(5)# 输出结果ifcorr_idincoord_agent.results:print(f最终结果:{coord_agent.results[corr_id]})coordinator.stop_all()if__name____main__:asyncio.run(main())输出示例[coordinator] 收到任务开始协调: 多智能体协作框架选型 [researcher] 开始研究: 多智能体协作框架选型 [writer] 开始写作基于研究结果 [coordinator] 收到最终文章任务完成 ✅ 任务完成文章 基于研究结果生成的文章... 关于研究主题多智能体协作框架选型列出5个关键点的研究结果...6. 企业真实业务中的协作策略6.1 电商平台并行分工 汇总用户提问 → 意图识别Agent路由 ├── 商品推荐Agent并行 ├── 优惠计算Agent并行 └── 库存查询Agent并行 → 汇总Agent → 返回结果关键点并行Agent用asyncio.gather()同时执行设置超时单个Agent超时不影响整体降级返回部分结果结果汇总用我们上一篇文章的降级策略6.2 内容平台流水线 人工审批选题Agent → 研究Agent → 写作Agent → 审核Agent → 发布Agent ↓ 人工审批节点关键点用LangGraph的Human-in-the-Loop人工审批节点每个节点完成后保存检查点断点续传见上一篇文章审核不通过 → 返回写作Agent重新生成6.3 金融风控辩论协商 裁判贷款申请 → 信用评估Agent看多 → 风险评估Agent看空 → 裁判Agent综合打分 → 拒绝/通过/人工复核关键点辩论轮次控制最多3轮防止死循环裁判Agent用结构化输出Pydantic模型强制输出JSON所有决策记录审计日志合规要求7. 实战建议混合策略真实企业项目往往是框架 自研的组合LangGraph处理核心流程状态管理 人工审批节点 自研协调层与现有系统API对接 CrewAI式角色定义业务人员可配置Agent角色具体做法用LangGraph定义核心工作流有状态、可持久化、支持Human-in-the-Loop用自研Adapter对接企业内部系统ERP/CRM/数据库用配置文件定义Agent角色业务人员可以改prompt不需要改代码用我们前两篇文章的方案重试 断点续传 降级8. 选型Checklist回答这8个问题你就知道该选什么了问题选框架自研业务流程是否经常变化✅变化频繁建议自研是否需要人工审批节点LangGraph需自己实现是否要对接企业内部系统需写Adapter✅ 更灵活团队是否有异步编程经验✅ 框架封装好了需要是否要求极低延迟500ms框架有开销✅ 自研更轻量是否需要多模态图片/语音CrewAI/LangGraph需自己实现是否金融/医疗等强监管行业需深度定制✅ 自研可控性更高只是做Demo/MVP✅ CrewAI最快没必要综合判断0-3个✅→ 用框架4-6个✅→ 混合框架核心 自研Adapter7-8个✅→ 自研9. 总结多智能体协作框架是工具业务是目的。先搞清楚协作模式流水线/并行/辩论/层级再选合适的框架CrewAI/LangGraph/AutoGen需要深度定制就自研最小可行架构约200行Python生产环境一定要加重试机制 断点续传 降级策略见前两篇文章参考CrewAI vs LangGraph深度对比https://blog.csdn.net/m0_63171455/article/details/160901604三大框架电商实战性能对决https://blog.csdn.net/weixin_33725722/article/details/159839219AutoGen多智能体协作详解https://download.csdn.net/blog/column/12651786/147266060LangGraph官方文档https://langchain-ai.github.io/langgraph/CrewAI官方文档https://docs.crewai.com/下一篇预告《多智能体系统的监控与调试如何知道哪个Agent在偷懒》