LLM-Gateway in Python: Aufruf eines Backend-Endpunkts

Was ein dünnes Gateway in Python tut

Ein nützliches LLM-Gateway-Python-Setup ist kleiner, als viele erwarten. Du benötigst keine vollständige Plattform. Ein dünnes Gateway ist ein kleiner Webdienst, der zwischen deinen Apps und einem Backend-Endpunkt sitzt und vier Dinge hinzufügt: ein gemeinsames Auth-Token für deine eigenen Clients, eine Timeout- und Wiederholungsrichtlinie, eine Begrenzung für parallele Upstream-Aufrufe und sauberes Streaming-Passthrough. Deine Apps kommunizieren damit, als wäre es OpenAI, und es leitet an das echte Backend weiter.

Dieser Artikel baut das mit FastAPI und httpx auf, wobei https://api.openllmgateway.com/v1 als Backend und die Modell-ID uncensored verwendet werden. Wenn du noch entscheidest, ob du überhaupt ein Gateway brauchst, lies zuerst was ein LLM-Gateway ist. Der Fokus liegt hier auf dem Code.

Einrichtung und die Nicht-Streaming-Route

Installiere die Abhängigkeiten und setze zwei Umgebungsvariablen: den Backend-Schlüssel und eine durch Komma getrennte Liste von Tokens, die du an deine eigenen Apps ausgibst. Deine Apps sehen den Backend-Schlüssel nie.

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

Die erste Hälfte des Dienstes umfasst Konfiguration, Auth und die reine JSON-Route mit Wiederholungen:

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")

Beachte, dass das Gateway model auf uncensored erzwingt. Clients können nichts anderes anfordern, was eine einfache Möglichkeit ist, eine Richtlinie an einem Ort zu halten.

Streaming-Passthrough, das keine Slots leakt

Streaming ist der Punkt, an dem dünne Gateways scheitern. Wenn du den gesamten Upstream-Körper liest und dann sendest, sieht der Benutzer nichts, bis das Ende erreicht ist. Wenn du vergisst, den Parallelitäts-Slot freizugeben, wenn ein Client die Verbindung abbricht, blockierst du dich schließlich selbst. Die API behandelt eine gestreamte Anfrage so, als würde sie ihren Parallelitäts-Slot bis zum Ende des Streams belegen, also sollte dein lokales Semaphore dasselbe tun.

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")

Der finally-Block wird ausgeführt, wenn der Stream abgeschlossen ist, der Client abbricht oder ein Fehler auftritt, sodass der Slot immer zurückkehrt. Wiederholungen sind hier bewusst nicht enthalten. Sobald Bytes an den Client gesendet wurden, kannst du nicht sauber neu starten, also wiederhole nur vor dem ersten Chunk, und in dieser Version ist die Statusprüfung oben die einzige Hürde.

Aufruf deines Gateways von einem Client

Führe es mit uvicorn gateway:app --port 8000 aus. Deine Apps verwenden jetzt den Standard-OpenAI-Client mit dem Gateway als Basis-URL und einem deiner eigenen Tokens als Schlüssel:

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)

Du kannst die anderen Routen auf die gleiche Weise hinzufügen. GET /v1/models und POST /v1/responses sind auf dem Backend verfügbar, und Function Calling sowie JSON-Modus benötigen keine spezielle Behandlung, da das Gateway den Körper unverändert weiterleitet. Der maximale Anfragekörper beträgt 8 MB, also setze ein passendes Limit vor das Gateway, wenn es dem Internet ausgesetzt ist.

429 korrekt behandeln

Das Backend antwortet in drei Situationen mit 429, und dein Gateway sollte sie unterscheiden. Ein Fehlercode rate_limited bedeutet, dass du die Anfragen pro Minute überschritten hast: 300 bei einem Vollschlüssel, 60 bei einem Testschlüssel. Ein Code concurrency bedeutet zu viele Anfragen gleichzeitig: 8 bei einem Vollschlüssel, 2 bei einem Testschlüssel. Temporäre Überlastung gibt ebenfalls 429 mit Retry-After: 5 zurück.

Das Semaphore oben verhindert das Parallelitätsproblem von vornherein, solange das Gateway der einzige Client auf diesem Schlüssel ist. Für Anfragen pro Minute füge einen einfachen Token-Bucket oder einen Zähler pro Minute hinzu. Der obige Code berücksichtigt Retry-After bei jeder 429, was Überlastung abdeckt. Wenn du trotzdem concurrency-Fehler siehst, verwendet ein anderer Prozess denselben Schlüssel oder ein alter Stream ist noch offen.

Logge den X-Request-Id-Header für jede Upstream-Anfrage, damit du eine fehlerhafte Antwort nachverfolgen kannst, und logge X-Balance-USD, um dein Guthaben zu überwachen. Ein Gateway ist ein guter Ort, um zu warnen, wenn das Guthaben unter einen Schwellenwert fällt. Die AI-Gateway-Produktions-Checkliste deckt den Rest ab: Geheimnisse, Timeouts, Health Checks und Monitoring.

Kostenverfolgung im Gateway

Da jede Antwort Token-Nutzung enthält, kann das Gateway Ausgaben pro Client-Token aufzeichnen. Der Preis beträgt $0,15 pro 1M Input-Token und $0,50 pro 1M Output-Token. Für einen Aufruf, der 1.200 Prompt-Token und 350 Completion-Token verwendet hat, beträgt die Kosten 1.200 x $0,15 / 1M plus 350 x $0,50 / 1M, was $0,00018 plus $0,000175 oder $0,000355 entspricht.

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

Füge es hinzu, nachdem die JSON-Antwort analysiert wurde, und speichere sie mit dem Client-Token. Für Streams bitte den Client, die Nutzung im letzten Chunk mit stream_options anzugeben. Fehlgeschlagene oder verweigerte Anfragen werden nicht berechnet, also zähle nur erfolgreiche Antworten. Aktuelle Preise und Auflade-Boni findest du auf der Preisseite.

Nächste Schritte

Du hast jetzt ein Gateway mit Authentifizierung, Wiederholungen, einem Concurrency-Limit und Streaming-Passthrough in etwa 80 Zeilen. Natürliche Erweiterungen sind Token-Budgets, Anfrageprotokollierung in deinem eigenen Speicher und ein zweiter Backend für Failover. Halte den Code klein; sobald du viele davon gleichzeitig brauchst, passt ein dediziertes Gateway-Projekt besser.

Um es auszuführen, benötigst du einen Backend-Schlüssel. Melde dich an und beanspruche das $0,50 Testguthaben mit einem Klick auf der API-Schlüssel-Seite, überprüfe dann Parameter und Fehlerformate in den API-Dokumenten. Denke daran, dass ein Testschlüssel nur 2 parallele Anfragen erlaubt, also setze das Semaphore auf 2 während der Tests.

Weitere Anleitungen

• LLM Gateway vs. OpenRouter: Welches brauchst du?
• AI-Gateway-Checkliste: Unzensierte LLM-API