Skip to content
🔗 分享本题
查看我的学习进度 →

Manager 生成任务 DAG 管理依赖和验收 Executor 执行专业子任务 MCP Tool 完成原子动作

🧠 图解记忆: Manager 管目标与依赖,Executor 管专业子任务,Tool 管原子动作,层级越清楚越可控。

💡 答案要点

扁平 vs 层级架构对比:

扁平架构(Flat):
  ┌─────────────────────────────────────┐
  │           Orchestrator              │
  │     (单点协调,所有决策)            │
  └─────────────────────────────────────┘
         ↓            ↓            ↓
    ┌────────┐   ┌────────┐   ┌────────┐
    │检索Agent│   │生成Agent│   │审核Agent│
    └────────┘   └────────┘   └────────┘
    问题:Orchestrator 成为瓶颈,单点故障

层级架构(Hierarchical):
  ┌─────────────────────────────────────┐
  │        Manager Agent(管理层)        │
  │  理解任务 → 分解 → 分派 → 汇总       │
  └─────────────────────────────────────┘
         ↓            ↓            ↓
  ┌────────────┐ ┌────────────┐ ┌────────────┐
  │ 执行层Agent │ │ 执行层Agent │ │ 执行层Agent │
  │  (检索)    │ │  (生成)    │ │  (审核)   │
  └────────────┘ └────────────┘ └────────────┘
         ↓            ↓            ↓
  ┌────────────┐ ┌────────────┐ ┌────────────┐
  │ 工具层MCP  │ │ 工具层MCP  │ │ 工具层MCP  │
  │ (搜索/数据库)│ │(生成模型) │ │(质量验证)  │
  └────────────┘ └────────────┘ └────────────┘

三层 Agent 职责划分(企业级标准):

层级Agent职责特点
L1 管理(Manager)Task Decomposer理解高层需求,分解子任务,协调执行顺序高层推理、低频调用
L2 执行(Executor)Retrieval Agent, Generation Agent接收具体子任务,执行并返回结果中层技能、高频调用
L3 工具(Tool)MCP Server 调用底层工具执行(搜索、数据库、API)原子操作、超高频调用

层级任务分解原理:

展开 Python 代码示例(43 行)
python
class HierarchicalMultiAgent:
    """层级多 Agent 系统"""
    
    def __init__(self):
        self.manager = ManagerAgent()       # L1
        self.executors = {                 # L2
            "retrieval": RetrievalAgent(),
            "generation": GenerationAgent(),
            "review": ReviewAgent()
        }
        self.tool_registry = ToolRegistry() # L3 (MCP)
    
    def solve(self, user_request: str) -> str:
        # L1: Manager 理解需求,分解为子任务图
        task_graph = self.manager.decompose(user_request)
        # task_graph = {"steps": [("retrieval", "query"), ...], "order": "sequential"}
        
        # L2: 按依赖顺序执行子任务
        context = {}
        for step_name, step_type in task_graph["steps"]:
            executor = self.executors.get(step_type)
            result = executor.execute(task=step_name, context=context)
            context[f"{step_type}_result"] = result
        
        # L1: Manager 汇总结果,生成最终答案
        final_answer = self.manager.aggregate(context)
        return final_answer


# 扁平 vs 层级的核心区别
"""
扁平:
  - 所有 Agent 都能调用所有工具
  - 容易产生循环调用、重复调用
  - 难以追踪任务状态
  - 适用场景:<5个 Agent、简单任务

层级:
  - Manager 才有任务分解权限
  - Executor 只能调用被分配的工具
  - 任务状态清晰可追踪
  - 适用场景:>5个 Agent、复杂任务
"""

HTTP 协议对多 Agent 通信的影响(面试重点):

为什么 HTTP 是多 Agent 通信的事实标准?

1. 无状态 = 每个请求独立,Agent 可以水平扩展
   HTTP 请求可以负载均衡到任意 Agent 实例

2. 文本头 = 元数据传递(TraceID、Authorization)
   headers["X-Trace-ID"] = "abc123"  # 分布式追踪
   headers["X-Agent-Role"] = "retrieval"  # 角色标注

3. REST 语义 = 清晰的操作语义
   POST = 创建任务
   GET = 查询状态
   DELETE = 取消任务

4. 广泛工具支持 = MQTT/WebSocket/SSE 都能 tunnel HTTP
   企业内部几乎所有中间件都支持 HTTP

⚠️ HTTP 的局限(必须说):
- 延迟高:每次请求都要 TCP 握手
- 无内置 pub/sub:无法像 MQTT 那样 pub/sub
- 不适合超低延迟场景:高频工具调用走 MCP 更高效

生产级 Agent 职责划分决策树:

有多少 Agent?
├── ≤3 个 → 扁平架构足够
├── 4~10 个 → 层级架构(Manager + Executors)
└── >10 个 → 层级 + 领域分组(Domain-based)

任务复杂度?
├── 简单单一任务 → 扁平(直接执行)
├── 中等(子任务有依赖)→ 层级(Manager 分解)
└── 复杂(子任务并行/有条件分支)→ 层级 + 任务队列

Agent 间通信频率?
├── 高频(工具调用级别)→ MCP 直连
├── 低频(任务委派级别)→ A2A/HTTP
└── 混合 → A2A+MCP 分层

企业级案例:智能客服多 Agent 系统职责划分:

用户:"帮我查一下我上个月的话费,并和上上个月对比"

L1 Manager Agent:
  → 理解意图:查询 + 对比
  → 分解任务:[("retrieval", "查询本月话费"), ("retrieval", "查询上月话费"), ("comparison", "对比两个月")]
  → 派发任务给 L2 Executor

L2 Executor Agent(Retrieval):
  → 调用 MCP Server 查询数据库
  → 返回两条账单数据

L2 Executor Agent(Comparison):
  → 接收两条数据
  → 调用计算工具 MCP Server
  → 返回对比结果

L1 Manager Agent:
  → 汇总结果,生成用户友好的回复
  → 返回:"您上个月话费 128元,比上上个月多了15元,主要是因为..."

面试话术:

"是否使用层级架构取决于任务依赖、权限边界和协调成本,而不是 Agent 数量超过某个固定阈值。Manager—Worker 适合集中分解与汇总,点对点或共享状态适合其他协作模式;无论哪种都要限制扇出、工具权限、预算和终止条件。MCP 是工具与上下文互操作协议,不是 HTTP 的低延迟替代品;高频通信应根据同进程调用、RPC、消息队列或流式传输等实际需求选型。"

📚 参考:Anthropic:multi-agent research system(编排者-工作者分工)