到目前为止，你构建的Agent都是单Agent系统——一个LLM实例、一套工具、一个循环。这种架构在很多场景下够用，但它有一个天花板：一个Agent只能有一个"视角"。
考虑这个任务："分析我们产品的用户评价，找出主要问题，并撰写一份改进建议报告。"
这个任务实际上包含三个性质不同的子任务：
如果用单Agent处理，它需要在三种"思维模式"之间切换：一会儿是数据分析师，一会儿是问题诊断师，一会儿是报告撰写者。这种频繁的角色切换会导致：
多Agent协作的核心思想：将任务分配给专门化的Agent，每个Agent只负责自己擅长的部分，通过消息传递协调完成整体任务。
在2.3节中你已经了解了多Agent的三种编排模式。现在深入到实现层面：
模式一：流水线（Pipeline）
数据收集Agent → 分析Agent → 报告Agent
每个Agent完成一个明确的阶段，输出直接作为下一个Agent的输入。最简单、最可控。
模式二：编排者-工作者（Orchestrator-Workers）
┌→ 工作者Agent A ─┐
编排者Agent ─→ 工作者Agent B ──→ 编排者Agent → 最终结果
└→ 工作者Agent C ─┘
一个主Agent负责任务分解和结果汇总，工作Agent负责执行具体子任务。
模式三：对等协作（Peer-to-Peer）
Agent A ←→ Agent B
↕ ↕
Agent C ←→ Agent D
Agent之间自由对话，没有固定层级。最灵活但最难控制。
本节重点实现模式一和模式二，因为它们是生产环境中最实用的。对等协作的不确定性太高，更适合研究场景。
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Callable
import json
import uuid
@dataclass
class AgentMessage:
"""Agent之间传递的消息"""
message_id: str
sender: str # 发送者Agent名称
receiver: str # 接收者Agent名称（"broadcast"表示广播）
content: str # 消息内容
message_type: str = "info" # info / request / result / error
timestamp: float = field(default_factory=lambda: __import__('time').time())
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
return {
"message_id": self.message_id,
"sender": self.sender,
"receiver": self.receiver,
"content": self.content,
"message_type": self.message_type,
"timestamp": self.timestamp,
"metadata": self.metadata,
}
@dataclass
class AgentConfig:
"""单个Agent的配置"""
name: str
role: str # 角色描述
system_prompt: str # 系统提示词
tools: List[str] = field(default_factory=list) # 可用工具名
model: str = "gpt-4o-mini"
temperature: float = 0.0
max_iterations: int = 10
Agent之间的消息传递需要一个消息总线来管理：
class MessageBus:
"""Agent间的消息总线"""
def __init__(self):
self.messages: List[AgentMessage] = []
self.subscribers: Dict[str, List[Callable]] = {}
def send(self, message: AgentMessage):
"""发送消息"""
self.messages.append(message)
# 通知订阅者
receivers = [message.receiver] if message.receiver != "broadcast" else list(self.subscribers.keys())
for receiver in receivers:
if receiver in self.subscribers:
for callback in self.subscribers[receiver]:
callback(message)
def subscribe(self, agent_name: str, callback: Callable):
"""订阅消息"""
if agent_name not in self.subscribers:
self.subscribers[agent_name] = []
self.subscribers[agent_name].append(callback)
def get_messages_for(self, agent_name: str) -> List[AgentMessage]:
"""获取发送给某个Agent的所有消息"""
return [
m for m in self.messages
if m.receiver == agent_name or m.receiver == "broadcast"
]
def get_conversation(self) -> List[AgentMessage]:
"""获取完整的消息对话记录"""
return self.messages
将你之前构建的Agent封装为一个可独立运行的单元：
class CollaborativeAgent:
"""可协作的Agent"""
def __init__(self, config: AgentConfig, llm_client,
tool_registry, message_bus: MessageBus):
self.config = config
self.client = llm_client
self.tools = tool_registry
self.bus = message_bus
self.memory = WorkingMemory(max_tokens=8000)
self.memory.set_system_prompt(config.system_prompt)
# 订阅发给自己的消息
self.bus.subscribe(config.name, self._on_message)
self.received_messages: List[AgentMessage] = []
def _on_message(self, message: AgentMessage):
"""收到消息时的处理"""
self.received_messages.append(message)
def run(self, task: str) -> str:
"""执行任务"""
self.memory.add_message("user", task, importance=5)
for iteration in range(self.config.max_iterations):
response = self.client.chat.completions.create(
model=self.config.model,
messages=self.memory.get_context(),
tools=self._get_available_tool_schemas(),
temperature=self.config.temperature,
)
message = response.choices[0].message
if message.tool_calls:
self.memory.add_message("assistant",
f"[调用工具] {message.tool_calls[0].function.name}")
for tool_call in message.tool_calls:
name = tool_call.function.name
args = json.loads(tool_call.function.arguments)
result = self.tools.execute(name, args)
self.memory.add_message("tool", result.content,
importance=4,
tool_call_id=tool_call.id)
else:
answer = message.content or ""
self.memory.add_message("assistant", answer)
return answer
return "达到最大迭代次数。"
def send_to(self, receiver: str, content: str,
message_type: str = "info") -> AgentMessage:
"""向另一个Agent发送消息"""
msg = AgentMessage(
message_id=str(uuid.uuid4())[:8],
sender=self.config.name,
receiver=receiver,
content=content,
message_type=message_type,
)
self.bus.send(msg)
return msg
def _get_available_tool_schemas(self) -> List[Dict[str, Any]]:
"""获取此Agent可用的工具Schema"""
if not self.config.tools:
return []
schemas = []
for tool_name in self.config.tools:
schema = self.tools.get_schema(tool_name)
if schema:
schemas.append(schema)
return schemas
流水线模式：一个Agent完成后，将结果传递给下一个Agent。
class PipelineOrchestrator:
"""流水线编排器"""
def __init__(self, agents: List[CollaborativeAgent],
verbose: bool = True):
self.agents = agents
self.verbose = verbose
def run(self, initial_input: str) -> str:
"""按顺序执行流水线中的Agent"""
current_input = initial_input
for i, agent in enumerate(self.agents):
if self.verbose:
print(f"\n{'='*50}")
print(f"阶段{i + 1}：{agent.config.name}（{agent.config.role}）")
print(f"{'='*50}")
# 执行当前Agent
result = agent.run(current_input)
if self.verbose:
print(f"输出：{result[:200]}...")
# 将结果传递给下一个Agent
current_input = result
return current_input
流水线示例
# 创建三个专门的Agent
data_agent = CollaborativeAgent(
AgentConfig(
name="data_collector",
role="数据收集员",
system_prompt="你是数据收集员。你的任务是收集用户评价数据。只收集和整理数据，不做分析。",
tools=["search_feedback", "filter_feedback"],
),
client, registry, bus,
)
analysis_agent = CollaborativeAgent(
AgentConfig(
name="analyst",
role="数据分析师",
system_prompt="你是数据分析师。你的任务是从数据中识别问题和趋势。基于数据说话，不要编造。",
tools=["analyze_sentiment", "find_patterns"],
),
client, registry, bus,
)
report_agent = CollaborativeAgent(
AgentConfig(
name="report_writer",
role="报告撰写者",
system_prompt="你是报告撰写者。你的任务是将分析结果转化为结构化的改进建议报告。",
tools=[],
),
client, registry, bus,
)
# 创建流水线
pipeline = PipelineOrchestrator([data_agent, analysis_agent, report_agent])
# 执行
final_report = pipeline.run("收集并分析用户对我们新产品的评价，生成改进建议")
编排者-工作者模式：主Agent分解任务并分配给工作Agent，最后汇总结果。
class OrchestratorWorkersSystem:
"""编排者-工作者多Agent系统"""
def __init__(self, orchestrator: CollaborativeAgent,
workers: Dict[str, CollaborativeAgent],
verbose: bool = True):
self.orchestrator = orchestrator
self.workers = workers
self.verbose = verbose
def run(self, task: str) -> str:
"""执行编排者-工作者流程"""
# 第1步：编排者分解任务
if self.verbose:
print(f"\n编排者 {self.orchestrator.config.name} 开始分解任务...")
decomposition = self.orchestrator.run(
f"请将以下任务分解为子任务，并指定每个子任务应该由哪个工作者完成。"
f"可用的工作者：{list(self.workers.keys())}\n\n任务：{task}"
)
# 第2步：解析任务分配
assignments = self._parse_assignments(decomposition)
if self.verbose:
print(f"\n任务分配：")
for assignment in assignments:
print(f" {assignment['worker']} ← {assignment['subtask']}")
# 第3步：并行/串行执行子任务
results = {}
for assignment in assignments:
worker_name = assignment.get("worker")
subtask = assignment.get("subtask")
if worker_name in self.workers:
worker = self.workers[worker_name]
if self.verbose:
print(f"\n工作者 {worker_name} 执行：{subtask}")
result = worker.run(subtask)
results[worker_name] = result
# 第4步：编排者汇总结果
if self.verbose:
print(f"\n编排者汇总结果...")
summary_input = (
f"以下是各工作者的执行结果，请汇总为最终答案：\n\n"
)
for worker_name, result in results.items():
summary_input += f"[{worker_name}]\n{result}\n\n"
final_answer = self.orchestrator.run(summary_input)
return final_answer
def _parse_assignments(self, decomposition: str) -> List[Dict[str, str]]:
"""从编排者的输出中解析任务分配"""
assignments = []
# 简单的解析：查找 "WorkerName: subtask" 模式
import re
lines = decomposition.split('\n')
for line in lines:
line = line.strip()
if not line:
continue
for worker_name in self.workers.keys():
if worker_name in line:
# 提取子任务描述
match = re.search(rf'{worker_name}[：:]\s*(.+)', line)
if match:
assignments.append({
"worker": worker_name,
"subtask": match.group(1),
})
break
elif worker_name in line:
# 尝试其他格式
assignments.append({
"worker": worker_name,
"subtask": line,
})
break
return assignments
编排者 coordinator 开始分解任务...
任务分配：
data_collector ← 收集最近30天用户评价数据
analyst ← 分析评价中的主要问题和趋势
report_writer ← 基于分析结果撰写改进建议
工作者 data_collector 执行：收集最近30天用户评价数据
输出：收集到156条评价...
工作者 analyst 执行：分析评价中的主要问题和趋势
输出：主要问题包括：登录速度慢（23%）、界面混乱（18%）...
工作者 report_writer 执行：基于分析结果撰写改进建议
输出：改进建议报告已完成...
编排者汇总结果...
最终输出：
根据对156条用户评价的分析，我们发现以下主要问题：
1. 登录速度慢（23%的用户提及）
2. 界面混乱（18%的用户提及）
...
建议的改进方向：
1. 优先优化登录流程，目标将登录时间缩短50%
2. 简化主导航菜单...
注意一：消息传递中的信息损失。
Agent之间传递的是文本结果，而非完整的推理过程。数据收集Agent的中间发现可能在传递给分析Agent时丢失。
应对：要求每个Agent在输出中包含关键数据和来源，而不仅仅是结论。在编排者汇总时，保留各工作者的原始输出。
注意二：成本叠加。
每个Agent都有自己的循环和LLM调用。一个三Agent的流水线，总token消耗可能是单Agent的三倍以上。
应对：只在确实需要角色分离时使用多Agent。对于简单任务，单Agent更经济。
注意三：错误的级联放大。
流水线中前一个Agent的错误会传递给后一个Agent，并可能被放大。如果数据收集Agent收集了错误的数据，分析Agent的分析结果必然错误。
应对：在每个阶段添加验证步骤——检查输入是否合理。编排者在汇总时应该质疑工作者的结果，而非直接接受。
多Agent协作让Agent系统突破单Agent的能力天花板。本节实现了：
核心原则：多Agent的价值在于角色分离——让每个Agent专注于一种思维模式。但这种价值是有代价的：更高的成本、更复杂的信息传递、更多的故障点。在引入多Agent之前，先确认单Agent确实不足以胜任。
在下一节中，我们将通过一个完整的实战案例，将ReAct、规划、反思和多Agent协作整合到一个真实的任务中。
评论专区
评论加载中...登录后即可发表评论