Die Dokumenten-Pipeline: Prefect, Airflow oder Data Factory

TL;DR

Der Orchestrierer ist die unwichtigste Entscheidung an dieser Stelle. Wichtiger ist, dass die Pipeline inkrementell arbeitet, Fehler pro Dokument isoliert und den Verarbeitungsstand nachvollziehbar festhält. Wer unter 50.000 Dokumenten bleibt, kommt lange mit einem Skript und einem Zeitplan aus.

Was die Pipeline tatsächlich tun muss

Bevor wir Werkzeuge vergleichen, die Aufgabenliste — denn sie erklärt, warum die meisten Eigenbauten nach einem halben Jahr an ihre Grenzen kommen.

1. Quellen abfragen und feststellen, was sich seit dem letzten Lauf geändert hat
2. Dokumente holen und parsen — PDF, Word, Excel, gescannte Seiten
3. In Abschnitte teilen, möglichst an der Dokumentstruktur statt nach Zeichenzahl
4. Berechtigungen mitziehen aus der Quelle
5. Einbettungen berechnen, in Stapeln, mit Wiederholung bei Fehlern
6. In den Index schreiben und gelöschte Dokumente entfernen
7. Festhalten, was verarbeitet wurde, was fehlschlug und warum

Punkt 7 ist der, den Eigenbauten weglassen, und Punkt 6 der zweite Teil davon, den fast alle vergessen: Ein Dokument, das in der Quelle gelöscht wurde, bleibt im Index und wird weiter als Antwortgrundlage verwendet. Bei einer zurückgezogenen Arbeitsanweisung ist das ein echtes Problem.

Erst die unbequeme Frage: brauchen Sie überhaupt einen Orchestrierer?

Bei 20.000 Dokumenten aus zwei Quellen, die sich täglich um ein paar Dutzend ändern, ist die ehrliche Antwort: nein.

Ein Python-Skript mit einer Zustandstabelle in PostgreSQL, ausgelöst per systemd-Timer oder Cron, erledigt das. Sie brauchen keine Weboberfläche, um zu sehen, was gelaufen ist — Sie brauchen eine Tabelle mit einer Zeile pro Dokument:

CREATE TABLE ingestion_status (
    quelle        text NOT NULL,
    dokument_id   text NOT NULL,
    hash          text NOT NULL,      -- erkennt echte Änderungen
    verarbeitet   timestamptz,
    abschnitte    int,
    fehler        text,
    PRIMARY KEY (quelle, dokument_id)
);

Der hash ist der Kern der inkrementellen Verarbeitung: Nur wenn er sich geändert hat, wird neu verarbeitet. Das spart bei einem Nachtlauf über 30.000 Dokumente den Großteil der Einbettungskosten.

Wir empfehlen diesen Weg für den Anfang ausdrücklich. Ein Orchestrierer, den niemand betreiben kann, ist schlechter als ein Skript, das alle verstehen.

Wann es sich ändert

Drei Auslöser, ab denen ein Orchestrierer sinnvoll wird:

• Mehr als vier oder fünf Quellen mit unterschiedlichen Zeitplänen und Abhängigkeiten
• Verarbeitungszeiten über zwei Stunden, bei denen ein Neustart nicht von vorn beginnen darf
• Mehrere Personen, die Läufe anstoßen und Fehler nachverfolgen müssen

Die drei im Vergleich

Prefect passt am besten zu diesem Aufgabentyp, weil Dokumentenverarbeitung dynamisch ist: Sie wissen vor dem Lauf nicht, wie viele Dokumente sich geändert haben. In Prefect ist das eine Schleife über eine Liste, die zur Laufzeit entsteht.

from prefect import flow, task

@task(retries=3, retry_delay_seconds=30)
def dokument_verarbeiten(dok: dict) -> int:
    """Ein Dokument: parsen, teilen, einbetten, schreiben."""
    text = parse(dok["pfad"])
    abschnitte = teile(text, dok["berechtigungen"])
    schreibe_index(einbetten(abschnitte))
    return len(abschnitte)

@flow(name="rag-ingestion")
def ingestion():
    geaendert = finde_geaenderte_dokumente()      # Liste zur Laufzeit
    # Jedes Dokument scheitert für sich — ein kaputtes PDF
    # stoppt nicht den ganzen Lauf
    ergebnisse = dokument_verarbeiten.map(geaendert)
    entferne_geloeschte()
    return sum(r for r in ergebnisse if r)

Die Zeile, auf die es ankommt, ist retries=3 auf Aufgabenebene. Ein Dokument, das an einem Zeichenfehler scheitert, wird dreimal versucht und dann als fehlerhaft vermerkt — der Lauf geht weiter.

