AI Agent 架构实战：多 Agent 协作系统的设计陷阱与工程解法

____simple_html_dom__voku__html_wrapper____>

AI Agent 架构实战：多 Agent 协作系统的设计陷阱与工程解法

一、从单 Agent 到多协作：当 LLM 不再是"一个人在战斗"

单个 Agent 能完成简单任务，但面对复杂业务流程——比如"分析用户反馈、提取关键问题、查询知识库、生成回复并审核合规性"——单 Agent 的 Prompt 会膨胀到难以维护，工具选择准确率急剧下降。实测数据表明，当工具数量超过 15 个时，Function Calling 的首次调用准确率从 92% 降至 67%。

多 Agent 协作的核心动机是"分而治之"：让每个 Agent 只负责一个领域，拥有少量精准工具，从而提升单步决策的可靠性。但多 Agent 系统引入了新的工程难题——Agent 间的状态传递、任务编排、错误隔离和上下文管理。如果设计不当，系统复杂度不降反升，调试成本远超单体方案。

二、多 Agent 协作架构与消息流转机制

flowchart TB
    subgraph Orchestrator["编排层"]
        ORC["Orchestrator<br/>任务分解与调度"]
        STATE["Shared State<br/>共享状态存储"]
    end
    subgraph Agents["Agent 集群"]
        A1["Analyzer Agent<br/>意图分析"]
        A2["Retriever Agent<br/>知识检索"]
        A3["Generator Agent<br/>内容生成"]
        A4["Reviewer Agent<br/>合规审核"]
    end
    subgraph Infra["基础设施"]
        MB["Message Bus<br/>消息总线"]
        MEM["Memory Store<br/>长期记忆"]
        TOOL["Tool Registry<br/>工具注册中心"]
    end
    ORC -->|"分解任务"| MB
    MB -->|"分发"| A1
    MB -->|"分发"| A2
    MB -->|"分发"| A3
    MB -->|"分发"| A4
    A1 -->|"分析结果"| MB
    A2 -->|"检索结果"| MB
    A3 -->|"生成内容"| MB
    A4 -->|"审核结果"| MB
    MB -->|"汇聚"| ORC
    ORC <-->|"读写"| STATE
    A1 <-->|"工具调用"| TOOL
    A2 <-->|"工具调用"| TOOL
    A3 <-->|"工具调用"| TOOL
    A4 <-->|"工具调用"| TOOL
    A1 <-->|"记忆读写"| MEM
    A2 <-->|"记忆读写"| MEM

多 Agent 系统的架构核心是"编排模式"。目前主流有三种：

顺序编排（Sequential）：Agent 按固定链路依次执行，前一个的输出是后一个的输入。类似 Pipeline 模式，优点是流程可预测，缺点是延迟累加。

层级编排（Hierarchical）：一个主 Agent 负责任务分解和调度，子 Agent 各自执行子任务。主 Agent 持有全局视图，能做动态决策。

去中心化编排（Decentralized）：Agent 之间通过消息总线直接通信，自主决定下一步。灵活但不可预测，调试困难。

生产环境中，层级编排是最务实的选择——它在灵活性和可控性之间取得了平衡。

三、生产级多 Agent 系统的代码实现

3.1 Agent 基础抽象与工具绑定

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import Any, Optional
import asyncio
@dataclass
class AgentMessage:
    """Agent 间传递的消息协议"""
    sender: str               # 发送方 Agent ID
    receiver: str             # 接收方 Agent ID，"broadcast" 表示广播
    content: Any              # 消息内容
    msg_type: str             # 消息类型：task/result/error
    correlation_id: str       # 关联 ID，用于追踪请求-响应链路
    metadata: dict = field(default_factory=dict)
class BaseAgent(ABC):
    """Agent 基础抽象，所有 Agent 必须实现 process 方法"""
    def __init__(self, agent_id: str, tools: list, llm_client: Any):
        self.agent_id = agent_id
        self.tools = {t.name: t for t in tools}
        self.llm = llm_client
        self.max_retries = 3   # 工具调用最大重试次数
    @abstractmethod
    async def process(self, message: AgentMessage) -> AgentMessage:
        """处理接收到的消息，返回响应消息"""
        pass
    async def call_tool(self, tool_name: str, params: dict) -> Any:
        """带重试和超时的工具调用"""
        tool = self.tools.get(tool_name)
        if not tool:
            raise ValueError(f"Agent {self.agent_id} 无可用工具: {tool_name}")
        last_err = None
        for attempt in range(self.max_retries):
            try:
                result = await asyncio.wait_for(
                    tool.run(**params),
                    timeout=tool.timeout
                )
                return result
            except asyncio.TimeoutError:
                last_err = f"工具 {tool_name} 第 {attempt+1} 次调用超时"
            except Exception as e:
                last_err = f"工具 {tool_name} 调用异常: {str(e)}"
        raise RuntimeError(f"工具 {tool_name} 调用失败: {last_err}")

