尧图网页设计 
在业务系统接入大模型之后最先出现的往往不是模型效果问题而是“所有模型调用都各自为政”同样的鉴权逻辑在不同服务里重复实现、每个团队自己处理限流与重试、线上出现 429 时无法统一降级、月底账单出来才发现成本根本没人能说清楚来源。这些问题积累到一定程度团队自然会想到引入一个统一的 LLM 网关。本文围绕 LLM 网关生产化这一主题从概念、核心职责、架构权衡、最小可落地实现、常见问题到工程教训展开一套可以照着用的实践思路。适合正在做 LLM 应用工程化、需要统一管理多个模型服务、或者准备搭建内部 AI 中台的后端开发者阅读。读完你可以掌握 LLM 网关的基本架构拆分方式并能基于 FastAPI 快速搭建一个具备路由、限流、缓存、可观测和成本追踪思路的网关原型。1. LLM 网关是什么为什么需要生产化1.1 从一次“失控的调用”说起先看一个常见的场景业务线 A 直接调用 OpenAI 的接口业务线 B 接的是 Azure OpenAI业务线 C 则在使用某家国产大模型。每个业务线都有自己的 API Key都自行管理超时和重试。表面上看起来没有什么问题直到模型供应商的某个接口开始抖动A 服务重试了 3 次导致上游请求量瞬间翻倍B 服务没有设置超时数据库连接池被慢请求耗尽C 服务把 API Key 明文写在了代码仓库里月底财务要分摊成本谁也无法提供精确的 token 用量。这些问题的根源是“模型调用”没有被当成一类需要统一治理的基础设施流量来对待。LLM 网关本质上就是这层统一入口所有模型请求都经过它再由它转发到不同的模型服务商。从专业角度定义LLM 网关是位于应用与 LLM 后端之间的一层代理服务负责统一处理认证、路由、限流、重试、缓存、可观测和成本计量。它解决的问题不是让模型变聪明而是让模型调用变得可控、可观测、可治理。1.2 生产化与 Demo 的本质区别很多团队在原型阶段就写了一个简单的 Proxy 脚本转发请求到 OpenAI觉得“这就够了”。但在生产环境里一个 LLM 网关必须回答下面这些问题如果上游服务超时网关是重试还是快速失败多个业务共用网关如何分租户限流流式响应场景下客户端断连时网关如何释放后端连接如何追踪一次完整的请求链路Key 泄露后如何快速隔离模型供应商升级接口版本如何做到业务无感这些问题都不可能在“转发请求”这一层得到解决必须在一开始做架构设计时就考虑清楚。因此“LLM 网关生产化”并不是简单加一层代理而是要做一次面向治理能力的基础设施建设。1.3 适用场景LLM 网关最典型的落地场景包括这几类多业务线共享同一批模型资源需要统一分配配额和成本公司需要接入多家大模型供应商并在故障时自动切换网络安全团队要求所有外部模型请求都必须经过审计业务方需要缓存常见 Prompt 的响应来降低成本和延迟平台团队要统一收集 token 消耗用于财务分账。如果只是个人开发者在笔记本上调用 API网关不是必须项。但一旦模型调用开始影响多团队、多系统网关就应该被提上日程。2. LLM 网关的核心职责与架构定位2.1 六大核心职责在设计网关时不要把它当成一个简单的 Reverse Proxy而是从六个维度拆解职责。一是认证与授权。网关需要统一管理上游模型的 API Key、租户身份、调用权限。业务方不直接接触模型密钥而是使用网关签发的 Token 或服务账号。这样即使某个业务项目代码泄露也不会直接暴露供应商密钥。二是请求路由。根据请求参数例如 model 字段、项目 ID、目标部署环境网关决定把请求发往哪一个具体模型或供应商。路由规则可以是静态映射也可以支持灰度比例和故障转移。三是流量治理。包括限流、熔断、超时控制、重试策略。例如某个租户每秒钟只能调用 100 次某个上游连续 5 次超时则触发熔断所有重试必须带指数退避。四是语义缓存。大模型调用通常有延迟和成本。如果多个用户使用相同的 Prompt并且结果可以被安全复用例如知识库问答的基础查询可以在网关层做语义缓存用 embedding 计算文本相似度来判断是否命中。五是可观测性。网关是流量的必经之路非常适合输出日志、指标和链路数据。包括每次请求的模型名称、Token 使用量、延迟、状态码、缓存命中情况、租户成本等。六是成本计量。把每次调用折算成单价并按租户、业务线、模型类型、时间段聚合方便成本分摊和异常消费告警。2.2 网关与业务服务的边界一个需要明确的边界问题是网关到底做多少业务逻辑常见的错误是网关越做越重最后把 Prompt 模板、知识库检索、Agent 编排逻辑都放进网关里。从架构维护角度我建议网关只关注“横切治理”不关注具体业务逻辑。业务逻辑如 Prompt 拼装、RAG 检索、工具调用都应该留在业务服务内。网关的服务对象是“模型调用”这个动作而不是“业务结果”。保持这个边界网关才能足够稳定并且可以被多个业务方长期依赖。2.3 网关形态选择LLM 网关在部署形态上通常有三种选择。第一种是独立部署的集中式网关例如搭建一个内部服务所有模型请求都请求它。优点是治理能力强权责清晰缺点是增加一跳网络延迟并且网关自身成为高可用重点。第二种是边车Sidecar形态每个业务服务旁边部署一个网关代理。优点是延迟较低故障爆炸半径小缺点是治理策略分散运维成本上升。第三种是直接使用云厂商的 API 网关或托管型 LLM Gateway。优点是开箱即用缺点是可能绑定云厂商且部分高级语义能力仍需自建。这三种形态没有绝对优劣更多是组织规模和运维能力的取舍。对于大多数中小团队集中式网关是性价比最高的起步形态后续如果延迟敏感型业务增多再针对部分服务做边车下沉。3. 生产化前必须想清楚的架构权衡3.1 同步代理还是异步管道LLM 调用和普通 REST API 最大的不同是响应时间波动大。普通接口 50ms 以内是常态而 LLM 接口可能 1 秒到 60 秒不等。于是网关架构必须考虑是以“同步转发”为核心还是以“事件队列 异步任务”为核心同步代理的好处是模型天然贴近用户请求流式返回容易实现代码逻辑简单排错直观。但同步代理对并发和连接管理要求很高每一个上游请求都会占用一个连接如果业务方大量调用网关需要精细控制连接池、超时和并发数。异步管道则是将请求写入消息队列由 Worker 异步调用模型并回传结果。这种设计适合离线批量生成、异步任务等等不需要用户立刻等待的场景但实现复杂度高而且要处理回调通知和结果持久化。我的建议是第一版网关优先实现同步代理因为大多数在线应用都需要流式返回。同时把“任务类型”设计成请求的一个属性为后续异步通道留出扩展位。同步和异步并存是生产化网关的常态但不要把两个模式混在一条代码路径里。3.2 多供应商抽象到什么程度很多团队接入第一个模型服务后会立即思考“未来要支持多家怎么办”。于是开始设计一个高层抽象Provider 接口、统一模型枚举、统一错误码。这个方向本身没错但要小心过度设计。多供应商抽象的核心是把不变的部分固定下来把变化的部分通过配置驱动。实际经验是不要一开始就试图把手上的所有模型 API 统一成一个格式。更好的做法是对业务提供一套“内部模型名”到“供应商真实模型名”的映射对响应体做统一封装至少要统一错误结构和 Token 用量字段对上游 SDK 或 HTTP Client 做薄封装不要把所有差异化参数都提前抽象掉。因为不同的模型供应商在参数细节上差异很大例如部分供应商的temperature范围不同或者不支持logprobs。过早抽象会导致网关层频繁发布版本稳定性下降。3.3 缓存命中与成本/延迟的权衡缓存能显著降低成本但也有三个很现实的坑。第一个坑是动态 Prompt 导致缓存命中率低。如果请求参数中带时间、用户 ID、随机数即使语义相同也无法命中。第二个坑是缓存污染。如果同一个缓存 Key 先被低质量模型的回答命中后续高质量模型永远没有机会被调用用户体验会受影响。第三个坑是数据安全。某些业务 Prompt 包含敏感信息不应写入 Redis 或对象存储。因此网关的缓存策略不应该默认对所有请求生效而应该按路由规则配置。通常建议对这类请求开启语义缓存读多写少、Prompt 相对固定、结果允许过期、不涉及敏感数据。普通随机性较强的 Prompt 不要缓存。3.4 流式与超时控制LLM 网关必须支持流式否则会让用户首 token 延迟变得不可接受。但流式与网关治理之间有冲突网关需要在拿到完整响应后记录 token 用量但流式情况下响应体是一块块到达的无法同时做到“一边转发一边精确计量”。实际设计中流式请求的用量记录通常有两种做法。一种是在业务侧传入预估参数或从响应中解析usage字段但并非所有供应商都会在每个 chunk 里带上 usage 信息。另一种是通过网关对内容流做累计统计估算 token 数。生产环境可以两者结合消息头里如果有usage就优先用没有则用内容长度估算。请求超时也要分级设计而不要只设置一个全局超时。例如连接超时 3 秒、首 token 超时 10 秒、整体响应超时 60 秒。不同模型和场景的超时策略差别很大建议在路由配置中为每个上游单独设置。4. 最小可落地 LLM 网关实现下面我们来动手实现一个最小可落地的 LLM 网关。示例会尽量简单方便你理解核心流程代码可以直接复制运行但生产环境还需要按团队规范扩充。4.1 技术选型与环境准备示例采用 Python FastAPI httpx原因有三FastAPI 对异步支持好适合转发流式响应httpx 支持 HTTP/2 和流式传输和 FastAPI 配合自然Python 生态便于快速加入 Redis、Prometheus 等组件。环境要求如下版本以你本机实际安装为准Python 3.10 或更高版本FastAPI 0.100uvicorn 0.23httpx 0.24安装依赖pip install fastapi uvicorn httpx pydantic-settings示例项目结构llm-gateway/ ├── app │ ├── __init__.py │ ├── main.py # 网关入口 │ ├── config.py # 配置管理 │ ├── models.py # 请求/响应模型 │ ├── provider.py # 上游供应商抽象 │ ├── cache.py # 简单 TTL 缓存 │ └── middleware.py # 限流与日志 └── .env.example4.2 创建配置管理先在app/config.py中定义网关配置。使用pydantic-settings读取环境变量避免把密钥写死在代码中。# 文件路径app/config.py from functools import lru_cache from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) # 网关服务端口 gateway_host: str 0.0.0.0 gateway_port: int 8000 # 示例用单一上游生产环境通常有一张路由表 default_upstream: str https://api.openai.com default_model: str gpt-4o-mini # 上游 API Key生产环境建议从密钥管理服务拉取 upstream_api_key: str # 限流配置单租户每秒最大请求数 rate_limit_per_second: int 10 # 缓存配置TTL 秒数0 表示关闭 cache_ttl_seconds: int 0 # 超时配置 connect_timeout_seconds: float 3.0 first_token_timeout_seconds: float 10.0 total_timeout_seconds: float 60.0 lru_cache def get_settings() - Settings: return Settings()在.env.example中# 复制为 .env 后填写 UPSTREAM_API_KEYsk-xxxx DEFAULT_UPSTREAMhttps://api.openai.com DEFAULT_MODELgpt-4o-mini RATE_LIMIT_PER_SECOND10 CACHE_TTL_SECONDS0这里有一个生产实践要强调环境变量不要提交进 Git 仓库。upstream_api_key是敏感信息建议在 CI/CD 流程中通过密钥管理平台注入。4.3 定义请求与响应模型网关需要对外提供统一的请求结构。为了兼容 OpenAI 接口风格我们直接使用类似 ChatCompletion 的格式这样业务方从直连切换到网关的成本最低。# 文件路径app/models.py from typing import Any, Dict, List from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: str content: str model_config {extra: allow} class ChatCompletionRequest(BaseModel): model: str Field(defaultgpt-4o-mini) messages: List[ChatMessage] temperature: float | None None max_tokens: int | None None stream: bool False model_config {extra: allow} class Usage(BaseModel): prompt_tokens: int 0 completion_tokens: int 0 total_tokens: int 0 class ChatCompletionResponse(BaseModel): id: str object: str chat.completion created: int model: str choices: List[Dict[str, Any]] usage: Usage | None None在真实网关中你还需要把供应商返回的各种字段映射成统一结构。这里为了说明核心流程只定义最常用的几个字段。model_config {extra: allow}用于透传额外参数避免网关因为多了一个供应商扩展字段而报错。4.4 封装上游供应商网关的核心逻辑都在 Provider 这一层。这里实现一个OpenAICompatibleProvider它使用httpx.AsyncClient转发/chat/completions请求并支持超时和流式处理。# 文件路径app/provider.py import uuid import time from typing import AsyncIterator import httpx from app.config import Settings from app.models import ChatCompletionRequest class OpenAICompatibleProvider: 面向 OpenAI 兼容接口的上游 Provider。 其他供应商如 Claude可以通过实现同样的 call/stream 方法接入。 def __init__(self, settings: Settings): self.settings settings self.headers { Authorization: fBearer {settings.upstream_api_key}, Content-Type: application/json, } self._client httpx.AsyncClient(timeoutsettings.total_timeout_seconds) async def call(self, req: ChatCompletionRequest) - dict: 非流式调用返回完整响应 JSON。 url f{self.settings.default_upstream}/v1/chat/completions payload req.model_dump(exclude_noneTrue) resp await self._client.post(url, jsonpayload, headersself.headers) resp.raise_for_status() return resp.json() async def stream(self, req: ChatCompletionRequest) - AsyncIterator[bytes]: 流式调用把上游响应以 bytes 形式逐块返回。 httpx 的 aiter_bytes 会在流结束时自动关闭响应体。 url f{self.settings.default_upstream}/v1/chat/completions payload req.model_dump(exclude_noneTrue) async with self._client.stream( POST, url, jsonpayload, headersself.headers ) as resp: resp.raise_for_status() async for chunk in resp.aiter_bytes(): yield chunk上面的实现思路里有两个生产要点httpx.AsyncClient应该被复用否则每个请求都创建新连接会非常浪费stream方法使用async with self._client.stream()确保响应流关闭。4.5 实现缓存与限流缓存可以先用内存字典实现一个最小 TTL 版本。生产环境建议换成 Redis但思路是一样的。# 文件路径app/cache.py import time import threading from typing import Any class TTLCache: def __init__(self, ttl_seconds: int 0, max_size: int 1024): self.ttl_seconds ttl_seconds self.max_size max_size self._data: dict[str, tuple[float, Any]] {} self._lock threading.Lock() def _is_expired(self, key: str) - bool: if key not in self._data: return True expire_at, _ self._data[key] return time.time() expire_at def get(self, key: str): with self._lock: if self._is_expired(key): self._data.pop(key, None) return None return self._data[key][1] def set(self, key: str, value: Any): if self.ttl_seconds 0: return with self._lock: if len(self._data) self.max_size: # 简单淘汰清掉所有过期 key for k in list(self._data.keys()): if self._is_expired(k): self._data.pop(k, None) self._data[key] (time.time() self.ttl_seconds, value) def make_key(self, req) - str: 生产环境建议在这里加入租户 ID避免跨租户串数据。 messages req.messages # 序列化时排除可能造成缓存击穿的随机参数 import json return json.dumps({ model: req.model, messages: [m.model_dump() for m in messages], temperature: req.temperature, }, ensure_asciiFalse, sort_keysTrue)缓存 key 是重点。如果直接把整个请求体包含进来那么任何细微参数变化都会导致不命中。建议只使用影响语义结果的字段model、messages、temperature。限流模块用token bucket算法做一个简单的内存限流器按租户维度限流# 文件路径app/middleware.py import time import threading class TokenBucketLimiter: def __init__(self, rate: int): rate: 每秒允许通过的请求数。 self.rate rate self.tokens rate self.last time.monotonic() self._lock threading.Lock() def acquire(self, tokens: int 1) - bool: with self._lock: now time.monotonic() elapsed now - self.last self.tokens min(self.rate, self.tokens elapsed * self.rate) self.last now if self.tokens tokens: self.tokens - tokens return True return False限流返回的错误应该统一否则客户端无法处理。通常生产网关会返回429 Too Many Requests并提供Retry-After头部。4.6 创建网关主入口现在把以上模块组合起来生成 FastAPI 应用。# 文件路径app/main.py import time import uuid from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, StreamingResponse from app.cache import TTLCache from app.config import get_settings from app.middleware import TokenBucketLimiter from app.models import ChatCompletionRequest, ChatCompletionResponse from app.provider import OpenAICompatibleProvider settings get_settings() app FastAPI(titleLLM Gateway, version0.1.0) provider OpenAICompatibleProvider(settings) cache TTLCache(ttl_secondssettings.cache_ttl_seconds) limiter TokenBucketLimiter(ratesettings.rate_limit_per_second) app.middleware(http) async def add_request_id(request: Request, call_next): request_id request.headers.get(X-Request-ID, str(uuid.uuid4())) start time.time() response await call_next(request) response.headers[X-Request-ID] request_id response.headers[X-LLM-Gateway-Version] 0.1.0 # 记录基础访问日志 print({ request_id: request_id, path: request.url.path, status: response.status_code, duration_ms: round((time.time() - start) * 1000, 2), }) return response app.get(/health) async def health(): return {status: ok} app.post(/v1/chat/completions) async def chat_completions(req: ChatCompletionRequest, request: Request): # 租户信息可以从 Header 传入例如 X-Tenant-Id tenant request.headers.get(X-Tenant-Id, default) # 限流 if not limiter.acquire(): return JSONResponse( status_code429, content{ error: { message: rate limit exceeded, type: rate_limit, request_id: request.state.__dict__.get(request_id), } }, headers{Retry-After: 1}, ) # 缓存逻辑仅针对非流式请求 if not req.stream: cache_key cache.make_key(req) cached cache.get(cache_key) if cached: return JSONResponse(contentcached) # 转发到上游 if req.stream: async def event_stream(): async for chunk in provider.stream(req): yield chunk return StreamingResponse(event_stream(), media_typetext/event-stream) try: upstream_data await provider.call(req) except Exception as exc: return JSONResponse( status_code502, content{error: {message: fupstream error: {exc}, type: upstream_error}}, ) # 写入缓存 if not req.stream and settings.cache_ttl_seconds 0: cache.set(cache_key, upstream_data) return JSONResponse(contentupstream_data)需要注意request.state.__dict__在 FastAPI 中并不是推荐做法上面的错误响应里没有真正写入 request_id。更好的方式是把 request_id 放到中间件中并显式赋值给request.state.request_id。下面是修正片段app.middleware(http) async def add_request_id(request: Request, call_next): request_id request.headers.get(X-Request-ID, str(uuid.uuid4())) request.state.request_id request_id ...然后在限流分支中使用request.state.request_id即可。4.7 运行与验证在项目根目录创建.envUPSTREAM_API_KEYsk-your-key启动网关uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload使用 curl 测试非流式请求curl --location http://localhost:8000/v1/chat/completions \ --header Content-Type: application/json \ --header X-Tenant-Id: demo \ --data { model: gpt-4o-mini, messages: [ {role: user, content: 你好请用一句话介绍你自己} ], stream: false }预期会返回上游模型的 JSON 响应。如果上游配置正确响应里应包含choices和usage字段。再测试流式请求curl --location http://localhost:8000/v1/chat/completions \ --header Content-Type: application/json \ --header X-Tenant-Id: demo \ --data { model: gpt-4o-mini, messages: [ {role: user, content: 讲一个简短的笑话} ], stream: true }如果stream为 true你会看到一段 SSE 格式的流式响应通常每个 chunk 都是一个data: {...}数据块最后以data: [DONE]结束。5. 常见问题与排查思路网关上线后运维排错比功能开发更频繁。下面整理几个真实场景中高频出现的问题。5.1 流式请求超时现象客户端通过网关进行流式对话超过 10 秒后才收到第一个 token然后连接被断开。原因大多数供应商在真实生成前也会有排队耗时。如果首 token 超时设置过短会在模型尚未开始返回内容时误杀连接。解决将“连接超时”和“首 token 超时”分开设置。连接超时可以短一些比如 3 秒首 token 超时建议放宽到 15 到 30 秒因为 LLM 的 Prefill 阶段耗时可能长。整体响应超时应根据最大生成长度调整。5.2 重试导致上游重复扣费现象上游返回超时网关自动重试一次但实际第一次请求已经在模型端执行成功结果用户收到两个相同回答成本也翻倍。原因没有做幂等控制。LLM 请求天然不是幂等的除非业务方传入自定义请求 ID并在供应商侧支持去重。解决在网关层默认关闭“不确定类错误”的自动重试。只有当错误码是明确的429或5xx并且确认请求未被执行时才允许重试。对非流式请求可以在重试前用同一个请求 ID 查询上游状态如果供应商不支持就宁可失败也不要盲目重试。5.3 不同模型返回结构不一致现象网关统一封装后业务方反馈解析错误。某个国产模型把usage字段放在别的位置或者choices里没有finish_reason。原因不是所有模型都严格兼容 OpenAI 响应格式尤其是部分自建模型服务。解决每个 Provider 实现内部都要完成“响应标准化”。网关对外暴露的统一模型必须稳定不能把上游的差异直接透传。建议写一个normalize_response方法在call和stream返回前统一处理字段。5.4 缓存导致不同租户数据串线现象租户 A 和租户 B 请求了同样的问题租户 B 直接拿到了租户 A 的回答但回答里包含租户 A 的内部信息。原因缓存 key 只包含 model 和 messages没有隔离租户维度。解决key 中必须加入租户 ID、应用 ID 等隔离字段。更安全的做法是只对明确标记为“可缓存”的租户启用语义缓存默认全局关闭。排查清单可以参考下表问题现象常见原因解决思路网关返回 502上游服务不可达或超时检查连接超时设置查看上游健康状态网关返回 429单租户限流触发调整限流阈值增加队列或降级返回流式响应中断客户端或网关到上游连接被掐断加大首 token 超时开启 TCP KeepAlive响应结果与直连不一致多 Provider 响应未标准化在 Provider 层做字段映射token 用量统计缺失流式响应中未解析 usage 字段对流式 chunks 累计估算或仅记录 non-stream 精确用量6. 生产化教训与最佳实践6.1 不要一开始就追求“万能抽象”第一版 LLM 网关最常见的错误是试图设计一个“能接入全世界所有模型”的抽象层。结果是接口设计被各种模型参数撑得难以维护团队花在抽象上的时间远超接入实际业务的时间。更务实的做法是先接入 1 到 2 个真正使用的模型服务定义好从业务侧到网关的最小公共契约等出现第三个供应商时再根据前两个差异点演进抽象层。这样抽象出来的接口有真实使用场景支撑而不是拍脑袋设计。6.2 网关要与密钥管理、审计联动网关集中保存各类模型密钥安全级别应当等同于堡垒机或数据库账号。生产环境不要依赖.env文件保存长期密钥建议对接 Vault、KMS 或云厂商的 Secret Manager。同时所有经过网关的请求都应该记录审计日志。审计日志至少包含请求时间、租户、调用方 IP、模型名、Token 用量、延迟、供应商请求 ID、错误信息。这不仅是安全需要也是排查问题的关键依据。6.3 从第一天就做成本计量成本计量如果滞后后面补起来会非常痛苦。网关在转发请求时就应该把usage数据写入一个时序数据库或消息队列。建议按以下标签建模tenant_id租户app_id业务应用model_name真实模型名provider供应商request_modestream 或 non-streamcache_hit是否命中缓存这样月底可以快速生成成本报表也能在某个应用出现异常高调用量时及时告警。不要在没有成本数据的网关版本上继续加功能否则后面只能做“估算”无法核实。6.4 灰度与降级预案必须提前设计生产环境的单个模型供应商必然会出现故障或者性能退化。网关发布策略上要支持按租户、流量比例、模型标签做灰度。最简单的实现是路由配置中心化# 伪代码示意生产环境可放到配置中心 routes { default: { provider: openai, model: gpt-4o-mini, }, fallback: { provider: azure, model: gpt-4o-mini-east, }, }当主路由连续失败超过阈值时网关自动把流量切换到备用路由。降级策略应该提前和业务方约定是否允许降低模型规格是否允许返回一个固定兜底回答这些决策涉及业务风险不能全部压给网关开发同学。6.5 监控与排错围绕“链路”展开因为网关增加了一次网络跳转排错难度会变大。建议在中间件中为每个请求注入统一 request_id并在日志里透传到上游。例如通过X-Request-ID和X-LLM-Gateway-Version头传递。监控指标至少要有网关 QPS、成功率、错误码分布各上游的平均延迟、P95/P99 延迟缓存命中率限流触发次数流式请求的平均首 token 延迟。建议接入 Prometheus 指标体系。FastAPI 可以通过prometheus-fastapi-instrumentator快速暴露默认指标再结合自定义 Counter/Histogram 统计模型相关指标。7. 总结与学习路线本文围绕 LLM 网关生产化展开核心可以概括成三句话第一LLM 网关是模型调用进入生产环境的“基础设施”不是简单代理第二架构设计的重点应从功能转向权衡包括同步与异步、抽象与稳定、缓存与成本、流式与计量第三生产化过程中最容易踩坑的不是接口开发而是超时、重试、幂等、成本和安全。建议你从本文的最小网关开始先跑通一条 OpenAI 兼容接口的链路然后补充多租户限流、请求级监控和成本日志。下一步可以继续学习语义缓存的 embedding 相似度计算方法基于 Redis 的分布式限流替代进程内令牌桶网关与 RAG 架构的边界划分支持 Anthropic 或自建模型时如何抽象出不脆弱的标准响应结构结合 Kubernetes 部署时网关的高可用与优雅缩容方案。生产环境没有银弹LLM 网关的每一项能力都需要结合团队现有基础和真实流量反复调整。但有一条经验几乎适用于所有团队把模型调用当成“基础设施流量”来治理越早做越省心。如果你正在搭建或计划搭建 LLM 网关建议先从一条路由、一套日志、一个成本指标开始跑通后再逐步增加能力。 LLM网关生产化：统一模型调用的架构设计与实践
在业务系统接入大模型之后最先出现的往往不是模型效果问题而是“所有模型调用都各自为政”同样的鉴权逻辑在不同服务里重复实现、每个团队自己处理限流与重试、线上出现 429 时无法统一降级、月底账单出来才发现成本根本没人能说清楚来源。这些问题积累到一定程度团队自然会想到引入一个统一的 LLM 网关。本文围绕 LLM 网关生产化这一主题从概念、核心职责、架构权衡、最小可落地实现、常见问题到工程教训展开一套可以照着用的实践思路。适合正在做 LLM 应用工程化、需要统一管理多个模型服务、或者准备搭建内部 AI 中台的后端开发者阅读。读完你可以掌握 LLM 网关的基本架构拆分方式并能基于 FastAPI 快速搭建一个具备路由、限流、缓存、可观测和成本追踪思路的网关原型。1. LLM 网关是什么为什么需要生产化1.1 从一次“失控的调用”说起先看一个常见的场景业务线 A 直接调用 OpenAI 的接口业务线 B 接的是 Azure OpenAI业务线 C 则在使用某家国产大模型。每个业务线都有自己的 API Key都自行管理超时和重试。表面上看起来没有什么问题直到模型供应商的某个接口开始抖动A 服务重试了 3 次导致上游请求量瞬间翻倍B 服务没有设置超时数据库连接池被慢请求耗尽C 服务把 API Key 明文写在了代码仓库里月底财务要分摊成本谁也无法提供精确的 token 用量。这些问题的根源是“模型调用”没有被当成一类需要统一治理的基础设施流量来对待。LLM 网关本质上就是这层统一入口所有模型请求都经过它再由它转发到不同的模型服务商。从专业角度定义LLM 网关是位于应用与 LLM 后端之间的一层代理服务负责统一处理认证、路由、限流、重试、缓存、可观测和成本计量。它解决的问题不是让模型变聪明而是让模型调用变得可控、可观测、可治理。1.2 生产化与 Demo 的本质区别很多团队在原型阶段就写了一个简单的 Proxy 脚本转发请求到 OpenAI觉得“这就够了”。但在生产环境里一个 LLM 网关必须回答下面这些问题如果上游服务超时网关是重试还是快速失败多个业务共用网关如何分租户限流流式响应场景下客户端断连时网关如何释放后端连接如何追踪一次完整的请求链路Key 泄露后如何快速隔离模型供应商升级接口版本如何做到业务无感这些问题都不可能在“转发请求”这一层得到解决必须在一开始做架构设计时就考虑清楚。因此“LLM 网关生产化”并不是简单加一层代理而是要做一次面向治理能力的基础设施建设。1.3 适用场景LLM 网关最典型的落地场景包括这几类多业务线共享同一批模型资源需要统一分配配额和成本公司需要接入多家大模型供应商并在故障时自动切换网络安全团队要求所有外部模型请求都必须经过审计业务方需要缓存常见 Prompt 的响应来降低成本和延迟平台团队要统一收集 token 消耗用于财务分账。如果只是个人开发者在笔记本上调用 API网关不是必须项。但一旦模型调用开始影响多团队、多系统网关就应该被提上日程。2. LLM 网关的核心职责与架构定位2.1 六大核心职责在设计网关时不要把它当成一个简单的 Reverse Proxy而是从六个维度拆解职责。一是认证与授权。网关需要统一管理上游模型的 API Key、租户身份、调用权限。业务方不直接接触模型密钥而是使用网关签发的 Token 或服务账号。这样即使某个业务项目代码泄露也不会直接暴露供应商密钥。二是请求路由。根据请求参数例如 model 字段、项目 ID、目标部署环境网关决定把请求发往哪一个具体模型或供应商。路由规则可以是静态映射也可以支持灰度比例和故障转移。三是流量治理。包括限流、熔断、超时控制、重试策略。例如某个租户每秒钟只能调用 100 次某个上游连续 5 次超时则触发熔断所有重试必须带指数退避。四是语义缓存。大模型调用通常有延迟和成本。如果多个用户使用相同的 Prompt并且结果可以被安全复用例如知识库问答的基础查询可以在网关层做语义缓存用 embedding 计算文本相似度来判断是否命中。五是可观测性。网关是流量的必经之路非常适合输出日志、指标和链路数据。包括每次请求的模型名称、Token 使用量、延迟、状态码、缓存命中情况、租户成本等。六是成本计量。把每次调用折算成单价并按租户、业务线、模型类型、时间段聚合方便成本分摊和异常消费告警。2.2 网关与业务服务的边界一个需要明确的边界问题是网关到底做多少业务逻辑常见的错误是网关越做越重最后把 Prompt 模板、知识库检索、Agent 编排逻辑都放进网关里。从架构维护角度我建议网关只关注“横切治理”不关注具体业务逻辑。业务逻辑如 Prompt 拼装、RAG 检索、工具调用都应该留在业务服务内。网关的服务对象是“模型调用”这个动作而不是“业务结果”。保持这个边界网关才能足够稳定并且可以被多个业务方长期依赖。2.3 网关形态选择LLM 网关在部署形态上通常有三种选择。第一种是独立部署的集中式网关例如搭建一个内部服务所有模型请求都请求它。优点是治理能力强权责清晰缺点是增加一跳网络延迟并且网关自身成为高可用重点。第二种是边车Sidecar形态每个业务服务旁边部署一个网关代理。优点是延迟较低故障爆炸半径小缺点是治理策略分散运维成本上升。第三种是直接使用云厂商的 API 网关或托管型 LLM Gateway。优点是开箱即用缺点是可能绑定云厂商且部分高级语义能力仍需自建。这三种形态没有绝对优劣更多是组织规模和运维能力的取舍。对于大多数中小团队集中式网关是性价比最高的起步形态后续如果延迟敏感型业务增多再针对部分服务做边车下沉。3. 生产化前必须想清楚的架构权衡3.1 同步代理还是异步管道LLM 调用和普通 REST API 最大的不同是响应时间波动大。普通接口 50ms 以内是常态而 LLM 接口可能 1 秒到 60 秒不等。于是网关架构必须考虑是以“同步转发”为核心还是以“事件队列 异步任务”为核心同步代理的好处是模型天然贴近用户请求流式返回容易实现代码逻辑简单排错直观。但同步代理对并发和连接管理要求很高每一个上游请求都会占用一个连接如果业务方大量调用网关需要精细控制连接池、超时和并发数。异步管道则是将请求写入消息队列由 Worker 异步调用模型并回传结果。这种设计适合离线批量生成、异步任务等等不需要用户立刻等待的场景但实现复杂度高而且要处理回调通知和结果持久化。我的建议是第一版网关优先实现同步代理因为大多数在线应用都需要流式返回。同时把“任务类型”设计成请求的一个属性为后续异步通道留出扩展位。同步和异步并存是生产化网关的常态但不要把两个模式混在一条代码路径里。3.2 多供应商抽象到什么程度很多团队接入第一个模型服务后会立即思考“未来要支持多家怎么办”。于是开始设计一个高层抽象Provider 接口、统一模型枚举、统一错误码。这个方向本身没错但要小心过度设计。多供应商抽象的核心是把不变的部分固定下来把变化的部分通过配置驱动。实际经验是不要一开始就试图把手上的所有模型 API 统一成一个格式。更好的做法是对业务提供一套“内部模型名”到“供应商真实模型名”的映射对响应体做统一封装至少要统一错误结构和 Token 用量字段对上游 SDK 或 HTTP Client 做薄封装不要把所有差异化参数都提前抽象掉。因为不同的模型供应商在参数细节上差异很大例如部分供应商的temperature范围不同或者不支持logprobs。过早抽象会导致网关层频繁发布版本稳定性下降。3.3 缓存命中与成本/延迟的权衡缓存能显著降低成本但也有三个很现实的坑。第一个坑是动态 Prompt 导致缓存命中率低。如果请求参数中带时间、用户 ID、随机数即使语义相同也无法命中。第二个坑是缓存污染。如果同一个缓存 Key 先被低质量模型的回答命中后续高质量模型永远没有机会被调用用户体验会受影响。第三个坑是数据安全。某些业务 Prompt 包含敏感信息不应写入 Redis 或对象存储。因此网关的缓存策略不应该默认对所有请求生效而应该按路由规则配置。通常建议对这类请求开启语义缓存读多写少、Prompt 相对固定、结果允许过期、不涉及敏感数据。普通随机性较强的 Prompt 不要缓存。3.4 流式与超时控制LLM 网关必须支持流式否则会让用户首 token 延迟变得不可接受。但流式与网关治理之间有冲突网关需要在拿到完整响应后记录 token 用量但流式情况下响应体是一块块到达的无法同时做到“一边转发一边精确计量”。实际设计中流式请求的用量记录通常有两种做法。一种是在业务侧传入预估参数或从响应中解析usage字段但并非所有供应商都会在每个 chunk 里带上 usage 信息。另一种是通过网关对内容流做累计统计估算 token 数。生产环境可以两者结合消息头里如果有usage就优先用没有则用内容长度估算。请求超时也要分级设计而不要只设置一个全局超时。例如连接超时 3 秒、首 token 超时 10 秒、整体响应超时 60 秒。不同模型和场景的超时策略差别很大建议在路由配置中为每个上游单独设置。4. 最小可落地 LLM 网关实现下面我们来动手实现一个最小可落地的 LLM 网关。示例会尽量简单方便你理解核心流程代码可以直接复制运行但生产环境还需要按团队规范扩充。4.1 技术选型与环境准备示例采用 Python FastAPI httpx原因有三FastAPI 对异步支持好适合转发流式响应httpx 支持 HTTP/2 和流式传输和 FastAPI 配合自然Python 生态便于快速加入 Redis、Prometheus 等组件。环境要求如下版本以你本机实际安装为准Python 3.10 或更高版本FastAPI 0.100uvicorn 0.23httpx 0.24安装依赖pip install fastapi uvicorn httpx pydantic-settings示例项目结构llm-gateway/ ├── app │ ├── __init__.py │ ├── main.py # 网关入口 │ ├── config.py # 配置管理 │ ├── models.py # 请求/响应模型 │ ├── provider.py # 上游供应商抽象 │ ├── cache.py # 简单 TTL 缓存 │ └── middleware.py # 限流与日志 └── .env.example4.2 创建配置管理先在app/config.py中定义网关配置。使用pydantic-settings读取环境变量避免把密钥写死在代码中。# 文件路径app/config.py from functools import lru_cache from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): model_config SettingsConfigDict(env_file.env, env_file_encodingutf-8) # 网关服务端口 gateway_host: str 0.0.0.0 gateway_port: int 8000 # 示例用单一上游生产环境通常有一张路由表 default_upstream: str https://api.openai.com default_model: str gpt-4o-mini # 上游 API Key生产环境建议从密钥管理服务拉取 upstream_api_key: str # 限流配置单租户每秒最大请求数 rate_limit_per_second: int 10 # 缓存配置TTL 秒数0 表示关闭 cache_ttl_seconds: int 0 # 超时配置 connect_timeout_seconds: float 3.0 first_token_timeout_seconds: float 10.0 total_timeout_seconds: float 60.0 lru_cache def get_settings() - Settings: return Settings()在.env.example中# 复制为 .env 后填写 UPSTREAM_API_KEYsk-xxxx DEFAULT_UPSTREAMhttps://api.openai.com DEFAULT_MODELgpt-4o-mini RATE_LIMIT_PER_SECOND10 CACHE_TTL_SECONDS0这里有一个生产实践要强调环境变量不要提交进 Git 仓库。upstream_api_key是敏感信息建议在 CI/CD 流程中通过密钥管理平台注入。4.3 定义请求与响应模型网关需要对外提供统一的请求结构。为了兼容 OpenAI 接口风格我们直接使用类似 ChatCompletion 的格式这样业务方从直连切换到网关的成本最低。# 文件路径app/models.py from typing import Any, Dict, List from pydantic import BaseModel, Field class ChatMessage(BaseModel): role: str content: str model_config {extra: allow} class ChatCompletionRequest(BaseModel): model: str Field(defaultgpt-4o-mini) messages: List[ChatMessage] temperature: float | None None max_tokens: int | None None stream: bool False model_config {extra: allow} class Usage(BaseModel): prompt_tokens: int 0 completion_tokens: int 0 total_tokens: int 0 class ChatCompletionResponse(BaseModel): id: str object: str chat.completion created: int model: str choices: List[Dict[str, Any]] usage: Usage | None None在真实网关中你还需要把供应商返回的各种字段映射成统一结构。这里为了说明核心流程只定义最常用的几个字段。model_config {extra: allow}用于透传额外参数避免网关因为多了一个供应商扩展字段而报错。4.4 封装上游供应商网关的核心逻辑都在 Provider 这一层。这里实现一个OpenAICompatibleProvider它使用httpx.AsyncClient转发/chat/completions请求并支持超时和流式处理。# 文件路径app/provider.py import uuid import time from typing import AsyncIterator import httpx from app.config import Settings from app.models import ChatCompletionRequest class OpenAICompatibleProvider: 面向 OpenAI 兼容接口的上游 Provider。 其他供应商如 Claude可以通过实现同样的 call/stream 方法接入。 def __init__(self, settings: Settings): self.settings settings self.headers { Authorization: fBearer {settings.upstream_api_key}, Content-Type: application/json, } self._client httpx.AsyncClient(timeoutsettings.total_timeout_seconds) async def call(self, req: ChatCompletionRequest) - dict: 非流式调用返回完整响应 JSON。 url f{self.settings.default_upstream}/v1/chat/completions payload req.model_dump(exclude_noneTrue) resp await self._client.post(url, jsonpayload, headersself.headers) resp.raise_for_status() return resp.json() async def stream(self, req: ChatCompletionRequest) - AsyncIterator[bytes]: 流式调用把上游响应以 bytes 形式逐块返回。 httpx 的 aiter_bytes 会在流结束时自动关闭响应体。 url f{self.settings.default_upstream}/v1/chat/completions payload req.model_dump(exclude_noneTrue) async with self._client.stream( POST, url, jsonpayload, headersself.headers ) as resp: resp.raise_for_status() async for chunk in resp.aiter_bytes(): yield chunk上面的实现思路里有两个生产要点httpx.AsyncClient应该被复用否则每个请求都创建新连接会非常浪费stream方法使用async with self._client.stream()确保响应流关闭。4.5 实现缓存与限流缓存可以先用内存字典实现一个最小 TTL 版本。生产环境建议换成 Redis但思路是一样的。# 文件路径app/cache.py import time import threading from typing import Any class TTLCache: def __init__(self, ttl_seconds: int 0, max_size: int 1024): self.ttl_seconds ttl_seconds self.max_size max_size self._data: dict[str, tuple[float, Any]] {} self._lock threading.Lock() def _is_expired(self, key: str) - bool: if key not in self._data: return True expire_at, _ self._data[key] return time.time() expire_at def get(self, key: str): with self._lock: if self._is_expired(key): self._data.pop(key, None) return None return self._data[key][1] def set(self, key: str, value: Any): if self.ttl_seconds 0: return with self._lock: if len(self._data) self.max_size: # 简单淘汰清掉所有过期 key for k in list(self._data.keys()): if self._is_expired(k): self._data.pop(k, None) self._data[key] (time.time() self.ttl_seconds, value) def make_key(self, req) - str: 生产环境建议在这里加入租户 ID避免跨租户串数据。 messages req.messages # 序列化时排除可能造成缓存击穿的随机参数 import json return json.dumps({ model: req.model, messages: [m.model_dump() for m in messages], temperature: req.temperature, }, ensure_asciiFalse, sort_keysTrue)缓存 key 是重点。如果直接把整个请求体包含进来那么任何细微参数变化都会导致不命中。建议只使用影响语义结果的字段model、messages、temperature。限流模块用token bucket算法做一个简单的内存限流器按租户维度限流# 文件路径app/middleware.py import time import threading class TokenBucketLimiter: def __init__(self, rate: int): rate: 每秒允许通过的请求数。 self.rate rate self.tokens rate self.last time.monotonic() self._lock threading.Lock() def acquire(self, tokens: int 1) - bool: with self._lock: now time.monotonic() elapsed now - self.last self.tokens min(self.rate, self.tokens elapsed * self.rate) self.last now if self.tokens tokens: self.tokens - tokens return True return False限流返回的错误应该统一否则客户端无法处理。通常生产网关会返回429 Too Many Requests并提供Retry-After头部。4.6 创建网关主入口现在把以上模块组合起来生成 FastAPI 应用。# 文件路径app/main.py import time import uuid from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, StreamingResponse from app.cache import TTLCache from app.config import get_settings from app.middleware import TokenBucketLimiter from app.models import ChatCompletionRequest, ChatCompletionResponse from app.provider import OpenAICompatibleProvider settings get_settings() app FastAPI(titleLLM Gateway, version0.1.0) provider OpenAICompatibleProvider(settings) cache TTLCache(ttl_secondssettings.cache_ttl_seconds) limiter TokenBucketLimiter(ratesettings.rate_limit_per_second) app.middleware(http) async def add_request_id(request: Request, call_next): request_id request.headers.get(X-Request-ID, str(uuid.uuid4())) start time.time() response await call_next(request) response.headers[X-Request-ID] request_id response.headers[X-LLM-Gateway-Version] 0.1.0 # 记录基础访问日志 print({ request_id: request_id, path: request.url.path, status: response.status_code, duration_ms: round((time.time() - start) * 1000, 2), }) return response app.get(/health) async def health(): return {status: ok} app.post(/v1/chat/completions) async def chat_completions(req: ChatCompletionRequest, request: Request): # 租户信息可以从 Header 传入例如 X-Tenant-Id tenant request.headers.get(X-Tenant-Id, default) # 限流 if not limiter.acquire(): return JSONResponse( status_code429, content{ error: { message: rate limit exceeded, type: rate_limit, request_id: request.state.__dict__.get(request_id), } }, headers{Retry-After: 1}, ) # 缓存逻辑仅针对非流式请求 if not req.stream: cache_key cache.make_key(req) cached cache.get(cache_key) if cached: return JSONResponse(contentcached) # 转发到上游 if req.stream: async def event_stream(): async for chunk in provider.stream(req): yield chunk return StreamingResponse(event_stream(), media_typetext/event-stream) try: upstream_data await provider.call(req) except Exception as exc: return JSONResponse( status_code502, content{error: {message: fupstream error: {exc}, type: upstream_error}}, ) # 写入缓存 if not req.stream and settings.cache_ttl_seconds 0: cache.set(cache_key, upstream_data) return JSONResponse(contentupstream_data)需要注意request.state.__dict__在 FastAPI 中并不是推荐做法上面的错误响应里没有真正写入 request_id。更好的方式是把 request_id 放到中间件中并显式赋值给request.state.request_id。下面是修正片段app.middleware(http) async def add_request_id(request: Request, call_next): request_id request.headers.get(X-Request-ID, str(uuid.uuid4())) request.state.request_id request_id ...然后在限流分支中使用request.state.request_id即可。4.7 运行与验证在项目根目录创建.envUPSTREAM_API_KEYsk-your-key启动网关uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload使用 curl 测试非流式请求curl --location http://localhost:8000/v1/chat/completions \ --header Content-Type: application/json \ --header X-Tenant-Id: demo \ --data { model: gpt-4o-mini, messages: [ {role: user, content: 你好请用一句话介绍你自己} ], stream: false }预期会返回上游模型的 JSON 响应。如果上游配置正确响应里应包含choices和usage字段。再测试流式请求curl --location http://localhost:8000/v1/chat/completions \ --header Content-Type: application/json \ --header X-Tenant-Id: demo \ --data { model: gpt-4o-mini, messages: [ {role: user, content: 讲一个简短的笑话} ], stream: true }如果stream为 true你会看到一段 SSE 格式的流式响应通常每个 chunk 都是一个data: {...}数据块最后以data: [DONE]结束。5. 常见问题与排查思路网关上线后运维排错比功能开发更频繁。下面整理几个真实场景中高频出现的问题。5.1 流式请求超时现象客户端通过网关进行流式对话超过 10 秒后才收到第一个 token然后连接被断开。原因大多数供应商在真实生成前也会有排队耗时。如果首 token 超时设置过短会在模型尚未开始返回内容时误杀连接。解决将“连接超时”和“首 token 超时”分开设置。连接超时可以短一些比如 3 秒首 token 超时建议放宽到 15 到 30 秒因为 LLM 的 Prefill 阶段耗时可能长。整体响应超时应根据最大生成长度调整。5.2 重试导致上游重复扣费现象上游返回超时网关自动重试一次但实际第一次请求已经在模型端执行成功结果用户收到两个相同回答成本也翻倍。原因没有做幂等控制。LLM 请求天然不是幂等的除非业务方传入自定义请求 ID并在供应商侧支持去重。解决在网关层默认关闭“不确定类错误”的自动重试。只有当错误码是明确的429或5xx并且确认请求未被执行时才允许重试。对非流式请求可以在重试前用同一个请求 ID 查询上游状态如果供应商不支持就宁可失败也不要盲目重试。5.3 不同模型返回结构不一致现象网关统一封装后业务方反馈解析错误。某个国产模型把usage字段放在别的位置或者choices里没有finish_reason。原因不是所有模型都严格兼容 OpenAI 响应格式尤其是部分自建模型服务。解决每个 Provider 实现内部都要完成“响应标准化”。网关对外暴露的统一模型必须稳定不能把上游的差异直接透传。建议写一个normalize_response方法在call和stream返回前统一处理字段。5.4 缓存导致不同租户数据串线现象租户 A 和租户 B 请求了同样的问题租户 B 直接拿到了租户 A 的回答但回答里包含租户 A 的内部信息。原因缓存 key 只包含 model 和 messages没有隔离租户维度。解决key 中必须加入租户 ID、应用 ID 等隔离字段。更安全的做法是只对明确标记为“可缓存”的租户启用语义缓存默认全局关闭。排查清单可以参考下表问题现象常见原因解决思路网关返回 502上游服务不可达或超时检查连接超时设置查看上游健康状态网关返回 429单租户限流触发调整限流阈值增加队列或降级返回流式响应中断客户端或网关到上游连接被掐断加大首 token 超时开启 TCP KeepAlive响应结果与直连不一致多 Provider 响应未标准化在 Provider 层做字段映射token 用量统计缺失流式响应中未解析 usage 字段对流式 chunks 累计估算或仅记录 non-stream 精确用量6. 生产化教训与最佳实践6.1 不要一开始就追求“万能抽象”第一版 LLM 网关最常见的错误是试图设计一个“能接入全世界所有模型”的抽象层。结果是接口设计被各种模型参数撑得难以维护团队花在抽象上的时间远超接入实际业务的时间。更务实的做法是先接入 1 到 2 个真正使用的模型服务定义好从业务侧到网关的最小公共契约等出现第三个供应商时再根据前两个差异点演进抽象层。这样抽象出来的接口有真实使用场景支撑而不是拍脑袋设计。6.2 网关要与密钥管理、审计联动网关集中保存各类模型密钥安全级别应当等同于堡垒机或数据库账号。生产环境不要依赖.env文件保存长期密钥建议对接 Vault、KMS 或云厂商的 Secret Manager。同时所有经过网关的请求都应该记录审计日志。审计日志至少包含请求时间、租户、调用方 IP、模型名、Token 用量、延迟、供应商请求 ID、错误信息。这不仅是安全需要也是排查问题的关键依据。6.3 从第一天就做成本计量成本计量如果滞后后面补起来会非常痛苦。网关在转发请求时就应该把usage数据写入一个时序数据库或消息队列。建议按以下标签建模tenant_id租户app_id业务应用model_name真实模型名provider供应商request_modestream 或 non-streamcache_hit是否命中缓存这样月底可以快速生成成本报表也能在某个应用出现异常高调用量时及时告警。不要在没有成本数据的网关版本上继续加功能否则后面只能做“估算”无法核实。6.4 灰度与降级预案必须提前设计生产环境的单个模型供应商必然会出现故障或者性能退化。网关发布策略上要支持按租户、流量比例、模型标签做灰度。最简单的实现是路由配置中心化# 伪代码示意生产环境可放到配置中心 routes { default: { provider: openai, model: gpt-4o-mini, }, fallback: { provider: azure, model: gpt-4o-mini-east, }, }当主路由连续失败超过阈值时网关自动把流量切换到备用路由。降级策略应该提前和业务方约定是否允许降低模型规格是否允许返回一个固定兜底回答这些决策涉及业务风险不能全部压给网关开发同学。6.5 监控与排错围绕“链路”展开因为网关增加了一次网络跳转排错难度会变大。建议在中间件中为每个请求注入统一 request_id并在日志里透传到上游。例如通过X-Request-ID和X-LLM-Gateway-Version头传递。监控指标至少要有网关 QPS、成功率、错误码分布各上游的平均延迟、P95/P99 延迟缓存命中率限流触发次数流式请求的平均首 token 延迟。建议接入 Prometheus 指标体系。FastAPI 可以通过prometheus-fastapi-instrumentator快速暴露默认指标再结合自定义 Counter/Histogram 统计模型相关指标。7. 总结与学习路线本文围绕 LLM 网关生产化展开核心可以概括成三句话第一LLM 网关是模型调用进入生产环境的“基础设施”不是简单代理第二架构设计的重点应从功能转向权衡包括同步与异步、抽象与稳定、缓存与成本、流式与计量第三生产化过程中最容易踩坑的不是接口开发而是超时、重试、幂等、成本和安全。建议你从本文的最小网关开始先跑通一条 OpenAI 兼容接口的链路然后补充多租户限流、请求级监控和成本日志。下一步可以继续学习语义缓存的 embedding 相似度计算方法基于 Redis 的分布式限流替代进程内令牌桶网关与 RAG 架构的边界划分支持 Anthropic 或自建模型时如何抽象出不脆弱的标准响应结构结合 Kubernetes 部署时网关的高可用与优雅缩容方案。生产环境没有银弹LLM 网关的每一项能力都需要结合团队现有基础和真实流量反复调整。但有一条经验几乎适用于所有团队把模型调用当成“基础设施流量”来治理越早做越省心。如果你正在搭建或计划搭建 LLM 网关建议先从一条路由、一套日志、一个成本指标开始跑通后再逐步增加能力。