Airflow ist die richtige Wahl, wenn es bereits im Haus läuft. Es hat den größten Fundus an Anbindern und die meiste Betriebserfahrung im Markt. Für eine reine RAG-Pipeline neu einzuführen, ist es überdimensioniert: Sie bekommen Scheduler, Arbeiterprozesse und eine Metadatenbank für eine Aufgabe, die im Kern eine Schleife ist.

Azure Data Factory lohnt sich, wenn Ihre Quellen ohnehin in Azure liegen und die Verarbeitung wenig Eigenlogik braucht. Die Anbindungen an Blob Storage, SQL und SharePoint sind fertig. Sobald Sie eigene Python-Bibliotheken für das Parsen brauchen — und das brauchen Sie fast immer —, hängen Sie Azure Functions oder Batch daran, und der Vorteil der grafischen Oberfläche ist weg.

Der Teil, der wirklich über die Qualität entscheidet

Nicht der Orchestrierer, sondern die Abschnittsbildung. Zwei Punkte, die in jeder Installation den Unterschied machen:

Schneiden Sie an der Struktur, nicht an der Zeichenzahl. Eine Arbeitsanweisung mit Überschriften, Aufzählungen und einer Tabelle wird durch einen Schnitt alle 512 Zeichen zerstört. Ein Abschnitt, der mitten in einer Tabelle beginnt, ist als Antwortgrundlage wertlos. Nutzen Sie Parser, die Struktur erhalten, und schneiden Sie an Überschriften.

Geben Sie jedem Abschnitt Kontext mit. Ein Abschnitt, der nur "Alle 5 Jahre, bei Behältern über 1000 Liter alle 3 Jahre" enthält, ist ohne die Überschrift "Prüfintervalle Druckbehälter" nicht auffindbar. Stellen Sie die Überschriftenkette dem Abschnittstext voran, bevor Sie einbetten. Dieser eine Kniff hebt die Wiederfindung in unseren Messungen regelmäßig um 10 bis 20 Punkte — mehr als jeder Wechsel des Einbettungsmodells. Wie Sie das nachmessen, steht im Beitrag zur Messung der RAG-Qualität.

Welche Parser dafür taugen, behandelt der Vergleich von Azure Document Intelligence und offenen Parsern.

Häufig gestellte Fragen

Brauche ich für eine RAG-Pipeline einen Orchestrierer wie Airflow?

Unterhalb von etwa 50.000 Dokumenten aus wenigen Quellen nicht. Ein Python-Skript mit einer Zustandstabelle und einem Zeitplan erfüllt dieselbe Aufgabe und ist von jedem im Team wartbar. Sinnvoll wird ein Orchestrierer bei vielen Quellen mit Abhängigkeiten, bei Laufzeiten über zwei Stunden oder wenn mehrere Personen Läufe verwalten müssen.

Wie erkenne ich, welche Dokumente sich geändert haben?

Über einen Hashwert des Inhalts, der pro Dokument in einer Zustandstabelle festgehalten wird. Zeitstempel allein sind unzuverlässig, weil viele Systeme sie bei rein technischen Vorgängen aktualisieren — Sie würden dann unverändert gebliebene Dokumente erneut einbetten und dafür bezahlen. Der Hash erkennt echte inhaltliche Änderungen.

Was passiert mit gelöschten Dokumenten im Index?

Nichts, wenn Sie es nicht ausdrücklich behandeln — und genau das ist ein häufiges Problem. Eine zurückgezogene Arbeitsanweisung bleibt im Index und dient weiter als Antwortgrundlage. Die Pipeline muss deshalb nach jedem Lauf abgleichen, welche Kennungen in der Quelle nicht mehr existieren, und die zugehörigen Abschnitte entfernen.

Wie oft sollte die Pipeline laufen?

Für die meisten Bestände reicht ein nächtlicher Lauf. Häufiger wird es nötig, wenn Dokumente unmittelbar nach Freigabe auffindbar sein müssen — etwa bei tagesaktuellen Preislisten. Dann ist ein zusätzlicher Lauf alle ein bis zwei Stunden für die betroffene Quelle praktikabler, als die gesamte Pipeline zu beschleunigen.

Was kostet die Einbettung eines Dokumentenbestands?

Bei 30.000 Dokumenten mit rund einer Million Abschnitten und einem Cloud-Einbettungsmodell liegt der erste vollständige Durchlauf im niedrigen dreistelligen Eurobereich. Danach fallen nur noch die geänderten Dokumente an, also ein Bruchteil. Mit einem lokal betriebenen Einbettungsmodell entfallen diese Kosten vollständig — dafür brauchen Sie GPU-Kapazität während des Laufs.

Der nächste Schritt

Bauen Sie zuerst die Zustandstabelle, nicht die Pipeline. Wenn Sie für jedes Dokument wissen, wann es zuletzt verarbeitet wurde, mit welchem Hash und ob es dabei einen Fehler gab, haben Sie das eigentliche Problem gelöst — der Rest ist eine Schleife. Beim Zuschnitt helfen wir gern.