简介：面向计算机与软件工程专业毕业设计场景，这份完整交付包以一个基于Go与React技术栈构建的LLM API统一管理系统为核心。系统通过统一API适配机制，将OpenAI、DeepSeek、文心一言、讯飞星火、通义千问及字节豆包等主流大语言模型接口整合至同一管理平台，提供API密钥管理、流量控制、计费统计、多租户支持、实时监控等核心能力；后端Go、前端React、Node.js中间件与Docker容器化部署构成完整技术链路，并采用分层架构、适配器模式、工厂模式与单例模式保障模块化与可扩展性。

压缩包共551个文件、约1.45MB，其中233个Go文件承载后端服务逻辑，242个JavaScript文件对应前端交互界面，另有SQL数据库脚本、Dockerfile部署配置、docx设计文档与论文材料、Markdown使用说明，目录组织清晰，便于按模块查找学习。

已有112人浏览学习。资源内含完整源码、系统设计文档、部署指南与毕业设计论文模板，代码注释完整且支持一键部署，可开箱即用。既能支撑学生高效完成毕业设计，也可作为学习统一API网关、多租户架构及主流LLM接入实践的参考案例。

1. 为什么要把 LLM API 统一管理当成一件正经事来做

我见过太多团队把多家大模型 API 的 key 直接散落在各个业务代码里：A 服务用 OpenAI 格式，B 服务用 DeepSeek 的 endpoint，C 服务偷偷接了一个聚合平台，月底账单下来财务对着三份来源完全不同的 Excel 发呆。LLM API 统一管理系统要解决的，就是这类越滚越大的混乱——它不是一个“调用中转站”那么简单，而是把路由、鉴权、计费、限流、失败重试这些原本每个业务都要重复写一遍的活，全部收拢到一个入口层，让上游的大模型厂商对业务侧完全透明。如果你正在搭企业内部 AI 平台、做多供应商容灾，或者准备拿一个“源码+论文”性质的完整工程去交差，这套系统的价值不是省几行代码，而是让你第一次能看清楚：每个业务到底烧了多少钱、每次失败到底翻车在哪一环。

2. LLM API 统一管理系统的架构：网关层到底在管哪些事

2.1 统一管理层拆开看：路由、鉴权、计费、限流四件事

先把“统一管理”这个说法拆成四个能落地的职能。路由解决的是“这个请求该发给哪家模型”——同样的“帮我写一段 Python 代码”的 prompt，你可以配置成默认走 DeepSeek、代码类任务走 Claude、长上下文任务走 GLM，路由规则写死还是动态调整都行。鉴权解决的是“谁有资格用哪个模型”——内部系统里不可能每个人都有 GPT-4 的调用权限，你要在网关层做一层应用级别的 API key 校验，再在后端映射到真正的大模型 key。计费是这里最容易被人低估的一块，因为大模型计费不是简单按次，而是按输入 token 加输出 token 分开计价，不同模型的价格表还经常调整，不在网关层统一记账，后面对账就是一场灾难。限流则是保护你的预算和上游配额：单个应用调用量突增时，是直接 429 拒绝，还是排队等待，得有个明确的策略。

这套系统在架构上处于业务后端和大模型厂商 API 之间，所有请求都先打到网关，由网关完成身份校验、配额检查、路由选择，再以网关自己的身份去调用上游。响应回来时，网关还要做用量采集、日志记录、缓存命中判断，最后把结果还给业务方。对业务方来说，它只知道有一个“内部 LLM API 平台”，至于这个请求是发给 OpenAI 还是智谱，是直连还是走了聚合商，业务方完全不关心。这个“业务无感”才是统一管理的核心价值。

2.2 为什么不能省掉这一层：直连模式的三笔糊涂账

有人会问：我不就调个 API 吗，为什么非要套一层网关？我自己写过直连代码，也知道直连的痛快——不用维护网关服务、不用处理额外的延迟、出问题直接看上游的错误信息。但直连模式有两个结构性问题绕不开。第一个是故障隔离缺失：上游模型某天抽风开始疯狂超时，你的业务代码里如果没有统一的超时和重试逻辑，每个服务各自为战，有的重试三次，有的重试十次，一个上游抖动就能把整个系统的负载打上去。第二个是成本失控：每个服务各自记账，没人汇总，月底看到账单根本说不清是哪条业务线烧掉了 80% 的调用量。

