如果你手上已经有两三个能跑的 LLM 智能体agent并且正在犹豫要不要把它们推向生产环境我猜你很快就会撞上同一个问题agent 本身跑得好好的但把它们放在一起就像一间办公室里每个人都想当负责人谁也没在管报销、排班和对外接口。我们团队这半年做的内部项目代号就叫 agency-agents。它不是一个新的大模型框架而是一层位于多个 agent 之上的中介层统一接收外部请求负责给任务派发合适的 agent、跟踪执行过程、限制工具权限、并且在出错时兜底。这篇文章我把我从命名、架构设计、核心代码骨架到上线第一周踩坑的全过程都记录下来。如果你想给自己的多智能体系统加一层治理和编排这篇应该能帮你省掉不少试错时间。1. 先从名字说起为什么叫 agency不叫 orchestrator很多团队管这类组件叫 orchestrator编排器听起来很工程化。但我们在命名阶段争论了几轮最后坚持叫 agency是有原因的。翻译成中文agency 是代理机构、中介Orchestrator 更像一个调度指挥中心。我希望团队成员每次看到这个项目名的时候都能想起它真正的定位它不是指挥官它是中介。1.1 agent 是被代理的那个人而 agency 是替所有人办事的中介大模型语境下的 agent本质上是替人去使用工具和对话的代理。那 agent 多了谁来帮 agent 对接资源、分配任务、审核权限这就是更上一层的代理的代理。你想象一个真实的经纪人公司agent 是业务员agency 是那家公司。业务员出去见客户之前公司得先帮他安排好工位、分配好客户线索、规定他能动用多少钱、还要在客户投诉的时候有人接电话。我们需要的正是这样一个角色。所以 agency-agents 这个项目名的意思就是管理 agent 的那个中介层。它不产生新的 agent它只是把 agent 变成一种可被统一调度、审计、计费的资源。这个定位说出来很轻但真正设计起来牵扯的东西非常多。1.2 和 LangChain、CrewAI 这类框架到底什么关系好多人一听多智能体编排第一反应是LangChain 不是有 Agent Executor 吗CrewAI 不是专门做角色扮演协作的吗我们为什么还要自己写一层我用一句话说清楚LangChain、CrewAI 解决的是怎么构建一个 agent 出来跑通任务agency-agents 解决的是多个 agent 上线之后怎么管。前者是技能后者是治理。打个比方LangChain 就是教你开车一脚油门能跑。而我们的 agency 解决的是车队管理哪辆车出勤、走哪条路线、每辆车能加多少油、出了事故怎么处理。你再会开车管一个车队还是另一套逻辑。实际架构里我们跑过 LangChain 的 agent、也跑过纯自写的 prompt 循环。agency-agents 根本不关心你底下跑的是什么框架它只认 agent 暴露出来的能力描述和调用协议。这样做的最大好处是以后团队里有人想换框架只要协议不变上层完全无感。1.3 我们明确不做的事比做了的事更重要项目初期大家很容易把什么东西都往里塞。我拉了几个硬边界写进了项目 README 的第一屏。不重新实现 LLM 调用封装。OpenAI、Anthropic 的 SDK 已经够好我们不需要再包一层。所有 agent 内部自己管模型调用。不做训练和微调。agent 的能力是跑出来的不是 our agency 能教会的。不绑定特定云厂商。部署在哪都可以只要走标准 HTTP 或消息队列。不做具体的业务功能。查订单、写文案、分析报表这些全部是 agent 自己的事agency 只负责把它送到对的地方、确保它能被信任。边界划清楚以后后面所有的技术决策都变得好做了。遇到一个需求先问这属于调车还是开车是后者就丢给 agent 层去搞。2. 整体架构把原来的单体 agent 拆成四层我们最初的版本其实是一个巨大的单体 agent什么工具都往里塞system prompt 写了几千字效果一开始还行后面越加功能越混乱。后来我们把一个单体 agent 按职责拆成了四层对应到 agency-agents 里就形成了这套架构接入层、路由层、执行层、治理层。下面我逐层讲清楚每一层具体干什么。2.1 接入层把人话请求变成标准意图接入层是对外窗口。用户或者上游系统发过来一句话接入层要做的第一件事是把这句话变成结构化的请求对象。这里不是简单的意图识别而是意图解析 参数抽取。举个例子。用户说帮我查一下昨天那笔 988 的订单退款到账了没。我们在接入层用一个轻量级模型甚至有时直接规则优先把这句话标准化成{ intent: order_refund_status, entities: {order_amount: 988, time: yesterday}, trace_id: req_7f3a2c9e }为什么不用大模型直接做这件事因为意图标准化是高频动作每来一个请求都要跑一次。用大模型做延迟高、费用高、结果还不稳定。我们的做法是先用规则和少量样本训练的意图分类模型置信度低于阈值才升级到大模型兜底。这样 90% 的请求走的是轻量链路平均耗时控制在 50ms 以内。2.2 路由层三问法决定任务去哪个 agent请求标准化之后进入路由层。这是我们这套系统里最核心的地方。路由层做三个判断我管它叫三问法。第一问谁会做能力匹配我们为每个 agent 注册了一张能力卡声明它能处理哪些意图。路由层拿到 intent在能力卡里做匹配。这一步可以是精确匹配也可以是语义相似度匹配。我们用的是 embedding 相似度加规则权重效果比纯关键词匹配好很多。第二问他该不该做权限与成本能力匹配上了不代表就让人家干。比如删除用户数据这个意图可能客服 agent 也会但权限上明确规定只有 admin agent 能执行。成本上如果一个 intent 既可以被便宜的快模型处理又可以交给贵的高阶 agent路由层会优先选便宜的除非用户明确要高阶分析。这个成本感知的路由策略一个月下来能帮我们省下 30% 以上的 token 开销。第三问他现在能不能做负载与熔断被选中的 agent 如果正在重试风暴里挣扎或者下游依赖已经熔断路由层会把请求引导到备用 agent 或者直接进入降级队列。别小看这一问没有它链路雪崩几乎不可避免。这三问跑完路由层输出一个目标 agent 引用 路由理由这个信息会被完整记录到审计日志后面排查问题全靠它。2.3 执行层从多轮闲聊变成任务图编排执行层的主要工作是任务分解和状态追踪。我发现很多人对 agent 编排有个误解以为让多个 agent 开会讨论、来回对话就是编排了。折腾一番之后你会发现只要对话轮数一多token 开销巨大还容易出现逻辑漂移。我们改成了任务图编排执行层先把一个复杂请求拆成 DAG有向无环图上的节点每个节点对应一个 agent 的一次调用。节点之间有明确的依赖关系比如查库存必须在确认商品之后执行。执行层用状态机推进整个图每个节点有 pending、running、succeeded、failed、retrying 这五个状态。这个设计的优点是可监控、可断点续跑。某个节点失败了不需要整个任务重来只需要重跑依赖链上的节点。而且因为是图我们可以做并行优化没有依赖关系的两个节点同时跑整体延迟能明显下降。2.4 治理层权限、限流、审计一起前置治理层是 agency-agents 区别于一般编排器的地方。别人交给你一个任务机构里不能只有业务员在跑还得有合规和风控。治理层做三件事。第一工具权限的最小化控制。每个 agent 有一个工具白名单它只能调用自己注册过的工具跨 agent 调工具必须经过 pre-hook 审核。我们在调用链路上埋了一个拦截器所有工具调用都会先经过治理层校验。第二限流和预算控制。每个调用方上游应用有一个配额每个 agent 每小时的调用量也有配额。超出后不是简单地拒绝而是把请求放到队列里排队或者降级到异步处理。这样流量高峰期不会打爆底层依赖。第三全量审计。每个请求从进来到结束每一步的决策依据、prompt 摘要、工具调用参数、返回结果摘要、耗时、token 数全部结构化落库。出错时你不是靠猜而是像放录像一样把整个执行过程倒放一遍。3. 一个最小可用的 agency-agents 核心骨架讲完架构我直接把当时跑通第一个版本的核心代码骨架放出来。麻雀虽小五脏俱全。你照着搭一个运行起来就能理解前面说的那些抽象概念到底落在什么代码上。3.1 能力卡让每个 agent 自己说清楚会什么能力卡是整个系统里最重要的数据结构我用 Pydantic 定义保证运行时数据合法性。from pydantic import BaseModel, Field class Capability(BaseModel): intent: str description: str required_permissions: list[str] Field(default_factorylist) cost_level: int Field(1, ge1, le5) # 1最便宜5最贵 class AgentSpec(BaseModel): agent_id: str display_name: str capabilities: list[Capability] endpoint: str # agent 的服务地址 timeout_ms: int 15_000 max_retries: int 2 backup_agent_id: str | None None enabled: bool True每个 agent 启动时会把自己的能力卡注册到 agency 的注册中心。这里有个经验能力卡上的 intent 一定要型号化命名比如order_refund_status而不是查退款状态。因为后面路由匹配会用 embedding命名规整的 intent 向量相似度也更稳定。3.2 意图路由与动态转发路由模块的核心就是一个 match 函数。输入标准意图输出一个 agent 引用。我贴一个简化版class Router: def __init__(self, registry: AgentRegistry, cost_optimize: bool True): self.registry registry self.cost_optimize cost_optimize async def route(self, req: NormalizedRequest) - RoutingDecision: candidates [] for spec in self.registry.list_enabled(): if not self._capable(spec, req.intent): continue if not self._authorized(spec, req): continue if not self._healthy(spec): continue score self._score(spec, req) candidates.append((score, spec)) if not candidates: return RoutingDecision(actionfallback, reasonno_capable_agent) # 根据成本优化开关选择最高分或者最便宜的 candidates.sort(keylambda x: x[0], reverseTrue) if self.cost_optimize and candidates[0][0] - candidates[1][0] 0.1: best min(candidates, keylambda x: x[1].capabilities_cost()) return RoutingDecision(actioninvoke, specbest[1]) return RoutingDecision(actioninvoke, speccandidates[0][1])里面有几个细节值得说明。_capable用的是 intent 精确匹配 embedding 相似度兜底。_authorized会检查请求方的 token 权限和 agent 的 required_permissions 有没有交集。_healthy会查 agent 的健康状态和熔断开关。_score结合能力匹配度、历史成功率、当前负载计算。因为用了 async这个 route 过程可以并发跑多个候选 agent 的评估不会阻塞太长时间。实测中注册了十几个 agent单次路由耗时在 5ms 以内完全无感。3.3 超时、重试与熔断的三级降级单次调 agent 总会有意外。我们的策略是三级降级一级调当前 agent超时后自动换到 backup agent如果 backup 也失败则任务降级为异步队列处理而不是直接丢给用户一个报错。class ExecutionNode: async def run(self, ctx: TaskContext) - NodeResult: for attempt in range(1 self.spec.max_retries): try: if self.is_circuit_open(self.spec): return await self._fallback_to_backup(ctx) return await self._invoke_with_timeout(ctx) except AgentTimeoutError: self._note_failure(ctx, timeout) if self.spec.backup_agent_id: return await self._invoke_backup(ctx) except AgentUnavailableError: self._open_circuit(self.spec, duration_seconds30) return NodeResult(statusasync_deferred, queue_urlctx.fallback_queue)这套逻辑解决的是单个 agent 挂掉后全局雪崩的问题。熔断器一旦打开30 秒内这个 agent 的所有请求直接走降级路径不再等待超时下游服务有了喘息时间。重试策略上我们用的是指数退避 抖动第一次等 0.5 秒第二次等 1.5 秒两次以后不再重试同一个 agent。3.4 决策轨迹可回放的审计日志最后一块骨架是日志。不是让开发自己打印几行 info 就完事而是在关键节点打结构化事件的 anchor。class TraceEvent(BaseModel): trace_id: str node_id: str event_type: Literal[route, invoke, tool_call, retry, fallback, error] decision: str payload: dict cost_tokens: int 0 latency_ms: int 0 happened_at: str我们把这些事件序列化成 JSON通过消息队列写入 ClickHouse。查询端写了一个简单界面输入 trace_id就能看到整个任务的生命线。有一次一个 agent 悄悄调用了本该禁用的工具就是靠这个日志抓出来的。没有全链路轨迹多 agent 系统出问题的时候你连该找谁谈话都不知道。4. 正式上线第一周我们踩到的四个深坑骨架能跑和能扛生产中间隔着一堆血泪。我把第一周最痛的问题拿出来说每个都对应一条真实的告警记录。4.1 两个 agent 共享连接池结果互相等成了死锁第一个坑来得特别快。上线第一天下午某两个 agent 的请求突然集体超时QPS 直线掉到 0.1。查了半天发现这两个 agent 会互相调用对方的工具而两个 agent 共用同一个底层的 HTTP 连接池对象。连接池使用了有界信号量最多允许同时 8 个连接。Agent A 发起了任务需要调用 B 的工具但它把连接池里的配额全占满了B 在等 A 释放。这样一层套一层整个池子被死锁堵死。解决方式有两步。第一步把共享连接池拆开每个 agent 独立连接池跨 agent 调用走标准 HTTP 网关不再直接复用对方的池子。第二步给工具调用边界加上带超时的信号量宁可快速失败也不要无限期等待。这个坑提醒我编排层可以调 agent但下面的基础设施资源绝不能共享得太随意。4.2 一次网络抖动触发了全链路的重试风暴上线第三天我们依赖的一个第三方库存接口出现约两分钟的网络抖动。正常情况下这个级别的抖动应该马上自己恢复。但我们忽略了所有 agent 的重试策略是独立配置的而且每层都乐观地认为重试一次就好。结果抖动一出现十几个 agent 在同一时刻对自己的下游发起重试库存接口原本只需要承受平时 3 倍的请求硬生生被打到了 20 倍直接把别人家的限流打满。这一天学到的东西我写进了项目文档的第一条运维规范全局必须有一个统一的断路器而不是每个 agent 各自为战。我们在 agency 层加了一个全局 CircuitBreakerManager一旦检测到某个下游服务连续失败率超过 20%直接让所有依赖它的 agent 都进入熔断状态同时启动 token bucket 限流把重试请求压到一个极小速率确保不会形成重试风暴。4.3 第三跳就失忆上下文压缩把关键约束压没了这个坑是最隐蔽的。我们的执行链是客服 agent - 订单 agent - 退款 agent任务请求里有一个关键约束只允许查询 2024 年之后的订单。任务传到第三跳时因为我们用了上下文压缩策略长对话历史会被摘要。结果摘要过程把这条硬性约束弄丢了退款 agent 直接在数据库里查了全时段订单差点把三年前的退款记录拉出来发给用户。教训也很干脆关键约束不能放对话里更不能赌摘要不会丢必须结构化传递。我们后面在 TaskContext 里加了一个 immutable_fields这个字段在整个执行链路上只读任何一层都不能修改或压缩。凡是业务层面的硬约束比如时间范围、金额上限、用户 ID、权限范围全部走这个卡槽。prompt 里的东西可以被模型自由发挥但 TaskContext 里的是底线程序化保证。4.4 YAML 配置深度合并时列表默认替换而不是追加这是个很小但是杀伤力很大的问题。我们有一个全局默认配置各个 agent 的配置再叠加用 deep merge 的方式合并。某天我更新配置给某个 agent 加了一个新工具结果上线后它的工具列表反而少了三个因为它自己的配置列表在 deep merge 时把全局配置里的工具列表整个覆盖掉了。yaml.safe_load 解析出来的 list很多 merge 库默认是替换语义而不是追加语义这跟我们预期的完全相反。后面我们的解决方案是配置文件里所有列表字段全部改成显式的extend_from_base: true/false标记并且在 YAML schema 校验阶段就强制开发者写清楚这个字段。没有标记的列表字段在合并时直接报错逼你去面对歧义。从那以后配置合并这件事再也没有出过幺蛾子。5. 什么时候值得上这个方案什么时候别硬上写到最后一部分我想聊点更实在的。很多人看到架构和代码很容易兴奋回去就想给自己的项目套上。但任何技术方案都有它的适用半径我把自己的判断标准写下来。5.1 三个真正值得冒险引入的场景第一个场景是你有超过 5 个 agent 同时在对外服务而且这些 agent 会互相调用。到这个规模人脑已经管不过来谁在用哪个工具、谁在调用谁了没有治理层出问题一定是最低效的方式——人肉翻日志。第二个场景是你们要把 agent 能力开放给其他团队或者外部系统使用。一旦开放你就要给别人 SLA、给别人鉴权、给别人计费。这些东西往 agent 里塞是塞不下的必须独立出来。第三个场景是你们对成本敏感想把 LLM 调用做成可观测、可优化的资源。我们的成本感知路由能让便宜的模型处理简单请求贵的模型只处理复杂请求。这个能力要写在单体 agent 里非常别扭做成独立路由层反而自然。5.2 哪些情况我劝你别用如果你的项目还处于一个 demo 跑通就行的阶段完全没必要上这套治理体系。多一层架构意味着多一层延迟、多一份部署和运维工作。你一个 agent 自己就能搞定的任务硬套一个中介层只会让问题变复杂。另外如果你们的场景是单链路、低延迟、且没有跨团队协作的需求比如一个非常固定的小工具型 agent走微服务调用已经够好不需要引入编排。用自己的话说就是不要因为框架好看就去重构你是在解决某个具体问题的不是为了把系统做复杂。5.3 如果继续做下去我会优先补这三件事我目前的路线图上有三件优先级最高的事。第一把路由策略从规则 embedding升级为带在线反馈的语义路由每一次 agent 执行的结果都能回流更新路由权重让路由越来越准。第二把审计日志从回放升级为回放 离线分析用聚类算法找出哪些失败其实是同一类根因。第三把治理层从内部库变成独立可部署的 sidecar 服务让外部团队也能很轻地接入而不需要跟我们共享一套代码库。做这个项目的这段时间我自己最大的体会是多智能体系统真正难的地方不是让单个 agent 变聪明而是让一堆 agent 在一起还能有序、可信、不出重大事故。你可能不需要叫它 agency-agents也不需要照搬我们四层的每个细节。但如果你的系统里 agent 已经开始变多试着在它们之上加一层统一的路由和治理哪怕只是一个很薄的中介层都会让你接下来的维护工作轻松非常多。