Build a RAG engine for your notes, one stage at a time.

mkdir notes-rag && cd notes-rag
uv init backend                          # pyproject.toml, .python-version, .venv-ready scaffold
cd backend
mkdir -p app/{models,ingestion,retrieval,generation,api}
mkdir -p tests scripts
touch app/__init__.py app/models/__init__.py tests/__init__.py

uv add pydantic pydantic-settings numpy
uv add sentence-transformers          # local embeddings, no key
uv add openai anthropic google-generativeai   # only the provider(s) you plan to wire up
uv add --dev pytest

from functools import lru_cache
from typing import Literal
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    # which components to use (the pluggable switch)
    embedder: Literal["local", "openai", "gemini", "fake"] = "local"
    llm: Literal["openai", "anthropic", "gemini", "fake"] = "fake"   # fake => runs with ZERO keys

    # model identifiers
    local_embed_model: str = "sentence-transformers/all-MiniLM-L6-v2"   # 384-dim
    openai_embed_model: str = "text-embedding-3-small"                  # 1536-dim
    gemini_embed_model: str = "models/text-embedding-004"               # 768-dim
    openai_chat_model: str = "gpt-4.1-nano"
    anthropic_chat_model: str = "claude-haiku-4-5"
    gemini_chat_model: str = "gemini-2.5-flash-lite"
    reranker_model: str = "cross-encoder/ms-marco-MiniLM-L-6-v2"

    # secrets (only needed when the matching provider is selected)
    openai_api_key: str = ""
    anthropic_api_key: str = ""
    gemini_api_key: str = ""

    # ingestion / retrieval
    chunk_min_chars: int = 200
    chunk_max_chars: int = 1200
    index_dir: str = "./data/index"
    retrieval_top_k: int = 15    # FAISS returns this many
    rerank_top_k: int = 5        # this many survive reranking

    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8",
        case_sensitive=False, extra="ignore",
    )


@lru_cache(maxsize=1)
def get_settings() -> Settings:
    return Settings()

# .env.example  (copy to .env, fill only what you need)
EMBEDDER=local
LLM=fake
OPENAI_API_KEY=
ANTHROPIC_API_KEY=
GEMINI_API_KEY=

from typing import Protocol, runtime_checkable
import numpy as np


@runtime_checkable
class Embedder(Protocol):
    dim: int                                   # retrieval needs the vector size
    def embed(self, texts: list[str]) -> np.ndarray: ...   # (n, dim) float32


@runtime_checkable
class Chatter(Protocol):
    def answer(self, prompt: str) -> str: ...

import hashlib
import numpy as np
from app.config import Settings


class LocalEmbedder:                # sentence-transformers, free + offline
    def __init__(self, settings: Settings):
        from sentence_transformers import SentenceTransformer   # lazy import
        self._model = SentenceTransformer(settings.local_embed_model)
        self.dim = self._model.get_sentence_embedding_dimension()

    def embed(self, texts: list[str]) -> np.ndarray:
        v = self._model.encode(texts, convert_to_numpy=True, normalize_embeddings=True)
        return v.astype(np.float32)


class OpenAIEmbedder:               # hosted, needs OPENAI_API_KEY, 1536-dim
    def __init__(self, settings: Settings):
        from openai import OpenAI
        if not settings.openai_api_key:
            raise ValueError("EMBEDDER=openai but OPENAI_API_KEY is empty")
        self._client = OpenAI(api_key=settings.openai_api_key)
        self._model, self.dim = settings.openai_embed_model, 1536

    def embed(self, texts: list[str]) -> np.ndarray:
        r = self._client.embeddings.create(model=self._model, input=texts)
        return np.array([d.embedding for d in r.data], dtype=np.float32)


class GeminiEmbedder:               # hosted, needs GEMINI_API_KEY, 768-dim
    def __init__(self, settings: Settings):
        import google.generativeai as genai
        if not settings.gemini_api_key:
            raise ValueError("EMBEDDER=gemini but GEMINI_API_KEY is empty")
        genai.configure(api_key=settings.gemini_api_key)
        self._genai, self._model, self.dim = genai, settings.gemini_embed_model, 768

    def embed(self, texts: list[str]) -> np.ndarray:
        out = [self._genai.embed_content(model=self._model, content=t)["embedding"]
               for t in texts]
        return np.array(out, dtype=np.float32)