我在实际项目中还遇到过更隐蔽的问题：各个服务自己保存的大模型 API key 分散在配置文件、环境变量里，某天 key 轮换，运维要挨个服务去改，漏掉一个就导致某个线上功能悄悄降级。统一网关之后，key 只存在网关这一处，业务侧拿到的都是网关签发的临时凭证，轮换成本从“改十几个服务”变成“改一个服务”。这个收益在团队超过十个人之后会变得非常明显。对于“源码+论文”形式的工程来说，在论文里把这个“为什么需要这一层”讲清楚，本身就是架构设计能力最好的体现。

2.3 技术选型：为什么这套系统最适合用异步框架加 Redis

聊到具体实现，我一般推荐 FastAPI 加异步 HTTP 客户端（httpx），再配一个 Redis 做限流和计数的存储。选 FastAPI 有几个现实理由：第一，它对流式响应（SSE）的支持非常原生，大模型 API 基本都是流式返回，网关层如果不能高效转发流，就会拖垮首字时延；第二，它是异步框架，网关大部分时间都在等上游响应，同步框架一个 worker 阻塞住就只能等，异步可以轻松扛住上百路并发等待；第三，FastAPI 自带 OpenAPI 文档，调试和给前端同学联调都很方便。

Redis 在这里不是缓存那么简单。令牌桶限流需要原子操作，INCR 和 EXPIRE 配合使用可以精确控制每一秒的请求量；计费计数需要高频写入，Redis 的 INCRBY 比写数据库靠谱得多，每隔一段时间再异步落库。我见过有人用内存字典做限流计数，单机自测没问题，一旦部署两个副本，流量分配不均，限流立刻失真。Redis 在这个场景下是性价比最高的选择，不是因为它流行，而是因为它把“原子操作”和“共享状态”这两件事同时解决了。

3. 核心实现：接入多家大模型 API 的路由与适配层

3.1 Provider 抽象：把 OpenAI、DeepSeek、智谱的差异抹平

做统一管理系统的第一步，不是写路由，而是定义一套“中间格式”。举例来说，OpenAI 的 Chat Completions 接口长这样：

{
  "model": "gpt-4o-mini",
  "messages": [{"role": "user", "content": "你好"}],
  "max_tokens": 1024,
  "temperature": 0.7
}

DeepSeek 官方 API 兼容 OpenAI 格式，但要把 model 换成 deepseek-chat ；智谱的接口虽然也宣称兼容 OpenAI 格式，但实际调用时鉴权头用的是不同的 key 名；Anthropic 则是完全不同的结构， messages 里要求带 role 的同时还要区分 system 和 user 的独立字段。如果每个上游都写一套适配代码，网关的核心逻辑就会被 if provider == "anthropic" 这样的分支淹没。

我的做法是在网关内部定义一套统一的“网关请求模型”，所有业务方只认这一套格式，由适配器负责翻译成各家上游的真实请求。核心模型长这样：

from pydantic import BaseModel
from typing import List, Optional, Dict, Any

class ChatMessage(BaseModel):
    role: str          # system / user / assistant
    content: str

class UnifiedChatRequest(BaseModel):
    messages: List[ChatMessage]
    model: Optional[str] = None          # 业务方可指定，不指定则走默认路由
    temperature: Optional[float] = 0.7
    max_tokens: Optional[int] = None
    stream: bool = True

# 内部路由标，业务方通过 headers 里的 X-LLM-App 标识自己
class RoutingContext(BaseModel):
    app_id: str
    user_id: Optional[str] = None
    priority: int = 0

每个上游一个适配函数，输入 UnifiedChatRequest ，输出该厂商的请求体。适配层只做“翻译”，不做业务判断，这样新增一家模型厂商时，你只需要写一个新适配器，路由和计费逻辑完全不用动。这个设计在写论文时也很好展开，可以画一张适配器模式的类图，评审老师看到“面向扩展开放”的设计基本就认可了。

