Python LLM 閘道器：呼叫後端端點

Python 輕量閘道器的功能

實用的 llm gateway python 設定比預期更輕量。你不需要完整的平台。輕量閘道是一個小型網頁服務，位於你的應用程式與後端端點之間，並提供四項功能：為你的用戶端提供單一共享驗證權杖、逾時與重試政策、對上游呼叫的並行限制，以及簡潔的串流直通。你的應用程式會將其視為 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，並將你自己的其中一個權杖作為金鑰：

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。暫時過載也會返回帶有 Retry-After: 5 的 429。

上方的訊號量可防止並行情況，前提是閘道是該金鑰的唯一用戶端。針對每分鐘請求數，請新增簡單的令牌桶或每分鐘計數器。上述程式碼會遵守每個 429 的 Retry-After，這涵蓋了過載情況。如果你仍看到 concurrency 錯誤，表示有其他程序正在使用相同金鑰，或舊的串流仍處於開啟狀態。

記錄每個上游呼叫的 X-Request-Id 標頭，以便追蹤不良回應，並記錄 X-Balance-USD 以監控你的額度。閘道是設定餘額低於閾值時發出警報的好地方。AI 閘道生產環境檢查清單 涵蓋其餘部分：機密、逾時、健康檢查和監控。

閘道內的成本追蹤

由於每個回應都包含 token 用量，閘道器可以記錄每個客戶端 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