有人问过我为什么你们内部做AI能力接入时文档那么少新同事一进来就能把模型调用跑通。其实不是我们人厉害而是我们把“接上游模型”这件事变成了一个统一的服务内部代号叫ax全称Agent eXecutor。业务方不用关心对面是哪个供应商、参数怎么配、密钥放哪里只需要按ax的定义发一次请求就行。今天就把这个服务从设计到落地到踩坑的完整过程翻出来讲讲有类似需求的朋友可以直接参考。这套东西放在团队里解决的是一个特别现实的问题多人、多项目、多条产品线都要用模型能力但又不能每个人都各接各的。一开始我们团队也踩过“各管各”的坑后来把接口收口才真正把AI能力变成随手可用的内部基础设施。下面我从头拆解。1. ax到底是什么为什么说它是“AI能力调度层”1.1 团队接入AI能力为什么会失控如果你所在的团队同时维护好几个产品每天有几十个人在跟不同的模型供应商打交道你一定见过这样的场景A项目的同学自己注册账号、拿Key、照着文档写了一套调用代码B项目因为业务数据更敏感用的是本地部署的模型C项目嫌上游响应慢自己封装了一套超时重试逻辑。表面上大家各忙各的实际上大量工作都在重复。更难办的是变更。供应商某个模型下线了、参数格式调整了、鉴权方式改了团队里每个接过的项目都要跟着改一遍。有的人改得及时有的人改完还是跑不通最后出问题时谁也说不清当前线上代码到底连的是哪家供应商的哪个模型。调度层的第一个作用就是切断这种混乱所有接入方只面对ax上游变更由ax统一消化。1.2 ax在整体架构里站在哪个位置从架构位置上看ax是夹在“业务服务层”和“模型供应商层”之间的调度节点。业务服务发请求到axax按照配置好的策略决定把请求转发给哪一个上游能力源拿到结果后再按统一格式返回。业务服务A ——┐ 业务服务B ——┼—— ax调度层 —— 上游模型A 业务服务C ——┘ |—— 上游模型B |—— 本地GPU服务我选择把调度层做成一个独立HTTP服务而不是塞进某个公共库原因是独立的服务可以统一升级、统一监控、统一控制流量。公共库方式看着轻量但每个业务服务升级库都要发版调度策略调整也没法做到全局实时生效。独立服务虽然多了一个网络跳转但只要控制在毫秒级开销对这单价值来说完全值得。1.3 “ax调度”到底调度什么“调度”这个词听起来很重落到我们这个场景里其实是四件事。第一是路由调度决定一次请求该由哪个上游模型处理第二是流量调度控制每个调用方和每个上游的速率与配额第三是异常调度上游超时、限流、故障时自动切换或降级第四是资源调度统一管理密钥、模型参数映射和上下文额度。很多人以为调度层就是“转发”那确实是把问题想简单了。转发只解决“按地址送达”调度还要解决“送的途中发生了什么、失败了该谁接、代价由谁承担”。后面我会把这几个部分逐个展开分析。2. ax的核心设计统一协议与模型路由2.1 用OpenAI兼容协议统一入口为什么是个好选择ax对外暴露最核心的接口我直接选择了兼容OpenAI Chat Completions风格的协议。请求长这样POST /v1/chat/completions { model: text-gendefault, messages: [ {role: system, content: 你是一名内容编辑}, {role: user, content: 把这篇文章改得更自然一些} ], temperature: 0.7, stream: false }选OpenAI兼容格式的考量很务实生态里大量的开源工具、客户端SDK、监控面板都原生支持这个协议业务方不需要为了接入ax额外发明一套格式。即使某个调用方原来已经是这么写代码的迁移到ax时只需要改base_url和鉴权头代码改动量非常小。协议统一之后业务方拿到的返回值也是固定结构。不管上游供应商返回的字段名是choices还是contents到了ax这一层都会归一化成一致的JSON结构。这一步极大降低了下游的解析成本。2.2 模型路由表的配置思路统一协议能够成立的关键是model字段背后有一套路由映射。业务方请求里写的不是“xxx供应商的某个模型名”而是ax内部定义的别名。比如text-gendefault、text-gencheap、embeddingzh这样的逻辑名。ax根据这些逻辑名去查路由表再决定实际请求哪一个上游模型。路由表我放在一个YAML配置文件里管理原因是这样可以让非开发同事也能看懂和微调。一段简化的示例routes: - model: text-gendefault upstreams: - name: provider-a type: openai_compat api_key_env: PROVIDER_A_KEY model_name: claude-3-5-sonnet weight: 80 - name: local-gpu type: local endpoint: http://127.0.0.1:8001/v1/chat/completions weight: 20这里weight就是流量权重80%走云端供应商20%走本地GPU服务。业务方完全不需要知道背后发生了这种拆分它们的每一次请求都还是发到ax由ax按权重往上分配。把路由配置抽出来的好处是模型升级和灰度可以只改配置不动代码。比如把本地上一个新版模型从0%权重逐渐加到100%过程中出问题随时回滚整个过程对业务方透明。2.3 参数映射与归一化处理模型路由还算直观真正麻烦的是不同供应商之间参数不统一。有的供应商支持temperature和top_p有的只支持temperature还有的把max_tokens叫成max_new_tokens。ax在适配层做了一层参数归一化。我采用的策略是协议层只暴露几个核心参数temperature、top_p、max_tokens、stream、stop。适配层收到这些通用参数后按供应商规则转换成它认识的样子。不支持的参数直接丢弃不做强行模拟。例如某个供应商没有top_p概念即便业务方传了top_p适配层在构造上游请求时也会忽略它不让错误参数透传出去。温度参数也很典型。A模型的temperature范围是0到2B模型只有0到1.5业务方却统一传0.7。ax会做范围归一化确保0.7在不同模型上对应的随机性尽量接近。这个映射关系没有行业通用标准需要每个团队根据自己对模型的使用经验去微调但至少比直接透传要可控。2.4 密钥统一管理与调用方鉴权传统的接入方式是每个项目各持一个上游密钥。问题是一旦密钥泄露无法定位是谁泄露的。ax把上游密钥全部收口到服务端对业务方另外发一套内部身份凭证。调用方通过app_id和app_secret换取JWT后续请求带上JWT即可。# 伪代码示意鉴权中间件 async def auth_middleware(request: Request, call_next): token request.headers.get(Authorization, ).replace(Bearer , ) try: payload jwt.decode(token, settings.jwt_secret, algorithms[HS256]) request.state.app_id payload[app_id] request.state.quota payload.get(quota, {}) except Exception: return JSONResponse(status_code401, content{code: unauthorized}) return await call_next(request)这一步的意义是让每个调用方都能被标识、被计量、被限流。因为上游密钥统一由ax管理就算某个调用方身份泄露ax单独吊销它的JWT就行不需要动上游的任何配置。团队内接入很快密钥权限控制也比较干净。3. 实操过程用FastAPI把ax跑起来3.1 工程结构怎么搭先用FastAPI搭骨架原因很简单异步性能好作者维护活跃类型提示完整写这种IO密集型的转发服务非常顺手。工程目录如下ax/ ├── app/ │ ├── main.py # FastAPI实例和中间件注册 │ ├── auth.py # 鉴权相关 │ ├── routes.py # HTTP接口定义 │ ├── dispatch.py # 调度核心逻辑 │ ├── upstreams/ │ │ ├── base.py # 上游适配器抽象 │ │ ├── openai_compat.py # OpenAI兼容上游 │ │ ├── local.py # 本地模型服务 │ │ └── ... │ ├── metrics.py # 监控指标记录 │ └── config.py # 配置加载 ├── config/ │ ├── routes.yaml # 路由表 │ └── settings.yaml # 全局配置 ├── requirements.txt └── deploy/ └── Dockerfile这种拆分方式主要服务于两个目标每个上游适配器互相隔离异常处理不会串核心调度逻辑独立于HTTP层以后想加消息队列或异步任务时可以复用。3.2 调度核心代码拆解调度核心最简版本如下已经足够跑通整体链路async def dispatch(request: ChatRequest, state: AppState): route get_route(request.model) # 根据model别名查路由表 upstreams route.upstreams # 获取可用的上游列表 candidates weighted_pick(upstreams, attempts3) last_error None for upstream in candidates: try: adapted upstream.adapt(request) # 参数归一化 resp await upstream.call(adapted, streamrequest.stream) if request.stream: return StreamingResponse(resp) return resp except RetryableError as err: last_error err continue raise ServiceUnavailable(detailstr(last_error))weighted_pick做的是按权重选出候选上游列表第一次优先选权重最高的失败后自动尝试下一个。这样“调度”的含义就具体化了它不是随机转发而是带着优先级和容错逻辑的智能分配。这里有一个重要的设计决策重试的候选列表要提前算好而不是每次失败后临时再选。临时再选很可能又会选到同一个已经故障的上游造成无意义重试。提前算好并按顺序记录可以让每个候选只被尝试一次保证重试之间真正换了节点。3.3 多供应商适配器怎么做base适配器定义一个统一接口所有上游实现它class BaseUpstream: name: str def adapt(self, request: ChatRequest) - dict: raise NotImplementedError async def call(self, adapted_request: dict) - UpstreamResponse: raise NotImplementedError def is_retryable(self, status_code: int) - bool: return status_code in {429, 500, 502, 503, 504}OpenAI兼容适配器做的事情最简单因为协议接近通常只需要把model替换成供应商真实模型名。本地模型服务适配器则要做一点点URL拼接和鉴权头转换。正是通过这种适配器机制之后接入任何新供应商基本都只需要新增一个文件主流程代码一行不用改。我建议适配器里一定要把非预期异常包成统一的内部异常类型。比如上游返回状态码200但响应体缺失这种“假成功”是最难排查的。通过适配器兜底校验把它转成ServiceError记录日志再走到统一错误码返回业务方看到的日志会清晰很多。3.4 部署与基础性能配置部署形态我用Docker加Gunicorn运行Uvicorn worker多worker模式可以更好地利用多核CPU。命令简化如下gunicorn app.main:app \ -k uvicorn.workers.UvicornWorker \ -w 4 \ -b 0.0.0.0:8000 \ --timeout 120 \ --max-requests 10000 \ --max-requests-jitter 1000--max-requests设置的意义是防止某个worker内存增长后一直不被回收定期重启可以缓解长尾内存问题。--max-requests-jitter用来避免所有worker在同一时刻到期、集体重启的惊群效应。由于ax做的事情本质是转发而不是计算它的CPU消耗主要集中在JSON序列化和请求调度上所以绝大多数场景4个worker就够。真正的瓶颈在上游响应时间和业务侧并发数这些要靠限流和排队策略去保护而不是无限增加worker。4. 调度策略负载均衡、重试、降级与限流4.1 权重负载均衡与故障转移负载均衡我用了带权重的轮询方式。为什么不用简单轮询因为不同模型成本和能力差异很大。一个高频低价值场景比如标题生成应该尽量命中便宜模型一个复杂推理场景应该尽量命中强模型。权重不是玄学而是“成本与效果”的量化。权重配置可以动态调整。本地GPU负载上去之后某段时间效果开始变差我直接把权重从30调低到10流量自动转移不需要重新发布代码。这个体验在出故障时特别值钱相当于拥有了一个不用重启的流量开关。故障转移的粒度也要设计好。我选择的是“上游实例级别”而不是“供应商级别”。比如同一家供应商配置了两个endpoint其中一个超时只跳过那一个继续尝试另一个而不是整个供应商都标记不可用。这样能尽量保住可用性也不会因为误判把健康实例也拉黑。4.2 重试策略与幂等控制重试是调度层最容易踩坑的地方不是所有失败都应该重试。我的策略分三类错误类型是否重试说明429限流有限重试最多2次重试间隔递增5xx/网络异常重试尝试下一个候选上游4xx参数错误不重试重试也一样会失败内容安全拦截不重试应返还给业务方检查一个细节点是限流重试时的退避时间。上游已经过载立刻重试只会加剧问题。我用的是指数退避加抖动第一次等200ms第二次等500ms加上随机0到100ms偏移避免多个请求同时打上去。真实业务里效果比固定间隔好很多。还需要注意流式请求的重试。一次流式请求已经发出去了上游产生了部分响应这时候重试会把重复的token带给用户。我的做法是流式请求在“建立连接后、收到首个字节前”允许重试一旦收到首个字节就开始透传中途断开不重试只记录错误。这样用户最多看到连接失败的错误提示不会看到语义重复的内容。4.3 限流与配额保护ax是团队内部服务但照样会出现突发流量。限流的作用有两条保护上游请求配额不被单个人打爆保护其他业务方不被一个异常调用方拖下水。我做了三层限流。第一层是按app限流每个业务方每秒最多N个请求超了直接返回429。第二层是按model限流某个模型别名一秒钟最多M次调用防止某个低成本模型被高频刷出天价费用。第三层是按上游实例限流这是真正保护上游资源的最后一层。除了QPS限流还要有长期配额控制。我倾向于按token使用量来限额因为模型计费是按token算的QPS限流没法防止一个调用方每次传超长上下文。实现上用Redis做计数key是app_id 日期每次请求结束后累加本次使用的token数超过阈值直接拒绝后续请求。成本控制也要在调度层体现。ax每次转发完会记录本次请求的输入token、输出token、使用的模型和上游单价定期汇总出各业务方的成本报表。没有这一步月底看账单就是一笔糊涂账根本定位不到是哪个业务方把预算烧掉的。5. 踩坑记录与问题排查实录5.1 流式输出中断客户端一直转圈第一次做流式转发时我以为只要把上游响应原样透传就行。实际客户端经常卡住不动。原因在于上游的SSE响应有严格的格式要求每个数据块要以data:开头以两个换行分隔结束时还要发一个data: [DONE]。有一类是上游响应的每个块都正常但ax在转发时对HTTP响应做了缓冲没有实时flush。客户端拿到的是一个攒了很久的大包等不到后续token体验就是“第一句话迟迟不出来”。解决方式是关掉HTTP代理层的缓冲同时确保转发响应时逐块写入并flush。5.2 参数不透传导致的模型行为差异业务方反馈在ax上调用模型生成效果没有直连时好。排查发现ax把temperature过滤掉了因为供应商适配器里没定义这个参数的转换逻辑。对业务方来说它们传了temperatureax静默丢掉了输出自然变“老实”。这个教训是参数映射不明确时宁可报错也不要静默丢弃。后来我给适配器加了一个参数白名单机制凡是没被适配器显式支持的参数在团队内调试环境直接报invalid_request开发阶段就能发现而不是等到线上生成质量出问题才开始查。生产环境保留丢弃策略但一定记录warn日志方便追踪。5.3 上游限流阈值识别错位导致重试风暴有一次上游供应商限流但返回的状态码是200响应体里带了一个限流错误码和提示文案。我们的重试判断只看HTTP状态码没有识别业务错误码导致这些“假成功”请求没有被重试也没被正确报错业务方拿到的结果是一段错误的文本还在继续拿它做后续处理。从那以后所有适配器都必须做响应体语义校验。HTTP 200只代表“传输成功”不代表“请求成功”。ax在转发前先检查响应体内的错误码字段发现问题就转换为统一错误码。这是调度层对业务方负责的底线。5.4 上下文过长上游直接422报错团队里一个业务方提交了一个PDF总结场景上下文动不动就超过上游模型的窗口。ax直接把超长请求转发过去上游返回参数错误。对方体验是“模型动不动就报错”毫无提示。我在ax里加了一个上下文长度预算模块。配置里为每个模型别名定义最大窗口请求进来先估算token长度超过警告线就拒绝并提示业务方使用长文本专用别名超过硬限制则直接返回清晰的错误码。这里估算不需要特别精确用字数和字节数近似估算就行。宁可多留一点余量也不要预估偏小导致误杀正常请求。5.5 可观测性调度层必须要有的三张表排查调度层问题最怕的是没有“现场”。我在ax里记录了三个维度的数据。第一张是请求日志表包含app_id、请求模型、实际上游、响应时间、重试次数、最终结果。第二张是上游健康表统计每个上游过去一分钟的成功率、平均延迟、首token延迟。第三张是业务方用量表展示每个调用方今天的请求量和token消耗。这里最值得一提的是首token延迟。对于流式请求用户感受到的“快慢”主要由首token延迟决定不是总耗时。ax在接到流式响应第一个字节时就记录该指标一旦它出现恶化优先检查上游排队情况和网络链路而不是瞎猜业务代码。这个指标也是我在实践里排查体验问题最有效的数据。最后说一点我的体会ax做完大概一年了最大的感受不是代码多厉害而是“接入方式被统一之后团队协作成本真的降下来了”。以前我一天要回答好几个“怎么连模型”的问题现在只需要把ax的接入文档丢过去。上游模型想升级就升级想切换就切换业务方的感受是零。如果你想在团队里做类似的东西我建议不要一开始就追求大而全。先做成一个只支持一个上游、只提供最基础转发的HTTP服务跑通之后再慢慢加路由、加权重、加重试、加监控。ax的完整能力也不是一天长出来的是靠一次一次线上故障和业务方反馈迭代出来的。最后再分享一个小技巧调度层的接口文档一定要保持OpenAI兼容哪怕你内部已经定义得很舒服。只要协议足够标准你在换模型、换工具链、做压测、接监控时都会省很多事。这大概是ax这个项目里我做的最正确的一个决定。