feat(scripts): bootstrap_corpus.py — idempotent Qdrant ingestion (Phase 2.3)

CLI that walks --pdf-dir, ingests each PDF via src.ingestion.pipeline.
ingest_paper, and writes chunks + bge-m3 embeddings to a configured Qdrant
collection. Skips re-ingestion if the collection already has points
(override with --force) so re-running is a safe noop.

Designed as the foundation for Phase 4's Container Apps init job; runs
locally against `docker compose up qdrant ollama` for dev parity. There's
no /ingest HTTP endpoint yet — deferred until live paper upload becomes
a demo requirement.

QdrantVectorStore gains a `count()` method that returns 0 for missing
collections (instead of raising) so the idempotency check is one call.
3 unit tests cover empty, missing, and populated paths against the
in-memory Qdrant client.

Caveats documented in the script docstring:
 - BM25 + chunks_by_id are in-process state and don't survive this
   script's process exit. The FastAPI app would need a separate
   persistence path to use the populated Qdrant collection
   end-to-end. Phase 2.x follow-up; eval scripts rebuild fresh per run.

Author: NorthernLightx

scripts/bootstrap_corpus.py

@@ -0,0 +1,137 @@
+"""Bootstrap the corpus into a configured Qdrant (Phase 2.3).
+
+Runs once per deploy / fresh local setup: walks --pdf-dir, ingests each PDF
+through src.ingestion.pipeline.ingest_paper, and writes the resulting chunks
++ embeddings to the named Qdrant collection. Idempotent — refuses to re-ingest
+if the collection already has points (override with --force).
+
+Designed to be the foundation for Phase 4's Container Apps init job; the same
+script runs locally against `docker compose up qdrant ollama` for development
+parity. There's no /ingest HTTP endpoint yet — that's deferred until live
+paper upload becomes a demo requirement.
+
+Caveats (recorded so future-me doesn't have to re-derive them):
+
+  * BM25 lives in process memory (`src/rag/bm25.py`) — this script populates
+    Qdrant but its BM25 dies with the process. The FastAPI app would need to
+    rebuild BM25 from scratch at startup (or persist it separately). That's
+    a Phase 2.x follow-up; for now eval scripts rebuild BM25 per run.
+  * Same for `chunks_by_id` — the PipelineRetriever needs a chunk dict to
+    resolve text after Qdrant returns chunk-ids. This script doesn't
+    persist that either; eval / retrieval workflows materialize chunks
+    fresh per run.
+
+Usage:
+
+    .venv/Scripts/python.exe -m scripts.bootstrap_corpus \\
+        --pdf-dir data/papers \\
+        --qdrant http://localhost:6333 \\
+        --ollama http://localhost:11434 \\
+        --collection rag_corpus
+
+Pass `--force` to re-ingest into a non-empty collection (drops + recreates).
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+from pathlib import Path
+
+from src.embeddings.ollama_bge import OllamaBgeEmbedder
+from src.ingestion.pipeline import ingest_paper
+from src.observability.logging import configure_logging, get_logger
+from src.rag.bm25 import Bm25Index
+from src.rag.vectorstore import QdrantVectorStore
+from src.types import Paper
+
+
+async def _main(
+    *,
+    pdf_dir: Path,
+    qdrant_url: str,
+    ollama_url: str,
+    collection: str,
+    force: bool,
+) -> None:
+    log = get_logger("scripts.bootstrap_corpus")
+    pdf_paths = sorted(pdf_dir.glob("*.pdf"))
+    if not pdf_paths:
+        raise SystemExit(f"No .pdf files found in {pdf_dir}")
+    print(f"Found {len(pdf_paths)} PDFs in {pdf_dir}")
+
+    embedder = OllamaBgeEmbedder(base_url=ollama_url)
+    vectorstore = QdrantVectorStore(url=qdrant_url, collection_name=collection, dim=embedder.dim)
+
+    existing = await vectorstore.count()
+    if existing > 0 and not force:
+        print(
+            f"Collection {collection!r} already has {existing} points — "
+            f"skipping ingestion (pass --force to re-ingest)."
+        )
+        log.info("bootstrap.skip", collection=collection, existing_points=existing, force=force)
+        return
+
+    if existing > 0 and force:
+        print(f"--force: dropping existing {existing} points in {collection!r}")
+
+    await vectorstore.ensure_collection()
+    bm25 = Bm25Index()  # in-process, throwaway — see module docstring caveats
+
+    total_chunks = 0
+    for pdf_path in pdf_paths:
+        paper = Paper(paper_id=pdf_path.stem, title=pdf_path.stem, pdf_path=pdf_path)
+        ingested = await ingest_paper(
+            paper=paper,
+            embedder=embedder,
+            vectorstore=vectorstore,
+            bm25=bm25,
+            contextualizer_llm=None,
+            contextualizer_model=None,
+            contextualizer_concurrency=4,
+            extract_figures_enabled=False,
+            extract_tables_enabled=False,
+            vlm_captioner=None,
+        )
+        total_chunks += ingested.chunk_count
+        print(f"  {pdf_path.name}: {ingested.chunk_count} chunks")
+
+    final = await vectorstore.count()
+    print(
+        f"\nIngested {total_chunks} chunks across {len(pdf_paths)} papers into "
+        f"{collection!r} (collection now contains {final} points)"
+    )
+    log.info(
+        "bootstrap.done",
+        collection=collection,
+        papers=len(pdf_paths),
+        chunks_ingested=total_chunks,
+        collection_size=final,
+    )
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Idempotently ingest every PDF in --pdf-dir into a Qdrant collection."
+    )
+    parser.add_argument("--pdf-dir", type=Path, default=Path("data/papers"))
+    parser.add_argument("--qdrant", default="http://localhost:6333")
+    parser.add_argument("--ollama", default="http://localhost:11434")
+    parser.add_argument("--collection", default="rag_corpus")
+    parser.add_argument(
+        "--force",
+        action="store_true",
+        help="Re-ingest even if the collection already contains points.",
+    )
+    args = parser.parse_args()
+
+    configure_logging(level="INFO", env="local", log_file=None)
+    asyncio.run(
+        _main(
+            pdf_dir=args.pdf_dir,
+            qdrant_url=args.qdrant,
+            ollama_url=args.ollama,
+            collection=args.collection,
+            force=args.force,
+        )
+    )

src/rag/vectorstore.py

@@ -64,6 +64,18 @@ async def upsert_chunks(self, chunks: list[Chunk], vectors: list[list[float]]) -
         ]
         await self._client.upsert(collection_name=self._collection, points=points)
 
+    async def count(self) -> int:
+        """Return the number of points in the collection; 0 if it doesn't exist.
+
+        Used by `scripts/bootstrap_corpus.py` for idempotent re-runs — a populated
+        collection means ingestion already happened, skip unless --force.
+        """
+        existing = await self._client.get_collections()
+        if not any(c.name == self._collection for c in existing.collections):
+            return 0
+        result = await self._client.count(collection_name=self._collection, exact=True)
+        return int(result.count)
+
     async def search(self, vector: list[float], top_k: int) -> list[VectorMatch]:
         response = await self._client.query_points(
             collection_name=self._collection,

tests/unit/test_vectorstore_count.py

@@ -0,0 +1,41 @@
+"""QdrantVectorStore.count — used by scripts/bootstrap_corpus.py for idempotency.
+
+Uses the in-memory Qdrant client (`url=':memory:'`) so the test runs without
+docker. Exercises the empty-collection, missing-collection, and populated
+paths.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from src.rag.vectorstore import QdrantVectorStore
+from src.types import Chunk
+
+
+@pytest.mark.asyncio
+async def test_count_returns_zero_when_collection_missing() -> None:
+    """Newly-constructed store has no collection yet — count is 0, not an error."""
+    store = QdrantVectorStore(url=":memory:", collection_name="not_yet_created", dim=4)
+    assert await store.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_count_returns_zero_when_collection_empty() -> None:
+    """Created but empty collection counts as 0."""
+    store = QdrantVectorStore(url=":memory:", collection_name="empty", dim=4)
+    await store.ensure_collection()
+    assert await store.count() == 0
+
+
+@pytest.mark.asyncio
+async def test_count_reflects_upserted_chunks() -> None:
+    store = QdrantVectorStore(url=":memory:", collection_name="populated", dim=4)
+    await store.ensure_collection()
+    chunks = [
+        Chunk(chunk_id=f"p1::p1::c{i}", paper_id="p1", page_numbers=[1], text=f"chunk {i}")
+        for i in range(3)
+    ]
+    vectors = [[0.1, 0.2, 0.3, 0.4], [0.5, 0.6, 0.7, 0.8], [0.9, 0.0, 0.1, 0.2]]
+    await store.upsert_chunks(chunks, vectors)
+    assert await store.count() == 3