3.2 路由规则：自己直连各家 API 还是走 OpenRouter 这类聚合商

这里有个选型分岔要提前想清楚：做统一管理，上游到底是直接接各家官方 API，还是接 OpenRouter 这类聚合平台。两种我都试过，各自有明确适用场景。直接接官方 API 的好处是延迟低、出错时你能直接看到是哪一家的错误码、数据链路短，而且 key 的管理完全在自己手里；坏处是要维护的适配器多，每家厂商的限流策略、计费规则都要逐一适配，上线初期工作量大不少。

接 OpenRouter 这类聚合商则相当于“统一格式 + 统一计费 + 多模型切换”一步到位，它自己就是一个大号的统一管理层，你再做一层自己的网关等于在它上面再套一层，架构上略显冗余，但适合想要快速验证多方模型效果、又不想每家都注册一遍的场景。这里有个实际的坑：聚合商的 key 管理是黑匣子，你并不知道它把请求路由给了哪家上游，出问题排查时只能看到聚合商的一句错误信息，根本定位不到具体厂商。我的建议很明确：生产系统直接接官方 API，网关层自己掌握路由；个人项目或前期探索可以用聚合商，但别把生产依赖建在别人没有 SLA 的转发层上。

路由规则的实现通常是基于权重或优先级。我常用的配置是设置一个默认路由表，例如 GPT-4o 为主力、DeepSeek 作为降级备胎、GLM 作为特定任务路由。请求进来先查规则表，规则表没命中就走默认模型。规则表存在哪？本地配置文件就够了，但更灵活的做法是存 Redis 哈希表，支持运行时调整而不用重启网关。

3.3 最小可跑通的网关核心：鉴权、转发、流式返回

下面这段代码是一个简化但完整的网关主干，实现了三个职能：校验业务方 API key、按路由表选择上游、以流式方式转发请求。这是整套系统的地基，我建议先跑通它再谈限流和计费。

import json
import httpx
from fastapi import FastAPI, Header, HTTPException
from fastapi.responses import StreamingResponse

app = FastAPI()

# 业务方 key -> 应用身份；生产环境应从 Redis/DB 加载
VALID_APP_KEYS = {"sk-app-001": "crm-system", "sk-app-002": "data-lab"}

# 上游模型路由表：业务标识 -> (厂商, 模型名)
ROUTE_TABLE = {
    "crm-system": ("openai", "gpt-4o-mini"),
    "data-lab": ("deepseek", "deepseek-chat"),
}

# 各家上游的 endpoint 和鉴权 key；生产环境不要写在代码里
UPSTREAM_CONFIG = {
    "openai": {"base": "https://api.openai.com/v1", "api_key": "sk-xxx"},
    "deepseek": {"base": "https://api.deepseek.com/v1", "api_key": "sk-xxx"},
}

async def call_upstream(provider: str, model: str, payload: dict):
    cfg = UPSTREAM_CONFIG[provider]
    # 适配层伪代码：OpenAI 兼容接口都按同一格式发，非兼容的在这里做翻译
    target_url = f"{cfg['base']}/chat/completions"
    headers = {"Authorization": f"Bearer {cfg['api_key']}"}
    payload = {**payload, "model": model}
    client = httpx.AsyncClient(timeout=httpx.Timeout(connect=5, read=60))
    req = client.build_request("POST", target_url, json=payload, headers=headers)
    return await client.send(req, stream=True)

@app.post("/v1/chat/completions")
async def chat_completions(request: dict, x_llm_key: str = Header(...)):
    # 1. 鉴权：业务方身份
    if x_llm_key not in VALID_APP_KEYS:
        raise HTTPException(status_code=401, detail="invalid app key")
    app_id = VALID_APP_KEYS[x_llm_key]

    # 2. 路由：从请求里取 model 字段，没指定则走应用默认
    requested_model = request.get("model")
    if requested_model:
        provider, model = requested_model, requested_model
    else:
        provider, model = ROUTE_TABLE[app_id]

    # 3. 转发上游，流式返回
    upstream_resp = await call_upstream(provider, model, request)

    async def event_stream():
        async for line in upstream_resp.aiter_lines():
            if line.strip() == "data: [DONE]":
                yield "data: [DONE]\n\n"
                break
            if line.startswith("data:"):
                # 这里可以做用量采集：解析 usage 字段，写 Redis 计数
                yield line + "\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

