企业官网建设流程全解析

在实际的 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，如果业务服务直接持有这些 Key，Key 的保存、轮换、权限回收都很难控制，泄露风险也会被放大。第三是没法统一计量。不同团队调用同一个模型，产生的 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，路由、守卫、计费的设计思路是一致的，本文的代码只负责把思路讲清楚。

最小依赖如下：

fastapi>=0.110 uvicorn[standard]>=0.29 httpx>=0.27 sqlalchemy>=2.0 pydantic>=2.6

把这些写入 requirements.txt：

fastapi==0.110.0 uvicorn[standard]==0.29.0 httpx==0.27.0 sqlalchemy==2.0.29 pydantic==2.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_key=True) name = Column(String(64), unique=True, nullable=False) api_key = Column(String(64), unique=True, nullable=False, index=True) balance = Column(Float, default=100.0) created_at = Column(DateTime, server_default=func.now()) class Provider(Base): __tablename__ = "providers" id = Column(Integer, primary_key=True) name = Column(String(32), unique=True, nullable=False) base_url = Column(String(255), nullable=False) upstream_api_key = Column(String(255), nullable=False) priority = Column(Integer, default=100) class ModelRoute(Base): __tablename__ = "model_routes" id = Column(Integer, primary_key=True) model_alias = Column(String(64), index=True, nullable=False) provider_name = Column(String(32), nullable=False) upstream_model = Column(String(64), nullable=False) enabled = Column(Integer, default=1) class BillRecord(Base): __tablename__ = "bill_records" id = Column(BigInteger, primary_key=True) tenant_id = Column(Integer, ForeignKey("tenants.id"), nullable=False) request_id = Column(String(64), unique=True, nullable=False) model = Column(String(64), nullable=False) prompt_tokens = Column(Integer, default=0) completion_tokens = Column(Integer, default=0) fee = Column(Float, default=0.0) created_at = Column(DateTime, server_default=func.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: str

Router 的核心职责是根据模型别名解析出 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(f"no 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_name=provider.name, base_url=provider.base_url, upstream_api_key=provider.upstream_api_key, upstream_model=route.upstream_model, ) raise RouteNotFoundError(f"all 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(f"all 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": f"Bearer {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, json=request_payload, headers=headers) except httpx.TimeoutException as exc: raise UpstreamError("timeout") from exc if resp.status_code >= 500: raise UpstreamError(f"upstream 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_code=401, detail="missing 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_code=401, detail="invalid 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_code=429, detail="rate 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_id=tenant.id, request_id=request_id, model=model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, fee=fee, ) 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(title="Zerker AI Gateway") guard = GuardPipeline(session_factory=session_factory) router = Router(session_factory=session_factory, client=httpx.AsyncClient(timeout=60)) charge = ChargeService(price_table=PRICE_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, payload=req.model_dump(), ) except RouteNotFoundError as exc: return JSONResponse(status_code=404, content={"detail": str(exc)}) except UpstreamError as exc: return JSONResponse(status_code=502, 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(bind=engine) Base.metadata.create_all(bind=engine) 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，并把它透传到上游请求头和下游响应中。日志至少要包含以下字段：

具备这些日志后，才可以回答常见的运营问题：哪个应用消耗最多、哪个模型最贵、哪个供应商最近故障率上升。

审计日志和业务日志要分开。审计日志记录谁在什么时间调用了什么模型、传了什么内容，用于安全追溯。业务日志用于排查链路问题。审计日志的保存时间通常更长，且需要权限控制。

8.3 多租户隔离与模型运营建议

多租户是 AI Gateway 的价值放大器，但也带来额外复杂度。不同租户需要有独立的 Key、独立的余额、独立的价格策略，甚至独立的模型路由范围。最简单的方式是在所有核心表上增加 tenant_id 维度，查询时强制带上租户过滤条件，避免数据串用。

模型运营层面，建议为每个模型别名建立独立的健康检查和指标看板。例如监控某个模型别名的平均耗时、错误率、失败转移次数和费用走势。这样当某家供应商质量下降时，运维人员能快速调整路由优先级，而不是等下游业务方反馈。

对于刚接触 AI Gateway 的开发者，建议不要一开始就实现全部功能。先用本文的骨架搭一个最小可用版本，跑通“一个模型、一个租户、一次计费”，再逐步加入多供应商、内容审核、配额管理、分布式限流。网关的复杂度会随着接入方的增加指数上升，前期把路由、守卫、计费三个模块的边界划清楚，后面才能稳定迭代。