自建AI模型网关：从转发工具到成本治理层的实战改造

自建AI模型网关这件事，我得从一次并不风光的对账说起。上个月财务把模型API账单甩到我桌上，我看了一眼，整个上半月的费用比去年季度总额还高。我们的网关明明上线了，把所有模型调用都统一走了网关，还做了转发、鉴权、日志，理论上一切尽在掌控，怎么就烧出去这么多钱？后来我把这几周的调用记录翻了个底朝天，才慢慢意识到一个很扎心的事实：你把流量收拢到自己的AI模型网关，只是把事情“看清楚”了，并没有把成本“管住”。这个网关，如果只做转发，本质上就是一条更粗的管道，模型贵不贵、调用合不合理、重复请求多不多，这些真正的成本问题一个都没碰。这篇文章就把我这次自建AI模型网关踩过的坑，以及后来从“转发工具”硬改成“成本治理层”的过程完整记录下来，希望能帮到正在自建或准备自建网关的团队。

1. 自建AI模型网关，我当初为什么这么干

1.1 没有网关之前，业务侧的模型调用乱成什么样

我们团队最初面临的局面，估计很多公司都一样：多个业务线都在做AI功能，有智能客服、文本摘要、搜索排序，还有几个内部效率工具。每一条业务线都自己申请模型API的Key，自己连供应商OpenAI、国内大模型厂商等等。代码里到处都是散装的模型调用，你根本不知道今天有多少个Key在线上跑。

这种状态带来的问题，不是一天两天了。首先是Key管理混乱，权限收不住，有人离职了Key还在线上用，换也不是不换也不是。其次是模型选择完全靠自觉，有的业务明明只是做一个文本分类，顺手就把旗舰模型配上了；有的业务为了省事，把几千行历史聊天记录一并塞进上下文，根本不看token消耗。再就是失败重试策略不统一，有一次一个下游供应商抖动，有个业务的重试逻辑一口气打了20遍，费用哗哗地涨。出了问题还特别难排查，到底哪个业务、哪个场景、哪个用户在调用，完全没有一个统一入口能看清楚。

我当时就提了一个方案：自建一个AI模型网关，把模型调用的入口统一收敛到一个服务上。核心目标有三个：统一接入、统一鉴权、统一日志。说得更直白一点，我先不看成本优化，先把所有流量集中到一个口子上，让团队能知道每天有多少人在调用、调用的是什么模型、有没有报错。这个目标在当时看来非常合理，大家也都觉得早该这么干了。

1.2 初版网关设计：把“转发”做好，我就以为万事大吉

初版网关设计得很简单，我当时的想法是，只要能让业务方把代码从直连模型API改成请求网关，就算成功。技术选型上，我用了一个轻量的中间层服务，Python FastAPI加Redis，接收所有模型请求，校验API Key，然后把请求原样转发给对应的模型供应商，拿到响应后再原样透传给调用方。核心代码摊开看，其实就是一个转发层。

from fastapi import FastAPI, Header, HTTPException import httpx app = FastAPI() UPSTREAM_MAP = { "openai": "https://api.openai.com/v1/chat/completions", "qwen": "https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation", } @app.post("/v1/chat/completions") async def chat_completions(request: dict, x_api_key: str = Header(...), x_upstream: str = Header("openai")): # 鉴权：校验这个 Key 是否有权限调用 if not check_key(x_api_key): raise HTTPException(status_code=401, detail="invalid key") # 转发：把请求体原样打给上游 upstream_url = UPSTREAM_MAP.get(x_upstream) if not upstream_url: raise HTTPException(status_code=400, detail="unknown upstream") async with httpx.AsyncClient(timeout=60) as client: resp = await client.post(upstream_url, json=request, headers={ "Authorization": f"Bearer {get_upstream_key(x_upstream)}" }) return resp.json()

当时我还挺满意，因为这个服务上线很快，两天就接完了主要业务。上线第一周，所有请求都正常流转，日志也统一了，业务方反馈接入成本很低，只要把base_url换掉就行。我一度以为这个项目已经成了，接下来只需要修修补补。