class FakeEmbedder:                 # deterministic, for tests. no model, no network
    def __init__(self, settings: Settings | None = None, dim: int = 384):
        self.dim = dim

    def embed(self, texts: list[str]) -> np.ndarray:
        out = np.zeros((len(texts), self.dim), dtype=np.float32)
        for i, t in enumerate(texts):
            seed = int(hashlib.sha256(t.encode()).hexdigest(), 16) % (2**32)
            v = np.random.default_rng(seed).standard_normal(self.dim).astype(np.float32)
            out[i] = v / (np.linalg.norm(v) + 1e-9)   # unit length, like the real ones
        return out

from app.config import Settings


class OpenAIChatter:
    def __init__(self, settings: Settings):
        from openai import OpenAI
        if not settings.openai_api_key:
            raise ValueError("LLM=openai but OPENAI_API_KEY is empty")
        self._client = OpenAI(api_key=settings.openai_api_key)
        self._model = settings.openai_chat_model

    def answer(self, prompt: str) -> str:
        r = self._client.chat.completions.create(
            model=self._model,
            messages=[{"role": "user", "content": prompt}])
        return r.choices[0].message.content


class AnthropicChatter:
    def __init__(self, settings: Settings):
        from anthropic import Anthropic
        if not settings.anthropic_api_key:
            raise ValueError("LLM=anthropic but ANTHROPIC_API_KEY is empty")
        self._client = Anthropic(api_key=settings.anthropic_api_key)
        self._model = settings.anthropic_chat_model

    def answer(self, prompt: str) -> str:
        r = self._client.messages.create(
            model=self._model, max_tokens=1024,
            messages=[{"role": "user", "content": prompt}])
        return r.content[0].text


class GeminiChatter:
    def __init__(self, settings: Settings):
        import google.generativeai as genai
        if not settings.gemini_api_key:
            raise ValueError("LLM=gemini but GEMINI_API_KEY is empty")
        genai.configure(api_key=settings.gemini_api_key)
        self._model = genai.GenerativeModel(settings.gemini_chat_model)

    def answer(self, prompt: str) -> str:
        return self._model.generate_content(prompt).text


class FakeChatter:                 # echoes prompt evidence; for tests + key-free dev
    def answer(self, prompt: str) -> str:
        head = prompt.strip().splitlines()[0][:80] if prompt.strip() else ""
        return f"[FAKE ANSWER] prompt started: {head!r}"

from functools import lru_cache
from app.config import Settings, get_settings
from app.models.base import Embedder, Chatter
from app.models import embedders, chatters


def build_embedder(s: Settings) -> Embedder:
    return {"local": embedders.LocalEmbedder,
            "openai": embedders.OpenAIEmbedder,
            "gemini": embedders.GeminiEmbedder,
            "fake": embedders.FakeEmbedder}[s.embedder](s)


def build_chatter(s: Settings) -> Chatter:
    return {"openai": chatters.OpenAIChatter,
            "anthropic": chatters.AnthropicChatter,
            "gemini": chatters.GeminiChatter,
            "fake": lambda _s: chatters.FakeChatter()}[s.llm](s)


@lru_cache(maxsize=1)
def get_embedder() -> Embedder: return build_embedder(get_settings())

@lru_cache(maxsize=1)
def get_chatter() -> Chatter: return build_chatter(get_settings())

uv add beautifulsoup4 lxml

from __future__ import annotations
from dataclasses import dataclass, field, asdict
import hashlib


@dataclass
class Chunk:
    text: str                    # shown to the user
    subject: str                 # "Java", "DSA", "React"...
    sheet: str                   # "Thread Synchronization"
    section: str = ""            # "Producer-Consumer pattern"
    tags: list[str] = field(default_factory=list)
    source: str = ""             # "java/thread-sync.html#producer-consumer"
    chunk_id: str = ""

    def __post_init__(self):
        if not self.chunk_id:
            key = f"{self.source}|{self.section}|{self.text[:64]}"
            self.chunk_id = hashlib.sha1(key.encode()).hexdigest()[:16]

    def embedding_text(self, with_crumb: bool = True) -> str:
        """What we EMBED: breadcrumb + tags + text, so the topic lives in the
        vector. Toggle with_crumb=False for the M1 experiment."""
        if not with_crumb:
            return self.text
        crumb = " > ".join(x for x in [self.subject, self.sheet, self.section] if x)
        tagline = f"tags: {', '.join(self.tags)}" if self.tags else ""
        return f"[{crumb}]\n{tagline}\n{self.text}".strip()

    def to_dict(self) -> dict:
        return asdict(self)

from __future__ import annotations
from dataclasses import dataclass
from bs4 import BeautifulSoup, Tag

@dataclass
class RawSection:
    section: str
    text: str
    tags: list[str]

_NOISE = ["script", "style", "nav", "footer", "header", "aside", "button"]

