Gateway LLM em Python: Chamando um Endpoint de Backend

O que um gateway fino em Python faz

Uma configuração útil de gateway LLM em Python é menor do que as pessoas esperam. Você não precisa de uma plataforma completa. Um gateway fino é um pequeno serviço web que fica entre seus aplicativos e um endpoint de backend e adiciona quatro coisas: um token de autenticação compartilhado para seus próprios clientes, uma política de tempo limite e retentativa, um limite de chamadas a servidores upstream simultâneas e passagem de streaming limpa. Seus aplicativos falam com ele como se fosse OpenAI, e ele encaminha para o backend real.

Este artigo constrói isso com FastAPI e httpx, usando https://api.openllmgateway.com/v1 como backend e o ID do modelo uncensored. Se você ainda está decidindo se precisa de um gateway, leia o que é um gateway LLM primeiro. O foco aqui é o código.

Configuração e a rota sem streaming

Instale as dependências e defina duas variáveis de ambiente: a chave do backend e uma lista separada por vírgulas de tokens que você distribui para seus próprios aplicativos. Seus aplicativos nunca veem a chave do backend.

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

A primeira metade do serviço cobre configuração, autenticação e a rota JSON simples com 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")

Observe que o gateway força model para uncensored. Clientes não podem solicitar nada além disso, o que é uma maneira simples de manter uma política em um só lugar.

Streaming passthrough que não vaza slots

O streaming é onde gateways simples falham. Se você ler o corpo inteiro do upstream e só então enviá-lo, o usuário não vê nada até o fim. Se você esquecer de liberar o slot de concorrência quando um cliente se desconecta, você eventualmente se bloqueia. A API trata uma requisição em streaming como ocupando seu slot de concorrência até o fim do streaming, então seu semáforo local deve fazer o mesmo.

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

O bloco finally é executado quando o streaming termina, quando o cliente se desconecta ou quando ocorre um erro, então o slot sempre é liberado. Retentativas estão deliberadamente ausentes aqui. Uma vez que bytes foram enviados ao cliente, você não pode reiniciar de forma limpa, então a retentativa só deve ocorrer antes do primeiro chunk, e nesta versão a verificação de status acima é a única barreira.

Chamando seu gateway de um cliente

Execute-o com uvicorn gateway:app --port 8000. Seus aplicativos agora usam o cliente OpenAI padrão com o gateway como base URL e um dos seus próprios tokens como chave:

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)

Você pode adicionar as outras rotas da mesma forma. GET /v1/models e POST /v1/responses estão disponíveis no backend, e chamada de funções e modo JSON não precisam de tratamento especial porque o gateway encaminha o corpo intacto. O corpo máximo da requisição é 8 MB, então defina um limite correspondente na frente do gateway se ele estiver exposto à internet.

Tratando 429 corretamente

O backend responde 429 em três situações, e seu gateway deve diferenciá-las. Um código de erro rate_limited significa que você excedeu requisições por minuto: 300 em uma chave completa, 60 em uma chave de teste. Um código concurrency significa muitas requisições ao mesmo tempo: 8 em uma chave completa, 2 em uma chave de teste. Sobrecarga temporária também retorna 429 com Retry-After: 5.

O semáforo acima evita o caso de concorrência desde o início, desde que o gateway seja o único cliente usando essa chave. Para requisições por minuto, adicione um bucket de tokens simples ou um contador por minuto. O código acima respeita o cabeçalho Retry-After em cada 429, o que cobre sobrecarga. Se você ver erros de concurrency mesmo assim, outro processo está usando a mesma chave, ou um streaming antigo ainda está aberto.

Registre o cabeçalho X-Request-Id para cada chamada ao upstream para rastrear uma resposta ruim, e registre X-Balance-USD para acompanhar seu crédito. Um gateway é um bom lugar para alertar quando o saldo cai abaixo de um limite. A lista de verificação de produção de gateways AI cobre o resto: segredos, timeouts, verificações de saúde e monitoramento.

Rastreamento de custos dentro do gateway

Como toda resposta traz o uso de tokens, o gateway pode registrar o gasto por token de cliente. O preço é $0,15 por 1M de tokens de entrada e $0,50 por 1M de tokens de saída. Para uma chamada que usou 1.200 tokens de prompt e 350 tokens de conclusão, o custo é 1.200 x $0,15 / 1M mais 350 x $0,50 / 1M, que é $0,00018 mais $0,000175, ou $0,000355.

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

Adicione-o após a análise da resposta JSON e armazene-o com o token de cliente. Para streams, peça ao cliente para incluir o uso no último chunk com stream_options. Requisições falhas ou recusadas não são cobradas, então conte apenas as respostas bem-sucedidas. As taxas atuais e bônus de recarga estão na página de preços.

Próximos passos

Você agora tem um gateway com autenticação, retentativas, limite de concorrência e streaming em cerca de 80 linhas. Extensões naturais são orçamentos por token, registro de requisições no seu próprio armazenamento e um segundo backend para failover. Mantenha o código pequeno; quando você precisar de muitos desses ao mesmo tempo, um projeto de gateway dedicado pode ser mais adequado.

Para executá-lo, você precisa de uma chave de backend. Cadastre-se e resgate o crédito de teste de $0,50 com um clique na página de chaves de API, depois verifique parâmetros e formatos de erro na documentação da API. Lembre-se de que uma chave de teste permite apenas 2 requisições simultâneas, então defina o semáforo para 2 durante os testes.

Mais guias

• LLM Gateway vs OpenRouter: Qual Você Precisa?
• Lista de verificação de Gateway de IA: API LLM sem censura