多Agent协作系统架构设计：角色分工、对话管理与流程编排的工程实践

单Agent跑通一个简单任务不难，但一旦任务需要多种能力、多个视角或严格的流程约束，单Agent的短板就会暴露：它既当产品经理又当工程师，既做规划又做执行，结果往往是规划不清晰、执行不稳定、错误无法自查。

多Agent协作系统的价值，不是“多几个人干活”，而是通过角色分工实现能力解耦，通过对话管理实现信息同步，通过流程编排实现执行可控。

但这三件事每一个都有工程上的坑。角色分不好，Agent之间会互相推诿；对话管不好，信息会在传递中丢失；流程编排不当，系统会陷入无限循环或死锁。

这篇文章从工程落地角度，拆解多Agent协作系统的三个核心设计维度。附可运行的代码骨架。

角色分工：不是“多几个Prompt”那么简单

角色分工的第一步是回答一个问题：这个任务需要哪些不可替代的能力？

以软件开发场景为例，MetaGPT把软件公司的SOP硬编码进Agent角色：产品经理负责需求分析和PRD产出，架构师负责系统设计，项目经理负责任务拆解，工程师负责编码，QA负责测试。每个角色有明确的输入和输出，输出物是结构化的交付物（PRD文档、架构图、代码、测试报告），而不是自然语言的“聊天记录”。

这种设计的核心优势是解耦。如果你把产品经理Agent换成更强的模型，只要它输出的PRD仍然符合规范，其他所有Agent都无需修改。

但角色分工也有反模式。最常见的错误是按“能力”而不是按“职责”划分角色。比如设置一个“搜索Agent”和一个“分析Agent”，看起来分工清晰，实际执行时分析Agent需要搜索能力，搜索Agent也需要理解分析目标，职责边界模糊导致协作效率低下。

正确的划分原则是按交付物划分：每个角色负责产出一种明确的交付物，交付物的格式和内容约束是角色之间的“接口协议”。

fromdataclassesimportdataclassfromtypingimportList,Dict,Any@dataclassclassRole:name:strdescription:strinput_schema:Dict[str,Any]# 输入交付物的格式output_schema:Dict[str,Any]# 输出交付物的格式system_prompt:str# 角色定义示例PRODUCT_MANAGER=Role(name="product_manager",description="负责需求分析和PRD产出",input_schema={"user_requirement":"str"},output_schema={"prd":"str","features":"List[str]","priority":"Dict[str, int]"},system_prompt="""你是一名资深产品经理。根据用户需求，输出结构化的PRD文档。 要求： 1. 明确核心功能列表，按优先级排序 2. 每个功能点包含：功能描述、验收标准、依赖关系 3. 输出格式为JSON，包含 prd、features、priority 三个字段""")ARCHITECT=Role(name="architect",description="负责系统架构设计",input_schema={"prd":"str","features":"List[str]"},output_schema={"architecture":"str","modules":"List[Dict]","tech_stack":"List[str]"},system_prompt="""你是一名系统架构师。根据PRD文档，设计系统架构。 要求： 1. 明确模块划分和模块间依赖关系 2. 给出技术选型建议 3. 输出格式为JSON""")

每个角色的Prompt中必须明确输出格式，这是多Agent协作的“接口契约”。如果输出是自由文本，下游Agent需要花大量精力去解析，协作效率会急剧下降。

对话管理：从“聊天记录”到“结构化消息池”

多Agent系统最常见的失败模式是信息在传递中丢失或失真。

基于自然语言对话的协作（比如两个Agent互相“聊天”）看起来灵活，但有两个致命问题：一是上下文长度迅速膨胀，二是关键信息容易被淹没在闲聊中。

MetaGPT的做法值得借鉴：Agent之间不直接对话，而是通过发布和订阅结构化消息来协作。所有Agent把消息发布到共享消息池，其他Agent根据角色订阅关系获取所需信息。

