How to build an ingestion pipeline in LlamaIndex

Search quality depends on the nodes that reach an index, including how source text is split and which embedding model encodes each chunk. A named IngestionPipeline keeps those transformations in a repeatable order before the nodes cross into a VectorStoreIndex.

The SentenceSplitter transformation runs before the deterministic BaseEmbedding implementation. The returned nodes therefore carry both chunked text and vectors when the index accepts them.

The small keyword embedding keeps the smoke test local and API-free. Replace it with the application's production embedding integration before indexing real data, and use that same model when creating or querying the index so stored and query vectors share one vector space.

1. Create ingestion_pipeline.py with the document, embedding, ingestion, splitting, and indexing imports.

   ingestion_pipeline.py

   from typing import List
    
   from llama_index.core import Document, VectorStoreIndex
   from llama_index.core.embeddings import BaseEmbedding
   from llama_index.core.ingestion import IngestionPipeline
   from llama_index.core.node_parser import SentenceSplitter

   Python's standard library provides List. The llama-index-core package provides Document, VectorStoreIndex, BaseEmbedding, IngestionPipeline, and SentenceSplitter.
   Related: How to install LlamaIndex with pip

2. Add the deterministic KeywordEmbedding class below the imports.

   class KeywordEmbedding(BaseEmbedding):
       @classmethod
       def class_name(cls) -> str:
           return "KeywordEmbedding"
    
       def _vector(self, text: str) -> List[float]:
           lowered = text.lower()
           return [
               float(sum(word in lowered for word in ("refund", "billing", "invoice"))),
               float(sum(word in lowered for word in ("shipping", "delivery", "courier"))),
               float(sum(word in lowered for word in ("password", "account", "login"))),
           ]
    
       def _get_text_embedding(self, text: str) -> List[float]:
           return self._vector(text)
    
       def _get_query_embedding(self, query: str) -> List[float]:
           return self._vector(query)
    
       async def _aget_query_embedding(self, query: str) -> List[float]:
           return self._get_query_embedding(query)

   The three vector positions represent billing, shipping, and account terms. A production embedding integration replaces this class without changing the pipeline order.

3. Define two source documents below the embedding class.

   documents = [
       Document(
           text="Refund requests follow the billing review process before approval.",
           metadata={"source": "billing.txt"},
       ),
       Document(
           text="Delayed deliveries are escalated to the shipping and courier desk.",
           metadata={"source": "shipping.txt"},
       ),
   ]

   The source metadata stays attached to each resulting node, which allows retrieval results to identify their origin.

4. Configure the splitter and embedding transformations below the document list.

   embedding = KeywordEmbedding()
   pipeline = IngestionPipeline(
       transformations=[
           SentenceSplitter(chunk_size=128, chunk_overlap=16),
           embedding,
       ]
   )

   IngestionPipeline applies transformations in list order. Nodes entering a vector index or vector store require the embedding stage.

5. Run the configured pipeline below the transformation block to create nodes from the source documents.

   nodes = pipeline.run(documents=documents)

6. Create an in-memory vector index from the pipeline nodes below the pipeline block.

   index = VectorStoreIndex(nodes, embed_model=embedding)

7. Retrieve the closest node for a refund question below the index block.

   retriever = index.as_retriever(similarity_top_k=1)
   match = retriever.retrieve("Where is a refund request reviewed?")[0].node

8. Add fail-capable assertions and concise output below the retrieval block.

   assert len(nodes) == 2
   assert all(node.embedding is not None for node in nodes)
   assert match.metadata["source"] == "billing.txt"
    
   print(f"pipeline_nodes={len(nodes)}")
   print(f"embedded_nodes={sum(node.embedding is not None for node in nodes)}")
   print(f"top_source={match.metadata['source']}")
   print(f"top_text={match.get_content(metadata_mode='none')}")

   The process stops with an AssertionError if chunking changes the node count, an embedding is missing, or retrieval returns the shipping document.

9. Confirm the completed ingestion_pipeline.py file contains the constructed sections in dependency order.

   ingestion_pipeline.py

   from typing import List
    
   from llama_index.core import Document, VectorStoreIndex
   from llama_index.core.embeddings import BaseEmbedding
   from llama_index.core.ingestion import IngestionPipeline
   from llama_index.core.node_parser import SentenceSplitter
    
    
   class KeywordEmbedding(BaseEmbedding):
       @classmethod
       def class_name(cls) -> str:
           return "KeywordEmbedding"
    
       def _vector(self, text: str) -> List[float]:
           lowered = text.lower()
           return [
               float(sum(word in lowered for word in ("refund", "billing", "invoice"))),
               float(sum(word in lowered for word in ("shipping", "delivery", "courier"))),
               float(sum(word in lowered for word in ("password", "account", "login"))),
           ]
    
       def _get_text_embedding(self, text: str) -> List[float]:
           return self._vector(text)
    
       def _get_query_embedding(self, query: str) -> List[float]:
           return self._vector(query)
    
       async def _aget_query_embedding(self, query: str) -> List[float]:
           return self._get_query_embedding(query)
    
    
   documents = [
       Document(
           text="Refund requests follow the billing review process before approval.",
           metadata={"source": "billing.txt"},
       ),
       Document(
           text="Delayed deliveries are escalated to the shipping and courier desk.",
           metadata={"source": "shipping.txt"},
       ),
   ]
    
   embedding = KeywordEmbedding()
   pipeline = IngestionPipeline(
       transformations=[
           SentenceSplitter(chunk_size=128, chunk_overlap=16),
           embedding,
       ]
   )
   nodes = pipeline.run(documents=documents)
    
   index = VectorStoreIndex(nodes, embed_model=embedding)
   retriever = index.as_retriever(similarity_top_k=1)
   match = retriever.retrieve("Where is a refund request reviewed?")[0].node
    
   assert len(nodes) == 2
   assert all(node.embedding is not None for node in nodes)
   assert match.metadata["source"] == "billing.txt"
    
   print(f"pipeline_nodes={len(nodes)}")
   print(f"embedded_nodes={sum(node.embedding is not None for node in nodes)}")
   print(f"top_source={match.metadata['source']}")
   print(f"top_text={match.get_content(metadata_mode='none')}")

10. Run the completed ingestion pipeline.

    $ python3 ingestion_pipeline.py
    pipeline_nodes=2
    embedded_nodes=2
    top_source=billing.txt
    top_text=Refund requests follow the billing review process before approval.