1.3 上线两周，代价立刻出现了

结果就是文章开头那一幕，月底对账的时候，账单不仅没降，反而比上个月还高。我第一反应是网关有Bug，或者是有人恶意刷接口，赶紧拉日志出来查。查完我才明白，网关确实把所有调用收拢了，但它本质上只是把费用从多个供应商账单收拢到了一个出口。业务方以前怎么调用，现在还是怎么调用；以前用贵模型拍脑袋，现在依然用贵模型拍脑袋；以前重复请求反复打，现在依然重复请求反复打。

换句话说，网关帮我把问题的可见度提高了，却没帮我解决任何一个成本问题。你看到每一笔token在消耗，但你阻止不了它们。这种感觉很难受，就像装了个水表，发现家里漏水很严重，但你手上没有扳手，关不掉阀门。那段时间我反复跟团队说一句话：“转发”解决的是技术通道问题，而“成本浪费”是治理问题，通道通了，不代表成本会被管住。

后来我把这次经历做了一次复盘，核心结论是：一个只负责转发的AI模型网关，充其量是个API代理层，它没有路由、没有缓存、没有配额、没有计量，就谈不上成本治理。我决定开始改造。

2. 只做转发，成本浪费到底出在哪

2.1 贵模型被当成默认选项：杀鸡一直在用牛刀

成本浪费的第一个大头，是模型选型严重不合理。我拉了一周的调用日志，统计各模型的使用占比，发现几个旗舰模型占了总调用量的六成以上。但实际上很多业务场景根本不需要那么强的模型，比如文本标签分类、关键词抽取、意图识别、格式改写，这类任务用中等规格的模型完全能打。

这里算一笔简单的账，假设某旗舰模型输入价格是15元/百万token，输出价格是60元/百万token；一个便宜模型输入3元/百万token，输出15元/百万token。一个文本分类场景每天调用10万次，每次平均输入2000 token、输出200 token。旗舰模型一天的成本大约是10万乘以（2000除以100万乘以15，再加上200除以100万乘以60），算下来是4200元。便宜模型一天的成本是10万乘以（2000除以100万乘以3，再加上200除以100万乘以15），只有900元。一天就差3300元，一个月就近10万元。而这只是一个场景。

这种浪费的根源，在于业务方没有成本和模型能力匹配的意识。对他们来说，我在代码里写死了旗舰模型，能完成任务就行，不会有人关心token单价。网关如果不提供路由能力，它就永远只能眼睁睁看着这些请求打向最贵的模型。

2.2 重复请求反复计费：同一个Prompt烧了好几遍

第二个大头是重复请求。我发现有些接口一天之内传进来的Prompt几乎完全一样，比如“判断用户这句话是否属于投诉”“给这篇文档生成一段摘要”，这类任务的输入高度相似，但因为没有缓存，每一次都实打实打给了模型API，每一分钱都在重复计费。

举一个真实案例，我查日志的时候看到一个接口，一天内完全相同的Prompt跑了八千多次。单次调用成本按2分钱算，一天就白烧160元。你可能觉得160元不算多，但同样的情况在十几个接口里都存在，有些更夸张，一天重复调用上万次。积少成多，一个月下来就是几万块。

这里面的问题在于，很多内部工具和自动化流程的调用模式是高度可预测的。同一个模板、同一份输入，隔几分钟跑一次，结果几乎不会变。对这类请求，网关如果只是转发，它就是一台没有感情的“烧钱机器”；但如果有一个精确缓存，命中后直接返回历史结果，这部分成本就能直接归零。

2.3 重试与并发放大了损失：模型一抖动，费用翻倍

第三个大头是重试和并发带来的费用放大器。当时网关里配了一个很粗暴的逻辑：只要上游返回超时或5xx，就自动重试3次。这个策略在单次请求上看起来没什么问题，但一旦遇到模型供应商大规模抖动，所有业务同时开始重试，问题就非常恐怖了。

