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 RAG pipeline that processes raw documents into stable, queryable chunks, retrieves relevant information through vector similarity, and generates grounded answers with validated citations. Understand the importance of repeatable ingestion, separation of retrieval and generation, and structured output validation to ensure reliable AI responses.

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 ...