这段代码的关键逻辑在于： x_llm_key 这个 Header 是业务方身份的凭证，不是上游的 key——业务方永远接触不到真实的大模型 key； ROUTE_TABLE 做应用级路由，crm-system 默认走 GPT-4o，data-lab 默认走 DeepSeek，业务方也可以用 model 字段临时指定其他模型，但权限控制可以后续细分成“允许模型白名单”； event_stream 是流式转发的心脏，它逐行读取上游返回的 SSE 数据，原样转发给业务方，在这中间是插入计费、日志、缓存钩子的最佳位置。

注意 httpx.Timeout 参数的设置：连接超时 5 秒，读超时 60 秒。大模型流式响应有时会出现一两分钟没有新 token 的情况，读超时设太短会导致中间断开，设太长又会让故障响应变慢。建议读超时在 60 到 120 秒之间调整，具体看你上游厂商的响应特征。

3.4 限流与调用量计费的 Redis 实现：令牌桶与 INCRBY

网关能转发是第一步，能控制住流量才算生产可用。限流我用的是令牌桶算法的 Redis 实现，相比固定窗口，令牌桶允许一定突发流量，更适合大模型调用这种“单次请求耗时差异极大”的场景——有的请求几百毫秒就返回了，有的要跑十几秒，只看 QPS 意义不大，按桶限能让峰值更平滑。

import time
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

class TokenBucket:
    def __init__(self, key: str, capacity: int, refill_rate: float):
        self.key = key
        self.capacity = capacity
        self.refill_rate = refill_rate  # 每秒补充 token 数

    def acquire(self) -> bool:
        # 用 Lua 脚本保证原子性，避免并发下多扣
        script = """
        local tokens = tonumber(redis.call('GET', KEYS[1]))
        local ts = tonumber(redis.call('GET', KEYS[1]..':ts'))
        if tokens == nil then
            tokens = ARGV[1]
            ts = ARGV[2]
        end
        local elapsed = tonumber(ARGV[2]) - ts
        tokens = math.min(tonumber(ARGV[1]), tokens + elapsed * tonumber(ARGV[3]))
        if tokens < 1 then
            return 0
        end
        redis.call('SET', KEYS[1], tokens - 1)
        redis.call('SET', KEYS[1]..':ts', ARGV[2])
        return 1
        """
        ok = r.eval(script, 1, self.key, self.capacity, time.time(), self.refill_rate)
        return bool(ok)

# 每个应用一个桶：容量 10，每秒补充 2 个 token
bucket = TokenBucket("bucket:crm-system", 10, 2)
if bucket.acquire():
    # 放行
    pass
else:
    # 返回 429，提示稍后重试
    pass

这个方案的参数设置是有讲究的。容量不能设太大，因为大模型请求都是昂贵请求，一个应用突发 50 个并发连续跑大模型，账单会很难看；refill_rate 也不能设太保守，否则正常业务高峰会被误伤。我一般按“该应用预估峰值 QPS 的 2 倍”来设 capacity，按“平均 QPS 的 1.2 倍”来设 refill_rate，先这样跑一周，再根据 Redis 里的实际拒绝数和业务投诉反向调参。

计费用的是另一组 Redis key，在流式转发的 event_stream 函数里解析上游返回的 usage 字段，然后把 input_tokens 和 output_tokens 分别累加：

async def event_stream():
    async for line in upstream_resp.aiter_lines():
        if line.startswith("data:") and "[DONE]" not in line:
            try:
                data = json.loads(line[5:])
                usage = data.get("usage")
                if usage:
                    r.hincrby(f"usage:{app_id}", "input_tokens", usage.get("prompt_tokens", 0))
                    r.hincrby(f"usage:{app_id}", "output_tokens", usage.get("completion_tokens", 0))
            except Exception:
                pass
        yield line + "\n\n"

