Интеграция нейросети в Django: вьюхи, стриминг, Celery и учёт затрат

Интеграция LLM в Django обычно начинается с одного вызова API во вьюхе, а потом появляются вопросы: как не держать воркер минуту, как не платить за повторные вопросы, как считать расходы. Ниже — рабочая структура для Django-проекта на OpenAI SDK: настройки, модуль-обёртка с кэшем, обычная и потоковая вьюхи, фоновая задача Celery и учёт затрат в админке. Все модели в примерах вызываются через шлюз Zukko по OpenAI-совместимому адресу.

Установка и настройки

pip install django openai celery redis

В settings.py добавьте параметры клиента и кэш на Redis (бэкенд django.core.cache.backends.redis.RedisCache встроен в Django начиная с версии 4.0):

import os

ZUKKO_API_KEY = os.environ["ZUKKO_API_KEY"]
ZUKKO_BASE_URL = "https://api.zukko.pro/v1"
ZUKKO_MODEL = os.environ.get("ZUKKO_MODEL", "claude-sonnet-5")

CACHES = {
    "default": {
        "BACKEND": "django.core.cache.backends.redis.RedisCache",
        "LOCATION": "redis://127.0.0.1:6379/1",
    }
}

CELERY_BROKER_URL = "redis://127.0.0.1:6379/0"

Ключ передавайте через переменную окружения, а не храните в settings.py. Создать его можно на странице API-ключи↗; для сайта удобно завести отдельный ключ с дневным лимитом.

Модуль-обёртка: один клиент, кэш, учёт токенов

Файл myapp/llm.py:

import hashlib
import json
from decimal import Decimal

from django.conf import settings
from django.core.cache import cache
from openai import OpenAI

client = OpenAI(
    base_url=settings.ZUKKO_BASE_URL,
    api_key=settings.ZUKKO_API_KEY,
    timeout=60,
    max_retries=2,
)

Что здесь важно:

• Клиент создаётся один раз при импорте модуля и переиспользует соединения. timeout=60 не даст запросу висеть бесконечно, max_retries=2 повторит запрос при сетевых сбоях.
• Ключ кэша строится из модели и всех сообщений через json.dumps(..., sort_keys=True). Частая ошибка — склеивать строки вроде system or "" + user: из-за приоритета операций при заданном системном промпте пользовательский текст в ключ не попадает, и все вопросы получают один и тот же закэшированный ответ.
• Цены взяты из прайса Zukko (в долларах за 1M токенов) и хранятся как строки, чтобы Decimal считал без ошибок округления. Цены и набор моделей меняются: актуальный список — GET https://api.zukko.pro/v1/models или страница «Модели» в кабинете.

Помимо кэша Django, у Zukko есть серверный кэш для полностью одинаковых запросов: повтор с тем же телом отдаётся бесплатно (подробнее — в разделе лимиты ключа, идемпотентность и кэш↗). Кэш Django при этом экономит ещё и сетевой запрос.

Вьюха с обработкой ошибок

Файл myapp/views.py:

import logging

from django.http import JsonResponse, StreamingHttpResponse
from django.views.decorators.http import require_POST
from openai import APIConnectionError, APIStatusError

from .llm import ask

logger = logging.getLogger(__name__)

