| """Converte un indice FAISS IVF (IndexIVFFlat) in ricerca esatta (IndexFlatIP). |
| |
| Motivazione: l'indice IVF con nprobe basso salta la maggior parte dei cluster e |
| manca i chunk rilevanti per le entita' meno comuni (recall crollata), causando |
| allucinazioni nel RAG. L'IVFFlat memorizza gia' tutti i vettori interi, quindi |
| la conversione in IndexFlatIP non costa RAM aggiuntiva e garantisce recall |
| perfetta con latenza comparabile (~27 ms/query su 3.2M vettori). |
| |
| Uso: |
| python scripts/convert_index_to_flat.py \ |
| --input rag/index/faiss/index.faiss \ |
| --output rag/index/faiss/index_flat.faiss |
| |
| Se --output coincide con --input, l'originale viene salvato in <input>.ivf.bak |
| prima della sovrascrittura. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import logging |
| import os |
| import shutil |
| import time |
|
|
| import faiss |
|
|
| logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") |
| log = logging.getLogger(__name__) |
|
|
|
|
| def parse_args() -> argparse.Namespace: |
| p = argparse.ArgumentParser(description="Convert FAISS IVF index to exact IndexFlatIP.") |
| p.add_argument("--input", default="rag/index/faiss/index.faiss", |
| help="Indice IVF di input.") |
| p.add_argument("--output", default="rag/index/faiss/index_flat.faiss", |
| help="Indice FlatIP di output.") |
| return p.parse_args() |
|
|
|
|
| def convert(input_path: str, output_path: str) -> None: |
| """Ricostruisce i vettori da un IVF e li inserisce in un IndexFlatIP. |
| |
| Parameters |
| ---------- |
| input_path : str |
| Percorso dell'indice IVF (IndexIVFFlat, METRIC_INNER_PRODUCT). |
| output_path : str |
| Percorso dove salvare l'indice FlatIP risultante. |
| |
| Raises |
| ------ |
| FileNotFoundError |
| Se input_path non esiste. |
| ValueError |
| Se l'indice non espone reconstruct (vettori non recuperabili). |
| """ |
| if not os.path.exists(input_path): |
| raise FileNotFoundError(f"{input_path} non trovato.") |
|
|
| log.info("Caricamento indice IVF: %s", input_path) |
| t = time.time() |
| index = faiss.read_index(input_path) |
| log.info(" caricato in %.1fs type=%s ntotal=%s d=%s", |
| time.time() - t, type(index).__name__, f"{index.ntotal:,}", index.d) |
|
|
| log.info("Ricostruzione vettori (make_direct_map + reconstruct_n)...") |
| t = time.time() |
| index.make_direct_map() |
| vectors = index.reconstruct_n(0, index.ntotal) |
| log.info(" ricostruiti %s vettori in %.1fs shape=%s", |
| f"{index.ntotal:,}", time.time() - t, vectors.shape) |
|
|
| log.info("Costruzione IndexFlatIP (ricerca esatta, inner product)...") |
| t = time.time() |
| flat = faiss.IndexFlatIP(index.d) |
| flat.add(vectors) |
| log.info(" aggiunti %s vettori in %.1fs", f"{flat.ntotal:,}", time.time() - t) |
|
|
| if os.path.abspath(output_path) == os.path.abspath(input_path): |
| backup = input_path + ".ivf.bak" |
| log.info("Output == input: backup dell'IVF in %s", backup) |
| shutil.copy2(input_path, backup) |
|
|
| log.info("Scrittura indice FlatIP: %s", output_path) |
| faiss.write_index(flat, output_path) |
| size_gb = os.path.getsize(output_path) / 1e9 |
| log.info("Fatto: %s (%.2f GB)", output_path, size_gb) |
|
|
|
|
| def main() -> None: |
| args = parse_args() |
| convert(args.input, args.output) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
