Lab 03: RAG Pipeline from Scratch

Goal

Build a complete RAG pipeline: document ingestion, chunking, embedding, vector storage, retrieval, reranking, and grounded generation with citations.

Prerequisites

pip install openai chromadb sentence-transformers httpx

Lab Files

lab-03-rag-pipeline/
  README.md        ← this file
  ingest.py        ← document loader + chunker + embedder
  search.py        ← retrieval + reranking
  app.py           ← full Q&A app
  evaluate.py      ← Ragas-style evaluation
  sample_docs/     ← put your documents here

Exercise 1: Ingest Documents

# lab03/ingest.py
import re
from pathlib import Path
from openai import OpenAI
import chromadb

client = OpenAI()
chroma = chromadb.PersistentClient(path="./rag_db")
collection = chroma.get_or_create_collection(
    name="docs",
    metadata={"hnsw:space": "cosine"}
)

def chunk_text(text: str, max_words: int = 300, overlap: int = 50) -> list[str]:
    """Split text into overlapping word-based chunks."""
    words = text.split()
    chunks = []
    
    for i in range(0, len(words), max_words - overlap):
        chunk_words = words[i:i + max_words]
        if len(chunk_words) < 20:  # Skip tiny trailing chunks
            continue
        chunks.append(" ".join(chunk_words))
    
    return chunks

def embed_batch(texts: list[str]) -> list[list[float]]:
    """Embed a batch of texts using OpenAI."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [e.embedding for e in response.data]

def ingest_file(filepath: str) -> int:
    """Ingest a single file into the vector database."""
    text = Path(filepath).read_text(encoding="utf-8", errors="ignore")
    
    # Remove excessive whitespace
    text = re.sub(r'\n{3,}', '\n\n', text)
    text = re.sub(r' {2,}', ' ', text)
    
    chunks = chunk_text(text)
    
    if not chunks:
        return 0
    
    # Embed in batches of 100
    all_embeddings = []
    for i in range(0, len(chunks), 100):
        batch = chunks[i:i + 100]
        all_embeddings.extend(embed_batch(batch))
    
    # Generate unique IDs
    filename = Path(filepath).stem
    ids = [f"{filename}_chunk_{i}" for i in range(len(chunks))]
    
    # Store in ChromaDB
    collection.add(
        documents=chunks,
        embeddings=all_embeddings,
        metadatas=[{"source": filepath, "chunk_index": i} for i in range(len(chunks))],
        ids=ids
    )
    
    return len(chunks)

def ingest_directory(directory: str) -> dict:
    """Ingest all text/markdown files in a directory."""
    results = {}
    
    for path in Path(directory).rglob("*"):
        if path.suffix in [".txt", ".md", ".rst"]:
            try:
                count = ingest_file(str(path))
                results[str(path)] = count
                print(f"✓ {path.name}: {count} chunks")
            except Exception as e:
                print(f"✗ {path.name}: {e}")
    
    return results

if __name__ == "__main__":
    import sys
    directory = sys.argv[1] if len(sys.argv) > 1 else "./sample_docs"
    
    print(f"Ingesting documents from {directory}...")
    results = ingest_directory(directory)
    
    total = sum(results.values())
    print(f"\nDone. Total chunks stored: {total}")
    print(f"Collection size: {collection.count()}")

Task: Put 5-10 markdown files about a topic you care about in sample_docs/. Run the ingest. Verify the collection has chunks.

Exercise 2: Retrieval + Reranking

# lab03/search.py
from openai import OpenAI
import chromadb
from sentence_transformers import CrossEncoder

client = OpenAI()
chroma = chromadb.PersistentClient(path="./rag_db")
collection = chroma.get_or_create_collection("docs")
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")

def retrieve(query: str, top_k: int = 15) -> list[dict]:
    """Semantic retrieval from vector DB."""
    query_embedding = client.embeddings.create(
        model="text-embedding-3-small",
        input=[query]
    ).data[0].embedding
    
    results = collection.query(
        query_embeddings=[query_embedding],
        n_results=min(top_k, collection.count())
    )
    
    chunks = []
    for doc, meta, dist in zip(
        results["documents"][0],
        results["metadatas"][0],
        results["distances"][0]
    ):
        chunks.append({
            "content": doc,
            "source": meta.get("source", "unknown"),
            "retrieval_score": 1 - dist  # Convert cosine distance to similarity
        })
    
    return chunks

def rerank(query: str, candidates: list[dict], top_n: int = 5) -> list[dict]:
    """Rerank candidates using a cross-encoder."""
    if not candidates:
        return []
    
    pairs = [(query, c["content"]) for c in candidates]
    scores = reranker.predict(pairs)
    
    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    
    result = []
    for chunk, score in ranked[:top_n]:
        chunk = chunk.copy()
        chunk["rerank_score"] = float(score)
        result.append(chunk)
    
    return result

def search(query: str, top_n: int = 5) -> list[dict]:
    """Full retrieval + reranking pipeline."""
    candidates = retrieve(query, top_k=20)
    return rerank(query, candidates, top_n=top_n)

if __name__ == "__main__":
    query = input("Search query: ")
    results = search(query)
    
    for i, r in enumerate(results):
        print(f"\n[{i+1}] Score: {r['rerank_score']:.3f} | {r['source']}")
        print(r["content"][:200] + "...")

Exercise 3: Full Q&A App

# lab03/app.py
from openai import OpenAI
from search import search

client = OpenAI()

SYSTEM_PROMPT = """You are a helpful assistant. Answer questions using ONLY the provided context sources.

