很多团队做Agent系统的路径惊人地相似：先做一个单Agent跑通核心场景，效果不错，然后业务方提出更多需求——要能查数据库、要能调外部API、要能生成报告、要能处理异常。于是往单Agent里不断加工具、加Prompt、加逻辑分支，直到某一天发现这个Agent已经变成了一个难以维护的“怪物”：改一个工具的描述可能影响另一个场景的表现，排查一个问题需要翻几百行System Prompt。

这时候团队会考虑引入多Agent。但多Agent不是“把单Agent拆成几个”那么简单——拆不好，协作成本会超过分工收益；拆得好，系统才能从“能跑”走向“可扩展”。

这篇文章从工程演进视角，拆解从单Agent到多Agent的架构升级路径，以及如何设计一个真正可扩展的协作框架。附可运行的代码骨架。

单Agent的边界在哪里

先明确一个问题：什么情况下单Agent还能撑住，什么情况下必须拆？

单Agent适合的场景有三个特征：工具数量可控（<10个）、任务路径相对固定、错误容忍度较高。比如一个查询天气和日程的助手，一个根据模板生成文案的工具，这些场景单Agent完全够用。

但以下信号出现时，单Agent就会开始力不从心：

信号一：工具数量超过15个，模型选型准确率明显下降。 当工具描述占据大量上下文时，模型对每个工具的理解深度会下降。实测中，工具从10个增加到30个时，选型准确率从92%跌到67%。

信号二：Prompt长度超过3000字，且包含大量条件分支。 当System Prompt里写满了“如果用户问A则…，如果用户问B则…”时，这个Prompt已经变成了隐式的流程编排，维护成本极高。

信号三：不同场景对模型能力的要求差异显著。 症状识别需要快速响应，病历摘要需要长上下文理解，药物交互需要严格推理链——用一个模型服务所有场景，必然在某些场景上妥协。

信号四：任务需要多种不可替代的能力。 如果一个任务同时需要“检索知识库”、“生成结构化文档”、“校验输出质量”，这些能力由不同的Prompt和模型承担更高效。

当这些信号出现时，多Agent架构就值得考虑了。

拆分的核心原则：按交付物划分，而非按能力划分

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

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

以医疗AI场景为例，可以这样划分：

• 分诊Agent：输入患者症状描述，输出结构化分诊建议（科室、紧急程度、建议）
• 病历摘要Agent：输入患者历史病历，输出结构化摘要（主诉、既往史、用药史）
• 用药检查Agent：输入当前处方和患者过敏史，输出用药风险提示
• 报告生成Agent：输入以上所有结果，输出面向患者的可读报告

每个Agent的输入和输出都是结构化数据，而不是自然语言。这是可扩展性的基础——如果Agent之间靠“聊天”传递信息，上下文长度会迅速膨胀，关键信息容易被淹没。

from dataclasses import dataclass
from typing import Dict, Any, List

@dataclass
class AgentSpec:
    """Agent的接口协议定义"""
    name: str
    description: str
    input_schema: Dict[str, Any]   # 输入数据的schema
    output_schema: Dict[str, Any]  # 输出数据的schema
    system_prompt: str
    
# 病历摘要Agent的协议定义
RECORD_SUMMARY_AGENT = AgentSpec(
    name="record_summary",
    description="根据患者历史病历生成结构化摘要",
    input_schema={
        "patient_id": "str",
        "raw_records": "List[dict]"  # 原始病历记录
    },
    output_schema={
        "chief_complaint": "str",     # 主诉
        "medical_history": "str",     # 既往史
        "medication_history": "str",  # 用药史
        "allergies": "List[str]"      # 过敏史
    },
    system_prompt="""你是一名医疗记录整理助手。根据患者历史病历，
输出结构化的病历摘要。

要求：
1. 仅提取病历中明确记录的信息，不做推断
2. 如果某项信息缺失，对应字段返回 "未记录"
3. 输出格式为JSON，包含 chief_complaint、medical_history、
   medication_history、allergies 四个字段"""
)

输入输出schema是Agent之间的“契约”。只要契约不变，你可以随时替换某个Agent的实现——换更便宜的模型、换更精准的Prompt、甚至换成一个规则引擎——其他Agent无需任何改动。

协作框架的三个核心组件

一个可扩展的多Agent协作框架需要三个核心组件：Agent注册中心、消息总线、执行引擎。

组件一：Agent注册中心

注册中心管理所有Agent的元信息，让执行引擎可以动态发现和调用Agent。

