复杂系统拆解为子图如何用分层架构图让分布式 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_exceptionsTrue) 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 分钟内理解关键交互。分层子图的方法论自顶向下先给全貌再展开细节。每层专注一件事不在一张图里塞两层的细节。层间接口明确连线只连到端点不连到内部组件。图和代码同仓管理避免图过时。我现在的习惯是每个新项目都在docs/architecture/下建一个 Mermaid 文件分层画出四层架构。代码评审时架构图也必须一起 review。这个顺手动作帮我少开了至少 20 次这个服务到底做什么的对齐会。架构不在于画得多细而在于让看的人一眼懂。希望这个方法能帮你省下几沓 A4 纸的打印费。下一篇预告Agent 可解释性让模型的每一步推理都透明可见。