Scaffold YouTube Channel Knowledge Base (RAG)

.env.example

@@ -0,0 +1,24 @@
+# ---- LLM provider ----
+# Options: "ollama" (local, free) or "anthropic" (Claude, for deployed demo)
+LLM_PROVIDER=ollama
+
+# Local Ollama settings
+OLLAMA_MODEL=qwen2.5:7b-instruct
+OLLAMA_HOST=http://localhost:11434
+
+# Anthropic (Claude) settings — used when LLM_PROVIDER=anthropic
+ANTHROPIC_API_KEY=
+ANTHROPIC_MODEL=claude-haiku-4-5-20251001
+
+# ---- Embeddings (local, free) ----
+EMBED_MODEL=sentence-transformers/all-MiniLM-L6-v2
+
+# ---- Retrieval ----
+TOP_K=6
+CHUNK_TARGET_WORDS=180
+CHUNK_OVERLAP_WORDS=30
+
+# ---- Storage paths ----
+CHROMA_DIR=data/chroma
+TRANSCRIPT_DIR=data/transcripts
+COLLECTION_NAME=youtube_kb

.gitignore

@@ -0,0 +1,18 @@
+# Python
+__pycache__/
+*.py[cod]
+.venv/
+venv/
+*.egg-info/
+
+# Secrets
+.env
+
+# Data artifacts (rebuilt by ingestion)
+data/transcripts/
+data/chroma/
+data/*.json
+
+# Models / caches
+.cache/
+*.log

README.md

@@ -0,0 +1,78 @@
+---
+title: YouTube Channel Knowledge Base
+emoji: 🎥
+colorFrom: indigo
+colorTo: purple
+sdk: streamlit
+sdk_version: 1.40.0
+app_file: app.py
+pinned: false
+---
+
+# 🎥 YouTube Channel Knowledge Base (RAG)
+
+Turn an entire YouTube channel into a searchable, conversational knowledge base.
+Ask a question in plain language and get an answer **grounded in the video
+transcripts**, with citations that deep-link to the exact **video and timestamp**.
+
+## How it works
+
+```
+channel ─▶ yt-dlp (video list)
+         └▶ youtube-transcript-api (timed captions)
+                    │
+                    ▼
+        chunk (timestamp-anchored) ─▶ sentence-transformers embeddings ─▶ ChromaDB
+                                                                              │
+ question ─▶ embed ─▶ vector search ─▶ top-k excerpts ─▶ LLM (grounded) ─▶ answer + [n] citations
+```
+
+- **Embeddings:** local `sentence-transformers` — free, no API calls.
+- **LLM:** pluggable — `ollama` (local, free) for dev, `anthropic` (Claude) for the deployed demo.
+- **Citations:** every answer links back to `youtube.com/watch?v=…&t=…s`.
+
+## Quickstart (local)
+
+```powershell
+python -m venv .venv
+.\.venv\Scripts\Activate.ps1
+pip install -r requirements.txt
+copy .env.example .env          # then edit if needed
+
+# 1) Build the index for a channel (transcripts are cached to disk)
+python -m scripts.build_index --channel "@SomeChannel" --limit 50
+
+# 2) Run the app
+streamlit run app.py
+```
+
+Local answers use Ollama by default (`LLM_PROVIDER=ollama`). Make sure a model is
+pulled, e.g. `ollama pull qwen2.5:7b-instruct`.
+
+## Switching to Claude
+
+Set in `.env` (or as environment variables):
+
+```
+LLM_PROVIDER=anthropic
+ANTHROPIC_API_KEY=sk-ant-...
+ANTHROPIC_MODEL=claude-haiku-4-5-20251001
+```
+
+## Evaluation
+
+```powershell
+python -m eval.run_eval --testset eval/testset.json
+```
+
+Runs deterministic checks (citation validity, out-of-scope refusal) plus an
+LLM-as-judge score for faithfulness and citation quality.
+
+## Deploy (Hugging Face Spaces)
+
+1. Build the index locally, then commit `data/chroma/` (remove it from
+   `.gitignore` for the deploy commit) so the Space ships with a prebuilt index —
+   Spaces cannot run ingestion or Ollama.
+2. Create a **Streamlit** Space and push this repo.
+3. In the Space **Settings → Secrets**, set `LLM_PROVIDER=anthropic` and
+   `ANTHROPIC_API_KEY`.

app.py

@@ -0,0 +1,73 @@
+"""Streamlit UI for the YouTube Channel Knowledge Base.
+
+Ask a question -> grounded answer with clickable video + timestamp citations.
+"""
+from __future__ import annotations
+
+import streamlit as st
+
+from src.chunk import _fmt_timestamp
+from src.config import CONFIG
+from src.llm import get_provider
+from src.rag import answer_question
+from src.store import count
+
+st.set_page_config(page_title="YouTube Channel Knowledge Base", page_icon="🎥", layout="centered")
+
+st.title("🎥 YouTube Channel Knowledge Base")
+st.caption("Ask anything about the channel's videos — answers are grounded in the "
+           "transcripts and cite the exact video and timestamp.")
+
+with st.sidebar:
+    st.header("Status")
+    n = count()
+    st.metric("Indexed transcript chunks", n)
+    if n == 0:
+        st.warning("No index found. Run:\n\n`python -m scripts.build_index --channel \"@handle\"`")
+
+    st.divider()
+    st.subheader("Settings")
+    provider_choice = st.selectbox(
+        "Answer model",
+        options=["(use .env default)", "ollama", "anthropic"],
+        index=0,
+    )
+    top_k = st.slider("Sources retrieved", 3, 12, CONFIG.top_k)
+    st.caption(f"Embeddings: `{CONFIG.embed_model}` (local)")
+
+if "history" not in st.session_state:
+    st.session_state.history = []
+
+for q, ans in st.session_state.history:
+    with st.chat_message("user"):
+        st.markdown(q)
+    with st.chat_message("assistant"):
+        st.markdown(ans.text)
+
+question = st.chat_input("Ask a question about the channel...")
+if question:
+    with st.chat_message("user"):
+        st.markdown(question)
+
+    with st.chat_message("assistant"):
+        provider = None if provider_choice.startswith("(") else get_provider(provider_choice)
+        with st.spinner("Searching the channel and composing an answer..."):
+            try:
+                ans = answer_question(question, provider=provider, top_k=top_k)
+            except Exception as e:  # surface provider/setup errors clearly
+                st.error(f"Error: {e}")
+                st.stop()
+
+        st.markdown(ans.text)
+
+        if ans.sources:
+            st.markdown("**Sources**")
+            for i, s in enumerate(ans.sources, 1):
+                st.markdown(
+                    f"{i}. [{s.title} @ {_fmt_timestamp(s.start)}]({s.url})  "
+                    f"<span style='color:gray'>· similarity {s.score:.2f}</span>",
+                    unsafe_allow_html=True,
+                )
+        st.caption(f"Answered by: {ans.provider}")
+
+    st.session_state.history.append((question, ans))

eval/run_eval.py

@@ -0,0 +1,92 @@
+"""Lightweight evaluation harness.
+
+Two layers, matching the JD's "deterministic + LLM-as-judge + human review":
+
+1. Deterministic checks (no model): every cited [n] in the answer must map to a
+   real retrieved source, and out-of-scope questions must trigger a refusal.
+2. LLM-as-judge: a model scores faithfulness (is the answer supported by the
+   retrieved excerpts?) and citation usefulness on a 1-5 scale.
+
+Usage:
+    python -m eval.run_eval --testset eval/testset.json
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import re
+
+from src.llm import get_provider
+from src.rag import answer_question
+
+CITATION_RE = re.compile(r"\[(\d+)\]")
+REFUSAL_MARKERS = ("don't know", "do not know", "couldn't find", "could not find", "no information")
+
+JUDGE_SYSTEM = (
+    "You are a strict evaluator. Given a question, the retrieved transcript "
+    "excerpts, and an assistant answer, rate two things from 1-5:\n"
+    "  faithfulness: is every claim supported by the excerpts (5) or are there "
+    "unsupported/hallucinated claims (1)?\n"
+    "  citation_quality: are the [n] citations present and pointing at relevant "
+    "excerpts (5) or missing/wrong (1)?\n"
+    "Respond ONLY as JSON: {\"faithfulness\": int, \"citation_quality\": int, \"reason\": str}."
+)
+
+
+def deterministic_checks(answer_text: str, n_sources: int) -> dict:
+    cited = {int(m) for m in CITATION_RE.findall(answer_text)}
+    invalid = {c for c in cited if c < 1 or c > n_sources}
+    refused = any(m in answer_text.lower() for m in REFUSAL_MARKERS)
+    return {
+        "has_citations": bool(cited),
+        "all_citations_valid": len(invalid) == 0,
+        "invalid_citations": sorted(invalid),
+        "refused": refused,
+    }
+
+
+def judge(question: str, ans, judge_provider) -> dict:
+    excerpts = "\n\n".join(f"[{i}] {s.text}" for i, s in enumerate(ans.sources, 1))
+    user = (
+        f"Question: {question}\n\nExcerpts:\n{excerpts}\n\nAnswer:\n{ans.text}\n\n"
+        "Return the JSON now."
+    )
+    raw = judge_provider.complete(JUDGE_SYSTEM, user)
+    try:
+        start, end = raw.find("{"), raw.rfind("}")
+        return json.loads(raw[start : end + 1])
+    except Exception:
+        return {"faithfulness": None, "citation_quality": None, "reason": f"unparseable: {raw[:120]}"}
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--testset", default="eval/testset.json")
+    args = parser.parse_args()
+
+    with open(args.testset, "r", encoding="utf-8") as f:
+        cases = json.load(f)
+
+    answer_provider = get_provider()
+    judge_provider = get_provider()  # same backend; swap to a stronger model if desired
+
+    rows = []
+    for case in cases:
+        q = case["question"]
+        ans = answer_question(q, provider=answer_provider)
+        det = deterministic_checks(ans.text, len(ans.sources))
+        verdict = judge(q, ans, judge_provider) if ans.sources else {"faithfulness": None, "citation_quality": None, "reason": "no sources"}
+        rows.append({"question": q, "deterministic": det, "judge": verdict})
+        print(f"\nQ: {q}\n  deterministic: {det}\n  judge: {verdict}")
+
+    faith = [r["judge"]["faithfulness"] for r in rows if isinstance(r["judge"].get("faithfulness"), int)]
+    if faith:
+        print(f"\nMean faithfulness: {sum(faith)/len(faith):.2f}/5 over {len(faith)} cases")
+
+    with open("eval/results.json", "w", encoding="utf-8") as f:
+        json.dump(rows, f, indent=2, ensure_ascii=False)
+    print("\nWrote eval/results.json")
+
+
+if __name__ == "__main__":
+    main()

eval/testset.example.json

@@ -0,0 +1,10 @@
+[
+  {
+    "question": "What is the main topic the channel covers most often?",
+    "expected_points": ["mentions a recurring theme present in multiple videos"]
+  },
+  {
+    "question": "Ask something the channel has clearly never discussed (out-of-scope)",
+    "expected_points": ["assistant says it doesn't know based on this channel's videos"]
+  }
+]

requirements.txt

@@ -0,0 +1,13 @@
+# Core RAG stack
+streamlit>=1.40
+yt-dlp>=2024.12.13
+youtube-transcript-api>=0.6.2
+sentence-transformers>=3.0
+chromadb>=0.5.5
+anthropic>=0.40
+python-dotenv>=1.0
+tqdm>=4.66
+requests>=2.31
+
+# Optional: Whisper fallback for videos without captions (heavier install)
+# faster-whisper>=1.0

scripts/build_index.py

@@ -0,0 +1,33 @@
+"""CLI: ingest a YouTube channel and build the vector index.
+
+Usage:
+    python -m scripts.build_index --channel "@SomeChannel" --limit 50
+"""
+from __future__ import annotations
+
+import argparse
+
+from src.chunk import chunk_all
+from src.ingest import ingest_channel
+from src.store import count, index_chunks
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Build the YouTube knowledge base index.")
+    parser.add_argument("--channel", required=True, help="Channel handle (@name), URL, or name")
+    parser.add_argument("--limit", type=int, default=None, help="Max number of videos to ingest")
+    args = parser.parse_args()
+
+    records = ingest_channel(args.channel, limit=args.limit)
+    if not records:
+        print("No transcripts ingested — nothing to index.")
+        return
+
+    chunks = chunk_all(records)
+    print(f"Built {len(chunks)} chunks from {len(records)} videos. Embedding + indexing...")
+    index_chunks(chunks)
+    print(f"Done. Collection now holds {count()} chunks.")
+
+
+if __name__ == "__main__":
+    main()

src/chunk.py

@@ -0,0 +1,90 @@
+"""Turn transcript segments into overlapping, timestamp-anchored chunks.
+
+Each chunk carries the start time of its first segment so we can build a
+deep-link citation (youtube.com/watch?v=ID&t=SECONDS) back to the exact moment.
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .config import CONFIG
+
+
+@dataclass
+class Chunk:
+    id: str
+    text: str
+    video_id: str
+    title: str
+    start: float          # seconds into the video
+    url: str              # deep link with timestamp
+
+
+def _fmt_timestamp(seconds: float) -> str:
+    s = int(seconds)
+    h, s = divmod(s, 3600)
+    m, s = divmod(s, 60)
+    return f"{h:d}:{m:02d}:{s:02d}" if h else f"{m:d}:{s:02d}"
+
+
+def chunk_record(record: dict,
+                 target_words: int | None = None,
+                 overlap_words: int | None = None) -> list[Chunk]:
+    """Chunk one ingested {meta, segments} record into Chunk objects."""
+    target_words = target_words or CONFIG.chunk_target_words
+    overlap_words = overlap_words or CONFIG.chunk_overlap_words
+
+    meta = record["meta"]
+    segments = record["segments"]
+    video_id = meta["video_id"]
+    title = meta["title"]
+
+    chunks: list[Chunk] = []
+    buf_words: list[str] = []
+    buf_start: float | None = None
+    idx = 0
+
+    def flush(start: float):
+        nonlocal buf_words, idx
+        if not buf_words:
+            return
+        text = " ".join(buf_words).strip()
+        ts = int(start)
+        chunk = Chunk(
+            id=f"{video_id}-{idx}",
+            text=text,
+            video_id=video_id,
+            title=title,
+            start=start,
+            url=f"https://www.youtube.com/watch?v={video_id}&t={ts}s",
+        )
+        chunks.append(chunk)
+        idx += 1
+
+    for seg in segments:
+        if buf_start is None:
+            buf_start = seg["start"]
+        buf_words.extend(seg["text"].split())
+
+        if len(buf_words) >= target_words:
+            flush(buf_start)
+            # keep an overlap tail for context continuity
+            tail = buf_words[-overlap_words:] if overlap_words else []
+            buf_words = list(tail)
+            buf_start = seg["start"]
+
+    if buf_words and buf_start is not None:
+        flush(buf_start)
+
+    return chunks
+
+
+def chunk_all(records: list[dict]) -> list[Chunk]:
+    out: list[Chunk] = []
+    for rec in records:
+        out.extend(chunk_record(rec))
+    return out
+
+
+def citation_label(chunk: Chunk) -> str:
+    return f"{chunk.title} @ {_fmt_timestamp(chunk.start)}"

src/config.py

@@ -0,0 +1,42 @@
+"""Central configuration, loaded from environment / .env."""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+
+def _int(name: str, default: int) -> int:
+    try:
+        return int(os.getenv(name, str(default)))
+    except ValueError:
+        return default
+
+
+@dataclass(frozen=True)
+class Config:
+    # LLM
+    llm_provider: str = os.getenv("LLM_PROVIDER", "ollama").lower()
+    ollama_model: str = os.getenv("OLLAMA_MODEL", "qwen2.5:7b-instruct")
+    ollama_host: str = os.getenv("OLLAMA_HOST", "http://localhost:11434")
+    anthropic_api_key: str = os.getenv("ANTHROPIC_API_KEY", "")
+    anthropic_model: str = os.getenv("ANTHROPIC_MODEL", "claude-haiku-4-5-20251001")
+
+    # Embeddings
+    embed_model: str = os.getenv("EMBED_MODEL", "sentence-transformers/all-MiniLM-L6-v2")
+
+    # Retrieval / chunking
+    top_k: int = _int("TOP_K", 6)
+    chunk_target_words: int = _int("CHUNK_TARGET_WORDS", 180)
+    chunk_overlap_words: int = _int("CHUNK_OVERLAP_WORDS", 30)
+
+    # Storage
+    chroma_dir: str = os.getenv("CHROMA_DIR", "data/chroma")
+    transcript_dir: str = os.getenv("TRANSCRIPT_DIR", "data/transcripts")
+    collection_name: str = os.getenv("COLLECTION_NAME", "youtube_kb")
+
+
+CONFIG = Config()

src/embed.py

@@ -0,0 +1,23 @@
+"""Local sentence-transformers embeddings (free, no API calls)."""
+from __future__ import annotations
+
+from functools import lru_cache
+
+from .config import CONFIG
+
+
+@lru_cache(maxsize=1)
+def _model():
+    from sentence_transformers import SentenceTransformer
+
+    return SentenceTransformer(CONFIG.embed_model)
+
+
+def embed_texts(texts: list[str]) -> list[list[float]]:
+    model = _model()
+    vecs = model.encode(texts, normalize_embeddings=True, show_progress_bar=len(texts) > 64)
+    return [v.tolist() for v in vecs]
+
+
+def embed_query(text: str) -> list[float]:
+    return embed_texts([text])[0]

src/ingest.py

@@ -0,0 +1,130 @@
+"""Fetch a channel's video list and transcripts.
+
+- Video discovery: yt-dlp (flat playlist) -> ids, titles, upload dates.
+- Transcripts: youtube-transcript-api (timed caption segments).
+- Everything is cached to disk so re-runs are cheap and offline-friendly.
+"""
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import asdict, dataclass
+from typing import Optional
+
+from tqdm import tqdm
+
+from .config import CONFIG
+
+
+@dataclass
+class VideoMeta:
+    video_id: str
+    title: str
+    url: str
+    upload_date: Optional[str] = None
+
+
+@dataclass
+class Segment:
+    text: str
+    start: float  # seconds
+
+
+def _channel_videos_url(channel: str) -> str:
+    """Normalise a channel handle/URL into its /videos tab URL for yt-dlp."""
+    channel = channel.strip()
+    if channel.startswith("http"):
+        return channel if channel.rstrip("/").endswith("/videos") else channel.rstrip("/") + "/videos"
+    if channel.startswith("@"):
+        return f"https://www.youtube.com/{channel}/videos"
+    return f"https://www.youtube.com/@{channel}/videos"
+
+
+def list_channel_videos(channel: str, limit: Optional[int] = None) -> list[VideoMeta]:
+    """Return video metadata for a channel using yt-dlp's flat extractor."""
+    import yt_dlp
+
+    url = _channel_videos_url(channel)
+    opts = {"extract_flat": True, "quiet": True, "skip_download": True}
+    if limit:
+        opts["playlistend"] = limit
+
+    with yt_dlp.YoutubeDL(opts) as ydl:
+        info = ydl.extract_info(url, download=False)
+
+    entries = info.get("entries") or []
+    videos: list[VideoMeta] = []
+    for e in entries:
+        if not e or not e.get("id"):
+            continue
+        vid = e["id"]
+        videos.append(
+            VideoMeta(
+                video_id=vid,
+                title=e.get("title") or vid,
+                url=f"https://www.youtube.com/watch?v={vid}",
+                upload_date=e.get("upload_date"),
+            )
+        )
+    return videos
+
+
+def fetch_transcript(video_id: str, languages: Optional[list[str]] = None) -> Optional[list[Segment]]:
+    """Fetch timed caption segments for a video, or None if unavailable."""
+    from youtube_transcript_api import (
+        NoTranscriptFound,
+        TranscriptsDisabled,
+        YouTubeTranscriptApi,
+    )
+
+    languages = languages or ["en", "en-US", "en-GB"]
+    try:
+        raw = YouTubeTranscriptApi.get_transcript(video_id, languages=languages)
+    except (TranscriptsDisabled, NoTranscriptFound):
+        # Try any available language as a fallback.
+        try:
+            listing = YouTubeTranscriptApi.list_transcripts(video_id)
+            transcript = next(iter(listing))
+            raw = transcript.fetch()
+        except Exception:
+            return None
+    except Exception:
+        return None
+
+    return [Segment(text=s["text"].strip(), start=float(s["start"])) for s in raw if s["text"].strip()]
+
+
+def _cache_path(video_id: str) -> str:
+    return os.path.join(CONFIG.transcript_dir, f"{video_id}.json")
+
+
+def ingest_channel(channel: str, limit: Optional[int] = None) -> list[dict]:
+    """Discover videos + transcripts for a channel, caching each to disk.
+
+    Returns a list of {meta, segments} dicts for videos that had a transcript.
+    """
+    os.makedirs(CONFIG.transcript_dir, exist_ok=True)
+    videos = list_channel_videos(channel, limit=limit)
+    print(f"Found {len(videos)} videos for channel '{channel}'.")
+
+    ingested: list[dict] = []
+    missing = 0
+    for meta in tqdm(videos, desc="Transcripts"):
+        path = _cache_path(meta.video_id)
+        if os.path.exists(path):
+            with open(path, "r", encoding="utf-8") as f:
+                ingested.append(json.load(f))
+            continue
+
+        segments = fetch_transcript(meta.video_id)
+        if not segments:
+            missing += 1
+            continue
+
+        record = {"meta": asdict(meta), "segments": [asdict(s) for s in segments]}
+        with open(path, "w", encoding="utf-8") as f:
+            json.dump(record, f, ensure_ascii=False)
+        ingested.append(record)
+
+    print(f"Ingested {len(ingested)} transcripts ({missing} videos had no captions).")
+    return ingested

src/llm.py

@@ -0,0 +1,75 @@
+"""Pluggable LLM provider: Ollama (local) or Anthropic (Claude).
+
+Selected via LLM_PROVIDER. Both expose the same .complete(system, user) API so
+the rest of the app never needs to know which backend is in use.
+"""
+from __future__ import annotations
+
+from typing import Protocol
+
+import requests
+
+from .config import CONFIG
+
+
+class LLMProvider(Protocol):
+    name: str
+
+    def complete(self, system: str, user: str) -> str: ...
+
+
+class OllamaProvider:
+    name = "ollama"
+
+    def __init__(self, model: str | None = None, host: str | None = None):
+        self.model = model or CONFIG.ollama_model
+        self.host = (host or CONFIG.ollama_host).rstrip("/")
+
+    def complete(self, system: str, user: str) -> str:
+        resp = requests.post(
+            f"{self.host}/api/chat",
+            json={
+                "model": self.model,
+                "messages": [
+                    {"role": "system", "content": system},
+                    {"role": "user", "content": user},
+                ],
+                "stream": False,
+                "options": {"temperature": 0.2},
+            },
+            timeout=180,
+        )
+        resp.raise_for_status()
+        return resp.json()["message"]["content"].strip()
+
+
+class AnthropicProvider:
+    name = "anthropic"
+
+    def __init__(self, model: str | None = None, api_key: str | None = None):
+        from anthropic import Anthropic
+
+        key = api_key or CONFIG.anthropic_api_key
+        if not key:
+            raise RuntimeError("ANTHROPIC_API_KEY is not set but LLM_PROVIDER=anthropic.")
+        self.model = model or CONFIG.anthropic_model
+        self.client = Anthropic(api_key=key)
+
+    def complete(self, system: str, user: str) -> str:
+        msg = self.client.messages.create(
+            model=self.model,
+            max_tokens=1024,
+            temperature=0.2,
+            system=system,
+            messages=[{"role": "user", "content": user}],
+        )
+        return "".join(block.text for block in msg.content if block.type == "text").strip()
+
+
+def get_provider(name: str | None = None) -> LLMProvider:
+    name = (name or CONFIG.llm_provider).lower()
+    if name == "anthropic":
+        return AnthropicProvider()
+    if name == "ollama":
+        return OllamaProvider()
+    raise ValueError(f"Unknown LLM_PROVIDER: {name!r}")

src/rag.py

@@ -0,0 +1,54 @@
+"""Retrieval-Augmented Generation: retrieve -> ground -> answer with citations."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .config import CONFIG
+from .llm import LLMProvider, get_provider
+from .store import Retrieved, search
+
+SYSTEM_PROMPT = (
+    "You are a knowledge-base assistant for a YouTube channel. "
+    "Answer ONLY using the provided transcript excerpts. "
+    "Every factual sentence must cite its source with a bracket like [1], [2] "
+    "matching the numbered excerpts. "
+    "If the excerpts do not contain the answer, say you don't know based on this "
+    "channel's videos — do not use outside knowledge. Be concise and specific."
+)
+
+
+@dataclass
+class Answer:
+    text: str
+    sources: list[Retrieved]
+    provider: str
+
+
+def _build_context(hits: list[Retrieved]) -> str:
+    blocks = []
+    for i, h in enumerate(hits, 1):
+        blocks.append(f"[{i}] (from \"{h.title}\")\n{h.text}")
+    return "\n\n".join(blocks)
+
+
+def answer_question(question: str,
+                    provider: LLMProvider | None = None,
+                    top_k: int | None = None) -> Answer:
+    provider = provider or get_provider()
+    hits = search(question, top_k=top_k or CONFIG.top_k)
+
+    if not hits:
+        return Answer(
+            text="I couldn't find anything in this channel's videos about that.",
+            sources=[],
+            provider=provider.name,
+        )
+
+    context = _build_context(hits)
+    user = (
+        f"Transcript excerpts:\n\n{context}\n\n"
+        f"Question: {question}\n\n"
+        "Answer using only the excerpts above, with [n] citations."
+    )
+    text = provider.complete(SYSTEM_PROMPT, user)
+    return Answer(text=text, sources=hits, provider=provider.name)

src/store.py

@@ -0,0 +1,85 @@
+"""ChromaDB-backed vector store (persistent on disk)."""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+from .chunk import Chunk
+from .config import CONFIG
+from .embed import embed_query, embed_texts
+
+
+@dataclass
+class Retrieved:
+    text: str
+    title: str
+    url: str
+    start: float
+    score: float
+
+
+def _client():
+    import chromadb
+
+    os.makedirs(CONFIG.chroma_dir, exist_ok=True)
+    return chromadb.PersistentClient(path=CONFIG.chroma_dir)
+
+
+def _collection():
+    client = _client()
+    # We supply our own embeddings, so no embedding_function is needed.
+    return client.get_or_create_collection(
+        name=CONFIG.collection_name,
+        metadata={"hnsw:space": "cosine"},
+    )
+
+
+def index_chunks(chunks: list[Chunk], batch_size: int = 256) -> int:
+    """Embed and upsert chunks into the collection. Returns count indexed."""
+    col = _collection()
+    for i in range(0, len(chunks), batch_size):
+        batch = chunks[i : i + batch_size]
+        col.upsert(
+            ids=[c.id for c in batch],
+            documents=[c.text for c in batch],
+            embeddings=embed_texts([c.text for c in batch]),
+            metadatas=[
+                {"video_id": c.video_id, "title": c.title, "url": c.url, "start": c.start}
+                for c in batch
+            ],
+        )
+    return len(chunks)
+
+
+def count() -> int:
+    try:
+        return _collection().count()
+    except Exception:
+        return 0
+
+
+def search(query: str, top_k: int | None = None) -> list[Retrieved]:
+    top_k = top_k or CONFIG.top_k
+    col = _collection()
+    if col.count() == 0:
+        return []
+    res = col.query(
+        query_embeddings=[embed_query(query)],
+        n_results=top_k,
+        include=["documents", "metadatas", "distances"],
+    )
+    docs = res["documents"][0]
+    metas = res["metadatas"][0]
+    dists = res["distances"][0]
+    out: list[Retrieved] = []
+    for doc, meta, dist in zip(docs, metas, dists):
+        out.append(
+            Retrieved(
+                text=doc,
+                title=meta.get("title", ""),
+                url=meta.get("url", ""),
+                start=float(meta.get("start", 0.0)),
+                score=1.0 - float(dist),  # cosine distance -> similarity
+            )
+        )
+    return out