Introducing dynamic index update (#108)
* feat: Add GitHub PR and issue templates for better contributor experience * simplify: Make templates more concise and user-friendly * fix: enable is_compact=False, is_recompute=True * feat: update when recompute * test * fix: real recompute * refactor * fix: compare with no-recompute * fix: test
This commit is contained in:
0
examples/__init__.py
Normal file
0
examples/__init__.py
Normal file
404
examples/dynamic_update_no_recompute.py
Normal file
404
examples/dynamic_update_no_recompute.py
Normal file
@@ -0,0 +1,404 @@
|
|||||||
|
"""Dynamic HNSW update demo without compact storage.
|
||||||
|
|
||||||
|
This script reproduces the minimal scenario we used while debugging on-the-fly
|
||||||
|
recompute:
|
||||||
|
|
||||||
|
1. Build a non-compact HNSW index from the first few paragraphs of a text file.
|
||||||
|
2. Print the top results with `recompute_embeddings=True`.
|
||||||
|
3. Append additional paragraphs with :meth:`LeannBuilder.update_index`.
|
||||||
|
4. Run the same query again to show the newly inserted passages.
|
||||||
|
|
||||||
|
Run it with ``uv`` (optionally pointing LEANN_HNSW_LOG_PATH at a file to inspect
|
||||||
|
ZMQ activity)::
|
||||||
|
|
||||||
|
LEANN_HNSW_LOG_PATH=embedding_fetch.log \
|
||||||
|
uv run -m examples.dynamic_update_no_recompute \
|
||||||
|
--index-path .leann/examples/leann-demo.leann
|
||||||
|
|
||||||
|
By default the script builds an index from ``data/2501.14312v1 (1).pdf`` and
|
||||||
|
then updates it with LEANN-related material from ``data/2506.08276v1.pdf``.
|
||||||
|
It issues the query "What's LEANN?" before and after the update to show how the
|
||||||
|
new passages become immediately searchable. The script uses the
|
||||||
|
``sentence-transformers/all-MiniLM-L6-v2`` model with ``is_recompute=True`` so
|
||||||
|
Faiss pulls existing vectors on demand via the ZMQ embedding server, while
|
||||||
|
freshly added passages are embedded locally just like the initial build.
|
||||||
|
|
||||||
|
To make storage comparisons easy, the script can also build a matching
|
||||||
|
``is_recompute=False`` baseline (enabled by default) and report the index size
|
||||||
|
delta after the update. Disable the baseline run with
|
||||||
|
``--skip-compare-no-recompute`` if you only need the recompute flow.
|
||||||
|
"""
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import json
|
||||||
|
from collections.abc import Iterable
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
from leann.api import LeannBuilder, LeannSearcher
|
||||||
|
from leann.registry import register_project_directory
|
||||||
|
|
||||||
|
from apps.chunking import create_text_chunks
|
||||||
|
|
||||||
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||||
|
|
||||||
|
DEFAULT_QUERY = "What's LEANN?"
|
||||||
|
DEFAULT_INITIAL_FILES = [REPO_ROOT / "data" / "2501.14312v1 (1).pdf"]
|
||||||
|
DEFAULT_UPDATE_FILES = [REPO_ROOT / "data" / "2506.08276v1.pdf"]
|
||||||
|
|
||||||
|
|
||||||
|
def load_chunks_from_files(paths: list[Path]) -> list[str]:
|
||||||
|
from llama_index.core import SimpleDirectoryReader
|
||||||
|
|
||||||
|
documents = []
|
||||||
|
for path in paths:
|
||||||
|
p = path.expanduser().resolve()
|
||||||
|
if not p.exists():
|
||||||
|
raise FileNotFoundError(f"Input path not found: {p}")
|
||||||
|
if p.is_dir():
|
||||||
|
reader = SimpleDirectoryReader(str(p), recursive=False)
|
||||||
|
documents.extend(reader.load_data(show_progress=True))
|
||||||
|
else:
|
||||||
|
reader = SimpleDirectoryReader(input_files=[str(p)])
|
||||||
|
documents.extend(reader.load_data(show_progress=True))
|
||||||
|
|
||||||
|
if not documents:
|
||||||
|
return []
|
||||||
|
|
||||||
|
chunks = create_text_chunks(
|
||||||
|
documents,
|
||||||
|
chunk_size=512,
|
||||||
|
chunk_overlap=128,
|
||||||
|
use_ast_chunking=False,
|
||||||
|
)
|
||||||
|
return [c for c in chunks if isinstance(c, str) and c.strip()]
|
||||||
|
|
||||||
|
|
||||||
|
def run_search(index_path: Path, query: str, top_k: int, *, recompute_embeddings: bool) -> list:
|
||||||
|
searcher = LeannSearcher(str(index_path))
|
||||||
|
try:
|
||||||
|
return searcher.search(
|
||||||
|
query=query,
|
||||||
|
top_k=top_k,
|
||||||
|
recompute_embeddings=recompute_embeddings,
|
||||||
|
batch_size=16,
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
searcher.cleanup()
|
||||||
|
|
||||||
|
|
||||||
|
def print_results(title: str, results: Iterable) -> None:
|
||||||
|
print(f"\n=== {title} ===")
|
||||||
|
res_list = list(results)
|
||||||
|
print(f"results count: {len(res_list)}")
|
||||||
|
print("passages:")
|
||||||
|
if not res_list:
|
||||||
|
print(" (no passages returned)")
|
||||||
|
for res in res_list:
|
||||||
|
snippet = res.text.replace("\n", " ")[:120]
|
||||||
|
print(f" - {res.id}: {snippet}... (score={res.score:.4f})")
|
||||||
|
|
||||||
|
|
||||||
|
def build_initial_index(
|
||||||
|
index_path: Path,
|
||||||
|
paragraphs: list[str],
|
||||||
|
model_name: str,
|
||||||
|
embedding_mode: str,
|
||||||
|
is_recompute: bool,
|
||||||
|
) -> None:
|
||||||
|
builder = LeannBuilder(
|
||||||
|
backend_name="hnsw",
|
||||||
|
embedding_model=model_name,
|
||||||
|
embedding_mode=embedding_mode,
|
||||||
|
is_compact=False,
|
||||||
|
is_recompute=is_recompute,
|
||||||
|
)
|
||||||
|
for idx, passage in enumerate(paragraphs):
|
||||||
|
builder.add_text(passage, metadata={"id": str(idx)})
|
||||||
|
builder.build_index(str(index_path))
|
||||||
|
|
||||||
|
|
||||||
|
def update_index(
|
||||||
|
index_path: Path,
|
||||||
|
start_id: int,
|
||||||
|
paragraphs: list[str],
|
||||||
|
model_name: str,
|
||||||
|
embedding_mode: str,
|
||||||
|
is_recompute: bool,
|
||||||
|
) -> None:
|
||||||
|
updater = LeannBuilder(
|
||||||
|
backend_name="hnsw",
|
||||||
|
embedding_model=model_name,
|
||||||
|
embedding_mode=embedding_mode,
|
||||||
|
is_compact=False,
|
||||||
|
is_recompute=is_recompute,
|
||||||
|
)
|
||||||
|
for offset, passage in enumerate(paragraphs, start=start_id):
|
||||||
|
updater.add_text(passage, metadata={"id": str(offset)})
|
||||||
|
updater.update_index(str(index_path))
|
||||||
|
|
||||||
|
|
||||||
|
def ensure_index_dir(index_path: Path) -> None:
|
||||||
|
index_path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
|
||||||
|
|
||||||
|
def cleanup_index_files(index_path: Path) -> None:
|
||||||
|
"""Remove leftover index artifacts for a clean rebuild."""
|
||||||
|
|
||||||
|
parent = index_path.parent
|
||||||
|
if not parent.exists():
|
||||||
|
return
|
||||||
|
stem = index_path.stem
|
||||||
|
for file in parent.glob(f"{stem}*"):
|
||||||
|
if file.is_file():
|
||||||
|
file.unlink()
|
||||||
|
|
||||||
|
|
||||||
|
def index_file_size(index_path: Path) -> int:
|
||||||
|
"""Return the size of the primary .index file for the given index path."""
|
||||||
|
|
||||||
|
index_file = index_path.parent / f"{index_path.stem}.index"
|
||||||
|
return index_file.stat().st_size if index_file.exists() else 0
|
||||||
|
|
||||||
|
|
||||||
|
def load_metadata_snapshot(index_path: Path) -> dict[str, Any] | None:
|
||||||
|
meta_path = index_path.parent / f"{index_path.name}.meta.json"
|
||||||
|
if not meta_path.exists():
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
return json.loads(meta_path.read_text())
|
||||||
|
except json.JSONDecodeError:
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def run_workflow(
|
||||||
|
*,
|
||||||
|
label: str,
|
||||||
|
index_path: Path,
|
||||||
|
initial_paragraphs: list[str],
|
||||||
|
update_paragraphs: list[str],
|
||||||
|
model_name: str,
|
||||||
|
embedding_mode: str,
|
||||||
|
is_recompute: bool,
|
||||||
|
query: str,
|
||||||
|
top_k: int,
|
||||||
|
) -> dict[str, Any]:
|
||||||
|
prefix = f"[{label}] " if label else ""
|
||||||
|
|
||||||
|
ensure_index_dir(index_path)
|
||||||
|
cleanup_index_files(index_path)
|
||||||
|
|
||||||
|
print(f"{prefix}Building initial index...")
|
||||||
|
build_initial_index(
|
||||||
|
index_path,
|
||||||
|
initial_paragraphs,
|
||||||
|
model_name,
|
||||||
|
embedding_mode,
|
||||||
|
is_recompute=is_recompute,
|
||||||
|
)
|
||||||
|
|
||||||
|
initial_size = index_file_size(index_path)
|
||||||
|
before_results = run_search(
|
||||||
|
index_path,
|
||||||
|
query,
|
||||||
|
top_k,
|
||||||
|
recompute_embeddings=is_recompute,
|
||||||
|
)
|
||||||
|
|
||||||
|
print(f"\n{prefix}Updating index with additional passages...")
|
||||||
|
update_index(
|
||||||
|
index_path,
|
||||||
|
start_id=len(initial_paragraphs),
|
||||||
|
paragraphs=update_paragraphs,
|
||||||
|
model_name=model_name,
|
||||||
|
embedding_mode=embedding_mode,
|
||||||
|
is_recompute=is_recompute,
|
||||||
|
)
|
||||||
|
|
||||||
|
after_results = run_search(
|
||||||
|
index_path,
|
||||||
|
query,
|
||||||
|
top_k,
|
||||||
|
recompute_embeddings=is_recompute,
|
||||||
|
)
|
||||||
|
updated_size = index_file_size(index_path)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"initial_size": initial_size,
|
||||||
|
"updated_size": updated_size,
|
||||||
|
"delta": updated_size - initial_size,
|
||||||
|
"before_results": before_results,
|
||||||
|
"after_results": after_results,
|
||||||
|
"metadata": load_metadata_snapshot(index_path),
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
parser = argparse.ArgumentParser(description=__doc__)
|
||||||
|
parser.add_argument(
|
||||||
|
"--initial-files",
|
||||||
|
type=Path,
|
||||||
|
nargs="+",
|
||||||
|
default=DEFAULT_INITIAL_FILES,
|
||||||
|
help="Initial document files (PDF/TXT) used to build the base index",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--index-path",
|
||||||
|
type=Path,
|
||||||
|
default=Path(".leann/examples/leann-demo.leann"),
|
||||||
|
help="Destination index path (default: .leann/examples/leann-demo.leann)",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--initial-count",
|
||||||
|
type=int,
|
||||||
|
default=8,
|
||||||
|
help="Number of chunks to use from the initial documents (default: 8)",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--update-files",
|
||||||
|
type=Path,
|
||||||
|
nargs="*",
|
||||||
|
default=DEFAULT_UPDATE_FILES,
|
||||||
|
help="Additional documents to add during update (PDF/TXT)",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--update-count",
|
||||||
|
type=int,
|
||||||
|
default=4,
|
||||||
|
help="Number of chunks to append from update documents (default: 4)",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--update-text",
|
||||||
|
type=str,
|
||||||
|
default=(
|
||||||
|
"LEANN (Lightweight Embedding ANN) is an indexing toolkit focused on "
|
||||||
|
"recompute-aware HNSW graphs, allowing embeddings to be regenerated "
|
||||||
|
"on demand to keep disk usage minimal."
|
||||||
|
),
|
||||||
|
help="Fallback text to append if --update-files is omitted",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--top-k",
|
||||||
|
type=int,
|
||||||
|
default=4,
|
||||||
|
help="Number of results to show for each search (default: 4)",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--query",
|
||||||
|
type=str,
|
||||||
|
default=DEFAULT_QUERY,
|
||||||
|
help="Query to run before/after the update",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--embedding-model",
|
||||||
|
type=str,
|
||||||
|
default="sentence-transformers/all-MiniLM-L6-v2",
|
||||||
|
help="Embedding model name",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--embedding-mode",
|
||||||
|
type=str,
|
||||||
|
default="sentence-transformers",
|
||||||
|
choices=["sentence-transformers", "openai", "mlx", "ollama"],
|
||||||
|
help="Embedding backend mode",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--compare-no-recompute",
|
||||||
|
dest="compare_no_recompute",
|
||||||
|
action="store_true",
|
||||||
|
help="Also run a baseline with is_recompute=False and report its index growth.",
|
||||||
|
)
|
||||||
|
parser.add_argument(
|
||||||
|
"--skip-compare-no-recompute",
|
||||||
|
dest="compare_no_recompute",
|
||||||
|
action="store_false",
|
||||||
|
help="Skip building the no-recompute baseline.",
|
||||||
|
)
|
||||||
|
parser.set_defaults(compare_no_recompute=True)
|
||||||
|
args = parser.parse_args()
|
||||||
|
|
||||||
|
ensure_index_dir(args.index_path)
|
||||||
|
register_project_directory(REPO_ROOT)
|
||||||
|
|
||||||
|
initial_chunks = load_chunks_from_files(list(args.initial_files))
|
||||||
|
if not initial_chunks:
|
||||||
|
raise ValueError("No text chunks extracted from the initial files.")
|
||||||
|
|
||||||
|
initial = initial_chunks[: args.initial_count]
|
||||||
|
if not initial:
|
||||||
|
raise ValueError("Initial chunk set is empty after applying --initial-count.")
|
||||||
|
|
||||||
|
if args.update_files:
|
||||||
|
update_chunks = load_chunks_from_files(list(args.update_files))
|
||||||
|
if not update_chunks:
|
||||||
|
raise ValueError("No text chunks extracted from the update files.")
|
||||||
|
to_add = update_chunks[: args.update_count]
|
||||||
|
else:
|
||||||
|
if not args.update_text:
|
||||||
|
raise ValueError("Provide --update-files or --update-text for the update step.")
|
||||||
|
to_add = [args.update_text]
|
||||||
|
if not to_add:
|
||||||
|
raise ValueError("Update chunk set is empty after applying --update-count.")
|
||||||
|
|
||||||
|
recompute_stats = run_workflow(
|
||||||
|
label="recompute",
|
||||||
|
index_path=args.index_path,
|
||||||
|
initial_paragraphs=initial,
|
||||||
|
update_paragraphs=to_add,
|
||||||
|
model_name=args.embedding_model,
|
||||||
|
embedding_mode=args.embedding_mode,
|
||||||
|
is_recompute=True,
|
||||||
|
query=args.query,
|
||||||
|
top_k=args.top_k,
|
||||||
|
)
|
||||||
|
|
||||||
|
print_results("initial search", recompute_stats["before_results"])
|
||||||
|
print_results("after update", recompute_stats["after_results"])
|
||||||
|
print(
|
||||||
|
f"\n[recompute] Index file size change: {recompute_stats['initial_size']} -> {recompute_stats['updated_size']} bytes"
|
||||||
|
f" (Δ {recompute_stats['delta']})"
|
||||||
|
)
|
||||||
|
|
||||||
|
if recompute_stats["metadata"]:
|
||||||
|
meta_view = {k: recompute_stats["metadata"].get(k) for k in ("is_compact", "is_pruned")}
|
||||||
|
print("[recompute] metadata snapshot:")
|
||||||
|
print(json.dumps(meta_view, indent=2))
|
||||||
|
|
||||||
|
if args.compare_no_recompute:
|
||||||
|
baseline_path = (
|
||||||
|
args.index_path.parent / f"{args.index_path.stem}-norecompute{args.index_path.suffix}"
|
||||||
|
)
|
||||||
|
baseline_stats = run_workflow(
|
||||||
|
label="no-recompute",
|
||||||
|
index_path=baseline_path,
|
||||||
|
initial_paragraphs=initial,
|
||||||
|
update_paragraphs=to_add,
|
||||||
|
model_name=args.embedding_model,
|
||||||
|
embedding_mode=args.embedding_mode,
|
||||||
|
is_recompute=False,
|
||||||
|
query=args.query,
|
||||||
|
top_k=args.top_k,
|
||||||
|
)
|
||||||
|
|
||||||
|
print(
|
||||||
|
f"\n[no-recompute] Index file size change: {baseline_stats['initial_size']} -> {baseline_stats['updated_size']} bytes"
|
||||||
|
f" (Δ {baseline_stats['delta']})"
|
||||||
|
)
|
||||||
|
|
||||||
|
after_texts = [res.text for res in recompute_stats["after_results"]]
|
||||||
|
baseline_after_texts = [res.text for res in baseline_stats["after_results"]]
|
||||||
|
if after_texts == baseline_after_texts:
|
||||||
|
print(
|
||||||
|
"[no-recompute] Search results match recompute baseline; see above for the shared output."
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
print("[no-recompute] WARNING: search results differ from recompute baseline.")
|
||||||
|
|
||||||
|
if baseline_stats["metadata"]:
|
||||||
|
meta_view = {k: baseline_stats["metadata"].get(k) for k in ("is_compact", "is_pruned")}
|
||||||
|
print("[no-recompute] metadata snapshot:")
|
||||||
|
print(json.dumps(meta_view, indent=2))
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
@@ -5,6 +5,8 @@ import os
|
|||||||
import struct
|
import struct
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
|
from dataclasses import dataclass
|
||||||
|
from typing import Any, Optional
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
@@ -237,6 +239,288 @@ def write_compact_format(
|
|||||||
f_out.write(storage_data)
|
f_out.write(storage_data)
|
||||||
|
|
||||||
|
|
||||||
|
@dataclass
|
||||||
|
class HNSWComponents:
|
||||||
|
original_hnsw_data: dict[str, Any]
|
||||||
|
assign_probas_np: np.ndarray
|
||||||
|
cum_nneighbor_per_level_np: np.ndarray
|
||||||
|
levels_np: np.ndarray
|
||||||
|
is_compact: bool
|
||||||
|
compact_level_ptr: Optional[np.ndarray] = None
|
||||||
|
compact_node_offsets_np: Optional[np.ndarray] = None
|
||||||
|
compact_neighbors_data: Optional[list[int]] = None
|
||||||
|
offsets_np: Optional[np.ndarray] = None
|
||||||
|
neighbors_np: Optional[np.ndarray] = None
|
||||||
|
storage_fourcc: int = NULL_INDEX_FOURCC
|
||||||
|
storage_data: bytes = b""
|
||||||
|
|
||||||
|
|
||||||
|
def _read_hnsw_structure(f) -> HNSWComponents:
|
||||||
|
original_hnsw_data: dict[str, Any] = {}
|
||||||
|
|
||||||
|
hnsw_index_fourcc = read_struct(f, "<I")
|
||||||
|
if hnsw_index_fourcc not in EXPECTED_HNSW_FOURCCS:
|
||||||
|
raise ValueError(
|
||||||
|
f"Unexpected HNSW FourCC: {hnsw_index_fourcc:08x}. Expected one of {EXPECTED_HNSW_FOURCCS}."
|
||||||
|
)
|
||||||
|
|
||||||
|
original_hnsw_data["index_fourcc"] = hnsw_index_fourcc
|
||||||
|
original_hnsw_data["d"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["ntotal"] = read_struct(f, "<q")
|
||||||
|
original_hnsw_data["dummy1"] = read_struct(f, "<q")
|
||||||
|
original_hnsw_data["dummy2"] = read_struct(f, "<q")
|
||||||
|
original_hnsw_data["is_trained"] = read_struct(f, "?")
|
||||||
|
original_hnsw_data["metric_type"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["metric_arg"] = 0.0
|
||||||
|
if original_hnsw_data["metric_type"] > 1:
|
||||||
|
original_hnsw_data["metric_arg"] = read_struct(f, "<f")
|
||||||
|
|
||||||
|
assign_probas_np = read_numpy_vector(f, np.float64, "d")
|
||||||
|
cum_nneighbor_per_level_np = read_numpy_vector(f, np.int32, "i")
|
||||||
|
levels_np = read_numpy_vector(f, np.int32, "i")
|
||||||
|
|
||||||
|
ntotal = len(levels_np)
|
||||||
|
if ntotal != original_hnsw_data["ntotal"]:
|
||||||
|
original_hnsw_data["ntotal"] = ntotal
|
||||||
|
|
||||||
|
pos_before_compact = f.tell()
|
||||||
|
is_compact_flag = None
|
||||||
|
try:
|
||||||
|
is_compact_flag = read_struct(f, "<?")
|
||||||
|
except EOFError:
|
||||||
|
is_compact_flag = None
|
||||||
|
|
||||||
|
if is_compact_flag:
|
||||||
|
compact_level_ptr = read_numpy_vector(f, np.uint64, "Q")
|
||||||
|
compact_node_offsets_np = read_numpy_vector(f, np.uint64, "Q")
|
||||||
|
|
||||||
|
original_hnsw_data["entry_point"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["max_level"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["efConstruction"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["efSearch"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["dummy_upper_beam"] = read_struct(f, "<i")
|
||||||
|
|
||||||
|
storage_fourcc = read_struct(f, "<I")
|
||||||
|
compact_neighbors_data_np = read_numpy_vector(f, np.int32, "i")
|
||||||
|
compact_neighbors_data = compact_neighbors_data_np.tolist()
|
||||||
|
storage_data = f.read()
|
||||||
|
|
||||||
|
return HNSWComponents(
|
||||||
|
original_hnsw_data=original_hnsw_data,
|
||||||
|
assign_probas_np=assign_probas_np,
|
||||||
|
cum_nneighbor_per_level_np=cum_nneighbor_per_level_np,
|
||||||
|
levels_np=levels_np,
|
||||||
|
is_compact=True,
|
||||||
|
compact_level_ptr=compact_level_ptr,
|
||||||
|
compact_node_offsets_np=compact_node_offsets_np,
|
||||||
|
compact_neighbors_data=compact_neighbors_data,
|
||||||
|
storage_fourcc=storage_fourcc,
|
||||||
|
storage_data=storage_data,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Non-compact case
|
||||||
|
f.seek(pos_before_compact)
|
||||||
|
|
||||||
|
pos_before_probe = f.tell()
|
||||||
|
try:
|
||||||
|
suspected_flag = read_struct(f, "<B")
|
||||||
|
if suspected_flag != 0x00:
|
||||||
|
f.seek(pos_before_probe)
|
||||||
|
except EOFError:
|
||||||
|
f.seek(pos_before_probe)
|
||||||
|
|
||||||
|
offsets_np = read_numpy_vector(f, np.uint64, "Q")
|
||||||
|
neighbors_np = read_numpy_vector(f, np.int32, "i")
|
||||||
|
|
||||||
|
original_hnsw_data["entry_point"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["max_level"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["efConstruction"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["efSearch"] = read_struct(f, "<i")
|
||||||
|
original_hnsw_data["dummy_upper_beam"] = read_struct(f, "<i")
|
||||||
|
|
||||||
|
storage_fourcc = NULL_INDEX_FOURCC
|
||||||
|
storage_data = b""
|
||||||
|
try:
|
||||||
|
storage_fourcc = read_struct(f, "<I")
|
||||||
|
storage_data = f.read()
|
||||||
|
except EOFError:
|
||||||
|
storage_fourcc = NULL_INDEX_FOURCC
|
||||||
|
|
||||||
|
return HNSWComponents(
|
||||||
|
original_hnsw_data=original_hnsw_data,
|
||||||
|
assign_probas_np=assign_probas_np,
|
||||||
|
cum_nneighbor_per_level_np=cum_nneighbor_per_level_np,
|
||||||
|
levels_np=levels_np,
|
||||||
|
is_compact=False,
|
||||||
|
offsets_np=offsets_np,
|
||||||
|
neighbors_np=neighbors_np,
|
||||||
|
storage_fourcc=storage_fourcc,
|
||||||
|
storage_data=storage_data,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _read_hnsw_structure_from_file(path: str) -> HNSWComponents:
|
||||||
|
with open(path, "rb") as f:
|
||||||
|
return _read_hnsw_structure(f)
|
||||||
|
|
||||||
|
|
||||||
|
def write_original_format(
|
||||||
|
f_out,
|
||||||
|
original_hnsw_data,
|
||||||
|
assign_probas_np,
|
||||||
|
cum_nneighbor_per_level_np,
|
||||||
|
levels_np,
|
||||||
|
offsets_np,
|
||||||
|
neighbors_np,
|
||||||
|
storage_fourcc,
|
||||||
|
storage_data,
|
||||||
|
):
|
||||||
|
"""Write non-compact HNSW data in original FAISS order."""
|
||||||
|
|
||||||
|
f_out.write(struct.pack("<I", original_hnsw_data["index_fourcc"]))
|
||||||
|
f_out.write(struct.pack("<i", original_hnsw_data["d"]))
|
||||||
|
f_out.write(struct.pack("<q", original_hnsw_data["ntotal"]))
|
||||||
|
f_out.write(struct.pack("<q", original_hnsw_data["dummy1"]))
|
||||||
|
f_out.write(struct.pack("<q", original_hnsw_data["dummy2"]))
|
||||||
|
f_out.write(struct.pack("<?", original_hnsw_data["is_trained"]))
|
||||||
|
f_out.write(struct.pack("<i", original_hnsw_data["metric_type"]))
|
||||||
|
if original_hnsw_data["metric_type"] > 1:
|
||||||
|
f_out.write(struct.pack("<f", original_hnsw_data["metric_arg"]))
|
||||||
|
|
||||||
|
write_numpy_vector(f_out, assign_probas_np, "d")
|
||||||
|
write_numpy_vector(f_out, cum_nneighbor_per_level_np, "i")
|
||||||
|
write_numpy_vector(f_out, levels_np, "i")
|
||||||
|
|
||||||
|
write_numpy_vector(f_out, offsets_np, "Q")
|
||||||
|
write_numpy_vector(f_out, neighbors_np, "i")
|
||||||
|
|
||||||
|
f_out.write(struct.pack("<i", original_hnsw_data["entry_point"]))
|
||||||
|
f_out.write(struct.pack("<i", original_hnsw_data["max_level"]))
|
||||||
|
f_out.write(struct.pack("<i", original_hnsw_data["efConstruction"]))
|
||||||
|
f_out.write(struct.pack("<i", original_hnsw_data["efSearch"]))
|
||||||
|
f_out.write(struct.pack("<i", original_hnsw_data["dummy_upper_beam"]))
|
||||||
|
|
||||||
|
f_out.write(struct.pack("<I", storage_fourcc))
|
||||||
|
if storage_fourcc != NULL_INDEX_FOURCC and storage_data:
|
||||||
|
f_out.write(storage_data)
|
||||||
|
|
||||||
|
|
||||||
|
def prune_hnsw_embeddings(input_filename: str, output_filename: str) -> bool:
|
||||||
|
"""Rewrite an HNSW index while dropping the embedded storage section."""
|
||||||
|
|
||||||
|
start_time = time.time()
|
||||||
|
try:
|
||||||
|
with open(input_filename, "rb") as f_in, open(output_filename, "wb") as f_out:
|
||||||
|
original_hnsw_data: dict[str, Any] = {}
|
||||||
|
|
||||||
|
hnsw_index_fourcc = read_struct(f_in, "<I")
|
||||||
|
if hnsw_index_fourcc not in EXPECTED_HNSW_FOURCCS:
|
||||||
|
print(
|
||||||
|
f"Error: Expected HNSW Index FourCC ({list(EXPECTED_HNSW_FOURCCS)}), got {hnsw_index_fourcc:08x}.",
|
||||||
|
file=sys.stderr,
|
||||||
|
)
|
||||||
|
return False
|
||||||
|
|
||||||
|
original_hnsw_data["index_fourcc"] = hnsw_index_fourcc
|
||||||
|
original_hnsw_data["d"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["ntotal"] = read_struct(f_in, "<q")
|
||||||
|
original_hnsw_data["dummy1"] = read_struct(f_in, "<q")
|
||||||
|
original_hnsw_data["dummy2"] = read_struct(f_in, "<q")
|
||||||
|
original_hnsw_data["is_trained"] = read_struct(f_in, "?")
|
||||||
|
original_hnsw_data["metric_type"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["metric_arg"] = 0.0
|
||||||
|
if original_hnsw_data["metric_type"] > 1:
|
||||||
|
original_hnsw_data["metric_arg"] = read_struct(f_in, "<f")
|
||||||
|
|
||||||
|
assign_probas_np = read_numpy_vector(f_in, np.float64, "d")
|
||||||
|
cum_nneighbor_per_level_np = read_numpy_vector(f_in, np.int32, "i")
|
||||||
|
levels_np = read_numpy_vector(f_in, np.int32, "i")
|
||||||
|
|
||||||
|
ntotal = len(levels_np)
|
||||||
|
if ntotal != original_hnsw_data["ntotal"]:
|
||||||
|
original_hnsw_data["ntotal"] = ntotal
|
||||||
|
|
||||||
|
pos_before_compact = f_in.tell()
|
||||||
|
is_compact_flag = None
|
||||||
|
try:
|
||||||
|
is_compact_flag = read_struct(f_in, "<?")
|
||||||
|
except EOFError:
|
||||||
|
is_compact_flag = None
|
||||||
|
|
||||||
|
if is_compact_flag:
|
||||||
|
compact_level_ptr = read_numpy_vector(f_in, np.uint64, "Q")
|
||||||
|
compact_node_offsets_np = read_numpy_vector(f_in, np.uint64, "Q")
|
||||||
|
|
||||||
|
original_hnsw_data["entry_point"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["max_level"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["efConstruction"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["efSearch"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["dummy_upper_beam"] = read_struct(f_in, "<i")
|
||||||
|
|
||||||
|
_storage_fourcc = read_struct(f_in, "<I")
|
||||||
|
compact_neighbors_data_np = read_numpy_vector(f_in, np.int32, "i")
|
||||||
|
compact_neighbors_data = compact_neighbors_data_np.tolist()
|
||||||
|
_storage_data = f_in.read()
|
||||||
|
|
||||||
|
write_compact_format(
|
||||||
|
f_out,
|
||||||
|
original_hnsw_data,
|
||||||
|
assign_probas_np,
|
||||||
|
cum_nneighbor_per_level_np,
|
||||||
|
levels_np,
|
||||||
|
compact_level_ptr,
|
||||||
|
compact_node_offsets_np,
|
||||||
|
compact_neighbors_data,
|
||||||
|
NULL_INDEX_FOURCC,
|
||||||
|
b"",
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
f_in.seek(pos_before_compact)
|
||||||
|
|
||||||
|
pos_before_probe = f_in.tell()
|
||||||
|
try:
|
||||||
|
suspected_flag = read_struct(f_in, "<B")
|
||||||
|
if suspected_flag != 0x00:
|
||||||
|
f_in.seek(pos_before_probe)
|
||||||
|
except EOFError:
|
||||||
|
f_in.seek(pos_before_probe)
|
||||||
|
|
||||||
|
offsets_np = read_numpy_vector(f_in, np.uint64, "Q")
|
||||||
|
neighbors_np = read_numpy_vector(f_in, np.int32, "i")
|
||||||
|
|
||||||
|
original_hnsw_data["entry_point"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["max_level"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["efConstruction"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["efSearch"] = read_struct(f_in, "<i")
|
||||||
|
original_hnsw_data["dummy_upper_beam"] = read_struct(f_in, "<i")
|
||||||
|
|
||||||
|
_storage_fourcc = None
|
||||||
|
_storage_data = b""
|
||||||
|
try:
|
||||||
|
_storage_fourcc = read_struct(f_in, "<I")
|
||||||
|
_storage_data = f_in.read()
|
||||||
|
except EOFError:
|
||||||
|
_storage_fourcc = NULL_INDEX_FOURCC
|
||||||
|
|
||||||
|
write_original_format(
|
||||||
|
f_out,
|
||||||
|
original_hnsw_data,
|
||||||
|
assign_probas_np,
|
||||||
|
cum_nneighbor_per_level_np,
|
||||||
|
levels_np,
|
||||||
|
offsets_np,
|
||||||
|
neighbors_np,
|
||||||
|
NULL_INDEX_FOURCC,
|
||||||
|
b"",
|
||||||
|
)
|
||||||
|
|
||||||
|
print(f"[{time.time() - start_time:.2f}s] Pruned embeddings from {input_filename}")
|
||||||
|
return True
|
||||||
|
except Exception as exc:
|
||||||
|
print(f"Failed to prune embeddings: {exc}", file=sys.stderr)
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
# --- Main Conversion Logic ---
|
# --- Main Conversion Logic ---
|
||||||
|
|
||||||
|
|
||||||
@@ -700,6 +984,29 @@ def convert_hnsw_graph_to_csr(input_filename, output_filename, prune_embeddings=
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def prune_hnsw_embeddings_inplace(index_filename: str) -> bool:
|
||||||
|
"""Convenience wrapper to prune embeddings in-place."""
|
||||||
|
|
||||||
|
temp_path = f"{index_filename}.prune.tmp"
|
||||||
|
success = prune_hnsw_embeddings(index_filename, temp_path)
|
||||||
|
if success:
|
||||||
|
try:
|
||||||
|
os.replace(temp_path, index_filename)
|
||||||
|
except Exception as exc: # pragma: no cover - defensive
|
||||||
|
logger.error(f"Failed to replace original index with pruned version: {exc}")
|
||||||
|
try:
|
||||||
|
os.remove(temp_path)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
return False
|
||||||
|
else:
|
||||||
|
try:
|
||||||
|
os.remove(temp_path)
|
||||||
|
except OSError:
|
||||||
|
pass
|
||||||
|
return success
|
||||||
|
|
||||||
|
|
||||||
# --- Script Execution ---
|
# --- Script Execution ---
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
parser = argparse.ArgumentParser(
|
parser = argparse.ArgumentParser(
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ from leann.interface import (
|
|||||||
from leann.registry import register_backend
|
from leann.registry import register_backend
|
||||||
from leann.searcher_base import BaseSearcher
|
from leann.searcher_base import BaseSearcher
|
||||||
|
|
||||||
from .convert_to_csr import convert_hnsw_graph_to_csr
|
from .convert_to_csr import convert_hnsw_graph_to_csr, prune_hnsw_embeddings_inplace
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -92,6 +92,8 @@ class HNSWBuilder(LeannBackendBuilderInterface):
|
|||||||
|
|
||||||
if self.is_compact:
|
if self.is_compact:
|
||||||
self._convert_to_csr(index_file)
|
self._convert_to_csr(index_file)
|
||||||
|
elif self.is_recompute:
|
||||||
|
prune_hnsw_embeddings_inplace(str(index_file))
|
||||||
|
|
||||||
def _convert_to_csr(self, index_file: Path):
|
def _convert_to_csr(self, index_file: Path):
|
||||||
"""Convert built index to CSR format"""
|
"""Convert built index to CSR format"""
|
||||||
@@ -133,10 +135,10 @@ class HNSWSearcher(BaseSearcher):
|
|||||||
if metric_enum is None:
|
if metric_enum is None:
|
||||||
raise ValueError(f"Unsupported distance_metric '{self.distance_metric}'.")
|
raise ValueError(f"Unsupported distance_metric '{self.distance_metric}'.")
|
||||||
|
|
||||||
self.is_compact, self.is_pruned = (
|
backend_meta_kwargs = self.meta.get("backend_kwargs", {})
|
||||||
self.meta.get("is_compact", True),
|
self.is_compact = self.meta.get("is_compact", backend_meta_kwargs.get("is_compact", True))
|
||||||
self.meta.get("is_pruned", True),
|
default_pruned = backend_meta_kwargs.get("is_recompute", self.is_compact)
|
||||||
)
|
self.is_pruned = bool(self.meta.get("is_pruned", default_pruned))
|
||||||
|
|
||||||
index_file = self.index_dir / f"{self.index_path.stem}.index"
|
index_file = self.index_dir / f"{self.index_path.stem}.index"
|
||||||
if not index_file.exists():
|
if not index_file.exists():
|
||||||
|
|||||||
@@ -24,13 +24,26 @@ logger = logging.getLogger(__name__)
|
|||||||
log_level = getattr(logging, LOG_LEVEL, logging.WARNING)
|
log_level = getattr(logging, LOG_LEVEL, logging.WARNING)
|
||||||
logger.setLevel(log_level)
|
logger.setLevel(log_level)
|
||||||
|
|
||||||
# Ensure we have a handler if none exists
|
# Ensure we have handlers if none exist
|
||||||
if not logger.handlers:
|
if not logger.handlers:
|
||||||
handler = logging.StreamHandler()
|
stream_handler = logging.StreamHandler()
|
||||||
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
||||||
handler.setFormatter(formatter)
|
stream_handler.setFormatter(formatter)
|
||||||
logger.addHandler(handler)
|
logger.addHandler(stream_handler)
|
||||||
logger.propagate = False
|
|
||||||
|
log_path = os.getenv("LEANN_HNSW_LOG_PATH")
|
||||||
|
if log_path:
|
||||||
|
try:
|
||||||
|
file_handler = logging.FileHandler(log_path, mode="a", encoding="utf-8")
|
||||||
|
file_formatter = logging.Formatter(
|
||||||
|
"%(asctime)s - %(levelname)s - [pid=%(process)d] %(message)s"
|
||||||
|
)
|
||||||
|
file_handler.setFormatter(file_formatter)
|
||||||
|
logger.addHandler(file_handler)
|
||||||
|
except Exception as exc: # pragma: no cover - best effort logging
|
||||||
|
logger.warning(f"Failed to attach file handler for log path {log_path}: {exc}")
|
||||||
|
|
||||||
|
logger.propagate = False
|
||||||
|
|
||||||
|
|
||||||
def create_hnsw_embedding_server(
|
def create_hnsw_embedding_server(
|
||||||
|
|||||||
Submodule packages/leann-backend-hnsw/third_party/faiss updated: ed96ff7dba...1d51f0c074
@@ -15,6 +15,7 @@ from pathlib import Path
|
|||||||
from typing import Any, Literal, Optional, Union
|
from typing import Any, Literal, Optional, Union
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
from leann_backend_hnsw.convert_to_csr import prune_hnsw_embeddings_inplace
|
||||||
|
|
||||||
from leann.interface import LeannBackendSearcherInterface
|
from leann.interface import LeannBackendSearcherInterface
|
||||||
|
|
||||||
@@ -476,9 +477,7 @@ class LeannBuilder:
|
|||||||
is_compact = self.backend_kwargs.get("is_compact", True)
|
is_compact = self.backend_kwargs.get("is_compact", True)
|
||||||
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
||||||
meta_data["is_compact"] = is_compact
|
meta_data["is_compact"] = is_compact
|
||||||
meta_data["is_pruned"] = (
|
meta_data["is_pruned"] = bool(is_recompute)
|
||||||
is_compact and is_recompute
|
|
||||||
) # Pruned only if compact and recompute
|
|
||||||
with open(leann_meta_path, "w", encoding="utf-8") as f:
|
with open(leann_meta_path, "w", encoding="utf-8") as f:
|
||||||
json.dump(meta_data, f, indent=2)
|
json.dump(meta_data, f, indent=2)
|
||||||
|
|
||||||
@@ -598,13 +597,157 @@ class LeannBuilder:
|
|||||||
is_compact = self.backend_kwargs.get("is_compact", True)
|
is_compact = self.backend_kwargs.get("is_compact", True)
|
||||||
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
||||||
meta_data["is_compact"] = is_compact
|
meta_data["is_compact"] = is_compact
|
||||||
meta_data["is_pruned"] = is_compact and is_recompute
|
meta_data["is_pruned"] = bool(is_recompute)
|
||||||
|
|
||||||
with open(leann_meta_path, "w", encoding="utf-8") as f:
|
with open(leann_meta_path, "w", encoding="utf-8") as f:
|
||||||
json.dump(meta_data, f, indent=2)
|
json.dump(meta_data, f, indent=2)
|
||||||
|
|
||||||
logger.info(f"Index built successfully from precomputed embeddings: {index_path}")
|
logger.info(f"Index built successfully from precomputed embeddings: {index_path}")
|
||||||
|
|
||||||
|
def update_index(self, index_path: str):
|
||||||
|
"""Append new passages and vectors to an existing HNSW index."""
|
||||||
|
if not self.chunks:
|
||||||
|
raise ValueError("No new chunks provided for update.")
|
||||||
|
|
||||||
|
path = Path(index_path)
|
||||||
|
index_dir = path.parent
|
||||||
|
index_name = path.name
|
||||||
|
index_prefix = path.stem
|
||||||
|
|
||||||
|
meta_path = index_dir / f"{index_name}.meta.json"
|
||||||
|
passages_file = index_dir / f"{index_name}.passages.jsonl"
|
||||||
|
offset_file = index_dir / f"{index_name}.passages.idx"
|
||||||
|
index_file = index_dir / f"{index_prefix}.index"
|
||||||
|
|
||||||
|
if not meta_path.exists() or not passages_file.exists() or not offset_file.exists():
|
||||||
|
raise FileNotFoundError("Index metadata or passage files are missing; cannot update.")
|
||||||
|
if not index_file.exists():
|
||||||
|
raise FileNotFoundError(f"HNSW index file not found: {index_file}")
|
||||||
|
|
||||||
|
with open(meta_path, encoding="utf-8") as f:
|
||||||
|
meta = json.load(f)
|
||||||
|
backend_name = meta.get("backend_name")
|
||||||
|
if backend_name != self.backend_name:
|
||||||
|
raise ValueError(
|
||||||
|
f"Index was built with backend '{backend_name}', cannot update with '{self.backend_name}'."
|
||||||
|
)
|
||||||
|
|
||||||
|
meta_backend_kwargs = meta.get("backend_kwargs", {})
|
||||||
|
index_is_compact = meta.get("is_compact", meta_backend_kwargs.get("is_compact", True))
|
||||||
|
if index_is_compact:
|
||||||
|
raise ValueError(
|
||||||
|
"Compact HNSW indices do not support in-place updates. Rebuild required."
|
||||||
|
)
|
||||||
|
|
||||||
|
distance_metric = meta_backend_kwargs.get(
|
||||||
|
"distance_metric", self.backend_kwargs.get("distance_metric", "mips")
|
||||||
|
).lower()
|
||||||
|
needs_recompute = bool(
|
||||||
|
meta.get("is_pruned")
|
||||||
|
or meta_backend_kwargs.get("is_recompute")
|
||||||
|
or self.backend_kwargs.get("is_recompute")
|
||||||
|
)
|
||||||
|
|
||||||
|
with open(offset_file, "rb") as f:
|
||||||
|
offset_map: dict[str, int] = pickle.load(f)
|
||||||
|
existing_ids = set(offset_map.keys())
|
||||||
|
|
||||||
|
valid_chunks: list[dict[str, Any]] = []
|
||||||
|
for chunk in self.chunks:
|
||||||
|
text = chunk.get("text", "")
|
||||||
|
if not isinstance(text, str) or not text.strip():
|
||||||
|
continue
|
||||||
|
metadata = chunk.setdefault("metadata", {})
|
||||||
|
passage_id = chunk.get("id") or metadata.get("id")
|
||||||
|
if passage_id and passage_id in existing_ids:
|
||||||
|
raise ValueError(f"Passage ID '{passage_id}' already exists in the index.")
|
||||||
|
valid_chunks.append(chunk)
|
||||||
|
|
||||||
|
if not valid_chunks:
|
||||||
|
raise ValueError("No valid chunks to append.")
|
||||||
|
|
||||||
|
texts_to_embed = [chunk["text"] for chunk in valid_chunks]
|
||||||
|
embeddings = compute_embeddings(
|
||||||
|
texts_to_embed,
|
||||||
|
self.embedding_model,
|
||||||
|
self.embedding_mode,
|
||||||
|
use_server=False,
|
||||||
|
is_build=True,
|
||||||
|
)
|
||||||
|
|
||||||
|
embedding_dim = embeddings.shape[1]
|
||||||
|
expected_dim = meta.get("dimensions")
|
||||||
|
if expected_dim is not None and expected_dim != embedding_dim:
|
||||||
|
raise ValueError(
|
||||||
|
f"Dimension mismatch during update: existing index uses {expected_dim}, got {embedding_dim}."
|
||||||
|
)
|
||||||
|
|
||||||
|
from leann_backend_hnsw import faiss # type: ignore
|
||||||
|
|
||||||
|
embeddings = np.ascontiguousarray(embeddings, dtype=np.float32)
|
||||||
|
if distance_metric == "cosine":
|
||||||
|
norms = np.linalg.norm(embeddings, axis=1, keepdims=True)
|
||||||
|
norms[norms == 0] = 1
|
||||||
|
embeddings = embeddings / norms
|
||||||
|
|
||||||
|
index = faiss.read_index(str(index_file))
|
||||||
|
if hasattr(index, "is_recompute"):
|
||||||
|
index.is_recompute = needs_recompute
|
||||||
|
if getattr(index, "storage", None) is None:
|
||||||
|
if index.metric_type == faiss.METRIC_INNER_PRODUCT:
|
||||||
|
storage_index = faiss.IndexFlatIP(index.d)
|
||||||
|
else:
|
||||||
|
storage_index = faiss.IndexFlatL2(index.d)
|
||||||
|
index.storage = storage_index
|
||||||
|
index.own_fields = True
|
||||||
|
if index.d != embedding_dim:
|
||||||
|
raise ValueError(
|
||||||
|
f"Existing index dimension ({index.d}) does not match new embeddings ({embedding_dim})."
|
||||||
|
)
|
||||||
|
|
||||||
|
base_id = index.ntotal
|
||||||
|
for offset, chunk in enumerate(valid_chunks):
|
||||||
|
new_id = str(base_id + offset)
|
||||||
|
chunk.setdefault("metadata", {})["id"] = new_id
|
||||||
|
chunk["id"] = new_id
|
||||||
|
|
||||||
|
index.add(embeddings.shape[0], faiss.swig_ptr(embeddings))
|
||||||
|
faiss.write_index(index, str(index_file))
|
||||||
|
|
||||||
|
with open(passages_file, "a", encoding="utf-8") as f:
|
||||||
|
for chunk in valid_chunks:
|
||||||
|
offset = f.tell()
|
||||||
|
json.dump(
|
||||||
|
{
|
||||||
|
"id": chunk["id"],
|
||||||
|
"text": chunk["text"],
|
||||||
|
"metadata": chunk.get("metadata", {}),
|
||||||
|
},
|
||||||
|
f,
|
||||||
|
ensure_ascii=False,
|
||||||
|
)
|
||||||
|
f.write("\n")
|
||||||
|
offset_map[chunk["id"]] = offset
|
||||||
|
|
||||||
|
with open(offset_file, "wb") as f:
|
||||||
|
pickle.dump(offset_map, f)
|
||||||
|
|
||||||
|
meta["total_passages"] = len(offset_map)
|
||||||
|
with open(meta_path, "w", encoding="utf-8") as f:
|
||||||
|
json.dump(meta, f, indent=2)
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
"Appended %d passages to index '%s'. New total: %d",
|
||||||
|
len(valid_chunks),
|
||||||
|
index_path,
|
||||||
|
len(offset_map),
|
||||||
|
)
|
||||||
|
|
||||||
|
self.chunks.clear()
|
||||||
|
|
||||||
|
if needs_recompute:
|
||||||
|
prune_hnsw_embeddings_inplace(str(index_file))
|
||||||
|
|
||||||
|
|
||||||
class LeannSearcher:
|
class LeannSearcher:
|
||||||
def __init__(self, index_path: str, enable_warmup: bool = False, **backend_kwargs):
|
def __init__(self, index_path: str, enable_warmup: bool = False, **backend_kwargs):
|
||||||
|
|||||||
Reference in New Issue
Block a user