Brama LLM w Pythonie: Wywoływanie endpointu backendu

Co robi cienka brama w Pythonie

Użyteczna konfiguracja bramy LLM w Pythonie jest mniejsza, niż ludzie się spodziewają. Nie potrzebujesz pełnej platformy. Cienka brama to mała usługa sieciowa stojąca między Twoimi aplikacjami a endpointem backendu, dodająca cztery rzeczy: jeden wspólny token autoryzacji dla Twoich klientów, politykę limitu czasu i ponawiania prób, limit równoległych wywołań upstream oraz czyste przesyłanie strumieniowe. Twoje aplikacje rozmawiają z nią tak, jakby była OpenAI, a ona przekierowuje do prawdziwego backendu.

Ten artykuł buduje to z FastAPI i httpx, używając https://api.openllmgateway.com/v1 jako backendu i ID modelu uncensored. Jeśli wciąż zastanawiasz się, czy potrzebujesz bramy, przeczytaj najpierw co to jest brama LLM. Chodzi tu o kod.

Konfiguracja i trasa bez strumieniowania

Zainstaluj zależności i ustaw dwie zmienne środowiskowe: klucz backendu oraz listę tokenów, oddzielonych przecinkami, które przekazujesz Twoim aplikacjom. Twoje aplikacje nigdy nie widzą klucza backendu.

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

Pierwsza połowa usługi obejmuje konfigurację, autoryzację i zwykłą trasę JSON z ponawianiem prób:

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

Zauważ, że brama wymusza model na uncensored. Klienci nie mogą żądać niczego innego, co jest prostym sposobem na utrzymanie jednej polityki w jednym miejscu.

Przesyłanie strumieniowe, które nie wycieka slotami

Strumieniowanie to miejsce, w którym cienkie bramy zawodzą. Jeśli przeczytasz całe ciało odpowiedzi z upstreamu i wyślesz je na raz, użytkownik niczego nie zobaczy do końca. Jeśli zapomnisz zwolnić slotu współbieżności po rozłączeniu się klienta, w końcu zablokujesz się sam. API traktuje zapytanie strumieniowane jako zajmujące slot współbieżności do momentu zakończenia strumienia, więc Twój lokalny semafor powinien robić to samo.

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

Blok finally wykonuje się, gdy strumień się zakończy, klient się rozłączy lub wystąpi błąd, więc slot zawsze wraca. Powtórzenia są tu celowo pominięte. Gdy bajty trafią do klienta, nie można zrestartować czysto, więc powtarzaj tylko przed pierwszym chunkiem, a w tej wersji powyższa kontrola statusu jest jedynym filtrem.

Wywoływanie Twojej bramy przez klienta

Uruchom go poleceniem uvicorn gateway:app --port 8000. Twoje aplikacje używają teraz standardowego klienta OpenAI z bramką jako bazowym URL i jednym ze swoich tokenów jako kluczem:

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)

Możesz dodać pozostałe routy w ten sam sposób. GET /v1/models i POST /v1/responses są dostępne po stronie backendu, a wywoływanie funkcji i tryb JSON nie wymagają specjalnej obsługi, ponieważ bramka przekazuje ciało zapytania bez zmian. Maksymalny rozmiar ciała zapytania to 8 MB, więc ustaw odpowiedni limit przed bramką, jeśli jest ona wystawiona na internet.

Poprawne obsłużenie 429

Backend odpowiada 429 w trzech sytuacjach, a Twoja brama powinna je rozróżniać. Kod błędu rate_limited oznacza przekroczenie zapytań na minutę: 300 dla pełnego klucza, 60 dla klucza próbnego. Kod concurrency oznacza zbyt wiele żądań naraz: 8 dla pełnego klucza, 2 dla klucza próbnego. Tymczasowe przeciążenie również zwraca 429 z Retry-After: 5.

Powyższy semafor zapobiega przypadkowi współbieżności z góry, o ile brama jest jedynym klientem na tym kluczu. Dla żądań na minutę dodaj prosty kubelek tokenów lub licznik na minutę. Powyższy kod szanuje Retry-After przy każdym 429, co pokrywa przeciążenie. Jeśli widzisz błędy concurrency mimo to, inny proces używa tego samego klucza lub stare strumienie są wciąż otwarte.

Loguj nagłówek X-Request-Id dla każdego wywołania upstreamu, abyś mógł prześledzić błędną odpowiedź, oraz loguj X-Balance-USD, aby kontrolować swój kredyt. Bramka to dobre miejsce na alert, gdy saldo spadnie poniżej progu. Checklista produkcyjna bramki AI obejmuje resztę: sekrety, limity czasu, kontrolę stanu i monitorowanie.

Śledzenie kosztów wewnątrz bramy

Ponieważ każda odpowiedź zawiera użycie tokenów, brama może rejestrować wydatki na token klienta. Cena to $0,15 za 1M tokenów wejściowych i $0,50 za 1M tokenów wyjściowych. Dla wywołania, które użyło 1 200 tokenów prompt i 350 tokenów completion, koszt to 1 200 x $0,15 / 1M plus 350 x $0,50 / 1M, co daje $0,00018 plus $0,000175, czyli $0,000355.

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

Dodaj go po przeanalizowaniu odpowiedzi JSON i zapisz go z tokenem klienta. W przypadku strumieni poproś klienta o dołączenie zużycia w ostatnim chunku z stream_options. Zapytania nieudane lub odrzucone nie są rozliczane, więc zliczaj tylko odpowiedzi pomyślne. Aktualne stawki i bonusy za doładowanie znajdziesz na stronie z cenami.

Kolejne kroki

Masz teraz bramę z autoryzacją, ponawianiem prób, limitem współbieżności i przesyłaniem strumieniowym w około 80 liniach. Naturalne rozszerzenia to budżety na token, logowanie zapytań do własnego magazynu i drugi backend na awarię. Trzymaj kod mały; gdy potrzebujesz wielu takich naraz, dedykowany projekt bramy może pasować lepiej.

Aby uruchomić, potrzebujesz klucza backendowego. Zarejestruj się i odbierz 0,50 $ kredytu próbnego jednym kliknięciem na stronie klucza API, a następnie sprawdź parametry i formaty błędów w dokumentacji API. Pamiętaj, że klucz próbny pozwala tylko na 2 równoległe zapytania, więc ustaw semafor na 2 podczas testowania.

Więcej przewodników

• Brama LLM vs OpenRouter: Czego potrzebujesz?
• Lista kontrolna AI Gateway: API LLM bez cenzury