def _clean(node: Tag) -> str:
    for bad in node.find_all(_NOISE):
        bad.decompose()
    return " ".join(node.get_text(separator=" ", strip=True).split())

def parse_sheet(html: str, *, subject_hint="", sheet_hint=""):
    soup = BeautifulSoup(html, "lxml")
    title_el = soup.select_one("h1")                       # ADJUST
    sheet = (title_el.get_text(strip=True) if title_el else "") or sheet_hint
    subject = subject_hint

    sections: list[RawSection] = []
    nodes = soup.select("section")                         # ADJUST
    if nodes:
        for sec in nodes:
            head = sec.find(["h2", "h3", "h4"])            # ADJUST
            name = head.get_text(strip=True) if head else ""
            tags = [t.get_text(strip=True) for t in sec.select(".tag")]   # ADJUST
            body = _clean(sec)
            if body:
                sections.append(RawSection(name, body, tags))
    else:                                                  # heading-walk fallback
        cur = RawSection(sheet, "", [])
        for el in (soup.body or soup).descendants:
            if isinstance(el, Tag) and el.name in ("h2", "h3"):        # ADJUST
                if cur.text.strip():
                    sections.append(cur)
                cur = RawSection(el.get_text(strip=True), "", [])
            elif isinstance(el, Tag) and el.name in ("p", "li", "pre", "code"):
                cur.text += " " + el.get_text(" ", strip=True)
        if cur.text.strip():
            sections.append(cur)
    return subject, sheet, sections

from __future__ import annotations
import re
from app.ingestion.types import Chunk
from app.ingestion.html_parser import RawSection
from app.config import Settings

_SENT = re.compile(r"(?<=[.!?])\s+")

def _split_long(text, max_chars):
    if len(text) <= max_chars:
        return [text]
    out, buf = [], ""
    for s in _SENT.split(text):
        if len(buf) + len(s) + 1 > max_chars and buf:
            out.append(buf.strip()); buf = s
        else:
            buf = f"{buf} {s}".strip()
    if buf.strip():
        out.append(buf.strip())
    return out

def _slug(s):
    return re.sub(r"[^a-z0-9]+", "-", s.lower()).strip("-")

def chunk_sections(subject, sheet, sections: list[RawSection], source, settings: Settings):
    chunks, carry = [], ""
    for sec in sections:
        text = sec.text.strip()
        if not text:
            continue
        # carry a too-short section forward onto the next one
        if len(text) < settings.chunk_min_chars and len(text) + len(carry) < settings.chunk_max_chars:
            carry = f"{carry} {text}".strip()
            continue
        body = f"{carry} {text}".strip() if carry else text
        carry = ""
        for piece in _split_long(body, settings.chunk_max_chars):
            chunks.append(Chunk(text=piece, subject=subject, sheet=sheet,
                                section=sec.section, tags=sec.tags,
                                source=f"{source}#{_slug(sec.section)}"))
    if carry:
        chunks.append(Chunk(text=carry, subject=subject, sheet=sheet, source=source))
    return chunks

from __future__ import annotations
from pathlib import Path
from app.ingestion.html_parser import parse_sheet
from app.ingestion.chunker import chunk_sections
from app.config import Settings

def ingest_directory(notes_dir, settings: Settings):
    notes_dir = Path(notes_dir)
    all_chunks = []
    for f in sorted(notes_dir.rglob("*.html")):
        subject_hint = f.parent.name.replace("-", " ").title()   # ADJUST
        source = str(f.relative_to(notes_dir))
        html = f.read_text(encoding="utf-8", errors="ignore")
        subject, sheet, sections = parse_sheet(html, subject_hint=subject_hint, sheet_hint=f.stem)
        all_chunks.extend(chunk_sections(subject, sheet, sections, source, settings))
    return all_chunks

from __future__ import annotations
from typing import Protocol, runtime_checkable
import json, os
import numpy as np
from app.ingestion.types import Chunk


@runtime_checkable
class VectorStore(Protocol):
    dim: int
    embedder_name: str
    def add(self, chunks: list[Chunk], vectors: np.ndarray) -> None: ...
    def search(self, query_vec: np.ndarray, k: int) -> list[tuple[Chunk, float]]: ...
    def save(self, path: str) -> None: ...


