AI Learning Assistant
An Incremental Build Walkthrough

A tiny helper the LLM/embedding calls can lean on: retry transient failures (429 / 5xx / dropped connections) with exponential backoff instead of crashing. Built on tenacity.

uv add tenacity httpx

"""
Centralised retry / backoff decorators built on `tenacity`.

Usage
-----
    from app.utils.retry import llm_retry, embed_retry

    @llm_retry
    def call_llm(...): ...
"""

import logging

from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    before_sleep_log,
    after_log,
)

logger = logging.getLogger(__name__)

# Exceptions that are safe to retry (network hiccups, rate-limits, timeouts)
_RETRYABLE = (
    ConnectionError,
    TimeoutError,
    OSError,
)

try:
    import httpx

    _RETRYABLE = (*_RETRYABLE, httpx.HTTPStatusError, httpx.ConnectError, httpx.ReadTimeout)
except ImportError:
    pass


def _is_retryable(exc: BaseException) -> bool:
    """Return True for 429 / 5xx HTTP errors and standard I/O errors."""
    if isinstance(exc, _RETRYABLE):
        return True
    # Inspect an HTTP status code if the exception carries a response
    status = getattr(getattr(exc, "response", None), "status_code", None)
    if status and (status == 429 or status >= 500):
        return True
    return False


# ── LLM calls (Claude / OpenAI / OpenRouter) ─────────────────────────────────
llm_retry = retry(
    retry=lambda retry_state: _is_retryable(retry_state.outcome.exception())
    if retry_state.outcome.failed
    else False,
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(6),
    before_sleep=before_sleep_log(logger, logging.WARNING),
    after=after_log(logger, logging.DEBUG),
    reraise=True,
)

# ── Embedding calls ──────────────────────────────────────────────────────────
embed_retry = retry(
    retry=lambda retry_state: _is_retryable(retry_state.outcome.exception())
    if retry_state.outcome.failed
    else False,
    wait=wait_exponential(multiplier=1, min=1, max=30),
    stop=stop_after_attempt(5),
    before_sleep=before_sleep_log(logger, logging.WARNING),
    reraise=True,
)

The heart of a Claude-first build. One function, get_llm(), returns a LangChain chat model for whichever provider .env selects. Every route and service imports this and never names a provider directly — that is why swapping providers later is a one-line change.

uv add langchain langchain-core langchain-anthropic langchain-openai
# (langchain-anthropic pulls in the official `anthropic` SDK)

"""
The LLM provider seam.

One function, get_llm(), returns a LangChain chat model for whichever provider
is configured. Everything downstream — query rewrite, answer generation, topic
extraction, question suggestions, and the TTS text rewrite — calls this and
never names a provider itself. Swap providers by editing .env, not code.

  LLM_PROVIDER=anthropic   -> ChatAnthropic (Claude Opus 4.8)   <- primary
  LLM_PROVIDER=openai      -> ChatOpenAI    (GPT-4o)
  LLM_PROVIDER=openrouter  -> ChatOpenAI against the OpenRouter gateway (demo fallback)
"""

from functools import lru_cache

from langchain_core.language_models import BaseChatModel

from app.config import get_settings


@lru_cache(maxsize=1)
def get_llm() -> BaseChatModel:
    """Build the chat model for the configured provider (once per process)."""
    settings = get_settings()
    provider = settings.llm_provider.lower()

    if provider == "anthropic":
        # Uses the official Anthropic SDK under the hood via langchain-anthropic.
        # IMPORTANT: do not pass temperature/top_p/top_k — Claude Opus 4.8
        # rejects sampling parameters with a 400. Steer with prompting instead.
        from langchain_anthropic import ChatAnthropic

        return ChatAnthropic(
            model=settings.anthropic_model,
            api_key=settings.anthropic_api_key,
            max_tokens=2048,
            timeout=60,
        )

    if provider == "openai":
        from langchain_openai import ChatOpenAI

        return ChatOpenAI(
            model=settings.openai_model,
            api_key=settings.openai_api_key,
            temperature=0.2,
            max_tokens=2048,
        )

    if provider == "openrouter":
        # OpenRouter exposes an OpenAI-compatible endpoint — it is a gateway/
        # aggregator, so ChatOpenAI + base_url is the correct tool here (this is
        # NOT "Claude through a shim"; Claude goes through langchain-anthropic).
        from langchain_openai import ChatOpenAI

        return ChatOpenAI(
            model=settings.openrouter_model,
            api_key=settings.openrouter_api_key,
            base_url=settings.openrouter_base_url,
            temperature=0.2,
            max_tokens=2048,
            default_headers={
                "HTTP-Referer": "https://my-learning-ai-assistant.local",
                "X-Title": "My Learning AI Assistant",
            },
        )

    raise ValueError(f"Unknown LLM_PROVIDER: {settings.llm_provider!r}")

Put your key in .env and prove the seam works before building anything on it:

ANTHROPIC_API_KEY=sk-ant-...     # add this line to backend/.env

uv run python -c "from app.services.llm import get_llm; print(get_llm().invoke('Say hi in 3 words').content)"
# -> a short Claude reply, e.g. "Hey there, friend!"

A thin LangChain-compatible wrapper around a local sentence-transformers model (with a small retry). Runs on-device, so there is nothing to swap by provider — and no API cost for the high-volume embedding of your whole PDF.

uv add langchain-huggingface sentence-transformers

"""
Local sentence-transformers embeddings via langchain-huggingface.
Inherits from LangChain's Embeddings base so FAISS accepts it directly.
"""

import logging
from functools import lru_cache

from langchain_core.embeddings import Embeddings
from langchain_huggingface import HuggingFaceEmbeddings
from tenacity import retry, stop_after_attempt, wait_exponential, before_sleep_log

from app.config import get_settings

logger = logging.getLogger(__name__)


class ResilientEmbeddings(Embeddings):
    """
    LangChain-compatible embeddings wrapper around sentence-transformers.
    Runs fully locally — no API key required.
    """

    def __init__(self) -> None:
        settings = get_settings()
        logger.info("Loading embedding model: %s", settings.embedding_model)
        self._model = HuggingFaceEmbeddings(
            model_name=settings.embedding_model,
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )
        logger.info("Embedding model ready.")

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return self._embed_docs(texts)

    def embed_query(self, text: str) -> list[float]:
        return self._embed_q(text)

    @retry(
        retry=lambda rs: isinstance(rs.outcome.exception(), (ConnectionError, OSError, TimeoutError))
            if rs.outcome.failed else False,
        wait=wait_exponential(min=1, max=30),
        stop=stop_after_attempt(3),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )
    def _embed_docs(self, texts: list[str]) -> list[list[float]]:
        return self._model.embed_documents(texts)

    @retry(
        retry=lambda rs: isinstance(rs.outcome.exception(), (ConnectionError, OSError, TimeoutError))
            if rs.outcome.failed else False,
        wait=wait_exponential(min=1, max=30),
        stop=stop_after_attempt(3),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        reraise=True,
    )
    def _embed_q(self, text: str) -> list[float]:
        return self._model.embed_query(text)


@lru_cache(maxsize=1)
def get_embeddings() -> ResilientEmbeddings:
    return ResilientEmbeddings()

Test it (first run downloads the model, a few seconds):

uv run python -c "from app.services.embeddings import get_embeddings; v=get_embeddings().embed_query('hello world'); print('dims:', len(v), '| first 3:', v[:3])"
# -> dims: 384 | first 3: [...]

Create the schemas file, but only add what this slice needs: the upload response. We append query and eval schemas later, when those features arrive. Growing models.py per feature keeps each step self-contained.

"""
Pydantic request / response models shared across routes.

We GROW this file feature by feature. Right now it only needs the upload
response; query and evaluation models get appended when we build those routes.
"""

from pydantic import BaseModel, Field  # noqa: F401  (Field is used by later models)


# ── Upload ────────────────────────────────────────────────────────────────────

class UploadResponse(BaseModel):
    message: str
    document_id: str
    chunk_count: int
    description: str

The ingestion engine, written and tested before the upload route calls it. Every page is tried through four extractors in order; the first that yields text for a page wins, so a hybrid PDF (some digital, some scanned) works. Then chunk, embed, and persist to FAISS.

uv add faiss-cpu langchain-community langchain-text-splitters \
       pypdf pdfplumber pymupdf pytesseract pdf2image pillow
# OCR (stage 4) also needs system binaries:  brew install tesseract poppler

