复杂系统拆解为子图：如何用分层架构图让分布式 Agent 架构一目了然

一、深度引言与场景痛点

大家好，我是赵咕咕。

前阵子做一个多 Agent 协作系统，架构评审时我把一张图放到屏幕上——同事们都沉默了。那张图上有 14 个 Agent、8 个消息队列、6 个数据库、无数条连线。我自己画的时候都觉得头晕。

后来我意识到一个问题：一张图装不下整个系统。强行塞进去的架构图，谁也看不懂。更致命的是，改一个字或者重排一下布局，整张图就得重画。

解决方法是：用分层子图把复杂系统拆解开。每一层只展示够用的细节，层与层之间用明确的接口连接。

今天分享我在多 Agent 分布式系统中学到的"分层画图"方法论。

二、底层机制与原理深度剖析

2.1 分层架构图的核心思想

软件架构的分层设计已经很成熟了——表现层、业务层、数据层，大家都能理解。但画图的时候，很多人没有把这种分层思维用上。

分层架构图的关键：

• 每层独立成图：一层一张图，专注描述该层的内部结构和交互。
• 层间用明确接口连接：不画线连到另一个系统内部，而是连到一个"接口端点"。
• 自顶向下渐进展开：先看整个系统的高层全貌，感兴趣的话 zoom in 到某一层。

2.2 多 Agent 系统的四层架构

一个典型的生产级多 Agent 系统可以拆成四层：

flowchart TB subgraph L1["第一层：用户接入层"] UI[Web / API / 企业微信] end subgraph L2["第二层：编排层"] ORCH[Orchestrator Agent] end subgraph L3["第三层：Agent 执行层"] direction LR A1[检索 Agent] A2[数据分析 Agent] A3[代码生成 Agent] A4[通知 Agent] end subgraph L4["第四层：基础设施层"] direction LR DB[(向量数据库)] MQ[消息队列] CACHE[(Redis 缓存)] LLM[LLM 服务] end UI --> ORCH ORCH --> A1 ORCH --> A2 ORCH --> A3 ORCH --> A4 A1 --> DB A1 --> LLM A2 --> CACHE A3 --> LLM A4 --> MQ

这张高层全貌图回答的是"系统由哪些部分构成、怎么连"。但每个子系统的内部细节，需要单独展开。

2.3 子图展开示例：编排层内部

flowchart TB A[接收用户请求] --> B[意图识别 Agent] B --> C{任务类型?} C -->|检索类| D[分发给检索 Agent] C -->|分析类| E[分发给数据分析 Agent] C -->|生成类| F[分发给代码生成 Agent] C -->|复杂多步| G[任务分解 Planner] G --> H[生成子任务列表] H --> I[逐个分发给对应 Agent] I --> J[收集子任务结果] J --> K[结果聚合与验证] K --> L[返回最终结果] D --> L E --> L F --> L style B fill:#4a9eff,color:#fff style G fill:#ff9f43,color:#fff

把 Orchestrator Agent 的内部展开后，可以看到它的职责：意图识别、任务分解、结果聚合。在高层图中，它只是一个"黑盒"；在子图中，它变得清晰。

三、生产级代码实现

分层架构不只是一种画图方法，也是代码的组织方式。下面给出对应的 Python 项目结构：