class NumpyStore:
    """In-memory brute-force search. Great for tests and small note corpora
    (a few thousand chunks is instant). Same interface as FAISS later."""
    def __init__(self, dim: int, embedder_name: str = ""):
        self.dim, self.embedder_name = dim, embedder_name
        self._vecs = np.zeros((0, dim), dtype=np.float32)
        self._chunks: list[Chunk] = []

    def add(self, chunks, vectors):
        if vectors.shape[1] != self.dim:
            raise ValueError(f"vector dim {vectors.shape[1]} != store dim {self.dim}")
        self._vecs = np.vstack([self._vecs, vectors.astype(np.float32)])
        self._chunks.extend(chunks)

    def search(self, query_vec, k):
        if not self._chunks:
            return []
        sims = self._vecs @ query_vec.reshape(-1)          # cosine (vectors are unit-norm)
        idx = np.argsort(-sims)[:k]
        return [(self._chunks[i], float(sims[i])) for i in idx]

    def save(self, path):
        os.makedirs(path, exist_ok=True)
        np.save(os.path.join(path, "vecs.npy"), self._vecs)
        meta = {"dim": self.dim, "embedder_name": self.embedder_name,
                "chunks": [c.to_dict() for c in self._chunks]}
        with open(os.path.join(path, "meta.json"), "w") as f:
            json.dump(meta, f)

    @classmethod
    def load(cls, path, expected_embedder=None):
        with open(os.path.join(path, "meta.json")) as f:
            meta = json.load(f)
        if expected_embedder and meta["embedder_name"] != expected_embedder:
            raise ValueError(
                f"index built with '{meta['embedder_name']}' but current embedder is "
                f"'{expected_embedder}'. Re-index (vectors are not interchangeable).")
        s = cls(meta["dim"], meta["embedder_name"])
        s._vecs = np.load(os.path.join(path, "vecs.npy"))
        s._chunks = [Chunk(**c) for c in meta["chunks"]]
        return s

uv add faiss-cpu

from __future__ import annotations
import json, os
import numpy as np
import faiss
from app.ingestion.types import Chunk


class FaissStore:
    def __init__(self, dim: int, embedder_name: str = ""):
        self.dim, self.embedder_name = dim, embedder_name
        self.index = faiss.IndexFlatIP(dim)      # inner product == cosine on unit vectors
        self._chunks: list[Chunk] = []

    def add(self, chunks, vectors):
        if vectors.shape[1] != self.dim:
            raise ValueError(f"vector dim {vectors.shape[1]} != store dim {self.dim}")
        self.index.add(vectors.astype(np.float32))
        self._chunks.extend(chunks)

    def search(self, query_vec, k):
        scores, idx = self.index.search(query_vec.astype(np.float32).reshape(1, -1), k)
        out = []
        for s, i in zip(scores[0], idx[0]):
            if i != -1:
                out.append((self._chunks[i], float(s)))
        return out

    def save(self, path):
        os.makedirs(path, exist_ok=True)
        faiss.write_index(self.index, os.path.join(path, "index.faiss"))
        meta = {"dim": self.dim, "embedder_name": self.embedder_name,
                "chunks": [c.to_dict() for c in self._chunks]}
        with open(os.path.join(path, "meta.json"), "w") as f:
            json.dump(meta, f)

    @classmethod
    def load(cls, path, expected_embedder=None):
        with open(os.path.join(path, "meta.json")) as f:
            meta = json.load(f)
        if expected_embedder and meta["embedder_name"] != expected_embedder:
            raise ValueError(
                f"index built with '{meta['embedder_name']}' but current embedder is "
                f"'{expected_embedder}'. Re-index.")
        s = cls(meta["dim"], meta["embedder_name"])
        s.index = faiss.read_index(os.path.join(path, "index.faiss"))
        s._chunks = [Chunk(**c) for c in meta["chunks"]]
        return s

import sys
from app.config import get_settings
from app.models.factory import get_embedder
from app.ingestion.pipeline import ingest_directory
from app.retrieval.faiss_store import FaissStore

def build(notes_dir, with_crumb=True):
    s, emb = get_settings(), get_embedder()
    chunks = ingest_directory(notes_dir, s)
    vectors = emb.embed([c.embedding_text(with_crumb) for c in chunks])
    store = FaissStore(emb.dim, s.embedder)
    store.add(chunks, vectors)
    store.save(s.index_dir)
    print(f"indexed {len(chunks)} chunks -> {s.index_dir} (embedder={s.embedder}, crumb={with_crumb})")

if __name__ == "__main__":
    build(sys.argv[1] if len(sys.argv) > 1 else "./notes")

reranker: Literal["cross-encoder", "noop"] = "cross-encoder"   # add to Settings

from app.config import Settings
from app.models.base import Embedder
from app.retrieval.store import VectorStore

class Retriever:
    def __init__(self, embedder: Embedder, store: VectorStore, settings: Settings):
        self.embedder, self.store, self.k = embedder, store, settings.retrieval_top_k
    def retrieve(self, query: str):
        qv = self.embedder.embed([query])[0]
        return self.store.search(qv, self.k)        # [(Chunk, score), ...]

