Python में LLM गेटवे: बैकएंड एंडपॉइंट पर कॉल करना

Python में एक पतला गेटवे क्या करता है

एक उपयोगी llm गेटवे Python सेटअप लोगों की उम्मीद से छोटा होता है। आपको एक पूरा प्लेटफॉर्म नहीं चाहिए। एक पतला गेटवे एक छोटा वेब सर्विस है जो आपके ऐप्स और एक बैकएंड एंडपॉइंट के बीच बैठता है और चार चीजें जोड़ता है: आपके अपने क्लाइंट्स के लिए एक शेयर्ड auth टोकन, एक टाइमआउट और retry पॉलिसी, concurrent upstream कॉल्स पर एक cap, और साफ स्ट्रीमिंग पासथ्रू। आपके ऐप्स इससे ऐसे बात करते हैं जैसे यह OpenAI हो, और यह असली बैकएंड को आगे बढ़ाता है।

यह आलेख FastAPI और httpx का उपयोग करके इसे बनाता है, बैकएंड के रूप में https://api.openllmgateway.com/v1 और मॉडल आईडी uncensored का उपयोग करके। यदि आप अभी भी यह तय कर रहे हैं कि आपको गेटवे की आवश्यकता है या नहीं, तो पहले एक एलएलएम गेटवे क्या है पढ़ें। यहाँ मुख्य बात कोड है।

सेटअप और नॉन-स्ट्रीमिंग राउट

डिपेंडेंसिज़ इंस्टॉल करें और दो एनवायरनमेंट वेरिएबल सेट करें: बैकएंड कुंजी, और एक कॉमा-सेपरेटेड लिस्ट टोकन की, जो आप अपने ऐप्स को देते हैं। आपके ऐप्स कभी बैकएंड कुंजी नहीं देखते।

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

सर्विस का पहला आधा हिस्सा कॉन्फ़िगरेशन, ऑथ और रेट्राइज़ के साथ प्लेन JSON राउट को कवर करता है:

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

ध्यान दें कि गेटवे model को uncensored पर फोर्स करता है। क्लाइंट कुछ और रिक्वेस्ट नहीं कर सकते, जो एक ही जगह पर एक पॉलिसी रखने का एक सरल तरीका है।

स्ट्रीमिंग पासथ्रू जो स्लॉट्स लीक नहीं करता

स्ट्रीमिंग वह जगह है जहाँ थिन गेटवे गलतियाँ करते हैं। यदि आप पूरा अपस्ट्रीम बॉडी पढ़ते हैं और फिर उसे भेजते हैं, तो यूजर अंत तक कुछ नहीं देखता। यदि आप क्लाइंट डिस्कनेक्ट होने पर कन्करेंसी स्लॉट रिलीज़ करना भूल जाते हैं, तो आप अंत में खुद को ब्लॉक कर देते हैं। API एक स्ट्रीम किए गए रिक्वेस्ट को स्ट्रीम खत्म होने तक कन्करेंसी स्लॉट को घेरे हुए मानती है, इसलिए आपके लोकल सेमाफोर को भी वही करना चाहिए।

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 ब्लॉक तब चलता है जब स्ट्रीम पूरी हो जाती है, क्लाइंट डिस्कनेक्ट हो जाता है, या कोई एरर होता है, इसलिए स्लॉट हमेशा वापस आ जाता है। रेट्राइज़ यहाँ जानबूझकर नहीं दिए गए हैं। एक बार बाइट्स क्लाइंट को चले जाने के बाद आप साफ़-सुथरे तरीके से रीस्टार्ट नहीं कर सकते, इसलिए केवल पहले चंक से पहले ही रेट्राई करें, और इस वर्जन में ऊपर दिया गया स्टेटस चेक ही एकमात्र गेट है।

क्लाइंट से अपने गेटवे को कॉल करना

इसे uvicorn gateway:app --port 8000 के साथ चलाएँ। अब आपका एप्लिकेशन मानक OpenAI क्लाइंट का उपयोग करता है, जिसमें गेटवे को बेस URL के रूप में और आपके स्वयं के टोकन को कुंजी के रूप में उपयोग किया जाता है:

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)

आप अन्य राउट्स को उसी तरीके से जोड़ सकते हैं। बैकएंड पर GET /v1/models और POST /v1/responses उपलब्ध हैं, और फ़ंक्शन कॉलिंग व JSON मोड को विशेष हैंडलिंग की आवश्यकता नहीं है क्योंकि गेटवे बॉडी को बिना किसी बदलाव के आगे बढ़ाता है। अधिकतम अनुरोध बॉडी 8 MB है, इसलिए यदि यह इंटरनेट के सामने है तो गेटवे के सामने एक मेल खाने वाली सीमा सेट करें।

