Search⌘ K
AI Features
Home
Courses
Applied AI Engineer: Build, Evaluate, and Operate Production LLM

Building an End-to-End RAG Pipeline

Explore how to build an end-to-end Retrieval-Augmented Generation pipeline by ingesting documents into a vector store, retrieving relevant chunks for a question, and generating answers with verified citations. Understand chunking, vector embeddings, context construction, retrieval validation, and integration of a structured generation model. By mastering this pipeline, you'll ensure AI responses are grounded, debuggable, and ready for production use.

We'll cover the following...

A working RAG flow has one observable end state. Given a question, answer(question) retrieves the top kk chunks, formats a CONTEXT block that labels each chunk, calls the chat model, then returns {answer, citations} where citations point back to the retrieved chunk markers. If we can print the retrieved chunk IDs plus sources and the returned citations match those IDs, we have grounding we can verify.

def ingest(documents: list[Document]) -> None:
    """Chunk, embed, and upsert documents into the vector store."""
def retrieve(question: str, k: int = 4, filters: dict | None = None) -> list[Chunk]:
    """Return the top-k chunks for a question, each with metadata and a score."""
def answer(question: str) -> dict:
    """Retrieve chunks, build a prompt, call the model, and return {answer, citations}."""
The pipeline reduced to its three testable units: no implementation yet, just the contract each one owes

  • Lines 1–2 (ingest): Takes raw documents and is responsible for everything up to "the store now has stable, queryable chunks"; nothing about questions or the model yet.

  • Lines 4–5 (retrieve): Takes a question and returns chunks with their metadata and score; nothing about prompts or generation yet.

  • Lines 7–8 (answer): The only function that touches the model, and it does so by composing the other two. If it returns a wrong answer, retrieve is the first thing to inspect, not the prompt.

The code behind these functions separates two data products that are easy to inspect. The retriever returns a list of chunks that each carry page_content plus metadata such as chunk_id and source, and the generation step returns a structured payload that includes citations referencing those chunk IDs. That separation is the reason debugging works, because we can log the chunk list before the model call and validate citations after the model call.

Ingestion that stays repeatable

With the contract visible, the next step is making ingestion deterministic so retrieval has stable targets. ingest() takes raw documents, splits them into chunks, attaches metadata, computes embeddings for each chunk, then upserts them into a named Chroma collection that we can persist on disk.

Two invariants keep ingestion repeatable across runs.

  • Stable chunk IDs so the same chunk does not get duplicated under a new identifier when we re-ingest.

  • Stable metadata schema so downstream code can always read fields like source, section, and chunk_id without conditional logic.

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
import hashlib
def load_and_split(source_id: str, text: str, chunk_size: int = 800, overlap: int = 120) -> list[Document]:
    splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=overlap)
    raw_chunks = splitter.create_documents([text])
    docs = []
    for i, chunk in enumerate(raw_chunks):
        content_hash = hashlib.sha256(chunk.page_content.encode("utf-8")).hexdigest()
        chunk_id = f"{source_id}::chunk-{i:05d}::{content_hash[:16]}"
        chunk.metadata.update({"source": source_id, "chunk_id": chunk_id})
        docs.append(chunk)
    return docs
Splitting documents into stably identified chunks, using a real splitter instead of a hand-written sliding window

  • Line 2 (from langchain_core.documents import Document): Imports LangChain's Document type, the standard container for page_content + metadata used across the whole pipeline.

  • Line 3 (import hashlib): Used to hash chunk content for a deterministic ID.

  • Line 5 (def load_and_split(...)): Takes a document's raw text plus a source_id, and returns a list of ready-to-embed Document chunks.

  • Line 6 (splitter = RecursiveCharacterTextSplitter(chunk_size=chunk_size, chunk_overlap=overlap)): Configures the splitter.

  • Line 7 (raw_chunks = splitter.create_documents([text])): Runs the split and returns a list of Document objects with page_content.

  • Lines 10–14 (the for loop): For each chunk, hashes its content, builds a stable chunk_id from source_id + index + hash, and writes source and chunk_id into the chunk's metadata dict. Re-ingesting the same source reproduces the same IDs, so a later upsert overwrites instead of duplicating.

from langchain_huggingface import HuggingFaceEmbeddings
from langchain_chroma import Chroma
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
vectorstore = Chroma(
    collection_name="internal_docs",
    embedding_function=embeddings,
    persist_directory="./chroma_store",
)
def ingest(documents: list[Document]) -> None:
    vectorstore.add_documents(documents, ids=[d.metadata["chunk_id"] for d in documents])
Embedding and writing chunks into a persistent vector store

  • ...