from functools import lru_cache
from app.config import Settings


@lru_cache(maxsize=1)
def _cross_encoder(model_name: str):
    from sentence_transformers import CrossEncoder     # lazy + cached (loads once)
    return CrossEncoder(model_name)


class CrossEncoderReranker:
    def __init__(self, settings: Settings):
        self.model_name, self.top_k = settings.reranker_model, settings.rerank_top_k

    def rerank(self, query, candidates):               # candidates: [(Chunk, score)]
        if not candidates:
            return []
        ce = _cross_encoder(self.model_name)
        scores = ce.predict([(query, c.text) for c, _ in candidates])
        ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
        return [(c, float(s)) for (c, _old), s in ranked[: self.top_k]]


class NoopReranker:                                   # for tests / A-B comparison
    def __init__(self, settings: Settings):
        self.top_k = settings.rerank_top_k
    def rerank(self, query, candidates):
        return candidates[: self.top_k]


def build_reranker(settings: Settings):
    return (CrossEncoderReranker if settings.reranker == "cross-encoder" else NoopReranker)(settings)

REWRITE_PROMPT = """Rewrite the user's question into a concise search query that \
will retrieve the most relevant study notes. Keep the key technical terms. \
Return ONLY the rewritten query, nothing else.

Question: {question}"""

RAG_PROMPT = """You are a study assistant. Answer the question using ONLY the notes \
below. Cite the source of each claim inline as [sheet > section]. If the notes do \
not contain the answer, say exactly: "The notes don't cover this." Do not use outside \
knowledge.

NOTES:
{context}

QUESTION: {question}

ANSWER:"""

from app.config import Settings
from app.models.base import Embedder, Chatter
from app.retrieval.store import VectorStore
from app.retrieval.retriever import Retriever
from app.retrieval.reranker import build_reranker
from app.generation.prompts import REWRITE_PROMPT, RAG_PROMPT


def _context(reranked):
    blocks = []
    for i, (c, _s) in enumerate(reranked, 1):
        crumb = " > ".join(x for x in [c.subject, c.sheet, c.section] if x)
        blocks.append(f"[{i}] ({crumb})\n{c.text}")
    return "\n\n".join(blocks)


class RagPipeline:
    def __init__(self, embedder: Embedder, store: VectorStore,
                 chatter: Chatter, settings: Settings):
        self.retriever = Retriever(embedder, store, settings)
        self.reranker = build_reranker(settings)
        self.chatter = chatter

    def answer(self, question: str, rewrite: bool = True) -> dict:
        query = self._rewrite(question) if rewrite else question
        reranked = self.reranker.rerank(query, self.retriever.retrieve(query))
        if not reranked:
            return {"answer": "The notes don't cover this.",
                    "sources": [], "contexts": [], "rewritten_query": query}
        answer = self.chatter.answer(
            RAG_PROMPT.format(context=_context(reranked), question=question))
        sources = [{"subject": c.subject, "sheet": c.sheet, "section": c.section,
                    "source": c.source, "score": round(s, 3)} for c, s in reranked]
        contexts = [c.text for c, _ in reranked]   # the exact note text the LLM saw
        return {"answer": answer, "sources": sources,
                "contexts": contexts, "rewritten_query": query}

    def _rewrite(self, question: str) -> str:
        try:
            return self.chatter.answer(REWRITE_PROMPT.format(question=question)).strip()
        except Exception:
            return question         # rewriting is a nice-to-have, never a hard failure

# .env  — pick ONE
LLM=openai
OPENAI_API_KEY=your_key_here
# or:
# LLM=anthropic
# ANTHROPIC_API_KEY=your_key_here
# or (has a free tier):
# LLM=gemini
# GEMINI_API_KEY=your_key_here

from app.ingestion.types import Chunk
from app.models.base import Chatter

TOPICS_PROMPT = """From these study notes, list the 5-8 main topics. \
Format each line exactly as: Title -- one-sentence description. No numbering.

NOTES:
{context}"""

QUIZ_PROMPT = """You are a study coach. From these notes, write exactly 4 questions a \
student should be able to answer to prove they understand the material. One per line, \
no numbering, no answers.

NOTES:
{context}"""


def _sample(chunks: list[Chunk], n: int = 12) -> str:
    """Spread the sample across the corpus so we don't only see the first sheet."""
    if len(chunks) <= n:
        picked = chunks
    else:
        step = len(chunks) // n
        picked = chunks[::step][:n]
    return "\n\n".join(c.text[:400] for c in picked)


