Młode pipeline’y: tworzenie pipeline’ów danych RAG klasy korporacyjnej
Jak docling-pipelines zastępuje kruche skrypty typu glue operatorami komponowalnymi oraz strumieniami JSON: pobieranie danych, wydobywanie informacji, kontrola jakości, embeddingi i przechowywanie wektorowe.
Źródła: Introducing Docling Pipelines: Building Enterprise-Grade RAG Data Pipelines
Gdzie popełnia się błędy przy używaniu kodu łączącego
Załóżmy, że mamy wspólny dysk z 10 000 plikami PDF, stronę SharePoint pełną plików Word oraz zaległości w nagraniach rozmów z klientami. Zespół odpowiedzialny za produkt chce, aby wszystko to znalazło się w bazie danych wektorowej, była możliwa jej wyszukiwanie i podział na fragmenty, a dane osobowe zostały usunięte, a duplikaty wyeliminowane.
Zwykłą reakcją inżynierów jest stworzenie skryptu, a potem drugiego skryptu. Dodaje się kilka bibliotek, notatnik Pandas łączy je ze sobą, zadań typu cron uruchamia się całość co noc, i system jest gotowy do użycia. Działa, przynajmniej przez jakiś czas.
Trzy miesiące później ktoś pyta, dlaczego brakuje nowych dokumentów. Ślad kończy się na zmianie schematu w jednej z kolumn, a skrypt cicho odrzucił te wiersze. Nikt tego nie zauważył, ponieważ nie było nic widocznego. Po naprawie następuje ponowne uruchomienie, a wraz z nim uświadomienie sobie, że nie ma żadnego zapisu o tym, które dokumenty zostały już przetworzone, więc wszystkie 10 000 zostaje ponownie przetworzonych.
Taki jest typowy los „kleju” w łańcuchu przetwarzania dokumentów. Zaczyna się prosto i powoli się psuje, a potem nagle. Dodawane są próby ponownego uruchomienia, filtrowanie jakości pojawia się jako dodatek, a śledzenie zależności między etapami nie istnieje, więc sukces zależy od tego, czy skrypty uruchomią się we właściwej kolejności. Aktualizacje stopniowe są trudne bez stabilnej tożsamości dokumentu. Pozostaje plątanina, którą może naprawić tylko jej autor, a nawet on już nie jest pewien.
Czym jest docling-pipelines
Docling-pipelines to modułowa, oparta na operatorach platforma do przetwarzania dokumentów, przeznaczona do budowania kompletnych pipeline’ów danych RAG. Obejmuje ona cały cykl życia dokumentu – począwszy od pobierania danych z lokalnych lub chmurowych źródeł, przez wyodrębnianie tekstu i entytetów, filtrowanie jakości, dzielenie na fragmenty i tworzenie wektorów, aż po zapisanie ich do bazy danych wektorowej. To standardowa ścieżka, a większość pipeline’ów produkcyjnych stanowi jej wariację.
Instaluje się go jako zwykły pakiet Pythona:
bash
pip install docling-pipelines
Działa wszędzie tam, gdzie działa Python. Do podstawowego użytku nie wymaga żadnej zewnętrznej infrastruktury: ani Kafka, ani klastra Spark, nic do konfiguracji. Przepływ jest opisany w pliku JSON, interfejs wiersza poleceń jest skierowany na ten plik i rozpoczyna się działanie.
Istnieją trzy sposoby korzystania z niego. Interfejs wiersza poleceń nadaje się do jednorazowych wykonywań i lokalnego rozwoju. Biblioteka Python, oparta na DocpipeFlowManager, służy do uruchamiania przepływów z kodu aplikacji lub notatek. REST API jest przeznaczony do przepływów uruchamianych jako usługa współdzielona przez kilka zespołów. Wszystkie trzy opcje zostaną omówione poniżej.
Projekt jest otwarty kodowo, a jego pierwsza wersja, v0.1.0, ukazała się w lipcu 2025 roku.
Główna idea: operatorzy i przepływy
Każdy, kto dopiero poznaje ten framework, powinien poświęcić tu najwięcej czasu, ponieważ reszta staje się jasna po zrozumieniu tego zagadnienia.
Operator to podstawowa jednostka. Operator to klasa w Pythonie, która przyjmuje tabelę dokumentów i zwraca tabelę, ewentualnie przetworzoną. Każda funkcjonalność w tym frameworku – czy to pobieranie danych, wydobywanie informacji, wykrywanie języka, dzielenie na fragmenty czy tworzenie wektorów reprezentacji – jest operatorem.
Obowiązki są niewiele:
short_name: ciąg znaków używany przez JSON do odnoszenia się do operatoratransform(table): przyjmuje obiektpa.Tablei zwraca listę tabel wraz z słownikiem metadanychget_metadata(): metoda@staticmethodzwracająca metadane operatora, aby mógł zostać odkrytyget_required_features(): określa, jakie kolumny są potrzebne operatorowi jako dane wejściowevalidate(): opcjonalna metoda służąca do sprawdzenia konfiguracji podczas uruchamiania
Minimalny szkielet wygląda następująco:
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": "…"}
Wdrożenie tych metod pozwala na stworzenie komponentu pierwszej klasy, który może być łączony z każdym innym operatorem w ekosystemie.
Tabele PyArrow przechowują dane. Każdemu operatorowi jest przekazywany obiekt pa.Table, a on z kolei zwraca taki sam obiekt. Jest to świadoma decyzja, ponieważ kolumnowy format PyArrow jest wydajny i typowany, a dane przemieszczają się pomiędzy etapami bez konieczności kopiowania. Surowe łańcuchy tekstowe oraz słowniki Pythona nigdy nie przechodzą pomiędzy poszczególnymi krokami.
Piękną konsekwencją tego jest to, że kolumny gromadzą się w miarę przetwarzania danych. Etap pobierania dodaje pola id, name, path oraz source. Etap wydobywania dodaje pole content. Narzędzie do dzielenia na fragmenty dodaje chunked_content, a etap embeddingów dodaje embeddings. Gdy wiersz trafia do magazynu wektorowego, zawiera on pełny zapis, wraz ze wszystkimi kolumnami i metadanymi, gotowy do indeksowania.
Operator dotyka tylko tych kolumn, które go interesują, a wszystko inne przechodzi obok niego nietknięte. Dlatego nowy operator może zostać umieszczony w dowolnym miejscu bez wpływu na sąsiednie elementy.
Przepływ to DAG. Pipeline znajduje się w pliku JSON. Każdy węzeł określa type (krótką nazwę operatora), name unikalny w grafie, blok config oraz listę depends_on. Podczas uruchamiania silnik rozwiązuje graf zależności i uruchamia węzły w porządku topologicznym, przy czym równoległe gałęzie są wykonywane jednocześnie.
Zatem struktura pipeline to dane, a nie kod. Może być wersjonowana, porównywana, statycznie weryfikowana i zastępowana bez konieczności edytowania żadnego kodu w Pythonie. Opcja --validate wykrywa naruszenia zależności, brakujące kolumny oraz błędy konfiguracji jeszcze przed przetworzeniem pierwszego dokumentu.
Dlaczego model działa. W praktyce wyróżniają się dwie zalety.
Pierwszą jest kompozytywność. Operator może następować po dowolnym innym, pod warunkiem że poprzednie wygenerowały kolumny wymienione w metodzie get_required_features(). Silnik sprawdza to przy uruchamianiu i podaje dokładną brakującą kolumnę oraz węzeł, który powinien ją dostarczyć, wszystko to przed rozpoczęciem wykonywania.
Drugą zaletą jest możliwość testowania. Każdy operator zachowuje się jak funkcja czysta, przenosząca dane z jednej tabeli do drugiej, więc można go testować osobno: utworzyć pa.Table z potrzebnymi kolumnami, wywołać transform() i sprawdzić wynik. Nie są potrzebne mocki, efekty uboczne ani aktywny pipeline.
Przegląd operatorów
Istnieje 22 wbudowane operatorów w pięciu kategoriach. To krótki przegląd.
Przyjmowanie danych. IngestSource to jedyny operator przyjmowania danych, który obsługuje zarówno źródła lokalne, jak i chmurowe. Ścieżka lokalna wykorzystuje dostawcę filesystem, a ten sam węzeł może przełączyć się na dowolnego z ośmiu dostawców chmurowych (S3, IBM COS, SharePoint, OneDrive, Google Drive, Box, pliki CSV oraz strony internetowe) bez żadnych zmian w pozostałej części procesu. Wystawia on jeden wiersz na plik, zawierający jedynie metadane, bez żadnej treści. Krok wyodrębniania treści następuje później, co zapewnia szybkie przyjmowanie danych i niewielkie obciążenie pamięci, nawet w dużych skali.
Wyodrębnianie treści. Funkcja Extract przekształca surowe pliki w tekst ustrukturyzowany. Pliki PDF, DOCX, HTML, obrazy, dźwięk i wideo są przekształcane w format markdown za pomocą Docling, albo w trybie biblioteki (lokalnie, bez połączenia sieciowego), albo za pośrednictwem Docling Serve w trybie API, co umożliwia przeniesienie obciążonego procesu wyodrębniania treści na dedykowaną usługę.
Ekstrakcja entytetów jest dodatkową opcją, dostępną u trzech dostawców: LiteLLM (który łączy się z ponad 100 dostawcami modeli), IBM watsonx.ai oraz systemu ekstrakcji opartego na szablonach od Docling. Wybór zależy od dostępu do modeli oraz kosztów.
ACLOperator jest częścią etapu ekstrakcji. Wyodrębnia on metadane dotyczące kontroli dostępu z dokumentów do osobnej kolumny, dzięki czemu późniejsze systemy mogą stosować zasady bezpieczeństwa na poziomie dokumentu bez konieczności ponownego otwierania pliku źródłowego.
Funkcjonalny. Do tej kategorii należy logika procesu: operatorzy, które przekształcają, kierują i przekształcają strukturę danych, zamiast je wzbogacać.
ChunkerOperator dzieli wydobyty tekst na fragmenty, stosując jedną z trzech strategii: prostą (stała długość znaku, konfigurowalne nakładanie się), semantyczną (uwzględniającą treść) lub hybrydową (fragmentowanie typowe dla Docling, które zachowuje strukturę dokumentu). Może również streszczać każdy fragment za pomocą modelu LLM, co zapewnia bogatsze przedstawienie danych przy ich wyszukiwaniu.
EmbeddingsOperator tworzy wektorowe reprezentacje dla każdego fragmentu. Działa z modelami HuggingFace uruchamianymi lokalnie, LiteLLM (ponad 100 dostawców chmury) oraz IBM watsonx.ai, a zmiana modelu reprezentacji polega jedynie na jednej modyfikacji w pliku JSON opisującym proces.
BranchingOperator to element, który sprawia, że procesy stają się ciekawe. Dzieli dokumenty na oddzielne podprocesy na podstawie wartości kolumn, warunków w stylu SQL lub etykiet klasyfikacyjnych. Każda gałąź działa niezależnie i jest odwoływana pod względem nazwy w polu depends_on za pomocą notacji z kropką, na przykład "depends_on": ["branch_node.branch_name"].
MergeOperator łączy ponownie połączone strumienie, albo układając wiersze obok siebie, albo łącząc kolumny, zgodnie z wymaganiami operatorów poniżej w łańcuchu przetwarzania.
DocIdHashOperator wytwarza deterministyczny, stabilny identyfikator hash na podstawie treści każdego dokumentu. Ten identyfikator stanowi podstawę usuwania duplikatów oraz przetwarzania inkrementalnego: ten sam dokument zawsze otrzymuje ten sam identyfikator, niezależnie od miejsca i czasu jego wprowadzenia.
EntityCurationOperator oraz NOOPOperator uzupełniają tę zbiór. Pierwszy zajmuje się oczyszczaniem entytetów po ich wyekstrakcji, natomiast drugi jest węzłem przekierowującym, który pomaga w łączeniu elementów DAG.
Jakość. Operatory z tej grupy pokazują różnicę pomiędzy posiadaniem zwykłego procesu przetwarzania a posiadaniem takiego, któremu można zaufać.
LanguageDetect identyfikuje język dokumentu za pomocą fastText i dodaje kolumnę lang_name, która może służyć do filtrowania, kierowania przepływem danych lub tagowania.
ReadabilityOperator ocenia każdy dokument pod względem czytelności na pięć sposobów: wskaźnik Flesch Reading Ease, poziom trudności Flesch-Kincaid, wskaźnik Gunning Fog, wskaźnik Dale-Chall oraz liczbę trudnych słów. Dostarczają one przydatnych sygnałów do kierowania przepływem danych, na przykład do odróżnienia dokumentacji technicznej od treści ogólnych przed wyborem strategii dzielenia na fragmenty.
PIIAndHAPAnnotator wykorzystuje detekcję opartą na modelach językowych, aby znajdować i maskować informacje osobiste (adresy e-mail, numery telefonów, numery ubezpieczenia społecznego, numery kart, konta bankowe, adresy IP), a także treści pełne nienawiści, obraźliwości i wulgaryzmów, przy czym progi są ustalane dla każdej kategorii oddzielnie.
RedactionOperator służy do maskowania na podstawie wzorów, gdy formaty są znane, takie jak wewnętrzne ID, kody produktów lub cokolwiek, co może zostać dopasowane za pomocą regularnych wyrażeń.
EdedupOperator usuwa dokładne duplikaty w całym zbiorze dokumentów, korzystając z kolumny doc_id_hash, aby działać w sposób deterministyczny.
SQLFilterOperator należy do najmniej docenianych elementów. Filtruje wiersze za pomocą warunku SQL WHERE opartego na wartościach kolumn, więc "docq_total_words > 100 AND lang_name = 'en'" tworzy kompletną bramkę kontroli jakości bez użycia Pythona. W połączeniu z ocenami czytelności i wykrywaniem języka umożliwia dość złożone kierowanie przepływem danych bez pisania kodu.
DocumentClassifierOperator sortuje dokumenty do konfigurowalnych kategorii. DocQuality oblicza metryki jakości dla całego korpusu, a MLEnrichmentOperator dodaje przewidywania ML jako dodatkowe kolumny. Szerszy wniosek, który podnoszą również inne opisy pipeline’ów, jest taki, że jakość danych często ma większy wpływ na wyniki RAG niż wybór modelu.
VectorDB i przechowywanie. VectorDBOperator zapisuje dokumenty w formie fragmentów do magazynu wektorowego, przy czym obsługiwane są backendy takie jak OpenSearch i Milvus. Mapowania pól, ustawienia indeksów oraz normalizacja metadanych są określane w pliku JSON opisującym przepływ.
DocumentSetOperator zapisuje przetworzoną tabelę do zestawu dokumentów opartego na DuckDB. Jest to przydatne, gdy przetworzone dokumenty muszą być dostępne do wyszukiwania bez użycia magazynu wektorowego lub gdy potrzebny jest trwały punkt kontrolny w trakcie przetwarzania.
StorageOutputOperator zapisuje dokumenty z powrotem do zewnętrznych destynacji (S3, SharePoint, Google Drive lub lokalny system plików), w formie przetworzonej lub oryginalnej, co odpowiada wzorcom „przetwarzanie i archiwizacja”.
Wszystkie te elementy mogą być łączone w dowolnej kolejności, pod warunkiem spełnienia zależności wstępnych, dzięki czemu przepływ nie jest skrępowany określoną strukturą.
Rzeczywisty przepływ
Oto kanoniczny proces szybkiego rozpoczęcia pracy, kompletny przepływ od pobierania danych do przechowywania we wektorach, który wykorzystuje Ollama do generowania embeddingów oraz OpenSearch do przechowywania danych:
{
"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"
]
}
]
}
Przechodząc węzeł po węźle:
- ingest_local_folder przeszukuje katalog
./sample_documentsw poszukiwaniu plików PDF, tekstowych i DOCX za pośrednictwemingest_sourceoraz jego dostawcy systemu plików. Każdy plik staje się jednym wierszem zawierającym identyfikator, nazwę, ścieżkę i źródło. W tym momencie treść nie jest czytana, więc operacja jest szybka – znajduje się pliki zamiast je ładować. - extract_with_docling bierze te wiersze metadanych, czyta każdy plik za pomocą Docling i dodaje kolumnę
contentz wydobytym tekstem w formacie markdown. To właśnie w tym momencie pliki PDF przekształcają się w ustrukturyzowany tekst, nadający się do wyszukiwania. Dostawcadocling_librarydziała wyłącznie wewnątrz procesu, więc nie jest potrzebna żadna usługa zewnętrzna.
chunked_content – ustrukturyzowaną listę obiektów fragmentów, z których każdy zawiera tekst oraz pozycję w dokumencie źródłowym.embeddings. Jest to w zasadzie wersja LiteLLM dostosowana do Ollamy – zmiana wartości model_id i api_base kieruje przetwarzanie do dowolnego z ponad 100 dostawców obsługiwanych przez LiteLLM, bez konieczności modyfikowania czegokolwiek innego.doc_id_hash jako klucza głównego. Dlatego ponowne uruchomienie tego narzędzia nigdy nie powoduje powstania duplikatów.Kilka kwestii strukturalnych wymaga uwagi. Jedyne wymagane połączenie to depends_on: nie ma wyraźnego przekazywania danych, żadnej rejestracji funkcji zwrotnej ani wspólnego stanu, ponieważ silnik rozwiązuje graf i przenosi dane. Blok global_config ustawia wartości domyślne, które dziedziczą wszystkie operatory. doc_column określa, która kolumna zawiera główny tekst, a storage: „in-memory” przechowuje tabele pośrednie w pamięci RAM pomiędzy krokami, co nadaje się do większości zastosowań i może zostać zastąpione w przypadku dużych zbiorów danych.
Cały proces przebiega przy użyciu jednego polecenia:
docling-pipelines - flow-file pipeline.json
Zaawansowane wzory
Przepływy nie muszą być liniowymi łańcuchami. W rzeczywistych implementacjach powtarzają się trzy wzory.
Rozgałęzianie i łączenie. Rozgałęzianie kieruje dokumenty do różnych podprzepływów zgodnie z konfigurowalnymi kryteriami. Każde rozgałęzienie to nazwany wynik, do którego odwołują się późniejsze węzły za pomocą notacji z kropką w polu depends_on. Rozgałęzienie routingu sterowane oceną czytelności wygląda w ten sposób:
{
"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"
]
}
Wersja o wysokiej jakości może przejść przez proces wzbogacania za pomocą ML przed embedowaniem, natomiast wersja o niskiej jakości przechodzi przez kolejnego SQLFilterOperator, który usuwa dokumenty bez żadnych słów przed embedowaniem. Oba rodzaje dokumentów trafiają do oddzielnych indeksów OpenSearch, a MergeOperator może później ponownie połączyć te strumienie, gdy późniejsza logika wymaga pojedynczego widoku.
Ścieżki routingu oparte na kontrolie jakości. Chodzi o to, aby najpierw uruchomić tanie operatory kontrolujące jakość, a dopiero potem pozwolić SQLFilterOperator usunąć niepotrzebne wiersze przed jakimkolwiek kosztownym krokiem wykorzystującym LLM. Pełna struktura takiego mechanizmu może wyglądać w ten sposób:
lang_detect → ededup → doc_quality → readability → sql_filter → embeddings
Warunek sql_filter to po prostu klauzula WHERE w SQL wyrażona w postaci danych:
{
"criteria_list": [
{
"variable": "docq_total_words",
"operator": ">",
"value": 100
},
{
"variable": "lang_name",
"operator": "=",
"value": "en"
}
],
"logical_operator": "AND"
}
Nie ma żadnego Pythona ani żadnego dostosowanego operatora – tylko warunek w JSON, a każdy wiersz, który go nie spełnia, nigdy nie trafia do wywołania embeddingu.
Wzbogacanie wieloetapowe. Sekwencja klasyfikacja, kierowanie, wydobywanie, wzbogacanie, przechowywanie jest stale używana w środowisku produkcyjnym. Dokumenty mogą najpierw zostać sklasyfikowane – dokumentacja techniczna jest kierowana do ekstraktora entytetów dostosowanego do treści technicznych, a dokumenty polityczne do innego. Każda z tych ścieżek dodaje kolumny wzbogacające specyficzne dla swojego typu, a strumienie łączą się przed podziałem na fragmenty i zastosowaniem embeddingu, dzięki czemu reszta procesu jest wspólna dla wszystkich.
Każdy etap dodaje nowe kolumny, więc kolejne operatory widzą coraz bogatszą tabelę. Nic z tego nie polega na specjalnych trybach ani flagach – są to zwykłe operatory połączone ze sobą, co stanowi zaletę modelu operatorów.
Trzy sposoby jego uruchomienia
To narzędzie to coś więcej niż tylko tool CLI i pasuje do różnych kontekstów inżynieryjnych.
CLI. To najszybsza droga do uruchomionego pipeline’a:
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 jest przydatny podczas tworzenia przepływów: naruszenia zależności, brak wymaganych kolumn oraz błędy typu konfiguracji ujawniają się jeszcze przed odczytaniem lub zapisaniem jakichkolwiek danych. Wykonywanie go w środowisku CI przed wdrożeniem zmian w przepływie to dobra praktyka.
Biblioteka Python. Gdy pipeline’y wymagają uruchomienia z kodu aplikacji:
from docpipe.lib.docpipe_flow_manager import DocpipeFlowManager
manager = DocpipeFlowManager(flow_file="pipeline.json")
manager.execute()
To nadaje się do kodu wplecionego w większe aplikacje, notatniki lub skrypty orkiestracji, które muszą zarządzać samym cyklem życia wykonywania.
REST API. Do działania jako usługa o długim czasie trwania:
uvicorn docpipe.api.main:app - host 0.0.0.0 - port 8000
Serwer FastAPI udostępnia trzy grupy tras: /api/v1/flows do zarządzania przepływami, /api/v1/job_runs do kontrolowania wykonywania zadań oraz /api/v1/operators do wyszukiwania operatorów. Użytkownicy uwierzytelniają się za pomocą LDAP lub OAuth2/OIDC przy użyciu tokenów JWT, a dokumentacja OpenAPI/Swagger jest generowana automatycznie pod adresem /docs. Jest to interfejs usługi wspólnej, którą używają różne zespoły, która jest uruchamiana przez zdarzenia zewnętrzne, a także do której inne zespoły wysyłają zapytania w celu uzyskania informacji o stanie i wynikach.
Rozszerzanie o własnych operatorach
Dla trwałego wdrożenia w rzeczywistych organizacjach jest to bez wątpienia najważniejsza część projektu.
Umowa jest na tyle zwięzła, że jej opis zajmuje około 30 wierszy. Klasa w Pythonie spełnia warunki, jeśli dziedziczy od AbstractOperator oraz implementuje metody transform(), get_metadata() i get_required_features(). Łączy się z każdym wbudowanym operatorem, pojawia się w wyniku wyświetlonym przez opcję --list-operators, a plik JSON opisujący przepływ odnosi się do niej za pomocą short_name, tak jak do wbudowanego operatora.
Rejestracja odbywa się na trzy sposoby, dostosowane do etapu rozwoju.
Ścieżka w systemie plików służy do celów rozwojowych. Wskazz zmienną środowiskową na lokalną katalog:
DOCPIPE_CUSTOM_OPERATORS=./my_operators docling-pipelines - flow-file pipeline.json
Wszystko, co znajduje się w tym katalogu, jest importowane podczas uruchamiania. Nie jest potrzebna żadna instalacja, dzięki czemu proces jest szybki.
Pakiet Pythona służy do dystrybucji. Opublikuj operatorów jako pakiet pip z odpowiednim punktem wejścia, zainstaluj go w tym samym środowisku, a one same się zarejestrują. W ten sposób operatorzy mogą dotrzeć do innych zespołów bez konieczności ingerencji w podstawową bazę kodu.
S3 URI to opcja typu enterprise. Ustaw DOCPIPE_CUSTOM_OPERATORS na odległy pakiet operatora, a framework pobraje go i zarejestruje w czasie wykonywania. Pozwala to na centralne zarządzanie operatorami we wszystkich implementacjach: jedno miejsce do aktualizacji, a każda implementacja przyjmuje zmiany przy następnym uruchomieniu.
Logika specyficzna dla przedsiębiorstw pozostaje zatem poza jądrzem frameworka. Procesory sprawdzania zgodności, narzędzia do wyodrębniania entytetów specyficznych dla danej dziedziny oraz własne modele oceny mogą być wszystkie zapakowane jako operatorzy, połączone z elementami wbudowanymi w plik JSON opisujący przepływ danych, i wdrożone bez konieczności modyfikacji całego frameworka.
Kwestie operacyjne
Umiarkowanym pytaniem dotyczącym takiego narzędzia jest to, czy stanowi ono dobry prototyp, czy faktycznie nadaje się do użycia w produkcji. Szczegóły poniżej.
Obserwowalność. Wbudowana jest natywna obsługa OpenTelemetry, a każda eksploatacja operatora tworzy element typu span. Logi mają postać zstrukturyzowanego JSON z informacjami o powiązaniach śledzenia, zawierającymi na każdej linii wartości trace_id i span_id. Działa to z Jaegerem, Grafana Tempo, Datadog i AWS X-Ray bez konieczności konfiguracji po stronie aplikacji, a obciążenie jest zerowe, gdy telemetry jest wyłączona.
Przetwarzanie stopniowe. Pipeline wykorzystuje deterministyczny identyfikator z DocIdHashOperator do śledzenia tego, co zostało już przetworzone. Podczas ponownego uruchomienia obsługiwane są tylko nowe lub zmienione dokumenty, a ten sam dokument wysłany dwukrotnie otrzymuje ten sam hash i jest pomijany. Ustawienie force_ingest: true w global_config anuluje to i przetwarza wszystko ponownie. Metadane dotyczące przetwarzania stopniowego mogą znajdować się w DuckDB (standardowo), PostgreSQL lub w systemie plików.
Bезpieczeństwo. ACLOperator prezentuje metadane uprawnień jako kolumnę w celu dalszego egzekwowania zasad, a PIIAndHAPAnnotator znajduje i maskuje treści wrażliwe, zanim dotrą one do magazynu wektorów. Sekrety w konfiguracji przepływu mogą pochodzić z HashiCorp Vault: ścieżka taka jak vault://path/to/secret zapisana w pliku JSON jest rozwiązywana przez silnik podczas wykonywania procesu. Dostęp do REST API jest zabezpieczony za pomocą LDAP oraz OAuth2/OIDC.
Skalowanie. Wykonywanie zadań odbywa się w trzech trybach, od laptopa po zdystrybuowaną flotę urządzeń. Domyślnym rozwiązaniem jest grupa wątków wewnątrz procesu, która jest szybka, prosta i nie wymaga żadnych zależności. Grupa procesów uruchamia operatory w oddzielnych procesach do zadań obciążających CPU na jednym węźle. Dla zadań produkcyjnych w środowisku zdystrybuowanym istnieje grupa zadań Prefect uruchamiająca kontenery Docker, co umożliwia horyzontalne skalowanie oraz centralne zarządzanie procesami. Ustawienia mikrogrupowania kontrolują zużycie pamięci we wszystkich trybach, co ma znaczenie, gdy duże dokumenty są przetwarzane za pomocą technologii LLM do wydobywania informacji.
Dalej co robić
Projekt jest hostowany na GitHubie w ramach organizacji IBM, a jego przewodnik szybkiego rozpoczęcia pracy to najszybsza droga do uruchomienia pipeline’a. Obszary wymagające dokładniejszego zbadania to warstwa ekstrakcji (integracja z Doclingiem, pipeline’y VLM dla dokumentów bogatych w obrazy, ASR dla audio i wideo), pisanie własnego operatora o złożonej logice – od pierwszej deklaracji klasy po testowanie, rejestrację i wdrożenie, uruchamianie Prefect z Dockerem dla dużych obciążeń pracy, a także opcje mapowania pól i szablonów schematu w VectorDBOperator dla OpenSearch.
Literatura pokrewna
- Notatki praktyczne: Budowa lokalnego pipeline RAG klasy produkcyjnej — 100% darmowe, bez ograniczeń — Szczegółowy przewodnik po notatkach praktycznych: Budowa lokalnego pipeline RAG klasy produkcyjnej — 100% darmowe, bez ograniczeń: umowy, sprawdzenia oraz miejsca na kod do wklejenia dla zespołów wdrażających ten model.
- Hybrydowe wyszukiwanie w Postgres dla asystenta zdrowotnego: schemat, dzielenie na fragmenty, RRF i ocena — Budowa warstwy wyszukiwania dla asystenta zdrowotnego w Postgres: dzielenie na fragmenty z uwzględnieniem struktury, hybrydowe wyszukiwanie za pomocą pgvector z RRF, kwantyfikacja binarna, progi i ocena.