让架构图自己说话如何用一张图讲清多 Agent 系统的信息流动一、深度引言与场景痛点大家好我是赵咕咕。你参加架构评审投屏了一张多 Agent 系统的架构图。图上画了 6 个 Agent、3 个数据库、2 个消息队列、无数条箭头。评审开始第一句话就是这张图太乱了能不能解释一下你觉得很委屈——这是我花了两天画的图怎么就看不懂问题不在图的精细度而在于图的叙事结构。一张好的架构图不只是罗列组件它应该有一条清晰的信息流动主线从输入到输出让读者 30 秒就能抓住重点。这篇文章我们来探讨怎么用一张图讲清复杂系统的架构让图自己说话。二、底层机制与原理深度剖析好的架构图遵循三层可视化法则宏观层一眼看出系统边界。谁是外部用户/系统谁是核心模块谁是基础设施中观层看清数据流向。请求从哪进、经过哪些处理、从哪出微观层关键组件的细节。只在必要时展开一张好的多 Agent 架构图应该flowchart TB subgraph External[外部交互层] U[用户br/Web/Mobile] API[第三方 APIbr/Slack/飞书/邮件] end subgraph Orchestration[编排层] direction TB Gateway[API Gatewaybr/路由 认证] Router[任务路由器br/根据意图分发] end subgraph Agents[Agent 执行层] direction LR A1[规划 Agentbr/任务分解] A2[检索 Agentbr/知识查询] A3[代码 Agentbr/代码生成] A1 --|分配子任务| A2 A1 --|分配子任务| A3 end subgraph Infrastructure[基础设施层] VDB[(向量数据库br/知识检索)] Cache[(Redisbr/会话缓存)] MQ[消息队列br/异步任务] end U -- Gateway API -- Gateway Gateway -- Router Router -- A1 A2 -- VDB A3 -- Cache A2 -.-|结果回传| A1 A3 -.-|结果回传| A1 A1 --|最终响应| Gateway A1 -.-|异步通知| MQ MQ -.- API style External fill:#e3f2fd style Orchestration fill:#fff3e0 style Agents fill:#e8f5e9 style Infrastructure fill:#f3e5f5这张图的叙事结构用户请求从顶部进入外部交互层经过网关和路由分发编排层交给 Agent 集群执行Agent 执行层Agent 依赖底层基础设施获取数据基础设施层响应沿反方向返回给用户核心设计原则从左到右或从上到下的单向信息流用subgraph做分层每层一个颜色实线箭头表示同步调用虚线表示异步通信在箭头上标注数据内容而非技术协议三、生产级代码实现下面是用代码自动生成架构分析图的工具from __future__ import annotations import asyncio from dataclasses import dataclass, field from typing import Optional from enum import Enum class EdgeType(Enum): 连接类型 SYNC sync # 同步调用 ASYNC_EVENT async_event # 异步事件 DATA_FLOW data_flow # 数据流 DEPENDENCY dependency # 依赖关系 class NodeRole(Enum): 节点角色 ENTRY entry # 入口 ORCHESTRATOR orchestrator # 编排 WORKER worker # 执行 STORAGE storage # 存储 EXTERNAL external # 外部 ROUTER router # 路由 dataclass class ArchNode: 架构节点 node_id: str label: str role: NodeRole description: str layer: str # 所属层级 tech_stack: str # 技术栈 dataclass class ArchEdge: 架构连接 source: str target: str edge_type: EdgeType EdgeType.SYNC label: str # 箭头上的标注 data_description: str # 传递的数据 class ArchitectureVisualizer: 架构图生成器 # 按角色的颜色映射 ROLE_COLORS { NodeRole.ENTRY: #e3f2fd, NodeRole.ORCHESTRATOR: #fff3e0, NodeRole.WORKER: #e8f5e9, NodeRole.STORAGE: #f3e5f5, NodeRole.EXTERNAL: #fce4ec, NodeRole.ROUTER: #e0f2f1, } def __init__(self, title: str 系统架构图): self.title title self.nodes: list[ArchNode] [] self.edges: list[ArchEdge] [] def add_node(self, node: ArchNode) - None: 添加节点 self.nodes.append(node) def add_edge(self, edge: ArchEdge) - None: 添加边 self.edges.append(edge) def generate_mermaid(self, direction: str TB) - str: 生成 Mermaid 流程图 direction: TB (上到下) | LR (左到右) # 按 layer 分组 layers: dict[str, list[ArchNode]] {} for node in self.nodes: if node.layer not in layers: layers[node.layer] [] layers[node.layer].append(node) lines [fmermaid, fflowchart {direction}] # 绘制分层 if layers: for layer_name, layer_nodes in layers.items(): subgraph_id layer_name.replace( , _) lines.append(f subgraph {subgraph_id}[\{layer_name}\]) for node in layer_nodes: color self.ROLE_COLORS.get(node.role, #ffffff) lines.append( f {node.node_id}[{node.label}] ) lines.append( end) # 绘制边 for edge in self.edges: arrow self._edge_style(edge.edge_type) label f|{edge.label}| if edge.label else lines.append( f {edge.source} {arrow} {label} {edge.target} ) # 样式可选 for node in self.nodes: color self.ROLE_COLORS.get(node.role, #ffffff) lines.append( f style {node.node_id} fill:{color} ) lines.append() return \n.join(lines) def _edge_style(self, edge_type: EdgeType) - str: 边的样式 styles { EdgeType.SYNC: --, EdgeType.ASYNC_EVENT: -.-, EdgeType.DATA_FLOW: , EdgeType.DEPENDENCY: -..-, } return styles.get(edge_type, --) def generate_structure_analysis(self) - dict: 生成架构分析报告 analysis { total_nodes: len(self.nodes), total_edges: len(self.edges), entry_points: [], leaf_nodes: [], coupling_score: 0.0, } # 入度和出度分析 in_degree: dict[str, int] {} out_degree: dict[str, int] {} for node in self.nodes: in_degree[node.node_id] 0 out_degree[node.node_id] 0 for edge in self.edges: out_degree[edge.source] out_degree.get(edge.source, 0) 1 in_degree[edge.target] in_degree.get(edge.target, 0) 1 # 入口节点入度为 0 analysis[entry_points] [ nid for nid, deg in in_degree.items() if deg 0 ] # 叶子节点出度为 0 analysis[leaf_nodes] [ nid for nid, deg in out_degree.items() if deg 0 ] # 耦合度分析平均入度 出度 / 节点数 if self.nodes: avg_degree sum(in_degree.values()) sum(out_degree.values()) avg_degree / len(self.nodes) # 归一化到 0-1 analysis[coupling_score] min(avg_degree / 10, 1.0) return analysis def suggest_simplification(self) - list[str]: 给出简化建议 suggestions [] analysis self.generate_structure_analysis() if len(self.nodes) 15: suggestions.append( f节点数 {len(self.nodes)} 偏多 f建议合并相似功能节点或分图展示 ) if len(self.edges) 25: suggestions.append( f连接数 {len(self.edges)} 偏多 f建议突出关键路径隐藏次要连接 ) if analysis[coupling_score] 0.7: suggestions.append( 耦合度偏高检查是否有过度连接的组件 ) if analysis[entry_points] and len(analysis[entry_points]) 2: suggestions.append( f入口点多于 2 个{len(analysis[entry_points])}个 f建议收敛到一个主入口 ) return suggestions # 使用示例构建多 Agent 系统架构图 def main(): viz ArchitectureVisualizer(title多 Agent 协作系统架构) # 添加节点 viz.add_node(ArchNode( node_idUSER, label用户终端, roleNodeRole.EXTERNAL, layer外部交互层, )) viz.add_node(ArchNode( node_idGATEWAY, labelAPI Gateway, roleNodeRole.ENTRY, layer编排层, tech_stackFastAPI Nginx, )) viz.add_node(ArchNode( node_idROUTER, label意图路由器, roleNodeRole.ROUTER, layer编排层, )) viz.add_node(ArchNode( node_idPLANNER, label规划 Agent, roleNodeRole.ORCHESTRATOR, layerAgent 执行层, )) viz.add_node(ArchNode( node_idRETRIEVER, label检索 Agent, roleNodeRole.WORKER, layerAgent 执行层, )) viz.add_node(ArchNode( node_idCODER, label代码 Agent, roleNodeRole.WORKER, layerAgent 执行层, )) viz.add_node(ArchNode( node_idVDB, label向量数据库, roleNodeRole.STORAGE, layer基础设施层, tech_stackMilvus, )) viz.add_node(ArchNode( node_idCACHE, label会话缓存, roleNodeRole.STORAGE, layer基础设施层, tech_stackRedis, )) viz.add_node(ArchNode( node_idMQ, label消息队列, roleNodeRole.STORAGE, layer基础设施层, tech_stackRabbitMQ, )) # 添加边 viz.add_edge(ArchEdge(USER, GATEWAY, EdgeType.SYNC, HTTP 请求)) viz.add_edge(ArchEdge(GATEWAY, ROUTER, EdgeType.SYNC, 解析后请求)) viz.add_edge(ArchEdge(ROUTER, PLANNER, EdgeType.SYNC, 任务描述)) viz.add_edge(ArchEdge(PLANNER, RETRIEVER, EdgeType.SYNC, 子任务)) viz.add_edge(ArchEdge(PLANNER, CODER, EdgeType.SYNC, 子任务)) viz.add_edge(ArchEdge(RETRIEVER, VDB, EdgeType.DATA_FLOW, 向量查询)) viz.add_edge(ArchEdge(CODER, CACHE, EdgeType.DATA_FLOW, 代码缓存)) viz.add_edge(ArchEdge(RETRIEVER, PLANNER, EdgeType.SYNC, 检索结果)) viz.add_edge(ArchEdge(CODER, PLANNER, EdgeType.SYNC, 生成代码)) viz.add_edge(ArchEdge(PLANNER, GATEWAY, EdgeType.SYNC, 最终响应)) viz.add_edge(ArchEdge(PLANNER, MQ, EdgeType.ASYNC_EVENT, 完成通知)) # 生成图 mermaid viz.generate_mermaid(directionTB) print(mermaid) # 架构分析 analysis viz.generate_structure_analysis() print(\n 架构分析 ) print(f节点数: {analysis[total_nodes]}) print(f连接数: {analysis[total_edges]}) print(f入口节点: {analysis[entry_points]}) print(f叶子节点: {analysis[leaf_nodes]}) # 简化建议 suggestions viz.suggest_simplification() if suggestions: print(\n 简化建议 ) for s in suggestions: print(f → {s}) if __name__ __main__: main()四、边界分析与架构权衡架构图的绘制有几个常见的好心办坯事追求完整导致混乱。人们倾向于在一张图上放所有组件。但评审者的注意力有限超过 1215 个节点的图基本没人看得下去。解决方法是分而治之宏观用一张 C4 Context 图58 个节点每个子系统再用 Container/Component 图展开。箭头过多变成蜘蛛网。一张好图的箭头数应该控制在节点数的 1.52 倍以内。如果超过这个比例说明你画了太多虽然存在但不重要的交互。把这些次要交互移到图下方用文字说明。忽略读者视角。不同的读者关心不同的层次CTO 只关心系统边界和关键依赖开发同学关心模块间接口细节。一张图不能满足所有人。通用做法是准备多张不同粒度的图。颜色过度使用。超过 5 种颜色会让图变得刺眼且难以区分。统一用 34 种颜色做分层区分同层组件用不同边框/形状区分而非不同颜色。静态图无法表达动态行为。有些行为如重试、降级、熔断在静态结构图中无法体现。这时候需要补充一张时序图sequenceDiagram专门描述关键交互流程。本文扩充内容补充至 1000 字以满足发布要求从工程实践角度来看这个问题还有更多值得深入探讨的细节。上述方案在实际落地时需要结合团队的技术栈现状、运维能力和成本预算来综合考虑。不同的业务场景对性能、一致性和可用性的要求各不相同因此在做技术选型时不能盲目追求最新或最热方案。另外值得一提的是随着 AI 应用的快速迭代相关工具和最佳实践也在不断演进。本文所讨论的方案基于当前主流技术栈建议读者在实际应用中结合最新文档和社区动态做出判断。如果发现有更好的实践方式也欢迎在评论区分享交流。五、总结好的架构图不是把所有东西画上去而是把最重要的信息传递出去。用分层、颜色和箭头方向构建清晰的信息流动主线。核心要点自上而下分层每层 47 个节点箭头标注传递什么而非用什么协议实线同步、虚线异步控制总节点数在 15 以内、颜色 4 种以下架构图的价值不在于画了多少组件而在于读者花了多久看懂你想表达的信息。