def get_topics(chatter: Chatter, chunks: list[Chunk]) -> list[dict]:
    raw = chatter.answer(TOPICS_PROMPT.format(context=_sample(chunks)))
    out = []
    for line in raw.splitlines():
        line = line.strip().lstrip("-*0123456789. ").strip()
        if "--" in line:
            title, desc = line.split("--", 1)
            out.append({"title": title.strip(), "description": desc.strip()})
    return out[:8]


def get_suggestions(chatter: Chatter, chunks: list[Chunk]) -> list[str]:
    raw = chatter.answer(QUIZ_PROMPT.format(context=_sample(chunks)))
    qs = [ln.strip().lstrip("-*0123456789. ").strip()
          for ln in raw.splitlines() if ln.strip()]
    return qs[:4]

uv add fastapi uvicorn tenacity

"""Central retry decorators. Do NOT re-implement retry inside providers -
decorate the provider method with one of these instead."""
import httpx
from tenacity import (retry, stop_after_attempt, wait_exponential,
                      retry_if_exception_type)

_NET = (httpx.TimeoutException, httpx.ConnectError, ConnectionError, OSError)

llm_retry = retry(reraise=True, stop=stop_after_attempt(4),
                  wait=wait_exponential(min=2, max=30),
                  retry=retry_if_exception_type(_NET))

embed_retry = retry(reraise=True, stop=stop_after_attempt(3),
                    wait=wait_exponential(min=1, max=15),
                    retry=retry_if_exception_type(_NET))

from pydantic import BaseModel, Field

class QueryIn(BaseModel):
    question: str = Field(min_length=3, max_length=2000)
    rewrite: bool = True

class Source(BaseModel):
    subject: str; sheet: str; section: str; source: str; score: float

class QueryOut(BaseModel):
    answer: str
    sources: list[Source]
    rewritten_query: str

class Topic(BaseModel):
    title: str; description: str

from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware

from app.config import get_settings
from app.models.factory import get_embedder, get_chatter
from app.retrieval.faiss_store import FaissStore
from app.generation.pipeline import RagPipeline
from app.generation.study import get_topics, get_suggestions
from app.ingestion.pipeline import ingest_directory
from app.api.schemas import QueryIn, QueryOut, Topic

state = {}

@asynccontextmanager
async def lifespan(app: FastAPI):
    s = get_settings()
    store = FaissStore.load(s.index_dir, expected_embedder=s.embedder)
    state["rag"] = RagPipeline(get_embedder(), store, get_chatter(), s)
    state["chunks"] = store._chunks
    yield
    state.clear()

app = FastAPI(title="Notes-RAG", version="1.0", lifespan=lifespan)
app.add_middleware(CORSMiddleware, allow_origins=["http://localhost:3000"],
                   allow_methods=["*"], allow_headers=["*"])

@app.get("/health")
def health():
    return {"status": "ok", "chunks": len(state.get("chunks", []))}

@app.post("/query", response_model=QueryOut)
def query(body: QueryIn):
    try:
        return state["rag"].answer(body.question, rewrite=body.rewrite)
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/topics", response_model=list[Topic])
def topics(subject: str | None = None):
    chunks = state["chunks"]
    if subject:
        chunks = [c for c in chunks if c.subject.lower() == subject.lower()]
    return get_topics(state["rag"].chatter, chunks)

@app.get("/suggestions")
def suggestions(subject: str | None = None):
    chunks = state["chunks"]
    if subject:
        chunks = [c for c in chunks if c.subject.lower() == subject.lower()]
    return {"suggestions": get_suggestions(state["rag"].chatter, chunks)}

const BASE = process.env.NEXT_PUBLIC_API_URL ?? "http://localhost:8000";

export type Source = { subject: string; sheet: string; section: string; source: string; score: number };
export type QueryOut = { answer: string; sources: Source[]; rewritten_query: string };

export async function ask(question: string): Promise<QueryOut> {
  const r = await fetch(`${BASE}/query`, {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({ question }),
  });
  if (!r.ok) throw new Error(`query failed: ${r.status}`);
  return r.json();
}

export async function getTopics(subject?: string) {
  const q = subject ? `?subject=${encodeURIComponent(subject)}` : "";
  return (await fetch(`${BASE}/topics${q}`)).json();
}

export async function getSuggestions(subject?: string) {
  const q = subject ? `?subject=${encodeURIComponent(subject)}` : "";
  return (await (await fetch(`${BASE}/suggestions${q}`)).json()).suggestions as string[];
}

"use client";
import { useEffect, useState } from "react";
import { ask, getTopics, getSuggestions, QueryOut } from "@/lib/api";

