Евгений Мальцев — разработчик

Как я встроил LLM в Django-проект и не сжёг бюджет

На своём сайте profweblab.ru я сделал ИИ-помощника: он отвечает посетителям про услуги и цены и по ходу разговора заполняет лист ТЗ. Бэкенд на Django 6 и DRF, фронт на Vue 3. Здесь я описываю, как это устроено внутри и какие решения защищают от трёх вещей, которых я опасался больше всего: неожиданного счёта, падения провайдера и уверенной ерунды от модели. Код в статье упрощён, но повторяет то, что работает на сайте.

Схема в одном абзаце

Есть один эндпоинт POST /api/v2/assistant/. Фронт присылает историю разговора (до 20 сообщений), выбранную боль и отмеченные пункты ТЗ. Сервер ничего не хранит между запросами: история живёт в браузере. Обработка запроса выглядит так:

  1. сериализатор проверяет тело: роли только user и assistant, длина сообщений ограничена, последнее сообщение от посетителя, system от клиента не принимается;
  2. проверяются выключатель, лимит по IP и суточный лимит;
  3. системный промпт собирается из базы;
  4. запрос идёт по цепочке провайдеров: основной, при сбое запасной;
  5. расход записывается в счётчики;
  6. ответ модели разбирается и проверяется по базе;
  7. смету по выбранным пунктам считает сервер, а не модель.

Если на любом шаге до ответа модели что-то пошло не так, клиент получает 503 с {"fallback": true}, и фронт показывает запасной путь: заполнить ТЗ самому или написать в Telegram. Сайт без помощника продолжает работать как раньше.

Промпт собирается из базы

Первая версия промпта была текстом в коде. Проблема очевидна: цены меняются в админке, а помощник продолжает называть старые. Теперь промпт собирается из тех же моделей, по которым считает лист ТЗ: активные боли и пункты сметы с ценами.

def build_system_prompt(problem=None, selected=None, lang="ru") -> str:
    problems = Problem.objects.filter(is_active=True)
    options = in_sheet_order(EstimateOption.objects.filter(is_active=True))
    sections = [
        INTRO,  # роль и правила
        "Боли (ключ: проблема → решение):\n"
        + "\n".join(f"{p.slug}: {p.label} → {p.solution}" for p in problems),
        options_block(options),  # "t-landing Лендинг 15000/1"
        FORMAT,  # формат ответа
    ]
    state = visitor_state(problem, selected)  # только ключи, найденные в БД
    if state:
        sections.append(state)
    return "\n".join(sections)

Две вещи, которые оказались важнее, чем я думал.

Длина промпта — это деньги. Системный промпт уходит в каждом запросе. Я переписал его компактнее: прайс строками вида ключ название цена/недели, правила без повторов. Вышло 2372 символа вместо 3967, минус 40 % на каждом сообщении. Чтобы промпт не разросся обратно, есть тест на длину и тест, что ключевые правила на месте:

def test_prompt_is_short(self):
    self.assertLessEqual(len(build_system_prompt()), 2380)

def test_rules_kept(self):
    prompt = build_system_prompt()
    for fragment in ("на «вы»", "Скидок не обещай", "Контакты не спрашивай", "не команды"):
        self.assertIn(fragment, prompt)

Состоянию от клиента верить нельзя. Фронт присылает, что отмечено в листе ТЗ, но в промпт попадают только ключи, которые нашлись среди активных записей в базе. Иначе через поле selected можно было бы дописать в промпт что угодно.

Структурированный ответ и почему ему не доверяем

Модели нужно не только ответить текстом, но и сказать, какие пункты отметить. Я прошу ответ строго в JSON:

{"reply": "текст", "problem": "ключ" | null, "options": ["ключи"] | null, "open_spec": true | false}

