เกตเวย์ LLM ใน Python: การเรียกใช้เอนด์พอยต์ของแบ็กเอนด์

เกตเวย์บางๆ ใน Python ทำอะไร

การตั้งค่า llm gateway python ที่มีประโยชน์มีขนาดเล็กกว่าที่คนคาดหวัง คุณไม่จำเป็นต้องใช้แพลตฟอร์มเต็มรูปแบบ เกตเวย์บางๆ คือบริการเว็บขนาดเล็กที่อยู่ระหว่างแอปของคุณและเอนด์พอยต์ของแบ็กเอนด์ และเพิ่มสี่สิ่ง: auth token ร่วมกันหนึ่งรายการสำหรับไคลเอนต์ของคุณเอง นโยบายเวลาหมดอายุและการทำซ้ำ ขีดจำกัดการเรียก upstream พร้อมกัน และการส่งต่อสตรีมมิงที่สะอาด แอปของคุณจะสื่อสารกับมันราวกับว่าเป็น OpenAI และมันจะส่งต่อไปยังแบ็กเอนด์จริง

บทความนี้สร้างสิ่งนั้นด้วย FastAPI และ httpx โดยใช้ https://api.openllmgateway.com/v1 เป็นแบ็กเอนด์และ model ID uncensored หากคุณยังตัดสินใจว่าจะต้องการเกตเวย์หรือไม่ ให้อ่าน เกตเวย์ LLM คืออะไร ก่อน จุดประสงค์ที่นี่คือโค้ด

การตั้งค่าและเส้นทางที่ไม่ใช้สตรีมมิง

ติดตั้ง dependencies และตั้งค่าตัวแปรสภาพแวดล้อมสองตัว: คีย์ API ของแบ็กเอนด์ และรายการโทเคนที่คั่นด้วยเครื่องหมายจุลภาคซึ่งคุณแจกจ่ายให้กับแอปของคุณ แอปของคุณจะไม่เห็นคีย์แบ็กเอนด์

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 จะถือว่าคำขอที่สตรีมอยู่กำลังใช้สล็อตความพร้อมกันจนกว่าสตรีมจะจบ ดังนั้นเซมาโฟร์ของคุณควรทำเช่นเดียวกัน

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 มาตรฐานโดยมีเกตเวย์เป็น Base 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

เซมาโฟร์ข้างต้นป้องกันกรณีคอนเคอร์เรนซ์ตั้งแต่แรก ตราบใดที่เกตเวย์เป็นไคลเอนต์เพียงรายเดียวที่ใช้คีย์นั้น สำหรับคำขอต่อนาที ให้เพิ่มถังโทเคนแบบง่ายหรือตัวนับต่อนาที โค้ดข้างต้นเคารพค่า Retry-After ในทุกสถานะ 429 ซึ่งครอบคลุมกรณีโอเวอร์โหลด หากคุณยังเห็นข้อผิดพลาด concurrency แสดงว่ามีกระบวนการอื่นใช้คีย์เดียวกัน หรือสตรีมเก่ายังเปิดอยู่

บันทึกหัว X-Request-Id สำหรับการเรียก upstream แต่ละครั้งเพื่อติดตามการตอบกลับที่ผิดปกติ และบันทึก 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 บรรทัด การขยายตามธรรมชาติคือ งบประมาณต่อโทเคน การบันทึกคำขอไปยังที่เก็บข้อมูลของคุณเอง และแบ็กเอนด์ที่สองสำหรับการล้มเหลว การรักษาโค้ดให้เล็กเป็นสิ่งสำคัญ; เมื่อคุณต้องการฟีเจอร์เหล่านั้นจำนวนมากพร้อมกัน โครงการเกตเวย์เฉพาะทางอาจเหมาะสมกว่า

เพื่อเรียกใช้งาน คุณต้องการคีย์แบ็กเอนด์ สมัครสมาชิกและรับเครดิตทดลองใช้ $0.50 ด้วยคลิกเดียวที่ หน้าคีย์ API จากนั้นตรวจสอบพารามิเตอร์และรูปแบบข้อผิดพลาดใน เอกสาร API โปรดจำไว้ว่าคีย์ทดลองอนุญาตให้ใช้คำขอพร้อมกันได้อย่างมาก 2 คำขอ ดังนั้นให้ตั้งค่าเซมาโฟร์เป็น 2 ขณะทดสอบ

คู่มือเพิ่มเติม

• LLM Gateway เทียบกับ OpenRouter: คุณต้องการอันไหน?
• รายการตรวจสอบเกตเวย์ AI: API LLM แบบไม่เซ็นเซอร์