Python LLM 网关：调用后端接口

Python 轻量网关的作用

一个实用的 LLM 网关 Python 环境搭建比人们预想的要简单。你不需要一个完整的平台。轻量级网关是一个小型 Web 服务，位于你的应用和后端接口之间，提供四项功能：为你的客户端提供统一的认证 token、超时与重试策略、限制上游并发请求数，以及干净的流式输出透传。你的应用可以像调用 OpenAI 一样与它交互，它会将请求转发至真正的后端。

本文使用 FastAPI 和 httpx 构建该网关，后端使用 https://api.openllmgateway.com/v1，模型 ID 为 uncensored。如果你仍在考虑是否真的需要网关，请先阅读 什么是 LLM 网关。本文的重点是代码实现。

设置与非流式路由

安装依赖项并设置两个环境变量：后端密钥，以及你分发给自有应用的令牌列表（以逗号分隔）。你的应用看不到后端密钥。

Copypip install fastapi uvicorn httpx openai
export BACKEND_KEY="your-backend-key"
export GATEWAY_TOKENS="app-one-token,app-two-token"

服务的前半部分涵盖配置、认证以及带有重试功能的普通 JSON 路由：

Copyimport asyncio
import os

import httpx
from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse, Response, StreamingResponse

BACKEND = "https://api.openllmgateway.com/v1"
BACKEND_KEY = os.environ["BACKEND_KEY"]
TOKENS = set(os.environ["GATEWAY_TOKENS"].split(","))

app = FastAPI()
http = httpx.AsyncClient(
    timeout=httpx.Timeout(connect=5, read=120, write=30, pool=10)
)
slots = asyncio.Semaphore(8)  # matches the full-key concurrency limit
UP = {"Authorization": "Bearer " + BACKEND_KEY, "Content-Type": "application/json"}


def check_auth(req: Request):
    token = req.headers.get("authorization", "").removeprefix("Bearer ").strip()
    if token not in TOKENS:
        raise HTTPException(401, "invalid gateway token")


@app.post("/v1/chat/completions")
async def chat(req: Request):
    check_auth(req)
    body = await req.json()
    body["model"] = "uncensored"
    if body.get("stream"):
        return await relay_stream(body)
    async with slots:
        for attempt in range(4):
            try:
                r = await http.post(BACKEND + "/chat/completions", json=body, headers=UP)
            except (httpx.ConnectError, httpx.ReadTimeout):
                await asyncio.sleep(2 ** attempt)
                continue
            if r.status_code == 429:
                await asyncio.sleep(float(r.headers.get("Retry-After", "2")))
                continue
            if r.status_code >= 500:
                await asyncio.sleep(2 ** attempt)
                continue
            return JSONResponse(r.json(), status_code=r.status_code)
    raise HTTPException(502, "backend unavailable after retries")

注意，网关强制将 model 映射为 uncensored。客户端无法请求其他模型，这是一种将策略集中管理在单一位置的简单方法。

不泄漏槽位的流式输出透传

流式输出是轻量网关容易出错的地方。如果你读取整个上游响应体后再发送，用户直到最后才能看到内容。如果客户端断开连接时未释放并发槽位，最终会导致阻塞。API 将流式请求视为占用并发槽位直到流结束，因此你的本地信号量也应如此操作。

Copyasync def relay_stream(body):
    await slots.acquire()
    try:
        req = http.build_request("POST", BACKEND + "/chat/completions", json=body, headers=UP)
        upstream = await http.send(req, stream=True)
    except httpx.HTTPError:
        slots.release()
        raise HTTPException(502, "backend unreachable")

    if upstream.status_code >= 300:
        data = await upstream.aread()
        await upstream.aclose()
        slots.release()
        return Response(data, status_code=upstream.status_code, media_type="application/json")

    async def pump():
        try:
            async for chunk in upstream.aiter_raw():
                yield chunk
        finally:
            await upstream.aclose()
            slots.release()

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

finally 块在流式输出完成、客户端断开连接或发生错误时运行，从而确保槽位始终被释放。此处特意未包含重试逻辑。一旦字节已发送给客户端，就无法干净地重启，因此仅在第一个数据块发送前进行重试。在此版本中，上方的状态检查是唯一的门控检查。

从客户端调用网关

使用 uvicorn gateway:app --port 8000 运行它。你的应用现在使用标准的 OpenAI 客户端，将网关作为 Base URL，并使用你自己的 token 作为 API 密钥：

Copyfrom openai import OpenAI

# Talks to the local gateway, which forwards to https://api.openllmgateway.com/v1
client = OpenAI(base_url="http://localhost:8000/v1", api_key="app-one-token")

for chunk in client.chat.completions.create(
    model="uncensored",
    messages=[{"role": "user", "content": "Give me three names for a hiking app."}],
    stream=True,
):
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

你可以以相同方式添加其他路由。GET /v1/models 和 POST /v1/responses 在后端可用，函数调用和 JSON 模式无需特殊处理，因为网关会原样转发请求体。最大请求体为 8 MB，如果网关面向互联网，请在网关前设置匹配的限制。

正确处理 429 错误

后端在三种情况下返回 429 状态码，你的网关应能区分这些情况。rate_limited 错误代码表示你超过了每分钟请求数限制：完整密钥为 300，试用密钥为 60。concurrency 代码表示并发请求过多：完整密钥为 8，试用密钥为 2。临时过载也返回 429 状态码，并附带 Retry-After: 5 头。

上述信号量首先防止了并发请求过多的情况，前提是网关是该密钥唯一的客户端。对于每分钟请求数限制，请添加一个简单的令牌桶或每分钟计数器。上述代码对每个 429 响应都遵循 Retry-After 头，这涵盖了过载情况。如果你仍然看到 concurrency 错误，说明有其他进程正在使用该密钥，或者旧的流式输出连接仍未关闭。

记录每次上游调用的 X-Request-Id 标头以便追踪不良响应，并记录 X-Balance-USD 以监控你的额度。网关是设置余额低于阈值时告警的好地方。AI 网关生产检查清单 涵盖了其余部分：密钥、超时、健康检查和监控。

网关内的成本追踪

由于每个响应都包含 token 使用情况，网关可以记录每个客户端令牌的支出。价格为每 1M 输入 token $0.15，每 1M 输出 token $0.50。对于使用 1,200 个提示词 token 和 350 个补全 token 的调用，成本为 1,200 x $0.15 / 1M 加上 350 x $0.50 / 1M，即 $0.00018 加上 $0.000175，总计 $0.000355。

Copydef cost_usd(usage):
    return usage["prompt_tokens"] * 0.15 / 1e6 + usage["completion_tokens"] * 0.50 / 1e6

在解析 JSON 响应后添加此计算，并与客户端令牌一起存储。对于流式输出，要求客户端在最后一个块中包含使用量，使用 stream_options。失败或被拒绝的请求不收费，因此只计算成功的响应。当前费率和充值奖励见 定价页面。

后续步骤

你现在拥有了一个包含认证、重试、并发限制和流式输出透传的网关，代码量约 80 行。自然的扩展包括每令牌预算、将请求日志记录到你自己的存储，以及用于故障转移的第二个后端。保持代码精简；一旦你需要同时使用其中的许多功能，专用的网关项目可能更合适。

要运行它，你需要后端密钥。注册并通过 API 密钥页面 一键领取 $0.50 试用额度，然后在 API 文档 中查看参数和错误格式。请记住，试用密钥仅允许 2 个并发请求，因此测试时将信号量设为 2。

更多指南

• LLM 网关与 OpenRouter：你需要哪一个？
• AI 网关检查清单：无审查 LLM API