TopolearnRepeatable, incremental ingestion is what keeps a RAG index correct as source data changes.
An `IngestionPipeline` applies a list of transformations to documents, in order, and can write the result straight to a vector store. Transformations include node parsers, metadata extractors and the embedding model itself.
pipeline = IngestionPipeline(
transformations=[SentenceSplitter(chunk_size=512), TitleExtractor(), embed_model],
vector_store=vector_store,
docstore=SimpleDocumentStore(),
)
pipeline.run(documents=documents)Why use it instead of from_documents:
- Caching. Each transformation's output is cached by input, so re-running on unchanged data skips the expensive steps.
- De-duplication. With a
docstoreattached, the pipeline remembers each document's id and content hash. Unchanged documents are skipped; changed ones are re-processed and their old nodes replaced (the defaultupsertsstrategy). Stable document ids, such as fromfilename_as_id=True, make this work. - One place for your ingestion logic, which you can run from a scheduled job.
Metadata extractors use an LLM to add metadata to nodes, for example TitleExtractor, SummaryExtractor or QuestionsAnsweredExtractor. They can improve retrieval but cost an LLM call per node, so use them where they measurably help.
Save the pipeline's cache and docstore with pipeline.persist("./pipeline_storage") and load them with pipeline.load(...) on the next run. pipeline.run(..., num_workers=4) processes in parallel.
After ingestion, query with VectorStoreIndex.from_vector_store(vector_store).
Example
from llama_index.core import Document, Settings, VectorStoreIndex
from llama_index.core.ingestion import IngestionPipeline
from llama_index.core.node_parser import SentenceSplitter
from llama_index.core.storage.docstore import SimpleDocumentStore
import chromadb
from llama_index.vector_stores.chroma import ChromaVectorStore
collection = chromadb.EphemeralClient().get_or_create_collection("docs") # in-memory demo
vector_store = ChromaVectorStore(chroma_collection=collection) # or Qdrant, pgvector...
pipeline = IngestionPipeline(
transformations=[SentenceSplitter(chunk_size=512, chunk_overlap=50),
Settings.embed_model], # embedding is the last step
docstore=SimpleDocumentStore(), # enables de-duplication
vector_store=vector_store,
)
docs = [Document(text="Refunds take 5 days.", id_="faq.md"),
Document(text="Office in Oslo.", id_="about.md")]
print(len(pipeline.run(documents=docs))) # 2 nodes processed
print(len(pipeline.run(documents=docs))) # 0 (unchanged, skipped)
docs[0] = Document(text="Refunds take 3 days.", id_="faq.md")
print(len(pipeline.run(documents=docs))) # 1 (only the changed doc)
pipeline.persist("./pipeline_storage")
index = VectorStoreIndex.from_vector_store(vector_store)Remember
- Pipelines chain parsers, extractors and the embedding model
- A docstore enables skipping unchanged documents and replacing changed ones
- Persist cache and docstore between runs
Watch out for
- Running a pipeline with a docstore but unstable document ids (random UUIDs), so every run looks like new documents.