举个例子，某次上游模型连续报警，有20个业务同时处于失败状态，每个业务都按约定重试3次，瞬时流量直接扩大了4倍。结果就是模型供应商更慢、更多请求超时，然后又有一些重试被触发。账单自然也跟着水涨船高。那个时候我才意识到，重试策略不是越简单越好，而是必须配合退避算法和熔断机制。真正应该做的是快速失败、有限重试、及时切走流量。

除了重试，还有并发问题。没有限流的网关，面对突发流量会全量转发给上游，结果上游被压垮，失败的请求又开始重试，整个链路进入恶性循环。如果限了流，至少能保证一部分请求正常完成，另一部分快速返回“稍后再试”，而不是把费用白白烧在失败请求上。

2.4 没有配额与预算熔断：接口在裸奔

最后一个大头，也是最容易被忽视的：没有配额和预算控制。我一直到账单爆炸之后才意识到，网关对调用方是完全信任的，只要Key有效，你想调多少次就调多少次，想调多贵的模型就调多贵的模型。

这就导致了一些典型的失控场景。测试人员对一个内部接口做压测，一个小时打出几十万次调用；某个定时任务因为数据问题陷入死循环，一晚上把所有数据重新跑了一遍；某个业务上线了新的实验策略，忘了关开关，把流量全部切到旗舰模型上跑了一整天。这些场景没有任何一道闸门能拦住。

我一直觉得，成本治理的关键不只是看每一次调用，而是要把每次调用的“额度”算清楚。就像公司预算一样，每个业务线每个月有多少模型调用额度，用完了是降级还是停止，必须提前定好规则。否则等账单出来再查，就已经太晚了。

3. 从“转发”到“经营成本”：网关必须做什么

3.1 模型路由：按任务、按场景、按成本分级

既然只转发不行，那网关第一步要做的就是模型路由。核心原则很简单：能用便宜模型解决的，绝不调用贵模型；能用小模型解决的场景，绝不上旗舰模型。把“业务要什么任务”和“哪个模型适合这个任务”对应起来。

路由的维度可以根据实际情况来定。我后来在网关里至少会考虑这几个维度。第一是业务场景，调用方请求时要传一个scene字段，比如text_classify、summary、code_gen、chat_demo，网关根据场景直接决定默认模型。第二是上下文长度，传入的Prompt或者历史消息非常长时，自动切到支持更大上下文且单价更合适的模型；如果输入很短，就走延迟低、价格便宜的模型。第三是用户等级，免费用户和付费用户可以走不同档次的模型，这不只是为了省钱，也是产品策略的一部分。第四是失败切换，上游模型不可用时，自动路由到备选模型，而不是直接报错。

我踩过的一个坑是，路由规则一开始不要太激进。先做观测，记录每个场景真实调用量、token消耗、模型效果，运行一两周之后，再根据数据调整路由策略。没有数据支撑的路由规则，很容易拍脑袋，拍完就出问题。

3.2 缓存：精确缓存先落地，语义缓存按需开

缓存是解决重复请求最直接的手段。我把缓存分成两档来设计。第一档是精确缓存，也就是请求体完全一样时直接返回历史结果。这个实现最简单，也最容易看到效果。我在网关里对model、messages、关键参数做一个哈希，Redis里存一份，命中就直接返回，完全不再打上游。

第二档是语义缓存，字面不完全一样但语义相同的请求，比如“帮我总结一下这个文档”和“请把这篇文档做个摘要”，可以通过embedding相似度来命中。语义缓存很诱人，因为它能覆盖更多请求，但风险也高。一是embedding计算本身有成本和延迟；二是相似度阈值调不好，容易误命中，返回一个不完全匹配的旧答案，业务方会投诉质量。所以我的建议是，语义缓存只对特定高频、结果相对稳定的场景开启，并且做灰度验证。

还有一个原则要记住，流式输出、个性化回答、涉及隐私或合规要求的请求，不适合做缓存。流式输出要缓存必须等完整结果生成完，首token延迟会变高；个性化回答几乎不会有重复；隐私内容留在缓存里本身就是隐患。

