Python + FastAPI 实现大模型 API 网关：统一路由、失败重试、并发控制与成本统计

当一个应用同时接入本地模型和多个云端模型时，把 base_url、API Key、模型名、超时和重试逻辑散落在业务代码中，很快就会失控。本文使用 Python、FastAPI、HTTPX 和 Pydantic 实现一个可以运行的大模型 API 网关：客户端只认识模型别名，网关负责上游路由、密钥隔离、并发控制、谨慎重试、Token 成本统计和统一错误返回。

本文配套项目已经完成 4 组自动化测试，覆盖模型路由、429 重试、成本计算、读取超时不重试、FastAPI Lifespan、健康检查和网关鉴权。

---

一、完成后的效果

假设我们有两个模型来源：

local-chat  → 本机 Ollama
cloud-chat  → 某个 OpenAI 兼容云服务

客户端不需要知道真实地址、真实模型名和上游密钥，只调用网关：

curl.exe http://127.0.0.1:9000/v1/chat/completions `
  -H "Authorization: Bearer change-me" `
  -H "Content-Type: application/json" `
  -d '{
    "model":"local-chat",
    "messages":[{"role":"user","content":"解释什么是大模型 API 网关"}],
    "temperature":0,
    "max_tokens":300,
    "stream":false
  }'

网关会把 local-chat 映射为配置中的真实模型，并将请求发送到对应上游。成功响应保持 OpenAI Chat Completions 风格，同时通过响应头返回网关信息：

X-Gateway-Trace-ID: 7c0c...
X-Gateway-Latency-MS: 823.41
X-Gateway-Cost-USD: 0.000600

还可以查询累计统计：

curl.exe http://127.0.0.1:9000/admin/stats `
  -H "Authorization: Bearer change-me"

返回示例：

{
  "local-chat": {
    "requests": 12,
    "success": 11,
    "failed": 1,
    "retries": 2,
    "prompt_tokens": 3200,
    "completion_tokens": 1400,
    "cost_usd": "0.000000",
    "average_latency_ms": 912.35
  }
}

本地模型的价格配置为 0，所以成本为 0。云端模型价格由使用者按照实际供应商官方价格手动配置，本文不会写入可能很快过期的价格数字。

---

二、为什么不应该在每个业务模块里直接调用模型

一个只有单模型的 Demo，通常这样写：

client = OpenAI(api_key="...", base_url="...")
response = client.chat.completions.create(...)

当系统逐渐接入多个模型，问题会出现：

• 每个模块保存一份 API Key；
• 模型名、地址和超时散落在代码里；
• 有的地方重试 3 次，有的地方完全不重试；
• 429、502、读取超时被当成同一种错误；
• 无法统一统计 Token 和费用；
• 无法限制某个模型的并发；
• 业务方直接依赖供应商模型名，迁移成本越来越高。

API 网关的价值，是把这些横切能力集中在一层：

业务应用
   ↓ 统一模型别名和请求格式
大模型 API 网关
   ├─ 鉴权
   ├─ 路由
   ├─ 并发控制
   ├─ 超时与重试
   ├─ 错误转换
   ├─ Token / 成本统计
   └─ Trace ID
       ↓
本地 Ollama / 云端 OpenAI 兼容服务

本文实现的是一个适合学习和小型内部项目的最小网关，不会把它包装成完整生产平台。

---

三、项目目录

020-llm-gateway/
├─ app.py
├─ providers.example.json
├─ requirements.txt
├─ test_app.py
└─ README.md

创建虚拟环境并安装依赖：

python -m venv .venv
.\.venv\Scripts\Activate.ps1
pip install -r requirements.txt

requirements.txt：

fastapi>=0.116,<1
httpx>=0.28,<1
pydantic>=2.11,<3
uvicorn>=0.35,<1
pytest>=8,<10

建议在真实项目中生成锁文件，避免未来安装到未经验证的新版本。本文测试时使用的具体版本记录在文末。

---

四、用模型别名隔离真实供应商

复制配置文件：

Copy-Item providers.example.json providers.json

