""" Generate the Jupyter notebook for Lab 07: The RAG Mega-Lab. Run: python generate_lab07_notebook.py """ import nbformat as nbf nb = nbf.v4.new_notebook() nb.metadata = { "kernelspec": { "display_name": "Python 3", "language": "python", "name": "python3", }, "language_info": { "name": "python", "version": "3.10.0", }, } cells = [] # ── Install & Setup ── cells.append(nbf.v4.new_code_cell("""# 1. Install required packages !pip install -q chromadb sentence-transformers google-genai rich python-dotenv # 2. Download Data from GitHub (if running in Colab) import os if not os.path.exists("data/policies"): print("Downloading data from GitHub...") !mkdir -p data/policies !wget -q https://raw.githubusercontent.com/KamyarZeinalipour/zucchetti-llm-course/main/lab_07_evaluation/data/golden_dataset.json -O data/golden_dataset.json !wget -q https://raw.githubusercontent.com/KamyarZeinalipour/zucchetti-llm-course/main/lab_07_evaluation/data/policies/data_privacy_policy.txt -O data/policies/data_privacy_policy.txt !wget -q https://raw.githubusercontent.com/KamyarZeinalipour/zucchetti-llm-course/main/lab_07_evaluation/data/policies/expense_policy.txt -O data/policies/expense_policy.txt !wget -q https://raw.githubusercontent.com/KamyarZeinalipour/zucchetti-llm-course/main/lab_07_evaluation/data/policies/onboarding_guide.txt -O data/policies/onboarding_guide.txt !wget -q https://raw.githubusercontent.com/KamyarZeinalipour/zucchetti-llm-course/main/lab_07_evaluation/data/policies/remote_work_policy.txt -O data/policies/remote_work_policy.txt !wget -q https://raw.githubusercontent.com/KamyarZeinalipour/zucchetti-llm-course/main/lab_07_evaluation/data/policies/sick_leave_policy.txt -O data/policies/sick_leave_policy.txt print("✓ Data downloaded successfully!") # 3. Colab API Key Setup try: from google.colab import userdata key = userdata.get('GOOGLE_API_KEY') if key: os.environ["GOOGLE_API_KEY"] = str(key).strip() print("✓ API key loaded from Colab Secrets") else: print("⚠ GOOGLE_API_KEY is empty in Colab Secrets") except Exception: print("Not in Colab — will use .env file") """)) # ── Title ── cells.append(nbf.v4.new_markdown_cell("""# 🔬 Lab 07: The RAG Mega-Lab — Step-by-Step Teaching Walkthrough ## Lecture 8 — RAG Evaluation & Observability This notebook is structured into **7 step-by-step interactive inspection cells**. Run each cell one at a time to demonstrate and explain every RAG technique learned in Lectures 4–8. | Part | Topic & Technique | Lecture | |------|-------------------|---------| | **Part 1** | Naive RAG & Vector Search Inspection | L4 | | **Part 2** | Golden Datasets, Retrieval Metrics & LLM-as-Judge | L8 | | **Part 3** | Two-Stage Retrieval with Cross-Encoder Re-ranking | L5 | | **Part 4** | HyDE (Hypothetical Document Embeddings) Inspection | L5 | | **Part 5** | Knowledge Graph Extraction & Hybrid Traversal | L6 | | **Part 6** | CRAG Self-Correcting Retrieval & Query Rewriting | L7 | | **Part 7** | Structured Traces, Leaderboard & Quality Gates | L8 | """)) # ── Imports & Setup ── cells.append(nbf.v4.new_code_cell("""import os import sys import json import time from pathlib import Path from dotenv import load_dotenv from rich.console import Console from rich.panel import Panel from rich.table import Table from rich import box console = Console() # Load environment variables load_dotenv(Path(".").parent / ".env") """)) # ── LLM Client ── cells.append(nbf.v4.new_markdown_cell("""## LLM Client Setup Connects to Google Gemini (`gemini-3.1-flash-lite`, `gemini-2.5-flash`, `gemini-2.0-flash-lite`, etc.) with automatic fallback. """)) cells.append(nbf.v4.new_code_cell('''def get_llm_client(): google_key = os.getenv("GOOGLE_API_KEY", "").strip() if google_key and len(google_key) > 5 and not google_key.startswith("your-"): try: from google import genai client = genai.Client(api_key=google_key) model_candidates = [ os.getenv("GOOGLE_MODEL", "").strip(), "gemini-3.1-flash-lite", "gemini-3.1-flash", "gemini-2.5-flash-lite", "gemini-2.5-flash", "gemini-2.0-flash-lite", "gemini-2.0-flash", "gemini-1.5-flash-lite", "gemini-1.5-flash", ] model_candidates = [m for m in model_candidates if m] working_model = None last_err = None for candidate in model_candidates: try: test = client.models.generate_content( model=candidate, contents="Say OK", config={"max_output_tokens": 5}, ) if test and test.text: working_model = candidate break except Exception as e: last_err = e continue if working_model: console.print(f"[bold green]✓[/] Connected to Gemini Model: [cyan]{working_model}[/]") def generate_gemini(prompt: str, system: str = None) -> str: full_prompt = f"{system}\\n\\n{prompt}" if system else prompt for attempt in range(3): try: response = client.models.generate_content( model=working_model, contents=full_prompt, config={ "temperature": 0.2, "max_output_tokens": 800, }, ) return response.text except Exception as e: if "429" in str(e) or "RESOURCE_EXHAUSTED" in str(e): time.sleep((2 ** attempt) * 5) else: raise raise RuntimeError("Gemini API rate limit exceeded.") return generate_gemini else: console.print(f"[yellow]⚠ Gemini models failed. Last error: {last_err}[/]") except Exception as e: console.print(f"[yellow]⚠ Gemini setup failed: {e}[/]") # DeepSeek fallback ds_key = os.getenv("DEEPSEEK_API_KEY", "").strip() if ds_key and not ds_key.startswith("your-"): from openai import OpenAI client = OpenAI( base_url=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"), api_key=ds_key, ) model_name = os.getenv("DEEPSEEK_CHAT_MODEL", "deepseek-chat") console.print(f"[bold green]✓[/] Connected to DeepSeek: [cyan]{model_name}[/]") def generate_deepseek(prompt: str, system: str = None) -> str: messages = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": prompt}) response = client.chat.completions.create( model=model_name, messages=messages, temperature=0.2, max_tokens=800, ) return response.choices[0].message.content return generate_deepseek print("Please set GOOGLE_API_KEY in Colab Secrets or .env file") sys.exit(1) generate_fn = get_llm_client() ''')) # ── Part 1 ── cells.append(nbf.v4.new_markdown_cell("""--- ## 🎬 Part 1: Naive RAG & Pipeline Inspection (Lecture 4) **Teaching Goal:** Inspect raw documents, chunk boundaries, ChromaDB vector distances, and final prompt. """)) cells.append(nbf.v4.new_code_cell("""# 1. Load Documents docs_dir = Path("data/policies") documents = [] for filepath in sorted(docs_dir.glob("*.txt")): content = filepath.read_text(encoding="utf-8") documents.append({ "filename": filepath.name, "title": filepath.stem.replace("_", " ").title(), "content": content, }) # 2. Chunking chunks = [] chunk_id = 0 chunk_size, overlap = 400, 50 for doc in documents: text = doc["content"] start = 0 while start < len(text): end = start + chunk_size chunk_text = text[start:end] if end < len(text): last_period = chunk_text.rfind(". ") if last_period > chunk_size * 0.5: chunk_text = chunk_text[:last_period + 1] end = start + last_period + 1 chunks.append({ "text": chunk_text.strip(), "source": doc["filename"], "title": doc["title"], "chunk_id": f"chunk_{chunk_id}", }) chunk_id += 1 start = end - overlap print(f"Loaded {len(documents)} documents → {len(chunks)} chunks.") t_chunks = Table(title="Sample Extracted Chunks", box=box.ROUNDED) t_chunks.add_column("Chunk ID", style="cyan") t_chunks.add_column("Source Document") t_chunks.add_column("Length") t_chunks.add_column("Snippet") for c in chunks[:3]: t_chunks.add_row(c["chunk_id"], c["source"], str(len(c["text"])), c["text"][:60] + "...") console.print(t_chunks) # 3. Vector Database (ChromaDB) import chromadb client = chromadb.Client() try: client.delete_collection("hr_policies") except Exception: pass collection = client.create_collection("hr_policies") collection.add( ids=[c["chunk_id"] for c in chunks], documents=[c["text"] for c in chunks], metadatas=[{"source": c["source"], "title": c["title"]} for c in chunks], ) # 4. Inspect Search test_query = "How many sick days am I entitled to per year?" results = collection.query(query_texts=[test_query], n_results=3) t_table = Table(title=f"Vector Results for: '{test_query}'", box=box.SIMPLE_HEAD) t_table.add_column("Rank", width=6) t_table.add_column("L2 Distance", width=14) t_table.add_column("Source", width=22) t_table.add_column("Snippet", width=50) for i in range(len(results["documents"][0])): t_table.add_row(f"#{i+1}", f"{results['distances'][0][i]:.4f}", results['metadatas'][0][i]['source'], results['documents'][0][i][:70] + "...") console.print(t_table) """)) # ── Part 2 ── cells.append(nbf.v4.new_markdown_cell("""--- ## 🎬 Part 2: Golden Dataset & Metrics Evaluation (Lecture 8) **Teaching Goal:** Step-by-step math for Precision@K, Recall@K, MRR, Hit Rate, and LLM-as-Judge JSON output. ### Quick Metric Reference | Metric | What It Measures | Formula (simplified) | |--------|-----------------|---------------------| | **Precision@K** | How many of the K retrieved chunks are relevant? | `relevant_in_top_K / K` | | **Recall@K** | Did we find all relevant chunks in top K? | `relevant_in_top_K / total_relevant` | | **MRR** | How high is the *first* relevant chunk ranked? | `1 / rank_of_first_relevant` | | **Hit Rate@K** | Is there *at least one* relevant chunk in top K? | `1 if any relevant in top K else 0` | > 💡 **Tip:** We compute these metrics for a single question first, then aggregate across the full Golden Dataset in Part 7. """)) cells.append(nbf.v4.new_code_cell("""# Load Golden Dataset with open("data/golden_dataset.json") as f: golden_data = json.load(f)["dataset"] print(f"Dataset contains {len(golden_data)} Q&A pairs. Example:") console.print(Panel(json.dumps(golden_data[0], indent=2), title="Raw Golden Dataset Item", border_style="yellow")) item = golden_data[0] print(f"\\nQuestion: {item['question']}") print(f"Ground Truth: {item['ground_truth']}") print(f"Expected Source: {item['expected_source']}\\n") # Run search res = collection.query(query_texts=[item['question']], n_results=3) sources = [m['source'] for m in res['metadatas'][0]] matches = [1 if s == item['expected_source'] else 0 for s in sources] precision = sum(matches) / len(sources) recall = 1.0 if sum(matches) > 0 else 0.0 hit = 1 if sum(matches) > 0 else 0 mrr = max([1.0 / (j + 1) for j, m in enumerate(matches) if m == 1] or [0.0]) print(f"Retrieved Sources: {sources}") print(f"Matches against expected: {matches}") print(f"• Precision@3: {precision:.2f}") print(f"• Recall@3: {recall:.2f}") print(f"• MRR: {mrr:.2f}") print(f"• Hit Rate@3: {hit:.2f}\\n") # LLM Judge Test context = "\\n\\n".join(res['documents'][0]) answer = generate_fn(f"Context:\\n{context}\\n\\nQuestion: {item['question']}") print("Generated RAG Answer:") console.print(Panel(answer.strip(), title="LLM Answer", border_style="green")) judge_prompt = f\"\"\"Given: QUESTION: {item['question']} CONTEXT: {context} ANSWER: {answer} Evaluate the ANSWER based on two criteria: 1. Faithfulness: Is the answer factually supported by the CONTEXT? (1=yes, 0=no) 2. Relevance: Does the answer directly address the QUESTION? (1=yes, 0=no) Respond ONLY in JSON format: {{ "faithfulness_score": 1 or 0, "faithfulness_reasoning": "...", "relevance_score": 1 or 0, "relevance_reasoning": "..." }}\"\"\" print("LLM Judge Raw JSON Response:") print(generate_fn(judge_prompt).strip()) """)) # ── Part 3 ── cells.append(nbf.v4.new_markdown_cell("""--- ## 🎬 Part 3: Two-Stage Retrieval with Cross-Encoder Re-ranking (Lecture 5) **Teaching Goal:** Observe how the Cross-Encoder re-scores 9 candidates and shifts ranks. ### How Two-Stage Retrieval Works 1. **Stage 1 — Bi-Encoder (Fast):** ChromaDB uses a bi-encoder to independently embed the query and each chunk, then ranks by vector distance. This is fast but shallow — it only compares embeddings, not actual word meaning. 2. **Stage 2 — Cross-Encoder (Precise):** We take the top 9 candidates from Stage 1 and pass each `(query, chunk)` pair through a Cross-Encoder neural network. The Cross-Encoder reads the query and chunk *together* through all its attention layers and outputs a relevance score (logit). Higher score = more relevant. > 💡 **Why 9 candidates?** We over-fetch to give the Cross-Encoder a larger pool to re-sort. The final top 3 after re-ranking are often different from the initial top 3! """)) cells.append(nbf.v4.new_code_cell("""from sentence_transformers import CrossEncoder reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2") query = "What is the policy for expense reimbursement on client dinners?" # Bi-Encoder fetch 9 res = collection.query(query_texts=[query], n_results=9) candidates = [{"text": res["documents"][0][i], "source": res["metadatas"][0][i]["source"], "bi_rank": i + 1} for i in range(len(res["documents"][0]))] # Cross-Encoder re-score scores = reranker.predict([(query, c["text"]) for c in candidates]) for c, score in zip(candidates, scores): c["cross_score"] = float(score) reranked = sorted(candidates, key=lambda x: x["cross_score"], reverse=True) t_rank = Table(title="Re-ranking Rank Shift Table", box=box.ROUNDED) t_rank.add_column("Bi-Rank", justify="center") t_rank.add_column("Cross-Rank", justify="center") t_rank.add_column("Cross Score", justify="center") t_rank.add_column("Source") for idx, item in enumerate(reranked[:5], 1): t_rank.add_row(f"#{item['bi_rank']}", f"#{idx}", f"{item['cross_score']:.4f}", item["source"]) console.print(t_rank) """)) # ── Part 4 ── cells.append(nbf.v4.new_markdown_cell("""--- ## 🎬 Part 4: HyDE — Hypothetical Document Embeddings (Lecture 5) **Teaching Goal:** Inspect the generated hypothetical answer and see how it bridges vocabulary gaps. ### How HyDE Works 1. The user asks a question using **casual language** (e.g., *"What happens if I lose my laptop?"*). 2. Instead of searching directly, we ask the LLM to **imagine** a formal policy paragraph that would answer the question. 3. We then search ChromaDB using the **hypothetical document** as the query instead of the original question. > 💡 **Why does this help?** The hypothetical document uses the same formal vocabulary as the real policy documents, so the vector similarity is much higher! """)) cells.append(nbf.v4.new_code_cell("""query = "What happens if I lose my company laptop while traveling?" # Generate Hypothesis hyde_prompt = f"Write a formal policy paragraph that answers: {query}\\nPolicy text:" hypothetical_doc = generate_fn(hyde_prompt) print("Generated Hypothetical Document (LLM Imagination):") console.print(Panel(hypothetical_doc.strip(), border_style="magenta")) # Compare search: Direct vs HyDE res_orig = collection.query(query_texts=[query], n_results=3) res_hyde = collection.query(query_texts=[hypothetical_doc], n_results=3) t_hyde = Table(title="Direct Search vs HyDE Search", box=box.ROUNDED) t_hyde.add_column("Rank", width=6) t_hyde.add_column("Direct Search Source", width=22) t_hyde.add_column("Direct Snippet", width=30) t_hyde.add_column("HyDE Search Source", width=22) t_hyde.add_column("HyDE Snippet", width=30) for i in range(min(3, len(res_orig['documents'][0]))): t_hyde.add_row( f"#{i+1}", res_orig['metadatas'][0][i]['source'], res_orig['documents'][0][i][:40] + "...", res_hyde['metadatas'][0][i]['source'], res_hyde['documents'][0][i][:40] + "...", ) console.print(t_hyde) """)) # ── Part 5 ── cells.append(nbf.v4.new_markdown_cell("""--- ## 🎬 Part 5: Knowledge Graph Extraction & Traversal (Lecture 6) **Teaching Goal:** Inspect entity node extraction and entity co-occurrence mapping. ### What is a Knowledge Graph in RAG? Instead of relying purely on vector similarity, we build a **graph index** of key entities (e.g., "sick days", "GDPR", "VPN") and map them to the chunks where they appear. This lets us: - **Find related chunks** that vector search might miss (e.g., chunks about "medical certificate" are related to "sick days" even if the words don't overlap). - **Traverse connections** between concepts to build richer context for the LLM. > 💡 Below we build a simplified entity index. In production, tools like LlamaIndex or Neo4j automate this with LLM-based entity extraction. """)) cells.append(nbf.v4.new_code_cell("""entity_to_chunks = {} terms = ["sick days", "medical certificate", "hotel", "expense", "gdpr", "vpn"] for chunk in chunks: t_lower = chunk["text"].lower() for term in terms: if term in t_lower: entity_to_chunks.setdefault(term, []).append(chunk["chunk_id"]) t_graph = Table(title="Knowledge Graph Node Index", box=box.SIMPLE) t_graph.add_column("Entity Node", style="cyan") t_graph.add_column("Chunk Count", justify="center") t_graph.add_column("Sample Chunks") for entity, cids in entity_to_chunks.items(): t_graph.add_row(entity, str(len(cids)), ", ".join(cids[:3])) console.print(t_graph) """)) # ── Part 6 ── cells.append(nbf.v4.new_markdown_cell("""--- ## 🎬 Part 6: CRAG — Corrective RAG & Query Rewriting (Lecture 7) **Teaching Goal:** Observe LLM chunk grading and self-correction query rewriting. """)) cells.append(nbf.v4.new_code_cell("""ambiguous_query = "What is the policy for work stuff?" # Initial retrieve res = collection.query(query_texts=[ambiguous_query], n_results=3) t_initial = Table(title="Initial Retrieval (Ambiguous Query)", box=box.ROUNDED) t_initial.add_column("Rank", width=6) t_initial.add_column("Source", width=22) t_initial.add_column("Snippet") for i in range(len(res['documents'][0])): t_initial.add_row(f"#{i+1}", res['metadatas'][0][i]['source'], res['documents'][0][i][:70] + "...") console.print(t_initial) # Grade chunks_text = "\\n".join([f"- {doc[:100]}..." for doc in res['documents'][0]]) grade_prompt = f"Are these chunks relevant to '{ambiguous_query}'?\\n{chunks_text}\\nAnswer RELEVANT or IRRELEVANT for each. Output JSON list of strings: ['...', '...', '...']" print("\\n📝 LLM-as-a-Judge Grades:") print(generate_fn(grade_prompt).strip()) # Rewrite console.print("\\n[yellow]⚠ Warning: Chunks graded as IRRELEVANT. Triggering CRAG Query Rewrite...[/]") rewrite_prompt = f"Rewrite this ambiguous query to be specific for HR policy search: '{ambiguous_query}'. Return ONLY the rewritten query text." rewritten = generate_fn(rewrite_prompt).strip() console.print(f" [bold]Original:[/] {ambiguous_query}") console.print(f" [bold]Rewritten:[/] [green]{rewritten}[/]") # Second retrieval with rewritten query res2 = collection.query(query_texts=[rewritten], n_results=3) t_corrected = Table(title="Corrected Retrieval (Rewritten Query)", box=box.ROUNDED) t_corrected.add_column("Rank", width=6) t_corrected.add_column("Source", width=22) t_corrected.add_column("Snippet") for i in range(len(res2['documents'][0])): t_corrected.add_row(f"#{i+1}", res2['metadatas'][0][i]['source'], res2['documents'][0][i][:70] + "...") console.print("\\n") console.print(t_corrected) """)) # ── Part 7 ── cells.append(nbf.v4.new_markdown_cell("""--- ## 🎬 Part 7: Dynamic Leaderboard & Quality Gates (Lecture 8) **Teaching Goal:** Dynamically compare Baseline vs Re-ranked retrieval across the full Golden Dataset and evaluate quality thresholds. > 💡 This cell loops over all 5 retrievable questions in the Golden Dataset (excluding the out-of-domain trick question) and computes Hit Rate@3 and MRR in real time for both strategies. """)) cells.append(nbf.v4.new_code_cell("""t_comp = Table(title="🏆 Dynamic Strategy Comparison (Computed Real-Time)", box=box.DOUBLE_EDGE, width=100) t_comp.add_column("Metric", width=25) t_comp.add_column("Baseline (Naive RAG)", justify="center") t_comp.add_column("Re-ranked (Cross-Encoder)", justify="center") def eval_method(use_reranker=False): hits = 0 mrrs = [] retrieval_data = [d for d in golden_data if d['expected_source'] != "NONE"] for i_data in retrieval_data: res_eval = collection.query(query_texts=[i_data['question']], n_results=9 if use_reranker else 3) if use_reranker: try: cands = [{"s": res_eval["metadatas"][0][i]["source"], "t": res_eval["documents"][0][i]} for i in range(len(res_eval["documents"][0]))] scores_eval = reranker.predict([(i_data['question'], c["t"]) for c in cands]) for i in range(len(cands)): cands[i]['score'] = float(scores_eval[i]) cands = sorted(cands, key=lambda x: x['score'], reverse=True)[:3] srcs = [c['s'] for c in cands] except NameError: srcs = [m['source'] for m in res_eval['metadatas'][0]] else: srcs = [m['source'] for m in res_eval['metadatas'][0]] matches_eval = [1 if s == i_data['expected_source'] else 0 for s in srcs] if sum(matches_eval) > 0: hits += 1 mrr_eval = max([1.0 / (j + 1) for j, m in enumerate(matches_eval) if m == 1] or [0.0]) mrrs.append(mrr_eval) return hits / len(retrieval_data), sum(mrrs) / len(mrrs) with console.status("Evaluating Baseline on 5 retrieval questions..."): base_hit, base_mrr = eval_method(use_reranker=False) with console.status("Evaluating Re-ranked on 5 retrieval questions..."): rerank_hit, rerank_mrr = eval_method(use_reranker=True) t_comp.add_row("Hit Rate@3", f"{base_hit:.2f}", f"[bold green]{rerank_hit:.2f} ★[/]" if rerank_hit > base_hit else f"{rerank_hit:.2f}") t_comp.add_row("MRR", f"{base_mrr:.2f}", f"[bold green]{rerank_mrr:.2f} ★[/]" if rerank_mrr > base_mrr else f"{rerank_mrr:.2f}") console.print(t_comp) print("\\nQuality Gates:") print(f"{'✓ PASS' if rerank_hit >= 0.90 else '⚠ FAIL'} — Re-ranked Hit Rate@3: {rerank_hit:.2f} (Target: ≥0.90)") print(f"{'✓ PASS' if rerank_mrr >= 0.70 else '⚠ FAIL'} — Re-ranked MRR: {rerank_mrr:.2f} (Target: ≥0.70)") """)) nb.cells = cells # Write notebook output_path = "lab_07_evaluation/lab_07_evaluation.ipynb" with open(output_path, "w") as f: nbf.write(nb, f) print(f"✓ Generated interactive notebook: {output_path} ({len(cells)} cells)")