3.3 限流、重试与降级：把失败成本控住

转发模式下的重试是灾难放大器，所以网关必须具备三个能力：限流、重试策略、熔断降级。

限流用令牌桶就够了，不用搞太复杂。每个业务、每个场景设置一个QPS上限，超过部分直接排队或快速失败，避免突发流量打爆上游，也避免费用失控。重试策略一定要改，不能无限重试，默认最多一次，并且要做指数退避，第一次失败等0.5秒，第二次失败等1秒，中间加随机抖动，防止所有请求同时重试。

熔断和降级是配套的。当某个上游模型连续失败率达到一定阈值，比如30秒内错误率超过50%，就把这个上游标记为熔断状态，后续请求自动切换到备用模型。降级则是在配额用尽时使用：某个业务的预算花完了，可以降级到更便宜的模型继续提供服务，而不是直接拒绝用户。这样用户感知不到服务中断，成本也不会失控。

3.4 成本计量与标签体系：让每笔token都有归属

前面说的路由、缓存、配额都依赖一件事情：你必须知道钱花在了哪里。所以我后来在网关里加了全面的成本计量，每个请求处理前后都记录一条日志，包含时间戳、业务线、场景、用户ID、模型名、prompt_tokens、completion_tokens、总token、费用估算、缓存是否命中、命中了哪条路由规则等等。

这些数据落到一张明细表里，每天定时聚合成成本报表。只有有了标签体系，你才能回答“哪个业务最烧钱”“哪个模型最浪费”“哪个用户一天就把配额烧光了”这些问题。没有计量的治理都是空话，这句话我后来反复跟团队讲。网关不是装完路由和缓存就完事了，它得像一个经营仪表盘，每天都让你看见成本的流向。

4. 关键方案落地：路由、缓存、配额怎么实现

4.1 整体架构调整：从纯转发到多级调度

改造后的网关不再是一个简单的转发层，而是变成了一个多级调度器。每个请求进入网关后的处理流程大致是：第一步鉴权，解析API Key和请求头；第二步记录原始请求元数据；第三步路由模块根据场景、上下文长度、用户等级选择模型；第四步查缓存，命中就直接返回；第五步做配额检查，判断这笔请求是否允许继续消耗预算；第六步调用上游，带超时、重试和熔断逻辑；第七步记录计量数据，写成本日志。

这个流程看着环节多，但每一步都是轻量操作，实际增加的时间在几十毫秒以内，还是能接受的。更重要的是，这个架构让网关从“搬运工”变成了“调度中心”，成本控制能力一下子就有了。

4.2 模型路由实现示例：请求级模型选择

路由模块我建议用配置驱动，不要硬编码在代码里。可以在配置中心放一份路由规则，运营人员直接调整，不需要重新发布服务。下面这个示例简化了配置格式，核心是讲清楚思路。

# 路由规则，可以在配置中心运行期更新 ROUTING_RULES = [ {"scene": "text_classify", "model": "fast-model", "max_input_chars": 4000}, {"scene": "extract", "model": "mid-model", "max_input_chars": 8000}, {"scene": "code_gen", "model": "flagship-model", "max_input_chars": 16000}, ] DEFAULT_MODEL = "fast-model" LONG_CONTEXT_MODEL = "long-context-model" def route_by_scene(scene: str, input_text: str) -> str: normalized_scene = (scene or "default").strip().lower() for rule in ROUTING_RULES: if rule["scene"] == normalized_scene: if len(input_text) <= rule["max_input_chars"]: return rule["model"] # 输入过长，专门切到支持长上下文的模型 return LONG_CONTEXT_MODEL return DEFAULT_MODEL

这里的关键点是，业务方在接入网关时必须在请求头或请求体里带上scene字段。如果有些老接口不方便改，也可以先通过URL路径映射，比如/v1/chat/completions/summary这样的路径直接对应summary场景。总之，路由的依据越清晰，后续调整越容易。

4.3 缓存落地示例：用Redis做精确缓存