注意流式响应的 usage 通常在最后一个 data 块里，所以这个采集逻辑只能在流结束前抓到，不会中途重复计数。这里有个小的性能考量：每一个流式 chunk 都做一次 json.loads 虽然开销不大，但在高并发下会被放大。建议先判断行内容是否包含 "usage" 字段再解析，能省掉大量无效解析。

4. 计费、用量统计与多 Key 池化：把账算清才能上线

4.1 用量统计的日志链路：从网关到账单的完整闭环

限流和计数跑通之后，下一个要面对的问题是：Redis 里的计数只是临时的，生产环境不能拿 Redis 当永久账本。我的习惯是两条链路并行。第一，网关每个请求都打印一条结构化日志，包含 app_id 、 model 、 prompt_tokens 、 completion_tokens 、 latency_ms 、 status 字段，打到 stdout 或专门的日志文件，由日志采集系统收到 Elasticsearch 或 ClickHouse。第二，定期（每小时或每天）把 Redis 里的增量计数批量写入 MySQL 或 PostgreSQL 的 usage_daily 表，作为最终账单依据。

这两条链路各有用处：日志链路用于实时排查单条请求的明细——哪个用户、什么时候、发了什么请求（注意脱敏）、花了多少钱；数据库链路用于月度对账和预算告警——每天增量导入，跑 SQL 一查就知道哪个业务线超支。我踩过只靠日志不做落库的坑，日志系统的保留期一过，历史账单就查不到了，财务要数据的时候只能干瞪眼。如果做的是论文工程，这个“明细日志 + 日汇总表”的双写设计，在论文里也可以作为“可观测性设计”章节的一个亮点。

4.2 多 Key 池化与故障摘除：让同一个模型有多个 key 可用

很多团队会用多个 key 轮询来分摊调用量。原因很现实：有些服务商的免费额度是分 key 的，或者单个 key 有每分钟调用上限，池化 key 可以避开单 key 限流。但池化不是一个字典遍历就完事，要考虑三个实际问题。第一，key 的权重可能不同——有些 key 是付费的，有些是免费额度，不能一律轮询，否则免费额度会先被打爆。第二，某个 key 被上游封了（403）时，网关要立刻把它摘除，而不是让它继续在每个请求里浪费一次失败的往返。第三，key 池要能动态调整，新 key 上线、旧 key 到期，都需要不停机生效。

我常用的实现是一个带健康状态的 key 队列：

import itertools
from collections import deque

class KeyPool:
    def __init__(self, keys: list[dict]):
        # keys: [{"key": "sk-a", "weight": 2}, {"key": "sk-b", "weight": 1}]
        self.keys = deque(keys)
        self._broken_keys = set()

    def next_key(self):
        # 简单轮询，带权重扩展
        for _ in range(len(self.keys)):
            item = self.keys.popleft()
            self.keys.append(item)
            if item["key"] not in self._broken_keys:
                return item["key"]
        raise RuntimeError("all keys are broken")

    def mark_broken(self, api_key: str):
        self._broken_keys.add(api_key)
        # 等 10 分钟后自动恢复，用线程或异步任务做

    def recover(self, api_key: str):
        self._broken_keys.discard(api_key)

调用上游时如果收到 401 或 403，就调 mark_broken 把这个 key 摘除；收到 429 说明限流但不代表 key 失效，不要摘除；收到 5xx 需要结合连续失败次数来判断，单次 5xx 可能是上游临时抖动，我一般是连续三次 5xx 才摘除。自动恢复的策略也重要，key 被封往往是临时的，比如触发了风控，过段时间能恢复。恢复机制别做太复杂的探活，简单做一个延时队列，10 分钟后再放回池子试一次即可。

4.3 语义缓存：同一个 prompt 能不能省下重复调用的钱

