Python ile LLM Ağ Geçidi: Arka Uç Noktayı Çağırmak

Python'daki ince bir ağ geçidi ne yapar

Kullanışlı bir LLM ağ geçidi Python kurulumu, insanların düşündüğünden daha küçüktür. Tam bir platforma ihtiyacınız yok. İnce bir ağ geçidi, uygulamalarınız ile bir arka uç uç noktası arasında duran ve dört şey ekleyen küçük bir web hizmetidir: kendi istemcileriniz için tek bir paylaşılan kimlik doğrulama belirteci, bir zaman aşımı ve yeniden deneme politikası, yukarı akış çağrıları için bir eşzamanlılık sınırı ve temiz akış aktarımı. Uygulamalarınız, sanki OpenAI ile konuşuyormuş gibi ona bağlanır ve gerçek arka uca yönlendirir.

Bu makale, arka uç olarak https://api.openllmgateway.com/v1 ve model kimliği uncensored kullanılarak bunu FastAPI ve httpx ile oluşturur. Hâlâ bir ağ geçidine gerçekten ihtiyacınız olup olmadığından emin değilseniz, önce bir LLM ağ geçidinin ne olduğunu okuyun. Buradaki amaç koddur.

Kurulum ve akış olmayan yol

Bağımlılıkları yükleyin ve iki ortam değişkeni ayarlayın: arka uç anahtarı ve kendi uygulamalarınıza vereceğiniz virgülle ayrılmış belirteç listesi. Uygulamalarınız arka uç anahtarını asla görmez.

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

Hizmetin ilk yarısı yapılandırma, kimlik doğrulama ve yeniden denemeli düz JSON yolunu kapsar:

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

Ağ geçidinin model değerini uncensored olarak zorladığına dikkat edin. İstemciler başka bir şey isteyemez; bu, bir politikayı tek bir yerde tutmanın basit bir yoludur.

Slotları sızdırmayan akış iletimi

Akış, ince ağ geçitlerinin hata yaptığı yerdir. Tüm üst akış gövdesini okuyup sonra gönderirseniz, kullanıcı sona kadar hiçbir şey görmez. Bir istemci bağlantısı koptuğunda eşzamanlılık yuvasını serbest bırakmayı unutursanız, sonunda kendinizi bloke edersiniz. API, akışlı bir isteği akış bitene kadar eşzamanlılık yuvasını işgal ediyor olarak kabul eder, bu yüzden yerel semafonunuz da aynı davranışı göstermelidir.

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

finally bloğu, akış tamamlandığında, istemci bağlantısı koptuğunda veya bir hata oluştuğunda çalışır, böylece yuva her zaman geri döner. Yeniden denemeler burada bilinçli olarak yoktur. Bir kere istemciye veri gönderildikten sonra temiz bir yeniden başlatma mümkün değildir, bu yüzden yeniden denemeyi ilk parça gönderilmeden önce yapın; bu sürümde yukarıdaki durum kontrolü tek kapıdır.

İstemciden ağ geçidinize çağrı

uvicorn gateway:app --port 8000 ile çalıştırın. Uygulamalarınız artık ağ geçidini temel URL olarak kullanan standart OpenAI istemcisini kullanır ve anahtar olarak kendi tokenlarınızdan birini kullanır:

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)

Diğer yolları aynı şekilde ekleyebilirsiniz. Arka uçta GET /v1/models ve POST /v1/responses mevcuttur ve fonksiyon çağırma ile JSON modu özel işlem gerektirmez çünkü ağ geçidi gövdeyi değişmeden iletir. Maksimum istek gövdesi 8 MB'tır, bu nedenle ağ geçidi internete açıksa önüne eşleşen bir sınır ayarlayın.

429 durumunun doğru şekilde işlenmesi

Arka uç üç durumda 429 yanıtı verir ve ağ geçidinizin bunları ayırt etmesi gerekir. rate_limited hata kodu, dakika başına istek sayısını aştığınızı gösterir: tam anahtar için 300, deneme anahtarı için 60. concurrency kodu, aynı anda çok fazla istek yapıldığını gösterir: tam anahtar için 8, deneme anahtarı için 2. Geçici aşırı yüklenme de Retry-After: 5 ile birlikte 429 yanıtı döndürür.

Yukarıdaki semafor, ağ geçidi o anahtar üzerindeki tek istemci olduğu sürece eşzamanlılık durumunu önler. Dakika başına istekler için basit bir token kovası veya dakika başına sayaç ekleyin. Yukarıdaki kod, her 429 yanıtında Retry-After değerine uyar, bu da aşırı yüklemeyi kapsar. Yine de concurrency hataları görürseniz, başka bir işlem aynı anahtarı kullanıyordur veya eski bir akış hala açıktır.

Her yukarı akış çağrısı için X-Request-Id başlığını günlüğe kaydedin, böylece kötü bir yanıtı izleyebilirsiniz ve kredinizi izlemek için X-Balance-USD değerini günlüğe kaydedin. Bakiye bir eşiğin altına düştüğünde uyarı vermek için bir ağ geçidi iyi bir yerdir. AI ağ geçidi üretim kontrol listesi gizli anahtarları, zaman aşımlarını, sağlık kontrollerini ve izlemeyi kapsar.

Ağ geçidi içinde maliyet takibi

Her yanıt token kullanımını taşıdığı için, ağ geçidi istemci tokenı başına harcamayı kaydedebilir. Fiyat 1M istem tokenı başına $0,15 ve 1M tamamlama tokenı başına $0,50'dir. 1.200 istem tokenı ve 350 tamamlama tokenı kullanan bir çağrı için maliyet, 1.200 x $0,15 / 1M artı 350 x $0,50 / 1M'dir; bu da $0,00018 artı $0,000175, yani $0,000355 eder.

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

Bunu JSON yanıtı ayrıldıktan sonra ekleyin ve istemci belirteci ile kaydedin. Akışlar için istemciden stream_options ile son parçada kullanım bilgisi içermesini isteyin. Başarısız veya reddedilen istekler için ücret alınmaz, bu nedenle yalnızca başarılı yanıtları sayın. Güncel oranlar ve bakiye yükleme bonusları fiyatlama sayfasında yer almaktadır.

Sonraki adımlar

Artık kimlik doğrulama, yeniden denemeler, bir eşzamanlılık sınırı ve akış aktarımı olan yaklaşık 80 satırlık bir ağ geçidiniz var. Doğal uzantılar belirteç başı bütçeler, kendi depolamanıza istek günlüğü ve yedekleme için ikinci bir arka uçtur. Kodu küçük tutun; aynı anda çok sayıda buna ihtiyacınız olduğunda, özel bir ağ geçidi projesi daha iyi uyum sağlayabilir.

Çalıştırmak için bir arka uç anahtarına ihtiyacınız vardır. API anahtarı sayfası üzerinde tek tıkla $0,50 deneme kredisi kazanmak için kaydolun ve API dokümantasyonu içinde parametreleri ve hata formatlarını kontrol edin. Bir deneme anahtarının yalnızca 2 eşzamanlı isteğe izin verdiğini unutmayın, bu yüzden test sırasında semafonu 2'ye ayarlayın.

Daha fazla kılavuz

• LLM Ağ Geçidi vs OpenRouter: Hangisine İhtiyacınız Var?
• AI Geçiş Kontrol Listesi: Sansürsüz LLM API