Cổng LLM bằng Python: Gọi một endpoint backend

Một gateway mỏng bằng Python làm những gì

Một thiết lập cổng LLM bằng Python hữu ích nhỏ hơn nhiều so với những gì mọi người nghĩ. Bạn không cần một nền tảng đầy đủ. Một gateway mỏng là một dịch vụ web nhỏ nằm giữa các ứng dụng của bạn và một endpoint backend, thêm bốn thứ: một token xác thực chia sẻ cho các client của riêng bạn, chính sách thời gian chờ và thử lại, giới hạn các lệnh gọi upstream đồng thời, và truyền phát streaming sạch sẽ. Các ứng dụng của bạn giao tiếp với nó như thể nó là OpenAI, và nó chuyển tiếp đến backend thực sự.

Bài viết này xây dựng điều đó với FastAPI và httpx, sử dụng https://api.openllmgateway.com/v1 làm backend và model ID uncensored. Nếu bạn vẫn đang phân vân có cần gateway không, hãy đọc gateway LLM là gì trước. Mục đích ở đây là mã nguồn.

Thiết lập và tuyến không streaming

Cài đặt các phụ thuộc và đặt hai biến môi trường: khóa backend và danh sách các token phân cách bằng dấu phẩy mà bạn cấp cho các ứng dụng của riêng mình. Các ứng dụng của bạn không bao giờ thấy khóa backend.

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

Nửa đầu của dịch vụ bao gồm cấu hình, xác thực và tuyến JSON đơn giản với các lần thử lại:

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

Lưu ý rằng gateway ép model thành uncensored. Các client không thể yêu cầu bất kỳ thứ gì khác, đây là một cách đơn giản để giữ một chính sách ở một nơi.

Truyền phát streaming không rò rỉ slot

Streaming là nơi các gateway mỏng thường sai. Nếu bạn đọc toàn bộ body upstream rồi gửi nó, người dùng sẽ không thấy gì cho đến khi kết thúc. Nếu bạn quên giải phóng slot độ đồng thời khi client ngắt kết nối, bạn cuối cùng sẽ tự chặn mình. API coi một yêu cầu được streaming là chiếm slot độ đồng thời của nó cho đến khi stream kết thúc, vì vậy semaphore cục bộ của bạn nên làm tương tự.

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

Khối finally chạy khi luồng hoàn tất, khi client ngắt kết nối hoặc khi xảy ra lỗi, nên slot luôn được hoàn trả. Các lần thử lại được cố ý loại bỏ ở đây. Một khi byte đã được gửi đến client, bạn không thể khởi động lại một cách sạch sẽ, vì vậy chỉ nên thử lại trước chunk đầu tiên, và trong phiên bản này, kiểm tra trạng thái ở trên là cửa kiểm duy nhất.

Gọi gateway của bạn từ một client

Chạy nó với uvicorn gateway:app --port 8000. Các ứng dụng của bạn bây giờ sử dụng client OpenAI tiêu chuẩn với gateway làm base URL và một trong các token của riêng bạn làm khóa:

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)

Bạn có thể thêm các tuyến khác theo cách tương tự. GET /v1/models và POST /v1/responses có sẵn trên backend, và gọi hàm cùng chế độ JSON không cần xử lý đặc biệt vì gateway chuyển tiếp thân yêu cầu không thay đổi. Kích thước thân yêu cầu tối đa là 8 MB, vì vậy hãy đặt giới hạn phù hợp phía trước gateway nếu nó tiếp xúc với internet.

Xử lý 429 đúng cách

Backend trả về 429 trong ba tình huống, và gateway của bạn nên phân biệt chúng. Mã lỗi rate_limited có nghĩa là bạn đã vượt quá số yêu cầu mỗi phút: 300 trên khóa đầy đủ, 60 trên khóa dùng thử. Mã concurrency có nghĩa là quá nhiều yêu cầu cùng lúc: 8 trên khóa đầy đủ, 2 trên khóa dùng thử. Quá tải tạm thời cũng trả về 429 với Retry-After: 5.

Semaphore ở trên ngăn chặn trường hợp độ đồng thời ngay từ đầu, miễn là gateway là khách hàng duy nhất trên khóa đó. Đối với số yêu cầu mỗi phút, hãy thêm một thùng token đơn giản hoặc bộ đếm mỗi phút. Mã ở trên tuân thủ Retry-After trên mọi phản hồi 429, điều này bao gồm trường hợp quá tải. Nếu bạn vẫn thấy lỗi concurrency, một tiến trình khác đang sử dụng cùng khóa, hoặc một luồng cũ vẫn đang mở.

Ghi lại tiêu đề X-Request-Id cho mỗi lệnh gọi upstream để bạn có thể truy vết phản hồi xấu, và ghi lại X-Balance-USD để theo dõi tín dụng của bạn. Một gateway là nơi tốt để cảnh báo khi số dư giảm xuống dưới ngưỡng. Danh sách kiểm tra sản xuất AI gateway bao gồm những phần còn lại: bí mật, thời gian chờ, kiểm tra sức khỏe và giám sát.

Theo dõi chi phí bên trong gateway

Bởi vì mọi phản hồi đều mang theo việc sử dụng token, gateway có thể ghi lại chi tiêu theo token client. Giá là $0,15 cho mỗi 1M token đầu vào và $0,50 cho mỗi 1M token đầu ra. Đối với một lệnh gọi đã sử dụng 1.200 token prompt và 350 token hoàn tất, chi phí là 1.200 x $0,15 / 1M cộng 350 x $0,50 / 1M, tức là $0,00018 cộng $0,000175, hoặc $0,000355.

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

Thêm nó sau khi phân tích cú pháp phản hồi JSON và lưu trữ nó với token client. Đối với các stream, hãy yêu cầu client bao gồm việc sử dụng trong chunk cuối cùng với stream_options. Các yêu cầu bị lỗi hoặc bị từ chối không bị tính phí, vì vậy chỉ đếm các phản hồi thành công. Các tỷ lệ hiện tại và bonus khi nạp tiền có trên trang giá.

Các bước tiếp theo

Bạn hiện có một gateway với xác thực, thử lại, giới hạn độ đồng thời và chuyển tiếp luồng chỉ trong khoảng 80 dòng. Các hướng mở rộng tự nhiên là ngân sách trên mỗi token, ghi lại yêu cầu vào kho lưu trữ của riêng bạn, và một backend thứ hai để dự phòng. Hãy giữ mã nguồn nhỏ gọn; một khi bạn cần nhiều cái như vậy cùng lúc, một dự án gateway chuyên dụng có thể phù hợp hơn.

Để chạy nó, bạn cần một khóa backend. Đăng ký và nhận tín dụng dùng thử $0,50 với một cú nhấp chuột trên trang khóa API, sau đó kiểm tra các tham số và định dạng lỗi trong tài liệu API. Hãy nhớ rằng khóa dùng thử chỉ cho phép 2 yêu cầu đồng thời, vì vậy hãy đặt semaphore thành 2 trong khi kiểm tra.

Hướng dẫn khác

• So sánh LLM Gateway với OpenRouter: Bạn cần cái nào?
• Danh sách kiểm tra AI Gateway: API LLM không kiểm duyệt