importasynciofromcollectionsimportdefaultdictfromtypingimportCallableclassMessagePool:"""共享消息池：Agent之间通过发布/订阅结构化消息协作"""def__init__(self):self.messages=[]# 所有消息self.subscribers=defaultdict(list)# 订阅关系self._lock=asyncio.Lock()defsubscribe(self,role_name:str,message_type:str,callback:Callable):"""Agent订阅特定类型的消息"""self.subscribers[message_type].append({"role":role_name,"callback":callback})asyncdefpublish(self,message:dict):"""发布消息，触发订阅者回调"""asyncwithself._lock:self.messages.append(message)message_type=message.get("type")forsubinself.subscribers.get(message_type,[]):awaitsub["callback"](message)defget_messages_for(self,role_name:str,message_types:List[str])->List[dict]:"""获取某个角色需要的消息"""return[mforminself.messagesifm.get("type")inmessage_typesandm.get("producer")!=role_name]

结构化消息的设计需要包含几个关键字段：

@dataclassclassAgentMessage:message_id:strproducer:str# 生产者角色message_type:str# 消息类型：PRD_CREATED / CODE_GENERATED / TEST_FAILEDpayload:dict# 结构化内容timestamp:floatcorrelation_id:str# 关联ID，用于追踪同一个任务的完整链路parent_id:str=None# 父消息ID，用于构建消息依赖树

correlation_id是实现可观测性的关键。一个任务从需求分析到代码生成可能产生几十条消息，通过correlation_id可以把它们串成一条完整的执行链路，用于问题排查和性能分析。

parent_id构建消息依赖树，让系统可以追溯“这个测试报告是基于哪个版本的代码生成的”。

流程编排：从固定流水线到动态协作

流程编排决定了Agent之间的执行顺序和依赖关系。有两种基本范式：

固定流水线（Pipeline）：角色按预定义顺序执行，像工厂装配线。MetaGPT采用的就是这种模式——产品经理→架构师→项目经理→工程师→QA，顺序固定，每个阶段的输出是下一阶段的输入。

动态协作（Orchestration）：由一个协调者Agent（Orchestrator）根据任务状态动态决定下一步调用哪个Agent。

固定流水线的优势是可预测、可审计。你清楚地知道每一步谁在做什么，出了问题容易定位。劣势是灵活性差，如果某个阶段的结果不满足要求，整个流程可能需要重跑。

动态协作的优势是灵活，可以根据执行结果实时调整策略。劣势是协调成本高，Orchestrator本身可能成为瓶颈或单点故障。

工程上的务实选择是混合模式：主流程用固定流水线保证可控性，在关键决策点引入动态路由处理异常情况。

classWorkflowOrchestrator:"""混合模式：固定流水线 + 关键节点动态路由"""def__init__(self,agents:dict,message_pool:MessagePool):self.agents=agents self.pool=message_pool self.max_iterations=5# 防止无限循环asyncdefexecute(self,task:dict)->dict:context={"task":task,"artifacts":{}}# 阶段1：需求分析（固定）prd=awaitself._run_agent("product_manager",context)context["artifacts"]["prd"]=prd# 阶段2：架构设计（固定）arch=awaitself._run_agent("architect",context)context["artifacts"]["architecture"]=arch# 阶段3：编码-测试循环（动态路由）iteration=0whileiteration<self.max_iterations:code=awaitself._run_agent("engineer",context)context["artifacts"]["code"]=code test_result=awaitself._run_agent("qa",context)iftest_result["passed"]:break# 测试失败，决定是让工程师修复还是让架构师重新设计decision=awaitself._route_decision(test_result,context)ifdecision=="REDESIGN":context["artifacts"]["architecture"]=awaitself._run_agent("architect",{**context,"feedback":test_result["errors"]})# decision == "FIX" 则继续循环，工程师基于错误信息修复iteration+=1ifiteration>=self.max_iterations:context["status"]="MAX_ITERATIONS_REACHED"returncontextasyncdef_route_decision(self,test_result:dict,context:dict)->str:"""根据测试结果决定下一步路由"""error_type=test_result.get("error_type","UNKNOWN")# 架构性问题需要重新设计iferror_typein["ARCHITECTURE_MISMATCH","MODULE_COUPLING"]:return"REDESIGN"# 实现性问题让工程师修复return"FIX"asyncdef_run_agent(self,agent_name:str,context:dict)->dict:agent=self.agents[agent_name]returnawaitagent.execute(context)

