{ "nbformat": 4, "nbformat_minor": 5, "metadata": { "colab": { "provenance": [], "name": "session_4_lab.ipynb" }, "kernelspec": { "name": "python3", "display_name": "Python 3" }, "language_info": { "name": "python" } }, "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Lab 4 — Chat With YOUR Notes\n", "**Session 4 · Embeddings + RAG · TCE** — the weekend's main build (~60 lines, all yours)\n", "\n", "You need: your 2–3 documents (PDF or .txt). Upload via folder icon. **File → Save a copy in Drive** first." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": "# Cell 1 — setup\n%pip install -q -U google-genai pypdf numpy\nfrom getpass import getpass\nfrom google import genai\nfrom google.genai import types\nimport numpy as np, time, re\n\nclient = genai.Client(api_key=getpass(\"Gemini API key: \"))\nMODEL = \"gemini-flash-latest\" # the free tier's current Flash (July 2026 → Gemini 3.5 Flash). 503 'high demand'? swap to \"gemini-flash-lite-latest\".\nEMBED_MODEL = \"gemini-embedding-2\" # check https://ai.google.dev/gemini-api/docs/embeddings\n\ndef ask(contents, temperature=0.0):\n for attempt in range(4):\n try:\n return client.models.generate_content(\n model=MODEL, contents=contents,\n config=types.GenerateContentConfig(temperature=temperature)).text\n except Exception as e:\n if \"429\" in str(e) and attempt < 3:\n print(\"rate limited...\"); time.sleep(20*(attempt+1))\n else: raise\nprint(\"ready ✓\")" }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part A — Ingest: load → chunk → embed → store", "\n\n> **Keep documents small** — a few pages or one chapter (a few thousand words). Everything runs in Google's cloud, so your laptop's speed/RAM don't matter; this cap just keeps you comfortably inside the free daily quota." ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Cell 2 — load your document\n", "from pypdf import PdfReader\n", "\n", "FILENAME = \"my_notes.pdf\" # ← your file (.pdf or .txt)\n", "\n", "if FILENAME.endswith(\".pdf\"):\n", " text = \"\\n\".join(page.extract_text() or \"\" for page in PdfReader(FILENAME).pages)\n", "else:\n", " text = open(FILENAME, encoding=\"utf-8\").read()\n", "\n", "print(len(text), \"characters loaded\")\n", "print(text[:400]) # sanity: is this YOUR text, readable?" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Cell 3 — chunk: paragraphs, merged to a target size, with overlap\n", "def chunk_text(text, target=800, overlap=150):\n", " paras = [p.strip() for p in text.split(\"\\n\") if p.strip()]\n", " chunks, cur = [], \"\"\n", " for p in paras:\n", " if len(cur) + len(p) > target and cur:\n", " chunks.append(cur.strip())\n", " cur = cur[-overlap:] + \" \" + p # overlap keeps thoughts intact\n", " else:\n", " cur += \" \" + p\n", " if cur.strip(): chunks.append(cur.strip())\n", " return chunks\n", "\n", "chunks = chunk_text(text)\n", "\n", "# Keep it laptop- and free-tier-friendly: a few dozen chunks is plenty to learn RAG.\n", "# (Big textbook? Use one chapter. You can always raise this later.)\n", "MAX_CHUNKS = 60\n", "if len(chunks) > MAX_CHUNKS:\n", " print(f\"note: {len(chunks)} chunks -> capping to first {MAX_CHUNKS} (stays well inside the free tier).\")\n", " chunks = chunks[:MAX_CHUNKS]\n", "\n", "print(len(chunks), \"chunks\")\n", "print(\"--- sample chunk ---\\n\", chunks[len(chunks)//2][:300])" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Cell 4 — embed all chunks (batched), store as one numpy matrix\n", "def embed(texts):\n", " res = client.models.embed_content(model=EMBED_MODEL, contents=texts)\n", " return np.array([e.values for e in res.embeddings])\n", "\n", "vecs = []\n", "B = 20\n", "for i in range(0, len(chunks), B):\n", " vecs.append(embed(chunks[i:i+B]))\n", " print(f\"embedded {min(i+B, len(chunks))}/{len(chunks)}\")\n", " time.sleep(1) # be polite to the free tier\n", "\n", "chunk_vecs = np.vstack(vecs)\n", "chunk_vecs = chunk_vecs / np.linalg.norm(chunk_vecs, axis=1, keepdims=True) # normalize once\n", "print(\"vector store:\", chunk_vecs.shape, \"← this numpy array IS your vector database\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### ✓ Checkpoint 1 — chunk count + vector store shape printed.\n", "\n", "---\n", "## Part B — Semantic search (the R in RAG)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Cell 5 — search = one matrix multiply\n", "def search(query, k=3):\n", " qv = embed([query])[0]\n", " qv = qv / np.linalg.norm(qv)\n", " scores = chunk_vecs @ qv # cosine similarity, all chunks at once\n", " top = np.argsort(scores)[::-1][:k]\n", " return [(float(scores[i]), chunks[i]) for i in top]\n", "\n", "# sanity check with 3 queries about YOUR material:\n", "for q in [\"\", \"\", \"\"]:\n", " print(\"=\" * 60, \"\\nQ:\", q)\n", " for s, c in search(q):\n", " print(f\" {s:.2f} | {c[:110]}...\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Do the top chunks LOOK right?\n", "If not: chunks too big/small (tune `target`)? PDF extracted garbage (check Cell 2 output)? Query too vague?\n", "\n", "### ✓ Checkpoint 2 — three sane searches.\n", "\n", "---\n", "## Part C — The full RAG loop (the A and G)" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Cell 6 — grounded, cited answers\n", "RAG_TEMPLATE = \"\"\"Answer the question using ONLY the context below.\n", "Cite which chunk you used, like [1] or [2].\n", "If the answer is not in the context, reply exactly: \"I don't know based on the provided documents.\"\n", "\n", "CONTEXT:\n", "{context}\n", "\n", "QUESTION: {question}\"\"\"\n", "\n", "def rag_ask(question, k=3, show_chunks=False):\n", " hits = search(question, k)\n", " context = \"\\n\\n\".join(f\"[{i+1}] {c}\" for i, (s, c) in enumerate(hits))\n", " if show_chunks:\n", " for i, (s, c) in enumerate(hits): print(f\" [{i+1}] ({s:.2f}) {c[:80]}...\")\n", " return ask(RAG_TEMPLATE.format(context=context, question=question))\n", "\n", "print(rag_ask(\"\", show_chunks=True))" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Cell 7 — interrogate your own notes (5 real questions)\n", "for q in [\n", " \"\", \"\", \"\", \"\", \"\",\n", "]:\n", " print(\"=\" * 60, \"\\nQ:\", q, \"\\n\")\n", " print(rag_ask(q), \"\\n\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Part D — Break it honestly\n", "\n", "1. Ask something **definitely NOT in your documents** → does it say \"I don't know\"? (If it invents instead, strengthen the ONLY/escape-hatch lines — this is real prompt hardening.)\n", "2. Ask something whose answer is **split across two places** → does it get half the truth?\n", "\n", "### ✓ Checkpoint 3 — one honest failure + what you changed to fix (or why it's hard).\n", "\n", "---\n", "## Stretch goals" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Stretch 1 — RAG eval (your S2 harness, now grading your app)\n", "rag_tests = [\n", " {\"q\": \"\", \"expected\": \"\"},\n", " # 4 more...\n", "]\n", "def norm(s): return re.sub(r\"[^a-z0-9 ]\", \"\", s.lower())\n", "hits = 0\n", "for t in rag_tests:\n", " ans = rag_ask(t[\"q\"])\n", " ok = norm(t[\"expected\"]) in norm(ans); hits += ok\n", " print(\"✓\" if ok else \"✗\", t[\"q\"])\n", "print(f\"RAG score: {hits}/{len(rag_tests)}\")" ] }, { "cell_type": "code", "execution_count": null, "metadata": {}, "outputs": [], "source": [ "# Stretch 2 — does k matter?\n", "q = \"\"\n", "for k in [1, 3, 5]:\n", " print(f\"===== k={k} =====\")\n", " print(rag_ask(q, k=k)[:300], \"\\n\")\n", "# Small k: may miss context. Big k: noise + tokens. Where's YOUR sweet spot?" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Capstone foundation — saved?\n", "\n", "**File → Save.** This notebook returns in Sessions 5 and 6: it gains tools after lunch and gets attacked (then hardened) in the finale. Short break — then AI that *does things*." ] } ] }