class AgentRegistry:
    """Agent注册中心：管理所有可用的Agent"""
    
    def __init__(self):
        self._agents: Dict[str, 'BaseAgent'] = {}
        self._specs: Dict[str, AgentSpec] = {}
    
    def register(self, spec: AgentSpec, agent: 'BaseAgent'):
        """注册一个Agent"""
        if spec.name in self._agents:
            raise ValueError(f"Agent {spec.name} 已存在")
        self._agents[spec.name] = agent
        self._specs[spec.name] = spec
    
    def get(self, name: str) -> 'BaseAgent':
        if name not in self._agents:
            raise KeyError(f"Agent {name} 未注册")
        return self._agents[name]
    
    def get_spec(self, name: str) -> AgentSpec:
        return self._specs[name]
    
    def list_agents(self) -> List[str]:
        return list(self._agents.keys())
    
    def find_by_capability(self, capability: str) -> List[str]:
        """根据能力标签查找Agent"""
        return [
            name for name, spec in self._specs.items()
            if capability in spec.description
        ]

注册中心的价值在于解耦。执行引擎不需要硬编码“我要调用record_summary_agent”，而是通过能力查询动态发现可用的Agent。新增Agent只需要注册，不需要修改执行引擎。

组件二：消息总线

消息总线是Agent之间通信的基础设施。设计要点是结构化、可追溯、支持异步。

import asyncio
import uuid
import time
from dataclasses import dataclass, field

@dataclass
class AgentMessage:
    message_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    producer: str = ""
    message_type: str = ""
    payload: dict = field(default_factory=dict)
    timestamp: float = field(default_factory=time.time)
    correlation_id: str = ""   # 关联ID，追踪完整任务链路
    parent_id: str = None      # 父消息ID，构建依赖树

class MessageBus:
    """消息总线：Agent之间的结构化通信"""
    
    def __init__(self):
        self._messages: List[AgentMessage] = []
        self._subscribers: Dict[str, List] = {}
        self._lock = asyncio.Lock()
    
    def subscribe(self, message_type: str, callback):
        """订阅特定类型的消息"""
        self._subscribers.setdefault(message_type, []).append(callback)
    
    async def publish(self, message: AgentMessage):
        """发布消息，触发订阅者回调"""
        async with self._lock:
            self._messages.append(message)
        
        for callback in self._subscribers.get(message.message_type, []):
            await callback(message)
    
    def get_messages(self, correlation_id: str) -> List[AgentMessage]:
        """获取某个任务的完整消息链路"""
        return [
            m for m in self._messages
            if m.correlation_id == correlation_id
        ]

correlation_id是实现可观测性的关键。一个任务从开始到结束可能产生几十条消息，通过correlation_id可以把它们串成完整的执行链路。

组件三：执行引擎

执行引擎负责调度Agent执行，管理任务状态和错误处理。

class ExecutionEngine:
    """执行引擎：调度Agent执行，管理任务生命周期"""
    
    def __init__(self, registry: AgentRegistry, bus: MessageBus):
        self.registry = registry
        self.bus = bus
        self.max_steps = 20  # 防止无限循环
    
    async def execute(self, workflow: 'Workflow', 
                       initial_input: dict) -> dict:
        """执行一个工作流"""
        correlation_id = str(uuid.uuid4())
        context = {
            "correlation_id": correlation_id,
            "input": initial_input,
            "artifacts": {},
            "status": "running"
        }
        
        step = 0
        current_node = workflow.entry_node
        
        while current_node and step < self.max_steps:
            step += 1
            
            # 1. 检查节点条件（是否跳过）
            if not workflow.should_execute(current_node, context):
                current_node = workflow.get_next_node(
                    current_node, context, skipped=True
                )
                continue
            
            # 2. 执行Agent
            try:
                agent = self.registry.get(current_node.agent_name)
                result = await agent.execute(context)
                context["artifacts"][current_node.name] = result
                
                # 3. 发布消息
                await self.bus.publish(AgentMessage(
                    producer=current_node.agent_name,
                    message_type=f"{current_node.name}_COMPLETED",
                    payload=result,
                    correlation_id=correlation_id
                ))
                
                # 4. 决定下一个节点
                current_node = workflow.get_next_node(
                    current_node, context
                )
                
            except Exception as e:
                # 5. 错误处理：尝试降级或中断
                handled = await self._handle_error(
                    current_node, context, e
                )
                if not handled:
                    context["status"] = "failed"
                    context["error"] = str(e)
                    break
                current_node = workflow.get_next_node(
                    current_node, context, error=True
                )
        
        if step >= self.max_steps:
            context["status"] = "max_steps_reached"
        
        return context