"""
PDF ingestion, intelligent chunking, and FAISS index management.

Extraction pipeline (each stage feeds the next as fallback)
------------------------------------------------------------
Stage 1 — pdfplumber  : best for complex layouts and tables
Stage 2 — pypdf       : fast, handles most standard digital PDFs
Stage 3 — pymupdf     : robust fallback for compressed / legacy PDFs
Stage 4 — OCR         : pytesseract + pdf2image for fully scanned PDFs

Every page is attempted through all four stages; the first stage that
yields non-empty text for a given page wins. This means a hybrid PDF
(some text pages, some scanned pages) is handled correctly.
"""

import io
import logging
import uuid
from pathlib import Path

from langchain_core.documents import Document
from langchain_community.vectorstores import FAISS
from langchain_text_splitters import RecursiveCharacterTextSplitter

from app.config import get_settings
from app.services.embeddings import get_embeddings

logger = logging.getLogger(__name__)


# ── Stage 1: pdfplumber ───────────────────────────────────────────────────────

def _extract_with_pdfplumber(pdf_bytes: bytes) -> dict[int, str]:
    import pdfplumber
    result: dict[int, str] = {}
    with pdfplumber.open(io.BytesIO(pdf_bytes)) as pdf:
        for page_num, page in enumerate(pdf.pages, start=1):
            text = (page.extract_text() or "").strip()
            if text:
                result[page_num] = text
    return result


# ── Stage 2: pypdf ────────────────────────────────────────────────────────────

def _extract_with_pypdf(pdf_bytes: bytes) -> dict[int, str]:
    from pypdf import PdfReader
    result: dict[int, str] = {}
    reader = PdfReader(io.BytesIO(pdf_bytes))
    for page_num, page in enumerate(reader.pages, start=1):
        try:
            text = page.extract_text(extraction_mode="layout") or ""
        except TypeError:
            text = page.extract_text() or ""
        text = text.strip()
        if text:
            result[page_num] = text
    return result


# ── Stage 3: pymupdf ──────────────────────────────────────────────────────────

def _extract_with_pymupdf(pdf_bytes: bytes) -> dict[int, str]:
    import fitz  # pymupdf
    result: dict[int, str] = {}
    with fitz.open(stream=pdf_bytes, filetype="pdf") as doc:
        for page_num, page in enumerate(doc, start=1):
            text = page.get_text("text").strip()
            if text:
                result[page_num] = text
    return result


# ── Stage 4: OCR via pytesseract + pdf2image ──────────────────────────────────

def _extract_with_ocr(pdf_bytes: bytes, missing_pages: set[int]) -> dict[int, str]:
    """
    Convert only the pages that all text-based engines failed on to images,
    then run Tesseract OCR on them. Requires:
      - tesseract  (brew install tesseract)
      - poppler    (brew install poppler)
    """
    import pytesseract
    from pdf2image import convert_from_bytes

    result: dict[int, str] = {}

    # Convert only the missing pages (1-indexed) to avoid processing the whole doc
    for page_num in sorted(missing_pages):
        try:
            images = convert_from_bytes(
                pdf_bytes,
                dpi=300,
                first_page=page_num,
                last_page=page_num,
            )
            if not images:
                continue
            text = pytesseract.image_to_string(images[0], lang="eng").strip()
            if text:
                result[page_num] = text
                logger.info("OCR extracted text from page %d", page_num)
            else:
                logger.warning("OCR returned no text for page %d", page_num)
        except Exception as exc:
            logger.warning("OCR failed on page %d: %s", page_num, exc)

    return result


# ── Multi-engine coordinator ──────────────────────────────────────────────────

def _get_total_pages(pdf_bytes: bytes) -> int:
    """Quick page count without full parsing."""
    try:
        import fitz
        with fitz.open(stream=pdf_bytes, filetype="pdf") as doc:
            return len(doc)
    except Exception:
        from pypdf import PdfReader
        return len(PdfReader(io.BytesIO(pdf_bytes)).pages)


def _extract_text_from_pdf(pdf_bytes: bytes) -> list[tuple[str, int]]:
    """
    Run all extraction stages and return a merged (text, page_num) list,
    sorted by page number.
    """
    total_pages = _get_total_pages(pdf_bytes)
    all_page_nums = set(range(1, total_pages + 1))
    page_map: dict[int, str] = {}

    text_stages = [
        ("pdfplumber", _extract_with_pdfplumber),
        ("pypdf",      _extract_with_pypdf),
        ("pymupdf",    _extract_with_pymupdf),
    ]

    for name, extractor in text_stages:
        missing = all_page_nums - page_map.keys()
        if not missing:
            break
        try:
            found = extractor(pdf_bytes)
            for page_num, text in found.items():
                if page_num not in page_map:
                    page_map[page_num] = text
            logger.debug("%s: captured %d page(s)", name, len(found))
        except Exception as exc:
            logger.warning("Extractor '%s' failed: %s", name, exc)

    # OCR pass for any pages still missing text
    missing_after_text = all_page_nums - page_map.keys()
    if missing_after_text:
        logger.info(
            "Falling back to OCR for %d page(s): %s",
            len(missing_after_text),
            sorted(missing_after_text),
        )
        try:
            ocr_results = _extract_with_ocr(pdf_bytes, missing_after_text)
            page_map.update(ocr_results)
        except Exception as exc:
            logger.warning("OCR stage failed: %s", exc)

    return [(text, num) for num, text in sorted(page_map.items())]


# ── Document builder ──────────────────────────────────────────────────────────

def _build_documents(
    pages: list[tuple[str, int]],
    description: str,
    document_id: str,
    filename: str,
) -> list[Document]:
    return [
        Document(
            page_content=text,
            metadata={
                "page": page_num,
                "document_id": document_id,
                "filename": filename,
                "description": description,
            },
        )
        for text, page_num in pages
    ]


# ── Public API ────────────────────────────────────────────────────────────────

def ingest_pdf(
    pdf_bytes: bytes,
    filename: str,
    description: str,
) -> tuple[str, int]:
    """
    Full ingestion pipeline: extract → chunk → embed → persist FAISS.
    Returns (document_id, total_chunks).
    """
    settings = get_settings()
    embeddings = get_embeddings()

    document_id = str(uuid.uuid4())
    logger.info("Starting ingestion for '%s' (id=%s)", filename, document_id)

    pages = _extract_text_from_pdf(pdf_bytes)

    if not pages:
        raise ValueError(
            "Could not extract text from any page of this PDF, even with OCR. "
            "The document may be corrupt, password-protected, or contain only graphics."
        )

    logger.info("Extracted text from %d/%d page(s) of '%s'",
                len(pages), _get_total_pages(pdf_bytes), filename)

    raw_docs = _build_documents(pages, description, document_id, filename)

    splitter = RecursiveCharacterTextSplitter(
        chunk_size=settings.chunk_size,
        chunk_overlap=settings.chunk_overlap,
        separators=["\n\n", "\n", ". ", " ", ""],
        length_function=len,
    )
    chunks = splitter.split_documents(raw_docs)

    for chunk in chunks:
        chunk.metadata["document_id"] = document_id

    logger.info("Split into %d chunks for document_id=%s", len(chunks), document_id)

    index_path = Path(settings.faiss_index_path)
    index_path.mkdir(parents=True, exist_ok=True)

    # Always rebuild the index from scratch so stale chunks from previous
    # uploads never accumulate.
    logger.info("Building fresh FAISS index.")
    vector_store = FAISS.from_documents(chunks, embeddings)

    vector_store.save_local(str(index_path))
    logger.info("FAISS index persisted to %s", index_path)

    return document_id, len(chunks)


def load_vector_store() -> FAISS:
    """Load the persisted FAISS index from disk."""
    settings = get_settings()
    index_path = Path(settings.faiss_index_path)
    if not (index_path / "index.faiss").exists():
        raise FileNotFoundError(
            "No FAISS index found. Please upload a document first."
        )
    return FAISS.load_local(
        str(index_path), get_embeddings(), allow_dangerous_deserialization=True,
    )

Drop any PDF next to your code as sample.pdf and ingest it:

uv run python -c "from app.services.document_processor import ingest_pdf; import pathlib; b=pathlib.Path('sample.pdf').read_bytes(); print(ingest_pdf(b, 'sample.pdf', 'a test document'))"
# -> ('a1b2c3-...uuid...', 42)     # (document_id, chunk_count)
# and a faiss_index/ folder now exists on disk

Only now, with ingest_pdf tested, do we add the route that calls it. Watch how little it does: validate, read bytes, delegate. Then add one line to main.py.

uv add python-multipart   # lets FastAPI accept file uploads

"""
POST /api/upload

Accepts a PDF file + a text description, triggers document ingestion,
and returns the assigned document_id and chunk count.
"""

import logging

from fastapi import APIRouter, File, Form, HTTPException, UploadFile, status

from app.models import UploadResponse
from app.services.document_processor import ingest_pdf

