Gerbang LLM di Python: Memanggil Endpoint Backend

Apa yang dilakukan gerbang tipis di Python

Pengaturan gerbang LLM Python yang berguna lebih kecil dari yang diharapkan orang. Anda tidak memerlukan platform lengkap. Gerbang tipis adalah layanan web kecil yang berada di antara aplikasi Anda dan endpoint backend serta menambahkan empat hal: satu token otentikasi bersama untuk klien Anda sendiri, kebijakan batas waktu dan coba ulang, batas panggilan upstream konkuren, dan penerusan streaming yang bersih. Aplikasi Anda berbicara dengannya seolah-olah itu adalah OpenAI, dan meneruskannya ke backend asli.

Artikel ini membangunnya dengan FastAPI dan httpx, menggunakan https://api.openllmgateway.com/v1 sebagai backend dan ID model uncensored. Jika Anda masih mempertimbangkan apakah Anda memerlukan gateway sama sekali, baca apa itu gateway LLM terlebih dahulu. Fokus di sini adalah kodenya.

Pengaturan dan rute non-streaming

Instal dependensi dan atur dua variabel lingkungan: kunci backend, dan daftar token yang dipisahkan koma yang Anda berikan ke aplikasi Anda sendiri. Aplikasi Anda tidak pernah melihat kunci backend.

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

Setengah pertama layanan mencakup konfigurasi, otentikasi, dan rute JSON biasa dengan percobaan ulang:

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

Perhatikan bahwa gerbang memaksa model menjadi uncensored. Klien tidak dapat meminta hal lain, yang merupakan cara sederhana untuk menjaga satu kebijakan di satu tempat.

Streaming passthrough yang tidak membocorkan slot

Streaming adalah tempat gerbang tipis melakukan kesalahan. Jika Anda membaca seluruh badan upstream dan kemudian mengirimkannya, pengguna tidak melihat apa pun hingga akhir. Jika Anda lupa melepaskan slot konkurensi ketika klien terputus, Anda akhirnya memblokir diri sendiri. API memperlakukan permintaan yang dialirkan sebagai menempati slot konkurensinya hingga aliran berakhir, sehingga semafor lokal Anda harus melakukan hal yang sama.

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 dijalankan ketika stream selesai, ketika klien terputus, atau ketika terjadi kesalahan, sehingga slot selalu kembali. Coba ulang sengaja dihilangkan di sini. Setelah byte dikirim ke klien, Anda tidak dapat memulai ulang dengan bersih, sehingga coba ulang hanya dilakukan sebelum chunk pertama, dan dalam versi ini pemeriksaan status di atas adalah satu-satunya gerbang.

Memanggil gerbang Anda dari klien

Jalankan dengan uvicorn gateway:app --port 8000. Aplikasi Anda sekarang menggunakan klien OpenAI standar dengan gerbang sebagai endpoint dan salah satu token Anda sendiri sebagai kunci:

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)

Anda dapat menambahkan rute lainnya dengan cara yang sama. GET /v1/models dan POST /v1/responses tersedia di backend, dan pemanggilan fungsi serta mode JSON tidak memerlukan penanganan khusus karena gerbang meneruskan badan permintaan tanpa perubahan. Badan permintaan maksimum adalah 8 MB, sehingga atur batas yang sesuai di depan gerbang jika menghadap internet.

Menangani 429 dengan benar

Backend menjawab 429 dalam tiga situasi, dan gerbang Anda harus membedakan mereka. Kode kesalahan rate_limited berarti Anda melebihi permintaan per menit: 300 pada kunci penuh, 60 pada kunci uji coba. Kode concurrency berarti terlalu banyak permintaan sekaligus: 8 pada kunci penuh, 2 pada kunci uji coba. Beban berlebih sementara juga mengembalikan 429 dengan Retry-After: 5.

Semaphore di atas mencegah kasus konkurensi sejak awal, selama gateway menjadi satu-satunya klien pada kunci tersebut. Untuk permintaan per menit, tambahkan token bucket sederhana atau penghitung per menit. Kode di atas mematuhi Retry-After pada setiap 429, yang mencakup kelebihan beban. Jika Anda masih melihat error concurrency, proses lain menggunakan kunci yang sama, atau stream lama masih terbuka.

Log header X-Request-Id untuk setiap panggilan upstream agar Anda dapat melacak respons yang buruk, dan log X-Balance-USD untuk memantau saldo Anda. Gerbang adalah tempat yang baik untuk memberi peringatan ketika saldo turun di bawah ambang batas. Daftar periksa produksi gerbang AI mencakup sisanya: rahasia, waktu tunggu, pemeriksaan kesehatan, dan pemantauan.

Pelacakan biaya di dalam gerbang

Karena setiap respons membawa penggunaan token, gerbang dapat mencatat pengeluaran per token klien. Harga adalah $0,15 per 1M token input dan $0,50 per 1M token output. Untuk panggilan yang menggunakan 1.200 token prompt dan 350 token penyelesaian, biayanya adalah 1.200 x $0,15 / 1M ditambah 350 x $0,50 / 1M, yaitu $0,00018 ditambah $0,000175, atau $0,000355.

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

Tambahkan setelah respons JSON diparsing dan simpan dengan token klien. Untuk streaming, minta klien untuk menyertakan penggunaan dalam chunk terakhir dengan stream_options. Permintaan yang gagal atau ditolak tidak dikenakan biaya, jadi hanya hitung respons yang berhasil. Tarif saat ini dan bonus isi saldo ada di halaman harga.

Langkah selanjutnya

Anda sekarang memiliki gerbang dengan autentikasi, coba ulang, batas konkurensi, dan terusan streaming dalam sekitar 80 baris. Ekstensi alami adalah anggaran per token, log permintaan ke penyimpanan Anda sendiri, dan backend kedua untuk failover. Jaga kode tetap kecil; begitu Anda membutuhkan banyak hal itu sekaligus, proyek gateway khusus mungkin lebih cocok.

Untuk menjalankannya, Anda memerlukan kunci backend. Daftar dan klaim kredit uji coba $0,50 dengan satu klik di halaman kunci API, lalu periksa parameter dan format kesalahan di dokumentasi API. Ingat bahwa kunci uji coba hanya memungkinkan 2 permintaan konkuren, sehingga atur semafor menjadi 2 saat pengujian.

Panduan lainnya

• Perbandingan Gerbang LLM vs OpenRouter: Mana yang Anda Butuhkan?
• Daftar Periksa Gateway AI: LLM API Tanpa Sensor