LLM-gateway in Python: Een backend-endpoint aanroepen

Wat een dunne gateway in Python doet

Een nuttige llm-gateway python-setup is kleiner dan mensen verwachten. Je hebt geen volledig platform nodig. Een dunne gateway is een kleine web-service die tussen je apps en een backend-endpoint staat en vier dingen toevoegt: één gedeelde auth token voor je eigen clients, een timeout en retry beleid, een cap op gelijktijdige upstream calls, en schone streaming passthrough. Je apps praten ermee alsof het OpenAI is, en het stuurt door naar de echte backend.

Dit artikel bouwt dat met FastAPI en httpx, met https://api.openllmgateway.com/v1 als backend en het model ID uncensored. Als je nog twijfelt of je überhaupt een gateway nodig hebt, lees dan eerst wat een LLM gateway is. Het punt hier is de code.

Setup en de niet-streaming route

Installeer de dependencies en stel twee environment variables in: de backend key, en een door komma's gescheiden lijst van tokens die je aan je eigen apps uitgeeft. Je apps zien de backend key nooit.

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

De eerste helft van de service behandelt configuratie, auth en de gewone JSON route met retries:

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

Merk op dat de gateway model dwingt naar uncensored. Clients kunnen niets anders opvragen, wat een eenvoudige manier is om één beleid op één plek te houden.

Streaming passthrough dat geen slots lekt

Streaming is waar dichte gateways het misdoen. Als je de hele upstream-body leest en daarna verstuurt, ziet de gebruiker niets tot het einde. Als je vergeet de concurrency-slot vrij te geven wanneer een client de verbinding verbreekt, blokkeer je jezelf uiteindelijk. De API behandelt een gestreamd verzoek als bezet van zijn concurrency-slot tot de stream eindigt, dus je lokale semaphore moet hetzelfde doen.

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

De finally-blok wordt uitgevoerd wanneer de stream compleet is, wanneer de client de verbinding verbreekt, of wanneer er een fout optreedt, dus het slot komt altijd terug. Retries zijn hier bewust weggelaten. Zodra er bytes naar de client zijn gestuurd, kun je niet schoon opnieuw beginnen, dus retry alleen voordat het eerste chunk is verstuurd, en in deze versie is de statuscontrole hierboven de enige poort.

Je gateway aanroepen vanaf een client

Voer het uit met uvicorn gateway:app --port 8000. Je apps gebruiken nu de standaard OpenAI-client met de gateway als basis-URL en een van je eigen tokens als sleutel:

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)

Je kunt de andere routes op dezelfde manier toevoegen. GET /v1/models en POST /v1/responses zijn beschikbaar op de backend, en function calling en JSON-modus vereisen geen speciale afhandeling omdat de gateway de body ongewijzigd doorstuurt. De maximale grootte van de request body is 8 MB, dus stel een overeenkomstige limiet in voor de gateway als deze het internet opgaat.

429 correct afhandelen

De backend antwoordt met 429 in drie situaties, en je gateway moet deze onderscheiden. Een rate_limited foutcode betekent dat je de limiet voor verzoeken per minuut hebt overschreden: 300 op een volledige sleutel, 60 op een proefsleutel. Een concurrency code betekent te veel verzoeken tegelijk: 8 op een volledige sleutel, 2 op een proefsleutel. Tijdelijke overbelasting retourneert ook 429 met Retry-After: 5.

De semaphore hierboven voorkomt het gelijktijdigheidsprobleem van tevoren, zolang de gateway de enige client op die sleutel is. Voor verzoeken per minuut kun je een eenvoudige token bucket of een teller per minuut toevoegen. De bovenstaande code respecteert Retry-After bij elke 429, wat overbelasting dekt. Als je toch concurrency fouten ziet, gebruikt een ander proces dezelfde sleutel, of een oude stream is nog steeds open.

Log de X-Request-Id header voor elke upstream-oproep zodat je een slecht antwoord kunt traceren, en log X-Balance-USD om je tegoed te bewaken. Een gateway is een goede plek om een waarschuwing te geven wanneer het saldo onder een bepaalde drempel daalt. De productiechecklist voor AI-gateways behandelt de rest: geheimen, time-outs, health checks en monitoring.

Kostentracking binnen de gateway

Omdat elk antwoord token-gebruik bevat, kan de gateway de kosten per client-token registreren. De prijs is $0,15 per 1M input tokens en $0,50 per 1M antwoord tokens. Voor een oproep die 1.200 prompt tokens en 350 completion tokens gebruikte, zijn de kosten 1.200 x $0,15 / 1M plus 350 x $0,50 / 1M, wat $0,00018 plus $0,000175 is, oftewel $0,000355.

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

Voeg het toe nadat de JSON response is geparsed en sla het op met de client token. Voor streams, vraag de client om usage in het laatste chunk op te nemen met stream_options. Mislukte of geweigerde requests worden niet in rekening gebracht, dus tel alleen succesvolle responses. Huidige tarieven en top-up bonussen staan op de prijspagina.

Volgende stappen

Je hebt nu een gateway met auth, retries, een concurrency cap en streaming passthrough in ongeveer 80 regels. Natuurlijke extensies zijn per-token budgets, request logging naar je eigen opslag, en een tweede backend voor failover. Houd de code klein; zodra je er veel tegelijk nodig hebt, kan een dedicated gateway project beter passen.

Om het te draaien, heb je een backend key nodig. Meld je aan en claim het $0,50 trial credit met één klik op de API-sleutelpagina, controleer dan parameters en error formats in de API-documentatie. Onthoud dat een trial key slechts 2 gelijktijdige requests toelaat, dus stel de semaphore in op 2 tijdens het testen.

Meer handleidingen

• LLM Gateway vs OpenRouter: Wat heb je nodig?
• AI Gateway Checklist: Ongecensureerd LLM API