在实际的 AI 应用项目中直接对接一两个大模型 API 往往很简单但当团队开始同时使用 OpenAI、Anthropic、Azure OpenAI、国内多家模型服务再叠加多个部门、多个产品线、不同的 API Key 和预算限制时问题就变了路由、安全、计量这些原本不需要关心的能力会成为独立的基础设施问题。Zerker AI Gateway 正是针对这个场景设计的一个轻量网关方案核心职责可以概括为三个词route、guard、charge。下面就以 Zerker 为例用 FastAPI 从零搭出一个可运行的网关骨架讲清楚每个模块为什么要这样做、怎么做以及上线后遇到问题该从哪里查。网关类项目最容易出现的误区是先花大量时间把每个上游模型的能力都接一遍却没有把“路由、守卫、计费”这三个横向能力设计好。单点调用可以靠硬编码解决网关存在的意义恰恰是把这些横切逻辑收敛到一个统一的入口。本文会沿着一条主线展开先理解 AI Gateway 解决什么问题再按“环境准备 - 路由模块 - 守卫模块 - 计费模块 - 串联验证 - 排错 - 生产化建议”的顺序把 Zerker 的骨架完整搭起来。1. 先理解 AI Gateway 要解决的三个核心问题1.1 为什么只接一个模型供应商不够很多项目一开始只接一个模型供应商代码里直接用厂商 SDK 发请求运行一段时间后会发现几个真实的问题。第一是切换成本高。模型供应商的 API 路径、鉴权方式、请求体格式、超时行为、错误码都不完全一致。一旦某家供应商涨价、限流、故障或者出现效果更好的新模型业务代码要跟着改测试也要重跑。第二是密钥管理难。每个上游服务都需要独立的 API Key如果业务服务直接持有这些 KeyKey 的保存、轮换、权限回收都很难控制泄露风险也会被放大。第三是没法统一计量。不同团队调用同一个模型产生的 token 消耗和费用如果散落在各个业务系统里月底对账基本靠人肉统计。AI Gateway 的价值就是把模型能力收口成“一个入口、一套协议、一份账单”。业务方只需要记住 Gateway 的地址和一把 Gateway 下发的 Key不需要关心请求最终落在哪家供应商、上游用的是哪个模型名、费用怎么算。Zerker 的定位也是这个它不是一个模型训练或推理框架而是一个位于业务系统和上游模型服务之间的代理层。它接收业务请求完成路由、守卫和计费三件工作再把请求转发给真正的大模型服务。1.2 route、guard、charge 分别承担什么这三个词可以理解为网关的三层能力顺序上也是请求进入网关后的处理顺序。route 是路由。路由不止是简单地把 model 字段映射到某个上游模型还包含多供应商选择、优先级、故障转移、按成本或延迟做策略调度。业务方传一个统一的模型别名网关负责找到合适的上游目标。guard 是守卫也就是保护层。包括身份认证、API Key 校验、租户识别、限流、配额检查、请求体校验、内容安全审核、上游密钥保护。这一层决定了“谁可以用、能用多少、不能传什么”。charge 是计量和计费。包括记录每一次请求的 token 用量、按价格表计算费用、扣减租户余额、生成账单记录。这一层决定了“用了多少、花了多少钱、谁来承担”。在 Zerker 的实现里这三个模块并不是孤立的三段代码而是同一个请求链路上的三个环节。请求先经过 guard 的认证和限流再由 route 决定转发目标转发完成后由 charge 根据上游返回的 usage 数据完成计量和记账。1.3 Zerker AI Gateway 的整体处理链路把三个能力串起来一次完整请求的链路大致是业务客户端 - 发起 POST /v1/chat/completions携带 Gateway API Key - guard 模块校验 Key、识别租户、检查限流和余额 - route 模块根据模型别名解析目标供应商和上游模型 - 网关携带上游 Key 转发请求到真实模型服务 - 上游返回响应网关解析 usage 数据 - charge 模块计算费用、扣减余额、写入账单 - 网关把原始响应返回给业务客户端这个链路里有一个容易被忽略的设计点业务客户端接触到的只有 Gateway 的 API Key真正的上游 Key 只存在于网关配置或数据库中。这样可以做到即使某个业务方的 Key 泄露也不会直接暴露上游模型服务的密钥guard 层的“密钥保护”指的就是这个隔离。2. 环境准备与项目骨架2.1 依赖与版本选择文章中的示例基于 Python 3.10 和 FastAPI。之所以选择 FastAPI是因为它自带异步能力和参数校验很适合做转发型服务。实际项目中也可以使用 Go、Java 或 Node.js路由、守卫、计费的设计思路是一致的本文的代码只负责把思路讲清楚。最小依赖如下fastapi0.110 uvicorn[standard]0.29 httpx0.27 sqlalchemy2.0 pydantic2.6把这些写入 requirements.txtfastapi0.110.0 uvicorn[standard]0.29.0 httpx0.27.0 sqlalchemy2.0.29 pydantic2.6.4安装命令pip install -r requirements.txt注意这里给出的版本号是写作时使用的组合。落地到真实项目前需要先确认这些版本在目标 Python 环境和部署平台上的兼容性不建议直接照搬版本号。2.2 项目目录结构Zerker 采用按模块拆分的结构每个 gateway 子模块对应一个核心职责zerker-gateway/ ├── main.py # FastAPI 入口串联三个模块 ├── database.py # 数据库引擎与会话 ├── models.py # SQLAlchemy 模型租户、供应商、路由表、账单 ├── schema.py # Pydantic 请求/响应模型 ├── config.py # 配置价格表、限流阈值、数据库地址 ├── gateway/ │ ├── __init__.py │ ├── route.py # 路由模块 │ ├── guard.py # 守卫模块 │ └── charge.py # 计费模块 └── requirements.txt这样的结构不是为了好看而是为了让三个核心职责的边界清晰。后续要加新的供应商改 route 模块要接入 Redis 做分布式限流改 guard 模块要对接财务系统改 charge 模块。各模块之间通过明确的函数接口通信而不是相互直接访问对方的内部状态。2.3 配置中心的抽象供应商、模型、计费规则Zerker 把三类核心配置抽象出来供应商信息、模型路由关系、价格表。供应商信息保存的是上游服务接入参数包括名称、Base URL、上游 API Key 和优先级。模型路由关系解决的是“模型别名到上游模型的映射”。价格表解决的是计算费用时的单价来源。实际项目里这些配置建议放在数据库或独立配置中心中而不是写死在代码里。因为供应商参数会变、模型价格会调、不同租户可能使用不同的价格策略。下面的代码以数据库模型为例# models.py from sqlalchemy import Column, Integer, String, Float, DateTime, ForeignKey, BigInteger from sqlalchemy.sql import func from database import Base class Tenant(Base): __tablename__ tenants id Column(Integer, primary_keyTrue) name Column(String(64), uniqueTrue, nullableFalse) api_key Column(String(64), uniqueTrue, nullableFalse, indexTrue) balance Column(Float, default100.0) created_at Column(DateTime, server_defaultfunc.now()) class Provider(Base): __tablename__ providers id Column(Integer, primary_keyTrue) name Column(String(32), uniqueTrue, nullableFalse) base_url Column(String(255), nullableFalse) upstream_api_key Column(String(255), nullableFalse) priority Column(Integer, default100) class ModelRoute(Base): __tablename__ model_routes id Column(Integer, primary_keyTrue) model_alias Column(String(64), indexTrue, nullableFalse) provider_name Column(String(32), nullableFalse) upstream_model Column(String(64), nullableFalse) enabled Column(Integer, default1) class BillRecord(Base): __tablename__ bill_records id Column(BigInteger, primary_keyTrue) tenant_id Column(Integer, ForeignKey(tenants.id), nullableFalse) request_id Column(String(64), uniqueTrue, nullableFalse) model Column(String(64), nullableFalse) prompt_tokens Column(Integer, default0) completion_tokens Column(Integer, default0) fee Column(Float, default0.0) created_at Column(DateTime, server_defaultfunc.now())这里的 Tenant 代表一个调用方可以是部门、应用或最终客户。api_key 是网关下发给调用方的凭证和上游 Key 完全隔离。这种模型设计背后的思路是计费的最小单位是“租户”而不是“请求”。3. 路由模块怎么把请求送到正确的模型3.1 供应商 Provider 抽象路由模块的第一步是把每个上游服务抽象成统一的 Provider 对象。不同供应商的差异包括请求路径、模型名、鉴权头、超时设置但对外暴露给业务方时这些差异应该被掩盖。定义 RouteTarget 数据结构保存一次转发需要的所有信息# gateway/route.py from dataclasses import dataclass dataclass class RouteTarget: provider_name: str base_url: str upstream_api_key: str upstream_model: strRouter 的核心职责是根据模型别名解析出 RouteTarget。这里有个关键取舍是每次请求实时查询数据库还是启动时把路由表加载到内存对于路由表这种变化频率低但查询频率极高的数据推荐启动时加载或者用缓存并监听变更。每次请求都查数据库在高并发下会白白消耗数据库连接。下面的示例使用数据库查询是为了保证代码最小可运行实际生产环境需要改为缓存。3.2 模型路由表与优先级策略模型路由表解决的是“外部模型别名”和“真实上游模型”的映射。常见的做法是给每个模型别名配置多条路由每条路由指向不同供应商并用 priority 表示优先级。例如对外提供模型别名gpt-4o-mini-primary它可以同时映射为 OpenAI 的gpt-4o-mini和 Azure OpenAI 的gpt-4o-mini优先级分别为 10 和 20。解析时优先选数字小的那条。# gateway/route.py class Router: def __init__(self, session_factory): self.session_factory session_factory def resolve(self, model_alias: str) - RouteTarget: with self.session_factory() as db: routes ( db.query(ModelRoute) .filter( ModelRoute.model_alias model_alias, ModelRoute.enabled 1, ) .order_by(ModelRoute.id.asc()) .all() ) if not routes: raise RouteNotFoundError(fno route for model alias: {model_alias}) for route in routes: provider ( db.query(Provider) .filter(Provider.name route.provider_name) .first() ) if not provider: continue return RouteTarget( provider_nameprovider.name, base_urlprovider.base_url, upstream_api_keyprovider.upstream_api_key, upstream_modelroute.upstream_model, ) raise RouteNotFoundError(fall providers unavailable for alias: {model_alias})这个实现的思路是先按别名找到所有可用路由再按顺序找到第一个有有效供应商配置的路由。这样做的好处是新增一个备用供应商只需要在 model_routes 表里加一行记录不需要改代码。实际项目中priority 字段比 id 排序更适合做层级控制。可以调整为routes ( db.query(ModelRoute) .filter( ModelRoute.model_alias model_alias, ModelRoute.enabled 1, ) .order_by(ModelRoute.priority.asc(), ModelRoute.id.asc()) .all() )priority 数字越小优先级越高这与常见运维习惯一致便于把主供应商设置为 10、备用设置为 20。3.3 故障转移与降级路由解析只是找到了目标真正发送请求时还需要考虑失败情况。上游可能返回 5xx、超时、限流甚至短暂不可用。网关不能因为单一供应商故障就导致整个调用失败要有故障转移能力。下面是一个带失败转移的转发逻辑示例# gateway/route.py class UpstreamError(Exception): pass class Router: def __init__(self, session_factory, client): self.session_factory session_factory self.client client async def call_with_failover(self, alias: str, payload: dict) - dict: targets self.resolve_all(alias) last_error: Exception | None None for target in targets: try: return await self._post(target, payload) except UpstreamError as exc: last_error exc logger.warning( model alias %s, provider %s failed: %s, alias, target.provider_name, exc, ) continue raise UpstreamError(fall providers failed for alias {alias}, last error: {last_error}) async def _post(self, target: RouteTarget, payload: dict) - dict: url target.base_url.rstrip(/) /v1/chat/completions headers {Authorization: fBearer {target.upstream_api_key}} request_payload { model: target.upstream_model, messages: payload.get(messages, []), temperature: payload.get(temperature, 1.0), stream: payload.get(stream, False), } try: resp await self.client.post(url, jsonrequest_payload, headersheaders) except httpx.TimeoutException as exc: raise UpstreamError(timeout) from exc if resp.status_code 500: raise UpstreamError(fupstream status {resp.status_code}) return resp.json()resolve_all 需要扩展为返回该别名下的全部可用 RouteTarget主备顺序由 priority 决定。这样当前一个供应商调用失败时会自动切换到下一个。故障转移不能只依赖异常还需要对返回状态码做判断否则上游返回 500 时请求会被误当作成功。这里要特别提醒故障转移是一把双刃剑。它提升了可用性但也会让同一个请求在多个供应商之间重试。对于模型调用这种耗时较长、费用较高的请求重试次数必须做上限并且要记录每次重试的请求 ID否则很容易造成费用翻倍。4. 守卫模块请求进来先过三关4.1 API Key 校验与租户识别守卫模块的目标是回答三个问题你是谁、你能用多少、你能不能传这个内容。第一个问题是身份认证。Zerker 使用简单的 Bearer Token 方式调用方在 Authorization 头里传入网关下发的 API Key。这个 Key 并不对应上游供应商而是对应数据库里的 Tenant 记录。# gateway/guard.py from fastapi import HTTPException, Request class GuardPipeline: def __init__(self, session_factory): self.session_factory session_factory def authenticate(self, request: Request) - Tenant: auth_header request.headers.get(Authorization, ) api_key auth_header.replace(Bearer , ).strip() if not api_key: raise HTTPException(status_code401, detailmissing api key) with self.session_factory() as db: tenant db.query(Tenant).filter(Tenant.api_key api_key).first() if not tenant: raise HTTPException(status_code401, detailinvalid api key) return tenant这里有几个细节值得注意。第一API Key 在数据库里应该保存哈希值而不是明文。这样即使数据库泄露攻击者也无法直接用 Key 调用网关。第二校验失败时返回 401不要返回过于具体的错误信息避免给攻击者提供排查线索。第三租户识别要在路由之前完成因为后续限流、计费都要基于租户身份。4.2 限流与配额检查第二个问题是“你能用多少”。限流通常分为两类速率限制和配额限制。速率限制关心的是每秒/每分钟能发多少请求配额限制关心的是这个月总共能消耗多少 token 或金额。教材中最简单的实现是内存滑动窗口# gateway/guard.py import time class InMemoryRateLimiter: def __init__(self, limit_per_minute: int 60): self.limit_per_minute limit_per_minute self._hits: dict[int, list[int]] {} def check(self, tenant_id: int) - None: now int(time.time()) window [ts for ts in self._hits.get(tenant_id, []) if now - ts 60] window.append(now) self._hits[tenant_id] window if len(window) self.limit_per_minute: raise HTTPException(status_code429, detailrate limit exceeded)这个实现的缺点很明显内存状态在网关重启后会丢失多实例部署时每个实例的计数是独立的无法做到全局限流。生产环境需要一个集中式的存储Redis 是常见选择。用 Redis 实现滑动窗口的示例思路如下以 tenant_id 和当前分钟为 Key对请求计数递增并设置过期时间。读取计数前检查是否超过阈值。这种方式在网关横向扩容后依然能保证全局统计正确。配额检查则是在请求转发前判断租户余额是否足够。Zerker 的简化做法是直接在 tenant 表维护 balance 字段每次调用后扣减。但真实的配额管理建议使用独立的配额表并按“预扣 - 结算 - 回滚”的方式处理避免请求过程中产生的并发问题。4.3 上游密钥保护与内容安全审核第三个问题是“你能传什么、不能传什么”。这一层包含请求内容校验和内容安全过滤。内容安全过滤是 AI 网关里必须重视的一环因为模型服务通常不会替网关方判断某个业务方是否允许发送某些敏感内容。Zerker 的设计是预留审核接口在请求转发前和响应返回后分别调用一次内容安全服务。这里的重点是“预留”而不是在内置代码里写死审核规则否则每次调整规则都要重新发布网关。# gateway/guard.py class ContentGuard: def __init__(self, audit_url: str | None None): self.audit_url audit_url async def inspect(self, text: str) - bool: if not self.audit_url: return True # 调用独立的内容安全服务返回是否通过 # 生产环境需要处理超时、降级和审计日志 return True内容安全服务超时时网关需要决定是放行还是拦截。推荐做法是默认拦截并记录日志避免高风险内容流出。但也要提供降级开关防止审核服务故障拖垮所有正常请求。上游密钥保护是 guard 里最容易理解也最容易做错的部分。正确做法是网关只保存一份上游 Key业务方完全不感知。网关转发请求时把上游 Key 写入转发请求的 Authorization 头但返回给业务方的响应绝不包含上游 Key 或上游请求 URL。日志里也不要打印请求头中的 Authorization 字段。5. 计费模块从一次调用到一条账单5.1 usage 数据从哪来计费的前提是拿到准确的用量数据。大模型 API 的响应体里通常包含 usage 字段{ usage: { prompt_tokens: 120, completion_tokens: 45, total_tokens: 165 } }Zerker 优先信任上游返回的 usage而不是自己用 tokenizer 重新计算。原因有两个一是上游计费就是以它们自己的统计为准自己数出来的数字对不上账单反而麻烦二是重新统计需要引入对应模型的 tokenizer增加了复杂度和出错概率。但如果上游没有返回 usage或者流式请求需要分段累计网关就必须做兜底计算。常见思路是使用 tiktoken 等开源工具对文本做估算。注意这里必须明确标注是估算不能和上游计费混为一谈。流式请求的计费要更复杂。响应不是一次性返回的网关需要在流式数据流中累计 usage 片段。很多供应商会在流式响应的最后一个 chunk 里附带 usage网关要等流结束后再落账。5.2 价格表与费用计算费用计算要解决“一个请求花了多少钱”。价格通常按每 1000 个 token 计费且输入和输出的单价不同。Zerker 使用一个价格表配置# config.py PRICE_TABLE { gpt-4o-mini: { input_per_1k: 0.00015, output_per_1k: 0.00060, }, claude-3-5-sonnet: { input_per_1k: 0.003, output_per_1k: 0.015, }, }实际价格会随供应商政策变化写死到代码里主要用于演示。生产环境应该把价格表放到数据库或配置中心并且支持按租户覆盖价格因为不同客户能拿到的折扣可能不同。# gateway/charge.py class ChargeService: def __init__(self, price_table: dict): self.price_table price_table def calculate_fee(self, model: str, prompt_tokens: int, completion_tokens: int) - float: price self.price_table.get(model) if not price: # 没有价格配置时不直接报错记录为 0 费用避免阻断请求 return 0.0 fee ( prompt_tokens / 1000 * price.get(input_per_1k, 0) completion_tokens / 1000 * price.get(output_per_1k, 0) ) return round(fee, 6)这段代码里有个很重要的取舍当某个模型没有配置价格时网关不应该直接报 500否则会因为计费模块的问题阻断正常业务。正确做法是费用记为 0同时记录告警让运营人员去补价格配置。5.3 写库与余额扣减计算出费用后需要创建账单记录并扣减租户余额。这两个操作必须放在同一个数据库事务里否则可能出现账单写了、余额没扣或者余额扣了、账单丢失的问题。# gateway/charge.py from uuid import uuid4 from models import BillRecord class ChargeService: def record_usage( self, tenant: Tenant, model: str, usage: dict, ) - float: prompt_tokens usage.get(prompt_tokens, 0) completion_tokens usage.get(completion_tokens, 0) fee self.calculate_fee(model, prompt_tokens, completion_tokens) request_id uuid4().hex with self.session_factory() as db: db_tenant db.query(Tenant).filter(Tenant.id tenant.id).first() db_tenant.balance round(db_tenant.balance - fee, 6) bill BillRecord( tenant_idtenant.id, request_idrequest_id, modelmodel, prompt_tokensprompt_tokens, completion_tokenscompletion_tokens, feefee, ) db.add(bill) db.commit() return fee余额扣减的逻辑里一个值得思考的问题是到底应该先扣费再转发还是转发成功后再扣费如果先扣费但上游调用失败需要回滚或退款如果转发成功后再扣遇到高并发会出现余额超扣。生产环境更稳妥的方案是“额度预占”请求开始时冻结一部分额度请求结束后按实际用量结算失败时释放冻结额度。Zerker 的简化实现直接在后置阶段扣减适合内部工具或低并发场景生产环境需要升级。6. 把三个模块串起来FastAPI 入口与验证6.1 请求入口代码现在把三个模块串成一条完整的链路。FastAPI 的请求入口代码如下# main.py from fastapi import FastAPI, Request from fastapi.responses import JSONResponse from pydantic import BaseModel, Field from gateway.guard import GuardPipeline from gateway.route import Router from gateway.charge import ChargeService from config import PRICE_TABLE app FastAPI(titleZerker AI Gateway) guard GuardPipeline(session_factorysession_factory) router Router(session_factorysession_factory, clienthttpx.AsyncClient(timeout60)) charge ChargeService(price_tablePRICE_TABLE) class ChatRequest(BaseModel): model: str Field(..., description模型别名) messages: list[dict] stream: bool False temperature: float | None 1.0 class ChatResponse(BaseModel): request_id: str provider: str usage: dict response: dict app.post(/v1/chat/completions) async def chat_completions(req: ChatRequest, http_request: Request): tenant guard.authenticate(http_request) guard.check_rate_limit(tenant.id) try: resp_data await router.call_with_failover( req.model, payloadreq.model_dump(), ) except RouteNotFoundError as exc: return JSONResponse(status_code404, content{detail: str(exc)}) except UpstreamError as exc: return JSONResponse(status_code502, content{detail: str(exc)}) usage resp_data.get(usage, {}) fee charge.record_usage(tenant, req.model, usage) return { request_id: uuid4().hex, provider: resolved, fee: fee, usage: usage, response: resp_data, }这段代码展示了完整的调用链路但还有一个明显缺陷真实网关不应把上游响应原样返回给业务方尤其是当响应中包含内部调试信息时。生产环境需要做响应裁剪只保留业务方需要的字段。另一个需要注意的点是守卫模块的“内容安全审核”在这里没有画出来。正式实现时它应该放在 authenticate 之后、route 之前对 messages 文本做检查对生成结果的审核则需要放在 upstream 返回之后、写计费之前。6.2 启动服务并构造调用启动数据库并初始化表结构。为了快速验证可以使用 SQLite# database.py from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from models import Base engine create_engine(sqlite:///./zerker.db, connect_args{check_same_thread: False}) SessionLocal sessionmaker(bindengine) Base.metadata.create_all(bindengine) def get_session(): return SessionLocal()提前写入一条测试路由和租户数据INSERT INTO providers (name, base_url, upstream_api_key, priority) VALUES (openai, https://api.openai.com, sk-upstream-demo-key, 1); INSERT INTO model_routes (model_alias, provider_name, upstream_model, enabled) VALUES (gpt-4o-mini, openai, gpt-4o-mini, 1); INSERT INTO tenants (name, api_key, balance) VALUES (demo-tenant, zk-demo-tenant-key-123, 100.0);启动网关uvicorn main:app --reload --port 8000调用接口curl -X POST http://127.0.0.1:8000/v1/chat/completions \ -H Authorization: Bearer zk-demo-tenant-key-123 \ -H Content-Type: application/json \ -d { model: gpt-4o-mini, messages: [{role: user, content: say hello}], stream: false }如果上游配置的是有效的 API Key这个请求会返回大模型生成的响应同时生成一条 BillRecord。6.3 验证路由结果、守卫行为和计费记录验证分三个维度。路由是否正确可以观察网关日志或响应中的 provider 字段确认请求被转发到了预期的供应商。如果 provider 解析失败会返回 404。守卫是否生效用错误 Key 再调用一次curl -X POST http://127.0.0.1:8000/v1/chat/completions \ -H Authorization: Bearer wrong-key \ -H Content-Type: application/json \ -d {model: gpt-4o-mini, messages: [{role: user, content: hi}]}预期返回 401。连续调用超过限流阈值预期返回 429。计费是否生效查询数据库sqlite3 zerker.db select tenant_id, model, prompt_tokens, completion_tokens, fee from bill_records;并确认 tenants 表中对应租户的 balance 减少了。这里的 fee 是模拟价格算出来的真实项目中需要和上游账单核对。不要只验证“能返回结果”。网关的正确性在于异常路径和计量路径至少要验证 401、429、404、502 这几个错误分支以及计费明细是否与 usage 一致。7. 常见问题排查现象、原因、处理7.1 请求总是路由到同一个模型现象配置了多个供应商和优先级但请求总是走到第一个供应商即使它已经报错。可能原因路由表的 order_by 没有按预期生效或者故障转移逻辑里的 resolve_all 返回列表不是预期的顺序又或者上游返回的是 4xx 错误但故障转移只处理了 5xx 和超时。检查方式先看 model_routes 表里该别名的全部记录和 priority 值再打开网关日志确认每次调用选择的 provider 是什么最后确认上游返回状态码。处理建议把故障转移条件写清楚是只要非 2xx 就重试还是只在 5xx、超时、网络错误时重试。对模型调用来说4xx 通常是请求参数错误重试没有意义不应触发转移。7.2 明明没超量却频繁被限流现象客户端每秒请求不超过限流阈值但还是收到 429。可能原因限流的 Key 维度不对例如把不同租户的请求归到了同一个 Key多实例部署时内存限流状态独立限流窗口按分钟但是客户端固定每秒打点计数在窗口边界发生重叠或者时钟不一致导致 Redis 过期时间计算错误。检查方式先看限流代码里的 Key 是什么再确认网关是否多实例最后检查 Redis 里当前 Key 的计数。处理建议限流 Key 必须包含租户 ID 和模型别名两个维度不能用 IP 或全局计数。多实例环境必须使用集中式存储不能使用内存计数。对分钟级窗口建议加上随机偏移避免大量请求在窗口边界集中触发。7.3 token 计数与上游账单不一致现象网关记录的 bill_records 中 token 数和上游供应商账单不一致。可能原因网关使用了本地 tokenizer 估算与上游统计口径不同流式请求只统计了最后一次 chunk 的 usage请求被故障转移重试后只有最后一条成功记录被计费但前几次失败请求可能也产生了费用。检查方式找到同一请求 ID比对网关日志、上游账单、数据库记录三方的 prompt_tokens 和 completion_tokens。处理建议优先使用上游返回的 usage流式请求等 final chunk 到达后再落账。对重试逻辑增加“请求是否已经发生过上游调用”的标记避免重复计费。无法从上游拿到 usage 时在网关日志中明确标记“估算值”不要混入正式账单。7.4 API Key 校验通过但无法调用现象调用方拿到的 Key 能通过 authenticate但请求仍然失败。可能原因租户余额不足guard 层虽然没写余额检查但配额检查拦截了或者模型别名没有配置路由或者供应商 Key 失效导致上游返回 401。检查方式检查网关日志里用户 ID、模型别名、余额。再测试上游 API Key 是否有效。处理建议把 Key 的有效性、余额、路由、上游 Key 四层状态分开排查。网关应记录请求流水至少包含租户 ID、模型别名、路由目标、上游状态码、错误信息这样排错时不用猜。8. 生产化建议与扩展方向8.1 从单机到分布式文章中的示例是单机教学版本生产环境至少要补齐四块能力配置外置化、集中式限流、异步落账、幂等控制。配置外置化是指供应商 Key、价格表、限流阈值不要写死在代码或本地配置文件里应该放到环境变量、配置中心或数据库。特别是上游 API Key一旦被提交到 Git 仓库就必须立即轮换。集中式限流使用 Redis 或类似组件替代内存限流保证多实例网关的计数全局一致。异步落账是指计费写入不应阻塞请求返回可以使用消息队列或本地异步任务。幂等控制要求同一个业务请求 ID 在重试时不会重复计费。8.2 可观测性与审计网关是请求流量的咽喉必须具有完整的可观测性。建议为每个请求生成唯一的 request_id并把它透传到上游请求头和下游响应中。日志至少要包含以下字段字段含义request_id请求唯一标识tenant_id租户标识model_alias外部模型别名route_provider实际选择的供应商upstream_status上游状态码prompt_tokens输入 token 数completion_tokens输出 token 数fee计算费用duration_ms请求耗时具备这些日志后才可以回答常见的运营问题哪个应用消耗最多、哪个模型最贵、哪个供应商最近故障率上升。审计日志和业务日志要分开。审计日志记录谁在什么时间调用了什么模型、传了什么内容用于安全追溯。业务日志用于排查链路问题。审计日志的保存时间通常更长且需要权限控制。8.3 多租户隔离与模型运营建议多租户是 AI Gateway 的价值放大器但也带来额外复杂度。不同租户需要有独立的 Key、独立的余额、独立的价格策略甚至独立的模型路由范围。最简单的方式是在所有核心表上增加 tenant_id 维度查询时强制带上租户过滤条件避免数据串用。模型运营层面建议为每个模型别名建立独立的健康检查和指标看板。例如监控某个模型别名的平均耗时、错误率、失败转移次数和费用走势。这样当某家供应商质量下降时运维人员能快速调整路由优先级而不是等下游业务方反馈。对于刚接触 AI Gateway 的开发者建议不要一开始就实现全部功能。先用本文的骨架搭一个最小可用版本跑通“一个模型、一个租户、一次计费”再逐步加入多供应商、内容审核、配额管理、分布式限流。网关的复杂度会随着接入方的增加指数上升前期把路由、守卫、计费三个模块的边界划清楚后面才能稳定迭代。