"""Regenerate chunks.jsonl metadata without re-embedding (CPU-only, offline). The dual-source retrieval work enriches each ``Chunk`` with structured ``metadata`` (IDA: sector/category/experience/measures; directive: article title). That metadata is **not embedded**, so the vectors and chunk order are unchanged — meaning the committed ``index.faiss`` stays byte-identical and we only need to rewrite ``chunks.jsonl``. This script loads the existing index + old chunks, rebuilds the corpus with the new metadata, asserts the rebuilt chunks line up 1:1 with the live vectors (same ids, same order, same count), then re-persists. No GPU, no embedding — runs in well under a second. Usage:: uv run python scripts/refresh_metadata.py """ from __future__ import annotations import sys from pathlib import Path # Allow running as a plain script (`python scripts/refresh_metadata.py`). sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from src.backend.indexing.ingest import ( # noqa: E402 build_corpus, load_directive_chunks, load_ida_chunks, ) from src.backend.indexing.store import DEFAULT_INDEX_DIR, VectorStore # noqa: E402 def main() -> None: store = VectorStore() if not store.load(): raise SystemExit( f"No index at {DEFAULT_INDEX_DIR}. Build it first (scripts/build_index.py " "or scripts/build_index_modal.py) before refreshing metadata." ) old_ids = [c.id for c in store._chunks] ntotal = store._index.ntotal n_dir = len(load_directive_chunks()) n_ida = len(load_ida_chunks()) new_chunks = build_corpus() new_ids = [c.id for c in new_chunks] # Alignment guard: the rebuilt chunks must line up exactly with the stored # vectors, else chunks.jsonl would point at the wrong embeddings. if len(new_chunks) != ntotal: raise SystemExit( f"Count mismatch: rebuilt {len(new_chunks)} chunks but index has " f"{ntotal} vectors. Re-embed (build_index) instead of refreshing." ) if new_ids != old_ids: raise SystemExit( "Order/id mismatch between rebuilt chunks and the existing index. " "The corpus changed shape — re-embed (build_index) instead." ) store._chunks = new_chunks store.persist() # rewrites chunks.jsonl (+ identical index.faiss) n_meta = sum(1 for c in new_chunks if c.metadata) print( f"refreshed: directive={n_dir} ida={n_ida} total={len(new_chunks)} " f"ntotal={ntotal} with_metadata={n_meta} → {DEFAULT_INDEX_DIR}" ) if __name__ == "__main__": main()