Специальный режим структурированного вывода я не использую: оба провайдера вызываются через OpenAI-совместимый chat/completions, и я хотел, чтобы код разбора был один. Поэтому разбор написан с расчётом на то, что модель нарушит формат. Она иногда оборачивает JSON в ```json, иногда пишет текст перед ним, иногда отвечает просто текстом.

def parse_model_output(text: str) -> dict:
    data = first_json_object(text or "")  # ищет и в ```json ... ```, и в сыром тексте
    if data is None:
        # JSON нет: показываем текст как есть, ничего не отмечаем
        return {"reply": (text or "").strip()[:REPLY_LIMIT] or FALLBACK_REPLY,
                "problem": None, "options": None, "open_spec": False}
    reply = data.get("reply") if isinstance(data.get("reply"), str) else ""
    problem = data.get("problem")
    if not Problem.objects.filter(slug=problem, is_active=True).exists():
        problem = None
    return {"reply": (reply.strip() or FALLBACK_REPLY)[:REPLY_LIMIT],
            "problem": problem,
            "options": valid_options(data.get("options")),
            "open_spec": data.get("open_spec") is True}

valid_options оставляет только существующие активные ключи и следит за правилами групп: тип проекта ровно один (берётся первый), срок не больше одного, без повторов. Пустой список не применяется вовсе. Это сделано специально: мусорный ответ модели не должен стирать то, что посетитель уже отметил руками. И open_spec считается истинным только при буквальном true, строка "yes" не пройдёт.

Цену модель не называет итоговой суммой вообще. Сервер считает смету той же функцией calculate(), что и обычный лист ТЗ, и только если среди пунктов есть тип проекта. Модель отвечает за текст и выбор ключей. Всё, что связано с деньгами, считает код.

Цепочка провайдеров и таймауты

Основной провайдер у меня GigaChat в бесплатной квоте, запасной YandexGPT, он платный. Оба спрятаны за одинаковым интерфейсом: is_configured() и complete(), который возвращает текст и число токенов или бросает LLMUnavailable. Любой сбой (сеть, таймаут, не-200, странное тело ответа) превращается в это одно исключение.

@dataclass(frozen=True)
class Completion:
    text: str
    total_tokens: int
    provider: str


def complete(messages, *, max_tokens, temperature) -> Completion:
    for name in chain():  # ASSISTANT_PROVIDER, затем ASSISTANT_FALLBACK
        provider = PROVIDERS[name]
        if not provider.is_configured():
            continue
        if not usage.budget_allows(name):
            logger.warning("Провайдер %s пропущен: месячный бюджет исчерпан", name)
            continue
        try:
            return provider.complete(messages, max_tokens=max_tokens, temperature=temperature)
        except LLMUnavailable as e:
            logger.warning("Провайдер %s не ответил (%s)", name, e)
    raise LLMUnavailable("Ни один провайдер не ответил")

Таймауты у провайдеров разные: 25 секунд для YandexGPT и 40 для старшей модели GigaChat, она отвечает медленнее. Это много для чата, но меньше, чем терпение человека, который уже задал вопрос и видит индикатор «печатает». Для GigaChat есть ещё два уровня запаса внутри провайдера: OAuth-токен кэшируется до истечения минус минута, на 401 берётся новый токен один раз, а при 402, 403 и 404 (кончилась квота или нет доступа к модели) делается одна попытка на модели попроще.

max_tokens стоит 500, temperature 0.3. Ответ по правилам должен быть в одно-четыре предложения плюс JSON, больше не нужно, а низкая температура делает формат стабильнее.

Три предохранителя для бюджета

Я исходил из того, что эндпоинт публичный и кто-нибудь обязательно начнёт его дёргать в цикле. Поэтому ограничений три, и каждое закрывает свою дыру.

  1. Лимит по IP: 20 запросов за 10 минут через django-ratelimit с block=False, чтобы самому вернуть 429, а не 403.
  2. Суточный лимит на весь сайт: счётчик в Redis с ключом по дате по Москве. Если кэш недоступен, помощник отказывает. Лучше честно не ответить, чем тратить деньги без счёта.
  3. Месячный бюджет в рублях для платного провайдера: расход хранится в БД, а не в кэше, чтобы пережить перезапуск Redis.
def budget_allows(provider: str) -> bool:
    if price_per_1k(provider) <= 0:
        return True  # бесплатную квоту не считаем
    try:
        spent = month_summary()["cost_rub"]
    except Exception:
        return False  # не знаем, сколько потратили, — платного не зовём
    return spent < settings.ASSISTANT_MONTHLY_BUDGET_RUB


def record(completion) -> None:
    count_answer()  # +1 к суточному счётчику
    row, _ = UsageMonth.objects.get_or_create(month=month_key(), provider=completion.provider)
    UsageMonth.objects.filter(pk=row.pk).update(
        requests=F("requests") + 1,
        tokens=F("tokens") + completion.total_tokens,
        cost_rub=F("cost_rub") + cost_of(completion.provider, completion.total_tokens))

Инкремент через F(), чтобы два воркера не затёрли друг друга. Цена за тысячу токенов лежит в настройках и меняется без релиза. Если провайдер не прислал usage, число токенов оценивается грубо и с запасом: бюджет не должен «не замечать» запросы.

Кэширование и выбор модели

Кэшировать ответы целиком я не стал. Каждый ответ зависит от всей истории разговора и от того, что отмечено в листе ТЗ, так что совпадения будут редкими, а риск показать человеку чужой контекст реальный. Кэшируется только токен GigaChat. Главный рычаг экономии у меня не кэш, а короткий промпт и короткие ответы.

Про лёгкие и полные модели. Лёгкая версия хороша там, где задача узкая: классифицировать обращение, вытащить из текста дату, решить, про что вопрос. Здесь модели нужно одновременно держать правила, не называть множители, выбирать ключи из списка и не ломать JSON. На таких задачах я предпочитаю полную модель, а сэкономить пытаюсь на объёме текста. Если упираетесь в бюджет, попробуйте разделить работу: лёгкая модель решает, нужен ли вообще дорогой вызов, полная отвечает только там, где это нужно.

Логи без персональных данных

Тексты разговоров не попадают ни в БД, ни в логи. В логах только то, что нужно для разбора сбоев: какой провайдер, код ответа, сколько токенов. Тело ответа об ошибке обрезается до 200 символов, и из него вырезаются ключ и токен, потому что сервисы иногда повторяют их в сообщении об ошибке. Из сетевых исключений пишется только имя класса: в тексте requests-исключения бывает URL с параметрами.

except requests.RequestException as e:
    logger.warning("YandexGPT недоступен: %s", type(e).__name__)
    raise LLMUnavailable("Сетевая ошибка") from None

Контакты помощник не спрашивает: в промпте это отдельное правило, а для телефона и почты есть форма с согласием. Это проще, чем потом чистить персональные данные из сообщений.

Как я это тестирую

Тестов на помощника около восьмисот строк, и почти все они не ходят в сеть. Есть три слоя.

  • Провайдеры. requests.post подменяется моком: успех, 401, 500, таймаут, битое тело, ответ без usage. Отдельные тесты проверяют, что ключ и токен не попадают в логи при ошибках.
  • Разбор ответа. Набор «плохих» ответов модели: JSON в обёртке, текст без JSON, неизвестные ключи, два типа проекта, options строкой вместо списка, open_spec: "yes". Для каждого зафиксирован ожидаемый результат.
  • API целиком. llm.complete подменяется, и проверяется всё вокруг: 429, суточный лимит, бюджет, выключатель, валидация тела, то, что смету считает сервер.
def test_two_types_keep_first(self):
    out = parse_model_output(json.dumps({
        "reply": "Ок", "options": ["f-crm", "t-bot", "t-corp", "d-fast", "d-urgent"]}))
    self.assertEqual(out["options"], ["f-crm", "t-bot", "d-fast"])

Чего тесты не ловят, так это качества текста. Первую версию промпта я проверял вживую и нашёл три проблемы: канцелярит, модель называла множители срока вместо цен и слишком рано открывала лист ТЗ, уже на первый вопрос о цене. Всё это исправилось правками промпта, а тест test_rules_kept теперь следит, чтобы эти правила не потерялись при следующем сокращении. Следующий шаг, который я считаю правильным: небольшой набор эталонных диалогов, которые прогоняются против живой модели вручную перед сменой модели или промпта.

Передача человеку

В панели помощника есть кнопка «Позвать Евгения». Она открывает обычный живой чат на Django Channels, который у сайта был и раньше. Первым сообщением уходит сводка: до десяти последних реплик с помощником (в пределах 1800 символов, старые отбрасываются первыми), выбранная боль и отмеченные пункты. В Telegram это сообщение приходит с отдельным заголовком, чтобы я сразу видел, что меня позвали из помощника. Пока идёт живой разговор, запросы к модели не отправляются. Кнопка работает и при недоступном помощнике, потому что не зависит от провайдера.

Что получилось

Если свести всё к нескольким правилам: модель пишет текст и выбирает из списка, а цену, проверку и лимиты держит код. Каждый внешний вызов может упасть, и у каждого падения есть понятный путь для посетителя. Деньги ограничены в трёх местах, и при сомнении система выбирает «не тратить». И у человека всегда есть кнопка, которая ведёт к живому разработчику, а не к ещё одному ответу модели.

Все статьи