Сбои ингеста документов в базу знаний

Пайплайн ингеста тащит PDF, HTML и docx в базу знаний RAG: парсит, режет на чанки, считает embeddings, делает upsert. Любой шаг спотыкается тихо — битый PDF, пустой OCR, таймаут embeddings — и документ просто не попадает в индекс. Пользователь потом получает «в базе знаний нет информации», хотя файл вы загружали. Ловим сбой в момент ингеста, с именем файла.

1. Ингест одного документа с алёртом

import os, requests

def ingest(path: str):
    try:
        text = parse_document(path)        # PDF/HTML/docx → текст
        if len(text.strip()) < 50:
            raise ValueError("извлечено < 50 символов (битый файл или пустой OCR)")
        chunks = chunk(text)
        vectors = embed_batch(chunks)      # может упасть по rate-limit / таймауту
        upsert(path, chunks, vectors)
        return True
    except Exception as e:
        notify("📄❌ Сбой ингеста",
               f"Файл: {os.path.basename(path)}\n"
               f"Этап: {_stage(e)}\n"
               f"Ошибка: {type(e).__name__}: {e}",
               priority=7)
        return False

def _stage(e):
    name = type(e).__name__.lower()
    if "pdf" in name or "parse" in name: return "парсинг"
    if "rate" in name or "timeout" in name: return "embeddings"
    return "неизвестен"

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)

2. Батч-дайджест вместо шторма пушей

Загрузили 500 файлов, 30 упали — 30 отдельных пушей никто читать не будет. Копим сбои по ходу батча и шлём один сводный push в конце:

def ingest_batch(paths):
    failures = []
    for p in paths:
        try:
            ingest_or_raise(p)
        except Exception as e:
            failures.append((os.path.basename(p), f"{type(e).__name__}"))

    if failures:
        head = "\n".join(f"• {name} — {err}" for name, err in failures[:15])
        more = f"\n… и ещё {len(failures) - 15}" if len(failures) > 15 else ""
        notify(f"📄 Ингест: {len(failures)}/{len(paths)} с ошибками",
               head + more,
               priority=8 if len(failures) > len(paths) * 0.2 else 6)
    return failures

Приоритет растёт, если доля сбоев большая: пять битых файлов из тысячи — рутина, а двести из тысячи — сломался парсер или упал embeddings-сервис.

3. Дедлеттер и повторный прогон

Складывайте упавшие пути в отдельный файл-очередь, чтобы перегнать их позже, не перезапуская весь батч:

def ingest_batch_with_dlq(paths, dlq="/var/lib/rag/failed.txt"):
    failures = ingest_batch(paths)
    if failures:
        with open(dlq, "a") as f:
            for name, _ in failures:
                f.write(name + "\n")

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

• имя файла (и путь / источник);
• этап, где споткнулось: парсинг, чанкинг, embeddings, upsert;
• класс ошибки;
• для батча — доля сбоев и путь к дедлеттер-очереди.

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

• Перестройка индекса завершена — сводка по всему прогону.
• Всплеск стоимости embeddings — ретраи ингеста жгут токены.
• Устаревший индекс знаний — если ингест тихо встал совсем.