Gateway LLM in Python: chiamata a un endpoint backend

Cosa fa un gateway sottile in Python

Un'utile configurazione gateway LLM in Python è più piccola del previsto. Non ti serve una piattaforma completa. Un gateway sottile è un piccolo servizio web che si posiziona tra le tue app e un endpoint backend aggiungendo quattro cose: un token auth condiviso per i tuoi client, una policy di timeout e retry, un limite alle chiamate upstream concorrenti e un clean streaming passthrough. Le tue app parlano con esso come se fosse OpenAI e lui inoltra al vero backend.

Questo articolo costruisce tutto con FastAPI e httpx, usando https://api.openllmgateway.com/v1 come backend e l'ID modello uncensored. Se stai ancora decidendo se ti serve un gateway, leggi prima cos'è un gateway LLM. Qui l'obiettivo è il codice.

Configurazione e percorso non streaming

Installa le dipendenze e imposta due variabili d'ambiente: la chiave del backend e un elenco separato da virgole di token che distribuisci alle tue app. Le tue app non vedono mai la chiave del backend.

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

La prima metà del servizio copre configurazione, autenticazione e il percorso JSON semplice con retry:

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

Nota che il gateway forza model a uncensored. I client non possono richiedere nient'altro, il che è un modo semplice per mantenere una policy in un unico punto.

Trasmissione in streaming che non perde slot

Lo streaming è dove i gateway sottili sbagliano. Se leggi l'intero body upstream e poi lo invii, l'utente non vede nulla fino alla fine. Se dimentichi di rilasciare lo slot di concorrenza quando un client si disconnette, finisci per bloccarti. L'API tratta una richiesta in streaming come occupante il suo slot di concorrenza fino alla fine dello stream, quindi il tuo semaforo locale dovrebbe fare lo stesso.

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

Il blocco finally viene eseguito quando lo stream termina, quando il client si disconnette o quando si verifica un errore, così lo slot viene sempre rilasciato. I retry sono deliberatamente assenti qui. Una volta che i byte sono stati inviati al client non puoi riavviare in modo pulito, quindi fai retry solo prima del primo chunk e in questa versione il controllo dello status sopra è l'unica porta.

Chiamare il tuo gateway da un client

Eseguilo con uvicorn gateway:app --port 8000. Le tue app ora usano il client OpenAI standard con il gateway come base URL e uno dei tuoi token come chiave:

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)

Puoi aggiungere gli altri percorsi allo stesso modo. GET /v1/models e POST /v1/responses sono disponibili sul backend, e tool calling e JSON mode non richiedono gestione speciale perché il gateway inoltra il body intatto. Il max body della richiesta è 8 MB, quindi imposta un limite corrispondente davanti al gateway se è esposto a internet.

Gestione corretta del 429

Il backend risponde 429 in tre situazioni e il tuo gateway dovrebbe distinguerle. Un codice di errore rate_limited significa che hai superato le richieste al minuto: 300 su una chiave completa, 60 su una chiave di prova. Un codice concurrency significa troppe richieste contemporaneamente: 8 su una chiave completa, 2 su una chiave di prova. Anche un sovraccarico temporaneo restituisce 429 con Retry-After: 5.

Il semaforo sopra previene il caso di concorrenza in primo luogo, purché il gateway sia l'unico client su quella chiave. Per le richieste al minuto, aggiungi un semplice token bucket o un contatore per minuto. Il codice sopra onora Retry-After su ogni 429, che copre il sovraccarico. Se vedi comunque errori concurrency, un altro processo sta usando la stessa chiave o uno stream vecchio è ancora aperto.

Logga l'intestazione X-Request-Id per ogni chiamata upstream così puoi tracciare una risposta errata, e logga X-Balance-USD per monitorare il tuo credito. Un gateway è un buon punto per inviare alert quando il saldo scende sotto una soglia. La checklist produzione gateway AI copre il resto: segreti, timeout, health check e monitoraggio.

Tracciamento dei costi all'interno del gateway

Poiché ogni risposta include l'uso di token, il gateway può registrare la spesa per token client. Il prezzo è $0,15 per 1M input tokens e $0,50 per 1M output tokens. Per una chiamata che ha usato 1.200 prompt tokens e 350 completion tokens, il costo è 1.200 x $0,15 / 1M più 350 x $0,50 / 1M, che è $0,00018 più $0,000175, o $0,000355.

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

Aggiungilo dopo che la risposta JSON è stata analizzata e memorizzala con il token client. Per gli stream, chiedi al client di includere l'utilizzo nell'ultimo chunk con stream_options. Le richieste fallite o rifiutate non vengono addebitate, quindi conta solo le risposte di successo. I tassi attuali e i bonus di ricarica sono nella pagina prezzi.

Prossimi passi

Ora hai un gateway con auth, retry, un limite di concorrenza e streaming passthrough in circa 80 righe. Estensioni naturali sono budget per token, logging delle richieste al tuo storage e un secondo backend per failover. Mantieni il codice piccolo; una volta che ti servono molti di questi contemporaneamente, un progetto gateway dedicato potrebbe adattarsi meglio.

Per eseguirlo, ti serve una chiave del backend. Iscriviti e richiedi il credito di prova da $0,50 con un clic sulla pagina API key, poi controlla i parametri e i formati di errore nella documentazione API. Ricorda che una chiave di prova permette solo 2 richieste concorrenti, quindi imposta il semaforo a 2 durante i test.

Altre guide

• Gateway LLM vs OpenRouter: quale ti serve?
• Checklist Gateway AI: API LLM senza censura