max_steps是必须的硬约束。 没有它，系统可能因为Agent之间的循环调用而永远跑不完。

工作流定义：可配置的编排

工作流定义了Agent之间的执行顺序和依赖关系。为了保持可扩展性，工作流应该配置化，而不是硬编码在代码里。

from dataclasses import dataclass
from typing import Callable, Optional

@dataclass
class WorkflowNode:
    name: str
    agent_name: str
    condition: Optional[Callable] = None  # 执行条件
    
@dataclass
class WorkflowEdge:
    from_node: str
    to_node: str
    condition: Optional[Callable] = None  # 路由条件

class Workflow:
    """可配置的工作流定义"""
    
    def __init__(self, name: str):
        self.name = name
        self.nodes: Dict[str, WorkflowNode] = {}
        self.edges: List[WorkflowEdge] = []
        self.entry_node: Optional[str] = None
    
    def add_node(self, node: WorkflowNode, is_entry: bool = False):
        self.nodes[node.name] = node
        if is_entry:
            self.entry_node = node.name
    
    def add_edge(self, edge: WorkflowEdge):
        self.edges.append(edge)
    
    def should_execute(self, node_name: str, context: dict) -> bool:
        node = self.nodes[node_name]
        if node.condition is None:
            return True
        return node.condition(context)
    
    def get_next_node(self, current: str, context: dict, 
                       skipped: bool = False, error: bool = False) -> Optional[str]:
        """根据当前节点和上下文决定下一个节点"""
        candidates = [
            e for e in self.edges if e.from_node == current
        ]
        for edge in candidates:
            if edge.condition is None:
                return edge.to_node
            if edge.condition(context, skipped=skipped, error=error):
                return edge.to_node
        return None

基于这个框架，医疗场景的工作流可以这样定义：

def build_medical_workflow() -> Workflow:
    wf = Workflow("medical_consultation")
    
    wf.add_node(WorkflowNode("triage", "triage_agent"), is_entry=True)
    wf.add_node(WorkflowNode("record_summary", "record_summary_agent"))
    wf.add_node(WorkflowNode("drug_check", "drug_check_agent"))
    wf.add_node(WorkflowNode("report", "report_generator_agent"))
    
    # 分诊后，如果是HIGH紧急程度，直接生成报告
    wf.add_edge(WorkflowEdge(
        "triage", "report",
        condition=lambda ctx, **kw: ctx["artifacts"]["triage"]["urgency"] == "HIGH"
    ))
    # 否则继续走完整流程
    wf.add_edge(WorkflowEdge("triage", "record_summary"))
    wf.add_edge(WorkflowEdge("record_summary", "drug_check"))
    wf.add_edge(WorkflowEdge("drug_check", "report"))
    
    return wf

工作流配置化的价值在于：新增场景不需要改代码。 如果医院新增了一个“检验报告解读”场景，只需要注册新的Agent，定义新的工作流，而不需要修改执行引擎。

可扩展性的三个检验标准

判断一个多Agent框架是否真正可扩展，看三个标准：

标准一：新增Agent是否零侵入。 新增一个Agent只需要实现接口、注册到Registry，不需要修改执行引擎或已有Agent的代码。这是最基本的可扩展性要求。

标准二：Agent实现是否可替换。 把某个Agent从“千问调用”换成“DeepSeek调用”，其他Agent是否完全无感知？如果替换一个Agent需要改其他Agent的代码，说明接口协议设计有问题。

标准三：工作流是否可组合。 能不能把两个已有的工作流组合成一个新的工作流？比如“分诊+病历摘要”的简单流程和“分诊+病历摘要+用药检查+报告生成”的完整流程，是否能共享同一套Agent定义？

写在最后

从单Agent到多Agent的演进，本质上是在**“能力解耦”和“协作成本”之间找平衡**。

拆得太细，协作成本会超过分工收益；拆得太粗，又失去了多Agent的意义。务实的设计原则是：按交付物划分Agent，用结构化消息传递信息，用配置化工作流编排执行顺序，用注册中心实现零侵入扩展。

如果你的团队正在考虑从单Agent升级到多Agent，建议从两个Agent的最小协作开始验证。先确认接口协议设计是否合理、消息传递是否顺畅、错误处理是否完备，再逐步扩展角色数量。不要一上来就设计五六个Agent的复杂系统，那样大概率会在协作的复杂性中迷失方向。

可扩展性的核心不是“能加多少Agent”，而是“加Agent的时候不用改已有的东西”。