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

RAG-ответ без источников (потеря grounding)

RAG-пайплайн находит релевантные чанки, кладёт их в контекст — а модель отвечает мимо: без ссылок, без [source]-маркеров, сочиняя поверх пустоты. Retrieval при этом отработал: чанки были. Ответ выглядит уверенно и правдоподобно, поэтому просадка тихая — пользователь просто получает «факты», которых нет в вашей базе.

Это отдельный от галлюцинаций сигнал: там модель выдумывает при пустом контексте, здесь — игнорирует непустой. Причина почти всегда в трёх местах: сломался retriever, сменилась версия эмбеддингов и старые чанки больше не находятся, или кто-то отредактировал системный промпт и выкинул строку «ссылайся на источники».

1. Дешёвая структурная проверка

Заголовок раздела «1. Дешёвая структурная проверка»

Самое дешёвое — без единого лишнего вызова модели. Если retrieval вернул чанки, а ответ не цитирует ни одного их id (нет [source], [1], doc_id в тексте), считаем ответ «ungrounded». Копим долю таких в скользящем окне и шлём push, когда она пробивает порог.

import os, re, json, time, statistics, requests


WIN = []                       # 1 = ungrounded, 0 = ok
FLAG = "/tmp/citation_spike.flag"
THRESHOLD = 0.30               # >30% ответов без источников за окно
MIN_WINDOW = 40


def is_ungrounded(answer_text, retrieved_ids):
    # retrieval что-то нашёл, но ответ не сослался ни на один id?
    if not retrieved_ids:
        return False           # пустой retrieval — это другой сценарий
    cited = set(re.findall(r"\[(?:source|src|doc)[:\s]*([\w\-]+)\]", answer_text))
    cited |= set(re.findall(r"\[(\d+)\]", answer_text))   # сноски вида [1]
    return len(cited & {str(i) for i in retrieved_ids}) == 0


def observe(answer_text, retrieved_ids):
    WIN.append(1 if is_ungrounded(answer_text, retrieved_ids) else 0)
    if len(WIN) > 200:
        WIN.pop(0)
    if len(WIN) < MIN_WINDOW:
        return
    rate = statistics.mean(WIN)
    if rate >= THRESHOLD and not os.path.exists(FLAG):
        open(FLAG, "w").write(str(time.time()))
        notify("📄❓ Ответы без источников",
               f"{int(rate*100)}% ответов не цитируют найденные чанки "
               f"(окно {len(WIN)}). Проверьте retriever / промпт.",
               priority=8)
    elif rate < THRESHOLD * 0.5 and os.path.exists(FLAG):
        os.remove(FLAG)        # всплеск прошёл — снимаем флаг


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)

Флаг-файл гарантирует один push на всплеск — иначе на каждом ответе в окне полетит новый алёрт. Флаг снимается, когда доля вернулась в норму.

2. Grounding-оценка судьёй-LLM

Заголовок раздела «2. Grounding-оценка судьёй-LLM»

Структурная проверка ловит «нет маркеров вообще», но пропускает ответ, который цитирует источник формально, а по сути — пересказ из головы модели. Точнее считать grounding-score: прослеживаются ли утверждения ответа к найденному контексту (0..1). Это дороже, поэтому гоняем на сэмпле (например каждый 10-й ответ) и следим за средним по окну.

import anthropic


client = anthropic.Anthropic()
GWIN = []
GTHRESHOLD = 0.6               # средний grounding упал ниже 0.6 → алёрт


def grounding_score(answer, context_chunks):
    ctx = "\n---\n".join(context_chunks)
    judge = client.messages.create(
        model="claude-haiku-4-5",
        max_tokens=8,
        messages=[{"role": "user", "content":
            f"Контекст:\n{ctx}\n\nОтвет:\n{answer}\n\n"
            "Какая доля утверждений ответа прямо следует из контекста? "
            "Верни только число от 0.0 до 1.0."}],
    )
    try:
        return max(0.0, min(1.0, float(judge.content[0].text.strip())))
    except ValueError:
        return 0.0


def observe_grounding(answer, context_chunks):
    GWIN.append(grounding_score(answer, context_chunks))
    if len(GWIN) > 100:
        GWIN.pop(0)
    if len(GWIN) >= 20:
        mean = statistics.mean(GWIN)
        if mean < GTHRESHOLD and not os.path.exists("/tmp/grounding.flag"):
            open("/tmp/grounding.flag", "w").write("1")
            notify("🧷 Grounding упал",
                   f"Средний grounding {mean:.2f} за {len(GWIN)} ответов "
                   f"(порог {GTHRESHOLD}). Ответы плохо опираются на контекст.",
                   priority=8)

Модель для судьи держите дешёвую — задача бинарно-простая, платить за флагман тут незачем.

3. Жёсткий per-answer алёрт для high-stakes

Заголовок раздела «3. Жёсткий per-answer алёрт для high-stakes»

Для ответов, где цена ошибки высокая — юридические, медицинские, финансовые — не ждём накопления статистики. Один ungrounded-ответ в такой категории — сразу громкий push, чтобы человек посмотрел до того, как ответ уйдёт пользователю.

HIGH_STAKES = {"legal", "medical", "finance"}


def guard_high_stakes(answer_text, retrieved_ids, category):
    if category in HIGH_STAKES and is_ungrounded(answer_text, retrieved_ids):
        notify("🚨 High-stakes ответ без источника",
               f"Категория «{category}»: ответ без цитат при "
               f"{len(retrieved_ids)} найденных чанках. Нужен ревью.",
               priority=10)
        return False           # заблокируйте выдачу до проверки
    return True

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

Заголовок раздела «Что положить в push»
  • доля ungrounded / средний grounding-score за окно;
  • сколько чанков вернул retrieval (чтобы отличить «пусто» от «проигнорил»);
  • категория, если это high-stakes;
  • намёк на причину: сменилась версия эмбеддингов?, правили системный промпт?, retriever отдаёт пусто?.

Если всплеск совпал по времени с деплоем — почти наверняка виноват апдейт эмбеддингов или правка промпта. Разложите проверки по слоям и запускайте счётчики в общем хранилище (Redis), чтобы окно переживало рестарты и охватывало все инстансы.

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

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