3.2 编排器实现

class Orchestrator:
    """层级编排器：负责任务分解、Agent 调度和结果汇聚"""
    def __init__(self):
        self.agents: dict[str, BaseAgent] = {}
        self.shared_state: dict = {}
        self.execution_log: list[dict] = []
    def register(self, agent: BaseAgent):
        self.agents[agent.agent_id] = agent
    async def execute_pipeline(
        self,
        task: str,
        pipeline: list[str],      # Agent 执行顺序列表
        timeout_per_step: float = 30.0
    ) -> dict:
        """按顺序执行 Pipeline，每步超时独立控制"""
        correlation_id = f"task_{id(task)}"
        current_input = task
        for agent_id in pipeline:
            agent = self.agents.get(agent_id)
            if not agent:
                return {"error": f"未注册的 Agent: {agent_id}"}
            msg = AgentMessage(
                sender="orchestrator",
                receiver=agent_id,
                content=current_input,
                msg_type="task",
                correlation_id=correlation_id
            )
            try:
                result = await asyncio.wait_for(
                    agent.process(msg),
                    timeout=timeout_per_step
                )
                # 记录执行日志，用于事后审计
                self.execution_log.append({
                    "agent": agent_id,
                    "input": current_input,
                    "output": result.content,
                    "status": "success"
                })
                current_input = result.content
            except asyncio.TimeoutError:
                self.execution_log.append({
                    "agent": agent_id,
                    "status": "timeout"
                })
                return {"error": f"Agent {agent_id} 执行超时"}
        return {"result": current_input, "log": self.execution_log}

3.3 共享状态与上下文管理

class SharedState:
    """Agent 间的共享状态管理，支持读写锁和版本控制"""
    def __init__(self):
        self._state: dict = {}
        self._versions: dict[str, int] = {}
        self._lock = asyncio.Lock()
    async def get(self, key: str) -> Any:
        async with self._lock:
            return self._state.get(key)
    async def set(self, key: str, value: Any):
        async with self._lock:
            self._state[key] = value
            self._versions[key] = self._versions.get(key, 0) + 1
    async def get_version(self, key: str) -> int:
        """获取数据的版本号，用于乐观锁冲突检测"""
        return self._versions.get(key, 0)

四、多 Agent 架构的代价与边界

上下文膨胀问题：每个 Agent 独立维护对话历史，当 Agent 数量增多时，总 Token 消耗线性增长。5 个 Agent 的系统，Token 消耗约为单 Agent 的 3-4 倍（考虑编排器自身的开销）。在成本敏感场景下，这是必须量化的指标。

调试复杂度指数级上升：单 Agent 出错只需检查一条链路；多 Agent 系统中，错误可能在 Agent 间的消息传递中产生，也可能在共享状态的并发修改中产生。建议从第一天起就建立结构化的执行日志，记录每步的输入、输出和耗时。

编排器的单点风险：层级编排中，编排器是单点。如果编排器自身出现 LLM 幻觉——错误分解任务或调度了错误的 Agent——整个流程都会偏离。缓解方案是给编排器增加"自省"环节：在调度前让编排器验证分解结果是否覆盖了原始任务的所有子目标。

适用边界：当任务可以明确分解为 3-5 个独立步骤，且步骤间的数据传递格式可标准化时，多 Agent 架构的收益最大。当任务本身是端到端的生成（如写一篇文章），单 Agent 加长上下文反而更高效。

五、总结

多 Agent 协作系统的核心价值在于"降低单步决策复杂度"，而非"用更多 Agent 解决更多问题"。工程落地的关键有三点：第一，编排模式选择层级编排，在可控性和灵活性间取得平衡；第二，Agent 间通过结构化消息协议通信，避免隐式状态依赖；第三，从第一天起建立执行日志和 Token 消耗监控，为后续优化提供数据支撑。

落地路线建议：先从顺序编排的 2-Agent 链路起步，验证消息协议和状态管理是否跑通；再逐步引入并行调度和动态路由；最后根据执行日志数据决定是否需要更复杂的去中心化编排。切忌一步到位搭建"全功能多 Agent 框架"，先让最小可用系统跑起来，再按需演进。