Document ingestion failures into the knowledge base

The ingestion pipeline pulls PDFs, HTML and docx into the RAG knowledge base: parse, chunk, embed, upsert. Any step trips silently — a corrupt PDF, empty OCR, an embeddings timeout — and the document just never makes it into the index. The user later gets “there’s no information in the knowledge base”, even though you did upload the file. Catch the failure at ingest time, with the file name.

1. Single-document ingest with an alert

import os, requests

def ingest(path: str):
    try:
        text = parse_document(path)        # PDF/HTML/docx → text
        if len(text.strip()) < 50:
            raise ValueError("extracted < 50 chars (corrupt file or empty OCR)")
        chunks = chunk(text)
        vectors = embed_batch(chunks)      # may fail on rate-limit / timeout
        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. A batch digest instead of a push storm

You uploaded 500 files, 30 failed — nobody will read 30 separate pushes. Accumulate failures as the batch runs and send one summary push at the end:

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

The priority rises when the failure share is large: five corrupt files out of a thousand is routine, while two hundred out of a thousand means the parser broke or the embeddings service went down.

3. A dead-letter queue and a rerun

Put failed paths into a separate queue file so you can re-run them later without restarting the whole batch:

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")

What to put in the alert text

• the file name (and path / source);
• the stage where it tripped: parsing, chunking, embeddings, upsert;
• the error class;
• for a batch — the failure share and the dead-letter queue path.

Related recipes

• Vector index rebuild finished — a summary of the whole run.
• Embedding cost spike — ingest retries burn tokens.
• Stale knowledge index — if ingestion silently stalled entirely.