providers.json：

{
  "providers": [
    {
      "alias": "local-chat",
      "base_url": "http://127.0.0.1:11434/v1",
      "upstream_model": "qwen3:8b",
      "api_key_env": null,
      "input_price_per_million": 0,
      "output_price_per_million": 0,
      "max_concurrency": 2,
      "timeout_seconds": 120,
      "retries": 1,
      "backoff_base_seconds": 0.5
    },
    {
      "alias": "cloud-chat",
      "base_url": "https://api.example.com/v1",
      "upstream_model": "replace-with-real-model",
      "api_key_env": "CLOUD_LLM_API_KEY",
      "input_price_per_million": 0,
      "output_price_per_million": 0,
      "max_concurrency": 8,
      "timeout_seconds": 120,
      "retries": 1,
      "backoff_base_seconds": 0.5
    }
  ]
}

云端地址、模型名和价格都是占位值，必须根据实际供应商官方文档修改。

模型别名带来三个好处：

1. 客户端只认识 local-chat，不知道真实供应商；
2. 更换上游模型时，不必修改所有业务代码；
3. 客户端不能提交任意模型名，模型使用范围由网关白名单控制。

---

五、请求模型与配置校验

使用 Pydantic 定义配置和请求：

from decimal import Decimal
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator


class ProviderConfig(BaseModel):
    alias: str = Field(pattern=r"^[a-zA-Z0-9_-]+$")
    base_url: str
    upstream_model: str
    api_key_env: str | None = None
    input_price_per_million: Decimal = Decimal("0")
    output_price_per_million: Decimal = Decimal("0")
    max_concurrency: int = Field(default=2, ge=1, le=1000)
    timeout_seconds: float = Field(default=120, gt=0)
    retries: int = Field(default=1, ge=0, le=5)
    backoff_base_seconds: float = Field(default=0.5, ge=0, le=30)

    @model_validator(mode="after")
    def validate_base_url(self):
        if not self.base_url.startswith(("http://", "https://")):
            raise ValueError("base_url must start with http:// or https://")
        self.base_url = self.base_url.rstrip("/")
        return self


class ChatMessage(BaseModel):
    role: Literal["system", "user", "assistant", "tool"]
    content: str


class ChatRequest(BaseModel):
    model_config = ConfigDict(extra="forbid")

    model: str
    messages: list[ChatMessage] = Field(min_length=1)
    temperature: float | None = Field(default=None, ge=0, le=2)
    top_p: float | None = Field(default=None, gt=0, le=1)
    max_tokens: int | None = Field(default=None, ge=1)
    stop: str | list[str] | None = None
    stream: Literal[False] = False

这里故意只支持非流式 Chat Completions，并禁止未知字段。

“OpenAI 兼容”不代表所有服务都支持完全相同的参数。如果网关不加约束，客户端可能把某个供应商专属字段传给另一个服务，最后得到难以解释的错误。

当需要支持视觉、工具调用或流式响应时，应逐项增加 Schema 和兼容测试，而不是直接允许任意 JSON 穿透。

---

六、为什么 HTTPX 客户端要在 Lifespan 中创建

HTTPX 官方文档提醒：不要在请求热循环中反复创建 AsyncClient，否则无法充分利用连接池。

FastAPI 推荐使用 Lifespan 管理跨请求共享并需要在关闭时释放的资源。我们在应用启动时创建一个客户端，在应用关闭时统一释放：

from contextlib import asynccontextmanager