max_iterations是必须的约束。没有它，编码-测试循环可能永远跑不完。这在生产环境里意味着持续的Token消耗和无法收敛的任务状态。

常见失败模式与工程约束

多Agent系统在生产环境中有几种典型的失败模式，需要在架构设计时就考虑好应对。

失败模式一：无限辩论。两个Agent在对话中互不相让，循环到max_rounds。解决方案是设置硬性轮次上限，并在达到上限时强制引入第三方仲裁或直接返回当前最佳结果。

失败模式二：指令竞态。两个Agent同时操作同一资源，导致状态不一致。解决方案是在消息池层面引入乐观锁或版本号机制。

classVersionedResource:def__init__(self,resource_id:str,content:dict):self.resource_id=resource_id self.content=content self.version=0asyncdefupdate(self,new_content:dict,expected_version:int)->bool:"""乐观锁更新：版本不匹配则拒绝"""ifexpected_version!=self.version:returnFalse# 说明已被其他Agent修改self.content=new_content self.version+=1returnTrue

失败模式三：状态幻觉。Agent基于过期的快照做决策。解决方案是在每次Agent执行前，重新从消息池拉取最新的相关消息，而不是依赖缓存的上下文。

失败模式四：责任扩散。出了问题没人负责，因为“我只是执行了上游的指令”。解决方案是每条消息记录producer和parent_id，形成完整的责任链。当测试失败时，系统可以追溯是需求定义不清晰、架构设计有缺陷、还是编码实现有bug。

可观测性：多Agent系统必须的“黑匣子”

多Agent系统的执行链路比单Agent复杂得多，没有可观测性根本没法调试。至少需要记录：

• 每个Agent的输入和输出（谁在什么时候基于什么信息产出了什么）
• 消息池的完整流转记录（消息在Agent之间如何传递）
• 每个阶段的耗时和Token消耗（成本分析的基础）
• 决策路由的依据（为什么走了这条路径而不是另一条）

importstructlogfromopentelemetryimporttrace logger=structlog.get_logger()tracer=trace.get_tracer(__name__)classObservableAgent:def__init__(self,role:Role,llm_client):self.role=role self.llm=llm_clientasyncdefexecute(self,context:dict)->dict:withtracer.start_as_current_span(f"agent.{self.role.name}")asspan:span.set_attribute("agent.role",self.role.name)span.set_attribute("agent.input_keys",list(context.keys()))start=time.time()try:result=awaitself._do_execute(context)span.set_attribute("agent.status","success")span.set_attribute("agent.duration_ms",(time.time()-start)*1000)logger.info("agent_executed",role=self.role.name,duration_ms=(time.time()-start)*1000,output_keys=list(result.keys()))returnresultexceptExceptionase:span.set_attribute("agent.status","failed")span.record_exception(e)logger.error("agent_failed",role=self.role.name,error=str(e))raise

Trace ID需要在消息传递中透传，这样才能把跨Agent的调用串成一条完整的链路。当用户反馈“这次任务执行有问题”时，可以通过Trace ID快速定位是哪个Agent、哪一步出了问题。

写在最后

多Agent协作系统的设计，本质上是在**“能力解耦”和“协作成本”之间找平衡**。

角色分得太细，协作成本会超过分工收益；分得太粗，又失去了多Agent的意义。对话管理太松散，信息会丢失；太严格，又限制了Agent的灵活性。流程编排太固定，无法应对异常；太动态，又难以预测和审计。

务实的设计原则是：按交付物划分角色，用结构化消息池替代自然语言对话，用混合编排模式平衡可控性和灵活性，用完整的可观测性保证问题可追溯。

如果你的团队正在构建多Agent系统，建议从最小可行的角色划分开始——先验证两个角色之间的协作是否顺畅，再逐步扩展。不要一上来就设计五六个Agent的复杂系统，那样大概率会在协作的复杂性中迷失方向。