Асинхронная обработка AI-запросов — это архитектурный паттерн, при котором клиент не ждёт завершения генерации, а получает уведомление через webhook, когда результат готов. Это критически важно для длительных задач: анализ документов, генерация больших текстов, пакетная обработка.
Зачем нужна асинхронность
Синхронный подход (клиент ждёт ответа) имеет серьёзные ограничения:
- Таймауты HTTP-соединения (обычно 30-60 секунд)
- Блокировка пользовательского интерфейса
- Невозможность масштабирования пакетных задач
- Потеря результата при разрыве соединения
Асинхронный паттерн решает эти проблемы: запрос ставится в очередь, обрабатывается фоново, а результат доставляется через webhook.
Архитектура с очередью задач
Реализуем систему на Python с использованием Celery, Redis и ModelSwitch API:
# tasks.py — Celery-задачи для AI-генерации
import os
import requests
from celery import Celery
from openai import OpenAI
app = Celery("ai_tasks", broker="redis://localhost:6379/0")
client = OpenAI(
base_url=os.environ["MODELSWITCH_BASE_URL"],
api_key=os.environ["MODELSWITCH_API_KEY"],
)
@app.task(bind=True, max_retries=3)
def generate_ai_response(self, task_id: str, prompt: str, webhook_url: str, model: str = "gpt-4o"):
"""Асинхронная генерация AI-ответа с уведомлением через webhook."""
try:
response = client.chat.completions.create(
model=model,
messages=[{"role": "user", "content": prompt}],
max_tokens=4000,
)
result = {
"task_id": task_id,
"status": "completed",
"response": response.choices[0].message.content,
"model": model,
"usage": {
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
},
}
# Отправляем результат на webhook
requests.post(webhook_url, json=result, timeout=10)
except Exception as exc:
# Уведомляем об ошибке
requests.post(webhook_url, json={
"task_id": task_id,
"status": "failed",
"error": str(exc),
}, timeout=10)
raise self.retry(exc=exc, countdown=5)API-эндпоинт для постановки задач
Создайте API, который принимает запрос, ставит его в очередь и возвращает идентификатор задачи:
# api.py — FastAPI-сервер
import uuid
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from tasks import generate_ai_response
app = FastAPI()
class TaskRequest(BaseModel):
prompt: str
model: str = "gpt-4o"
webhook_url: str
class TaskResponse(BaseModel):
task_id: str
status: str
@app.post("/api/tasks", response_model=TaskResponse)
def create_task(request: TaskRequest):
task_id = str(uuid.uuid4())
# Ставим задачу в очередь Celery
generate_ai_response.delay(
task_id=task_id,
prompt=request.prompt,
webhook_url=request.webhook_url,
model=request.model,
)
return TaskResponse(task_id=task_id, status="queued")
@app.post("/webhook/results")
async def receive_webhook(payload: dict):
"""Пример эндпоинта для приёма результатов."""
task_id = payload.get("task_id")
status = payload.get("status")
print(f"Task {task_id}: {status}")
# Здесь: сохранить в БД, уведомить пользователя, и т.д.
return {"received": True}Пакетная обработка
Для обработки большого числа задач одновременно отправляйте их пачками с отслеживанием прогресса:
# batch.py — пакетная обработка
import uuid
from tasks import generate_ai_response
def process_batch(prompts: list[str], webhook_url: str, model: str = "gpt-4o-mini"):
"""Ставим пакет задач в очередь."""
batch_id = str(uuid.uuid4())
tasks = []
for i, prompt in enumerate(prompts):
task_id = f"{batch_id}:{i}"
generate_ai_response.delay(
task_id=task_id,
prompt=prompt,
webhook_url=webhook_url,
model=model,
)
tasks.append(task_id)
return {"batch_id": batch_id, "total": len(tasks), "task_ids": tasks}
# Использование:
# result = process_batch(
# prompts=["Переведи: Hello", "Переведи: World"],
# webhook_url="https://your-app.com/webhook/results"
# )Для пакетных задач рекомендуется использовать GPT-4o-mini — он в 15 раз дешевле GPT-4o при достаточном качестве для большинства задач обработки данных.
Заключение
Асинхронная обработка с webhooks — это правильный подход для production AI-приложений. Celery + Redis обеспечивают надёжную очередь задач, а ModelSwitch API — единый интерфейс для всех моделей. Используйте пакетную обработку для массовых задач и дешёвые модели (GPT-4o-mini) для экономии бюджета.