尧图网络科技YAOTU DIGITAL 获取报价
获取报价
首页 / 资讯中心 / 文章详情

多Agent系统架构实战:从超级Agent到能力平台的设计与落地

发布时间:2026/9/28 15:59:03

资讯中心
01
ARTICLE

多Agent系统架构实战:从超级Agent到能力平台的设计与落地

多Agent系统架构实战:从超级Agent到能力平台的设计与落地
1. 从“超级 Agent”到能力平台多 Agent 系统的架构取舍1.1 为什么单 Agent 走到了天花板过去一年我经手过不下十个 Agent 项目从最早的“一个 Prompt 打天下”到后来动辄几十个工具挂在一个 Agent 身上踩的坑基本能写一本错题集。最开始大家的思路很朴素既然大模型能力这么强那就把所有工具、所有知识、所有流程都塞给一个 Agent让它自己判断该调什么、该怎么调。这个思路在 Demo 阶段非常性感一个入口、一套记忆、一条链路演示效果拉满。但真到了生产环境问题就藏不住了。最典型的是上下文污染一个 Agent 同时挂着数据库查询、文件操作、外部 API 调用、代码执行等十几类工具光是工具描述就吃掉几千 token模型在长对话里开始“记混”——把 A 场景的约束带到 B 场景把不该暴露的内部字段写进回复。我实测过一个客服场景单 Agent 挂 12 个工具时工具选择准确率从 92% 掉到 71%而且错误分布毫无规律排查起来极其痛苦。第二个问题是能力耦合导致的迭代僵局。当所有逻辑都在一个 Agent 里你想优化“退款流程”的判断逻辑结果发现改动会影响到“订单查询”的意图识别因为它们在同一个 Prompt 里共享上下文。团队里三个人同时改一个 Agent 的配置合并冲突比代码还难解。这时候你才意识到Agent 不是越“超级”越好它需要被拆。第三个问题是可观测性缺失。单 Agent 的执行链路是一条黑盒你只能看到输入和最终输出中间它调了哪些工具、为什么这么调、哪一步耗时最长全靠猜。线上出问题时你连“它到底在哪一步开始跑偏”都定位不了。这三个问题叠加基本宣告了“超级 Agent”路线在生产环境的天花板。1.2 多 Agent 不是简单拆分而是能力平台化很多人一听多 Agent第一反应是“把一个大 Agent 拆成几个小 Agent各管一摊”。这个理解只对了一半。拆分只是手段真正的目标是把 Agent 从“一个会调工具的程序”升级成能力平台——每个 Agent 是一个独立的能力单元有明确的职责边界、独立的记忆空间、可观测的执行链路并且能通过统一的调度层被编排和复用。我习惯用一个类比单 Agent 像一家什么都卖的杂货铺多 Agent 像一条商业街。杂货铺里找东西全靠老板记性商业街里每家店招牌清晰你按需进店。但商业街要运转好需要三个基础设施统一的入口指引用户知道该进哪家店、标准化的店面规范每家店的服务流程一致、独立的库存管理各店记忆不串味。对应到技术架构上就是 Unified Invocation Gateway、Agent Runtime 和独立记忆层。这个转变带来的最大价值是能力复用。比如“代码审查”这个能力在研发助手场景里要用在 CI 流水线里也要用。单 Agent 时代你得复制两份 Prompt 和工具配置能力平台化之后它就是一个独立的 Agent 服务谁需要谁调用改一处全局生效。这才是多 Agent 架构真正的杠杆点。1.3 架构取舍的核心矛盾控制力与灵活性的平衡多 Agent 架构没有银弹所有设计决策本质上都是在控制力和灵活性之间做取舍。控制力强意味着流程确定、可预测、易调试但面对新场景时扩展成本高灵活性高意味着 Agent 能自主决策、适应性强但行为不可预测、排查困难。我见过两种极端。一种是全编排式所有 Agent 的调用顺序、输入输出格式全部由代码写死Agent 只负责在指定节点做指定的事。这种架构稳如老狗但基本退化成传统工作流Agent 的智能性被浪费了。另一种是全自主式给一个“总控 Agent”让它自己决定调哪个子 Agent、传什么参数。这种架构在 Demo 里很惊艳但生产环境里总控 Agent 的决策错误会级联放大而且你很难向业务方解释“为什么这次它选了 A 方案而不是 B 方案”。我的经验是核心链路用编排边缘场景用自主。比如支付、退款这类涉及资金的操作必须走确定性编排每一步都可审计而“帮我找一份资料并总结”这类任务可以交给自主调度允许 Agent 自己规划路径。这个边界划清楚架构就不会跑偏。2. 核心组件拆解Gateway、Runtime 与记忆层2.1 Unified Invocation Gateway统一入口不是简单路由Unified Invocation Gateway统一调用网关是多 Agent 系统的门面但它绝不只是“根据请求转发到对应 Agent”这么简单。我把它定位成请求的标准化层和治理层核心职责有四块协议适配、身份与权限、流量治理、可观测埋点。协议适配解决的是“不同客户端说不同方言”的问题。Web 端可能发 JSON移动端可能发 gRPC内部服务可能直接调 SDK。Gateway 要把这些统一成内部 Agent Runtime 能理解的调用格式。我通常会在这一层定义一个标准调用信封包含intent意图标识、payload业务参数、context会话上下文、constraints约束条件比如超时、预算、敏感级别。这个信封设计得好后面所有 Agent 的输入输出都规整。身份与权限是容易被忽视但极其关键的一环。多 Agent 系统里不是每个 Agent 都能访问所有工具和数据。比如“财务 Agent”能查账单“客服 Agent”就不能。Gateway 要在请求进入时就完成身份校验和权限裁剪把不该给这个 Agent 看的工具直接从可用列表里摘掉。这比在 Agent 内部做判断要安全得多因为 Agent 的 Prompt 可能被注入攻击绕过。流量治理包括限流、熔断、重试、降级。Agent 调用大模型有延迟和成本不加治理的话一个异常流量就能把整个系统拖垮。我一般会给每个 Agent 配置独立的并发上限和超时时间超时后走降级逻辑比如返回缓存结果或转人工。可观测埋点则是把每次调用的trace_id、agent_id、tool_calls、latency、token_usage全部记录下来这是后面排查问题的命根子。2.2 Agent Runtime执行引擎的隔离与调度Agent Runtime 是每个 Agent 的“运行容器”它决定了 Agent 以什么形态存在、如何被调度、如何与其他 Agent 通信。这里最大的架构取舍是进程隔离还是线程隔离。进程隔离每个 Agent 独立进程或容器的好处是故障不扩散、资源可独立限制、依赖不冲突。坏处是启动慢、通信开销大、内存占用高。线程隔离所有 Agent 跑在同一进程的不同线程则相反轻量但一个 Agent 崩溃可能拖垮整个进程。我的选择是核心 Agent 用进程隔离辅助 Agent 用线程隔离。比如订单处理 Agent 必须独立进程因为它涉及资金而“格式化输出”这种辅助 Agent 可以共享线程池。Runtime 的另一个核心是工具调用的沙箱化。Agent 要执行代码、访问文件、调外部 API这些操作必须在受控环境里进行。我通常会给每个 Agent 配一个工具白名单白名单外的调用直接拒绝。代码执行类工具跑在独立沙箱里限制 CPU、内存、网络访问。文件操作限制在指定目录禁止路径穿越。这些约束不是不信任模型而是安全底线。调度策略上我倾向于两级调度Gateway 做粗粒度路由根据 intent 找到候选 Agent 集合Runtime 做细粒度调度在候选集合里根据负载、优先级、亲和性选一个实例。这样既保证了路由的确定性又保留了负载均衡的灵活性。2.3 记忆层多 Agent 的记忆隔离与共享记忆是多 Agent 系统里最容易被低估的组件。单 Agent 时代记忆就是一段对话历史简单拼接就行。多 Agent 时代记忆要解决三个问题隔离A Agent 的记忆不能污染 B Agent、共享某些上下文需要跨 Agent 传递、持久化会话结束后记忆要能存下来供后续使用。我的做法是把记忆分成三层。会话记忆是当前对话的短期上下文每个 Agent 独立维护不共享。任务记忆是当前任务相关的上下文比如用户上传的文件、中间计算结果在参与同一任务的 Agent 之间共享。长期记忆是跨会话的知识比如用户偏好、历史工单存在向量数据库里按需检索注入。这里有个坑共享记忆的读写权限要严格控制。我见过一个案例客服 Agent 把用户的内部备注写进了共享记忆结果销售 Agent 在后续对话里把备注内容泄露给了用户。所以共享记忆的写入必须经过脱敏和审核敏感字段打标后才能进入共享层。3. 实操落地从零搭建一个多 Agent 能力平台3.1 环境准备与基础依赖假设我们要搭建一个最小可用的多 Agent 平台包含三个 Agent检索 Agent负责查资料、总结 Agent负责归纳、审核 Agent负责合规检查。技术栈选 Python FastAPI Redis PostgreSQL 向量数据库这里用 Chroma 做示例。先装依赖pip install fastapi uvicorn redis psycopg2-binary chromadb openai pydantic目录结构我习惯这样组织agent-platform/ ├── gateway/ │ ├── main.py # 网关入口 │ ├── router.py # 路由逻辑 │ └── middleware.py # 鉴权、限流、埋点 ├── agents/ │ ├── retriever/ │ │ ├── runtime.py # Agent 执行逻辑 │ │ └── config.yaml # 工具白名单、模型参数 │ ├── summarizer/ │ └── reviewer/ ├── memory/ │ ├── session.py # 会话记忆 │ ├── shared.py # 共享记忆 │ └── longterm.py # 长期记忆 └── common/ ├── envelope.py # 标准调用信封 └── observability.py # 埋点这个结构的好处是每个 Agent 完全独立可以单独部署、单独升级。Gateway 只依赖common里的信封定义不依赖具体 Agent 实现做到了解耦。3.2 标准调用信封的设计与实现信封是整个系统的“通用语言”设计好坏直接影响后续扩展成本。我的信封定义如下from pydantic import BaseModel, Field from typing import Any, Optional from datetime import datetime class CallEnvelope(BaseModel): trace_id: str Field(..., description全链路追踪ID) intent: str Field(..., description意图标识如 search_and_summarize) payload: dict Field(default_factorydict, description业务参数) context: dict Field(default_factorydict, description会话上下文) constraints: dict Field(default_factorydict, description约束条件) caller: str Field(..., description调用方标识) timestamp: datetime Field(default_factorydatetime.utcnow) class ResponseEnvelope(BaseModel): trace_id: str status: str # success / partial / failed data: Any agent_chain: list # 经过的 Agent 链路 token_usage: dict latency_ms: intintent是路由的核心依据我建议用命名空间风格比如search.web、search.internal、summarize.text、review.compliance。这样 Gateway 可以用前缀匹配做粗路由用完整 intent 做精路由。constraints里我通常会放timeout_ms、max_tokens、sensitivity敏感级别Runtime 根据这些约束决定用哪个模型、要不要走人工审核。3.3 Gateway 的路由与治理实现Gateway 的核心是路由表。我用一个 YAML 配置来定义 intent 到 Agent 的映射routes: - intent: search.* agents: [retriever] strategy: round_robin - intent: summarize.* agents: [summarizer] strategy: least_loaded - intent: review.* agents: [reviewer] strategy: consistent_hash - intent: search_and_summarize agents: [retriever, summarizer] strategy: sequential pipeline: truepipeline: true表示这是一个编排链路Gateway 会按顺序调用 retriever 和 summarizer把前者的输出作为后者的输入。这种设计让简单编排不需要写代码改配置就行。限流我用 Redis 做令牌桶每个 Agent 独立配额import redis import time class TokenBucket: def __init__(self, client: redis.Redis, key: str, rate: int, capacity: int): self.client client self.key key self.rate rate self.capacity capacity def allow(self) - bool: now time.time() pipe self.client.pipeline() pipe.hgetall(self.key) data pipe.execute()[0] tokens float(data.get(btokens, self.capacity)) last float(data.get(blast, now)) tokens min(self.capacity, tokens (now - last) * self.rate) if tokens 1: return False tokens - 1 self.client.hset(self.key, mapping{btokens: tokens, blast: now}) return True这个实现有个细节last时间戳要一起更新否则令牌补充会算错。我踩过这个坑一开始只更新 tokens 不更新 last结果限流形同虚设。3.4 Agent Runtime 的工具白名单与沙箱每个 Agent 的config.yaml里定义工具白名单agent_id: retriever model: gpt-4o-mini tools: - name: web_search enabled: true timeout_ms: 5000 - name: vector_search enabled: true timeout_ms: 2000 - name: file_read enabled: false # 检索 Agent 不需要读文件 max_tokens: 4096 temperature: 0.3Runtime 加载配置后只把enabled: true的工具注册到 Agent 的可用工具列表里。这样即使 Prompt 被注入攻击要求调用file_readRuntime 也会直接拒绝因为工具根本不在注册表里。代码执行类工具的沙箱我用subprocess 资源限制实现import subprocess import resource def run_in_sandbox(code: str, timeout: int 5) - str: def limit_resources(): resource.setrlimit(resource.RLIMIT_CPU, (timeout, timeout)) resource.setrlimit(resource.RLIMIT_AS, (256 * 1024 * 1024, 256 * 1024 * 1024)) resource.setrlimit(resource.RLIMIT_NOFILE, (10, 10)) try: result subprocess.run( [python, -c, code], capture_outputTrue, textTrue, timeouttimeout, preexec_fnlimit_resources ) return result.stdout or result.stderr except subprocess.TimeoutExpired: return 执行超时RLIMIT_AS限制内存 256MBRLIMIT_CPU限制 CPU 时间RLIMIT_NOFILE限制文件描述符数量。这三个限制能挡住大部分资源耗尽型攻击。注意preexec_fn在有多线程的进程里可能不安全生产环境建议用独立容器做沙箱。3.5 记忆层的读写与隔离实现会话记忆我用 Redis 的 Hash 结构key 是session:{session_id}:{agent_id}保证每个 Agent 独立class SessionMemory: def __init__(self, client: redis.Redis, ttl: int 3600): self.client client self.ttl ttl def append(self, session_id: str, agent_id: str, role: str, content: str): key fsession:{session_id}:{agent_id} self.client.rpush(key, f{role}:{content}) self.client.expire(key, self.ttl) def get_history(self, session_id: str, agent_id: str, limit: int 20) - list: key fsession:{session_id}:{agent_id} return self.client.lrange(key, -limit, -1)共享记忆用 PostgreSQL写入前做脱敏import re SENSITIVE_PATTERNS [ (r\d{11}, [PHONE]), # 手机号 (r\d{18}, [ID_CARD]), # 身份证 (r[\w\.-][\w\.-], [EMAIL]), # 邮箱 ] def sanitize(text: str) - str: for pattern, replacement in SENSITIVE_PATTERNS: text re.sub(pattern, replacement, text) return text def write_shared(task_id: str, agent_id: str, content: str): safe_content sanitize(content) # 写入 PostgreSQL ...长期记忆用向量库检索时按agent_id过滤避免跨 Agent 污染def retrieve_longterm(query: str, agent_id: str, top_k: int 5): results collection.query( query_texts[query], n_resultstop_k, where{agent_id: agent_id} ) return results4. 常见问题与排查技巧实录4.1 Agent 之间“踢皮球”怎么破多 Agent 系统最烦的问题之一是任务在 Agent 之间来回传递谁也不真正执行。我遇到过检索 Agent 把任务转给总结 Agent总结 Agent 觉得信息不够又转回检索 Agent循环三四次后超时。根因通常是职责边界模糊加上缺少终止条件。排查思路先看agent_chain埋点确认循环发生在哪两个 Agent 之间。然后检查这两个 Agent 的 Prompt看是否都认为“这不是我的活”。解决方法是给每个 Agent 加明确的拒绝条件和兜底行为。比如检索 Agent 的 Prompt 里写死“如果查询涉及内部数据直接返回NEED_INTERNAL_ACCESS不要转给其他 Agent。”总结 Agent 写死“如果输入少于 100 字直接返回INSUFFICIENT_INPUT不要请求更多信息。”另外在 Gateway 层加最大跳转次数限制比如max_hops: 3超过就强制返回当前最佳结果并标记partial。这个兜底能防止无限循环拖垮系统。4.2 工具调用参数错误的快速定位Agent 调工具时参数格式错误是高频问题。比如web_search期望query是字符串Agent 传了个对象。排查时不要只看最终报错要看模型原始输出。我通常会在 Runtime 里记录每次工具调用的原始参数和校验后的参数def call_tool(tool_name: str, raw_args: dict, schema: dict): try: validated validate(raw_args, schema) except ValidationError as e: logger.warning(ftool{tool_name} raw_args{raw_args} error{e}) return {error: 参数格式错误, hint: schema} return execute(tool_name, validated)把raw_args和schema一起记下来对比就能看出是模型理解错了 schema还是 schema 本身描述不清。我踩过的坑是 schema 里type: string但没写description模型不知道这个字段该填什么就瞎填。补上 description 后准确率明显提升。4.3 记忆串味的排查与修复记忆串味表现为A 用户的信息出现在 B 用户的回复里或者 A Agent 的内部状态被 B Agent 引用。排查第一步是确认 key 设计。我见过最离谱的 bug 是会话记忆的 key 只用了session_id没带agent_id导致同一会话里所有 Agent 共享一份历史。修复就是加上agent_id维度。第二步是检查共享记忆的写入时机。有些实现会在 Agent 每次输出后自动写入共享记忆结果把中间推理过程也写进去了。正确做法是只写入经过审核的最终结果中间过程留在会话记忆里。第三步是向量检索的过滤条件。如果长期记忆检索时忘了加agent_id过滤就会召回其他 Agent 的知识。这个在测试环境不容易发现因为测试数据少上线后数据一多就暴露。建议在向量库层面做命名空间隔离每个 Agent 一个 collection从物理上杜绝串味。4.4 性能瓶颈的定位与优化多 Agent 系统的延迟是累加的三个 Agent 串行调用每个 2 秒总延迟就是 6 秒起步。优化前先定位瓶颈看latency_ms埋点里每个 Agent 的耗时占比。常见瓶颈有三类模型推理慢换更小的模型或者对非核心 Agent 用流式输出降低首字延迟。我实测过把总结 Agent 从 GPT-4 换成 GPT-4o-mini延迟从 3.2 秒降到 1.1 秒质量下降在可接受范围内。工具调用慢外部 API 加缓存相同参数短时间内直接返回缓存结果。向量检索加索引Chroma 默认是暴力检索数据量上万后要换 HNSW 索引。串行改并行如果两个 Agent 之间没有依赖关系改成并行调用。比如检索和审核可以同时进行最后合并结果。Gateway 的pipeline配置里加parallel: true即可。问题现象可能原因排查动作解决方案Agent 循环调用职责边界模糊查 agent_chain 埋点加拒绝条件 max_hops工具参数错误schema 描述不清对比 raw_args 和 schema补全 description记忆串味key 缺 agent_id检查 Redis key 设计加 agent_id 维度延迟过高串行调用看 latency_ms 分布并行化 换小模型工具调用被拒白名单未配置查 config.yaml按需开启工具4.5 上线前的检查清单多 Agent 系统上线前我必查这几项每个 Agent 的工具白名单是否最小化不该有的工具坚决不开、共享记忆写入是否脱敏跑一遍敏感数据测试、Gateway 限流是否生效压测验证、埋点是否覆盖全链路trace_id 能否串起所有 Agent、降级逻辑是否可用手动触发验证。这五项过了基本不会出大事故。5. 架构演进从能力平台到生态5.1 什么时候该从多 Agent 升级到平台化多 Agent 和 Agent 平台的区别在于是否支持动态注册和发现。多 Agent 系统里Agent 列表是写死在配置里的平台化之后Agent 可以自己注册到服务发现中心Gateway 动态感知。触发平台化升级的信号通常有三个Agent 数量超过 10 个、有多个团队独立开发 Agent、需要按租户隔离 Agent 能力。我建议不要过早平台化。Agent 少于 5 个时写死配置最简单直接引入服务发现反而增加复杂度。等到配置维护成本明显上升再考虑平台化。5.2 能力复用的两种模式平台化之后能力复用有两种模式Agent 级复用和工具级复用。Agent 级复用是把整个 Agent 作为服务暴露出去其他系统通过 Gateway 调用。工具级复用是把 Agent 内部的某个工具抽出来做成独立的工具服务多个 Agent 共享。我的经验是先做工具级复用再做 Agent 级复用。因为工具级复用的粒度更细改造成本低而且能快速看到效果。比如把“敏感词检测”抽成独立工具服务所有 Agent 都能调不用每个 Agent 都实现一遍。Agent 级复用适合跨团队场景但需要更完善的版本管理和 SLA 保障。5.3 多 Agent 系统的安全边界最后聊聊安全。多 Agent 系统扩大了攻击面每个 Agent 是一个入口每个工具是一个出口Agent 之间的通信是新的通道。我的安全原则是零信任不信任任何 Agent 的输出所有跨 Agent 的数据传递都要经过校验和脱敏不信任任何工具调用所有调用都要经过白名单和沙箱不信任任何记忆写入所有写入都要经过审核。具体措施包括Gateway 层做统一的身份认证和权限校验Runtime 层做工具白名单和沙箱隔离记忆层做脱敏和访问控制全链路做审计日志任何异常调用都能追溯。这些措施会增加一些开发成本但比起数据泄露的代价完全值得。我在实际项目里踩过最深的坑是早期为了快速上线把工具白名单做成了“默认全开按需关闭”。结果一个测试用的文件读取工具忘了关上线后被 Prompt 注入攻击利用读到了服务器上的配置文件。从那以后我所有项目都改成“默认全关按需开启”这个原则再也没破过。
02
RELATED NEWS

相关资讯

更多网站建设与数字化升级内容

03
WHY YAOTU

想打造同款高转化官网?

懂行业、懂生意,从建站到增长一站式陪跑

◈

场景化定制

不做模板站,围绕你的业务场景量身设计,小众不撞款。

◐

营销型架构

以转化目标组织内容与路径,让官网真正带来询盘。

▲

全周期服务

设计、开发、运营、运维一体,上线只是开始。

免费获取你的建站方案

留下需求,专属顾问 24 小时内为你输出方案建议。