大模型 API 调用贵，有些场景的重复性其实很高——比如内部知识库问答里，同一个问题被不同用户反复问；或者报表生成场景，业务方每天跑同一套 prompt 模板。缓存是省钱的最直接手段。简单做法是 KV 缓存：以 messages JSON 的哈希值 为 key，缓存非流式响应和 token 用量，命中后直接返回缓存结果并扣除对应配额。这个方案实现成本极低，但对语义相同的不同问法完全无能为力。

进阶做法是语义缓存：将 prompt 做 embedding 向量化，存到向量数据库里，新请求进来先用余弦相似度找最相似的缓存条目，相似度超过阈值（比如 0.92）就返回缓存结果。这个方案效果好，但引入了 embedding 延迟和额外的向量检索组件，架构复杂度提升不少。我给一个务实的建议：先做 KV 缓存，把 (model, messages_hash) 作为 key，命中率通常能到 10% 左右，加上简单的相似度归一化（把空白字符、标点差异消掉），低成本就能见效；语义缓存等确认重复流量确实存在之后再引入，别一上来就上向量库。

缓存命中时有个容易忽略的细节：计费。业务方失去了 token 计数，但从业务角度这次回答省下了成本，账单上最好能体现“缓存命中节省了多少钱”。我在缓存命中时会把 usage 字段记成 "input_tokens": 0, "output_tokens": 0, "cached": true ，业务方看到这个字段就能区分真实调用和缓存调用。

4.4 论文视角：这个系统怎么写才能让评审看出工程量

如果这套系统是课程设计或毕业设计的“源码+论文”工程，论文结构不能只写“我做了什么功能”，要写“我遇到了什么问题、做了哪些方案对比、数据证明我选的方案更好”。我建议论文里至少包含三个可论证的点。第一，多厂商接入的适配器模式设计，和硬编码分支相比，新增一个厂商的代码改动量对比——这个可以用表格列出来，适配器模式下新增厂商只需要新增一个类和一行路由配置。第二，流式转发网关 vs 传统同步转发的性能对比，用压测数据说明异步流式方案下首字时延增加不超过 10% 到 20%，同时并发能力提升数倍。第三，限流算法的选型论证，比较固定窗口、滑动窗口和令牌桶在“大模型调用场景”下的表现，用 Redis 记录的拒绝率和误杀率来说明为什么选令牌桶。

论文里如果放了真实的压测数据和对比表，评审基本就不会觉得这是“玩具工程”。源码部分则要注意目录结构清晰—— adapter/ 、 router/ 、 ratelimiter/ 、 billing/ 、 storage/ 分层明确，每个模块带 README 说明设计意图，比塞一堆没有注释的工具函数更能体现工程能力。

5. 实战踩坑记录：流式超时、上下文长度、账单对不齐

5.1 流式响应在网关层中断，业务端毫无感知

现象：业务方反馈说“模型回答经常只有一半”，但网关日志里显示响应已经正常完成，状态码 200，流式数据全部转发成功。

原因：问题出在 httpx 客户端的读超时。大模型流式返回时，思考时间长的模型可能连续 20 秒都不吐出第一个 token，我的 read=60 超时设置从连接建立后开始计时，如果 60 秒内没有任何一个字节到达，客户端就会抛 ReadTimeout 中断连接。看起来是“网关完成”，实际上流在中间断了，业务端收到的 data: [DONE] 永远没有出现。

解决：超时设置不能只看平均值，要看 P95 首字时延。我后来改成 httpx.Timeout(connect=5, read=180, write=30) ，同时在上游响应的 headers 里透传一个自定义超时字段，让业务方知道这个请求的超时阈值。更稳妥的做法是区分“连接超时”和“空闲超时”：上游如果长时间没有任何数据，合理的处理是中断并重试；但首字前的等待期要格外宽容。这里调参没有银弹，必须结合上游厂商的实际响应特征来设，建议先跑一周日志，统计首字时延分布再定。

5.2 把上游的 400 Context Length 错误当成网关 bug

现象：网关日志里频繁出现 api error: 400 this model's maximum context length is 1048576 tokens ，业务方质疑“是不是你们的网关把请求撑爆了”。

