remove ipynb

.gitignore

@@ -18,6 +18,7 @@ demo/experiment_results/**/*.json
 *.eml
 *.emlx
 *.json
+*.png
 !.vscode/*.json
 *.sh
 *.txt
@@ -101,3 +102,6 @@ CLAUDE.local.md
 .claude/*.local.*
 .claude/local/*
 benchmarks/data/
+
+## multi vector
+apps/multimodal/vision-based-pdf-multi-vector/multi-vector-colpali-native-weaviate.py

apps/base_rag_example.py

@@ -11,7 +11,6 @@ from typing import Any
 import dotenv
 from leann.api import LeannBuilder, LeannChat
 from leann.registry import register_project_directory
-from leann.settings import resolve_ollama_host, resolve_openai_api_key, resolve_openai_base_url
 
 dotenv.load_dotenv()
 
@@ -79,24 +78,6 @@ class BaseRAGExample(ABC):
             choices=["sentence-transformers", "openai", "mlx", "ollama"],
             help="Embedding backend mode (default: sentence-transformers), we provide sentence-transformers, openai, mlx, or ollama",
         )
-        embedding_group.add_argument(
-            "--embedding-host",
-            type=str,
-            default=None,
-            help="Override Ollama-compatible embedding host",
-        )
-        embedding_group.add_argument(
-            "--embedding-api-base",
-            type=str,
-            default=None,
-            help="Base URL for OpenAI-compatible embedding services",
-        )
-        embedding_group.add_argument(
-            "--embedding-api-key",
-            type=str,
-            default=None,
-            help="API key for embedding service (defaults to OPENAI_API_KEY)",
-        )
 
         # LLM parameters
         llm_group = parser.add_argument_group("LLM Parameters")
@@ -116,8 +97,8 @@ class BaseRAGExample(ABC):
         llm_group.add_argument(
             "--llm-host",
             type=str,
-            default=None,
-            help="Host for Ollama-compatible APIs (defaults to LEANN_OLLAMA_HOST/OLLAMA_HOST)",
+            default="http://localhost:11434",
+            help="Host for Ollama API (default: http://localhost:11434)",
         )
         llm_group.add_argument(
             "--thinking-budget",
@@ -126,18 +107,6 @@ class BaseRAGExample(ABC):
             default=None,
             help="Thinking budget for reasoning models (low/medium/high). Supported by GPT-Oss:20b and other reasoning models.",
         )
-        llm_group.add_argument(
-            "--llm-api-base",
-            type=str,
-            default=None,
-            help="Base URL for OpenAI-compatible APIs",
-        )
-        llm_group.add_argument(
-            "--llm-api-key",
-            type=str,
-            default=None,
-            help="API key for OpenAI-compatible APIs (defaults to OPENAI_API_KEY)",
-        )
 
         # AST Chunking parameters
         ast_group = parser.add_argument_group("AST Chunking Parameters")
@@ -236,13 +205,9 @@ class BaseRAGExample(ABC):
 
         if args.llm == "openai":
             config["model"] = args.llm_model or "gpt-4o"
-            config["base_url"] = resolve_openai_base_url(args.llm_api_base)
-            resolved_key = resolve_openai_api_key(args.llm_api_key)
-            if resolved_key:
-                config["api_key"] = resolved_key
         elif args.llm == "ollama":
             config["model"] = args.llm_model or "llama3.2:1b"
-            config["host"] = resolve_ollama_host(args.llm_host)
+            config["host"] = args.llm_host
         elif args.llm == "hf":
             config["model"] = args.llm_model or "Qwen/Qwen2.5-1.5B-Instruct"
         elif args.llm == "simulated":
@@ -258,20 +223,10 @@ class BaseRAGExample(ABC):
         print(f"\n[Building Index] Creating {self.name} index...")
         print(f"Total text chunks: {len(texts)}")
 
-        embedding_options: dict[str, Any] = {}
-        if args.embedding_mode == "ollama":
-            embedding_options["host"] = resolve_ollama_host(args.embedding_host)
-        elif args.embedding_mode == "openai":
-            embedding_options["base_url"] = resolve_openai_base_url(args.embedding_api_base)
-            resolved_embedding_key = resolve_openai_api_key(args.embedding_api_key)
-            if resolved_embedding_key:
-                embedding_options["api_key"] = resolved_embedding_key
-
         builder = LeannBuilder(
             backend_name=args.backend_name,
             embedding_model=args.embedding_model,
             embedding_mode=args.embedding_mode,
-            embedding_options=embedding_options or None,
             graph_degree=args.graph_degree,
             complexity=args.build_complexity,
             is_compact=not args.no_compact,

apps/multimodal/vision-based-pdf-multi-vector/leann_multi_vector.py

@@ -0,0 +1,182 @@
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import numpy as np
+
+
+def _ensure_repo_paths_importable(current_file: str) -> None:
+    _repo_root = Path(current_file).resolve().parents[3]
+    _leann_core_src = _repo_root / "packages" / "leann-core" / "src"
+    _leann_hnsw_pkg = _repo_root / "packages" / "leann-backend-hnsw"
+    if str(_leann_core_src) not in sys.path:
+        sys.path.append(str(_leann_core_src))
+    if str(_leann_hnsw_pkg) not in sys.path:
+        sys.path.append(str(_leann_hnsw_pkg))
+
+
+_ensure_repo_paths_importable(__file__)
+
+from leann_backend_hnsw.hnsw_backend import HNSWBuilder, HNSWSearcher  # noqa: E402
+
+
+class LeannMultiVector:
+    def __init__(
+        self,
+        index_path: str,
+        dim: int = 128,
+        distance_metric: str = "mips",
+        m: int = 16,
+        ef_construction: int = 500,
+        is_compact: bool = False,
+        is_recompute: bool = False,
+        embedding_model_name: str = "colvision",
+    ) -> None:
+        self.index_path = index_path
+        self.dim = dim
+        self.embedding_model_name = embedding_model_name
+        self._pending_items: list[dict] = []
+        self._backend_kwargs = {
+            "distance_metric": distance_metric,
+            "M": m,
+            "efConstruction": ef_construction,
+            "is_compact": is_compact,
+            "is_recompute": is_recompute,
+        }
+        self._labels_meta: list[dict] = []
+
+    def _meta_dict(self) -> dict:
+        return {
+            "version": "1.0",
+            "backend_name": "hnsw",
+            "embedding_model": self.embedding_model_name,
+            "embedding_mode": "custom",
+            "dimensions": self.dim,
+            "backend_kwargs": self._backend_kwargs,
+            "is_compact": self._backend_kwargs.get("is_compact", True),
+            "is_pruned": self._backend_kwargs.get("is_compact", True)
+            and self._backend_kwargs.get("is_recompute", True),
+        }
+
+    def create_collection(self) -> None:
+        path = Path(self.index_path)
+        path.parent.mkdir(parents=True, exist_ok=True)
+
+    def insert(self, data: dict) -> None:
+        self._pending_items.append(
+            {
+                "doc_id": int(data["doc_id"]),
+                "filepath": data.get("filepath", ""),
+                "colbert_vecs": [np.asarray(v, dtype=np.float32) for v in data["colbert_vecs"]],
+            }
+        )
+
+    def _labels_path(self) -> Path:
+        index_path_obj = Path(self.index_path)
+        return index_path_obj.parent / f"{index_path_obj.name}.labels.json"
+
+    def _meta_path(self) -> Path:
+        index_path_obj = Path(self.index_path)
+        return index_path_obj.parent / f"{index_path_obj.name}.meta.json"
+
+    def create_index(self) -> None:
+        if not self._pending_items:
+            return
+
+        embeddings: list[np.ndarray] = []
+        labels_meta: list[dict] = []
+
+        for item in self._pending_items:
+            doc_id = int(item["doc_id"])
+            filepath = item.get("filepath", "")
+            colbert_vecs = item["colbert_vecs"]
+            for seq_id, vec in enumerate(colbert_vecs):
+                vec_np = np.asarray(vec, dtype=np.float32)
+                embeddings.append(vec_np)
+                labels_meta.append(
+                    {
+                        "id": f"{doc_id}:{seq_id}",
+                        "doc_id": doc_id,
+                        "seq_id": int(seq_id),
+                        "filepath": filepath,
+                    }
+                )
+
+        if not embeddings:
+            return
+
+        embeddings_np = np.vstack(embeddings).astype(np.float32)
+        # print shape of embeddings_np
+        print(embeddings_np.shape)
+
+        builder = HNSWBuilder(**{**self._backend_kwargs, "dimensions": self.dim})
+        ids = [str(i) for i in range(embeddings_np.shape[0])]
+        builder.build(embeddings_np, ids, self.index_path)
+
+        import json as _json
+
+        with open(self._meta_path(), "w", encoding="utf-8") as f:
+            _json.dump(self._meta_dict(), f, indent=2)
+        with open(self._labels_path(), "w", encoding="utf-8") as f:
+            _json.dump(labels_meta, f)
+
+        self._labels_meta = labels_meta
+
+    def _load_labels_meta_if_needed(self) -> None:
+        if self._labels_meta:
+            return
+        labels_path = self._labels_path()
+        if labels_path.exists():
+            import json as _json
+
+            with open(labels_path, encoding="utf-8") as f:
+                self._labels_meta = _json.load(f)
+
+    def search(
+        self, data: np.ndarray, topk: int, first_stage_k: int = 50
+    ) -> list[tuple[float, int]]:
+        if data.ndim == 1:
+            data = data.reshape(1, -1)
+        if data.dtype != np.float32:
+            data = data.astype(np.float32)
+
+        self._load_labels_meta_if_needed()
+
+        searcher = HNSWSearcher(self.index_path, meta=self._meta_dict())
+        raw = searcher.search(
+            data,
+            first_stage_k,
+            recompute_embeddings=False,
+            complexity=128,
+            beam_width=1,
+            prune_ratio=0.0,
+            batch_size=0,
+        )
+
+        labels = raw.get("labels")
+        distances = raw.get("distances")
+        if labels is None or distances is None:
+            return []
+
+        doc_scores: dict[int, float] = {}
+        B = len(labels)
+        for b in range(B):
+            per_doc_best: dict[int, float] = {}
+            for k, sid in enumerate(labels[b]):
+                try:
+                    idx = int(sid)
+                except Exception:
+                    continue
+                if 0 <= idx < len(self._labels_meta):
+                    doc_id = int(self._labels_meta[idx]["doc_id"])  # type: ignore[index]
+                else:
+                    continue
+                score = float(distances[b][k])
+                if (doc_id not in per_doc_best) or (score > per_doc_best[doc_id]):
+                    per_doc_best[doc_id] = score
+            for doc_id, best_score in per_doc_best.items():
+                doc_scores[doc_id] = doc_scores.get(doc_id, 0.0) + best_score
+
+        scores = sorted(((v, k) for k, v in doc_scores.items()), key=lambda x: x[0], reverse=True)
+        return scores[:topk] if len(scores) >= topk else scores

apps/multimodal/vision-based-pdf-multi-vector/multi-vector-leann-similarity-map.py

@@ -0,0 +1,477 @@
+## Jupyter-style notebook script
+# %%
+# uv pip install matplotlib qwen_vl_utils
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Any, Optional, cast
+
+from PIL import Image
+from tqdm import tqdm
+
+
+def _ensure_repo_paths_importable(current_file: str) -> None:
+    """Make local leann packages importable without installing (mirrors multi-vector-leann.py)."""
+    _repo_root = Path(current_file).resolve().parents[3]
+    _leann_core_src = _repo_root / "packages" / "leann-core" / "src"
+    _leann_hnsw_pkg = _repo_root / "packages" / "leann-backend-hnsw"
+    if str(_leann_core_src) not in sys.path:
+        sys.path.append(str(_leann_core_src))
+    if str(_leann_hnsw_pkg) not in sys.path:
+        sys.path.append(str(_leann_hnsw_pkg))
+
+
+_ensure_repo_paths_importable(__file__)
+
+from leann_multi_vector import LeannMultiVector  # noqa: E402
+
+# %%
+# Config
+os.environ["TOKENIZERS_PARALLELISM"] = "false"
+QUERY = "How does DeepSeek-V2 compare against the LLaMA family of LLMs?"
+MODEL: str = "colqwen2"  # "colpali" or "colqwen2"
+
+# Data source: set to True to use the Hugging Face dataset example (recommended)
+USE_HF_DATASET: bool = True
+DATASET_NAME: str = "weaviate/arXiv-AI-papers-multi-vector"
+DATASET_SPLIT: str = "train"
+MAX_DOCS: Optional[int] = None  # limit number of pages to index; None = all
+
+# Local pages (used when USE_HF_DATASET == False)
+PDF: Optional[str] = None  # e.g., "./pdfs/2004.12832v2.pdf"
+PAGES_DIR: str = "./pages"
+
+# Index + retrieval settings
+INDEX_PATH: str = "./indexes/colvision.leann"
+TOPK: int = 1
+FIRST_STAGE_K: int = 500
+REBUILD_INDEX: bool = False
+
+# Artifacts
+SAVE_TOP_IMAGE: Optional[str] = "./figures/retrieved_page.png"
+SIMILARITY_MAP: bool = True
+SIM_TOKEN_IDX: int = 13  # -1 means auto-select the most salient token
+SIM_OUTPUT: str = "./figures/similarity_map.png"
+ANSWER: bool = True
+MAX_NEW_TOKENS: int = 128
+
+
+# %%
+# Helpers
+def _natural_sort_key(name: str) -> int:
+    m = re.search(r"\d+", name)
+    return int(m.group()) if m else 0
+
+
+def _load_images_from_dir(pages_dir: str) -> tuple[list[str], list[Image.Image]]:
+    filenames = [n for n in os.listdir(pages_dir) if n.lower().endswith((".png", ".jpg", ".jpeg"))]
+    filenames = sorted(filenames, key=_natural_sort_key)
+    filepaths = [os.path.join(pages_dir, n) for n in filenames]
+    images = [Image.open(p) for p in filepaths]
+    return filepaths, images
+
+
+def _maybe_convert_pdf_to_images(pdf_path: Optional[str], pages_dir: str, dpi: int = 200) -> None:
+    if not pdf_path:
+        return
+    os.makedirs(pages_dir, exist_ok=True)
+    try:
+        from pdf2image import convert_from_path
+    except Exception as e:
+        raise RuntimeError(
+            "pdf2image is required to convert PDF to images. Install via pip install pdf2image"
+        ) from e
+    images = convert_from_path(pdf_path, dpi=dpi)
+    for i, image in enumerate(images):
+        image.save(os.path.join(pages_dir, f"page_{i + 1}.png"), "PNG")
+
+
+def _select_device_and_dtype():
+    import torch
+    from colpali_engine.utils.torch_utils import get_torch_device
+
+    device_str = (
+        "cuda"
+        if torch.cuda.is_available()
+        else (
+            "mps"
+            if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available()
+            else "cpu"
+        )
+    )
+    device = get_torch_device(device_str)
+    # Stable dtype selection to avoid NaNs:
+    # - CUDA: prefer bfloat16 if supported, else float16
+    # - MPS: use float32 (fp16 on MPS can produce NaNs in some ops)
+    # - CPU: float32
+    if device_str == "cuda":
+        dtype = torch.bfloat16 if torch.cuda.is_bf16_supported() else torch.float16
+        try:
+            torch.backends.cuda.matmul.allow_tf32 = True  # Better stability/perf on Ampere+
+        except Exception:
+            pass
+    elif device_str == "mps":
+        dtype = torch.float32
+    else:
+        dtype = torch.float32
+    return device_str, device, dtype
+
+
+def _load_colvision(model_choice: str):
+    import torch
+    from colpali_engine.models import ColPali, ColQwen2, ColQwen2Processor
+    from colpali_engine.models.paligemma.colpali.processing_colpali import ColPaliProcessor
+    from transformers.utils.import_utils import is_flash_attn_2_available
+
+    device_str, device, dtype = _select_device_and_dtype()
+
+    if model_choice == "colqwen2":
+        model_name = "vidore/colqwen2-v1.0"
+        # On CPU/MPS we must avoid flash-attn and stay eager; on CUDA prefer flash-attn if available
+        attn_implementation = (
+            "flash_attention_2"
+            if (device_str == "cuda" and is_flash_attn_2_available())
+            else "eager"
+        )
+        model = ColQwen2.from_pretrained(
+            model_name,
+            torch_dtype=torch.bfloat16,
+            device_map=device,
+            attn_implementation=attn_implementation,
+        ).eval()
+        processor = ColQwen2Processor.from_pretrained(model_name)
+    else:
+        model_name = "vidore/colpali-v1.2"
+        model = ColPali.from_pretrained(
+            model_name,
+            torch_dtype=torch.bfloat16,
+            device_map=device,
+        ).eval()
+        processor = cast(ColPaliProcessor, ColPaliProcessor.from_pretrained(model_name))
+
+    return model_name, model, processor, device_str, device, dtype
+
+
+def _embed_images(model, processor, images: list[Image.Image]) -> list[Any]:
+    import torch
+    from colpali_engine.utils.torch_utils import ListDataset
+    from torch.utils.data import DataLoader
+
+    # Ensure deterministic eval and autocast for stability
+    model.eval()
+
+    dataloader = DataLoader(
+        dataset=ListDataset[Image.Image](images),
+        batch_size=1,
+        shuffle=False,
+        collate_fn=lambda x: processor.process_images(x),
+    )
+
+    doc_vecs: list[Any] = []
+    for batch_doc in dataloader:
+        with torch.no_grad():
+            batch_doc = {k: v.to(model.device) for k, v in batch_doc.items()}
+            # autocast on CUDA for bf16/fp16; on CPU/MPS stay in fp32
+            if model.device.type == "cuda":
+                with torch.autocast(
+                    device_type="cuda",
+                    dtype=model.dtype if model.dtype.is_floating_point else torch.bfloat16,
+                ):
+                    embeddings_doc = model(**batch_doc)
+            else:
+                embeddings_doc = model(**batch_doc)
+        doc_vecs.extend(list(torch.unbind(embeddings_doc.to("cpu"))))
+    return doc_vecs
+
+
+def _embed_queries(model, processor, queries: list[str]) -> list[Any]:
+    import torch
+    from colpali_engine.utils.torch_utils import ListDataset
+    from torch.utils.data import DataLoader
+
+    model.eval()
+
+    dataloader = DataLoader(
+        dataset=ListDataset[str](queries),
+        batch_size=1,
+        shuffle=False,
+        collate_fn=lambda x: processor.process_queries(x),
+    )
+
+    q_vecs: list[Any] = []
+    for batch_query in dataloader:
+        with torch.no_grad():
+            batch_query = {k: v.to(model.device) for k, v in batch_query.items()}
+            if model.device.type == "cuda":
+                with torch.autocast(
+                    device_type="cuda",
+                    dtype=model.dtype if model.dtype.is_floating_point else torch.bfloat16,
+                ):
+                    embeddings_query = model(**batch_query)
+            else:
+                embeddings_query = model(**batch_query)
+        q_vecs.extend(list(torch.unbind(embeddings_query.to("cpu"))))
+    return q_vecs
+
+
+def _build_index(index_path: str, doc_vecs: list[Any], filepaths: list[str]) -> LeannMultiVector:
+    dim = int(doc_vecs[0].shape[-1])
+    retriever = LeannMultiVector(index_path=index_path, dim=dim)
+    retriever.create_collection()
+    for i, vec in enumerate(doc_vecs):
+        data = {
+            "colbert_vecs": vec.float().numpy(),
+            "doc_id": i,
+            "filepath": filepaths[i],
+        }
+        retriever.insert(data)
+    retriever.create_index()
+    return retriever
+
+
+def _load_retriever_if_index_exists(index_path: str, dim: int) -> Optional[LeannMultiVector]:
+    index_base = Path(index_path)
+    # Rough heuristic: index dir exists AND meta+labels files exist
+    meta = index_base.parent / f"{index_base.name}.meta.json"
+    labels = index_base.parent / f"{index_base.name}.labels.json"
+    if index_base.exists() and meta.exists() and labels.exists():
+        return LeannMultiVector(index_path=index_path, dim=dim)
+    return None
+
+
+def _generate_similarity_map(
+    model,
+    processor,
+    image: Image.Image,
+    query: str,
+    token_idx: Optional[int] = None,
+    output_path: Optional[str] = None,
+) -> tuple[int, float]:
+    import torch
+    from colpali_engine.interpretability import (
+        get_similarity_maps_from_embeddings,
+        plot_similarity_map,
+    )
+
+    batch_images = processor.process_images([image]).to(model.device)
+    batch_queries = processor.process_queries([query]).to(model.device)
+
+    with torch.no_grad():
+        image_embeddings = model.forward(**batch_images)
+        query_embeddings = model.forward(**batch_queries)
+
+    n_patches = processor.get_n_patches(
+        image_size=image.size,
+        spatial_merge_size=getattr(model, "spatial_merge_size", None),
+    )
+    image_mask = processor.get_image_mask(batch_images)
+
+    batched_similarity_maps = get_similarity_maps_from_embeddings(
+        image_embeddings=image_embeddings,
+        query_embeddings=query_embeddings,
+        n_patches=n_patches,
+        image_mask=image_mask,
+    )
+
+    similarity_maps = batched_similarity_maps[0]
+
+    # Determine token index if not provided: choose the token with highest max score
+    if token_idx is None:
+        per_token_max = similarity_maps.view(similarity_maps.shape[0], -1).max(dim=1).values
+        token_idx = int(per_token_max.argmax().item())
+
+    max_sim_score = similarity_maps[token_idx, :, :].max().item()
+
+    if output_path:
+        import matplotlib.pyplot as plt
+
+        fig, ax = plot_similarity_map(
+            image=image,
+            similarity_map=similarity_maps[token_idx],
+            figsize=(14, 14),
+            show_colorbar=False,
+        )
+        ax.set_title(f"Token #{token_idx}. MaxSim score: {max_sim_score:.2f}", fontsize=12)
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        plt.savefig(output_path, bbox_inches="tight")
+        plt.close(fig)
+
+    return token_idx, float(max_sim_score)
+
+
+class QwenVL:
+    def __init__(self, device: str):
+        from transformers import AutoProcessor, Qwen2_5_VLForConditionalGeneration
+        from transformers.utils.import_utils import is_flash_attn_2_available
+
+        attn_implementation = "flash_attention_2" if is_flash_attn_2_available() else "eager"
+        self.model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
+            "Qwen/Qwen2.5-VL-3B-Instruct",
+            torch_dtype="auto",
+            device_map=device,
+            attn_implementation=attn_implementation,
+        )
+
+        min_pixels = 256 * 28 * 28
+        max_pixels = 1280 * 28 * 28
+        self.processor = AutoProcessor.from_pretrained(
+            "Qwen/Qwen2.5-VL-3B-Instruct", min_pixels=min_pixels, max_pixels=max_pixels
+        )
+
+    def answer(self, query: str, images: list[Image.Image], max_new_tokens: int = 128) -> str:
+        import base64
+        from io import BytesIO
+
+        from qwen_vl_utils import process_vision_info
+
+        content = []
+        for img in images:
+            buffer = BytesIO()
+            img.save(buffer, format="jpeg")
+            img_base64 = base64.b64encode(buffer.getvalue()).decode("utf-8")
+            content.append({"type": "image", "image": f"data:image;base64,{img_base64}"})
+        content.append({"type": "text", "text": query})
+        messages = [{"role": "user", "content": content}]
+
+        text = self.processor.apply_chat_template(
+            messages, tokenize=False, add_generation_prompt=True
+        )
+        image_inputs, video_inputs = process_vision_info(messages)
+        inputs = self.processor(
+            text=[text], images=image_inputs, videos=video_inputs, padding=True, return_tensors="pt"
+        )
+        inputs = inputs.to(self.model.device)
+
+        generated_ids = self.model.generate(**inputs, max_new_tokens=max_new_tokens)
+        generated_ids_trimmed = [
+            out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
+        ]
+        return self.processor.batch_decode(
+            generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
+        )[0]
+
+
+# %%
+
+# Step 1: Prepare data
+if USE_HF_DATASET:
+    from datasets import load_dataset
+
+    dataset = load_dataset(DATASET_NAME, split=DATASET_SPLIT)
+    N = len(dataset) if MAX_DOCS is None else min(MAX_DOCS, len(dataset))
+    filepaths: list[str] = []
+    images: list[Image.Image] = []
+    for i in tqdm(range(N), desc="Loading dataset"):
+        p = dataset[i]
+        # Compose a descriptive identifier for printing later
+        identifier = f"arXiv:{p['paper_arxiv_id']}|title:{p['paper_title']}|page:{int(p['page_number'])}|id:{p['page_id']}"
+        print(identifier)
+        filepaths.append(identifier)
+        images.append(p["page_image"])  # PIL Image
+else:
+    _maybe_convert_pdf_to_images(PDF, PAGES_DIR)
+    filepaths, images = _load_images_from_dir(PAGES_DIR)
+    if not images:
+        raise RuntimeError(
+            f"No images found in {PAGES_DIR}. Provide PDF path in PDF variable or ensure images exist."
+        )
+
+
+# %%
+# Step 2: Load model and processor
+model_name, model, processor, device_str, device, dtype = _load_colvision(MODEL)
+print(f"Using model={model_name}, device={device_str}, dtype={dtype}")
+
+
+# %%
+
+# %%
+# Step 3: Build or load index
+retriever: Optional[LeannMultiVector] = None
+if not REBUILD_INDEX:
+    try:
+        one_vec = _embed_images(model, processor, [images[0]])[0]
+        retriever = _load_retriever_if_index_exists(INDEX_PATH, dim=int(one_vec.shape[-1]))
+    except Exception:
+        retriever = None
+
+if retriever is None:
+    doc_vecs = _embed_images(model, processor, images)
+    retriever = _build_index(INDEX_PATH, doc_vecs, filepaths)
+
+
+# %%
+# Step 4: Embed query and search
+q_vec = _embed_queries(model, processor, [QUERY])[0]
+results = retriever.search(q_vec.float().numpy(), topk=TOPK, first_stage_k=FIRST_STAGE_K)
+if not results:
+    print("No results found.")
+else:
+    print(f'Top {len(results)} results for query: "{QUERY}"')
+    top_images: list[Image.Image] = []
+    for rank, (score, doc_id) in enumerate(results, start=1):
+        path = filepaths[doc_id]
+        # For HF dataset, path is a descriptive identifier, not a real file path
+        print(f"{rank}) MaxSim: {score:.4f}, Page: {path}")
+        top_images.append(images[doc_id])
+
+    if SAVE_TOP_IMAGE:
+        from pathlib import Path as _Path
+
+        base = _Path(SAVE_TOP_IMAGE)
+        base.parent.mkdir(parents=True, exist_ok=True)
+        for rank, img in enumerate(top_images[:TOPK], start=1):
+            if base.suffix:
+                out_path = base.parent / f"{base.stem}_rank{rank}{base.suffix}"
+            else:
+                out_path = base / f"retrieved_page_rank{rank}.png"
+            img.save(str(out_path))
+            print(f"Saved retrieved page (rank {rank}) to: {out_path}")
+
+## TODO stange results of second page of DeepSeek-V2 rather than the first page
+
+# %%
+# Step 5: Similarity maps for top-K results
+if results and SIMILARITY_MAP:
+    token_idx = None if SIM_TOKEN_IDX < 0 else int(SIM_TOKEN_IDX)
+    from pathlib import Path as _Path
+
+    output_base = _Path(SIM_OUTPUT) if SIM_OUTPUT else None
+    for rank, img in enumerate(top_images[:TOPK], start=1):
+        if output_base:
+            if output_base.suffix:
+                out_dir = output_base.parent
+                out_name = f"{output_base.stem}_rank{rank}{output_base.suffix}"
+                out_path = str(out_dir / out_name)
+            else:
+                out_dir = output_base
+                out_dir.mkdir(parents=True, exist_ok=True)
+                out_path = str(out_dir / f"similarity_map_rank{rank}.png")
+        else:
+            out_path = None
+        chosen_idx, max_sim = _generate_similarity_map(
+            model=model,
+            processor=processor,
+            image=img,
+            query=QUERY,
+            token_idx=token_idx,
+            output_path=out_path,
+        )
+        if out_path:
+            print(
+                f"Saved similarity map for rank {rank}, token #{chosen_idx} (max={max_sim:.2f}) to: {out_path}"
+            )
+        else:
+            print(
+                f"Computed similarity map for rank {rank}, token #{chosen_idx} (max={max_sim:.2f})"
+            )
+
+
+# %%
+# Step 6: Optional answer generation
+if results and ANSWER:
+    qwen = QwenVL(device=device_str)
+    response = qwen.answer(QUERY, top_images[:TOPK], max_new_tokens=MAX_NEW_TOKENS)
+    print("\nAnswer:")
+    print(response)

apps/multimodal/vision-based-pdf-multi-vector/multi-vector-leann.py

@@ -0,0 +1,134 @@
+# pip install pdf2image
+# pip install pymilvus
+# pip install colpali_engine
+# pip install tqdm
+# pip install pillow
+
+# %%
+from pdf2image import convert_from_path
+
+pdf_path = "pdfs/2004.12832v2.pdf"
+images = convert_from_path(pdf_path)
+
+for i, image in enumerate(images):
+    image.save(f"pages/page_{i + 1}.png", "PNG")
+
+# %%
+import os
+from pathlib import Path
+
+# Make local leann packages importable without installing
+_repo_root = Path(__file__).resolve().parents[3]
+_leann_core_src = _repo_root / "packages" / "leann-core" / "src"
+_leann_hnsw_pkg = _repo_root / "packages" / "leann-backend-hnsw"
+import sys
+
+if str(_leann_core_src) not in sys.path:
+    sys.path.append(str(_leann_core_src))
+if str(_leann_hnsw_pkg) not in sys.path:
+    sys.path.append(str(_leann_hnsw_pkg))
+
+from leann_multi_vector import LeannMultiVector
+
+
+class LeannRetriever(LeannMultiVector):
+    pass
+
+
+# %%
+from typing import cast
+
+import torch
+from colpali_engine.models import ColPali
+from colpali_engine.models.paligemma.colpali.processing_colpali import ColPaliProcessor
+from colpali_engine.utils.torch_utils import ListDataset, get_torch_device
+from torch.utils.data import DataLoader
+
+# Auto-select device: CUDA > MPS (mac) > CPU
+_device_str = (
+    "cuda"
+    if torch.cuda.is_available()
+    else (
+        "mps"
+        if getattr(torch.backends, "mps", None) and torch.backends.mps.is_available()
+        else "cpu"
+    )
+)
+device = get_torch_device(_device_str)
+# Prefer fp16 on GPU/MPS, bfloat16 on CPU
+_dtype = torch.float16 if _device_str in ("cuda", "mps") else torch.bfloat16
+model_name = "vidore/colpali-v1.2"
+
+model = ColPali.from_pretrained(
+    model_name,
+    torch_dtype=_dtype,
+    device_map=device,
+).eval()
+print(f"Using device={_device_str}, dtype={_dtype}")
+
+queries = [
+    "How to end-to-end retrieval with ColBert",
+    "Where is ColBERT performance Table, including text representation results?",
+]
+
+processor = cast(ColPaliProcessor, ColPaliProcessor.from_pretrained(model_name))
+
+dataloader = DataLoader(
+    dataset=ListDataset[str](queries),
+    batch_size=1,
+    shuffle=False,
+    collate_fn=lambda x: processor.process_queries(x),
+)
+
+qs: list[torch.Tensor] = []
+for batch_query in dataloader:
+    with torch.no_grad():
+        batch_query = {k: v.to(model.device) for k, v in batch_query.items()}
+        embeddings_query = model(**batch_query)
+    qs.extend(list(torch.unbind(embeddings_query.to("cpu"))))
+print(qs[0].shape)
+# %%
+
+
+import re
+
+from PIL import Image
+from tqdm import tqdm
+
+page_filenames = sorted(os.listdir("./pages"), key=lambda n: int(re.search(r"\d+", n).group()))
+images = [Image.open(os.path.join("./pages", name)) for name in page_filenames]
+
+dataloader = DataLoader(
+    dataset=ListDataset[str](images),
+    batch_size=1,
+    shuffle=False,
+    collate_fn=lambda x: processor.process_images(x),
+)
+
+ds: list[torch.Tensor] = []
+for batch_doc in tqdm(dataloader):
+    with torch.no_grad():
+        batch_doc = {k: v.to(model.device) for k, v in batch_doc.items()}
+        embeddings_doc = model(**batch_doc)
+    ds.extend(list(torch.unbind(embeddings_doc.to("cpu"))))
+
+print(ds[0].shape)
+
+# %%
+# Build HNSW index via LeannRetriever primitives and run search
+index_path = "./indexes/colpali.leann"
+retriever = LeannRetriever(index_path=index_path, dim=int(ds[0].shape[-1]))
+retriever.create_collection()
+filepaths = [os.path.join("./pages", name) for name in page_filenames]
+for i in range(len(filepaths)):
+    data = {
+        "colbert_vecs": ds[i].float().numpy(),
+        "doc_id": i,
+        "filepath": filepaths[i],
+    }
+    retriever.insert(data)
+retriever.create_index()
+for query in qs:
+    query_np = query.float().numpy()
+    result = retriever.search(query_np, topk=1)
+    print(filepaths[result[0][1]])

docs/configuration-guide.md

@@ -83,81 +83,6 @@ ollama pull nomic-embed-text
 
 </details>
 
-## Local & Remote Inference Endpoints
-
-> Applies to both LLMs (`leann ask`) and embeddings (`leann build`).
-
-LEANN now treats Ollama, LM Studio, and other OpenAI-compatible runtimes as first-class providers. You can point LEANN at any compatible endpoint – either on the same machine or across the network – with a couple of flags or environment variables.
-
-### One-Time Environment Setup
-
-```bash
-# Works for OpenAI-compatible runtimes such as LM Studio, vLLM, SGLang, llamafile, etc.
-export OPENAI_API_KEY="your-key"            # or leave unset for local servers that do not check keys
-export OPENAI_BASE_URL="http://localhost:1234/v1"
-
-# Ollama-compatible runtimes (Ollama, Ollama on another host, llamacpp-server, etc.)
-export LEANN_OLLAMA_HOST="http://localhost:11434"   # falls back to OLLAMA_HOST or LOCAL_LLM_ENDPOINT
-```
-
-LEANN also recognises `LEANN_LOCAL_LLM_HOST` (highest priority), `LEANN_OPENAI_BASE_URL`, and `LOCAL_OPENAI_BASE_URL`, so existing scripts continue to work.
-
-### Passing Hosts Per Command
-
-```bash
-# Build an index with a remote embedding server
-leann build my-notes \
-  --docs ./notes \
-  --embedding-mode openai \
-  --embedding-model text-embedding-qwen3-embedding-0.6b \
-  --embedding-api-base http://192.168.1.50:1234/v1 \
-  --embedding-api-key local-dev-key
-
-# Query using a local LM Studio instance via OpenAI-compatible API
-leann ask my-notes \
-  --llm openai \
-  --llm-model qwen3-8b \
-  --api-base http://localhost:1234/v1 \
-  --api-key local-dev-key
-
-# Query an Ollama instance running on another box
-leann ask my-notes \
-  --llm ollama \
-  --llm-model qwen3:14b \
-  --host http://192.168.1.101:11434
-```
-
-⚠️ **Make sure the endpoint is reachable**: when your inference server runs on a home/workstation and the index/search job runs in the cloud, the server must be able to reach the host you configured. Typical options include:
-
-- Expose a public IP (and open the relevant port) on the machine that hosts LM Studio/Ollama.
-- Configure router or cloud provider port forwarding.
-- Tunnel traffic through tools like `tailscale`, `cloudflared`, or `ssh -R`.
-
-When you set these options while building an index, LEANN stores them in `meta.json`. Any subsequent `leann ask` or searcher process automatically reuses the same provider settings – even when we spawn background embedding servers. This makes the “server without GPU talking to my local workstation” workflow from [issue #80](https://github.com/yichuan-w/LEANN/issues/80#issuecomment-2287230548) work out-of-the-box.
-
-**Tip:** If your runtime does not require an API key (many local stacks don’t), leave `--api-key` unset. LEANN will skip injecting credentials.
-
-### Python API Usage
-
-You can pass the same configuration from Python:
-
-```python
-from leann.api import LeannBuilder
-
-builder = LeannBuilder(
-    backend_name="hnsw",
-    embedding_mode="openai",
-    embedding_model="text-embedding-qwen3-embedding-0.6b",
-    embedding_options={
-        "base_url": "http://192.168.1.50:1234/v1",
-        "api_key": "local-dev-key",
-    },
-)
-builder.build_index("./indexes/my-notes", chunks)
-```
-
-`embedding_options` is persisted to the index `meta.json`, so subsequent `LeannSearcher` or `LeannChat` sessions automatically reuse the same provider settings (the embedding server manager forwards them to the provider for you).
-
 ## Index Selection: Matching Your Scale
 
 ### HNSW (Hierarchical Navigable Small World)

examples/dynamic_update_no_recompute.py

@@ -1,404 +0,0 @@
-"""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()

packages/leann-backend-diskann/leann_backend_diskann/diskann_embedding_server.py

@@ -10,7 +10,7 @@ import sys
 import threading
 import time
 from pathlib import Path
-from typing import Any, Optional
+from typing import Optional
 
 import numpy as np
 import zmq
@@ -32,16 +32,6 @@ if not logger.handlers:
     logger.propagate = False
 
 
-_RAW_PROVIDER_OPTIONS = os.getenv("LEANN_EMBEDDING_OPTIONS")
-try:
-    PROVIDER_OPTIONS: dict[str, Any] = (
-        json.loads(_RAW_PROVIDER_OPTIONS) if _RAW_PROVIDER_OPTIONS else {}
-    )
-except json.JSONDecodeError:
-    logger.warning("Failed to parse LEANN_EMBEDDING_OPTIONS; ignoring provider options")
-    PROVIDER_OPTIONS = {}
-
-
 def create_diskann_embedding_server(
     passages_file: Optional[str] = None,
     zmq_port: int = 5555,
@@ -191,12 +181,7 @@ def create_diskann_embedding_server(
                     logger.debug(f"Text lengths: {[len(t) for t in texts[:5]]}")  # Show first 5
 
                 # Process embeddings using unified computation
-                embeddings = compute_embeddings(
-                    texts,
-                    model_name,
-                    mode=embedding_mode,
-                    provider_options=PROVIDER_OPTIONS,
-                )
+                embeddings = compute_embeddings(texts, model_name, mode=embedding_mode)
                 logger.info(
                     f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
                 )
@@ -311,12 +296,7 @@ def create_diskann_embedding_server(
                             continue
 
                     # Process the request
-                    embeddings = compute_embeddings(
-                        texts,
-                        model_name,
-                        mode=embedding_mode,
-                        provider_options=PROVIDER_OPTIONS,
-                    )
+                    embeddings = compute_embeddings(texts, model_name, mode=embedding_mode)
                     logger.info(f"Computed embeddings shape: {embeddings.shape}")
 
                     # Validation

packages/leann-backend-hnsw/leann_backend_hnsw/convert_to_csr.py

@@ -5,8 +5,6 @@ import os
 import struct
 import sys
 import time
-from dataclasses import dataclass
-from typing import Any, Optional
 
 import numpy as np
 
@@ -239,288 +237,6 @@ def write_compact_format(
         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 ---
 
 
@@ -984,29 +700,6 @@ def convert_hnsw_graph_to_csr(input_filename, output_filename, prune_embeddings=
             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 ---
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(

packages/leann-backend-hnsw/leann_backend_hnsw/hnsw_backend.py

@@ -14,7 +14,7 @@ from leann.interface import (
 from leann.registry import register_backend
 from leann.searcher_base import BaseSearcher
 
-from .convert_to_csr import convert_hnsw_graph_to_csr, prune_hnsw_embeddings_inplace
+from .convert_to_csr import convert_hnsw_graph_to_csr
 
 logger = logging.getLogger(__name__)
 
@@ -92,8 +92,6 @@ class HNSWBuilder(LeannBackendBuilderInterface):
 
         if self.is_compact:
             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):
         """Convert built index to CSR format"""
@@ -135,10 +133,10 @@ class HNSWSearcher(BaseSearcher):
         if metric_enum is None:
             raise ValueError(f"Unsupported distance_metric '{self.distance_metric}'.")
 
-        backend_meta_kwargs = self.meta.get("backend_kwargs", {})
-        self.is_compact = self.meta.get("is_compact", backend_meta_kwargs.get("is_compact", True))
-        default_pruned = backend_meta_kwargs.get("is_recompute", self.is_compact)
-        self.is_pruned = bool(self.meta.get("is_pruned", default_pruned))
+        self.is_compact, self.is_pruned = (
+            self.meta.get("is_compact", True),
+            self.meta.get("is_pruned", True),
+        )
 
         index_file = self.index_dir / f"{self.index_path.stem}.index"
         if not index_file.exists():

packages/leann-backend-hnsw/leann_backend_hnsw/hnsw_embedding_server.py

@@ -10,7 +10,7 @@ import sys
 import threading
 import time
 from pathlib import Path
-from typing import Any, Optional
+from typing import Optional
 
 import msgpack
 import numpy as np
@@ -24,35 +24,13 @@ logger = logging.getLogger(__name__)
 log_level = getattr(logging, LOG_LEVEL, logging.WARNING)
 logger.setLevel(log_level)
 
-# Ensure we have handlers if none exist
+# Ensure we have a handler if none exists
 if not logger.handlers:
-    stream_handler = logging.StreamHandler()
+    handler = logging.StreamHandler()
     formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
-    stream_handler.setFormatter(formatter)
-    logger.addHandler(stream_handler)
-
-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
-
-_RAW_PROVIDER_OPTIONS = os.getenv("LEANN_EMBEDDING_OPTIONS")
-try:
-    PROVIDER_OPTIONS: dict[str, Any] = (
-        json.loads(_RAW_PROVIDER_OPTIONS) if _RAW_PROVIDER_OPTIONS else {}
-    )
-except json.JSONDecodeError:
-    logger.warning("Failed to parse LEANN_EMBEDDING_OPTIONS; ignoring provider options")
-    PROVIDER_OPTIONS = {}
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.propagate = False
 
 
 def create_hnsw_embedding_server(
@@ -160,12 +138,7 @@ def create_hnsw_embedding_server(
                     ):
                         last_request_type = "text"
                         last_request_length = len(request)
-                        embeddings = compute_embeddings(
-                            request,
-                            model_name,
-                            mode=embedding_mode,
-                            provider_options=PROVIDER_OPTIONS,
-                        )
+                        embeddings = compute_embeddings(request, model_name, mode=embedding_mode)
                         rep_socket.send(msgpack.packb(embeddings.tolist()))
                         e2e_end = time.time()
                         logger.info(f"⏱️  Text embedding E2E time: {e2e_end - e2e_start:.6f}s")
@@ -214,10 +187,7 @@ def create_hnsw_embedding_server(
                         if texts:
                             try:
                                 embeddings = compute_embeddings(
-                                    texts,
-                                    model_name,
-                                    mode=embedding_mode,
-                                    provider_options=PROVIDER_OPTIONS,
+                                    texts, model_name, mode=embedding_mode
                                 )
                                 logger.info(
                                     f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
@@ -282,12 +252,7 @@ def create_hnsw_embedding_server(
 
                     if texts:
                         try:
-                            embeddings = compute_embeddings(
-                                texts,
-                                model_name,
-                                mode=embedding_mode,
-                                provider_options=PROVIDER_OPTIONS,
-                            )
+                            embeddings = compute_embeddings(texts, model_name, mode=embedding_mode)
                             logger.info(
                                 f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
                             )

packages/leann-core/src/leann/api.py

@@ -15,7 +15,6 @@ from pathlib import Path
 from typing import Any, Literal, Optional, Union
 
 import numpy as np
-from leann_backend_hnsw.convert_to_csr import prune_hnsw_embeddings_inplace
 
 from leann.interface import LeannBackendSearcherInterface
 
@@ -39,7 +38,6 @@ def compute_embeddings(
     use_server: bool = True,
     port: Optional[int] = None,
     is_build=False,
-    provider_options: Optional[dict[str, Any]] = None,
 ) -> np.ndarray:
     """
     Computes embeddings using different backends.
@@ -73,7 +71,6 @@ def compute_embeddings(
             model_name,
             mode=mode,
             is_build=is_build,
-            provider_options=provider_options,
         )
 
 
@@ -280,7 +277,6 @@ class LeannBuilder:
         embedding_model: str = "facebook/contriever",
         dimensions: Optional[int] = None,
         embedding_mode: str = "sentence-transformers",
-        embedding_options: Optional[dict[str, Any]] = None,
         **backend_kwargs,
     ):
         self.backend_name = backend_name
@@ -303,7 +299,6 @@ class LeannBuilder:
         self.embedding_model = embedding_model
         self.dimensions = dimensions
         self.embedding_mode = embedding_mode
-        self.embedding_options = embedding_options or {}
 
         # Check if we need to use cosine distance for normalized embeddings
         normalized_embeddings_models = {
@@ -411,7 +406,6 @@ class LeannBuilder:
                     self.embedding_model,
                     self.embedding_mode,
                     use_server=False,
-                    provider_options=self.embedding_options,
                 )[0]
             )
         path = Path(index_path)
@@ -451,7 +445,6 @@ class LeannBuilder:
             self.embedding_mode,
             use_server=False,
             is_build=True,
-            provider_options=self.embedding_options,
         )
         string_ids = [chunk["id"] for chunk in self.chunks]
         current_backend_kwargs = {**self.backend_kwargs, "dimensions": self.dimensions}
@@ -478,15 +471,14 @@ class LeannBuilder:
             ],
         }
 
-        if self.embedding_options:
-            meta_data["embedding_options"] = self.embedding_options
-
         # Add storage status flags for HNSW backend
         if self.backend_name == "hnsw":
             is_compact = self.backend_kwargs.get("is_compact", True)
             is_recompute = self.backend_kwargs.get("is_recompute", True)
             meta_data["is_compact"] = is_compact
-            meta_data["is_pruned"] = bool(is_recompute)
+            meta_data["is_pruned"] = (
+                is_compact and is_recompute
+            )  # Pruned only if compact and recompute
         with open(leann_meta_path, "w", encoding="utf-8") as f:
             json.dump(meta_data, f, indent=2)
 
@@ -601,166 +593,18 @@ class LeannBuilder:
             "embeddings_source": str(embeddings_file),
         }
 
-        if self.embedding_options:
-            meta_data["embedding_options"] = self.embedding_options
-
         # Add storage status flags for HNSW backend
         if self.backend_name == "hnsw":
             is_compact = self.backend_kwargs.get("is_compact", True)
             is_recompute = self.backend_kwargs.get("is_recompute", True)
             meta_data["is_compact"] = is_compact
-            meta_data["is_pruned"] = bool(is_recompute)
+            meta_data["is_pruned"] = is_compact and is_recompute
 
         with open(leann_meta_path, "w", encoding="utf-8") as f:
             json.dump(meta_data, f, indent=2)
 
         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,
-            provider_options=self.embedding_options,
-        )
-
-        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:
     def __init__(self, index_path: str, enable_warmup: bool = False, **backend_kwargs):
@@ -784,7 +628,6 @@ class LeannSearcher:
         self.embedding_model = self.meta_data["embedding_model"]
         # Support both old and new format
         self.embedding_mode = self.meta_data.get("embedding_mode", "sentence-transformers")
-        self.embedding_options = self.meta_data.get("embedding_options", {})
         # Delegate portability handling to PassageManager
         self.passage_manager = PassageManager(
             self.meta_data.get("passage_sources", []), metadata_file_path=self.meta_path_str
@@ -796,8 +639,6 @@ class LeannSearcher:
             raise ValueError(f"Backend '{backend_name}' not found.")
         final_kwargs = {**self.meta_data.get("backend_kwargs", {}), **backend_kwargs}
         final_kwargs["enable_warmup"] = enable_warmup
-        if self.embedding_options:
-            final_kwargs.setdefault("embedding_options", self.embedding_options)
         self.backend_impl: LeannBackendSearcherInterface = backend_factory.searcher(
             index_path, **final_kwargs
         )

packages/leann-core/src/leann/chat.py

@@ -12,8 +12,6 @@ from typing import Any, Optional
 
 import torch
 
-from .settings import resolve_ollama_host, resolve_openai_api_key, resolve_openai_base_url
-
 # Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
@@ -312,12 +310,11 @@ def search_hf_models(query: str, limit: int = 10) -> list[str]:
 
 
 def validate_model_and_suggest(
-    model_name: str, llm_type: str, host: Optional[str] = None
+    model_name: str, llm_type: str, host: str = "http://localhost:11434"
 ) -> Optional[str]:
     """Validate model name and provide suggestions if invalid"""
     if llm_type == "ollama":
-        resolved_host = resolve_ollama_host(host)
-        available_models = check_ollama_models(resolved_host)
+        available_models = check_ollama_models(host)
         if available_models and model_name not in available_models:
             error_msg = f"Model '{model_name}' not found in your local Ollama installation."
 
@@ -460,19 +457,19 @@ class LLMInterface(ABC):
 class OllamaChat(LLMInterface):
     """LLM interface for Ollama models."""
 
-    def __init__(self, model: str = "llama3:8b", host: Optional[str] = None):
+    def __init__(self, model: str = "llama3:8b", host: str = "http://localhost:11434"):
         self.model = model
-        self.host = resolve_ollama_host(host)
-        logger.info(f"Initializing OllamaChat with model='{model}' and host='{self.host}'")
+        self.host = host
+        logger.info(f"Initializing OllamaChat with model='{model}' and host='{host}'")
         try:
             import requests
 
             # Check if the Ollama server is responsive
-            if self.host:
-                requests.get(self.host)
+            if host:
+                requests.get(host)
 
             # Pre-check model availability with helpful suggestions
-            model_error = validate_model_and_suggest(model, "ollama", self.host)
+            model_error = validate_model_and_suggest(model, "ollama", host)
             if model_error:
                 raise ValueError(model_error)
 
@@ -481,11 +478,9 @@ class OllamaChat(LLMInterface):
                 "The 'requests' library is required for Ollama. Please install it with 'pip install requests'."
             )
         except requests.exceptions.ConnectionError:
-            logger.error(
-                f"Could not connect to Ollama at {self.host}. Please ensure Ollama is running."
-            )
+            logger.error(f"Could not connect to Ollama at {host}. Please ensure Ollama is running.")
             raise ConnectionError(
-                f"Could not connect to Ollama at {self.host}. Please ensure Ollama is running."
+                f"Could not connect to Ollama at {host}. Please ensure Ollama is running."
             )
 
     def ask(self, prompt: str, **kwargs) -> str:
@@ -742,31 +737,21 @@ class GeminiChat(LLMInterface):
 class OpenAIChat(LLMInterface):
     """LLM interface for OpenAI models."""
 
-    def __init__(
-        self,
-        model: str = "gpt-4o",
-        api_key: Optional[str] = None,
-        base_url: Optional[str] = None,
-    ):
+    def __init__(self, model: str = "gpt-4o", api_key: Optional[str] = None):
         self.model = model
-        self.base_url = resolve_openai_base_url(base_url)
-        self.api_key = resolve_openai_api_key(api_key)
+        self.api_key = api_key or os.getenv("OPENAI_API_KEY")
 
         if not self.api_key:
             raise ValueError(
                 "OpenAI API key is required. Set OPENAI_API_KEY environment variable or pass api_key parameter."
             )
 
-        logger.info(
-            "Initializing OpenAI Chat with model='%s' and base_url='%s'",
-            model,
-            self.base_url,
-        )
+        logger.info(f"Initializing OpenAI Chat with model='{model}'")
 
         try:
             import openai
 
-            self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url)
+            self.client = openai.OpenAI(api_key=self.api_key)
         except ImportError:
             raise ImportError(
                 "The 'openai' library is required for OpenAI models. Please install it with 'pip install openai'."
@@ -856,16 +841,12 @@ def get_llm(llm_config: Optional[dict[str, Any]] = None) -> LLMInterface:
     if llm_type == "ollama":
         return OllamaChat(
             model=model or "llama3:8b",
-            host=llm_config.get("host"),
+            host=llm_config.get("host", "http://localhost:11434"),
         )
     elif llm_type == "hf":
         return HFChat(model_name=model or "deepseek-ai/deepseek-llm-7b-chat")
     elif llm_type == "openai":
-        return OpenAIChat(
-            model=model or "gpt-4o",
-            api_key=llm_config.get("api_key"),
-            base_url=llm_config.get("base_url"),
-        )
+        return OpenAIChat(model=model or "gpt-4o", api_key=llm_config.get("api_key"))
     elif llm_type == "gemini":
         return GeminiChat(model=model or "gemini-2.5-flash", api_key=llm_config.get("api_key"))
     elif llm_type == "simulated":

packages/leann-core/src/leann/cli.py

@@ -9,7 +9,6 @@ from tqdm import tqdm
 
 from .api import LeannBuilder, LeannChat, LeannSearcher
 from .registry import register_project_directory
-from .settings import resolve_ollama_host, resolve_openai_api_key, resolve_openai_base_url
 
 
 def extract_pdf_text_with_pymupdf(file_path: str) -> str:
@@ -124,24 +123,6 @@ Examples:
             choices=["sentence-transformers", "openai", "mlx", "ollama"],
             help="Embedding backend mode (default: sentence-transformers)",
         )
-        build_parser.add_argument(
-            "--embedding-host",
-            type=str,
-            default=None,
-            help="Override Ollama-compatible embedding host",
-        )
-        build_parser.add_argument(
-            "--embedding-api-base",
-            type=str,
-            default=None,
-            help="Base URL for OpenAI-compatible embedding services",
-        )
-        build_parser.add_argument(
-            "--embedding-api-key",
-            type=str,
-            default=None,
-            help="API key for embedding service (defaults to OPENAI_API_KEY)",
-        )
         build_parser.add_argument(
             "--force", "-f", action="store_true", help="Force rebuild existing index"
         )
@@ -267,12 +248,7 @@ Examples:
         ask_parser.add_argument(
             "--model", type=str, default="qwen3:8b", help="Model name (default: qwen3:8b)"
         )
-        ask_parser.add_argument(
-            "--host",
-            type=str,
-            default=None,
-            help="Override Ollama-compatible host (defaults to LEANN_OLLAMA_HOST/OLLAMA_HOST)",
-        )
+        ask_parser.add_argument("--host", type=str, default="http://localhost:11434")
         ask_parser.add_argument(
             "--interactive", "-i", action="store_true", help="Interactive chat mode"
         )
@@ -301,18 +277,6 @@ Examples:
             default=None,
             help="Thinking budget for reasoning models (low/medium/high). Supported by GPT-Oss:20b and other reasoning models.",
         )
-        ask_parser.add_argument(
-            "--api-base",
-            type=str,
-            default=None,
-            help="Base URL for OpenAI-compatible APIs (e.g., http://localhost:10000/v1)",
-        )
-        ask_parser.add_argument(
-            "--api-key",
-            type=str,
-            default=None,
-            help="API key for OpenAI-compatible APIs (defaults to OPENAI_API_KEY)",
-        )
 
         # List command
         subparsers.add_parser("list", help="List all indexes")
@@ -1361,20 +1325,10 @@ Examples:
 
         print(f"Building index '{index_name}' with {args.backend} backend...")
 
-        embedding_options: dict[str, Any] = {}
-        if args.embedding_mode == "ollama":
-            embedding_options["host"] = resolve_ollama_host(args.embedding_host)
-        elif args.embedding_mode == "openai":
-            embedding_options["base_url"] = resolve_openai_base_url(args.embedding_api_base)
-            resolved_embedding_key = resolve_openai_api_key(args.embedding_api_key)
-            if resolved_embedding_key:
-                embedding_options["api_key"] = resolved_embedding_key
-
         builder = LeannBuilder(
             backend_name=args.backend,
             embedding_model=args.embedding_model,
             embedding_mode=args.embedding_mode,
-            embedding_options=embedding_options or None,
             graph_degree=args.graph_degree,
             complexity=args.complexity,
             is_compact=args.compact,
@@ -1522,12 +1476,7 @@ Examples:
 
         llm_config = {"type": args.llm, "model": args.model}
         if args.llm == "ollama":
-            llm_config["host"] = resolve_ollama_host(args.host)
-        elif args.llm == "openai":
-            llm_config["base_url"] = resolve_openai_base_url(args.api_base)
-            resolved_api_key = resolve_openai_api_key(args.api_key)
-            if resolved_api_key:
-                llm_config["api_key"] = resolved_api_key
+            llm_config["host"] = args.host
 
         chat = LeannChat(index_path=index_path, llm_config=llm_config)
 

packages/leann-core/src/leann/embedding_compute.py

@@ -7,13 +7,11 @@ Preserves all optimization parameters to ensure performance
 import logging
 import os
 import time
-from typing import Any, Optional
+from typing import Any
 
 import numpy as np
 import torch
 
-from .settings import resolve_ollama_host, resolve_openai_api_key, resolve_openai_base_url
-
 # Set up logger with proper level
 logger = logging.getLogger(__name__)
 LOG_LEVEL = os.getenv("LEANN_LOG_LEVEL", "WARNING").upper()
@@ -33,7 +31,6 @@ def compute_embeddings(
     adaptive_optimization: bool = True,
     manual_tokenize: bool = False,
     max_length: int = 512,
-    provider_options: Optional[dict[str, Any]] = None,
 ) -> np.ndarray:
     """
     Unified embedding computation entry point
@@ -49,8 +46,6 @@ def compute_embeddings(
     Returns:
         Normalized embeddings array, shape: (len(texts), embedding_dim)
     """
-    provider_options = provider_options or {}
-
     if mode == "sentence-transformers":
         return compute_embeddings_sentence_transformers(
             texts,
@@ -62,21 +57,11 @@ def compute_embeddings(
             max_length=max_length,
         )
     elif mode == "openai":
-        return compute_embeddings_openai(
-            texts,
-            model_name,
-            base_url=provider_options.get("base_url"),
-            api_key=provider_options.get("api_key"),
-        )
+        return compute_embeddings_openai(texts, model_name)
     elif mode == "mlx":
         return compute_embeddings_mlx(texts, model_name)
     elif mode == "ollama":
-        return compute_embeddings_ollama(
-            texts,
-            model_name,
-            is_build=is_build,
-            host=provider_options.get("host"),
-        )
+        return compute_embeddings_ollama(texts, model_name, is_build=is_build)
     elif mode == "gemini":
         return compute_embeddings_gemini(texts, model_name, is_build=is_build)
     else:
@@ -368,15 +353,12 @@ def compute_embeddings_sentence_transformers(
     return embeddings
 
 
-def compute_embeddings_openai(
-    texts: list[str],
-    model_name: str,
-    base_url: Optional[str] = None,
-    api_key: Optional[str] = None,
-) -> np.ndarray:
+def compute_embeddings_openai(texts: list[str], model_name: str) -> np.ndarray:
     # TODO: @yichuan-w add progress bar only in build mode
     """Compute embeddings using OpenAI API"""
     try:
+        import os
+
         import openai
     except ImportError as e:
         raise ImportError(f"OpenAI package not installed: {e}")
@@ -391,18 +373,16 @@ def compute_embeddings_openai(
             f"Found {invalid_count} empty/invalid text(s) in input. Upstream should filter before calling OpenAI."
         )
 
-    resolved_base_url = resolve_openai_base_url(base_url)
-    resolved_api_key = resolve_openai_api_key(api_key)
-
-    if not resolved_api_key:
+    api_key = os.getenv("OPENAI_API_KEY")
+    if not api_key:
         raise RuntimeError("OPENAI_API_KEY environment variable not set")
 
     # Cache OpenAI client
-    cache_key = f"openai_client::{resolved_base_url}"
+    cache_key = "openai_client"
     if cache_key in _model_cache:
         client = _model_cache[cache_key]
     else:
-        client = openai.OpenAI(api_key=resolved_api_key, base_url=resolved_base_url)
+        client = openai.OpenAI(api_key=api_key)
         _model_cache[cache_key] = client
         logger.info("OpenAI client cached")
 
@@ -527,10 +507,7 @@ def compute_embeddings_mlx(chunks: list[str], model_name: str, batch_size: int =
 
 
 def compute_embeddings_ollama(
-    texts: list[str],
-    model_name: str,
-    is_build: bool = False,
-    host: Optional[str] = None,
+    texts: list[str], model_name: str, is_build: bool = False, host: str = "http://localhost:11434"
 ) -> np.ndarray:
     """
     Compute embeddings using Ollama API with simplified batch processing.
@@ -541,7 +518,7 @@ def compute_embeddings_ollama(
         texts: List of texts to compute embeddings for
         model_name: Ollama model name (e.g., "nomic-embed-text", "mxbai-embed-large")
         is_build: Whether this is a build operation (shows progress bar)
-        host: Ollama host URL (defaults to environment or http://localhost:11434)
+        host: Ollama host URL (default: http://localhost:11434)
 
     Returns:
         Normalized embeddings array, shape: (len(texts), embedding_dim)
@@ -556,19 +533,17 @@ def compute_embeddings_ollama(
     if not texts:
         raise ValueError("Cannot compute embeddings for empty text list")
 
-    resolved_host = resolve_ollama_host(host)
-
     logger.info(
-        f"Computing embeddings for {len(texts)} texts using Ollama API, model: '{model_name}', host: '{resolved_host}'"
+        f"Computing embeddings for {len(texts)} texts using Ollama API, model: '{model_name}'"
     )
 
     # Check if Ollama is running
     try:
-        response = requests.get(f"{resolved_host}/api/version", timeout=5)
+        response = requests.get(f"{host}/api/version", timeout=5)
         response.raise_for_status()
     except requests.exceptions.ConnectionError:
         error_msg = (
-            f"❌ Could not connect to Ollama at {resolved_host}.\n\n"
+            f"❌ Could not connect to Ollama at {host}.\n\n"
             "Please ensure Ollama is running:\n"
             "  • macOS/Linux: ollama serve\n"
             "  • Windows: Make sure Ollama is running in the system tray\n\n"
@@ -580,7 +555,7 @@ def compute_embeddings_ollama(
 
     # Check if model exists and provide helpful suggestions
     try:
-        response = requests.get(f"{resolved_host}/api/tags", timeout=5)
+        response = requests.get(f"{host}/api/tags", timeout=5)
         response.raise_for_status()
         models = response.json()
         model_names = [model["name"] for model in models.get("models", [])]
@@ -643,9 +618,7 @@ def compute_embeddings_ollama(
         # Verify the model supports embeddings by testing it
         try:
             test_response = requests.post(
-                f"{resolved_host}/api/embeddings",
-                json={"model": model_name, "prompt": "test"},
-                timeout=10,
+                f"{host}/api/embeddings", json={"model": model_name, "prompt": "test"}, timeout=10
             )
             if test_response.status_code != 200:
                 error_msg = (
@@ -692,7 +665,7 @@ def compute_embeddings_ollama(
             while retry_count < max_retries:
                 try:
                     response = requests.post(
-                        f"{resolved_host}/api/embeddings",
+                        f"{host}/api/embeddings",
                         json={"model": model_name, "prompt": truncated_text},
                         timeout=30,
                     )

packages/leann-core/src/leann/embedding_server_manager.py

@@ -8,8 +8,6 @@ import time
 from pathlib import Path
 from typing import Optional
 
-from .settings import encode_provider_options
-
 # Lightweight, self-contained server manager with no cross-process inspection
 
 # Set up logging based on environment variable
@@ -84,40 +82,16 @@ class EmbeddingServerManager:
     ) -> tuple[bool, int]:
         """Start the embedding server."""
         # passages_file may be present in kwargs for server CLI, but we don't need it here
-        provider_options = kwargs.pop("provider_options", None)
-
-        config_signature = {
-            "model_name": model_name,
-            "passages_file": kwargs.get("passages_file", ""),
-            "embedding_mode": embedding_mode,
-            "provider_options": provider_options or {},
-        }
 
         # If this manager already has a live server, just reuse it
-        if (
-            self.server_process
-            and self.server_process.poll() is None
-            and self.server_port
-            and self._server_config == config_signature
-        ):
+        if self.server_process and self.server_process.poll() is None and self.server_port:
             logger.info("Reusing in-process server")
             return True, self.server_port
 
-        # Configuration changed, stop existing server before starting a new one
-        if self.server_process and self.server_process.poll() is None:
-            logger.info("Existing server configuration differs; restarting embedding server")
-            self.stop_server()
-
         # For Colab environment, use a different strategy
         if _is_colab_environment():
             logger.info("Detected Colab environment, using alternative startup strategy")
-            return self._start_server_colab(
-                port,
-                model_name,
-                embedding_mode,
-                provider_options=provider_options,
-                **kwargs,
-            )
+            return self._start_server_colab(port, model_name, embedding_mode, **kwargs)
 
         # Always pick a fresh available port
         try:
@@ -127,21 +101,13 @@ class EmbeddingServerManager:
             return False, port
 
         # Start a new server
-        return self._start_new_server(
-            actual_port,
-            model_name,
-            embedding_mode,
-            provider_options=provider_options,
-            config_signature=config_signature,
-            **kwargs,
-        )
+        return self._start_new_server(actual_port, model_name, embedding_mode, **kwargs)
 
     def _start_server_colab(
         self,
         port: int,
         model_name: str,
         embedding_mode: str = "sentence-transformers",
-        provider_options: Optional[dict] = None,
         **kwargs,
     ) -> tuple[bool, int]:
         """Start server with Colab-specific configuration."""
@@ -159,20 +125,8 @@ class EmbeddingServerManager:
 
         try:
             # In Colab, we'll use a more direct approach
-            self._launch_server_process_colab(
-                command,
-                actual_port,
-                provider_options=provider_options,
-            )
-            started, ready_port = self._wait_for_server_ready_colab(actual_port)
-            if started:
-                self._server_config = {
-                    "model_name": model_name,
-                    "passages_file": kwargs.get("passages_file", ""),
-                    "embedding_mode": embedding_mode,
-                    "provider_options": provider_options or {},
-                }
-            return started, ready_port
+            self._launch_server_process_colab(command, actual_port)
+            return self._wait_for_server_ready_colab(actual_port)
         except Exception as e:
             logger.error(f"Failed to start embedding server in Colab: {e}")
             return False, actual_port
@@ -180,13 +134,7 @@ class EmbeddingServerManager:
     # Note: No compatibility check needed; manager is per-searcher and configs are stable per instance
 
     def _start_new_server(
-        self,
-        port: int,
-        model_name: str,
-        embedding_mode: str,
-        provider_options: Optional[dict] = None,
-        config_signature: Optional[dict] = None,
-        **kwargs,
+        self, port: int, model_name: str, embedding_mode: str, **kwargs
     ) -> tuple[bool, int]:
         """Start a new embedding server on the given port."""
         logger.info(f"Starting embedding server on port {port}...")
@@ -194,20 +142,8 @@ class EmbeddingServerManager:
         command = self._build_server_command(port, model_name, embedding_mode, **kwargs)
 
         try:
-            self._launch_server_process(
-                command,
-                port,
-                provider_options=provider_options,
-            )
-            started, ready_port = self._wait_for_server_ready(port)
-            if started:
-                self._server_config = config_signature or {
-                    "model_name": model_name,
-                    "passages_file": kwargs.get("passages_file", ""),
-                    "embedding_mode": embedding_mode,
-                    "provider_options": provider_options or {},
-                }
-            return started, ready_port
+            self._launch_server_process(command, port)
+            return self._wait_for_server_ready(port)
         except Exception as e:
             logger.error(f"Failed to start embedding server: {e}")
             return False, port
@@ -237,12 +173,7 @@ class EmbeddingServerManager:
 
         return command
 
-    def _launch_server_process(
-        self,
-        command: list,
-        port: int,
-        provider_options: Optional[dict] = None,
-    ) -> None:
+    def _launch_server_process(self, command: list, port: int) -> None:
         """Launch the server process."""
         project_root = Path(__file__).parent.parent.parent.parent.parent
         logger.info(f"Command: {' '.join(command)}")
@@ -262,20 +193,14 @@ class EmbeddingServerManager:
 
         # Start embedding server subprocess
         logger.info(f"Starting server process with command: {' '.join(command)}")
-        env = os.environ.copy()
-        encoded_options = encode_provider_options(provider_options)
-        if encoded_options:
-            env["LEANN_EMBEDDING_OPTIONS"] = encoded_options
-
         self.server_process = subprocess.Popen(
             command,
             cwd=project_root,
             stdout=stdout_target,
             stderr=stderr_target,
-            env=env,
         )
         self.server_port = port
-        # Record config for in-process reuse (best effort; refined later when ready)
+        # Record config for in-process reuse
         try:
             self._server_config = {
                 "model_name": command[command.index("--model-name") + 1]
@@ -287,14 +212,12 @@ class EmbeddingServerManager:
                 "embedding_mode": command[command.index("--embedding-mode") + 1]
                 if "--embedding-mode" in command
                 else "sentence-transformers",
-                "provider_options": provider_options or {},
             }
         except Exception:
             self._server_config = {
                 "model_name": "",
                 "passages_file": "",
                 "embedding_mode": "sentence-transformers",
-                "provider_options": provider_options or {},
             }
         logger.info(f"Server process started with PID: {self.server_process.pid}")
 
@@ -399,27 +322,16 @@ class EmbeddingServerManager:
         # Removed: cross-process adoption no longer supported
         return
 
-    def _launch_server_process_colab(
-        self,
-        command: list,
-        port: int,
-        provider_options: Optional[dict] = None,
-    ) -> None:
+    def _launch_server_process_colab(self, command: list, port: int) -> None:
         """Launch the server process with Colab-specific settings."""
         logger.info(f"Colab Command: {' '.join(command)}")
 
         # In Colab, we need to be more careful about process management
-        env = os.environ.copy()
-        encoded_options = encode_provider_options(provider_options)
-        if encoded_options:
-            env["LEANN_EMBEDDING_OPTIONS"] = encoded_options
-
         self.server_process = subprocess.Popen(
             command,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
-            env=env,
         )
         self.server_port = port
         logger.info(f"Colab server process started with PID: {self.server_process.pid}")
@@ -433,7 +345,6 @@ class EmbeddingServerManager:
             "model_name": "",
             "passages_file": "",
             "embedding_mode": "sentence-transformers",
-            "provider_options": provider_options or {},
         }
 
     def _wait_for_server_ready_colab(self, port: int) -> tuple[bool, int]:

packages/leann-core/src/leann/searcher_base.py

@@ -41,7 +41,6 @@ class BaseSearcher(LeannBackendSearcherInterface, ABC):
             print("WARNING: embedding_model not found in meta.json. Recompute will fail.")
 
         self.embedding_mode = self.meta.get("embedding_mode", "sentence-transformers")
-        self.embedding_options = self.meta.get("embedding_options", {})
 
         self.embedding_server_manager = EmbeddingServerManager(
             backend_module_name=backend_module_name,
@@ -78,7 +77,6 @@ class BaseSearcher(LeannBackendSearcherInterface, ABC):
             passages_file=passages_source_file,
             distance_metric=distance_metric,
             enable_warmup=kwargs.get("enable_warmup", False),
-            provider_options=self.embedding_options,
         )
         if not server_started:
             raise RuntimeError(f"Failed to start embedding server on port {actual_port}")
@@ -127,12 +125,7 @@ class BaseSearcher(LeannBackendSearcherInterface, ABC):
         from .embedding_compute import compute_embeddings
 
         embedding_mode = self.meta.get("embedding_mode", "sentence-transformers")
-        return compute_embeddings(
-            [query],
-            self.embedding_model,
-            embedding_mode,
-            provider_options=self.embedding_options,
-        )
+        return compute_embeddings([query], self.embedding_model, embedding_mode)
 
     def _compute_embedding_via_server(self, chunks: list, zmq_port: int) -> np.ndarray:
         """Compute embeddings using the ZMQ embedding server."""

packages/leann-core/src/leann/settings.py

@@ -1,74 +0,0 @@
-"""Runtime configuration helpers for LEANN."""
-
-from __future__ import annotations
-
-import json
-import os
-from typing import Any
-
-# Default fallbacks to preserve current behaviour while keeping them in one place.
-_DEFAULT_OLLAMA_HOST = "http://localhost:11434"
-_DEFAULT_OPENAI_BASE_URL = "https://api.openai.com/v1"
-
-
-def _clean_url(value: str) -> str:
-    """Normalize URL strings by stripping trailing slashes."""
-
-    return value.rstrip("/") if value else value
-
-
-def resolve_ollama_host(explicit: str | None = None) -> str:
-    """Resolve the Ollama-compatible endpoint to use."""
-
-    candidates = (
-        explicit,
-        os.getenv("LEANN_LOCAL_LLM_HOST"),
-        os.getenv("LEANN_OLLAMA_HOST"),
-        os.getenv("OLLAMA_HOST"),
-        os.getenv("LOCAL_LLM_ENDPOINT"),
-    )
-
-    for candidate in candidates:
-        if candidate:
-            return _clean_url(candidate)
-
-    return _clean_url(_DEFAULT_OLLAMA_HOST)
-
-
-def resolve_openai_base_url(explicit: str | None = None) -> str:
-    """Resolve the base URL for OpenAI-compatible services."""
-
-    candidates = (
-        explicit,
-        os.getenv("LEANN_OPENAI_BASE_URL"),
-        os.getenv("OPENAI_BASE_URL"),
-        os.getenv("LOCAL_OPENAI_BASE_URL"),
-    )
-
-    for candidate in candidates:
-        if candidate:
-            return _clean_url(candidate)
-
-    return _clean_url(_DEFAULT_OPENAI_BASE_URL)
-
-
-def resolve_openai_api_key(explicit: str | None = None) -> str | None:
-    """Resolve the API key for OpenAI-compatible services."""
-
-    if explicit:
-        return explicit
-
-    return os.getenv("OPENAI_API_KEY")
-
-
-def encode_provider_options(options: dict[str, Any] | None) -> str | None:
-    """Serialize provider options for child processes."""
-
-    if not options:
-        return None
-
-    try:
-        return json.dumps(options)
-    except (TypeError, ValueError):
-        # Fall back to empty payload if serialization fails
-        return None

pyproject.toml

@@ -104,7 +104,11 @@ astchunk = { path = "packages/astchunk-leann", editable = true }
 [tool.ruff]
 target-version = "py39"
 line-length = 100
-extend-exclude = ["third_party"]
+extend-exclude = [
+    "third_party",
+    "apps/multimodal/vision-based-pdf-multi-vector/multi-vector-leann.py",
+    "apps/multimodal/vision-based-pdf-multi-vector/multi-vector-leann-similarity-map.py"
+]
 
 
 [tool.ruff.lint]