Passerelle LLM en Python : Appel à un endpoint backend

Ce que fait une passerelle légère en Python

Une configuration de passerelle LLM Python utile est plus petite que ce que les gens imaginent. Vous n'avez pas besoin d'une plateforme complète. Une passerelle légère est un petit service web qui se place entre vos applications et un point de terminaison backend et ajoute quatre éléments : un jeton d'authentification partagé pour vos propres clients, une politique de délai d'attente et de reconnexion, une limite sur les appels amont simultanés et un transfert de flux en mode pass-through propre. Vos applications s'y adressent comme s'il s'agissait d'OpenAI, et elle transmet vers le backend réel.

Cet article construit cela avec FastAPI et httpx, en utilisant https://api.openllmgateway.com/v1 comme backend et l'ID de modèle uncensored. Si vous hésitez encore sur le besoin d'une passerelle, lisez d'abord ce qu'est une passerelle LLM. Ici, l'objectif est le code.

Configuration et le chemin non streaming

Installez les dépendances et définissez deux variables d'environnement : la clé du backend, et une liste séparée par des virgules de jetons que vous distribuez à vos propres applications. Vos applications ne voient jamais la clé du backend.

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

La première moitié du service couvre la configuration, l'authentification et le chemin JSON simple avec tentatives :

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

Notez que la passerelle force model vers uncensored. Les clients ne peuvent rien demander d'autre, ce qui est un moyen simple de maintenir une politique en un seul endroit.

Streaming pass-through sans fuite de slots

Le streaming est là où les passerelles légères font des erreurs. Si vous lisez tout le corps amont avant de l'envoyer, l'utilisateur ne voit rien jusqu'à la fin. Si vous oubliez de libérer le slot de concurrence quand un client se déconnecte, vous finissez par vous bloquer. L'API traite une requête streamée comme occupant son slot de concurrence jusqu'à la fin du flux, donc votre sémaphore local doit faire de même.

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

Le bloc finally s'exécute quand le flux se termine, quand le client se déconnecte ou quand une erreur survient, donc le slot revient toujours. Les tentatives sont délibérément absentes ici. Une fois des octets envoyés au client, vous ne pouvez pas redémarrer proprement, donc ne réessayez qu'avant le premier chunk, et dans cette version, la vérification de statut ci-dessus est le seul point de contrôle.

Appel de votre passerelle depuis un client

Lancez-le avec uvicorn gateway:app --port 8000. Vos applications utilisent désormais le client OpenAI standard avec la passerelle comme base URL et un de vos propres jetons comme clé :

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)

Vous pouvez ajouter les autres routes de la même manière. GET /v1/models et POST /v1/responses sont disponibles sur le backend, et l'appel d'outils et le mode JSON ne nécessitent pas de traitement spécial car la passerelle transmet le corps de la requête tel quel. La taille maximale du corps de requête est de 8 Mo, donc définissez une limite correspondante devant la passerelle si elle est exposée à Internet.

Gestion correcte des 429

Le backend répond 429 dans trois situations, et votre passerelle devrait les distinguer. Un code d'erreur rate_limited signifie que vous avez dépassé les requêtes par minute : 300 sur une clé complète, 60 sur une clé d'essai. Un code concurrency signifie trop de requêtes à la fois : 8 sur une clé complète, 2 sur une clé d'essai. Une surcharge temporaire renvoie aussi 429 avec Retry-After: 5.

Le sémaphore ci-dessus empêche le cas de concurrence dès le départ, tant que la passerelle est le seul client sur cette clé. Pour les requêtes par minute, ajoutez un simple seau à jetons ou un compteur par minute. Le code ci-dessus honore Retry-After sur chaque 429, ce qui couvre la surcharge. Si vous voyez quand même des erreurs concurrency, un autre processus utilise la même clé, ou un ancien flux est encore ouvert.

Journalisez l'en-tête X-Request-Id pour chaque appel amont afin de tracer une mauvaise réponse, et journalisez X-Balance-USD pour surveiller votre crédit. Une passerelle est un bon endroit pour alerter quand le solde tombe en dessous d'un seuil. La checklist de production de passerelle IA couvre le reste : secrets, délais d'attente, vérifications de santé et surveillance.

Suivi des coûts dans la passerelle

Comme chaque réponse contient l'utilisation de tokens, la passerelle peut enregistrer la dépense par jeton client. Le prix est de $0,15 par 1M de tokens d'entrée et $0,50 par 1M de tokens de sortie. Pour un appel ayant utilisé 1 200 tokens de prompt et 350 tokens de complétion, le coût est 1 200 x $0,15 / 1M plus 350 x $0,50 / 1M, soit $0,00018 plus $0,000175, ou $0,000355.

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

Ajoutez-le après l'analyse de la réponse JSON et stockez-le avec le jeton client. Pour les flux, demandez au client d'inclure l'utilisation dans le dernier chunk avec stream_options. Les requêtes échouées ou refusées ne sont pas facturées, donc ne comptez que les réponses réussies. Les tarifs actuels et les bonus de rechargement sont sur la page de tarification.

Étapes suivantes

Vous avez maintenant une passerelle avec auth, tentatives, une limite de concurrence et un streaming pass-through en environ 80 lignes. Des extensions naturelles sont les budgets par jeton, la journalisation des requêtes vers votre propre stockage, et un deuxième backend pour la bascule. Gardez le code petit ; une fois que vous en avez besoin de beaucoup en même temps, un projet de passerelle dédié peut mieux convenir.

Pour le lancer, vous avez besoin d'une clé backend. Inscrivez-vous et réclamez le crédit d'essai de $0,50 en un clic sur la page des clés API, puis vérifiez les paramètres et les formats d'erreur dans la documentation API. N'oubliez pas qu'une clé d'essai permet seulement 2 requêtes concurrentes, donc réglez le sémaphore à 2 pendant les tests.

Plus de guides

• Passerelle LLM vs OpenRouter : Lequel vous faut-il ?
• Checklist AI Gateway : API LLM sans censure