logger = logging.getLogger(__name__)
router = APIRouter()

_MAX_PDF_SIZE_MB = 50
_MAX_PDF_BYTES = _MAX_PDF_SIZE_MB * 1024 * 1024


@router.post(
    "/upload",
    response_model=UploadResponse,
    status_code=status.HTTP_201_CREATED,
    summary="Upload a PDF document for RAG ingestion",
)
async def upload_document(
    file: UploadFile = File(..., description="PDF document to ingest"),
    description: str = Form(..., min_length=5, max_length=1000),
) -> UploadResponse:
    """
    Upload a PDF alongside a descriptive context string.

    The backend will:
    1. Extract and chunk the PDF text.
    2. Generate embeddings via Google embedding-001.
    3. Persist chunks to the local FAISS index.

    Returns a `document_id` you can pass to `/api/query` to restrict
    retrieval to this document.
    """
    # Validate content type
    if file.content_type not in ("application/pdf", "application/octet-stream"):
        raise HTTPException(
            status_code=status.HTTP_415_UNSUPPORTED_MEDIA_TYPE,
            detail="Only PDF files are accepted.",
        )

    pdf_bytes = await file.read()

    if len(pdf_bytes) > _MAX_PDF_BYTES:
        raise HTTPException(
            status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE,
            detail=f"PDF exceeds the {_MAX_PDF_SIZE_MB} MB size limit.",
        )

    if len(pdf_bytes) == 0:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Uploaded file is empty.",
        )

    try:
        document_id, chunk_count = ingest_pdf(
            pdf_bytes=pdf_bytes,
            filename=file.filename or "document.pdf",
            description=description,
        )
    except ValueError as exc:
        raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
    except Exception as exc:
        logger.exception("Ingestion failed for file '%s'", file.filename)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Ingestion error: {exc}",
        )

    return UploadResponse(
        message="Document ingested successfully.",
        document_id=document_id,
        chunk_count=chunk_count,
        description=description,
    )

Grow main.py — add the import and one router line to the (currently empty) Routers section:

# at the top, with the other imports:
from app.routes import upload

# in the "Routers" section:
app.include_router(upload.router, prefix="/api", tags=["Documents"])

Restart uvicorn, open /docs, and try POST /api/upload with a PDF + a description.

After FAISS returns 15 candidates, the Cross-Encoder scores every (query, chunk) pair together and keeps the top 5. A leaf service the pipeline will call — write it first.

"""
Cross-Encoder re-ranker using sentence-transformers.

After FAISS retrieves `retrieval_top_k` candidates, this module scores
each (query, chunk) pair and returns the top `reranker_top_k` by relevance.
"""

import logging
from functools import lru_cache

from langchain_core.documents import Document
from sentence_transformers import CrossEncoder

from app.config import get_settings

logger = logging.getLogger(__name__)


@lru_cache(maxsize=1)
def _get_cross_encoder() -> CrossEncoder:
    settings = get_settings()
    logger.info("Loading CrossEncoder model: %s", settings.reranker_model)
    return CrossEncoder(settings.reranker_model)


def rerank(query: str, documents: list[Document]) -> list[Document]:
    """
    Score every (query, document) pair and return the top-k documents
    sorted by descending cross-encoder score.

    Parameters
    ----------
    query     : The (possibly rewritten) user question.
    documents : Candidate chunks from initial FAISS retrieval.

    Returns
    -------
    Reranked list of Documents (length ≤ reranker_top_k), each annotated
    with a `rerank_score` metadata field.
    """
    settings = get_settings()

    if not documents:
        return []

    encoder = _get_cross_encoder()

    # Build (query, passage) pairs for batch scoring
    pairs = [(query, doc.page_content) for doc in documents]
    scores: list[float] = encoder.predict(pairs).tolist()

    # Annotate and sort
    for doc, score in zip(documents, scores):
        doc.metadata["rerank_score"] = round(float(score), 4)

    ranked = sorted(documents, key=lambda d: d.metadata["rerank_score"], reverse=True)
    top_k = ranked[: settings.reranker_top_k]

    logger.debug(
        "Re-ranked %d → %d documents. Top score=%.4f",
        len(documents),
        len(top_k),
        top_k[0].metadata["rerank_score"] if top_k else 0.0,
    )
    return top_k

Stitch it together: rewrite the query → FAISS retrieve top-k → cross-encoder rerank → generate with the LLM. All LLM calls go through the seam. Because the doc you indexed in Step 10 is still on disk, you can run the whole pipeline before any route exists.

"""
Advanced RAG pipeline.

Stages
------
1. Query Optimisation  — LLM rewrites the raw user question for better retrieval.
2. FAISS Retrieval     — Fetch `retrieval_top_k` candidate chunks.
3. Cross-Encoder Re-rank — Score & filter down to `reranker_top_k` chunks.
4. Generation          — Gemini 2.5 Flash (via OpenRouter) synthesises the answer.

All LLM calls are wrapped in tenacity exponential-backoff retry.
"""

import logging

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.documents import Document
from app.services.llm import get_llm
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    before_sleep_log,
)

from app.config import get_settings
from app.services.document_processor import load_vector_store
from app.services.reranker import rerank

logger = logging.getLogger(__name__)


# ── LLM ────────────────────────────────────────────────────────────
# get_llm() lives in app/services/llm.py, so every route and service
# shares one provider seam (Claude by default; OpenAI/OpenRouter switchable).


# ── Prompts ───────────────────────────────────────────────────────────────────

_REWRITE_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            (
                "You are an expert query optimiser for a document retrieval system. "
                "Your task is to rewrite the user's question to be more precise and "
                "retrieval-friendly, using technical language where appropriate. "
                "Return ONLY the rewritten query — no explanation, no preamble."
            ),
        ),
        ("human", "Original question: {question}"),
    ]
)

_RAG_PROMPT = ChatPromptTemplate.from_messages(
    [
        (
            "system",
            (
                "You are a helpful and knowledgeable AI learning assistant. "
                "Answer the user's question using ONLY the provided context. "
                "If the context does not contain enough information, say so honestly. "
                "Cite relevant page numbers when possible. "
                "Be thorough, structured, and educational in your response."
            ),
        ),
        (
            "human",
            (
                "Context:\n{context}\n\n"
                "Question: {question}\n\n"
                "Answer:"
            ),
        ),
    ]
)


# ── Retry-wrapped LLM invoke ──────────────────────────────────────────────────

def _is_retryable(exc: BaseException) -> bool:
    status = getattr(getattr(exc, "response", None), "status_code", None)
    if status and (status == 429 or status >= 500):
        return True
    return isinstance(exc, (ConnectionError, TimeoutError, OSError))


@retry(
    retry=lambda rs: _is_retryable(rs.outcome.exception()) if rs.outcome.failed else False,
    wait=wait_exponential(multiplier=1, min=2, max=60),
    stop=stop_after_attempt(6),
    before_sleep=before_sleep_log(logger, logging.WARNING),
    reraise=True,
)
def _invoke_llm(llm, prompt: ChatPromptTemplate, **kwargs) -> str:
    chain = prompt | llm
    response = chain.invoke(kwargs)
    return response.content


# ── Pipeline stages ───────────────────────────────────────────────────────────

def _optimise_query(question: str) -> str:
    """Stage 1: Rewrite the question for better vector-space retrieval."""
    llm = get_llm()
    try:
        rewritten = _invoke_llm(llm, _REWRITE_PROMPT, question=question)
        logger.info("Query rewrite: '%s' → '%s'", question, rewritten)
        return rewritten.strip()
    except Exception as exc:
        logger.warning("Query rewrite failed (%s), using original question.", exc)
        return question


def _retrieve(query: str, document_id: str | None) -> list[Document]:
    """Stage 2: FAISS similarity search, optionally filtered by document_id."""
    settings = get_settings()
    vector_store = load_vector_store()

    if document_id:
        # LangChain's FAISS filter= parameter is unreliable (returns far fewer
        # results than the matching chunks). Fetch the full index and filter manually.
        total = len(vector_store.docstore._dict)
        all_results = vector_store.similarity_search(query, k=total)
        docs = [
            d for d in all_results
            if d.metadata.get("document_id") == document_id
        ][:settings.retrieval_top_k]
    else:
        docs = vector_store.similarity_search(query, k=settings.retrieval_top_k)

    logger.debug("Retrieved %d raw candidates.", len(docs))
    return docs


def _build_context(docs: list[Document]) -> str:
    """Format reranked docs into a structured context block for the LLM."""
    parts = []
    for i, doc in enumerate(docs, start=1):
        page = doc.metadata.get("page", "?")
        score = doc.metadata.get("rerank_score", "n/a")
        parts.append(
            f"[Chunk {i} | Page {page} | Relevance {score}]\n{doc.page_content}"
        )
    return "\n\n---\n\n".join(parts)


