Docling-Pipelines: Erstellung von RAG-Datenpipelines auf Unternehmensebene
Wie docling-pipelines zerbrechliche Glue-Skripte durch komponierbare Operatoren und JSON-Flüsse ersetzt: Eingabe, Extraktion, Qualitätskontrollen, Embeddings und Vektorlagern.
Quellen: Introducing Docling Pipelines: Building Enterprise-Grade RAG Data Pipelines
Wo Glue-Code versagt
Stellen Sie sich einen gemeinsamen Speicher mit 10.000 PDF-Dateien, eine SharePoint-Seite voller Word-Dateien sowie eine Anhäufung von aufgezeichneten Kundengesprächen vor. Ein Produktteam möchte alles davon in einem Vektorlager haben – suchbar und in Blöcken strukturiert, wobei personenbezogene Daten entfernt und Duplikate beseitigt werden sollen.
Die übliche ingenieurtechnische Lösung besteht aus einem Skript, anschließend einem weiteren Skript. Einige Bibliotheken werden hinzugezogen, ein Pandas-Notebook verbindet sie miteinander, ein Cron-Job führt alles jede Nacht erneut aus – und das System ist fertig. Es funktioniert, zumindest für eine Weile.
Drei Monate später fragt jemand, warum neue Dokumente fehlen. Die Spur führt zu einer Änderung des Schemas in einer nachgelagerten Spalte, wodurch das Skript diese Zeilen stillschweigend verworfen hatte. Niemand bemerkte es, weil nichts Beobachtbares vorlag. Nach der Behebung wird der Vorgang erneut ausgeführt, und dabei wird klar, dass es keine Aufzeichnung darüber gibt, welche Dokumente bereits bearbeitet wurden – daher werden alle 10.000 Dokumente erneut verarbeitet.
So verläuft das typische Schicksal von Verbindungsmechanismen in Dokumenten-Pipelines. Sie beginnen einfach und verschlechtern sich langsam, bevor sie plötzlich zusammenbrechen. Versuche, das Problem zu umgehen, werden nachträglich hinzugefügt, die Qualitätssicherung kommt erst als Nachgedanke hinzu, und es gibt keine Überwachung der Abhängigkeiten zwischen den Schritten – daher hängt der Erfolg davon ab, dass die Skripte zufällig in der richtigen Reihenfolge ausgeführt werden. Ohne eine stabile Dokumentidentität sind schrittweise Aktualisierungen schwierig. Was übrig bleibt, ist ein Wirrwarr, den nur sein Ersteller debuggen kann – und selbst der Ersteller ist sich nicht mehr sicher.
Was docling-Pipelines sind
Docling-Pipelines ist ein modulares, auf Operatoren basierendes Framework für die Dokumentverarbeitung, das dazu dient, vollständige RAG-Datenpipelines zu erstellen. Sein Anwendungsbereich umfasst den gesamten Lebenszyklus – beginnend mit der Aufnahme von Inhalten aus lokalen oder Cloud-Quellen, über die Extraktion von Text und Entitäten, das Qualitätsfiltern, das Aufteilen in Blöcke sowie das Erstellen von Embeddings, bis hin zur Speicherung in einer Vektordatenbank. Das ist der Standardweg, und die meisten Produktionspipelines sind Variationen davon.
Es wird als gewöhnliches Python-Paket installiert:
bash
pip install docling-pipelines
Es läuft überall dort, wo Python läuft. Für die grundlegende Nutzung ist keine externe Infrastruktur erforderlich: weder Kafka noch ein Spark-Cluster, nichts, was eingerichtet werden muss. Ein Pipeline wird in einer JSON-Datei beschrieben, die CLI wird auf diese Datei ausgerichtet, und der Ausführungsprozess beginnt.
Es gibt drei Möglichkeiten, damit zu arbeiten. Die CLI eignet sich für einmalige Ausführungen und lokale Entwicklung. Die Python-Bibliothek, die auf DocpipeFlowManager basiert, dient dazu, Pipelines aus Anwendungscode oder Notebooks auszulösen. Die REST API ist für Pipelines gedacht, die als gemeinsamer Dienst von mehreren Teams genutzt werden. Alle drei Ansätze werden unten erläutert.
Das Projekt ist Open Source, und seine erste Veröffentlichung, v0.1.0, erschien im Juli 2025.
Die Kernidee: Operator und Flüsse
Jeder, der mit diesem Framework neu ist, sollte hier am meisten Zeit investieren, da der Rest erst verständlich wird, wenn dies klar ist.
Der Operator ist die grundlegende Einheit. Ein Operator ist eine Python-Klasse, die eine Tabelle mit Dokumenten entgegennimmt und möglicherweise transformierte Tabelle zurückgibt. Jede Funktionalität im Framework – sei es die Aufnahme, Extraktion, Spracherkennung, Aufteilung in Blöcke oder Erstellung von Embeddings – ist ein Operator.
Die Anforderungen sind gering:
short_name: Der Zeichenstrom, den JSON verwendet, um auf den Operator zu verweisentransform(table): Nimmt einepa.Tableentgegen und gibt eine Liste von Tabellen zusammen mit einem Metadaten-Dictionary zurückget_metadata(): Eine@staticmethod, die die Metadaten des Operators zurückgibt, damit er entdeckt werden kannget_required_features(): Gibt an, welche Spalten der Operator zur Eingabe benötigtvalidate(): Ein optioneller Hook, der bei der Startzeit die Konfiguration überprüft
Ein minimales Skelett sieht so aus:
from docpipe.core.operators.abstract_operator import AbstractOperator, OperatorCategory
import pyarrow as pa
class MyOperator(AbstractOperator):
short_name = "my_operator"
category = OperatorCategory.Quality
def __init__(self, config: dict) -> None:
super().__init__(config)
def transform(self, table: pa.Table) -> tuple[list[pa.Table], dict]:
metadata = self.create_base_metadata(total_docs_count=len(table))
# … process table …
return [table], metadata
@staticmethod
def get_metadata() -> dict:
return {"short_name": "my_operator", "description": "…"}
Durch die Implementierung dieser Methoden entsteht ein erstklassiges Komponente, das mit jedem anderen Operator im Ökosystem kombiniert werden kann.
PyArrow-Tabellen tragen die Daten. Jeder Operator erhält eine pa.Table und gibt eine zurück. Das ist eine bewusste Entscheidung, denn PyArrows spaltenbasiertes Format ist effizient und typisiert und ermöglicht den Datentransfer zwischen den Schritten ohne Kopieren. Rohstrings und Python-Dictionaries gelangen niemals zwischen den Schritten hin und her.
Die praktische Folge ist, dass sich die Spalten im Laufe des Datentransfers anhäufen. Die Eingabeschicht fügt id, name, path und source hinzu. Die Extraktionsphase fügt content hinzu. Die Chunking-Schicht fügt chunked_content hinzu, und die Embedding-Schicht fügt embeddings hinzu. Wenn eine Zeile schließlich im Vektorlagern ankommt, enthält sie ein vollständiges Datensatz mit allen Spalten und Metadaten, bereit zur Indizierung.
Ein Operator berührt nur die Spalten, die ihn interessieren, und alles andere fließt unberührt weiter. Daher kann ein neuer Operator überall eingefügt werden, ohne seine Nachbarn zu verändern.
Ein Fluss ist eine DAG. Die Pipeline befindet sich in einer JSON-Datei. Jeder Knoten gibt einen type (den Kurznamen des Operators), einen im Graphen eindeutigen name, einen config-Block sowie eine depends_on-Liste an. Beim Start löst der Engine das Abhängigkeitsgraphen auf und führt die Knoten in topologischer Reihenfolge aus, wobei parallele Zweige gleichzeitig ausgeführt werden.
Die Struktur der Pipeline besteht also aus Daten und nicht aus Code. Sie kann versioniert, verglichen, statisch überprüft und ersetzt werden, ohne irgendeinen Python-Code anzufassen. Die Option --validate weist vor der Verarbeitung eines einzigen Dokuments auf Abhängigkeitsverstöße, fehlende Spalten sowie Konfigurationsfehler hin.
Warum das Modell funktioniert. Im Praxistest heben sich zwei Vorteile hervor.
Einer ist die Kompositionsfähigkeit. Ein Operator kann nach jedem anderen folgen, sofern die vorangegangenen Operatoren die Spalten erzeugt haben, die er in get_required_features() auflistet. Der Engine überprüft dies beim Start und gibt die genaue fehlende Spalte sowie den Knoten an, der sie liefern sollte – und das bereits vor Beginn der Ausführung.
Der andere Vorteil ist die Testbarkeit. Jeder Operator verhält sich wie eine reine Funktion von einer Tabelle zur nächsten, sodass er einzeln getestet werden kann: Erstelle eine pa.Table mit den erforderlichen Spalten, rufe transform() auf und überprüfe das Ergebnis. Es sind weder Mocks noch Nebeneffekte noch ein laufender Pipeline erforderlich.
Ein Überblick über die Operatoren
Es gibt 22 eingebaute Operatoren in fünf Kategorien. Dies ist eine kurze Übersicht.
Einlesen. IngestSource ist der einzige Einlesevorgang, der sowohl lokale als auch Cloud-Quellen abdeckt. Ein lokaler Pfad verwendet den filesystem-Anbieter, und derselbe Knoten kann ohne Änderungen im Ablauf auf einen von acht Cloud-Anbietern (S3, IBM COS, SharePoint, OneDrive, Google Drive, Box, CSV-Dateien und Webseiten) umschalten. Er gibt pro Datei nur Metadaten aus, noch nicht den Inhalt. Der Extraktionsschritt holt den Inhalt später ab, wodurch das Einlesen schnell und auch bei großer Skalierung memoryeffizient bleibt.
Extrahieren. Extract wandelt Rohdateien in strukturierten Text um. PDF-, DOCX-, HTML-Dateien sowie Bilder, Audio- und Videodateien werden über Docling in Markdown umgewandelt – entweder im Bibliotheksmodus (lokal, ohne Netzwerkaufruf) oder über Docling Serve im API-Modus, der es ermöglicht, aufwändige Extraktionsaufgaben auf einen speziellen Dienst auszulagern.
Die Entitätenextraktion ist darüber hinaus optional und wird von drei Anbietern bereitgestellt: LiteLLM (das auf über 100 Modellanbieter verweist), IBM watsonx.ai sowie Doclings extraktion basierend auf Vorlagen. Die Wahl hängt vom Zugang zu Modellen und den Kosten ab.
ACLOperator ist Teil der Extraktionsphase. Er zieht die Zugriffssteuerungsmetadaten aus Dokumenten in eine eigene Spalte, sodass nachfolgende Systeme eine Dokumentebene-Sicherheit ohne erneuten Zugriff auf die Quelldatei anwenden können.
Funktional. In diese Kategorie fallen die Pipeline-Logik: Operatoren, die Daten transformieren, leiten und umformen, anstatt sie zu erweitern.
ChunkerOperator teilt den extrahierten Text in Blöcke auf, wobei er eine von drei Strategien anwendet: einfache (fester Zeichenzahl, konfigurierbare Überlappung), semantische (inhaltsgestützte) oder hybride (Docling-eigene Blockierung, die die Dokumentstruktur beibehält). Zudem kann er jeden Block mit einem LLM zusammenfassen, was einer Suche eine reichhaltigere Darstellung bietet.
EmbeddingsOperator erzeugt Vektor-Embeddings für jeden Block. Er arbeitet mit lokal ausgeführten HuggingFace-Modellen, LiteLLM (mehr als 100 Cloud-Anbieter) sowie IBM watsonx.ai, und das Wechseln des Embedding-Modells erfordert lediglich eine Änderung in einer Zeile des Flow-JSONs.
BranchingOperator ist der Punkt, an dem die Abläufe interessant werden. Er teilt Dokumente anhand von Spaltenwerten, Bedingungen im SQL-Stil oder Klassifizierungslabels in separate Unter-Pipelines auf. Jede Abzweigung läuft unabhängig und wird im Feld depends_on mithilfe der Punktnotation nach ihrem Namen referenziert, zum Beispiel "depends_on": ["branch_node.branch_name"].
MergeOperator vereint die in verschiedene Ströme aufgeteilten Daten wieder miteinander, indem er entweder Zeilen stapelt oder Spalten verknüpft – je nach den Anforderungen der nachfolgenden Operatoren.
DocIdHashOperator erzeugt aus dem Inhalt jedes Dokuments eine deterministische, stabile Hash-ID. Diese ID bildet die Grundlage für die Deduplizierung sowie den inkrementellen Verarbeitungsablauf: Dasselbe Dokument erhält immer dieselbe ID, egal wo und wann es eingeführt wird.
EntityCurationOperator und NOOPOperator vervollständigen das Set. Der erste kümmert sich um die Reinigung der Entitäten nach der Extraktion, während der zweite ein Durchleitungsnode ist, der bei der Verknüpfung des DAG hilft.
Qualität. Die Operatoren in dieser Gruppe markieren den Unterschied zwischen einem einfachen Pipeline-System und einem zuverlässigen System.
LanguageDetect erkennt mithilfe von fastText die Sprache eines Dokuments und fügt eine lang_name-Spalte hinzu, die zur Filterung, Weiterleitung oder Kennzeichnung genutzt werden kann.
ReadabilityOperator bewertet jedes Dokument in fünf Aspekten hinsichtlich der Lesbarkeit: Flesch-Reading-Ease, Flesch-Kincaid-Grad, Gunning Fog, Dale-Chall sowie die Anzahl schwieriger Wörter. Diese Bewertungen liefern nützliche Signale zur Weiterleitung, zum Beispiel um technische Dokumentation von allgemeinem Inhalt zu unterscheiden, bevor eine Aufteilungsstrategie gewählt wird.
PIIAndHAPAnnotator nutzt auf LLMs basierende Erkennungsmethoden, um personenbezogene Informationen (E-Mail-Adressen, Telefonnummern, Sozialversicherungsnummern, Kartennummern, Bankkonten, IP-Adressen) sowie Hassausdrücke, Missbrauch und Schimpfwörter zu finden und zu entfernen, wobei für jede Kategorie eigene Schwellenwerte festgelegt sind.
RedactionOperator kümmert sich um die entfernung basierend auf Mustern, bei denen die Formate bekannt sind, wie z. B. interne IDs, Produktcodes oder alles, was mit einer regulären Ausdrucksregel abgeglichen werden kann.
EdedupOperator entfernt exakte Duplikate im gesamten Dokumentensatz und verwendet dabei die doc_id_hash-Spalte, um dies deterministisch durchzuführen.
SQLFilterOperator gehört zu den am meisten unterschätzten Komponenten. Er filtert Zeilen mithilfe einer SQL WHERE-Bedingung anhand von Spaltenwerten, sodass "docq_total_words > 100 AND lang_name = 'en'" ohne jeglichen Python-Anteil eine vollständige Qualitätsprüfung darstellt. Zusammen mit Lesbarkeitsbewertungen und Spracherkennung ermöglicht er eine recht ausgefeilte Routierung ohne Code.
DocumentClassifierOperator sortiert Dokumente in konfigurierbare Kategorien. DocQuality berechnet Qualitätsmetriken für das gesamte Korpus, und MLEnrichmentOperator fügt ML-Vorhersagen als zusätzliche Spalten hinzu. Der allgemeine Punkt, den auch andere Beschreibungen von Pipelines betonen, ist, dass die Datenqualität oft einen größeren Einfluss auf die RAG-Ergebnisse hat als die Wahl des Modells.
VectorDB und Speicherung. VectorDBOperator schreibt in Blöcken gegliederte, eingebettete Dokumente in einen Vektor-Speicher, wobei OpenSearch und Milvus als unterstützte Backend-Systeme dienen. Feldzuordnungen, Indexeinstellungen sowie die Normalisierung von Metadaten werden alle im Flow-JSON festgelegt.
DocumentSetOperator speichert die verarbeiteten Tabellen in einem von DuckDB unterstützten Dokumentensatz. Dies ist nützlich, wenn die verarbeiteten Dokumente auch ohne Vektor-Speicher abgefragt werden sollen oder wenn ein dauerhafter Zwischencheckpoint im Pipeline-Prozess gewünscht ist.
StorageOutputOperator schreibt Dokumente wieder in externe Ziele (S3, SharePoint, Google Drive oder das lokale Dateisystem) – in verarbeiteter oder ursprünglicher Form – was sich für „verarbeiten und archivieren“-Szenarien eignet.
All diese Komponenten können in beliebiger Reihenfolge kombiniert werden, sofern die vorherigen Abhängigkeiten erfüllt sind, sodass die Pipeline nicht in eine feste Struktur gebunden ist.
Ein echter Flow
Hier ist der kanonische Quickstart-Fluss, ein vollständiger Pipeline von der Eingabe bis zum Vektorlagers, der Ollama für Embeddings und OpenSearch für die Speicherung verwendet:
{
"flow_name": "complete-document-pipeline",
"global_config": {
"doc_column": "content",
"storage": "in-memory"
},
"flow": [
{
"type": "ingest_source",
"name": "ingest_local_folder",
"config": {
"provider": "filesystem",
"connection_params": {
"paths": [
"./sample_documents"
]
},
"include_filter": "pdf,txt,docx"
}
},
{
"type": "extract_operator",
"name": "extract_with_docling",
"config": {
"text_extraction": {
"provider": "docling_library"
},
"entity_extraction": {
"provider": "none"
}
},
"depends_on": [
"ingest_local_folder"
]
},
{
"type": "chunker",
"name": "simple_chunker",
"config": {
"chunk_type": "simple",
"chunk_size": 512,
"chunk_overlap": 50
},
"depends_on": [
"extract_with_docling"
]
},
{
"type": "embeddings",
"name": "ollama_embeddings",
"config": {
"provider": "litellm",
"provider_config": {
"model_id": "openai/nomic-embed-text",
"api_base": "http://localhost:11434/v1"
},
"embeddings_column": "embeddings"
},
"depends_on": [
"simple_chunker"
]
},
{
"type": "vectordb",
"name": "opensearch_vector_store",
"config": {
"provider": "opensearch",
"doc_id_column": "doc_id_hash",
"embeddings_column": "embeddings",
"provider_config": {
"index_name": "sample-documents-index",
"host": "localhost",
"port": 9200
}
},
"depends_on": [
"ollama_embeddings"
]
}
]
}
Node für Node vorgehen:
- ingest_local_folder durchsucht
./sample_documentsnach PDF-, Text- und DOCX-Dateien mithilfe voningest_sourcesowie dessen Dateisystemanbieter. Jede Datei wird zu einer Zeile mit den Feldern id, Name, Pfad und Quelle. Der Inhalt wird zu diesem Zeitpunkt noch nicht gelesen, wodurch der Vorgang schnell abläuft – es werden lediglich die Dateien gefunden, anstatt sie zu laden. - extract_with_docling nimmt diese Metadatenzeilen, liest jede Datei mit Docling und fügt eine
content-Spalte hinzu, die den extrahierten Markdown-Text enthält. An diesem Punkt werden PDFs in strukturierten, abfragbaren Text umgewandelt. Derdocling_library-Anbieter läuft vollständig im Prozess selbst, sodass kein externer Dienst erforderlich ist.
chunked_content hinzu, eine strukturierte Liste von Blöckobjekten, die jeweils ihren Text sowie ihre Position im Quelldokument enthalten.embeddings-Spalte. Es handelt sich dabei im Grunde um LiteLLM für Ollama; durch Änderung von model_id und api_base wird der Datenfluss an einen der über 100 von LiteLLM unterstützten Anbieter umgeleitet, ohne dass sonst etwas geändert werden muss.doc_id_hash als Primärschlüssel. Dadurch entstehen beim erneuten Ausführen niemals Duplikate.Mehrere strukturelle Aspekte verdienen Aufmerksamkeit. depends_on ist die einzige notwendige Verbindung: Es findet keine explizite Datenübertragung statt, es gibt keine Callback-Registrierung und keinen gemeinsamen Zustand, da der Engine den Graphen auflöst und die Daten verschiebt. Der global_config-Block legt Standardwerte fest, die jeder Operator erbt. doc_column gibt an, in welcher Spalte der Haupttext gespeichert ist, und storage: „in-memory“ behält Zwischentabellen zwischen den Schritten im RAM, was für die meisten Ausführungen geeignet ist und bei großen Datensätzen durch eine andere Lösung ersetzt werden kann.
Die gesamte Pipeline wird mit einem Befehl ausgeführt:
docling-pipelines - flow-file pipeline.json
Fortgeschrittene Muster
Flüsse müssen nicht lineare Ketten sein. Drei Muster kommen in realen Implementierungen häufig vor.
Ableiten und Zusammenführen. Das Ableiten sendet Dokumente entsprechend konfigurierbaren Kriterien an verschiedene Unter-Pipelines. Jede Abzweigung ist eine benannte Ausgabe, auf die nachgelagerte Knoten mit Punktnotation in depends_on verweisen. Eine durch einen Lesbarkeitswert gesteuerte Routenabzweigung sieht so aus:
{
"name": "quality_branching",
"type": "branching",
"config": {
"branches": {
"high_quality_branch": {
"link_name": "High Quality Documents",
"criteria_json": {
"criteria_list": [
{
"variable": "flesch_reading_ease",
"operator": ">",
"value": 60
}
],
"logical_operator": "AND"
}
},
"low_quality_branch": {
"link_name": "Low Quality Documents",
"criteria_json": {
"criteria_list": [
{
"variable": "flesch_reading_ease",
"operator": "<=",
"value": 60
}
],
"logical_operator": "AND"
}
}
}
},
"depends_on": [
"readability"
]
}
Die hochwertige Abteilung kann vor dem Einbetten einer ML-Verfeinerung durchlaufen, während die niedrigwertige Abteilung einen weiteren SQLFilterOperator durchläuft, der Dokumente mit null Wörtern vor dem Einbetten entfernt. Beide landen schließlich in separaten OpenSearch-Indizes, und MergeOperator kann die Ströme anschließend wieder zusammenführen, wenn späterige Logik einen einzigen Überblick wünscht.
Qualitätsbasiertes Routing. Die Idee besteht darin, zunächst günstige Qualitätsoperator auszuführen und anschließend SQLFilterOperator dazu zu nutzen, Zeilen vor jedem kostspieligen LLM-Schritt zu entfernen. Ein vollständiges Qualitätsfilter kann wie folgt aussehen:
lang_detect → ededup → doc_quality → readability → sql_filter → embeddings
Die Bedingung sql_filter ist im Grunde eine SQL WHERE-Klausel, die als Daten dargestellt wird:
{
"criteria_list": [
{
"variable": "docq_total_words",
"operator": ">",
"value": 100
},
{
"variable": "lang_name",
"operator": "=",
"value": "en"
}
],
"logical_operator": "AND"
}
Es gibt weder Python noch benutzerdefinierte Operatoren – nur eine Bedingung im JSON – und jede Zeile, die dieser Bedingung nicht entspricht, erreicht niemals den Aufruf zur Embedding-Verarbeitung.
Mehrstufige Aufwertung. Die Abfolge klassifizieren, routen, extrahieren, aufwerten, speichern kommt in der Produktion ständig vor. Dokumente können zunächst klassifiziert werden, wobei technische Dokumente an einen für technischen Inhalt optimierten Entitätenextraktor weitergeleitet und Richtliniedokumente an einen anderen. Jeder Zweig fügt Spalten hinzu, die spezifisch für seinen Typ sind, und die Ströme werden vor dem Chunking und Embedding zusammengeführt, sodass der Rest des Pipelines gemeinsam genutzt werden kann.
Jede Stufe fügt Spalten hinzu, wodurch die nachfolgenden Operatoren eine immer umfangreichere Tabelle erhalten. Dabei wird auf spezielle Modi oder Flags verzichtet. Es handelt sich um gewöhnliche Operatoren, die miteinander kombiniert werden – das ist der Vorteil des Operator-Modells.
Drei Möglichkeiten, es auszuführen
Das Framework ist mehr als nur ein CLI-Tool und eignet sich für verschiedene ingenieurwissenschaftliche Kontexte.
CLI. Dies ist der schnellste Weg zu einem laufenden Pipeline:
docling-pipelines - flow-file pipeline.json
docling-pipelines - flow-file pipeline.json - validate # validate without running
docling-pipelines - list-operators - verbose # discover all registered operators
--validate ist während der Entwicklung eines Flows nützlich: Abhängigkeitsverstöße, fehlende erforderliche Spalten sowie Konfigurationsfehler werden bereits vor dem Lesen oder Schreiben aufgedeckt. Es ist eine gute Gewohnheit, es im CI vor dem Bereitstellen einer Flow-Änderung auszuführen.
Python-Bibliothek. Wenn Pipelines durch Anwendungscode ausgelöst werden müssen:
from docpipe.lib.docpipe_flow_manager import DocpipeFlowManager
manager = DocpipeFlowManager(flow_file="pipeline.json")
manager.execute()
Dies eignet sich für Code, der in einer größeren Anwendung, einem Notebook oder einem Orchestrierungsskript eingebettet ist und das eigene Lauflebenszyklus-Management übernehmen muss.
REST API. Für den Betrieb als langlebiger Dienst:
uvicorn docpipe.api.main:app - host 0.0.0.0 - port 8000
Der FastAPI-Server stellt drei Gruppen von Routen zur Verfügung: /api/v1/flows zur Verwaltung von Workflows, /api/v1/job_runs zur Steuerung von Job-Ausführungen sowie /api/v1/operators zur Erkennung von Operatoren. Die Benutzer melden sich über LDAP oder OAuth2/OIDC mit JWT-Tokens an, und Dokumentationen im OpenAPI/Swagger-Format werden automatisch unter /docs erzeugt. Dies ist die Schnittstelle für einen gemeinsam genutzten Dienst, den mehrere Teams nutzen, der durch externe Ereignisse ausgelöst werden kann und dessen Status sowie Ergebnisse von anderen abgerufen werden können.
Erweiterung mit benutzerdefinierten Operatoren
Für eine nachhaltige Einführung in echten Organisationen ist dies wohl der wichtigste Aspekt des Designs.
Der Vertrag ist so kompakt, dass ein Operator in etwa 30 Zeilen untergebracht werden kann. Eine Python-Klasse erfüllt die Voraussetzungen, wenn sie von AbstractOperator abgeleitet ist und transform(), get_metadata() sowie get_required_features() implementiert. Sie kann mit jedem eingebauten Operator kombiniert werden, erscheint in der Ausgabe von --list-operators, und das Flow-JSON bezieht sich auf sie über short_name, genauso wie auf einen eingebauten Operator.
Die Registrierung ist je nach Entwicklungsstadium auf drei Arten möglich.
Dateisystempfad wird für die Entwicklung verwendet. Weisen Sie eine Umgebungsvariable auf einen lokalen Ordner hin:
DOCPIPE_CUSTOM_OPERATORS=./my_operators docling-pipelines - flow-file pipeline.json
Alles in diesem Verzeichnis wird bei der Startzeit importiert. Eine Installation ist nicht erforderlich, wodurch die Iteration schnell erfolgt.
Python-Paket dient der Verbreitung. Veröffentlichen Sie Operator als pip-Paket mit dem richtigen Eingangspunkt, installieren Sie es in derselben Umgebung – dann registrieren sie sich automatisch. So können Operator andere Teams erreichen, ohne dass jemand den Kerncode verändert.
S3-URI ist die Unternehmenslösung. Weisen Sie DOCPIPE_CUSTOM_OPERATORS auf ein entferntes Operator-Paket hin, und das Framework lädt es zur Laufzeit herunter und registriert es. Dies eignet sich für eine zentrale Verwaltung von Operatorn in vielen Bereitstellungen: An einer Stelle können Updates vorgenommen werden, und jede Bereitstellung übernimmt die Änderungen beim nächsten Start.
Die unternehmensspezifische Logik bleibt daher außerhalb des Kerns. Konformitätsverarbeitungsfunktionen, domänenspezifische Entitätenextraktoren sowie proprietäre Bewertungsmodelle können alle als Operatoren verpackt werden, mit den eingebauten Komponenten im Flow JSON kombiniert und ohne Forking des Frameworks bereitgestellt werden.
Operative Aspekte
Eine berechtigte Frage zu einem solchen Tool ist, ob es sich um ein gutes Prototyp oder tatsächlich um eine produktreife Lösung handelt. Die Details folgen.
Überwachbarkeit. Eine native Unterstützung für OpenTelemetry ist bereits integriert, und jede Ausführung eines Operators erzeugt einen Span. Die Logs sind strukturiertes JSON mit Trace-Korrelation, das in jeder Zeile trace_id und span_id enthält. Dies funktioniert mit Jaeger, Grafana Tempo, Datadog und AWS X-Ray ohne Konfiguration auf der Anwendungsseite, und der Overhead ist bei deaktivierter Telemetrie null.
Schrittweises Verarbeiten. Die Pipeline verwendet die deterministische ID aus DocIdHashOperator, um nachzuvollziehen, was bereits verarbeitet wurde. Bei einer erneuten Ausführung werden nur neue oder geänderte Dokumente bearbeitet; dasselbe Dokument, das zweimal eingereicht wird, erhält denselben Hash und wird übersprungen. Die Einstellung force_ingest: true in global_config überschreibt dies und verarbeitet alles erneut. Metadaten für schrittweise Ausführungen können in DuckDB (Standard), PostgreSQL oder dem Dateisystem gespeichert werden.
Sicherheit. ACLOperator stellt Berechtigungsmetadaten als Spalte zur weiteren Umsetzung bereit, und PIIAndHAPAnnotator erkennt und entfernt sensible Inhalte, bevor diese den Vektor-Speicher erreichen. Geheimnisse in der Flusskonfiguration können aus HashiCorp Vault stammen: Ein Pfad wie vault://path/to/secret, der im JSON festgehalten ist, wird während der Ausführung vom Engine gelöst. Der Zugriff auf die REST-API wird durch LDAP sowie OAuth2/OIDC geschützt.
Skalierung. Die Ausführung ist in drei Modi verfügbar, die von einem Laptop bis zu einer verteilten Flotte reichen. Standardmäßig wird ein Thread-Pool innerhalb des Prozesses verwendet, der schnell, einfach und unabhängig von Abhängigkeiten ist. Ein Prozess-Pool führt Operationen in separaten Prozessen aus, um rechenintensive Aufgaben auf einem Knoten zu bewältigen. Für verteilte Produktivlasten steht ein Prefect-Work-Pool zur Verfügung, der Docker-Container ausführt und so horizontale Skalierung sowie zentrale Verwaltung der Ausführungen ermöglicht. Die Einstellungen für das Mikro-Batching steuern den Speicherverbrauch in jedem Modus – was wichtig ist, wenn große Dokumente der LLM-Extraktion unterzogen werden.
Was kommt als Nächstes?
Das Projekt wird auf GitHub unter der Organisation IBM gehostet, und sein Schnellstartleitfaden ist der schnellste Weg zu einem laufenden Pipeline-System. Bereiche, die einer genaueren Untersuchung bedürfen, umfassen die Extraktionsschicht (Integration von Docling, VLM-Pipelines für dokumentenreiche Bilder, ASR für Audio und Video), das Schreiben eines benutzerdefinierten Operators mit komplexer Logik – von der ersten Klassendeklaration über Tests bis hin zur Registrierung und Bereitstellung –, den Betrieb von Prefect mit Docker für große Arbeitslasten sowie die Optionen zur Feldzuordnung und Schema-Vorlage des VectorDBOperator für OpenSearch.
Verwandte Literatur
- Praktische Anmerkungen: Aufbau einer Produktionsreifen lokalen RAG-Pipeline – 100% kostenlos – Detaillierte Anleitung zu den Praktischen Anmerkungen: Aufbau einer Produktionsreifen lokalen RAG-Pipeline – 100% kostenlos: Verträge, Überprüfungen sowie Code-Blöcke für Teams, die dieses Muster einsetzen.
- Hybride Suche in Postgres für einen Gesundheitsassistenten: Schema, Chunking, RRF und Bewertung – Aufbau der Abrufschicht eines Gesundheitsassistenten in Postgres: strukturbezogenes Chunking, hybride Suche mit pgvector und RRF, binäre Quantisierung, Schwellenwerte sowie Bewertungsmethoden.