Gateway LLM en Python: Llamada a un endpoint de backend

Lo que hace un gateway ligero en Python

Una configuración útil de gateway LLM en Python es más pequeña de lo que la gente espera. No necesitas una plataforma completa. Un gateway ligero es un pequeño servicio web que se sitúa entre tus aplicaciones y un endpoint de backend y añade cuatro cosas: un token de autenticación compartido para tus propios clientes, una política de tiempo de espera y reintentos, un límite de llamadas upstream concurrentes y un passthrough de streaming limpio. Tus aplicaciones hablan con él como si fuera OpenAI, y él reenvía al backend real.

Este artículo construye eso con FastAPI y httpx, usando https://api.openllmgateway.com/v1 como backend y el model ID uncensored. Si aún estás decidiendo si necesitas un gateway, lee qué es un gateway LLM primero. El punto aquí es el código.

Configuración y la ruta sin streaming

Instala las dependencias y establece dos variables de entorno: la clave del backend y una lista separada por comas de tokens que entregas a tus propias aplicaciones. Tus aplicaciones nunca ven la clave del backend.

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

La primera mitad del servicio cubre la configuración, la autenticación y la ruta JSON simple con reintentos:

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

Observa que el gateway fuerza model a uncensored. Los clientes no pueden solicitar nada más, lo cual es una forma sencilla de mantener una política en un solo lugar.

Passthrough de streaming que no pierde slots

El streaming es donde los gateways ligeros fallan. Si lees todo el cuerpo upstream y luego lo envías, el usuario no ve nada hasta el final. Si olvidas liberar el slot de concurrencia cuando un cliente se desconecta, eventualmente te bloqueas a ti mismo. La API trata una petición en streaming como si ocupara su slot de concurrencia hasta que termine el stream, así que tu semáforo local debería hacer lo mismo.

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

El bloque finally se ejecuta cuando el stream se completa, cuando el cliente se desconecta o cuando ocurre un error, así que el slot siempre vuelve. Los reintentos están deliberadamente ausentes aquí. Una vez que los bytes han ido al cliente, no puedes reiniciar limpiamente, así que reintenta solo antes del primer chunk, y en esta versión la verificación de estado anterior es la única puerta.

Llamando a tu gateway desde un cliente

Ejecútalo con uvicorn gateway:app --port 8000. Tus aplicaciones ahora usan el cliente estándar de OpenAI con el gateway como base URL y uno de tus propios tokens como clave:

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)

Puedes añadir las otras rutas de la misma manera. GET /v1/models y POST /v1/responses están disponibles en el backend, y el tool calling y el modo JSON no necesitan manejo especial porque el gateway reenvía el cuerpo intacto. El cuerpo de la petición máximo es de 8 MB, así que establece un límite coincidente delante del gateway si está expuesto a internet.

Manejo correcto de 429

El backend responde 429 en tres situaciones, y tu gateway debería distinguirlos. Un código de error rate_limited significa que excediste las peticiones por minuto: 300 en una clave completa, 60 en una clave de prueba. Un código concurrency significa demasiadas peticiones a la vez: 8 en una clave completa, 2 en una clave de prueba. La sobrecarga temporal también devuelve 429 con Retry-After: 5.

El semáforo anterior previene el caso de concurrencia de antemano, siempre que el gateway sea el único cliente en esa clave. Para las peticiones por minuto, añade un bucket de tokens simple o un contador por minuto. El código anterior respeta Retry-After en cada 429, lo cual cubre la sobrecarga. Si ves errores concurrency de todos modos, otro proceso está usando la misma clave, o un stream antiguo aún está abierto.

Registra el encabezado X-Request-Id para cada llamada upstream para que puedas rastrear una respuesta defectuosa, y registra X-Balance-USD para vigilar tu crédito. Un gateway es un buen lugar para alertar cuando el saldo baja de un umbral. La lista de verificación de producción de gateways AI cubre el resto: secretos, tiempos de espera, comprobaciones de salud y monitoreo.

Seguimiento de costos dentro del gateway

Debido a que cada respuesta lleva el uso de tokens, el gateway puede registrar el gasto por token de cliente. El precio es $0,15 por 1M tokens de entrada y $0,50 por 1M tokens de salida. Para una llamada que usó 1.200 tokens de prompt y 350 tokens de completion, el costo es 1.200 x $0,15 / 1M más 350 x $0,50 / 1M, que es $0,00018 más $0,000175, o $0,000355.

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

Añádelo después de que se analice la respuesta JSON y guárdalo con el token de cliente. Para streams, pide al cliente que incluya usage en el último chunk con stream_options. Las peticiones fallidas o rechazadas no se cobran, así que cuenta solo las respuestas exitosas. Las tarifas actuales y los bonos de recarga están en la página de precios.

Próximos pasos

Ahora tienes un gateway con autenticación, reintentos, un límite de concurrencia y passthrough de streaming en unas 80 líneas. Las extensiones naturales son presupuestos por token, registro de peticiones en tu propio almacenamiento y un segundo backend para failover. Mantén el código pequeño; una vez que necesites muchos de esos a la vez, un proyecto de gateway dedicado puede encajar mejor.

Para ejecutarlo, necesitas una clave de backend. Regístrate y reclama el crédito de prueba de $0,50 con un clic en la página de claves de API, luego verifica los parámetros y los formatos de error en la documentación de API. Recuerda que una clave de prueba permite solo 2 peticiones concurrentes, así que establece el semáforo en 2 mientras pruebas.

Más guías

• Gateway LLM vs OpenRouter: ¿Cuál necesitas?
• Lista de verificación de AI Gateway: API de LLM sin censura