Building a RAG Pipeline End to End

What you will learn

Choose the right chunking strategy for your document type
Build a complete ingestion and retrieval pipeline in Python
Generate grounded answers with source citations

The four stages of RAG

RAG Pipeline — Ingest then Retrieve-and-Generate

  INGEST (run once, or on update)
  Documents
     |
     v
  Chunk (split into passages)
     |
     v
  Embed (encode each chunk)
     |
     v
  Index (store in vector DB)

  ---------------------------------

  QUERY (run per user question)
  User question
     |
     v
  Embed question
     |
     v
  Retrieve top-K chunks
     |
     v
  Build prompt [question + chunks]
     |
     v
  LLM generates answer with citations

Chunking strategies

Chunking is the most important decision in RAG. Too small: chunks lack context. Too large: irrelevant content pollutes the LLM context.

Strategy	How it works	Best for
Fixed-size	Split every N tokens with M overlap	Simple docs, fastest to implement
Recursive	Split by paragraph → sentence → word	General prose, code
Semantic	Split at meaning boundaries (embed sentences, split on drops)	Long-form documents with mixed topics
Late chunking	Embed the full document, then pool chunk embeddings	When chunk context depends on surrounding text

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,       # characters per chunk
    chunk_overlap=64,     # overlap between chunks (context continuity)
    separators=["\n\n", "\n", ". ", " ", ""],  # try these in order
)

with open("document.txt") as f:
    text = f.read()

chunks = splitter.split_text(text)
print(f"{len(chunks)} chunks from {len(text)} chars")

Complete ingestion pipeline

import uuid
from pathlib import Path
from sentence_transformers import SentenceTransformer
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from langchain_text_splitters import RecursiveCharacterTextSplitter

client = QdrantClient(host="localhost", port=6333)
model = SentenceTransformer("BAAI/bge-small-en-v1.5")
splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=64)

client.recreate_collection(
    collection_name="knowledge_base",
    vectors_config=VectorParams(size=384, distance=Distance.COSINE),
)


def ingest_file(path: Path):
    text = path.read_text(encoding="utf-8")
    chunks = splitter.split_text(text)
    print(f"  {path.name}: {len(chunks)} chunks")

    embeddings = model.encode(chunks, normalize_embeddings=True, show_progress_bar=False)

    points = [
        PointStruct(
            id=str(uuid.uuid4()),
            vector=emb.tolist(),
            payload={"text": chunk, "source": path.name, "chunk_index": i},
        )
        for i, (chunk, emb) in enumerate(zip(chunks, embeddings))
    ]
    client.upsert(collection_name="knowledge_base", points=points)


docs_dir = Path("./docs")
for doc in docs_dir.glob("*.txt"):
    ingest_file(doc)

print("Ingestion complete.")

Retrieval and generation with citations

import anthropic

anthropic_client = anthropic.Anthropic()


def retrieve(query: str, top_k: int = 5) -> list[dict]:
    q_vec = model.encode(query, normalize_embeddings=True).tolist()
    results = client.search(
        collection_name="knowledge_base",
        query_vector=q_vec,
        limit=top_k,
        with_payload=True,
    )
    return [{"text": r.payload["text"], "source": r.payload["source"], "score": r.score}
            for r in results]


def rag_answer(question: str) -> str:
    chunks = retrieve(question)

    context_parts = []
    for i, c in enumerate(chunks):
        context_parts.append(f"[{i+1}] (from {c['source']}, relevance {c['score']:.2f})\n{c['text']}")
    context = "\n\n".join(context_parts)

    prompt = f"""Answer the question using only the provided context. 
Cite sources by their number, e.g. [1] or [2][3].
If the context does not contain enough information, say so clearly.

Context:
{context}

Question: {question}"""

    response = anthropic_client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=1024,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.content[0].text


answer = rag_answer("How does MCP handle authentication?")
print(answer)
# -> "According to [1], MCP servers can implement OAuth 2.0..."

Query rewriting

Before retrieving, rewrite ambiguous queries to improve recall:

def rewrite_query(question: str) -> str:
    response = anthropic_client.messages.create(
        model="claude-haiku-4-5-20251001",
        max_tokens=128,
        messages=[{"role": "user", "content":
            f"Rewrite this search query to be more specific and retrieval-friendly. "
            f"Return only the rewritten query, nothing else.\n\nOriginal: {question}"
        }]
    )
    return response.content[0].text.strip()


original = "how does it authenticate?"
rewritten = rewrite_query(original)
# -> "How does MCP server authentication work?"

Chapter summary

Chunking strategy is the most impactful decision in RAG — 512 chars with 64 overlap is a good start
Include source and chunk_index in payload so you can cite and trace answers
Pass retrieved chunks with source labels in the prompt to enable citation
Query rewriting improves recall for vague or ambiguous questions before retrieval

Check your understanding

What is chunk overlap and why does it help?
Why does the RAG prompt instruct the model to "answer using only the provided context"?
What problem does query rewriting solve?