429 को सही तरीके से हैंडल करना

बैकएंड तीन स्थितियों में 429 उत्तर देता है, और आपके गेटवे को उन्हें अलग-अलग पहचानना चाहिए। एक rate_limited एरर कोड का मतलब है कि आपने प्रति मिनट अनुरोधों की सीमा पार कर दी है: पूरी कुंजी पर 300, ट्रायल कुंजी पर 60। एक concurrency कोड का मतलब है एक साथ बहुत सारे अनुरोध: पूरी कुंजी पर 8, ट्रायल कुंजी पर 2। अस्थायी ओवरलोड भी Retry-After: 5 के साथ 429 लौटाता है।

ऊपर दिया गया सेमाफोर कन्करेंसी केस को पहले ही रोकता है, जब तक कि गेटवे उस कुंजी पर एकमात्र क्लाइंट है। प्रति मिनट रिक्वेस्ट्स के लिए, एक साधारण टोकन बकेट या प्रति मिनट काउंटर जोड़ें। ऊपर दिया गया कोड हर 429 पर Retry-After का सम्मान करता है, जो ओवरलोड को कवर करता है। यदि आपको फिर भी concurrency एरर्स दिखते हैं, तो कोई दूसरा प्रोसेस उसी कुंजी का उपयोग कर रहा है, या एक पुरानी स्ट्रीम अभी भी ओपन है।

प्रत्येक अपस्ट्रीम कॉल के लिए X-Request-Id हेडर को लॉग करें ताकि आप खराब प्रतिक्रिया का पता लगा सकें, और अपने क्रेडिट की निगरानी के लिए X-Balance-USD को लॉग करें। बैलेंस एक थ्रेशोल्ड से नीचे गिरने पर अलर्ट करने के लिए गेटवे एक अच्छी जगह है। AI गेटवे प्रोडक्शन चेकलिस्ट शेष कवर करती है: सीक्रेट, टाइमआउट, हेल्थ चेक और मॉनिटरिंग।

गेटवे के अंदर कॉस्ट ट्रैकिंग

चूंकि प्रत्येक प्रतिक्रिया टोकन उपयोग लाती है, गेटवे क्लाइंट टोकन प्रति खर्च रिकॉर्ड कर सकता है। कीमत $0.15 प्रति 1M इनपुट टोकन और $0.50 प्रति 1M आउटपुट टोकन है। एक कॉल के लिए जिसने 1,200 प्रॉम्प्ट टोकन और 350 कंप्लीशन टोकन का उपयोग किया, लागत 1,200 x $0.15 / 1M प्लस 350 x $0.50 / 1M है, जो $0.00018 प्लस $0.000175 है, या $0.000355 है।

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

इसे JSON प्रतिक्रिया के पार्स करने के बाद जोड़ें और इसे क्लाइंट टोकन के साथ स्टोर करें। स्ट्रीम के लिए, stream_options के साथ क्लाइंट से अंतिम चंक में उपयोग शामिल करने के लिए कहें। विफल या अस्वीकृत अनुरोधों के लिए चार्ज नहीं किया जाता है, इसलिए केवल सफल प्रतिक्रियाओं को ही गिनें। वर्तमान दरें और टॉप-अप बोनस मूल्य निर्धारण पृष्ठ पर उपलब्ध हैं।

अगले कदम

अब आपके पास लगभग 80 लाइनों में ऑथ, रेट्रिज़, कंकरेंसी कैप और स्ट्रीमिंग पासथ्रू वाला गेटवे है। तार्किक विस्तारों में प्रति-टोकन बजट, अपने स्वयं के स्टोरेज में अनुरोध लॉगिंग, और फेलओवर के लिए एक दूसरा बैकएंड शामिल है। कोड को छोटा रखें; एक बार जब आपको एक साथ कई चीजों की आवश्यकता होती है, तो एक समर्पित गेटवे प्रोजेक्ट बेहतर फिट हो सकता है।

इसे चलाने के लिए आपको एक बैकएंड कुंजी की आवश्यकता है। साइन अप करें और API कुंजी पेज पर एक क्लिक के साथ $0.50 का ट्रायल क्रेडिट क्लेम करें, फिर API दस्तावेज़ में पैरामीटर और त्रुटि प्रारूप जांचें। याद रखें कि एक ट्रायल कुंजी केवल 2 समानांतर अनुरोधों की अनुमति देती है, इसलिए परीक्षण के दौरान सेमाफोर को 2 पर सेट करें।

अधिक गाइड

• LLM गेटवे बनाम OpenRouter: आपको किसकी आवश्यकता है?
• AI गेटवे चेकलिस्ट: बिना सेंसर LLM API