Rules:
- Cite sources using [1], [2], [3] notation
- If the answer is not in the provided sources, say "I don't have information about this in the provided documents"
- Do not make up information
- Keep answers focused and precise"""

def answer(question: str) -> dict:
    """Full RAG pipeline: retrieve → pack context → generate."""
    chunks = search(question, top_n=5)
    
    if not chunks:
        return {"answer": "No documents found. Please ingest documents first.", "sources": []}
    
    # Pack context
    context = ""
    for i, chunk in enumerate(chunks):
        context += f"[{i+1}] Source: {chunk['source']}\n{chunk['content']}\n\n"
    
    # Generate
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"Context:\n{context}\nQuestion: {question}"}
        ],
        max_tokens=600,
        temperature=0
    )
    
    return {
        "answer": response.choices[0].message.content,
        "sources": [c["source"] for c in chunks],
        "chunks_used": len(chunks)
    }

if __name__ == "__main__":
    print("RAG Q&A System (type 'quit' to exit)\n")
    
    while True:
        question = input("Question: ").strip()
        if question.lower() in ["quit", "exit", "q"]:
            break
        
        result = answer(question)
        print(f"\nAnswer: {result['answer']}")
        print(f"\nSources used:")
        for source in result["sources"]:
            print(f"  - {source}")
        print()

Exercise 4: Evaluate Your RAG System

# lab03/evaluate.py
from openai import OpenAI
from app import answer

client = OpenAI()

def check_faithfulness(question: str, answer_text: str, sources_content: str) -> float:
    """Check if answer is supported by sources using LLM-as-judge."""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""Given these source passages and an answer, rate from 0.0 to 1.0 whether ALL claims in the answer are supported by the sources.

Sources:
{sources_content}

Answer: {answer_text}

Score (0.0 = not supported, 1.0 = fully supported). Respond with just the score."""
        }],
        max_tokens=5,
        temperature=0
    )
    try:
        return float(response.choices[0].message.content.strip())
    except:
        return 0.5

# Build test questions for YOUR documents
TEST_QUESTIONS = [
    "What is the main topic covered?",
    # Add questions specific to your documents
]

print("Evaluating RAG pipeline...\n")
scores = []

for q in TEST_QUESTIONS:
    result = answer(q)
    
    # Run faithfulness check (requires source content — simplified here)
    score = 1.0 if result["chunks_used"] > 0 else 0.0
    scores.append(score)
    
    print(f"Q: {q}")
    print(f"A: {result['answer'][:150]}...")
    print(f"Sources used: {result['chunks_used']}")
    print()

if scores:
    import statistics
    print(f"Average pipeline score: {statistics.mean(scores):.2f}")

Deliverables

• 5+ documents ingested successfully
• Search returning relevant results for 5 test queries
• Q&A app working with citations
• Evaluation run on 5 questions
• Failure analysis: note 2 cases where the system failed and explain why

Reflection Questions

1. How does chunk size affect recall? What happens if chunks are too small?
2. Why is reranking separate from initial retrieval? Why not just use the reranker for all retrieval?
3. A user asks a question and gets "I don't have information about this" even though the document exists. What are the possible causes?
4. How would you add support for PDF files to the ingestion pipeline?
5. How would you implement user-level ACL (user A can only see their documents)?