单Agent跑通一个简单任务不难，但一旦任务需要多种能力、多个视角或严格的流程约束，单Agent的短板就会暴露：它既当产品经理又当工程师，既做规划又做执行，结果往往是规划不清晰、执行不稳定、错误无法自查。
多Agent协作系统的价值，不是“多几个人干活”，而是通过角色分工实现能力解耦，通过对话管理实现信息同步，通过流程编排实现执行可控。
但这三件事每一个都有工程上的坑。角色分不好，Agent之间会互相推诿；对话管不好，信息会在传递中丢失；流程编排不当，系统会陷入无限循环或死锁。
这篇文章从工程落地角度，拆解多Agent协作系统的三个核心设计维度。附可运行的代码骨架。
角色分工：不是“多几个Prompt”那么简单
角色分工的第一步是回答一个问题：这个任务需要哪些不可替代的能力？
以软件开发场景为例，MetaGPT把软件公司的SOP硬编码进Agent角色：产品经理负责需求分析和PRD产出，架构师负责系统设计，项目经理负责任务拆解，工程师负责编码，QA负责测试。每个角色有明确的输入和输出，输出物是结构化的交付物（PRD文档、架构图、代码、测试报告），而不是自然语言的“聊天记录”。
这种设计的核心优势是解耦。如果你把产品经理Agent换成更强的模型，只要它输出的PRD仍然符合规范，其他所有Agent都无需修改。
但角色分工也有反模式。最常见的错误是按“能力”而不是按“职责”划分角色。比如设置一个“搜索Agent”和一个“分析Agent”，看起来分工清晰，实际执行时分析Agent需要搜索能力，搜索Agent也需要理解分析目标，职责边界模糊导致协作效率低下。
正确的划分原则是按交付物划分：每个角色负责产出一种明确的交付物，交付物的格式和内容约束是角色之间的“接口协议”。
from dataclasses import dataclass
from typing import List, Dict, Any
@dataclass
class Role:
name: str
description: str
input_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根据角色订阅关系获取所需信息。
import asyncio
from collections import defaultdict
from typing import Callable
class MessagePool:
"""共享消息池：Agent之间通过发布/订阅结构化消息协作"""
def __init__(self):
self.messages = [] # 所有消息
self.subscribers = defaultdict(list) # 订阅关系
self._lock = asyncio.Lock()
def subscribe(self, role_name: str,
message_type: str,
callback: Callable):
"""Agent订阅特定类型的消息"""
self.subscribers[message_type].append({
"role": role_name,
"callback": callback
})
async def publish(self, message: dict):
"""发布消息，触发订阅者回调"""
async with self._lock:
self.messages.append(message)
message_type = message.get("type")
for sub in self.subscribers.get(message_type, []):
await sub["callback"](message)
def get_messages_for(self, role_name: str,
message_types: List[str]) -> List[dict]:
"""获取某个角色需要的消息"""
return [
m for m in self.messages
if m.get("type") in message_types
and m.get("producer") != role_name
]
结构化消息的设计需要包含几个关键字段：
@dataclass
class AgentMessage:
message_id: str
producer: str # 生产者角色
message_type: str # 消息类型：PRD_CREATED / CODE_GENERATED / TEST_FAILED
payload: dict # 结构化内容
timestamp: float
correlation_id: str # 关联ID，用于追踪同一个任务的完整链路
parent_id: str = None # 父消息ID，用于构建消息依赖树
correlation_id是实现可观测性的关键。 一个任务从需求分析到代码生成可能产生几十条消息，通过correlation_id可以把它们串成一条完整的执行链路，用于问题排查和性能分析。
parent_id构建消息依赖树，让系统可以追溯“这个测试报告是基于哪个版本的代码生成的”。
流程编排：从固定流水线到动态协作
流程编排决定了Agent之间的执行顺序和依赖关系。有两种基本范式：
固定流水线（Pipeline）：角色按预定义顺序执行，像工厂装配线。MetaGPT采用的就是这种模式——产品经理→架构师→项目经理→工程师→QA，顺序固定，每个阶段的输出是下一阶段的输入。
动态协作（Orchestration）：由一个协调者Agent（Orchestrator）根据任务状态动态决定下一步调用哪个Agent。
固定流水线的优势是可预测、可审计。你清楚地知道每一步谁在做什么，出了问题容易定位。劣势是灵活性差，如果某个阶段的结果不满足要求，整个流程可能需要重跑。
动态协作的优势是灵活，可以根据执行结果实时调整策略。劣势是协调成本高，Orchestrator本身可能成为瓶颈或单点故障。
工程上的务实选择是混合模式：主流程用固定流水线保证可控性，在关键决策点引入动态路由处理异常情况。
class WorkflowOrchestrator:
"""混合模式：固定流水线 + 关键节点动态路由"""
def __init__(self, agents: dict, message_pool: MessagePool):
self.agents = agents
self.pool = message_pool
self.max_iterations = 5 # 防止无限循环
async def execute(self, task: dict) -> dict:
context = {"task": task, "artifacts": {}}
# 阶段1：需求分析（固定）
prd = await self._run_agent("product_manager", context)
context["artifacts"]["prd"] = prd
# 阶段2：架构设计（固定）
arch = await self._run_agent("architect", context)
context["artifacts"]["architecture"] = arch
# 阶段3：编码-测试循环（动态路由）
iteration = 0
while iteration < self.max_iterations:
code = await self._run_agent("engineer", context)
context["artifacts"]["code"] = code
test_result = await self._run_agent("qa", context)
if test_result["passed"]:
break
# 测试失败，决定是让工程师修复还是让架构师重新设计
decision = await self._route_decision(test_result, context)
if decision == "REDESIGN":
context["artifacts"]["architecture"] = await self._run_agent(
"architect", {**context, "feedback": test_result["errors"]}
)
# decision == "FIX" 则继续循环，工程师基于错误信息修复
iteration += 1
if iteration >= self.max_iterations:
context["status"] = "MAX_ITERATIONS_REACHED"
return context
async def _route_decision(self, test_result: dict, context: dict) -> str:
"""根据测试结果决定下一步路由"""
error_type = test_result.get("error_type", "UNKNOWN")
# 架构性问题需要重新设计
if error_type in ["ARCHITECTURE_MISMATCH", "MODULE_COUPLING"]:
return "REDESIGN"
# 实现性问题让工程师修复
return "FIX"
async def _run_agent(self, agent_name: str, context: dict) -> dict:
agent = self.agents[agent_name]
return await agent.execute(context)
max_iterations是必须的约束。 没有它，编码-测试循环可能永远跑不完。这在生产环境里意味着持续的Token消耗和无法收敛的任务状态。
常见失败模式与工程约束
多Agent系统在生产环境中有几种典型的失败模式，需要在架构设计时就考虑好应对。
失败模式一：无限辩论。 两个Agent在对话中互不相让，循环到max_rounds。解决方案是设置硬性轮次上限，并在达到上限时强制引入第三方仲裁或直接返回当前最佳结果。
失败模式二：指令竞态。 两个Agent同时操作同一资源，导致状态不一致。解决方案是在消息池层面引入乐观锁或版本号机制。
class VersionedResource:
def __init__(self, resource_id: str, content: dict):
self.resource_id = resource_id
self.content = content
self.version = 0
async def update(self, new_content: dict, expected_version: int) -> bool:
"""乐观锁更新：版本不匹配则拒绝"""
if expected_version != self.version:
return False # 说明已被其他Agent修改
self.content = new_content
self.version += 1
return True
失败模式三：状态幻觉。 Agent基于过期的快照做决策。解决方案是在每次Agent执行前，重新从消息池拉取最新的相关消息，而不是依赖缓存的上下文。
失败模式四：责任扩散。 出了问题没人负责，因为“我只是执行了上游的指令”。解决方案是每条消息记录producer和parent_id，形成完整的责任链。当测试失败时，系统可以追溯是需求定义不清晰、架构设计有缺陷、还是编码实现有bug。
可观测性：多Agent系统必须的“黑匣子”
多Agent系统的执行链路比单Agent复杂得多，没有可观测性根本没法调试。至少需要记录：
- 每个Agent的输入和输出（谁在什么时候基于什么信息产出了什么）
- 消息池的完整流转记录（消息在Agent之间如何传递）
- 每个阶段的耗时和Token消耗（成本分析的基础）
- 决策路由的依据（为什么走了这条路径而不是另一条）
import structlog
from opentelemetry import trace
logger = structlog.get_logger()
tracer = trace.get_tracer(__name__)
class ObservableAgent:
def __init__(self, role: Role, llm_client):
self.role = role
self.llm = llm_client
async def execute(self, context: dict) -> dict:
with tracer.start_as_current_span(f"agent.{self.role.name}") as span:
span.set_attribute("agent.role", self.role.name)
span.set_attribute("agent.input_keys", list(context.keys()))
start = time.time()
try:
result = await self._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())
)
return result
except Exception as e:
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的复杂系统，那样大概率会在协作的复杂性中迷失方向。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/nanfengi77/article/details/167893027