精确缓存是最容易上手的，我用Redis的字符串结构就够了。核心思路是把“模型名、消息列表、关键参数”三个要素序列化后做SHA256哈希，作为Redis Key，缓存值就是模型返回的结果。

import hashlib import json import redis r = redis.Redis(host="localhost", port=6379, db=0) def _cache_key(model: str, messages: list, params: dict) -> str: raw = json.dumps({"model": model, "messages": messages, "params": params}, sort_keys=True) return "model_cache:" + hashlib.sha256(raw.encode()).hexdigest() def get_cached_response(model: str, messages: list, params: dict): key = _cache_key(model, messages, params) cached = r.get(key) if cached: return json.loads(cached) return None def set_cached_response(model: str, messages: list, params: dict, response: dict, ttl: int = 3600): key = _cache_key(model, messages, params) r.setex(key, ttl, json.dumps(response))

使用的时候，在处理请求前先查缓存，查到就直接返回，查不到再调模型API，拿到结果后写缓存。TTL我一般不会设太长，一小时到一天，根据场景调。这里有个小技巧：对于明显包含时间戳、随机数的请求，缓存命中率会很低。可以在生成缓存Key前先做一层归一化，把时间戳、随机数、无关紧要的空格等字段剔除，再生成Key。这个操作能明显提高命中率。

4.4 配额与预算控制示例：先估算费用，再决定放不放行

配额控制的思路是基于预算上限做前置判断。每次请求进来之前，先根据模型和token数估算本次费用，然后用Redis中的计数器看当前业务当天已经消耗了多少，如果加上本次费用会超过每日限额，就走降级模型或者直接拒绝。下面是一段精简的示例代码。

import time QUOTA_KEY = "quota:cost:{biz}:{day}" def check_and_consume_quota(biz: str, estimated_cost: float, daily_limit: float) -> bool: day = time.strftime("%Y-%m-%d") key = QUOTA_KEY.format(biz=biz, day=day) used = float(r.get(key) or 0.0) if used + estimated_cost > daily_limit: return False r.incrbyfloat(key, estimated_cost) return True

如果配额不足，网关可以根据策略决定是返回429让业务方感知，还是自动降级到便宜模型。我建议在初期先做“拒绝”模式，把问题暴露出来，等业务稳定了再切“降级”模式，避免用户在无感知的情况下拿到质量下降的答案。

成本明细表的结构也很重要，我用的是MySQL，核心字段包括调用时间、业务线、场景、用户ID、模型、prompt_tokens、completion_tokens、total_tokens、估算费用、缓存是否命中、路由命中的规则名、原始响应JSON。有了这张表，就可以写各种聚合SQL来做成本分析。

CREATE TABLE model_call_bill ( id BIGINT AUTO_INCREMENT PRIMARY KEY, call_time DATETIME NOT NULL, biz VARCHAR(64) NOT NULL, scene VARCHAR(64) NOT NULL, user_id VARCHAR(128), model VARCHAR(64) NOT NULL, prompt_tokens INT DEFAULT 0, completion_tokens INT DEFAULT 0, total_tokens INT DEFAULT 0, cost_cny DECIMAL(10, 4) DEFAULT 0, cache_hit TINYINT DEFAULT 0, route_rule VARCHAR(128), raw_response JSON, KEY idx_biz_time (biz, call_time), KEY idx_model_time (model, call_time) );

SELECT biz, model, SUM(total_tokens) AS total_tokens, SUM(cost_cny) AS total_cost FROM model_call_bill WHERE call_time >= NOW() - INTERVAL 7 DAY GROUP BY biz, model ORDER BY total_cost DESC LIMIT 20;

5. 踩坑记录与问题排查速查

5.1 缓存命中率为什么上不去

缓存落地之后，我遇到的第一个问题是命中率远远低于预期。查了一圈才发现，很多业务在Prompt里带了时间戳或者随机字符串，比如“今天是2025年6月1日，请分析以下内容”，每次请求内容都不一样，精确缓存当然永远不命中。