# 项目目录结构 # # multi_agent/ # ├── layer1_api/ # 用户接入层 # │ ├── routes.py # │ └── middleware.py # ├── layer2_orchestrator/ # 编排层 # │ ├── orchestrator.py # │ ├── intent_router.py # │ └── task_planner.py # ├── layer3_agents/ # Agent 执行层 # │ ├── base.py # │ ├── search_agent.py # │ ├── analyst_agent.py # │ └── code_agent.py # └── layer4_infra/ # 基础设施层 # ├── vector_store.py # ├── message_queue.py # └── llm_client.py import asyncio from abc import ABC, abstractmethod from dataclasses import dataclass from enum import Enum from typing import Any class TaskType(Enum): SEARCH = "search" ANALYSIS = "analysis" CODE_GEN = "code_generation" COMPLEX = "complex" @dataclass class SubTask: task_id: str task_type: TaskType payload: dict[str, Any] dependencies: list[str] = None # ─── 第三层：Agent 执行层 ─── class BaseAgent(ABC): """所有 Agent 的基类，定义统一接口。""" def __init__(self, name: str): self.name = name @abstractmethod async def execute(self, task: SubTask) -> dict[str, Any]: """执行子任务，返回结果。""" ... @abstractmethod def can_handle(self, task_type: TaskType) -> bool: """判断是否能处理该类型任务。""" ... class SearchAgent(BaseAgent): def __init__(self): super().__init__("search_agent") async def execute(self, task: SubTask) -> dict[str, Any]: # 调用向量检索服务 return {"results": [], "status": "ok"} def can_handle(self, task_type: TaskType) -> bool: return task_type == TaskType.SEARCH # ─── 第二层：编排层 ─── class Orchestrator: """编排层：任务分解 + Agent 调度 + 结果聚合。""" def __init__(self, agents: list[BaseAgent]): self._agents = {a.name: a for a in agents} self._agent_by_type: dict[TaskType, BaseAgent] = {} for agent in agents: for t in TaskType: if agent.can_handle(t): self._agent_by_type[t] = agent async def handle_request( self, user_input: str ) -> dict[str, Any]: # Step 1: 意图识别 task_type = await self._classify_intent(user_input) # Step 2: 任务分解（如果是复杂任务） if task_type == TaskType.COMPLEX: subtasks = await self._plan_tasks(user_input) results = await self._execute_parallel(subtasks) return await self._aggregate(results) else: # 简单任务直接分发给对应 Agent agent = self._agent_by_type.get(task_type) if agent is None: return {"error": f"无法处理的任务类型: {task_type}"} task = SubTask("direct", task_type, {"query": user_input}) return await agent.execute(task) async def _classify_intent(self, text: str) -> TaskType: # 通常调用 LLM 做意图分类 return TaskType.SEARCH async def _plan_tasks( self, text: str ) -> list[SubTask]: # 调用 Planner 做任务分解 return [] async def _execute_parallel( self, tasks: list[SubTask] ) -> list[dict]: coros = [] for task in tasks: agent = self._agent_by_type.get(task.task_type) if agent: coros.append(agent.execute(task)) return await asyncio.gather(*coros, return_exceptions=True) async def _aggregate( self, results: list[dict] ) -> dict[str, Any]: return {"results": results, "count": len(results)}

项目目录按四层架构组织，代码的 import 方向也是从上往下：高层的 orchestration 可以 import 底层的 agents，反之不行。

四、边界分析与架构权衡

4.1 分层不等于性能损耗

很多人担心分层会增加延迟——毕竟每多一层就多一次函数调用。但对于分布式 Agent 系统来说，进程间通信（网络 RPC、消息队列）的开销才是大头，几层 Python 函数调用完全可以忽略。

真正需要注意的是：不要让层间接口变成瓶颈。比如 Orchestrator 等待所有 Agent 返回结果时，如果某个 Agent 慢了，整体就慢了。这时候应该用超时 + 降级策略，而不是取消分层。

4.2 分层图的维护成本

架构图最大的痛点是：代码改了，图没更新，渐渐地图就"过时"了。

降低维护成本的几个建议：

• 不要把图放在架构文档里。用 Mermaid 写在代码仓库的 README 或docs/目录中，和代码一起 PR review、一起 merge。
• 从代码生成图。如果你的 Agent 是通过配置注册的，可以从配置自动生成架构图。
• 只保留关键路径。不是每个细节都需要画出来，保留 80% 的流量经过的路径即可。

4.3 何时需要引入"子图"

当一张图上的节点超过 15 个时，就应该拆分子图。当一张图上的连线交叉超过 5 处时，也应该拆。

判断标准很简单：把你的图给一个不了解系统的同事看，30 秒内他能不能说清系统做什么？能就合格，不能就继续拆。

五、总结

画架构图不是为了炫技，是为了传递信息。一张好的架构图应该让读者在 30 秒内理解系统的大致结构，5 分钟内理解关键交互。

分层子图的方法论：

1. 自顶向下：先给全貌，再展开细节。
2. 每层专注一件事：不在一张图里塞两层的细节。
3. 层间接口明确：连线只连到端点，不连到内部组件。
4. 图和代码同仓管理：避免图过时。

我现在的习惯是：每个新项目都在docs/architecture/下建一个 Mermaid 文件，分层画出四层架构。代码评审时架构图也必须一起 review。这个顺手动作，帮我少开了至少 20 次"这个服务到底做什么"的对齐会。

架构不在于画得多细，而在于让看的人一眼懂。希望这个方法能帮你省下几沓 A4 纸的打印费。

下一篇预告：Agent 可解释性，让模型的每一步推理都透明可见。