# ── Public entry point ────────────────────────────────────────────────────────

def run_rag(
    question: str,
    document_id: str | None = None,
) -> dict:
    """
    Execute the full RAG pipeline and return a result dict containing:
    - answer         : LLM-generated answer
    - rewritten_query: query after optimisation
    - sources        : list of SourceChunk-compatible dicts
    """
    # Stage 1 — query optimisation
    rewritten_query = _optimise_query(question)

    # Stage 2 — retrieval
    candidates = _retrieve(rewritten_query, document_id)
    if not candidates:
        return {
            "answer": "I could not find relevant information in the uploaded document.",
            "rewritten_query": rewritten_query,
            "sources": [],
        }

    # Stage 3 — re-ranking
    top_docs = rerank(rewritten_query, candidates)

    # Stage 4 — generation
    context = _build_context(top_docs)
    llm = get_llm()
    answer = _invoke_llm(llm, _RAG_PROMPT, context=context, question=rewritten_query)

    sources = [
        {
            "content": doc.page_content[:400],  # truncate for payload size
            "page": doc.metadata.get("page"),
            "score": doc.metadata.get("rerank_score"),
        }
        for doc in top_docs
    ]

    return {
        "answer": answer.strip(),
        "rewritten_query": rewritten_query,
        "sources": sources,
    }

Ask your indexed document a question, no server needed:

uv run python -c "from app.services.rag_pipeline import run_rag; r=run_rag('What is this document about?'); print('ANSWER:', r['answer'][:300]); print('REWRITTEN:', r['rewritten_query']); print('SOURCES:', len(r['sources']))"

Append the query schemas to the models file you started in Step 9. Same file, three more classes — added exactly when the feature needs them.

# ── Query ─────────────────────────────────────────────────────────────────────
# Append these to models.py when you build the query route.

class QueryRequest(BaseModel):
    question: str = Field(..., min_length=3, max_length=2000)
    document_id: str | None = Field(
        default=None,
        description="If provided, restrict retrieval to this document's chunks.",
    )
    stream: bool = False


class SourceChunk(BaseModel):
    content: str
    page: int | None = None
    score: float | None = None


class QueryResponse(BaseModel):
    answer: str
    sources: list[SourceChunk]
    rewritten_query: str
    document_id: str | None = None

With run_rag proven and the schemas in place, the route is trivial: validate, delegate, shape the response. Then one more line in main.py.

"""
POST /api/query

Runs the full RAG pipeline (query rewrite → retrieval → rerank → generation)
and returns the answer with source citations.
"""

import logging

from fastapi import APIRouter, HTTPException, status

from app.models import QueryRequest, QueryResponse, SourceChunk
from app.services.rag_pipeline import run_rag

logger = logging.getLogger(__name__)
router = APIRouter()


@router.post(
    "/query",
    response_model=QueryResponse,
    summary="Ask a question against ingested documents",
)
async def query_documents(payload: QueryRequest) -> QueryResponse:
    """
    Ask a natural-language question.

    The pipeline will:
    1. Rewrite/expand your question for optimal retrieval.
    2. Fetch the top-k semantically relevant chunks from FAISS.
    3. Re-rank them with a Cross-Encoder.
    4. Feed the top results + question to Gemini 2.5 Flash via OpenRouter.

    Optionally supply `document_id` to restrict answers to a single document.
    """
    try:
        result = run_rag(
            question=payload.question,
            document_id=payload.document_id,
        )
    except FileNotFoundError as exc:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc))
    except Exception as exc:
        logger.exception("RAG pipeline error for question: %s", payload.question)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Pipeline error: {exc}",
        )

    sources = [SourceChunk(**s) for s in result["sources"]]

    return QueryResponse(
        answer=result["answer"],
        sources=sources,
        rewritten_query=result["rewritten_query"],
        document_id=payload.document_id,
    )

from app.routes import query                                  # top imports
app.include_router(query.router, prefix="/api", tags=["RAG"])   # Routers section

Both services it needs already exist (load_vector_store, get_llm). It samples chunks evenly across the whole document and asks the LLM for 5–8 topics + one-liners.

"""
POST /api/topics

Scans the document's chunks and asks the LLM to identify the main topics/sections.
Returns a list of { title, description } objects for the sidebar.
"""

import logging

from fastapi import APIRouter, HTTPException
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel

from app.services.document_processor import load_vector_store
from app.services.llm import get_llm

logger = logging.getLogger(__name__)
router = APIRouter()

_TOPICS_PROMPT = ChatPromptTemplate.from_messages([
    (
        "system",
        (
            "You are a document analyst. Based on the document excerpts provided, "
            "identify 5-8 main topics or sections in the document. "
            "For each topic return exactly this format on its own line:\n"
            "TOPIC: <short title> | <one sentence description>\n"
            "Return nothing else."
        ),
    ),
    ("human", "Document excerpts:\n\n{context}"),
])


class TopicsRequest(BaseModel):
    document_id: str


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


class TopicsResponse(BaseModel):
    topics: list[Topic]