解决方法是做请求归一化。在生成缓存Key之前，先把明显不影响结果的噪声字段去掉，比如时间戳、随机ID、无意义的空格和换行。如果还是命中率低，就要看业务场景是否真的适合缓存。有些场景每次输入都完全不同，硬上语义缓存也没用，没必要为了缓存而缓存。

语义缓存我也试了，踩了不少坑。相似度阈值设高了，命中很少，形同虚设；阈值设低了，隔三差五返回一个语义上差不多但实际内容有偏差的旧结果，业务方立刻投诉质量。最后我学到的经验是：语义缓存只对低风险、模板化的场景开，比如“产品功能介绍”“内部知识问答”这类，结果和措辞不完全一样没关系，但不会出大错的场景。对生成代码、医疗诊断、法律文本这类高风险场景，我坚决不开。

5.2 路由规则上线后，效果回退了

路由规则刚上线的时候，其中一个文本摘要场景被我强制切到了便宜模型。省是真的省了，一个月少了将近八成的token费用，但业务方很快反馈，摘要质量偶发下降，有些长文档的提炼不够准。说白了，便宜模型不是不行，而是在某些边界场景下稳定性不如旗舰模型。

这件事给我的教训是，路由规则的调整必须灰度。不能一把梭把整个场景切过去，而是先放5%的流量，对比效果和成本，再逐步加大比例。同时要给每个场景配一个白名单，某几个核心客户或者高价值用户始终走旗舰模型，其他人走便宜模型。另外还要设置一键回退开关，一旦质量指标恶化，马上切回原模型，不要让业务方在那里干着急。

5.3 token计量口径不一致，对不上账

成本统计做完之后，我发现网关里算出来的费用和模型供应商账单对不上，差得还挺多。排查下来原因很多：不同模型对token的定义不一样，有的按字符，有的按token，有的对缓存命中token打折，有的把系统提示词算在输入里，有的不算。如果我们自己在网关里估算token，用字符数除以4这种粗略方式，误差会非常大。

这个问题的解法是，所有成本计量都以模型返回的usage字段为准，不要自己猜。网关在拿到上游响应后，把usage里的prompt_tokens、completion_tokens、total_tokens原封不动落库，再套用官方计费表计算费用。同时保留原始usage JSON，方便以后对账时追溯。如果供应商支持流式返回，记得在流结束时汇总各分段usage，避免漏算。

5.4 排查成本上涨的通用流程

现在我的日常工作中，排查成本问题已经形成一套固定流程。第一步看成本趋势图，找到涨幅最大的那天和业务线；第二步按标签下钻，看TOP模型、TOP场景、TOP用户，一眼定位钱烧在哪里；第三步翻对应的请求日志，看有没有异常重复、异常重试、异常大token输入；第四步对照路由规则和缓存配置，验证是不是规则没生效或者被绕过了；第五步修复问题后观察24小时，确认成本曲线回落。

下面整理了一个速查表，是我自己排查时常用的对照逻辑：

6. 说点心里话

这次改造做下来，我最后悔的是第一版网关太执着于“转发”这个动作，把网关当成了一个网络管道，只要数据能流过去就觉得任务完成了。但AI模型网关真正值钱的地方，根本不在“转发”，而是在“在不影响业务效果的前提下，用最便宜的方式把任务做完”。如果你现在正准备自建模型网关，我建议第一个版本就把四件事带上：计量标签、模型路由、精确缓存、配额熔断。不要像我一样先把转发上线，再被账单打醒，然后返工。

另外也给还在犹豫是否要自建的团队提个醒，自建网关本身有维护成本，要跟着上游API的变更新增适配逻辑，还要持续迭代路由策略。如果你们团队只有一两个AI应用，直接使用开源的LLM网关可能更划算，先借用现成方案把成本观测做起来，搞清楚自己的调用画像之后，再评估要不要自研。自研不是不行，但要清楚它解决的是“成本治理”问题，不是“转发通道”问题。我这次也就是栽在这个认知偏差上，希望你们能比我少走一段弯路。