Перейти к содержимому
Notifly Notifly

Ваше приложение выдало токсичный ответ

Одно дело — токсичный ввод от пользователя, его ловит safety-trigger. Совсем другое — когда токсичный, оскорбительный или небезопасный текст выдало ваше собственное приложение. Это уже не про злоумышленника на входе, а про репутационный и юридический инцидент: ваш сервис под вашим именем отправил пользователю то, чего отправлять не должен.

Про такое надо узнавать немедленно и громко — не из скриншота в соцсетях через день. Гоняем каждый ответ модели через модератор перед отдачей пользователю и на флаге шлём push с высоким приоритетом.

1. Модерация вывода перед отдачей + HIGH-priority алёрт

Заголовок раздела «1. Модерация вывода перед отдачей + HIGH-priority алёрт»

Классификатор может быть любым: OpenAI moderation endpoint, локальная модель (Llama Guard, detoxify) или дешёвый LLM-судья. Важно прогнать свой вывод до return пользователю. На флаг — редактированный сэмпл, сработавшие категории и request id.

import os, re, requests
from openai import OpenAI


client = OpenAI()


def moderate(text: str):
    # Возвращает (flagged: bool, categories: list[str]).
    r = client.moderations.create(model="omni-moderation-latest", input=text)
    res = r.results[0]
    cats = [c for c, on in res.categories.model_dump().items() if on]
    return res.flagged, cats


def redact(text: str, limit: int = 300) -> str:
    # маскируем PII и режем длину — сам токсичный текст в push не тащим целиком
    s = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "<email>", text)
    s = re.sub(r"\+?\d[\d\s()-]{8,}\d", "<phone>", s)
    return s[:limit] + ("…" if len(s) > limit else "")


def guarded_answer(user_input: str, request_id: str) -> str:
    answer = my_app.answer(user_input)          # ваш реальный пайплайн
    flagged, cats = moderate(answer)
    if flagged:
        notify("🚨 Приложение выдало токсичный ответ",
               f"request_id: {request_id}\n"
               f"Категории: {', '.join(cats) or '—'}\n\n"
               f"Фрагмент (редактир.):\n{redact(answer)}",
               priority=9)
        return SAFE_FALLBACK                    # не отдаём пользователю flagged-текст
    return answer


def notify(title, message, priority):
    requests.post(f"{os.environ['NOTIFLY_URL']}/message",
                  params={"token": os.environ["NOTIFLY_TOKEN"]},
                  json={"title": title, "message": message, "priority": priority},
                  timeout=5)

priority: 9 — почти самый громкий push: это инцидент, а не FYI. И обратите внимание: на флаг мы не только алёртим, но и подменяем ответ на безопасный fallback, чтобы токсичный текст вообще не дошёл до пользователя.

Нет доступа к OpenAI — тот же интерфейс даёт локальный detoxify или LLM-судья на дешёвой модели:

def moderate_via_judge(text: str):
    judge = anthropic.Anthropic().messages.create(
        model="claude-haiku-4-5", max_tokens=16,
        messages=[{"role": "user", "content":
            "Ты модератор. Есть ли в тексте оскорбления, угрозы, hate speech "
            "или небезопасный контент? Ответь строго: FLAG:<категория> или OK.\n\n"
            f"Текст:\n{text}"}],
    )
    verdict = judge.content[0].text.strip()
    if verdict.upper().startswith("FLAG"):
        return True, [verdict.split(":", 1)[-1].strip()]
    return False, []

2. Rate-view: разовый flag против всплеска

Заголовок раздела «2. Rate-view: разовый flag против всплеска»

Один flagged-ответ на миллион — редкая осечка модели. Но если флаги пошли пачкой, это уже система: кто-то нашёл jailbreak, сломался фильтр в промпте или подменилась модель. Считаем флаги в окне и эскалируем приоритет, если это всплеск, а не единичный случай:

import time
from collections import deque


WINDOW_SEC = 600
_flags = deque()          # timestamps сработавших флагов


def record_flag(request_id: str, cats: list[str], sample: str):
    now = time.time()
    _flags.append(now)
    while _flags and now - _flags[0] > WINDOW_SEC:
        _flags.popleft()


    n = len(_flags)
    prio = 10 if n >= 5 else 9      # всплеск → максимальный приоритет
    head = ("🚨🔥 Всплеск токсичных ответов" if n >= 5
            else "🚨 Приложение выдало токсичный ответ")
    notify(head,
           f"{n} флагов за {WINDOW_SEC // 60} мин.\n"
           f"request_id: {request_id}\nКатегории: {', '.join(cats) or '—'}\n\n"
           f"Фрагмент (редактир.):\n{redact(sample)}",
           priority=prio)

Порог n >= 5 подкрутите под свой трафик. Логика та же, что в safety-trigger, но здесь ставки выше: токсичит ваш сервис, поэтому даже одиночный случай идёт с priority: 9, а всплеск — с 10.

3. Не утечь токсичным контентом в сам push

Заголовок раздела «3. Не утечь токсичным контентом в сам push»

Push уходит на ваш телефон, но текст всё равно живёт в истории Notifly и в логах. Не тащите flagged-ответ целиком:

  • режьте длину — 300 символов хватает, чтобы понять суть;
  • маскируйте PII — email, телефоны, номера (см. redact выше);
  • если хочется совсем чисто — шлите в push только категории + request id, а сам сэмпл кладите в защищённый лог и передавайте ссылкой через msgextras, чтобы открыть его отдельно.
notify("🚨 Приложение выдало токсичный ответ",
       f"request_id: {request_id}\nКатегории: {', '.join(cats)}\n"
       "Полный (редактир.) текст — по ссылке в логе.",
       priority=9)
# URL полного лога прикрепляем через msgextras: /docs/msgextras/

Что положить в push

Заголовок раздела «Что положить в push»
  • request_id / session id — чтобы найти инцидент в логах за секунды;
  • сработавшие категории модератора (harassment, hate, violence…);
  • редактированный и усечённый сэмпл — ровно чтобы понять серьёзность;
  • признак всплеска (5 флагов за 10 мин) и, если знаете, вероятную причину.

Связанные рецепты

Заголовок раздела «Связанные рецепты»