@router.post("/topics", response_model=TopicsResponse)
async def get_topics(payload: TopicsRequest) -> TopicsResponse:
    try:
        vector_store = load_vector_store()
        all_docs = list(vector_store.docstore._dict.values())
        chunks = [
            d for d in all_docs
            if getattr(d, "metadata", {}).get("document_id") == payload.document_id
        ]
    except Exception as exc:
        logger.warning("Vector store error: %s", exc)
        raise HTTPException(status_code=404, detail="Document not found.")

    if not chunks:
        raise HTTPException(status_code=404, detail="No chunks for this document.")

    # Sample evenly across the document for broad coverage
    step = max(1, len(chunks) // 8)
    sampled = chunks[::step][:8]
    context = "\n\n---\n\n".join(c.page_content[:350] for c in sampled)

    llm = get_llm()
    try:
        chain = _TOPICS_PROMPT | llm
        result = chain.invoke({"context": context})
        raw = result.content.strip()
    except Exception as exc:
        logger.error("LLM topics error: %s", exc)
        raise HTTPException(status_code=502, detail=f"LLM error: {exc}")

    topics: list[Topic] = []
    for line in raw.splitlines():
        line = line.strip()
        if not line.startswith("TOPIC:"):
            continue
        rest = line[len("TOPIC:"):].strip()
        if "|" in rest:
            title, desc = rest.split("|", 1)
            topics.append(Topic(title=title.strip(), description=desc.strip()))
        else:
            topics.append(Topic(title=rest.strip(), description=""))

    if not topics:
        raise HTTPException(status_code=502, detail="Could not extract topics.")

    return TopicsResponse(topics=topics)

from app.routes import topics
app.include_router(topics.router, prefix="/api", tags=["Topics"])

Same shape as topics: sample a few chunks, ask for exactly four starter questions, return a clean list.

"""
POST /api/suggestions

Generates 4 document-specific question suggestions using the LLM + a sample
of document chunks. Returns a clean JSON list — no full RAG pipeline needed.
"""

import logging

from fastapi import APIRouter, HTTPException
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel

from app.services.document_processor import load_vector_store
from app.services.llm import get_llm

logger = logging.getLogger(__name__)
router = APIRouter()

_SUGGESTIONS_PROMPT = ChatPromptTemplate.from_messages([
    (
        "system",
        (
            "You are a helpful learning assistant. Based on the document excerpts provided, "
            "generate exactly 4 specific, insightful questions a student would genuinely want to ask. "
            "Each question must be unique, targeting a different concept or section. "
            "Return ONLY the 4 questions, one per line, no numbering, no bullets, no extra text."
        ),
    ),
    ("human", "Document excerpts:\n\n{context}"),
])


class SuggestionsRequest(BaseModel):
    document_id: str


class SuggestionsResponse(BaseModel):
    suggestions: list[str]


@router.post("/suggestions", response_model=SuggestionsResponse)
async def get_suggestions(payload: SuggestionsRequest) -> SuggestionsResponse:
    try:
        vector_store = load_vector_store()
        # Scan docstore directly — avoids FAISS filter k-cutoff issues
        all_docs = list(vector_store.docstore._dict.values())
        docs = [
            d for d in all_docs
            if getattr(d, "metadata", {}).get("document_id") == payload.document_id
        ][:6]
    except Exception as exc:
        logger.warning("Vector store lookup failed: %s", exc)
        raise HTTPException(status_code=404, detail="Document not found in index.")

    if not docs:
        raise HTTPException(status_code=404, detail="No chunks found for this document.")

    context = "\n\n---\n\n".join(doc.page_content[:400] for doc in docs)

    llm = get_llm()
    try:
        chain = _SUGGESTIONS_PROMPT | llm
        result = chain.invoke({"context": context})
        raw = result.content.strip()
    except Exception as exc:
        logger.error("LLM suggestions error: %s", exc)
        raise HTTPException(status_code=502, detail=f"LLM error: {exc}")

    suggestions = [
        line.strip()
        for line in raw.splitlines()
        if line.strip() and len(line.strip()) > 10
    ][:4]

    if len(suggestions) < 2:
        raise HTTPException(status_code=502, detail="Could not generate suggestions.")

    return SuggestionsResponse(suggestions=suggestions)

from app.routes import suggestions
app.include_router(suggestions.router, prefix="/api", tags=["Suggestions"])

A service (write before its route) that scores an answer on faithfulness, answer-relevancy, context-precision, and — only with a ground truth — context-recall, then appends each run to a JSONL log.

uv add ragas datasets

"""
RAGAS-based evaluation of RAG outputs.

Metrics computed
----------------
- faithfulness       : Is the answer grounded in the retrieved context?
- answer_relevancy   : How relevant is the answer to the question?
- context_precision  : Are the retrieved contexts precise and on-topic?
- context_recall     : (Requires ground_truth) Did we retrieve what was needed?

Results are logged to both stdout (structlog) and a JSONL file for persistence.
"""

import json
import logging
import os
from datetime import datetime, timezone
from pathlib import Path

import structlog

from app.config import get_settings

logger = structlog.get_logger(__name__)


def _ensure_log_dir(log_path: str) -> Path:
    path = Path(log_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    return path


def evaluate_rag(
    question: str,
    answer: str,
    contexts: list[str],
    ground_truth: str | None = None,
) -> dict:
    """
    Run RAGAS evaluation and return a metrics dict.

    Parameters
    ----------
    question     : The original user question.
    answer       : The LLM-generated answer.
    contexts     : List of retrieved context strings used for generation.
    ground_truth : Optional reference answer (enables context_recall).

    Returns
    -------
    Dict with metric names → float scores (None if metric could not be computed).
    """
    try:
        from datasets import Dataset
        from ragas import evaluate
        from ragas.metrics import (
            answer_relevancy,
            context_precision,
            faithfulness,
        )

        metrics_to_run = [faithfulness, answer_relevancy, context_precision]

        # context_recall requires ground_truth
        if ground_truth:
            from ragas.metrics import context_recall
            metrics_to_run.append(context_recall)

        # Build a single-row HuggingFace Dataset (ragas input format)
        data = {
            "question": [question],
            "answer": [answer],
            "contexts": [contexts],
        }
        if ground_truth:
            data["ground_truth"] = [ground_truth]

        dataset = Dataset.from_dict(data)

        settings = get_settings()

        # RAGAS uses LangChain under the hood; reuse the SAME provider seam
        # as the rest of the app (Claude by default) plus local embeddings.
        from langchain_community.embeddings import HuggingFaceEmbeddings

        from app.services.llm import get_llm

        ragas_llm = get_llm()

        ragas_embeddings = HuggingFaceEmbeddings(
            model_name=settings.embedding_model,
            model_kwargs={"device": "cpu"},
            encode_kwargs={"normalize_embeddings": True},
        )

        result = evaluate(
            dataset,
            metrics=metrics_to_run,
            llm=ragas_llm,
            embeddings=ragas_embeddings,
            raise_exceptions=False,
        )

        scores = result.to_pandas().iloc[0].to_dict()

        metrics = {
            "faithfulness": _safe_float(scores.get("faithfulness")),
            "answer_relevancy": _safe_float(scores.get("answer_relevancy")),
            "context_precision": _safe_float(scores.get("context_precision")),
            "context_recall": _safe_float(scores.get("context_recall")),
        }

    except Exception as exc:
        logger.warning("RAGAS evaluation failed", error=str(exc))
        metrics = {
            "faithfulness": None,
            "answer_relevancy": None,
            "context_precision": None,
            "context_recall": None,
        }

    _log_metrics(question, answer, metrics)
    return metrics


def _safe_float(value) -> float | None:
    try:
        return round(float(value), 4)
    except (TypeError, ValueError):
        return None


def _log_metrics(question: str, answer: str, metrics: dict) -> None:
    """Append evaluation record to the JSONL log file and emit to stdout."""
    settings = get_settings()
    log_path = _ensure_log_dir(settings.eval_log_path)

    record = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "question": question[:200],
        "answer_preview": answer[:200],
        "metrics": metrics,
    }

    # Structured log to stdout
    logger.info("ragas_eval", **record)

    # Persist to JSONL file
    with open(log_path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps(record) + "\n")

Append the eval schemas, then add the route that delegates to evaluate_rag, then one more line in main.py.

# ── Evaluation ────────────────────────────────────────────────────────────────
# Append these to models.py when you build the evaluate route.

class EvalRequest(BaseModel):
    question: str
    answer: str
    contexts: list[str]
    ground_truth: str | None = None


class EvalMetrics(BaseModel):
    faithfulness: float | None = None
    answer_relevancy: float | None = None
    context_precision: float | None = None
    context_recall: float | None = None


class EvalResponse(BaseModel):
    metrics: EvalMetrics
    logged: bool

"""
POST /api/evaluate

Trigger RAGAS evaluation on a previously generated answer.
Metrics are returned in the response AND persisted to a JSONL log file.
"""

import logging

from fastapi import APIRouter, HTTPException, status

from app.models import EvalMetrics, EvalRequest, EvalResponse
from app.services.evaluator import evaluate_rag

logger = logging.getLogger(__name__)
router = APIRouter()


@router.post(
    "/evaluate",
    response_model=EvalResponse,
    summary="Evaluate RAG output quality with RAGAS",
)
async def evaluate_output(payload: EvalRequest) -> EvalResponse:
    """
    Run RAGAS evaluation metrics on a question/answer/context triplet.

    - **faithfulness**      — Is the answer supported by the contexts?
    - **answer_relevancy**  — Is the answer relevant to the question?
    - **context_precision** — Are the contexts precise and on-topic?
    - **context_recall**    — (Only when `ground_truth` is provided.)

    Scores are appended to the `EVAL_LOG_PATH` JSONL file for
    offline analysis and observability.
    """
    if not payload.contexts:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="At least one context string is required for evaluation.",
        )

    try:
        metrics_dict = evaluate_rag(
            question=payload.question,
            answer=payload.answer,
            contexts=payload.contexts,
            ground_truth=payload.ground_truth,
        )
    except Exception as exc:
        logger.exception("Evaluation endpoint error.")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Evaluation error: {exc}",
        )

    return EvalResponse(
        metrics=EvalMetrics(**metrics_dict),
        logged=True,
    )

from app.routes import evaluate
app.include_router(evaluate.router, prefix="/api", tags=["Evaluation"])

The one place Claude cannot serve: audio. This route rewrites the answer into a warm spoken script (using the primary LLM — Claude), then streams PCM16 audio from an OpenAI-family audio model and wraps it in a WAV container for the browser.

"""
POST /api/tts

Uses openai/gpt-4o-audio-preview via OpenRouter.
Parses the SSE stream directly with httpx to reliably extract PCM16 audio chunks,
then wraps them in a WAV container for browser playback.
"""

import base64
import json
import logging
import re
import struct

import httpx
from fastapi import APIRouter, HTTPException
from fastapi.responses import Response
from pydantic import BaseModel

from app.config import get_settings
from app.services.llm import get_llm
from langchain_core.prompts import ChatPromptTemplate

logger = logging.getLogger(__name__)
router = APIRouter()

ALLOWED_VOICES = {"alloy", "echo", "fable", "onyx", "nova", "shimmer"}

_SPOKEN_REWRITE_PROMPT = ChatPromptTemplate.from_messages([
    (
        "system",
        (
            "You are a skilled audio narrator for a learning app. "
            "Rewrite the following text as a natural, engaging spoken explanation — "
            "like a knowledgeable teacher talking warmly to a curious student. "
            "Remove ALL markdown (headers, bullets, bold, asterisks, code blocks). "
            "Use smooth transitions. Be clear and engaging. "
            "Return ONLY the spoken script, nothing else."
        ),
    ),
    ("human", "{text}"),
])


def _strip_markdown(text: str) -> str:
    text = re.sub(r"#{1,6}\s+", "", text)
    text = re.sub(r"\*{1,3}(.+?)\*{1,3}", r"\1", text)
    text = re.sub(r"`{1,3}[^`]*`{1,3}", "", text)
    text = re.sub(r"\[([^\]]+)\]\([^)]+\)", r"\1", text)
    text = re.sub(r"^\s*[-*+]\s+", "", text, flags=re.MULTILINE)
    text = re.sub(r"\n{2,}", ". ", text)
    text = re.sub(r"\n", " ", text)
    return text.strip()


