IngestionPipeline: transformări, cache și deduplicare în LlamaIndex & Data RAG

from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=50),
        OpenAIEmbedding(model="text-embedding-3-small"),
    ],
)
nodes = pipeline.run(documents=documents)

IngestionPipeline din llama_index.core.ingestion primește transformations: componente aplicate în ordine peste documente, de exemplu un node parser, extractori de metadate (TitleExtractor) și un model de embedding. run() întoarce nodurile rezultate sau, dacă ai dat vector_store, le inserează în baza vectorială; indexul se creează apoi cu VectorStoreIndex.from_vector_store(). Documentația recomandă pipeline-ul când vrei control asupra chunking-ului, metadatelor și embedding-urilor.

Fiecare combinație nod + transformare este hash-uită și pusă în cache. O rulare repetată peste aceleași date refolosește rezultatul în loc să recalculeze embedding-urile. Cache-ul local se salvează cu pipeline.persist("./pipeline_storage") și se încarcă cu load(); există și cache-uri la distanță (RedisCache, MongoDBCache, FirestoreCache) prin IngestionCache.

Cu un docstore atașat, pipeline-ul gestionează documentele după doc_id: ține o hartă doc_id → hash. Dacă are și vector_store, un document cu același ID dar conținut schimbat e reprocesat și actualizat (upsert), iar unul neschimbat e sărit. Fără bază vectorială poate doar să elimine intrările duplicate. run(num_workers=4) împarte loturile pe mai multe procese, iar await pipeline.arun(...) este varianta async.

Concepte cheie și caracteristici

Concept / Parametru	Descriere practică
transformations	Lista de transformări, aplicate în ordine. Dacă scrii într-o bază vectorială, ultima trebuie să fie un model de embedding.
vector_store	Baza vectorială în care se inserează nodurile la run().
docstore	Activează gestiunea documentelor (deduplicare și upsert după doc_id).
cache	Un IngestionCache; implicit cache local în memorie, salvat cu persist().
run(documents=..., num_workers=...)	Rulează pipeline-ul; num_workers folosește multiprocessing.Pool.
arun(...)	Varianta asincronă a lui run().
persist(dir) / load(dir)	Salvează și reîncarcă cache-ul (și docstore-ul, dacă există) pe disc.

Exemple practice și scenarii reale

Exemplul 1: Ingestie directă într-un Qdrant local

Adaptat după exemplul din documentație; modelul de embedding este ultimul pas, altfel indexul creat din bază nu funcționează.

import qdrant_client
from llama_index.core import VectorStoreIndex
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.embeddings.openai import OpenAIEmbedding
from llama_index.vector_stores.qdrant import QdrantVectorStore

client = qdrant_client.QdrantClient(path="./qdrant_data")
vector_store = QdrantVectorStore(client=client, collection_name="documente")

pipeline = IngestionPipeline(
    transformations=[
        SentenceSplitter(chunk_size=512, chunk_overlap=50),
        OpenAIEmbedding(model="text-embedding-3-small"),
    ],
    vector_store=vector_store,
)
pipeline.run(documents=documents)
index = VectorStoreIndex.from_vector_store(vector_store)

Exemplul 2: Reindexare doar pentru fișierele modificate

from llama_index.core import SimpleDirectoryReader
from llama_index.core.storage.docstore import SimpleDocumentStore

docs = SimpleDirectoryReader("./date", filename_as_id=True).load_data()
pipeline = IngestionPipeline(
    transformations=[SentenceSplitter(), OpenAIEmbedding(model="text-embedding-3-small")],
    docstore=SimpleDocumentStore(),
    vector_store=vector_store,
)
pipeline.run(documents=docs)
pipeline.persist("./pipeline_storage")   # la rularea următoare: pipeline.load(...)

Greșeli frecvente și capcane

• Pipeline cu vector_store, dar fără model de embedding în transformations. Nodurile ajung în bază fără vectori, iar indexul creat apoi din bază nu poate căuta.
• Documente fără doc_id stabil când folosești docstore. Fiecare rulare vede documente noi și nu deduplică nimic; cu SimpleDirectoryReader folosește filename_as_id=True.
• Cache-ul local nesalvat. Fără persist() și load(), cache-ul dispare la oprirea procesului, iar embedding-urile se recalculează.

Noțiuni și module înrudite

• documents-nodes
• sentence-splitter-node-parser
• vector-store-integrations
• vector-store-index