import httpx
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI):
    configs = load_provider_configs("providers.json")
    max_connections = max(
        10, sum(item.max_concurrency for item in configs) * 2
    )
    limits = httpx.Limits(
        max_connections=max_connections,
        max_keepalive_connections=max(5, max_connections // 2),
    )
    timeout = httpx.Timeout(
        connect=10,
        read=120,
        write=30,
        pool=10,
    )

    async with httpx.AsyncClient(
        limits=limits,
        timeout=timeout,
    ) as client:
        app.state.gateway = Gateway(configs, client)
        yield


app = FastAPI(lifespan=lifespan)

HTTPX 把超时分为四类：

• connect：建立连接等待时间；
• read：等待响应数据块的时间；
• write：发送请求数据块的时间；
• pool：等待连接池可用连接的时间。

大模型首 Token 可能较慢，不能机械使用普通 Web API 的短读取超时。但也不应该把所有超时设为 None，否则故障请求可能长期占用连接和并发槽位。

---

七、并发控制不是简单限制连接数

每个模型服务的承载能力不同。一个本地 8B 模型可能只适合两路并发，而云端服务可以承载更多请求。

为每个模型别名创建独立 Semaphore：

import asyncio
from dataclasses import dataclass, field


@dataclass
class RuntimeProvider:
    config: ProviderConfig
    semaphore: asyncio.Semaphore = field(init=False)

    def __post_init__(self):
        self.semaphore = asyncio.Semaphore(
            self.config.max_concurrency
        )

发送请求时获取并发槽位：

async with provider.semaphore:
    response = await client.post(...)

这解决的是单进程内的并发上限，不是完整分布式限流。

如果 Uvicorn 启动 4 个 Worker，每个进程都有自己的 Semaphore。配置为 8 时，整体最多可能出现约 32 路并发。多进程、多实例部署必须使用 Redis、集中式网关或供应商侧配额控制，不能把内存 Semaphore 当作全局限流器。

---

八、重试最容易被写错

许多示例把所有异常统一重试 3 次：

try:
    ...
except Exception:
    retry()

对大模型接口，这可能造成重复生成和重复计费。

8.1 本文只自动重试这些情况

• ConnectError：没有成功建立上游连接；
• ConnectTimeout：建立连接超时；
• HTTP 429、502、503、504，并且没有超过配置次数。

429 按 RFC 6585 表示请求过多，响应可以提供 Retry-After。网关优先服从该字段，没有时才使用指数退避和随机抖动。

8.2 读取超时不自动重试

ReadTimeout 发生时，上游可能已经接收请求、开始生成甚至完成计费，只是客户端没有及时收到结果。此时自动重试会产生第二次调用。

所以本文返回：

{
  "error": {
    "code": "upstream_timeout",
    "message": "upstream result is unknown; request was not retried"
  }
}

这不是说读取超时永远不能重试，而是必须有可靠的幂等机制、请求状态查询或供应商明确保证后再做。

8.3 Retry-After 解析

Retry-After 可能是秒数，也可能是 HTTP 日期：

from datetime import datetime, timezone
from email.utils import parsedate_to_datetime


def parse_retry_after(value: str | None) -> float | None:
    if not value:
        return None
    try:
        return max(0.0, float(value))
    except ValueError:
        try:
            parsed = parsedate_to_datetime(value)
            if parsed.tzinfo is None:
                parsed = parsed.replace(tzinfo=timezone.utc)
            return max(
                0.0,
                (parsed - datetime.now(timezone.utc)).total_seconds(),
            )
        except (TypeError, ValueError, OverflowError):
            return None

退避计算：

def backoff(config, attempt, retry_after):
    server_delay = parse_retry_after(retry_after)
    if server_delay is not None:
        return min(server_delay, 60.0)

    exponential = config.backoff_base_seconds * (2 ** attempt)
    jitter = random.uniform(0, config.backoff_base_seconds)
    return min(exponential + jitter, 60.0)

设置 60 秒上限，是为了避免错误或异常大的响应头让网关任务无限等待。实际生产系统可以把任务交还队列，而不是一直占用 Web 请求。

---

九、核心路由代码

网关根据客户端传入的别名查找配置，再把 model 替换为真实上游模型名：

class Gateway:
    def __init__(self, configs, client: httpx.AsyncClient):
        self.client = client
        self.providers = {
            config.alias: RuntimeProvider(config)
            for config in configs
        }

    def get_provider(self, alias: str):
        provider = self.providers.get(alias)
        if provider is None:
            raise GatewayFailure(
                404,
                "unknown_model",
                f"unknown model alias: {alias}",
            )
        return provider

    async def complete(self, body: ChatRequest, trace_id: str):
        provider = self.get_provider(body.model)
        config = provider.config

        payload = body.model_dump(exclude_none=True)
        payload["model"] = config.upstream_model

        url = f"{config.base_url}/chat/completions"
        headers = self.upstream_headers(config, trace_id)

        async with provider.semaphore:
            for attempt in range(config.retries + 1):
                try:
                    response = await self.client.post(
                        url,
                        json=payload,
                        headers=headers,
                        timeout=config.timeout_seconds,
                    )
                except (
                    httpx.ConnectError,
                    httpx.ConnectTimeout,
                ) as exc:
                    if attempt >= config.retries:
                        raise GatewayFailure(
                            502,
                            "upstream_connect_failed",
                            str(exc),
                        ) from exc
                    await asyncio.sleep(
                        self._backoff(config, attempt, None)
                    )
                    continue
                except httpx.TimeoutException as exc:
                    raise GatewayFailure(
                        504,
                        "upstream_timeout",
                        "upstream result is unknown; "
                        "request was not retried",
                    ) from exc

                if (
                    response.status_code
                    in {429, 502, 503, 504}
                    and attempt < config.retries
                ):
                    await asyncio.sleep(
                        self._backoff(
                            config,
                            attempt,
                            response.headers.get("Retry-After"),
                        )
                    )
                    continue

                if response.is_error:
                    raise GatewayFailure(
                        response.status_code,
                        "upstream_rejected",
                        self._safe_upstream_error(response),
                        retry_after=response.headers.get("Retry-After"),
                    )

                try:
                    response_body = response.json()
                except ValueError as exc:
                    raise GatewayFailure(
                        502,
                        "invalid_upstream_response",
                        "upstream did not return JSON",
                    ) from exc

                return response_body

        raise AssertionError("retry loop exited unexpectedly")

配套源码还会记录请求数、成功数、失败数、重试数、Token、成本和平均延迟。为了让核心逻辑更容易阅读，这里省略了重复的统计加锁代码，完整文件已经通过自动化测试。

---

十、Token 成本如何计算

只有上游返回下面的 Usage 时，网关才计算成本：

{
  "usage": {
    "prompt_tokens": 100,
    "completion_tokens": 50
  }
}

公式：

成本 = 输入 Token × 输入单价 ÷ 1,000,000
     + 输出 Token × 输出单价 ÷ 1,000,000

代码使用 Decimal，避免用二进制浮点数累计金额：

from decimal import Decimal


def calculate_cost(
    config,
    prompt_tokens: int,
    completion_tokens: int,
) -> Decimal:
    million = Decimal("1000000")
    return (
        Decimal(prompt_tokens)
        * config.input_price_per_million
        / million
        + Decimal(completion_tokens)
        * config.output_price_per_million
        / million
    )

如果输入价格为 2 美元/百万 Token，输出价格为 8 美元/百万 Token，那么 100 输入 Token 和 50 输出 Token 的成本为：

100 × 2 / 1,000,000 + 50 × 8 / 1,000,000
= 0.0002 + 0.0004
= 0.0006 美元

这个数字只用于验证公式。真实价格必须由使用者根据实际供应商的最新官方价格配置。

如果上游没有 Usage，网关不会根据字符数伪造 Token 数，也不会给出看似精确的费用。

---

十一、网关鉴权与上游密钥隔离

启动网关前设置一个入口密钥：

$env:GATEWAY_API_KEY='change-me'

依赖函数检查请求头：

from fastapi import Header


def require_gateway_key(
    authorization: str | None = Header(default=None),
):
    expected = os.getenv("GATEWAY_API_KEY")
    if not expected:
        raise GatewayFailure(
            503,
            "gateway_not_configured",
            "GATEWAY_API_KEY is missing",
        )
    if authorization != f"Bearer {expected}":
        raise GatewayFailure(
            401,
            "invalid_gateway_key",
            "invalid gateway API key",
        )

云端上游密钥不写入 JSON，只保存环境变量名：

"api_key_env": "CLOUD_LLM_API_KEY"

运行时再读取：

api_key = os.getenv(config.api_key_env)
headers["Authorization"] = f"Bearer {api_key}"

这样业务客户端只持有网关密钥，不接触上游密钥。

这个例子仍然只是最小实现。生产环境应使用密钥管理服务、短期凭据、网关级身份体系和轮换机制，不应长期依赖进程环境变量保存所有秘密。

---

十二、定义接口

import uuid

from fastapi import Depends, FastAPI, Request
from fastapi.responses import JSONResponse


@app.get("/health")
async def health(request: Request):
    gateway = request.app.state.gateway
    return {
        "status": "ok",
        "models": sorted(gateway.providers),
    }


@app.get(
    "/v1/models",
    dependencies=[Depends(require_gateway_key)],
)
async def models(request: Request):
    gateway = request.app.state.gateway
    return {
        "object": "list",
        "data": [
            {
                "id": alias,
                "object": "model",
                "owned_by": "gateway",
            }
            for alias in sorted(gateway.providers)
        ],
    }


@app.post(
    "/v1/chat/completions",
    dependencies=[Depends(require_gateway_key)],
)
async def chat_completions(
    body: ChatRequest,
    request: Request,
):
    trace_id = uuid.uuid4().hex
    gateway = request.app.state.gateway
    response_body, cost, latency_ms = await gateway.complete(
        body,
        trace_id,
    )

    headers = {
        "X-Gateway-Trace-ID": trace_id,
        "X-Gateway-Latency-MS": f"{latency_ms:.2f}",
    }
    if cost is not None:
        headers["X-Gateway-Cost-USD"] = str(
            cost.quantize(Decimal("0.000001"))
        )

    return JSONResponse(
        content=response_body,
        headers=headers,
    )


@app.get(
    "/admin/stats",
    dependencies=[Depends(require_gateway_key)],
)
async def stats(request: Request):
    return await request.app.state.gateway.stats()

响应正文尽量保持上游格式，把 Trace、延迟和成本放进响应头，可以减少客户端适配成本。

---

十三、启动与调用

先启动本地 Ollama 并准备模型，具体模型名根据机器资源选择。Ollama 官方文档说明其兼容部分 OpenAI API，默认 OpenAI 风格地址可以配置为：

http://127.0.0.1:11434/v1

然后设置网关配置：

$env:GATEWAY_CONFIG='providers.json'
$env:GATEWAY_API_KEY='change-me'

启动：

uvicorn app:app --host 127.0.0.1 --port 9000

查看健康状态：

curl.exe http://127.0.0.1:9000/health

查看可用模型别名：

curl.exe http://127.0.0.1:9000/v1/models `
  -H "Authorization: Bearer change-me"

调用 Chat Completions：

curl.exe http://127.0.0.1:9000/v1/chat/completions `
  -H "Authorization: Bearer change-me" `
  -H "Content-Type: application/json" `
  -d '{"model":"local-chat","messages":[{"role":"user","content":"你好"}],"temperature":0,"max_tokens":128,"stream":false}'

不要把 --host 改成 0.0.0.0 后直接暴露到公网。即使网关有一个静态 API Key，也不等于已经具备 TLS、账户、权限、审计、限流和防攻击能力。

---

十四、自动化测试

这类网关至少要测试三个高风险点。

14.1 429 只重试规定次数

使用 HTTPX MockTransport，第一次返回 429，第二次返回成功：

def handler(request: httpx.Request):
    if calls == 1:
        return httpx.Response(
            429,
            headers={"Retry-After": "0"},
        )
    return httpx.Response(
        200,
        json={
            "id": "chatcmpl-test",
            "choices": [],
            "usage": {
                "prompt_tokens": 100,
                "completion_tokens": 50,
            },
        },
    )

测试同时断言：

• 请求模型别名被替换为上游模型名；
• 总调用次数为 2；
• 网关记录一次重试；
• 成本等于 0.0006 美元。

14.2 ReadTimeout 不重试

def handler(_: httpx.Request):
    raise httpx.ReadTimeout("read timed out")

即使配置 retries=3，实际调用次数也必须为 1。

14.3 API 鉴权和 Lifespan

使用 ASGITransport 直接调用 FastAPI：

• /health 返回模型别名；
• /v1/models 不带密钥返回 401；
• 正确密钥返回 200。

执行：

pytest -q

本文配套代码的执行结果：

....                                                                     [100%]
4 passed in 0.60s

测试环境版本：

Python 3.12
FastAPI 0.141.1
Starlette 1.3.1
HTTPX 0.28.1
Pydantic 2.13.4
pytest 9.1.1

项目依赖采用版本范围，正式部署前应生成并保存锁文件。

---

十五、这个版本故意没有实现什么

15.1 没有流式转发

流式转发不能简单使用普通 post()。HTTPX 官方文档说明，手动流式模式必须最终调用 Response.aclose()，否则可能泄漏连接。

增加流式功能时还要处理：

• 客户端中途断开；
• 上游响应关闭；
• SSE 格式差异；
• 首 Token 延迟；
• 流式 Usage；
• 生成中途失败后的计费和重试边界。

这些值得单独实现和测试，不适合塞进一个“看起来能跑”的代码分支。

15.2 没有自动模型降级

主模型失败后自动切换备用模型，可能改变答案质量、合规范围、上下文长度和成本。没有明确路由策略与业务授权时，网关不应该偷偷换模型。

15.3 没有持久化统计

当前统计保存在进程内：

• 重启后归零；
• 多 Worker 之间不共享；
• 不适合财务结算。

生产环境应把用量写入数据库或指标系统，并使用 Trace ID 关联请求、上游记录和业务订单。

15.4 没有完整用户体系

一个静态网关密钥只能用于本地或受控内部环境。多团队使用时需要：

• 不同调用方独立凭据；
• 模型权限；
• 每日/每月预算；
• QPS 和并发配额；
• 密钥轮换与吊销；
• 审计日志和敏感内容脱敏。

---

十六、生产化升级顺序

不要一开始就堆满所有功能。更稳妥的顺序是：

第一阶段：模型别名 + 密钥隔离 + 统一错误
第二阶段：超时 + 谨慎重试 + 并发控制
第三阶段：Token / 成本 / 延迟指标
第四阶段：持久化用量 + 调用方配额
第五阶段：流式转发 + 断开处理
第六阶段：多实例限流 + 高可用
第七阶段：经过验证的路由与降级策略

每增加一个能力，都要加入故障测试。大模型网关最危险的地方不是正常请求能否成功，而是超时、限流、客户端断开和上游异常时，系统是否会重复请求、泄漏连接、暴露密钥或错误计费。

---

十七、总结

一个可用的大模型 API 网关，不只是把请求转发到另一个 URL。

本文实现了这些核心能力：

1. 用模型别名隔离真实供应商与模型名；
2. 用环境变量隔离上游密钥；
3. 复用一个 HTTPX AsyncClient 和连接池；
4. 按模型配置单进程并发上限；
5. 对 429、502、503、504 和连接失败进行有限重试；
6. 不自动重试结果未知的读取超时；
7. 根据上游 Usage 和手动配置价格计算成本；
8. 返回 Trace ID、延迟与成本响应头；
9. 通过自动化测试验证关键故障路径。

如果应用只调用一个本地模型，直接连接 Ollama 完全没有问题。当模型来源、调用方和成本逐渐增多时，再引入这样一个边界清晰的网关，才能避免路由、密钥、重试和统计逻辑散落到整个项目。

真正的工程价值不是“多套一层 FastAPI”，而是把不确定的模型调用封装进可配置、可限制、可观察、可测试的系统边界。

---

参考资料

以下资料于 2026 年 7 月 30 日核对：

• FastAPI Lifespan Events
• FastAPI Handling Errors
• HTTPX Async Support
• HTTPX Timeouts
• HTTPX Resource Limits
• Ollama OpenAI Compatibility
• RFC 6585：429 Too Many Requests
• RFC 9110：Retry-After