def _pcm16_to_wav(pcm_data: bytes, sample_rate: int = 24000, channels: int = 1) -> bytes:
    data_size   = len(pcm_data)
    byte_rate   = sample_rate * channels * 2
    block_align = channels * 2
    header = struct.pack(
        "<4sI4s4sIHHIIHH4sI",
        b"RIFF", 36 + data_size,
        b"WAVE",
        b"fmt ", 16,
        1, channels, sample_rate,
        byte_rate, block_align, 16,
        b"data", data_size,
    )
    return header + pcm_data


async def _stream_pcm16(url: str, headers: dict, body: dict) -> list[bytes]:
    """Extracted for testability — streams OpenRouter SSE and returns PCM16 chunks."""
    pcm_chunks: list[bytes] = []
    async with httpx.AsyncClient(timeout=60) as client:
        async with client.stream("POST", url, headers=headers, json=body) as resp:
            if resp.status_code != 200:
                raw = await resp.aread()
                raise HTTPException(status_code=502, detail=raw.decode())
            async for line in resp.aiter_lines():
                if not line.startswith("data: "):
                    continue
                payload_str = line[6:]
                if payload_str.strip() == "[DONE]":
                    break
                try:
                    data = json.loads(payload_str)
                    choices = data.get("choices", [])
                    if not choices:
                        continue
                    audio = choices[0].get("delta", {}).get("audio") or {}
                    chunk_b64 = audio.get("data")
                    if chunk_b64:
                        pcm_chunks.append(base64.b64decode(chunk_b64))
                except Exception:
                    continue
    return pcm_chunks


class TTSRequest(BaseModel):
    text: str
    voice: str = "nova"


@router.post("/tts", summary="AI speech via gpt-4o-audio-preview on OpenRouter")
async def text_to_speech(payload: TTSRequest) -> Response:
    settings = get_settings()
    voice = payload.voice if payload.voice in ALLOWED_VOICES else "nova"

    # Claude has no audio model, so TTS uses an OpenAI-family model, either
    # OpenAI directly or the same model via the OpenRouter gateway.
    if settings.tts_provider.lower() == "openrouter":
        audio_base = settings.openrouter_base_url
        audio_key = settings.openrouter_api_key
        audio_model = f"openai/{settings.tts_model}"
    else:
        audio_base = settings.openai_base_url
        audio_key = settings.openai_api_key
        audio_model = settings.tts_model

    # Stage 1: LLM rewrite
    llm = get_llm()
    try:
        chain = _SPOKEN_REWRITE_PROMPT | llm
        result = chain.invoke({"text": payload.text})
        spoken_text = result.content.strip()
    except Exception as exc:
        logger.warning("Audio rewrite failed (%s) — using stripped text.", exc)
        spoken_text = _strip_markdown(payload.text)

    # Stage 2: Stream audio via raw httpx SSE
    try:
        pcm_chunks = await _stream_pcm16(
            url=f"{audio_base}/chat/completions",
            headers={
                "Authorization": f"Bearer {audio_key}",
                "Content-Type": "application/json",
                "HTTP-Referer": "https://my-learning-ai-assistant.local",
                "X-Title": "My Learning AI Assistant",
            },
            body={
                "model": audio_model,
                "modalities": ["text", "audio"],
                "audio": {"voice": voice, "format": "pcm16"},
                "messages": [{"role": "user", "content": spoken_text}],
                "stream": True,
            },
        )
    except HTTPException:
        raise
    except Exception as exc:
        logger.error("TTS stream error: %s", exc)
        raise HTTPException(status_code=502, detail=f"TTS error: {exc}")

    if not pcm_chunks:
        raise HTTPException(status_code=502, detail="No audio data received from model.")

    wav_bytes = _pcm16_to_wav(b"".join(pcm_chunks))
    return Response(
        content=wav_bytes,
        media_type="audio/wav",
        headers={"Content-Disposition": "inline; filename=speech.wav"},
    )

from app.routes import tts
app.include_router(tts.router, prefix="/api", tags=["Audio"])

You grew both files one feature at a time. Here is the finished shape to check yours against — six routers wired in, all schemas present. This matches the reference repo exactly.

"""
FastAPI application entry point.

Registers all routers, configures CORS, and sets up structured logging.
Run with: uvicorn app.main:app --reload
"""

import logging
import sys

import structlog
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse

from app.config import get_settings
from app.routes import evaluate, query, suggestions, topics, tts, upload

# ── Structured logging setup ─────────────────────────────────────────────────

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.dev.ConsoleRenderer() if sys.stderr.isatty()
        else structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(logging.INFO),
    context_class=dict,
    logger_factory=structlog.PrintLoggerFactory(),
)

settings = get_settings()
logging.basicConfig(
    level=getattr(logging, settings.log_level.upper(), logging.INFO),
    stream=sys.stdout,
)

# ── Application ───────────────────────────────────────────────────────────────

app = FastAPI(
    title="My Learning AI Assistant",
    description=(
        "Advanced RAG API powered by Gemini 2.5 Flash (OpenRouter), "
        "Google Embeddings, FAISS, and RAGAS evaluation."
    ),
    version="1.0.0",
    docs_url="/docs",
    redoc_url="/redoc",
)

# ── CORS ──────────────────────────────────────────────────────────────────────

app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.cors_origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ── Routers ───────────────────────────────────────────────────────────────────

app.include_router(upload.router,       prefix="/api", tags=["Documents"])
app.include_router(query.router,        prefix="/api", tags=["RAG"])
app.include_router(evaluate.router,     prefix="/api", tags=["Evaluation"])
app.include_router(tts.router,          prefix="/api", tags=["Audio"])
app.include_router(suggestions.router,  prefix="/api", tags=["Suggestions"])
app.include_router(topics.router,       prefix="/api", tags=["Topics"])


# ── Health check ──────────────────────────────────────────────────────────────

@app.get("/health", tags=["System"])
async def health_check() -> JSONResponse:
    return JSONResponse({"status": "ok", "service": "my-learning-ai-assistant"})


# ── Root ──────────────────────────────────────────────────────────────────────

@app.get("/", tags=["System"])
async def root() -> JSONResponse:
    return JSONResponse({
        "message": "My Learning AI Assistant API",
        "docs": "/docs",
    })

"""
Pydantic request / response models shared across routes.
"""

from pydantic import BaseModel, Field


# ── Upload ────────────────────────────────────────────────────────────────────

class UploadResponse(BaseModel):
    message: str
    document_id: str
    chunk_count: int
    description: str


# ── Query ─────────────────────────────────────────────────────────────────────

class QueryRequest(BaseModel):
    question: str = Field(..., min_length=3, max_length=2000)
    document_id: str | None = Field(
        default=None,
        description="If provided, restrict retrieval to this document's chunks.",
    )
    stream: bool = False


class SourceChunk(BaseModel):
    content: str
    page: int | None = None
    score: float | None = None


class QueryResponse(BaseModel):
    answer: str
    sources: list[SourceChunk]
    rewritten_query: str
    document_id: str | None = None


# ── Evaluation ────────────────────────────────────────────────────────────────

class EvalRequest(BaseModel):
    question: str
    answer: str
    contexts: list[str]
    ground_truth: str | None = None


class EvalMetrics(BaseModel):
    faithfulness: float | None = None
    answer_relevancy: float | None = None
    context_precision: float | None = None
    context_recall: float | None = None


class EvalResponse(BaseModel):
    metrics: EvalMetrics
    logged: bool

uv tracked your deps in pyproject.toml as you went. For the Dockerfile and CI (which install from a requirements file), export a pinned list — or write it out. This is the full set the project ends up needing:

uv pip compile pyproject.toml -o requirements.txt   # or: uv export --no-hashes -o requirements.txt

# Core API framework
fastapi==0.115.5
uvicorn[standard]==0.32.1
python-multipart==0.0.12

# LangChain ecosystem
langchain==0.3.9
langchain-community==0.3.9
langchain-openai==0.2.10
langchain-text-splitters==0.3.2
langchain-huggingface==0.1.2

# Vector store
faiss-cpu==1.9.0.post1

# PDF processing — multi-engine for maximum compatibility
pypdf==5.1.0
pdfplumber==0.11.4
pymupdf==1.25.3
# OCR for scanned / image-only PDFs (requires tesseract + poppler system packages)
pytesseract==0.3.13
pdf2image==1.17.0
pillow==11.0.0

# Re-ranking (Cross-Encoder)
sentence-transformers==3.3.1

# RAG Evaluation
ragas==0.2.6
datasets==3.1.0