原因：这是最典型的“黑匣子误判”。业务方通过网关调用了上下文窗口很大的模型，但他们把超长文档一次性塞进 messages，token 总和超过了模型上限。这个错误是上游模型拒绝请求，不是网关逻辑问题。但问题在于网关没有对这类 4xx 错误做归一化，业务方看到的是层层包装后的错误信息，自然会把锅甩给网关。

解决：网关在捕获上游 4xx 错误时，要区分“参数错误”“鉴权失败”“上下文超长”“内容违规”几类，把上游的错误信息解析后包装成统一错误码返回。特别是上下文超长，网关可以做一层预检查——把 messages 的 token 数估算出来，超过模型上限就提前返回 400，错误信息里带上“当前请求约 XX tokens，模型上限 XX tokens，请缩减内容或更换模型”。一来省掉一次无效的上游调用，二来业务方拿到明确的指引。token 估算不需要精确，用 len(content) // 4 加一个系数即可，准确率够用于预检。

5.3 计费按上游 usage 字段，账单经常对不齐

现象：网关记录的月度调用量和厂商后台的账单差了 5% 到 10%，财务对不上账，质疑系统的计费准确性。

原因：流式响应的 usage 字段只在最后一个数据块中出现，但如果我的转发层在中间某个环节丢弃了最后一块（比如前面提到的超时中断），usage 就永远采集不到。更隐蔽的是，有些模型的 usage 字段并不包含在流式响应的最后一块，而是有一个独立的响应对象，需要额外解析一次。还有部分厂商按字符计费，有些按 token 计费，网关不做归一化就会累计出系统性偏差。

解决：不要在流式转发的过程中依赖 usage 字段作为唯一计费依据。我在网关层加了一个“兜底策略”：流正常结束时解析 usage 并累加；如果流中途断裂，则根据 output_tokens 的近似值从业务侧传入的 max_tokens 扣减估算。同时在计费表里增加 billing_source 字段，标记本行数据来自“usage 精确统计”还是“估算”，月底对账时先把精确统计的账对上，估算部分单独列出来。这个做法不能完全消除误差，但至少让你能对财务解释清楚差额的来源。

5.4 多 Key 池化后触发上游风控，整个网关被限

现象：配置了 5 个 key 做轮询，第二天突然所有 key 全部返回 403，网关彻底瘫痪。

原因：上游的风控系统检测到这些 key 来自同一出口 IP，且请求频率总和超过单个组织的正常范围，把它们判定为异常使用，全部冻结。池化使用 key 确实能分散单 key 的速率限制，但如果出口 IP 恒定，风控模型很容易从“同 IP + 多 key + 高并发”这个组合特征里识别出池化行为。

解决：池化 key 的并发要控制在上游可接受的范围，不是 key 越多就能无限放大吞吐。我在池化之外加了一层全局 QPS 封顶——不管 pool 里有多少 key，网关发给每个上游的总请求速率有一个硬上限，同时随机加入 100 到 300 毫秒的抖动，避免请求呈现过于整齐的周期性。另外强烈建议池化只用于非核心、可重试的调用；生产核心链路用单个主 key 加一个备用 key 就够了，主 key 挂了再切备用。不要把池化当成规避风控的手段，它只是用来水平扩展吞吐的工程手段。

5.5 日志里堆满 prompt 原文，把用户隐私打进 ELK

现象：合规团队检查日志系统，发现网关日志记录了所有请求的完整 prompt 内容，里面包含工单描述、客户姓名、甚至内部讨论的敏感信息。

原因：为了排查问题方便，我在 gateway 日志里打印了完整的消息内容，图一时省事。而且日志采集系统对这些字段不做任何脱敏，直接进 Elasticsearch 后任何有日志查询权限的人都能看到。这在企业内部问题不大，一旦涉及用户数据或客户数据，就是合规事故。

