Python용 LLM 게이트웨이: 백엔드 엔드포인트 호출

Python의 얇은 게이트웨이가 수행하는 작업

유용한 LLM 게이트웨이 Python 설정은 사람들이 예상하는 것보다 작습니다. 전체 플랫폼이 필요하지 않습니다. 얇은 게이트웨이는 앱과 백엔드 엔드포인트 사이에 위치하여 네 가지 기능을 추가하는 작은 웹 서비스입니다: 자체 클라이언트를 위한 하나의 공유 인증 토큰, 시간 제한 및 재시도 정책, 상류 호출에 대한 동시성 제한 및 깨끗한 스트리밍 전달. 앱은 이를 OpenAI처럼 사용하고 실제 백엔드로 전달합니다.

이 글은 FastAPI와 httpx를 사용하여 https://api.openllmgateway.com/v1를 백엔드로 사용하고 모델 ID 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로 강제한다는 점에 유의하세요. 클라이언트는 다른 것을 요청할 수 없으며, 이는 하나의 정책을 한 곳에 유지하는 간단한 방법입니다.

슬롯 누수가 없는 스트리밍 전달

스트리밍은 얇은 게이트웨이가 실수하는 부분입니다. 전체 상류 본문을 읽고 나중에 보내면 사용자는 마지막까지 아무것도 보지 못합니다. 클라이언트가 연결을 끊을 때 동시성 슬롯을 해제하는 것을 잊으면 결국 자신을 차단하게 됩니다. 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로 실행하세요. 이제 앱은 게이트웨이를 base URL로 사용하고 표준 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 올바르게 처리하기

백엔드는 세 가지 상황에서 429를 반환하며 게이트웨이는 이를 구분해야 합니다. rate_limited 오류 코드는 분당 요청 수를 초과했음을 의미합니다: 전체 키는 300, 체험 키는 60입니다. concurrency 코드는 동시에 너무 많은 요청이 있음을 의미합니다: 전체 키는 8, 체험 키는 2입니다. 일시적인 과부하도 Retry-After: 5와 함께 429를 반환합니다.

위의 세마포어는 게이트웨이가 해당 키의 유일한 클라이언트인 한 동시성 사례를 처음부터 방지합니다. 분당 요청의 경우 간단한 토큰 버킷 또는 분당 카운터를 추가하세요. 위의 코드는 모든 429에서 Retry-After를 준수하며 이는 과부하를 처리합니다. 여전히 concurrency 오류가 발생하면 다른 프로세스가 동일한 키를 사용 중이거나 오래된 스트림이 여전히 열려 있습니다.

각 상류 호출에 대해 X-Request-Id 헤더를 로그에 기록하여 잘못된 응답을 추적하고, X-Balance-USD을 로그에 기록하여 크레딧을 모니터링하세요. 게이트웨이는 잔액이 임계값 아래로 떨어질 때 알림을 보내기에 좋은 장소입니다. AI 게이트웨이 프로덕션 체크리스트는 나머지 부분(시크릿, 시간 제한, 상태 확인 및 모니터링)을 다룹니다.

게이트웨이 내 비용 추적

모든 응답에는 토큰 사용량이 포함되므로 게이트웨이는 클라이언트 토큰당 지출을 기록할 수 있습니다. 가격은 입력 토큰 100만 개당 $0.15, 출력 토큰 100만 개당 $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