# Resilience / retry
tenacity==9.0.0

# Config & environment
pydantic-settings==2.6.1
python-dotenv==1.0.1

# HTTP client
httpx==0.28.0

# Logging
structlog==24.4.0

# Testing
pytest==8.3.4
pytest-asyncio==0.24.0
httpx==0.28.0

# Linting / formatting
ruff==0.8.3

langchain-anthropic==0.3.3
anthropic>=0.42.0

A two-stage build on python:3.11-slim: install deps in a builder, copy just the installed packages + app into a lean runtime, run as non-root. Provider-independent.

# ─── Builder stage ────────────────────────────────────────────────────────────
FROM python:3.11-slim AS builder

WORKDIR /build

# Install build dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    gcc \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --upgrade pip && \
    pip install --prefix=/install --no-cache-dir -r requirements.txt


# ─── Runtime stage ────────────────────────────────────────────────────────────
FROM python:3.11-slim AS runtime

WORKDIR /app

# Copy installed packages from builder
COPY --from=builder /install /usr/local

# Copy application source
COPY app/ ./app/

# Non-root user for security
RUN addgroup --system appgroup && adduser --system --ingroup appgroup appuser
USER appuser

# Persistent volume mounts (FAISS index and logs)
VOLUME ["/app/faiss_index", "/app/logs"]

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

Wires both services: backend on :8000, frontend on :3000, a shared network, named volumes for the FAISS index and logs, and a health-check gate so the frontend waits for the backend.

version: "3.9"

# ── Shared network ────────────────────────────────────────────────────────────
networks:
  app_net:
    driver: bridge

# ── Named volumes for persistence ─────────────────────────────────────────────
volumes:
  faiss_data:
  eval_logs:

services:

  # ── FastAPI Backend ──────────────────────────────────────────────────────────
  backend:
    build:
      context: ./backend
      dockerfile: Dockerfile
      target: runtime
    container_name: mlai_backend
    restart: unless-stopped
    env_file:
      - ./backend/.env          # copy .env.example → .env and fill values
    environment:
      - ALLOWED_ORIGINS=http://localhost:3000
      - FAISS_INDEX_PATH=/data/faiss_index
      - EVAL_LOG_PATH=/data/logs/ragas_eval.jsonl
    volumes:
      - faiss_data:/data/faiss_index
      - eval_logs:/data/logs
    ports:
      - "8000:8000"
    networks:
      - app_net
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3
      start_period: 15s

  # ── Next.js Frontend ─────────────────────────────────────────────────────────
  frontend:
    build:
      context: ./frontend
      dockerfile: Dockerfile
      args:
        NEXT_PUBLIC_API_URL: http://backend:8000
    container_name: mlai_frontend
    restart: unless-stopped
    environment:
      - NEXT_PUBLIC_API_URL=http://backend:8000
    ports:
      - "3000:3000"
    networks:
      - app_net
    depends_on:
      backend:
        condition: service_healthy

Commit a blank-key template (your real .env is git-ignored):

# ─── LLM provider ─────────────────────────────────────────────────────────────
# "anthropic" (primary) | "openai" | "openrouter" (demo fallback)
LLM_PROVIDER=anthropic

# Set ONLY the key for the provider you selected above.
# Anthropic: https://console.anthropic.com/settings/keys
ANTHROPIC_API_KEY=Your_Anthropic_API_KEY
# OpenAI:    https://platform.openai.com/api-keys
# OPENAI_API_KEY=Your_OpenAI_API_KEY
# OpenRouter (demo fallback): https://openrouter.ai/keys
# OPENROUTER_API_KEY=Your_OpenRouter_API_KEY

# ─── Text-to-speech (audio) ───────────────────────────────────────────────────
# Claude has no audio model, so TTS uses an OpenAI-family model.
# "openai" needs OPENAI_API_KEY; "openrouter" reuses OPENROUTER_API_KEY.
TTS_PROVIDER=openai

# NOTE: Embeddings run locally via sentence-transformers — no API key needed.

# ─── Application ──────────────────────────────────────────────────────────────
ALLOWED_ORIGINS=http://localhost:3000
FAISS_INDEX_PATH=./faiss_index
LOG_LEVEL=INFO
EVAL_LOG_PATH=./logs/ragas_eval.jsonl

The ruff + pytest config (uv already made a pyproject.toml; add these tool sections):

[project]
name = "my-learning-ai-assistant-backend"
version = "1.0.0"
requires-python = ">=3.11"

[tool.ruff]
line-length = 100
target-version = "py311"

[tool.ruff.lint]
select = ["E", "F", "I", "UP", "B", "SIM"]
ignore = ["E501"]

[tool.ruff.lint.isort]
known-first-party = ["app"]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

And the GitHub Actions workflow — lint, test, build both images, deploy placeholder, on every push:

name: CI/CD — My Learning AI Assistant

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

# Cancel redundant runs on the same branch
concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}
  cancel-in-progress: true

env:
  PYTHON_VERSION: "3.11"
  NODE_VERSION: "20"

jobs:

  # ════════════════════════════════════════════════════════════════════════════
  # 1. Backend — Lint, Type-check, Test
  # ════════════════════════════════════════════════════════════════════════════
  backend-ci:
    name: "Backend — Lint & Test"
    runs-on: ubuntu-latest
    defaults:
      run:
        working-directory: backend

    steps:
      - uses: actions/checkout@v4

      - name: Set up Python ${{ env.PYTHON_VERSION }}
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: pip

      - name: Install dependencies
        run: pip install --upgrade pip && pip install -r requirements.txt

      - name: Lint with Ruff
        run: ruff check . --output-format=github

      - name: Format check with Ruff
        run: ruff format --check .

      - name: Run tests
        env:
          # Provide dummy keys so settings validation passes in CI
          OPENROUTER_API_KEY: ci-dummy-key
          GOOGLE_API_KEY: ci-dummy-key
        run: pytest tests/ -v --tb=short

  # ════════════════════════════════════════════════════════════════════════════
  # 2. Frontend — Lint, Type-check, Build
  # ════════════════════════════════════════════════════════════════════════════
  frontend-ci:
    name: "Frontend — Lint, Type-check & Build"
    runs-on: ubuntu-latest
    defaults:
      run:
        working-directory: frontend

    steps:
      - uses: actions/checkout@v4

      - name: Set up Node.js ${{ env.NODE_VERSION }}
        uses: actions/setup-node@v4
        with:
          node-version: ${{ env.NODE_VERSION }}
          cache: npm
          cache-dependency-path: frontend/package-lock.json

      - name: Install dependencies
        run: npm ci

      - name: Lint (ESLint)
        run: npm run lint

      - name: TypeScript check
        run: npm run type-check

      - name: Build
        env:
          NEXT_PUBLIC_API_URL: http://localhost:8000
        run: npm run build

  # ════════════════════════════════════════════════════════════════════════════
  # 3. Docker Build — validate images build cleanly
  # ════════════════════════════════════════════════════════════════════════════
  docker-build:
    name: "Docker — Build Images"
    runs-on: ubuntu-latest
    needs: [backend-ci, frontend-ci]
    # Only build Docker images on pushes (not PRs from forks)
    if: github.event_name == 'push'

    steps:
      - uses: actions/checkout@v4

      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v3

      - name: Build backend image
        uses: docker/build-push-action@v5
        with:
          context: ./backend
          push: false
          tags: mlai-backend:${{ github.sha }}
          cache-from: type=gha
          cache-to: type=gha,mode=max

      - name: Build frontend image
        uses: docker/build-push-action@v5
        with:
          context: ./frontend
          push: false
          tags: mlai-frontend:${{ github.sha }}
          build-args: |
            NEXT_PUBLIC_API_URL=http://backend:8000
          cache-from: type=gha
          cache-to: type=gha,mode=max

  # ════════════════════════════════════════════════════════════════════════════
  # 4. Deploy (placeholder — wire up to your infra)
  # ════════════════════════════════════════════════════════════════════════════
  deploy:
    name: "Deploy to Production"
    runs-on: ubuntu-latest
    needs: [docker-build]
    if: github.ref == 'refs/heads/main' && github.event_name == 'push'
    environment: production

    steps:
      - uses: actions/checkout@v4

      - name: Deploy notification
        run: |
          echo "✅ All CI checks passed."
          echo "🚀 Add your deployment steps here (e.g. push to registry, SSH deploy, Fly.io, etc.)"
          echo "   SHA: ${{ github.sha }}"

A full API suite that patches every heavy service (FAISS, the LLM, embeddings, httpx) so it runs instantly with no GPU, network, or keys. It mocks at the route boundary, so it is provider-agnostic — repo verbatim.

uv add --dev pytest pytest-asyncio ruff

