AI-эндпоинт на FastAPI через AI Gateway: async и стриминг

Соберём в FastAPI пару эндпоинтов, которые обращаются к моделям из каталога AI Gateway: один отдаёт готовый ответ, другой стримит его по мере генерации. FastAPI асинхронный, поэтому берём AsyncOpenAI — он совместим с шлюзом на уровне base_url и не блокирует event loop.

Подготовка

Нужен ключ sk-aigate-… из кабинета и базовый URL https://ai-gatewey.ru/v1. Ключ держим в переменной окружения, а не в коде.

pip install fastapi uvicorn openai

Реализация

1. 1
   Настройте асинхронный клиент

   Один экземпляр AsyncOpenAI на всё приложение — создаём при старте и переиспользуем во всех обработчиках:

   python Копировать

   import os
   from openai import AsyncOpenAI
   
   client = AsyncOpenAI(
       base_url="https://ai-gatewey.ru/v1",
       api_key=os.environ["AIGATE_API_KEY"],
   )

2. 2
   Эндпоинт с обычным ответом

   Принимаем текст в теле запроса, отдаём ответ модели одним JSON:

   python Копировать

   from fastapi import FastAPI
   from pydantic import BaseModel
   
   app = FastAPI()
   
   class Prompt(BaseModel):
       message: str
   
   @app.post("/chat")
   async def chat(body: Prompt):
       resp = await client.chat.completions.create(
           model="deepseek/deepseek-chat",
           messages=[{"role": "user", "content": body.message}],
       )
       return {"reply": resp.choices[0].message.content}

3. 3
   Стриминг через StreamingResponse

   Для «печатающегося» ответа включаем stream=True и отдаём чанки генератором в StreamingResponse:

   python Копировать

   from fastapi.responses import StreamingResponse
   
   @app.post("/chat/stream")
   async def chat_stream(body: Prompt):
       async def gen():
           stream = await client.chat.completions.create(
               model="deepseek/deepseek-chat",
               messages=[{"role": "user", "content": body.message}],
               stream=True,
           )
           async for chunk in stream:
               delta = chunk.choices[0].delta.content
               if delta:
                   yield delta
       return StreamingResponse(gen(), media_type="text/plain")

4. 4
   Запустите сервер

   Поднимаем uvicorn и проверяем эндпоинты; расход по токенам виден в кабинете AI Gateway в реальном времени:

   bash Копировать

   uvicorn main:app --reload

Какие модели подойдут

Для бэкенд-эндпоинтов под нагрузкой обычно берут быструю недорогую модель, а флагман подключают точечно там, где нужен максимум качества:

Частые вопросы