export default function Home() {
  const [q, setQ] = useState("");
  const [res, setRes] = useState<QueryOut | null>(null);
  const [loading, setLoading] = useState(false);
  const [topics, setTopics] = useState<{ title: string; description: string }[]>([]);
  const [chips, setChips] = useState<string[]>([]);

  useEffect(() => {
    getTopics().then(setTopics).catch(() => {});
    getSuggestions().then(setChips).catch(() => {});
  }, []);

  async function run(question: string) {
    if (question.trim().length < 3) return;
    setLoading(true); setRes(null);
    try { setRes(await ask(question)); }
    catch (e) { setRes({ answer: "Request failed. Is the API running on :8000?", sources: [], rewritten_query: "" }); }
    finally { setLoading(false); }
  }

  return (
    <main style={{ display: "grid", gridTemplateColumns: "260px 1fr", gap: 32, maxWidth: 1100, margin: "40px auto", padding: "0 20px", fontFamily: "system-ui" }}>
      <aside>
        <h3>Topics</h3>
        <ul style={{ listStyle: "none", padding: 0 }}>
          {topics.map((t) => (
            <li key={t.title} style={{ marginBottom: 12 }}>
              <strong>{t.title}</strong>
              <div style={{ fontSize: 13, color: "#666" }}>{t.description}</div>
            </li>
          ))}
        </ul>
      </aside>

      <section>
        <h1>Ask my notes</h1>
        <div style={{ display: "flex", gap: 8 }}>
          <input value={q} onChange={(e) => setQ(e.target.value)}
                 onKeyDown={(e) => e.key === "Enter" && run(q)}
                 placeholder="e.g. difference between Comparable and Comparator"
                 style={{ flex: 1, padding: 12, fontSize: 15 }} />
          <button onClick={() => run(q)} disabled={loading} style={{ padding: "12px 20px" }}>
            {loading ? "..." : "Ask"}
          </button>
        </div>

        <div style={{ marginTop: 10, display: "flex", flexWrap: "wrap", gap: 8 }}>
          {chips.map((c) => (
            <button key={c} onClick={() => { setQ(c); run(c); }}
                    style={{ fontSize: 13, padding: "6px 10px", borderRadius: 16, cursor: "pointer" }}>
              {c}
            </button>
          ))}
        </div>

        {res && (
          <article style={{ marginTop: 28 }}>
            <p style={{ whiteSpace: "pre-wrap", lineHeight: 1.6 }}>{res.answer}</p>
            {res.sources.length > 0 && (
              <div style={{ marginTop: 20, fontSize: 13, color: "#555" }}>
                <strong>Sources</strong>
                <ul>
                  {res.sources.map((s, i) => (
                    <li key={i}>{s.subject} &rsaquo; {s.sheet} &rsaquo; {s.section} ({s.score})</li>
                  ))}
                </ul>
              </div>
            )}
          </article>
        )}
      </section>
    </main>
  );
}

# frontend/.env.local
NEXT_PUBLIC_API_URL=http://localhost:8000

[
  {"q": "difference between Comparable and Comparator", "expect_sheet": "Sorting"},
  {"q": "how does volatile differ from synchronized", "expect_sheet": "Thread Synchronization"},
  {"q": "what is Reciprocal Rank Fusion", "expect_sheet": "Hybrid Search"},
  {"q": "how does useEffect cleanup work", "expect_sheet": "React Foundations"},
  {"q": "what is the equals hashCode contract", "expect_sheet": "Object Class"}
]

import json
from app.config import get_settings
from app.models.factory import get_embedder
from app.retrieval.faiss_store import FaissStore
from app.retrieval.retriever import Retriever
from app.retrieval.reranker import build_reranker

s = get_settings()
store = FaissStore.load(s.index_dir, expected_embedder=s.embedder)
retr, rr = Retriever(get_embedder(), store, s), build_reranker(s)
data = json.load(open("eval/questions.json"))

hits_k, rr_sum = 0, 0.0
for row in data:
    ranked = rr.rerank(row["q"], retr.retrieve(row["q"]))
    sheets = [c.sheet for c, _ in ranked]
    want = row["expect_sheet"].lower()
    pos = next((i for i, sh in enumerate(sheets) if want in sh.lower()), None)
    if pos is not None:
        hits_k += 1
        rr_sum += 1.0 / (pos + 1)          # reciprocal rank
    print(f"{'OK ' if pos is not None else 'MISS'} rank={pos}  {row['q'][:45]}")

n = len(data)
print(f"\nrecall@{s.rerank_top_k}: {hits_k}/{n} = {hits_k/n:.0%}   MRR: {rr_sum/n:.3f}")

uv add ragas   # optional, generation-quality eval

