Запрос заблокирован content-policy провайдера

Отличие от safety-триггера: там модель ответила отказом, здесь провайдер вообще не выполнил запрос — вернул ошибку модерации (content_filter, responsible_ai_policy_violation, content_policy_violation). Для вас это выглядит как 400/403 в проде, а дальше пользователь видит пустой экран или generic-ошибку.

Такие блокировки почти всегда — про конкретный промпт или конкретный пользовательский ввод. Push с усечённым и обезличенным примером даёт вам ровно то, что нужно, чтобы починить, не роясь в логах.

Ловим ошибку модерации в коде

Ключевой момент — отличить content-policy ошибку от прочих 400. У каждого провайдера свой код, собираем их в один набор:

import os, re, requests

NOTIFLY_URL   = os.environ["NOTIFLY_URL"]
NOTIFLY_TOKEN = os.environ["NOTIFLY_TOKEN"]

POLICY_CODES = {
    "content_filter",                    # OpenAI / Azure
    "content_policy_violation",          # OpenAI images
    "responsible_ai_policy_violation",   # Azure
    "safety",                            # Vertex / Gemini
}

def notify(title, msg, prio):
    requests.post(f"{NOTIFLY_URL}/message",
                  params={"token": NOTIFLY_TOKEN},
                  json={"title": title, "message": msg, "priority": prio},
                  timeout=5)

# грубое обезличивание: убираем e-mail, телефоны, длинные цифровые последовательности
def redact(text: str, limit=400) -> str:
    text = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "[email]", text)
    text = re.sub(r"\+?\d[\d\s()-]{7,}\d", "[phone]", text)
    text = re.sub(r"\b\d{6,}\b", "[num]", text)
    return text[:limit]

Теперь оборачиваем вызов провайдера:

import openai

def ask(user_input: str):
    try:
        return client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": user_input}],
        )
    except openai.BadRequestError as e:
        body = getattr(e, "body", {}) or {}
        code = (body.get("code")
                or body.get("error", {}).get("code")
                or "")
        if code in POLICY_CODES:
            notify("🚫 Content-policy: запрос заблокирован",
                   f"Provider: openai\nCode: {code}\n\n"
                   f"Ввод (обезличен):\n{redact(user_input)}",
                   prio=7)
        raise

priority=7 — это не инцидент безопасности, а сигнал «почини промпт/вход», поэтому нормальная громкость, а не «десятка».

Дедупликация: один и тот же промпт бьётся сотни раз

Если сломан ваш системный промпт, а не разовый пользовательский ввод, ошибка полетит на каждом запросе. Слать нужно первый случай и потом периодический агрегат — общий шаблон:

import time
state = {"first_ts": 0, "count": 0}
WINDOW = 3600

def maybe_alert(code, sample):
    now = time.time()
    if now - state["first_ts"] > WINDOW:
        state.update(first_ts=now, count=0)
        notify("🚫 Content-policy (первый за час)",
               f"Code: {code}\n{sample}", prio=7)
    state["count"] += 1
    if state["count"] in (25, 250):
        notify(f"🚫 Content-policy ×{state['count']}",
               "Похоже, задет ваш системный промпт или шаблон — а не разовый ввод.",
               prio=9)

Другие провайдеры

• Anthropic — stop_reason == "refusal" в ответе (запрос выполнился, но модель отказалась), либо 400 с error.type == "invalid_request_error".
• Vertex / Gemini — в ответе есть promptFeedback.blockReason и safetyRatings; блокировка не кидает исключение, её надо проверять в теле.
• Провайдеры без webhook — если модерация приходит только письмом, заверните почтовый ящик в Email Inbox и алёртите на письма.

Что положить в текст алёрта

• провайдер и точный код политики;
• усечённый и обезличенный ввод (никакого сырого PII в push!);
• какая категория сработала, если провайдер её отдаёт (violence / hate / …);
• session/request id, чтобы найти полный лог в S3/Sentry.