解决：日志分级，正文不落明文日志。网关的访问日志只记录 app_id 、 model 、token 数、耗时、状态码；prompt 内容只在开启 DEBUG 模式且指定了 debug_request_id 时才打印，并且经过脱敏——手机号、邮箱用正则替换成掩码。这个教训很大，后来我写网关代码时养成了习惯：任何日志字段先问自己一句“这里能不能写明文”，不能的话就换 ID 引用。论文工程里如果写了日志模块，建议把这个脱敏设计写进去，也是加分项。

6. 压测与验证：上线前用一套脚本确认网关没拖后腿

6.1 并发压测脚本：先测网关本身的极限，再测转发后的劣化

网关层增加一次网络转发，必然引入延迟和并发瓶颈，上线前要用压测数据说服自己“这套系统没有成为性能短板”。我常用的方法是用 asyncio 写一个简单的并发脚本，模拟业务方同时调用网关：

import asyncio
import httpx
import time

GATEWAY_URL = "http://localhost:8000/v1/chat/completions"
APP_KEY = "sk-app-001"
CONCURRENCY = 20
TOTAL_REQUESTS = 100

async def one_call(client: httpx.AsyncClient, seq: int):
    payload = {
        "messages": [{"role": "user", "content": "用一句话介绍你自己"}],
        "stream": True,
        "max_tokens": 64,
    }
    headers = {"X-LLM-Key": APP_KEY}
    t0 = time.perf_counter()
    async with client.stream("POST", GATEWAY_URL, json=payload, headers=headers) as resp:
        chunks = [c async for c in resp.aiter_text()]
        first_byte_time = None
        # 简化统计：这里记录总耗时和首 chunk 到达时间
    cost = time.perf_counter() - t0
    first_token_latency = first_byte_time - t0 if first_byte_time else cost
    return cost, first_token_latency

async def main():
    limits = httpx.Limits(max_connections=CONCURRENCY)
    async with httpx.AsyncClient(limits=limits) as client:
        tasks = [one_call(client, i) for i in range(TOTAL_REQUESTS)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
    costs = [r[0] for r in results if isinstance(r, tuple)]
    first_latencies = [r[1] for r in results if isinstance(r, tuple)]
    print(f"成功率: {len(costs)}/{TOTAL_REQUESTS}")
    print(f"P50 总耗时: {sorted(costs)[len(costs)//2]*1000:.0f} ms")
    print(f"P95 首字时延: {sorted(first_latencies)[int(len(first_latencies)*0.95)]*1000:.0f} ms")

asyncio.run(main())

这个脚本的价值有两个。第一，能直接测出网关在 20 并发下是否出现连接堆积或超时；第二，对比“直连上游”和“经网关转发”两个模式下的首字时延差——一般会多 10 到 30 毫秒，如果超过 100 毫秒，说明你的网关转发层有问题，大概率是流式处理里做了逐 chunk 的同步 IO。压测时建议在网关日志里同时记录 gateway_start 和 upstream_response_start 两个时间戳，用它们算出网关自身耗时，避免把上游延迟也归罪到网关头上。

6.2 故障注入：把上游模拟成“挂了”“超时”“限流”三种状态

性能达标后，还要验证网关在故障下的行为。我的做法是做一个 mock 上游服务，能配置成三种故障模式：立即返回 500、一直不返回数据、返回 429 限流。然后分别测试网关的表现。预期结果是：500 时网关返回统一的 502 错误码并触发重试；不下发数据时读超时生效，网关返回 504；429 时网关返回 429 并带上 Retry-After 头。如果哪一项不符合预期，故障处理代码就需要补强。这个故障注入脚本不复杂，写个几十行的 FastAPI mock 服务就行，但能帮你发现很多只有生产环境才会暴露的问题。

做完压测和故障注入，这套网关系统才算真正能交出去。我在多个项目里重复这套流程的体会是：LLM API 统一管理的难点从来不在写请求转发那几十行代码，而在把路由、权限、计费、限流这些横切关注点都收口到一条链路上，并且每一条都有日志和数据兜底。先把最小核心跑通，再按“先能转、再能控、后能算”的顺序逐步加厚，应该是最省力的落地路径。希望帮到你。

本文还有配套的精品资源，点击获取