PythonでのLLMゲートウェイ：バックエンドエンドポイントの呼び出しの呼び出し

Pythonの軽量ゲートウェイの役割

有用な llm gateway Python 環境は、人が想定するよりも小さく済みます。フルプラットフォームは不要です。軽量なゲートウェイは、アプリとバックエンドエンドポイントの間に位置する小さな Web サービスで、次の 4 つの機能を追加します：クライアント用の共有認証トークン、タイムアウトとリトライポリシー、同時アップストリーム呼び出しの制限、そしてクリーンなストリーミングパススルーです。アプリは OpenAI のようにこれを操作し、実際のバックエンドに転送します。

この記事では FastAPI と httpx を使用して構築します。バックエンドには https://api.openllmgateway.com/v1 を、モデル ID には uncensored を使用します。まだゲートウェイが必要かどうか迷っている場合は、まず LLM ゲートウェイとは何か をお読みください。ここではコードが主眼です。

セットアップと非ストリーミングルート

依存関係をインストールし、2つの環境変数を設定します：バックエンドキーと、独自のアプリに配布するトークンのカンマ区切りリストです。アプリはバックエンドキーを直接確認できません。

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 に強制している点に注意してください。クライアントはそれ以外のものをリクエストできないため、これは 1 つの場所でのポリシー管理を維持する簡単な方法です。

スロットリークのないストリーミングパススルー

ストリーミングは軽量ゲートウェイが失敗する箇所です。アップストリームの本文全体を読み取ってから送信すると、ユーザーは最後まで何も見えません。クライアントが切断されたときに同時実行スロットを解放し忘れると、最終的に自身をブロックします。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で実行します。アプリは、ゲートウェイをベースURLとして、独自のトークンの1つをキーとして使用して標準のOpenAIクライアントを使用します：

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の適切な処理

バックエンドは3つの状況で429を返答し、ゲートウェイはそれらを区別するべきです。rate_limitedエラーコードは、1分あたりのリクエスト数を超えたことを意味します：フルキーでは300、トライアルキーでは60です。concurrencyコードは、一度に多すぎるリクエストがあることを意味します：フルキーでは8、トライアルキーでは2です。一時的な過負荷もRetry-After: 5で429を返します。

上記のセマフォは、ゲートウェイがそのキー上の唯一のクライアントである限り、まず同時実行ケースを防ぎます。1分あたりのリクエスト数については、単純なトークンバケットまたは1分あたりのカウンターを追加します。上記のコードは、すべての429でRetry-Afterを尊重し、過負荷をカバーします。それでもconcurrencyエラーが表示される場合は、別のプロセスが同じキーを使用しているか、古いストリームがまだ開いています。

各アップストリーム呼び出しのX-Request-Idヘッダーをログに記録して、不良応答を追跡し、X-Balance-USDをログに記録してクレジットを監視します。残高が閾値を下回ったときにアラートを出すには、ゲートウェイが最適な場所です。AIゲートウェイのプロダクションチェックリストは、シークレット、タイムアウト、ヘルスチェック、モニタリングの残りをカバーします。

ゲートウェイ内のコスト追跡

すべての応答にはトークン使用量が含まれるため、ゲートウェイはクライアントトークンごとに支出を記録できます。価格は1M入力トークンあたり$0.15、1M出力トークンあたり$0.50です。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ゲートウェイ vs OpenRouter：どちらが必要か？
• AIゲートウェイチェックリスト：無検閲LLM API