بوابة LLM في Python: استدعاء نقطة نهاية الخلفية

ما تفعله البوابة الرقيقة في Python

إعداد بوابة LLM في Python المفيد أصغر مما يتوقعه الناس. لا تحتاج إلى منصة كاملة. البوابة الرقيقة هي خدمة ويب صغيرة تقع بين تطبيقاتك ونقطة نهاية الخلفية وتضيف أربعة أشياء: رمز مصادقة مشترك واحد لعملائك، وسياسة مهلة زمنية وإعادة محاولة، حدّ للطلبات المتزامنة، وتمرير البث المتدفق النظيف. تتحدث تطبيقاتك معها كما لو كانت OpenAI، وهي تحوّل الطلبات إلى الخلفية الحقيقية.

يبني هذا المقال ذلك باستخدام FastAPI و httpx، باستخدام https://api.openllmgateway.com/v1 كخلفية ومعرف النموذج uncensored. إذا كنت لا تزال تقرر ما إذا كنت بحاجة إلى بوابة على الإطلاق، اقرأ ما هي بوابة LLM أولاً. النقطة هنا هي الكود.

الإعداد وطريقة غير البث المتدفق

ثبّت التبعيات واضبط متغيري بيئة: مفتاح الخادم الخلفي، وقائمة مفصولة بفواصل من الرموز التي تمنحها لتطبيقاتك. لا ترى تطبيقاتك مفتاح الخادم الخلفي أبدًا.

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. لا يمكن للعملاء طلب أي شيء آخر، وهي طريقة بسيطة للحفاظ على سياسة واحدة في مكان واحد.

تحويل البث المتدفق دون تسريب الشقوق

البث المتدفق هو حيث تفشل البوابات الخفيفة. إذا قرأت جسم الـ upstream بالكامل ثم أرسلته، لن يرى المستخدم شيئًا حتى النهاية. إذا نسيت تحرير شق التزامن عند قطع العميل للاتصال، فإنك ستحجب نفسك في النهاية. يعالج الـ API الطلب المتدفق على أنه يشغل شق التزامن الخاص به حتى ينتهي البث، لذا يجب أن يفعل الـ semaphore المحلي لديك الأمر نفسه.

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 عند اكتمال البث، أو عند قطع الاتصال من قِبل العميل، أو عند حدوث خطأ، مما يضمن عودة الشقّ دائماً. عمليات إعادة المحاولة غائبة هنا عمدًا. بمجرد إرسال البيانات إلى العميل، لا يمكن إعادة التشغيل بنظافة، لذا تُنفَّذ إعادة المحاولة فقط قبل إرسال الجزء (chunk) الأول، وفي هذا الإصدار، يُعدّ فحص الحالة أعلاه هو البوابة الوحيدة.

استدعاء البوابة من العميل

شغّله باستخدام 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 على مفتاح تجريبي. يعيد الحمل الزائد المؤقت أيضًا 429 مع Retry-After: 5.

يمنع الـ semaphore أعلاه حالة التزامن من الأساس، طالما أن البوابة هي العميل الوحيد على هذا المفتاح. بالنسبة للطلبات في الدقيقة، أضف دلو رمز بسيطًا أو عدادًا لكل دقيقة. يحترم الكود أعلاه Retry-After على كل 429، مما يغطي الحمل الزائد. إذا رأيت أخطاء concurrency على أي حال، فإن عملية أخرى تستخدم نفس المفتاح، أو أن تدفقًا قديمًا لا يزال مفتوحًا.

سجّل رأس X-Request-Id لكل استدعاء للخادم البعيد لتتبع استجابة سيئة، وسجّل X-Balance-USD لمراقبة رصيدك. البوابة مكان جيد للتنبيه عندما ينخفض الرصيد تحت عتبة معينة. يغطي قائمة مراجعة إنتاج بوابة AI الباقي: الأسرار، وفترات الانتظار، وفحوصات الصحة والمراقبة.

تتبع التكلفة داخل البوابة

نظرًا لأن كل استجابة تحمل استخدام الرموز، يمكن للبوابة تسجيل الإنفاق لكل رمز عميل. السعر هو $0.15 لكل مليون رمز مدخل و $0.50 لكل مليون رمز ناتج. لمكالمة استخدمت 1,200 رمز موجّه و 350 رمز إكمال، التكلفة هي 1,200 x $0.15 / مليون زائد 350 x $0.50 / مليون، وهو $0.00018 زائد $0.000175، أو $0.000355.

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

أضفه بعد تحليل استجابة JSON واحفظه مع رمز العميل. للبث المتدفق، اطلب من العميل تضمين الاستخدام في الجزء (chunk) الأخير عبر stream_options. الطلبات الفاشلة أو المرفوضة لا تُحسب، لذا احسب فقط الاستجابات الناجحة. الأسعار الحالية ومكافآت الشحن متوفرة في صفحة الأسعار.

الخطوات التالية

أصبح لديك الآن بوابة تتضمن المصادقة، وإعادة المحاولة، وحدًا للعمليات المتزامنة، وتمرير البث، في حوالي 80 سطرًا. من التوسعات الطبيعية إضافة ميزانيات لكل توكن، وتسجيل الطلبات في تخزينك الخاص، وإضافة خادم خلفي ثانٍ للتبديل عند الفشل. حافظ على حجم الكود صغيرًا؛ فعندما تحتاج إلى استخدام العديد منها في آن واحد، قد يكون مشروع البوابة المخصص أكثر ملاءمة.

لتشغيله، تحتاج إلى مفتاح للخادم الخلفي. سجّل واحصل على رصيد تجريبي بقيمة $0.50 بنقرة واحدة على صفحة مفتاح API، ثم تحقق من المعلمات وتنسيقات الأخطاء في وثائق API. تذكر أن المفتاح التجريبي يسمح بطلبين متزامنين فقط، لذا اضبط الـ semaphore على 2 أثناء الاختبار.

المزيد من الأدلة

• بوابة LLM مقابل OpenRouter: أيّهما تحتاج؟
• قائمة مراجعة بوابة AI: LLM بدون رقابة