原文：https://www.mstx.cn/harness/438.html · 创建：2026-09-08 · 更新：2026-09-08 · 阅读：1066

---

单Agent的天花板

到目前为止，你构建的Agent都是单Agent系统——一个LLM实例、一套工具、一个循环。这种架构在很多场景下够用，但它有一个天花板：一个Agent只能有一个"视角"。

考虑这个任务："分析我们产品的用户评价，找出主要问题，并撰写一份改进建议报告。"

这个任务实际上包含三个性质不同的子任务：

• 数据分析：从大量评价中提取统计信息（需要严谨、精确）
• 问题识别：从数据中找出关键问题（需要洞察、判断）
• 报告撰写：将发现转化为结构化的建议（需要表达、组织）

如果用单Agent处理，它需要在三种"思维模式"之间切换：一会儿是数据分析师，一会儿是问题诊断师，一会儿是报告撰写者。这种频繁的角色切换会导致：

• 注意力分散：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之间自由对话，没有固定层级。最灵活但最难控制。

本节重点实现模式一和模式二，因为它们是生产环境中最实用的。对等协作的不确定性太高，更适合研究场景。

实现多Agent的基础设施

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的封装

将你之前构建的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在输出中包含关键数据和来源，而不仅仅是结论。在编排者汇总时，保留各工作者的原始输出。

注意二：成本叠加。

每个Agent都有自己的循环和LLM调用。一个三Agent的流水线，总token消耗可能是单Agent的三倍以上。

应对： 只在确实需要角色分离时使用多Agent。对于简单任务，单Agent更经济。

注意三：错误的级联放大。

流水线中前一个Agent的错误会传递给后一个Agent，并可能被放大。如果数据收集Agent收集了错误的数据，分析Agent的分析结果必然错误。

应对： 在每个阶段添加验证步骤——检查输入是否合理。编排者在汇总时应该质疑工作者的结果，而非直接接受。

本节小结

多Agent协作让Agent系统突破单Agent的能力天花板。本节实现了：

• 消息总线：Agent之间的通信基础设施
• 可协作的Agent：独立运行、可订阅消息的Agent单元
• 流水线模式：顺序执行的Agent链
• 编排者-工作者模式：主Agent分解任务、工作Agent执行、主Agent汇总

核心原则： 多Agent的价值在于角色分离——让每个Agent专注于一种思维模式。但这种价值是有代价的：更高的成本、更复杂的信息传递、更多的故障点。在引入多Agent之前，先确认单Agent确实不足以胜任。

在下一节中，我们将通过一个完整的实战案例，将ReAct、规划、反思和多Agent协作整合到一个真实的任务中。