# RAGAS scores the ANSWER, not just retrieval. It calls an LLM judge (set its
# key), so keep it off your fast path. The two metrics below need NO ground truth.
from datasets import Dataset
from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

def ragas_row(rag, question):
    out = rag.answer(question)
    # contexts MUST be the real chunk text the model saw (out["contexts"]), NOT a
    # "sheet > section" label — faithfulness scores each claim against these strings.
    return {"question": question, "answer": out["answer"],
            "contexts": out["contexts"]}

# ds = Dataset.from_list([ragas_row(rag, r["q"]) for r in data])
# print(evaluate(ds, metrics=[faithfulness, answer_relevancy]))

# ---- builder: resolve + install the locked env ----
FROM python:3.13-slim AS builder
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
WORKDIR /app
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-install-project --no-dev
COPY app ./app
COPY scripts ./scripts
RUN uv sync --frozen --no-dev

# ---- runtime: copy only the built venv + app ----
FROM python:3.13-slim
WORKDIR /app
COPY --from=builder /app /app
ENV PATH="/app/.venv/bin:$PATH"
EXPOSE 8000
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

FROM node:20-alpine AS build
WORKDIR /app
COPY package*.json ./
RUN npm ci
COPY . .
RUN npm run build

FROM node:20-alpine
WORKDIR /app
COPY --from=build /app ./
EXPOSE 3000
CMD ["npm", "start"]

services:
  backend:
    build: ./backend
    ports: ["8000:8000"]
    env_file: ./backend/.env
    volumes:
      - ./backend/data:/app/data      # persist the FAISS index
      - ./notes:/app/notes:ro         # notes available for /reindex
  frontend:
    build: ./frontend
    ports: ["3000:3000"]
    environment:
      - NEXT_PUBLIC_API_URL=http://localhost:8000
    depends_on: [backend]

name: Reindex notes

on:
  push:
    paths: ["**/*.html"]          # only when a sheet changes
  workflow_dispatch: {}           # and let me trigger it by hand

jobs:
  build-index:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Install uv
        uses: astral-sh/setup-uv@v3
        with:
          enable-cache: true

      - name: Install backend deps
        working-directory: backend
        run: uv sync --frozen

      - name: Build FAISS index from the notes
        working-directory: backend
        env:
          EMBEDDER: local           # no API key needed in CI
          LLM: fake
        run: uv run python scripts/build_index.py ../   # repo root holds the .html sheets

      - name: Upload the index
        uses: actions/upload-artifact@v4
        with:
          name: notes-index
          path: backend/data/index/

uv add pymupdf
# optional OCR fallback for scanned pages:
# uv add pytesseract   (also needs the system 'tesseract' binary)

from __future__ import annotations
from pathlib import Path
import fitz   # PyMuPDF
from app.ingestion.html_parser import RawSection

def parse_pdf(path, *, subject_hint=""):
    doc = fitz.open(path)
    sections = []
    for i, page in enumerate(doc, 1):
        text = page.get_text().strip()
        if not text:
            text = _ocr(page)                       # scanned page fallback
        if text:
            sections.append(RawSection(section=f"p.{i}", text=" ".join(text.split()), tags=[]))
    sheet = Path(path).stem
    return subject_hint, sheet, sections

def _ocr(page) -> str:
    try:
        import pytesseract
        from PIL import Image
        pix = page.get_pixmap(dpi=200)
        img = Image.frombytes("RGB", [pix.width, pix.height], pix.samples)
        return pytesseract.image_to_string(img)
    except Exception:
        return ""

uv add rank-bm25

from rank_bm25 import BM25Okapi
from app.ingestion.types import Chunk


class Bm25Index:
    def __init__(self, chunks: list[Chunk]):
        self.chunks = chunks
        self.bm25 = BM25Okapi([c.text.lower().split() for c in chunks])

    def search(self, query: str, k: int):
        scores = self.bm25.get_scores(query.lower().split())
        top = sorted(range(len(scores)), key=lambda i: -scores[i])[:k]
        return [(self.chunks[i], float(scores[i])) for i in top]


def rrf(rankings: list[list[tuple[Chunk, float]]], k: int, c: int = 60):
    """Reciprocal Rank Fusion: each list votes 1/(c + rank). Robust because it
    uses positions, not raw scores (which aren't comparable across methods)."""
    fused: dict[str, list] = {}
    for ranking in rankings:
        for rank, (chunk, _score) in enumerate(ranking):
            entry = fused.setdefault(chunk.chunk_id, [chunk, 0.0])
            entry[1] += 1.0 / (c + rank)
    merged = sorted(fused.values(), key=lambda x: -x[1])
    return [(chunk, score) for chunk, score in merged[:k]]