"""
Full API test suite.

Patches all heavy services (FAISS, LLM, embeddings, httpx) so tests run
instantly without GPU, network, or API keys.
"""

import base64
import io
import struct
from unittest.mock import AsyncMock, MagicMock, patch

import pytest
from httpx import AsyncClient, ASGITransport

# ── Shared mock return values ─────────────────────────────────────────────────

_DOC_ID     = "test-doc-id-1234"
_CHUNK_COUNT = 11

_MOCK_RAG = {
    "answer": "The document discusses machine learning fundamentals.",
    "rewritten_query": "What are the key ML concepts?",
    "sources": [{"content": "ML is a subfield of AI.", "page": 1, "score": 0.95}],
}

_MOCK_EVAL = {
    "faithfulness": 0.91,
    "answer_relevancy": 0.88,
    "context_precision": 0.85,
    "context_recall": None,
}

# Minimal valid WAV bytes for TTS mock
def _fake_wav() -> bytes:
    pcm = b"\x00\x00" * 100
    header = struct.pack(
        "<4sI4s4sIHHIIHH4sI",
        b"RIFF", 36 + len(pcm), b"WAVE",
        b"fmt ", 16, 1, 1, 24000, 48000, 2, 16,
        b"data", len(pcm),
    )
    return header + pcm


# ── Mock document in FAISS docstore ──────────────────────────────────────────

class _FakeDoc:
    def __init__(self):
        self.page_content = "This is a chunk about machine learning concepts."
        self.metadata = {"document_id": _DOC_ID, "page": 1}

_fake_vs = MagicMock()
_fake_vs.docstore._dict = {f"id-{i}": _FakeDoc() for i in range(6)}


# ── Fixtures ──────────────────────────────────────────────────────────────────

@pytest.fixture(autouse=True)
def patch_services():
    """Patch every external service before each test."""
    mock_llm_result = MagicMock()
    mock_llm_result.content = (
        "TOPIC: Machine Learning | Core concepts of ML.\n"
        "TOPIC: Neural Networks | Deep learning architectures.\n"
        "TOPIC: Evaluation Metrics | How models are measured.\n"
    )

    mock_suggestions_result = MagicMock()
    mock_suggestions_result.content = (
        "What is supervised learning?\n"
        "How do neural networks work?\n"
        "What are evaluation metrics?\n"
        "What is overfitting?\n"
    )

    _pcm_bytes = b"\x00\x01" * 100

    with (
        patch("app.routes.upload.ingest_pdf",        MagicMock(return_value=(_DOC_ID, _CHUNK_COUNT))),
        patch("app.routes.query.run_rag",             MagicMock(return_value=_MOCK_RAG)),
        patch("app.routes.evaluate.evaluate_rag",     MagicMock(return_value=_MOCK_EVAL)),
        patch("app.routes.suggestions.load_vector_store", MagicMock(return_value=_fake_vs)),
        patch("app.routes.topics.load_vector_store",      MagicMock(return_value=_fake_vs)),
        # Patch _get_llm to prevent ChatOpenAI instantiation (avoids httpx.AsyncClient corruption)
        patch("app.routes.suggestions._get_llm", MagicMock(return_value=MagicMock())),
        patch("app.routes.topics._get_llm",      MagicMock(return_value=MagicMock())),
        patch("app.routes.tts._get_llm",         MagicMock(return_value=MagicMock())),
        # Patch each prompt chain's __or__ so invoke() returns the right fixture
        patch("app.routes.suggestions._SUGGESTIONS_PROMPT",
              MagicMock(__or__=MagicMock(return_value=MagicMock(invoke=MagicMock(return_value=mock_suggestions_result))))),
        patch("app.routes.topics._TOPICS_PROMPT",
              MagicMock(__or__=MagicMock(return_value=MagicMock(invoke=MagicMock(return_value=mock_llm_result))))),
        patch("app.routes.tts._SPOKEN_REWRITE_PROMPT",
              MagicMock(__or__=MagicMock(return_value=MagicMock(invoke=MagicMock(return_value=MagicMock(content="Spoken script.")))))),
        # Patch the extracted streaming helper — safe, no httpx module corruption
        patch("app.routes.tts._stream_pcm16", AsyncMock(return_value=[_pcm_bytes])),
        patch("app.config.Settings.openrouter_api_key", "test-or-key", create=True),
    ):
        yield


@pytest.fixture
def app():
    from app.main import app as fastapi_app
    return fastapi_app


@pytest.fixture
async def client(app):
    async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as ac:
        yield ac


# ── Health ────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_health(client):
    r = await client.get("/health")
    assert r.status_code == 200
    assert r.json()["status"] == "ok"


# ── Upload ────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_upload_success(client):
    r = await client.post(
        "/api/upload",
        files={"file": ("doc.pdf", io.BytesIO(b"%PDF-1.4 fake"), "application/pdf")},
        data={"description": "A document about machine learning."},
    )
    assert r.status_code == 201
    body = r.json()
    assert body["document_id"] == _DOC_ID
    assert body["chunk_count"] == _CHUNK_COUNT


@pytest.mark.asyncio
async def test_upload_rejects_non_pdf(client):
    r = await client.post(
        "/api/upload",
        files={"file": ("doc.txt", io.BytesIO(b"hello"), "text/plain")},
        data={"description": "Some description here."},
    )
    assert r.status_code == 415


@pytest.mark.asyncio
async def test_upload_requires_description(client):
    r = await client.post(
        "/api/upload",
        files={"file": ("doc.pdf", io.BytesIO(b"%PDF"), "application/pdf")},
        data={"description": "hi"},   # too short (min 5)
    )
    assert r.status_code == 422


# ── Query ─────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_query_success(client):
    r = await client.post("/api/query", json={"question": "What is machine learning?"})
    assert r.status_code == 200
    body = r.json()
    assert body["answer"] == _MOCK_RAG["answer"]
    assert len(body["sources"]) == 1
    assert "rewritten_query" in body


@pytest.mark.asyncio
async def test_query_with_document_id(client):
    r = await client.post("/api/query", json={"question": "Explain neural networks.", "document_id": _DOC_ID})
    assert r.status_code == 200


@pytest.mark.asyncio
async def test_query_too_short(client):
    r = await client.post("/api/query", json={"question": "Hi"})
    assert r.status_code == 422


# ── Evaluate ──────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_evaluate_success(client):
    r = await client.post("/api/evaluate", json={
        "question": "What is ML?",
        "answer": "ML is a subfield of AI.",
        "contexts": ["Machine learning is a subfield of artificial intelligence."],
    })
    assert r.status_code == 200
    body = r.json()
    assert "metrics" in body
    assert body["metrics"]["faithfulness"] == 0.91


# ── Suggestions ───────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_suggestions_success(client):
    r = await client.post("/api/suggestions", json={"document_id": _DOC_ID})
    assert r.status_code == 200
    body = r.json()
    assert "suggestions" in body
    assert len(body["suggestions"]) >= 2


@pytest.mark.asyncio
async def test_suggestions_invalid_doc(client):
    with patch("app.routes.suggestions.load_vector_store") as mock_vs:
        mock_vs.return_value.docstore._dict = {}
        r = await client.post("/api/suggestions", json={"document_id": "nonexistent"})
    assert r.status_code == 404


# ── Topics ────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_topics_success(client):
    r = await client.post("/api/topics", json={"document_id": _DOC_ID})
    assert r.status_code == 200
    body = r.json()
    assert "topics" in body
    assert len(body["topics"]) >= 1
    assert "title" in body["topics"][0]
    assert "description" in body["topics"][0]


# ── TTS ───────────────────────────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_tts_returns_wav(client):
    r = await client.post("/api/tts", json={"text": "Hello, this is a test.", "voice": "nova"})
    assert r.status_code == 200
    assert r.headers["content-type"] == "audio/wav"
    # Verify WAV header
    assert r.content[:4] == b"RIFF"
    assert r.content[8:12] == b"WAVE"


@pytest.mark.asyncio
async def test_tts_invalid_voice_falls_back(client):
    r = await client.post("/api/tts", json={"text": "Testing fallback voice.", "voice": "invalid_voice"})
    assert r.status_code == 200  # falls back to nova


@pytest.mark.asyncio
async def test_tts_empty_text_rejected(client):
    r = await client.post("/api/tts", json={"text": ""})
    # pydantic min_length not set, but empty string hits the "no audio" path
    assert r.status_code in (200, 422, 502)

uv run pytest -v

Everything is built and tested locally. Bring up both services in Docker:

# from the repo root — make sure backend/.env exists with your key
docker compose up --build

# backend  -> http://localhost:8000/docs
# frontend -> http://localhost:3000   (upload a PDF, ask by text or voice)