ERRORS = {
    401: "Ошибка конфигурации: неверный API-ключ",
    402: "Сервис временно недоступен: закончился баланс",
    403: "Модель не разрешена для этого ключа",

Пользователю уходит понятный текст, а подробности — в лог. Сырое str(e) в ответе показывать не стоит.

На фронтенде вставляйте ответ через textContent, а не innerHTML: и сообщение пользователя, и ответ модели — недоверенный текст, иначе получите XSS. CSRF-токен попадёт в запрос автоматически, если отправлять FormData формы с {% csrf_token %}.

Потоковый ответ через StreamingHttpResponse

Чтобы текст появлялся по мере генерации, отдавайте генератор в StreamingHttpResponse. Добавьте в myapp/llm.py:

def stream_answer(user_message, system=None, model=None):
    stream = client.chat.completions.create(
        model=model or settings.ZUKKO_MODEL,
        messages=build_messages(user_message, system),
        max_tokens=1024,
        stream=True,
    )
    for chunk in stream:
        if chunk.choices:
            text = chunk.choices[0].delta.content
            if text:
                yield text

И вьюху в myapp/views.py:

from .llm import stream_answer

@require_POST
def chat_stream(request):
    message = clean_message(request)
    if message is None:
        return JsonResponse({"error": "Пустое или слишком длинное сообщение"}, status=400)
    return StreamingHttpResponse(
        stream_answer(message), content_type="text/plain; charset=utf-8"
    )

Проверка if chunk.choices: обязательна: последний чанк потока содержит только usage и пустой список choices, и без проверки chunk.choices[0] упадёт с IndexError в конце ответа.

Учтите ограничения из документации Django: под WSGI потоковый ответ занимает воркер на всё время генерации, поэтому для долгих задач лучше вынести работу из цикла запрос-ответ. Под ASGI StreamingHttpResponse должен получать асинхронный итератор — для этого используйте AsyncOpenAI и async for. Ошибки API в потоковом варианте возникнут уже после отправки заголовков, внутри генератора, — логируйте их там.

Фоновые задачи в Celery

Длинные документы и пакетная обработка не должны держать HTTP-запрос. Модель myapp/models.py:

from django.conf import settings
from django.db import models

class ProcessedText(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
    input_text = models.TextField()
    result = models.TextField(blank=True)
    status = models.CharField(max_length=20, default="pending")
    error_message = models.TextField(blank=True)
    created_at = models.DateTimeField(auto_now_add=True)

class APIUsage(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, blank=True,
                             on_delete=models.SET_NULL)
    model_name = models.CharField(max_length=50)
    input_tokens = models.PositiveIntegerField()
    output_tokens = models.PositiveIntegerField()
    cost_usd = models.DecimalField(max_digits=12, decimal_places=8)
    created_at = models.DateTimeField(auto_now_add=True)

Задача myapp/tasks.py:

from celery import shared_task
from openai import APIConnectionError, APIStatusError, RateLimitError

from .llm import ask
from .models import ProcessedText

@shared_task(bind=True, max_retries=3)
def process_text(self, text_id):
    obj = ProcessedText.objects.get(pk=text_id)
    try:
        obj.result = ask(obj.input_text, use_cache=False, user=obj.user)
        obj.status = "completed"
    except (RateLimitError, APIConnectionError) as exc:
        raise self.retry(exc=exc, countdown=30)
    except APIStatusError as exc:
        obj.status = "failed"
        obj.error_message = f"{exc.status_code}: {exc.message}"
    obj.save(update_fields=["result", "status", "error_message"])

Повторяем только временные ошибки (429 и сетевые). Ошибки 401, 402, 403 повторять бессмысленно — задача сразу помечается как неудачная. Во вьюхе создайте объект и запустите задачу: process_text.delay(obj.pk). Для подключения Celery к Django нужен стандартный файл celery.py из документации Celery, а воркер запускается командой celery -A myproject worker -l info.

Расходы в админке

from django.contrib import admin
from .models import APIUsage

@admin.register(APIUsage)
class APIUsageAdmin(admin.ModelAdmin):
    list_display = ("created_at", "user", "model_name",
                    "input_tokens", "output_tokens", "cost_usd")
    list_filter = ("model_name", "created_at")

Пример расчёта: запрос на claude-sonnet-5 с 800 входными и 300 выходными токенами стоит (800 × 0.3 + 300 × 0.3) / 1 000 000 = $0.00033. Эти цифры можно сверить с разделом «Использование» в кабинете. Базовая настройка Python-клиента описана в руководстве по Python↗.

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

Зачем кэш, если запросы недорогие?

Кэш экономит и деньги, и время: повторный ответ приходит без запроса к модели. На сайте с типовыми вопросами (FAQ-бот, подсказки в форме) доля повторов бывает заметной — проверьте по своим логам.

Синхронный или асинхронный клиент?

Под WSGI используйте синхронный OpenAI, как в примерах. Если проект работает под ASGI и вы пишете асинхронные вьюхи, берите AsyncOpenAI с теми же параметрами.

Какую модель выбрать?

По умолчанию в примере claude-sonnet-5; дешевле всего glm-5.3-flash ($0.04 за 1M токенов). Модель задаётся переменной ZUKKO_MODEL, поэтому сравнить их на своих запросах можно без изменения кода.

Что если воркер Celery не запущен?

Задача будет ждать в очереди брокера и выполнится, когда воркер стартует. Статус pending в модели поможет показать пользователю, что обработка ещё идёт.