Шлюз 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. Клиенты не могут запросить что-то другое, что является простым способом хранить одну политику в одном месте.

Потоковая передача без утечки слотов

Потоковая передача — это то место, где тонкие шлюзы ошибаются. Если вы прочитаете все тело ответа бэкенда, а затем отправите его, пользователь ничего не увидит до конца. Если вы забудете освободить слот параллельных запросов при отключении клиента, вы в конечном итоге заблокируете себя. 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 с шлюзом в качестве базового 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 МБ, поэтому установите соответствующее ограничение перед шлюзом, если он обращен в интернет.

Правильная обработка 429

Бэкенд отвечает 429 в трех ситуациях, и ваш шлюз должен различать их. Код ошибки rate_limited означает, что вы превысили количество запросов в минуту: 300 для полного ключа, 60 для пробного ключа. Код concurrency означает слишком много запросов одновременно: 8 для полного ключа, 2 для пробного ключа. Временная перегрузка также возвращает 429 с Retry-After: 5.

Семафор выше предотвращает случай параллельных запросов изначально, пока шлюз является единственным клиентом на этом ключе. Для запросов в минуту добавьте простой токен-бакет или счетчик на минуту. Приведенный выше код учитывает Retry-After при каждом 429, что покрывает перегрузку. Если вы все равно видите ошибки concurrency, другой процесс использует тот же ключ или старый поток все еще открыт.

Логируйте заголовок X-Request-Id для каждого запроса к бэкенду, чтобы отслеживать плохие ответы, и логируйте 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 против OpenRouter: что вам нужно?
• Чек-лист AI-шлюза: API LLM без цензуры