Compare commits
3 Commits
fix/ask-cl
...
colqwen
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
df3350be43 | ||
|
|
94d9a203a2 | ||
|
|
72455bb269 |
4
.gitignore
vendored
4
.gitignore
vendored
@@ -18,6 +18,7 @@ demo/experiment_results/**/*.json
|
|||||||
*.eml
|
*.eml
|
||||||
*.emlx
|
*.emlx
|
||||||
*.json
|
*.json
|
||||||
|
*.png
|
||||||
!.vscode/*.json
|
!.vscode/*.json
|
||||||
*.sh
|
*.sh
|
||||||
*.txt
|
*.txt
|
||||||
@@ -101,3 +102,6 @@ CLAUDE.local.md
|
|||||||
.claude/*.local.*
|
.claude/*.local.*
|
||||||
.claude/local/*
|
.claude/local/*
|
||||||
benchmarks/data/
|
benchmarks/data/
|
||||||
|
|
||||||
|
## multi vector
|
||||||
|
apps/multimodal/vision-based-pdf-multi-vector/multi-vector-colpali-native-weaviate.py
|
||||||
|
|||||||
@@ -546,9 +546,6 @@ leann search my-docs "machine learning concepts"
|
|||||||
# Interactive chat with your documents
|
# Interactive chat with your documents
|
||||||
leann ask my-docs --interactive
|
leann ask my-docs --interactive
|
||||||
|
|
||||||
# Ask a single question (non-interactive)
|
|
||||||
leann ask my-docs "Where are prompts configured?"
|
|
||||||
|
|
||||||
# List all your indexes
|
# List all your indexes
|
||||||
leann list
|
leann list
|
||||||
|
|
||||||
|
|||||||
@@ -11,7 +11,6 @@ from typing import Any
|
|||||||
import dotenv
|
import dotenv
|
||||||
from leann.api import LeannBuilder, LeannChat
|
from leann.api import LeannBuilder, LeannChat
|
||||||
from leann.registry import register_project_directory
|
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()
|
dotenv.load_dotenv()
|
||||||
|
|
||||||
@@ -79,24 +78,6 @@ class BaseRAGExample(ABC):
|
|||||||
choices=["sentence-transformers", "openai", "mlx", "ollama"],
|
choices=["sentence-transformers", "openai", "mlx", "ollama"],
|
||||||
help="Embedding backend mode (default: sentence-transformers), we provide sentence-transformers, openai, mlx, or 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 parameters
|
||||||
llm_group = parser.add_argument_group("LLM Parameters")
|
llm_group = parser.add_argument_group("LLM Parameters")
|
||||||
@@ -116,8 +97,8 @@ class BaseRAGExample(ABC):
|
|||||||
llm_group.add_argument(
|
llm_group.add_argument(
|
||||||
"--llm-host",
|
"--llm-host",
|
||||||
type=str,
|
type=str,
|
||||||
default=None,
|
default="http://localhost:11434",
|
||||||
help="Host for Ollama-compatible APIs (defaults to LEANN_OLLAMA_HOST/OLLAMA_HOST)",
|
help="Host for Ollama API (default: http://localhost:11434)",
|
||||||
)
|
)
|
||||||
llm_group.add_argument(
|
llm_group.add_argument(
|
||||||
"--thinking-budget",
|
"--thinking-budget",
|
||||||
@@ -126,18 +107,6 @@ class BaseRAGExample(ABC):
|
|||||||
default=None,
|
default=None,
|
||||||
help="Thinking budget for reasoning models (low/medium/high). Supported by GPT-Oss:20b and other reasoning models.",
|
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 Chunking parameters
|
||||||
ast_group = parser.add_argument_group("AST Chunking Parameters")
|
ast_group = parser.add_argument_group("AST Chunking Parameters")
|
||||||
@@ -236,13 +205,9 @@ class BaseRAGExample(ABC):
|
|||||||
|
|
||||||
if args.llm == "openai":
|
if args.llm == "openai":
|
||||||
config["model"] = args.llm_model or "gpt-4o"
|
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":
|
elif args.llm == "ollama":
|
||||||
config["model"] = args.llm_model or "llama3.2:1b"
|
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":
|
elif args.llm == "hf":
|
||||||
config["model"] = args.llm_model or "Qwen/Qwen2.5-1.5B-Instruct"
|
config["model"] = args.llm_model or "Qwen/Qwen2.5-1.5B-Instruct"
|
||||||
elif args.llm == "simulated":
|
elif args.llm == "simulated":
|
||||||
@@ -258,20 +223,10 @@ class BaseRAGExample(ABC):
|
|||||||
print(f"\n[Building Index] Creating {self.name} index...")
|
print(f"\n[Building Index] Creating {self.name} index...")
|
||||||
print(f"Total text chunks: {len(texts)}")
|
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(
|
builder = LeannBuilder(
|
||||||
backend_name=args.backend_name,
|
backend_name=args.backend_name,
|
||||||
embedding_model=args.embedding_model,
|
embedding_model=args.embedding_model,
|
||||||
embedding_mode=args.embedding_mode,
|
embedding_mode=args.embedding_mode,
|
||||||
embedding_options=embedding_options or None,
|
|
||||||
graph_degree=args.graph_degree,
|
graph_degree=args.graph_degree,
|
||||||
complexity=args.build_complexity,
|
complexity=args.build_complexity,
|
||||||
is_compact=not args.no_compact,
|
is_compact=not args.no_compact,
|
||||||
|
|||||||
@@ -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
|
||||||
@@ -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)
|
||||||
@@ -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]])
|
||||||
@@ -83,81 +83,6 @@ ollama pull nomic-embed-text
|
|||||||
|
|
||||||
</details>
|
</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
|
## Index Selection: Matching Your Scale
|
||||||
|
|
||||||
### HNSW (Hierarchical Navigable Small World)
|
### HNSW (Hierarchical Navigable Small World)
|
||||||
|
|||||||
@@ -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()
|
|
||||||
@@ -10,7 +10,7 @@ import sys
|
|||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Optional
|
from typing import Optional
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import zmq
|
import zmq
|
||||||
@@ -32,16 +32,6 @@ if not logger.handlers:
|
|||||||
logger.propagate = False
|
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(
|
def create_diskann_embedding_server(
|
||||||
passages_file: Optional[str] = None,
|
passages_file: Optional[str] = None,
|
||||||
zmq_port: int = 5555,
|
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
|
logger.debug(f"Text lengths: {[len(t) for t in texts[:5]]}") # Show first 5
|
||||||
|
|
||||||
# Process embeddings using unified computation
|
# Process embeddings using unified computation
|
||||||
embeddings = compute_embeddings(
|
embeddings = compute_embeddings(texts, model_name, mode=embedding_mode)
|
||||||
texts,
|
|
||||||
model_name,
|
|
||||||
mode=embedding_mode,
|
|
||||||
provider_options=PROVIDER_OPTIONS,
|
|
||||||
)
|
|
||||||
logger.info(
|
logger.info(
|
||||||
f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
|
f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
|
||||||
)
|
)
|
||||||
@@ -311,12 +296,7 @@ def create_diskann_embedding_server(
|
|||||||
continue
|
continue
|
||||||
|
|
||||||
# Process the request
|
# Process the request
|
||||||
embeddings = compute_embeddings(
|
embeddings = compute_embeddings(texts, model_name, mode=embedding_mode)
|
||||||
texts,
|
|
||||||
model_name,
|
|
||||||
mode=embedding_mode,
|
|
||||||
provider_options=PROVIDER_OPTIONS,
|
|
||||||
)
|
|
||||||
logger.info(f"Computed embeddings shape: {embeddings.shape}")
|
logger.info(f"Computed embeddings shape: {embeddings.shape}")
|
||||||
|
|
||||||
# Validation
|
# Validation
|
||||||
|
|||||||
@@ -5,8 +5,6 @@ import os
|
|||||||
import struct
|
import struct
|
||||||
import sys
|
import sys
|
||||||
import time
|
import time
|
||||||
from dataclasses import dataclass
|
|
||||||
from typing import Any, Optional
|
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
|
|
||||||
@@ -239,288 +237,6 @@ def write_compact_format(
|
|||||||
f_out.write(storage_data)
|
f_out.write(storage_data)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
|
||||||
class HNSWComponents:
|
|
||||||
original_hnsw_data: dict[str, Any]
|
|
||||||
assign_probas_np: np.ndarray
|
|
||||||
cum_nneighbor_per_level_np: np.ndarray
|
|
||||||
levels_np: np.ndarray
|
|
||||||
is_compact: bool
|
|
||||||
compact_level_ptr: Optional[np.ndarray] = None
|
|
||||||
compact_node_offsets_np: Optional[np.ndarray] = None
|
|
||||||
compact_neighbors_data: Optional[list[int]] = None
|
|
||||||
offsets_np: Optional[np.ndarray] = None
|
|
||||||
neighbors_np: Optional[np.ndarray] = None
|
|
||||||
storage_fourcc: int = NULL_INDEX_FOURCC
|
|
||||||
storage_data: bytes = b""
|
|
||||||
|
|
||||||
|
|
||||||
def _read_hnsw_structure(f) -> HNSWComponents:
|
|
||||||
original_hnsw_data: dict[str, Any] = {}
|
|
||||||
|
|
||||||
hnsw_index_fourcc = read_struct(f, "<I")
|
|
||||||
if hnsw_index_fourcc not in EXPECTED_HNSW_FOURCCS:
|
|
||||||
raise ValueError(
|
|
||||||
f"Unexpected HNSW FourCC: {hnsw_index_fourcc:08x}. Expected one of {EXPECTED_HNSW_FOURCCS}."
|
|
||||||
)
|
|
||||||
|
|
||||||
original_hnsw_data["index_fourcc"] = hnsw_index_fourcc
|
|
||||||
original_hnsw_data["d"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["ntotal"] = read_struct(f, "<q")
|
|
||||||
original_hnsw_data["dummy1"] = read_struct(f, "<q")
|
|
||||||
original_hnsw_data["dummy2"] = read_struct(f, "<q")
|
|
||||||
original_hnsw_data["is_trained"] = read_struct(f, "?")
|
|
||||||
original_hnsw_data["metric_type"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["metric_arg"] = 0.0
|
|
||||||
if original_hnsw_data["metric_type"] > 1:
|
|
||||||
original_hnsw_data["metric_arg"] = read_struct(f, "<f")
|
|
||||||
|
|
||||||
assign_probas_np = read_numpy_vector(f, np.float64, "d")
|
|
||||||
cum_nneighbor_per_level_np = read_numpy_vector(f, np.int32, "i")
|
|
||||||
levels_np = read_numpy_vector(f, np.int32, "i")
|
|
||||||
|
|
||||||
ntotal = len(levels_np)
|
|
||||||
if ntotal != original_hnsw_data["ntotal"]:
|
|
||||||
original_hnsw_data["ntotal"] = ntotal
|
|
||||||
|
|
||||||
pos_before_compact = f.tell()
|
|
||||||
is_compact_flag = None
|
|
||||||
try:
|
|
||||||
is_compact_flag = read_struct(f, "<?")
|
|
||||||
except EOFError:
|
|
||||||
is_compact_flag = None
|
|
||||||
|
|
||||||
if is_compact_flag:
|
|
||||||
compact_level_ptr = read_numpy_vector(f, np.uint64, "Q")
|
|
||||||
compact_node_offsets_np = read_numpy_vector(f, np.uint64, "Q")
|
|
||||||
|
|
||||||
original_hnsw_data["entry_point"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["max_level"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["efConstruction"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["efSearch"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["dummy_upper_beam"] = read_struct(f, "<i")
|
|
||||||
|
|
||||||
storage_fourcc = read_struct(f, "<I")
|
|
||||||
compact_neighbors_data_np = read_numpy_vector(f, np.int32, "i")
|
|
||||||
compact_neighbors_data = compact_neighbors_data_np.tolist()
|
|
||||||
storage_data = f.read()
|
|
||||||
|
|
||||||
return HNSWComponents(
|
|
||||||
original_hnsw_data=original_hnsw_data,
|
|
||||||
assign_probas_np=assign_probas_np,
|
|
||||||
cum_nneighbor_per_level_np=cum_nneighbor_per_level_np,
|
|
||||||
levels_np=levels_np,
|
|
||||||
is_compact=True,
|
|
||||||
compact_level_ptr=compact_level_ptr,
|
|
||||||
compact_node_offsets_np=compact_node_offsets_np,
|
|
||||||
compact_neighbors_data=compact_neighbors_data,
|
|
||||||
storage_fourcc=storage_fourcc,
|
|
||||||
storage_data=storage_data,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Non-compact case
|
|
||||||
f.seek(pos_before_compact)
|
|
||||||
|
|
||||||
pos_before_probe = f.tell()
|
|
||||||
try:
|
|
||||||
suspected_flag = read_struct(f, "<B")
|
|
||||||
if suspected_flag != 0x00:
|
|
||||||
f.seek(pos_before_probe)
|
|
||||||
except EOFError:
|
|
||||||
f.seek(pos_before_probe)
|
|
||||||
|
|
||||||
offsets_np = read_numpy_vector(f, np.uint64, "Q")
|
|
||||||
neighbors_np = read_numpy_vector(f, np.int32, "i")
|
|
||||||
|
|
||||||
original_hnsw_data["entry_point"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["max_level"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["efConstruction"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["efSearch"] = read_struct(f, "<i")
|
|
||||||
original_hnsw_data["dummy_upper_beam"] = read_struct(f, "<i")
|
|
||||||
|
|
||||||
storage_fourcc = NULL_INDEX_FOURCC
|
|
||||||
storage_data = b""
|
|
||||||
try:
|
|
||||||
storage_fourcc = read_struct(f, "<I")
|
|
||||||
storage_data = f.read()
|
|
||||||
except EOFError:
|
|
||||||
storage_fourcc = NULL_INDEX_FOURCC
|
|
||||||
|
|
||||||
return HNSWComponents(
|
|
||||||
original_hnsw_data=original_hnsw_data,
|
|
||||||
assign_probas_np=assign_probas_np,
|
|
||||||
cum_nneighbor_per_level_np=cum_nneighbor_per_level_np,
|
|
||||||
levels_np=levels_np,
|
|
||||||
is_compact=False,
|
|
||||||
offsets_np=offsets_np,
|
|
||||||
neighbors_np=neighbors_np,
|
|
||||||
storage_fourcc=storage_fourcc,
|
|
||||||
storage_data=storage_data,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def _read_hnsw_structure_from_file(path: str) -> HNSWComponents:
|
|
||||||
with open(path, "rb") as f:
|
|
||||||
return _read_hnsw_structure(f)
|
|
||||||
|
|
||||||
|
|
||||||
def write_original_format(
|
|
||||||
f_out,
|
|
||||||
original_hnsw_data,
|
|
||||||
assign_probas_np,
|
|
||||||
cum_nneighbor_per_level_np,
|
|
||||||
levels_np,
|
|
||||||
offsets_np,
|
|
||||||
neighbors_np,
|
|
||||||
storage_fourcc,
|
|
||||||
storage_data,
|
|
||||||
):
|
|
||||||
"""Write non-compact HNSW data in original FAISS order."""
|
|
||||||
|
|
||||||
f_out.write(struct.pack("<I", original_hnsw_data["index_fourcc"]))
|
|
||||||
f_out.write(struct.pack("<i", original_hnsw_data["d"]))
|
|
||||||
f_out.write(struct.pack("<q", original_hnsw_data["ntotal"]))
|
|
||||||
f_out.write(struct.pack("<q", original_hnsw_data["dummy1"]))
|
|
||||||
f_out.write(struct.pack("<q", original_hnsw_data["dummy2"]))
|
|
||||||
f_out.write(struct.pack("<?", original_hnsw_data["is_trained"]))
|
|
||||||
f_out.write(struct.pack("<i", original_hnsw_data["metric_type"]))
|
|
||||||
if original_hnsw_data["metric_type"] > 1:
|
|
||||||
f_out.write(struct.pack("<f", original_hnsw_data["metric_arg"]))
|
|
||||||
|
|
||||||
write_numpy_vector(f_out, assign_probas_np, "d")
|
|
||||||
write_numpy_vector(f_out, cum_nneighbor_per_level_np, "i")
|
|
||||||
write_numpy_vector(f_out, levels_np, "i")
|
|
||||||
|
|
||||||
write_numpy_vector(f_out, offsets_np, "Q")
|
|
||||||
write_numpy_vector(f_out, neighbors_np, "i")
|
|
||||||
|
|
||||||
f_out.write(struct.pack("<i", original_hnsw_data["entry_point"]))
|
|
||||||
f_out.write(struct.pack("<i", original_hnsw_data["max_level"]))
|
|
||||||
f_out.write(struct.pack("<i", original_hnsw_data["efConstruction"]))
|
|
||||||
f_out.write(struct.pack("<i", original_hnsw_data["efSearch"]))
|
|
||||||
f_out.write(struct.pack("<i", original_hnsw_data["dummy_upper_beam"]))
|
|
||||||
|
|
||||||
f_out.write(struct.pack("<I", storage_fourcc))
|
|
||||||
if storage_fourcc != NULL_INDEX_FOURCC and storage_data:
|
|
||||||
f_out.write(storage_data)
|
|
||||||
|
|
||||||
|
|
||||||
def prune_hnsw_embeddings(input_filename: str, output_filename: str) -> bool:
|
|
||||||
"""Rewrite an HNSW index while dropping the embedded storage section."""
|
|
||||||
|
|
||||||
start_time = time.time()
|
|
||||||
try:
|
|
||||||
with open(input_filename, "rb") as f_in, open(output_filename, "wb") as f_out:
|
|
||||||
original_hnsw_data: dict[str, Any] = {}
|
|
||||||
|
|
||||||
hnsw_index_fourcc = read_struct(f_in, "<I")
|
|
||||||
if hnsw_index_fourcc not in EXPECTED_HNSW_FOURCCS:
|
|
||||||
print(
|
|
||||||
f"Error: Expected HNSW Index FourCC ({list(EXPECTED_HNSW_FOURCCS)}), got {hnsw_index_fourcc:08x}.",
|
|
||||||
file=sys.stderr,
|
|
||||||
)
|
|
||||||
return False
|
|
||||||
|
|
||||||
original_hnsw_data["index_fourcc"] = hnsw_index_fourcc
|
|
||||||
original_hnsw_data["d"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["ntotal"] = read_struct(f_in, "<q")
|
|
||||||
original_hnsw_data["dummy1"] = read_struct(f_in, "<q")
|
|
||||||
original_hnsw_data["dummy2"] = read_struct(f_in, "<q")
|
|
||||||
original_hnsw_data["is_trained"] = read_struct(f_in, "?")
|
|
||||||
original_hnsw_data["metric_type"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["metric_arg"] = 0.0
|
|
||||||
if original_hnsw_data["metric_type"] > 1:
|
|
||||||
original_hnsw_data["metric_arg"] = read_struct(f_in, "<f")
|
|
||||||
|
|
||||||
assign_probas_np = read_numpy_vector(f_in, np.float64, "d")
|
|
||||||
cum_nneighbor_per_level_np = read_numpy_vector(f_in, np.int32, "i")
|
|
||||||
levels_np = read_numpy_vector(f_in, np.int32, "i")
|
|
||||||
|
|
||||||
ntotal = len(levels_np)
|
|
||||||
if ntotal != original_hnsw_data["ntotal"]:
|
|
||||||
original_hnsw_data["ntotal"] = ntotal
|
|
||||||
|
|
||||||
pos_before_compact = f_in.tell()
|
|
||||||
is_compact_flag = None
|
|
||||||
try:
|
|
||||||
is_compact_flag = read_struct(f_in, "<?")
|
|
||||||
except EOFError:
|
|
||||||
is_compact_flag = None
|
|
||||||
|
|
||||||
if is_compact_flag:
|
|
||||||
compact_level_ptr = read_numpy_vector(f_in, np.uint64, "Q")
|
|
||||||
compact_node_offsets_np = read_numpy_vector(f_in, np.uint64, "Q")
|
|
||||||
|
|
||||||
original_hnsw_data["entry_point"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["max_level"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["efConstruction"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["efSearch"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["dummy_upper_beam"] = read_struct(f_in, "<i")
|
|
||||||
|
|
||||||
_storage_fourcc = read_struct(f_in, "<I")
|
|
||||||
compact_neighbors_data_np = read_numpy_vector(f_in, np.int32, "i")
|
|
||||||
compact_neighbors_data = compact_neighbors_data_np.tolist()
|
|
||||||
_storage_data = f_in.read()
|
|
||||||
|
|
||||||
write_compact_format(
|
|
||||||
f_out,
|
|
||||||
original_hnsw_data,
|
|
||||||
assign_probas_np,
|
|
||||||
cum_nneighbor_per_level_np,
|
|
||||||
levels_np,
|
|
||||||
compact_level_ptr,
|
|
||||||
compact_node_offsets_np,
|
|
||||||
compact_neighbors_data,
|
|
||||||
NULL_INDEX_FOURCC,
|
|
||||||
b"",
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
f_in.seek(pos_before_compact)
|
|
||||||
|
|
||||||
pos_before_probe = f_in.tell()
|
|
||||||
try:
|
|
||||||
suspected_flag = read_struct(f_in, "<B")
|
|
||||||
if suspected_flag != 0x00:
|
|
||||||
f_in.seek(pos_before_probe)
|
|
||||||
except EOFError:
|
|
||||||
f_in.seek(pos_before_probe)
|
|
||||||
|
|
||||||
offsets_np = read_numpy_vector(f_in, np.uint64, "Q")
|
|
||||||
neighbors_np = read_numpy_vector(f_in, np.int32, "i")
|
|
||||||
|
|
||||||
original_hnsw_data["entry_point"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["max_level"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["efConstruction"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["efSearch"] = read_struct(f_in, "<i")
|
|
||||||
original_hnsw_data["dummy_upper_beam"] = read_struct(f_in, "<i")
|
|
||||||
|
|
||||||
_storage_fourcc = None
|
|
||||||
_storage_data = b""
|
|
||||||
try:
|
|
||||||
_storage_fourcc = read_struct(f_in, "<I")
|
|
||||||
_storage_data = f_in.read()
|
|
||||||
except EOFError:
|
|
||||||
_storage_fourcc = NULL_INDEX_FOURCC
|
|
||||||
|
|
||||||
write_original_format(
|
|
||||||
f_out,
|
|
||||||
original_hnsw_data,
|
|
||||||
assign_probas_np,
|
|
||||||
cum_nneighbor_per_level_np,
|
|
||||||
levels_np,
|
|
||||||
offsets_np,
|
|
||||||
neighbors_np,
|
|
||||||
NULL_INDEX_FOURCC,
|
|
||||||
b"",
|
|
||||||
)
|
|
||||||
|
|
||||||
print(f"[{time.time() - start_time:.2f}s] Pruned embeddings from {input_filename}")
|
|
||||||
return True
|
|
||||||
except Exception as exc:
|
|
||||||
print(f"Failed to prune embeddings: {exc}", file=sys.stderr)
|
|
||||||
return False
|
|
||||||
|
|
||||||
|
|
||||||
# --- Main Conversion Logic ---
|
# --- Main Conversion Logic ---
|
||||||
|
|
||||||
|
|
||||||
@@ -984,29 +700,6 @@ def convert_hnsw_graph_to_csr(input_filename, output_filename, prune_embeddings=
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
def prune_hnsw_embeddings_inplace(index_filename: str) -> bool:
|
|
||||||
"""Convenience wrapper to prune embeddings in-place."""
|
|
||||||
|
|
||||||
temp_path = f"{index_filename}.prune.tmp"
|
|
||||||
success = prune_hnsw_embeddings(index_filename, temp_path)
|
|
||||||
if success:
|
|
||||||
try:
|
|
||||||
os.replace(temp_path, index_filename)
|
|
||||||
except Exception as exc: # pragma: no cover - defensive
|
|
||||||
logger.error(f"Failed to replace original index with pruned version: {exc}")
|
|
||||||
try:
|
|
||||||
os.remove(temp_path)
|
|
||||||
except OSError:
|
|
||||||
pass
|
|
||||||
return False
|
|
||||||
else:
|
|
||||||
try:
|
|
||||||
os.remove(temp_path)
|
|
||||||
except OSError:
|
|
||||||
pass
|
|
||||||
return success
|
|
||||||
|
|
||||||
|
|
||||||
# --- Script Execution ---
|
# --- Script Execution ---
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
parser = argparse.ArgumentParser(
|
parser = argparse.ArgumentParser(
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ from leann.interface import (
|
|||||||
from leann.registry import register_backend
|
from leann.registry import register_backend
|
||||||
from leann.searcher_base import BaseSearcher
|
from leann.searcher_base import BaseSearcher
|
||||||
|
|
||||||
from .convert_to_csr import convert_hnsw_graph_to_csr, prune_hnsw_embeddings_inplace
|
from .convert_to_csr import convert_hnsw_graph_to_csr
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -92,8 +92,6 @@ class HNSWBuilder(LeannBackendBuilderInterface):
|
|||||||
|
|
||||||
if self.is_compact:
|
if self.is_compact:
|
||||||
self._convert_to_csr(index_file)
|
self._convert_to_csr(index_file)
|
||||||
elif self.is_recompute:
|
|
||||||
prune_hnsw_embeddings_inplace(str(index_file))
|
|
||||||
|
|
||||||
def _convert_to_csr(self, index_file: Path):
|
def _convert_to_csr(self, index_file: Path):
|
||||||
"""Convert built index to CSR format"""
|
"""Convert built index to CSR format"""
|
||||||
@@ -135,10 +133,10 @@ class HNSWSearcher(BaseSearcher):
|
|||||||
if metric_enum is None:
|
if metric_enum is None:
|
||||||
raise ValueError(f"Unsupported distance_metric '{self.distance_metric}'.")
|
raise ValueError(f"Unsupported distance_metric '{self.distance_metric}'.")
|
||||||
|
|
||||||
backend_meta_kwargs = self.meta.get("backend_kwargs", {})
|
self.is_compact, self.is_pruned = (
|
||||||
self.is_compact = self.meta.get("is_compact", backend_meta_kwargs.get("is_compact", True))
|
self.meta.get("is_compact", True),
|
||||||
default_pruned = backend_meta_kwargs.get("is_recompute", self.is_compact)
|
self.meta.get("is_pruned", True),
|
||||||
self.is_pruned = bool(self.meta.get("is_pruned", default_pruned))
|
)
|
||||||
|
|
||||||
index_file = self.index_dir / f"{self.index_path.stem}.index"
|
index_file = self.index_dir / f"{self.index_path.stem}.index"
|
||||||
if not index_file.exists():
|
if not index_file.exists():
|
||||||
|
|||||||
@@ -10,7 +10,7 @@ import sys
|
|||||||
import threading
|
import threading
|
||||||
import time
|
import time
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Any, Optional
|
from typing import Optional
|
||||||
|
|
||||||
import msgpack
|
import msgpack
|
||||||
import numpy as np
|
import numpy as np
|
||||||
@@ -24,35 +24,13 @@ logger = logging.getLogger(__name__)
|
|||||||
log_level = getattr(logging, LOG_LEVEL, logging.WARNING)
|
log_level = getattr(logging, LOG_LEVEL, logging.WARNING)
|
||||||
logger.setLevel(log_level)
|
logger.setLevel(log_level)
|
||||||
|
|
||||||
# Ensure we have handlers if none exist
|
# Ensure we have a handler if none exists
|
||||||
if not logger.handlers:
|
if not logger.handlers:
|
||||||
stream_handler = logging.StreamHandler()
|
handler = logging.StreamHandler()
|
||||||
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
||||||
stream_handler.setFormatter(formatter)
|
handler.setFormatter(formatter)
|
||||||
logger.addHandler(stream_handler)
|
logger.addHandler(handler)
|
||||||
|
logger.propagate = False
|
||||||
log_path = os.getenv("LEANN_HNSW_LOG_PATH")
|
|
||||||
if log_path:
|
|
||||||
try:
|
|
||||||
file_handler = logging.FileHandler(log_path, mode="a", encoding="utf-8")
|
|
||||||
file_formatter = logging.Formatter(
|
|
||||||
"%(asctime)s - %(levelname)s - [pid=%(process)d] %(message)s"
|
|
||||||
)
|
|
||||||
file_handler.setFormatter(file_formatter)
|
|
||||||
logger.addHandler(file_handler)
|
|
||||||
except Exception as exc: # pragma: no cover - best effort logging
|
|
||||||
logger.warning(f"Failed to attach file handler for log path {log_path}: {exc}")
|
|
||||||
|
|
||||||
logger.propagate = False
|
|
||||||
|
|
||||||
_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_hnsw_embedding_server(
|
def create_hnsw_embedding_server(
|
||||||
@@ -160,12 +138,7 @@ def create_hnsw_embedding_server(
|
|||||||
):
|
):
|
||||||
last_request_type = "text"
|
last_request_type = "text"
|
||||||
last_request_length = len(request)
|
last_request_length = len(request)
|
||||||
embeddings = compute_embeddings(
|
embeddings = compute_embeddings(request, model_name, mode=embedding_mode)
|
||||||
request,
|
|
||||||
model_name,
|
|
||||||
mode=embedding_mode,
|
|
||||||
provider_options=PROVIDER_OPTIONS,
|
|
||||||
)
|
|
||||||
rep_socket.send(msgpack.packb(embeddings.tolist()))
|
rep_socket.send(msgpack.packb(embeddings.tolist()))
|
||||||
e2e_end = time.time()
|
e2e_end = time.time()
|
||||||
logger.info(f"⏱️ Text embedding E2E time: {e2e_end - e2e_start:.6f}s")
|
logger.info(f"⏱️ Text embedding E2E time: {e2e_end - e2e_start:.6f}s")
|
||||||
@@ -214,10 +187,7 @@ def create_hnsw_embedding_server(
|
|||||||
if texts:
|
if texts:
|
||||||
try:
|
try:
|
||||||
embeddings = compute_embeddings(
|
embeddings = compute_embeddings(
|
||||||
texts,
|
texts, model_name, mode=embedding_mode
|
||||||
model_name,
|
|
||||||
mode=embedding_mode,
|
|
||||||
provider_options=PROVIDER_OPTIONS,
|
|
||||||
)
|
)
|
||||||
logger.info(
|
logger.info(
|
||||||
f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
|
f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
|
||||||
@@ -282,12 +252,7 @@ def create_hnsw_embedding_server(
|
|||||||
|
|
||||||
if texts:
|
if texts:
|
||||||
try:
|
try:
|
||||||
embeddings = compute_embeddings(
|
embeddings = compute_embeddings(texts, model_name, mode=embedding_mode)
|
||||||
texts,
|
|
||||||
model_name,
|
|
||||||
mode=embedding_mode,
|
|
||||||
provider_options=PROVIDER_OPTIONS,
|
|
||||||
)
|
|
||||||
logger.info(
|
logger.info(
|
||||||
f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
|
f"Computed embeddings for {len(texts)} texts, shape: {embeddings.shape}"
|
||||||
)
|
)
|
||||||
|
|||||||
Submodule packages/leann-backend-hnsw/third_party/faiss updated: 1d51f0c074...ed96ff7dba
@@ -15,7 +15,6 @@ from pathlib import Path
|
|||||||
from typing import Any, Literal, Optional, Union
|
from typing import Any, Literal, Optional, Union
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
from leann_backend_hnsw.convert_to_csr import prune_hnsw_embeddings_inplace
|
|
||||||
|
|
||||||
from leann.interface import LeannBackendSearcherInterface
|
from leann.interface import LeannBackendSearcherInterface
|
||||||
|
|
||||||
@@ -39,7 +38,6 @@ def compute_embeddings(
|
|||||||
use_server: bool = True,
|
use_server: bool = True,
|
||||||
port: Optional[int] = None,
|
port: Optional[int] = None,
|
||||||
is_build=False,
|
is_build=False,
|
||||||
provider_options: Optional[dict[str, Any]] = None,
|
|
||||||
) -> np.ndarray:
|
) -> np.ndarray:
|
||||||
"""
|
"""
|
||||||
Computes embeddings using different backends.
|
Computes embeddings using different backends.
|
||||||
@@ -73,7 +71,6 @@ def compute_embeddings(
|
|||||||
model_name,
|
model_name,
|
||||||
mode=mode,
|
mode=mode,
|
||||||
is_build=is_build,
|
is_build=is_build,
|
||||||
provider_options=provider_options,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -280,7 +277,6 @@ class LeannBuilder:
|
|||||||
embedding_model: str = "facebook/contriever",
|
embedding_model: str = "facebook/contriever",
|
||||||
dimensions: Optional[int] = None,
|
dimensions: Optional[int] = None,
|
||||||
embedding_mode: str = "sentence-transformers",
|
embedding_mode: str = "sentence-transformers",
|
||||||
embedding_options: Optional[dict[str, Any]] = None,
|
|
||||||
**backend_kwargs,
|
**backend_kwargs,
|
||||||
):
|
):
|
||||||
self.backend_name = backend_name
|
self.backend_name = backend_name
|
||||||
@@ -303,7 +299,6 @@ class LeannBuilder:
|
|||||||
self.embedding_model = embedding_model
|
self.embedding_model = embedding_model
|
||||||
self.dimensions = dimensions
|
self.dimensions = dimensions
|
||||||
self.embedding_mode = embedding_mode
|
self.embedding_mode = embedding_mode
|
||||||
self.embedding_options = embedding_options or {}
|
|
||||||
|
|
||||||
# Check if we need to use cosine distance for normalized embeddings
|
# Check if we need to use cosine distance for normalized embeddings
|
||||||
normalized_embeddings_models = {
|
normalized_embeddings_models = {
|
||||||
@@ -411,7 +406,6 @@ class LeannBuilder:
|
|||||||
self.embedding_model,
|
self.embedding_model,
|
||||||
self.embedding_mode,
|
self.embedding_mode,
|
||||||
use_server=False,
|
use_server=False,
|
||||||
provider_options=self.embedding_options,
|
|
||||||
)[0]
|
)[0]
|
||||||
)
|
)
|
||||||
path = Path(index_path)
|
path = Path(index_path)
|
||||||
@@ -451,7 +445,6 @@ class LeannBuilder:
|
|||||||
self.embedding_mode,
|
self.embedding_mode,
|
||||||
use_server=False,
|
use_server=False,
|
||||||
is_build=True,
|
is_build=True,
|
||||||
provider_options=self.embedding_options,
|
|
||||||
)
|
)
|
||||||
string_ids = [chunk["id"] for chunk in self.chunks]
|
string_ids = [chunk["id"] for chunk in self.chunks]
|
||||||
current_backend_kwargs = {**self.backend_kwargs, "dimensions": self.dimensions}
|
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
|
# Add storage status flags for HNSW backend
|
||||||
if self.backend_name == "hnsw":
|
if self.backend_name == "hnsw":
|
||||||
is_compact = self.backend_kwargs.get("is_compact", True)
|
is_compact = self.backend_kwargs.get("is_compact", True)
|
||||||
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
||||||
meta_data["is_compact"] = is_compact
|
meta_data["is_compact"] = is_compact
|
||||||
meta_data["is_pruned"] = 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:
|
with open(leann_meta_path, "w", encoding="utf-8") as f:
|
||||||
json.dump(meta_data, f, indent=2)
|
json.dump(meta_data, f, indent=2)
|
||||||
|
|
||||||
@@ -601,166 +593,18 @@ class LeannBuilder:
|
|||||||
"embeddings_source": str(embeddings_file),
|
"embeddings_source": str(embeddings_file),
|
||||||
}
|
}
|
||||||
|
|
||||||
if self.embedding_options:
|
|
||||||
meta_data["embedding_options"] = self.embedding_options
|
|
||||||
|
|
||||||
# Add storage status flags for HNSW backend
|
# Add storage status flags for HNSW backend
|
||||||
if self.backend_name == "hnsw":
|
if self.backend_name == "hnsw":
|
||||||
is_compact = self.backend_kwargs.get("is_compact", True)
|
is_compact = self.backend_kwargs.get("is_compact", True)
|
||||||
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
is_recompute = self.backend_kwargs.get("is_recompute", True)
|
||||||
meta_data["is_compact"] = is_compact
|
meta_data["is_compact"] = is_compact
|
||||||
meta_data["is_pruned"] = bool(is_recompute)
|
meta_data["is_pruned"] = is_compact and is_recompute
|
||||||
|
|
||||||
with open(leann_meta_path, "w", encoding="utf-8") as f:
|
with open(leann_meta_path, "w", encoding="utf-8") as f:
|
||||||
json.dump(meta_data, f, indent=2)
|
json.dump(meta_data, f, indent=2)
|
||||||
|
|
||||||
logger.info(f"Index built successfully from precomputed embeddings: {index_path}")
|
logger.info(f"Index built successfully from precomputed embeddings: {index_path}")
|
||||||
|
|
||||||
def update_index(self, index_path: str):
|
|
||||||
"""Append new passages and vectors to an existing HNSW index."""
|
|
||||||
if not self.chunks:
|
|
||||||
raise ValueError("No new chunks provided for update.")
|
|
||||||
|
|
||||||
path = Path(index_path)
|
|
||||||
index_dir = path.parent
|
|
||||||
index_name = path.name
|
|
||||||
index_prefix = path.stem
|
|
||||||
|
|
||||||
meta_path = index_dir / f"{index_name}.meta.json"
|
|
||||||
passages_file = index_dir / f"{index_name}.passages.jsonl"
|
|
||||||
offset_file = index_dir / f"{index_name}.passages.idx"
|
|
||||||
index_file = index_dir / f"{index_prefix}.index"
|
|
||||||
|
|
||||||
if not meta_path.exists() or not passages_file.exists() or not offset_file.exists():
|
|
||||||
raise FileNotFoundError("Index metadata or passage files are missing; cannot update.")
|
|
||||||
if not index_file.exists():
|
|
||||||
raise FileNotFoundError(f"HNSW index file not found: {index_file}")
|
|
||||||
|
|
||||||
with open(meta_path, encoding="utf-8") as f:
|
|
||||||
meta = json.load(f)
|
|
||||||
backend_name = meta.get("backend_name")
|
|
||||||
if backend_name != self.backend_name:
|
|
||||||
raise ValueError(
|
|
||||||
f"Index was built with backend '{backend_name}', cannot update with '{self.backend_name}'."
|
|
||||||
)
|
|
||||||
|
|
||||||
meta_backend_kwargs = meta.get("backend_kwargs", {})
|
|
||||||
index_is_compact = meta.get("is_compact", meta_backend_kwargs.get("is_compact", True))
|
|
||||||
if index_is_compact:
|
|
||||||
raise ValueError(
|
|
||||||
"Compact HNSW indices do not support in-place updates. Rebuild required."
|
|
||||||
)
|
|
||||||
|
|
||||||
distance_metric = meta_backend_kwargs.get(
|
|
||||||
"distance_metric", self.backend_kwargs.get("distance_metric", "mips")
|
|
||||||
).lower()
|
|
||||||
needs_recompute = bool(
|
|
||||||
meta.get("is_pruned")
|
|
||||||
or meta_backend_kwargs.get("is_recompute")
|
|
||||||
or self.backend_kwargs.get("is_recompute")
|
|
||||||
)
|
|
||||||
|
|
||||||
with open(offset_file, "rb") as f:
|
|
||||||
offset_map: dict[str, int] = pickle.load(f)
|
|
||||||
existing_ids = set(offset_map.keys())
|
|
||||||
|
|
||||||
valid_chunks: list[dict[str, Any]] = []
|
|
||||||
for chunk in self.chunks:
|
|
||||||
text = chunk.get("text", "")
|
|
||||||
if not isinstance(text, str) or not text.strip():
|
|
||||||
continue
|
|
||||||
metadata = chunk.setdefault("metadata", {})
|
|
||||||
passage_id = chunk.get("id") or metadata.get("id")
|
|
||||||
if passage_id and passage_id in existing_ids:
|
|
||||||
raise ValueError(f"Passage ID '{passage_id}' already exists in the index.")
|
|
||||||
valid_chunks.append(chunk)
|
|
||||||
|
|
||||||
if not valid_chunks:
|
|
||||||
raise ValueError("No valid chunks to append.")
|
|
||||||
|
|
||||||
texts_to_embed = [chunk["text"] for chunk in valid_chunks]
|
|
||||||
embeddings = compute_embeddings(
|
|
||||||
texts_to_embed,
|
|
||||||
self.embedding_model,
|
|
||||||
self.embedding_mode,
|
|
||||||
use_server=False,
|
|
||||||
is_build=True,
|
|
||||||
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:
|
class LeannSearcher:
|
||||||
def __init__(self, index_path: str, enable_warmup: bool = False, **backend_kwargs):
|
def __init__(self, index_path: str, enable_warmup: bool = False, **backend_kwargs):
|
||||||
@@ -784,7 +628,6 @@ class LeannSearcher:
|
|||||||
self.embedding_model = self.meta_data["embedding_model"]
|
self.embedding_model = self.meta_data["embedding_model"]
|
||||||
# Support both old and new format
|
# Support both old and new format
|
||||||
self.embedding_mode = self.meta_data.get("embedding_mode", "sentence-transformers")
|
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
|
# Delegate portability handling to PassageManager
|
||||||
self.passage_manager = PassageManager(
|
self.passage_manager = PassageManager(
|
||||||
self.meta_data.get("passage_sources", []), metadata_file_path=self.meta_path_str
|
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.")
|
raise ValueError(f"Backend '{backend_name}' not found.")
|
||||||
final_kwargs = {**self.meta_data.get("backend_kwargs", {}), **backend_kwargs}
|
final_kwargs = {**self.meta_data.get("backend_kwargs", {}), **backend_kwargs}
|
||||||
final_kwargs["enable_warmup"] = enable_warmup
|
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(
|
self.backend_impl: LeannBackendSearcherInterface = backend_factory.searcher(
|
||||||
index_path, **final_kwargs
|
index_path, **final_kwargs
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -12,8 +12,6 @@ from typing import Any, Optional
|
|||||||
|
|
||||||
import torch
|
import torch
|
||||||
|
|
||||||
from .settings import resolve_ollama_host, resolve_openai_api_key, resolve_openai_base_url
|
|
||||||
|
|
||||||
# Configure logging
|
# Configure logging
|
||||||
logging.basicConfig(level=logging.INFO)
|
logging.basicConfig(level=logging.INFO)
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -312,12 +310,11 @@ def search_hf_models(query: str, limit: int = 10) -> list[str]:
|
|||||||
|
|
||||||
|
|
||||||
def validate_model_and_suggest(
|
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]:
|
) -> Optional[str]:
|
||||||
"""Validate model name and provide suggestions if invalid"""
|
"""Validate model name and provide suggestions if invalid"""
|
||||||
if llm_type == "ollama":
|
if llm_type == "ollama":
|
||||||
resolved_host = resolve_ollama_host(host)
|
available_models = check_ollama_models(host)
|
||||||
available_models = check_ollama_models(resolved_host)
|
|
||||||
if available_models and model_name not in available_models:
|
if available_models and model_name not in available_models:
|
||||||
error_msg = f"Model '{model_name}' not found in your local Ollama installation."
|
error_msg = f"Model '{model_name}' not found in your local Ollama installation."
|
||||||
|
|
||||||
@@ -460,19 +457,19 @@ class LLMInterface(ABC):
|
|||||||
class OllamaChat(LLMInterface):
|
class OllamaChat(LLMInterface):
|
||||||
"""LLM interface for Ollama models."""
|
"""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.model = model
|
||||||
self.host = resolve_ollama_host(host)
|
self.host = host
|
||||||
logger.info(f"Initializing OllamaChat with model='{model}' and host='{self.host}'")
|
logger.info(f"Initializing OllamaChat with model='{model}' and host='{host}'")
|
||||||
try:
|
try:
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
# Check if the Ollama server is responsive
|
# Check if the Ollama server is responsive
|
||||||
if self.host:
|
if host:
|
||||||
requests.get(self.host)
|
requests.get(host)
|
||||||
|
|
||||||
# Pre-check model availability with helpful suggestions
|
# 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:
|
if model_error:
|
||||||
raise ValueError(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'."
|
"The 'requests' library is required for Ollama. Please install it with 'pip install requests'."
|
||||||
)
|
)
|
||||||
except requests.exceptions.ConnectionError:
|
except requests.exceptions.ConnectionError:
|
||||||
logger.error(
|
logger.error(f"Could not connect to Ollama at {host}. Please ensure Ollama is running.")
|
||||||
f"Could not connect to Ollama at {self.host}. Please ensure Ollama is running."
|
|
||||||
)
|
|
||||||
raise ConnectionError(
|
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:
|
def ask(self, prompt: str, **kwargs) -> str:
|
||||||
@@ -742,31 +737,21 @@ class GeminiChat(LLMInterface):
|
|||||||
class OpenAIChat(LLMInterface):
|
class OpenAIChat(LLMInterface):
|
||||||
"""LLM interface for OpenAI models."""
|
"""LLM interface for OpenAI models."""
|
||||||
|
|
||||||
def __init__(
|
def __init__(self, model: str = "gpt-4o", api_key: Optional[str] = None):
|
||||||
self,
|
|
||||||
model: str = "gpt-4o",
|
|
||||||
api_key: Optional[str] = None,
|
|
||||||
base_url: Optional[str] = None,
|
|
||||||
):
|
|
||||||
self.model = model
|
self.model = model
|
||||||
self.base_url = resolve_openai_base_url(base_url)
|
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
|
||||||
self.api_key = resolve_openai_api_key(api_key)
|
|
||||||
|
|
||||||
if not self.api_key:
|
if not self.api_key:
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
"OpenAI API key is required. Set OPENAI_API_KEY environment variable or pass api_key parameter."
|
"OpenAI API key is required. Set OPENAI_API_KEY environment variable or pass api_key parameter."
|
||||||
)
|
)
|
||||||
|
|
||||||
logger.info(
|
logger.info(f"Initializing OpenAI Chat with model='{model}'")
|
||||||
"Initializing OpenAI Chat with model='%s' and base_url='%s'",
|
|
||||||
model,
|
|
||||||
self.base_url,
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
import openai
|
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:
|
except ImportError:
|
||||||
raise ImportError(
|
raise ImportError(
|
||||||
"The 'openai' library is required for OpenAI models. Please install it with 'pip install openai'."
|
"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":
|
if llm_type == "ollama":
|
||||||
return OllamaChat(
|
return OllamaChat(
|
||||||
model=model or "llama3:8b",
|
model=model or "llama3:8b",
|
||||||
host=llm_config.get("host"),
|
host=llm_config.get("host", "http://localhost:11434"),
|
||||||
)
|
)
|
||||||
elif llm_type == "hf":
|
elif llm_type == "hf":
|
||||||
return HFChat(model_name=model or "deepseek-ai/deepseek-llm-7b-chat")
|
return HFChat(model_name=model or "deepseek-ai/deepseek-llm-7b-chat")
|
||||||
elif llm_type == "openai":
|
elif llm_type == "openai":
|
||||||
return OpenAIChat(
|
return OpenAIChat(model=model or "gpt-4o", api_key=llm_config.get("api_key"))
|
||||||
model=model or "gpt-4o",
|
|
||||||
api_key=llm_config.get("api_key"),
|
|
||||||
base_url=llm_config.get("base_url"),
|
|
||||||
)
|
|
||||||
elif llm_type == "gemini":
|
elif llm_type == "gemini":
|
||||||
return GeminiChat(model=model or "gemini-2.5-flash", api_key=llm_config.get("api_key"))
|
return GeminiChat(model=model or "gemini-2.5-flash", api_key=llm_config.get("api_key"))
|
||||||
elif llm_type == "simulated":
|
elif llm_type == "simulated":
|
||||||
|
|||||||
@@ -9,7 +9,6 @@ from tqdm import tqdm
|
|||||||
|
|
||||||
from .api import LeannBuilder, LeannChat, LeannSearcher
|
from .api import LeannBuilder, LeannChat, LeannSearcher
|
||||||
from .registry import register_project_directory
|
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:
|
def extract_pdf_text_with_pymupdf(file_path: str) -> str:
|
||||||
@@ -124,24 +123,6 @@ Examples:
|
|||||||
choices=["sentence-transformers", "openai", "mlx", "ollama"],
|
choices=["sentence-transformers", "openai", "mlx", "ollama"],
|
||||||
help="Embedding backend mode (default: sentence-transformers)",
|
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(
|
build_parser.add_argument(
|
||||||
"--force", "-f", action="store_true", help="Force rebuild existing index"
|
"--force", "-f", action="store_true", help="Force rebuild existing index"
|
||||||
)
|
)
|
||||||
@@ -257,11 +238,6 @@ Examples:
|
|||||||
# Ask command
|
# Ask command
|
||||||
ask_parser = subparsers.add_parser("ask", help="Ask questions")
|
ask_parser = subparsers.add_parser("ask", help="Ask questions")
|
||||||
ask_parser.add_argument("index_name", help="Index name")
|
ask_parser.add_argument("index_name", help="Index name")
|
||||||
ask_parser.add_argument(
|
|
||||||
"query",
|
|
||||||
nargs="?",
|
|
||||||
help="Question to ask (omit for prompt or when using --interactive)",
|
|
||||||
)
|
|
||||||
ask_parser.add_argument(
|
ask_parser.add_argument(
|
||||||
"--llm",
|
"--llm",
|
||||||
type=str,
|
type=str,
|
||||||
@@ -272,12 +248,7 @@ Examples:
|
|||||||
ask_parser.add_argument(
|
ask_parser.add_argument(
|
||||||
"--model", type=str, default="qwen3:8b", help="Model name (default: qwen3:8b)"
|
"--model", type=str, default="qwen3:8b", help="Model name (default: qwen3:8b)"
|
||||||
)
|
)
|
||||||
ask_parser.add_argument(
|
ask_parser.add_argument("--host", type=str, default="http://localhost:11434")
|
||||||
"--host",
|
|
||||||
type=str,
|
|
||||||
default=None,
|
|
||||||
help="Override Ollama-compatible host (defaults to LEANN_OLLAMA_HOST/OLLAMA_HOST)",
|
|
||||||
)
|
|
||||||
ask_parser.add_argument(
|
ask_parser.add_argument(
|
||||||
"--interactive", "-i", action="store_true", help="Interactive chat mode"
|
"--interactive", "-i", action="store_true", help="Interactive chat mode"
|
||||||
)
|
)
|
||||||
@@ -306,18 +277,6 @@ Examples:
|
|||||||
default=None,
|
default=None,
|
||||||
help="Thinking budget for reasoning models (low/medium/high). Supported by GPT-Oss:20b and other reasoning models.",
|
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
|
# List command
|
||||||
subparsers.add_parser("list", help="List all indexes")
|
subparsers.add_parser("list", help="List all indexes")
|
||||||
@@ -1366,20 +1325,10 @@ Examples:
|
|||||||
|
|
||||||
print(f"Building index '{index_name}' with {args.backend} backend...")
|
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(
|
builder = LeannBuilder(
|
||||||
backend_name=args.backend,
|
backend_name=args.backend,
|
||||||
embedding_model=args.embedding_model,
|
embedding_model=args.embedding_model,
|
||||||
embedding_mode=args.embedding_mode,
|
embedding_mode=args.embedding_mode,
|
||||||
embedding_options=embedding_options or None,
|
|
||||||
graph_degree=args.graph_degree,
|
graph_degree=args.graph_degree,
|
||||||
complexity=args.complexity,
|
complexity=args.complexity,
|
||||||
is_compact=args.compact,
|
is_compact=args.compact,
|
||||||
@@ -1527,38 +1476,11 @@ Examples:
|
|||||||
|
|
||||||
llm_config = {"type": args.llm, "model": args.model}
|
llm_config = {"type": args.llm, "model": args.model}
|
||||||
if args.llm == "ollama":
|
if args.llm == "ollama":
|
||||||
llm_config["host"] = resolve_ollama_host(args.host)
|
llm_config["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
|
|
||||||
|
|
||||||
chat = LeannChat(index_path=index_path, llm_config=llm_config)
|
chat = LeannChat(index_path=index_path, llm_config=llm_config)
|
||||||
|
|
||||||
llm_kwargs: dict[str, Any] = {}
|
|
||||||
if args.thinking_budget:
|
|
||||||
llm_kwargs["thinking_budget"] = args.thinking_budget
|
|
||||||
|
|
||||||
def _ask_once(prompt: str) -> None:
|
|
||||||
response = chat.ask(
|
|
||||||
prompt,
|
|
||||||
top_k=args.top_k,
|
|
||||||
complexity=args.complexity,
|
|
||||||
beam_width=args.beam_width,
|
|
||||||
prune_ratio=args.prune_ratio,
|
|
||||||
recompute_embeddings=args.recompute_embeddings,
|
|
||||||
pruning_strategy=args.pruning_strategy,
|
|
||||||
llm_kwargs=llm_kwargs,
|
|
||||||
)
|
|
||||||
print(f"LEANN: {response}")
|
|
||||||
|
|
||||||
initial_query = (args.query or "").strip()
|
|
||||||
|
|
||||||
if args.interactive:
|
if args.interactive:
|
||||||
if initial_query:
|
|
||||||
_ask_once(initial_query)
|
|
||||||
|
|
||||||
print("LEANN Assistant ready! Type 'quit' to exit")
|
print("LEANN Assistant ready! Type 'quit' to exit")
|
||||||
print("=" * 40)
|
print("=" * 40)
|
||||||
|
|
||||||
@@ -1571,14 +1493,41 @@ Examples:
|
|||||||
if not user_input:
|
if not user_input:
|
||||||
continue
|
continue
|
||||||
|
|
||||||
_ask_once(user_input)
|
# Prepare LLM kwargs with thinking budget if specified
|
||||||
else:
|
llm_kwargs = {}
|
||||||
query = initial_query or input("Enter your question: ").strip()
|
if args.thinking_budget:
|
||||||
if not query:
|
llm_kwargs["thinking_budget"] = args.thinking_budget
|
||||||
print("No question provided. Exiting.")
|
|
||||||
return
|
|
||||||
|
|
||||||
_ask_once(query)
|
response = chat.ask(
|
||||||
|
user_input,
|
||||||
|
top_k=args.top_k,
|
||||||
|
complexity=args.complexity,
|
||||||
|
beam_width=args.beam_width,
|
||||||
|
prune_ratio=args.prune_ratio,
|
||||||
|
recompute_embeddings=args.recompute_embeddings,
|
||||||
|
pruning_strategy=args.pruning_strategy,
|
||||||
|
llm_kwargs=llm_kwargs,
|
||||||
|
)
|
||||||
|
print(f"LEANN: {response}")
|
||||||
|
else:
|
||||||
|
query = input("Enter your question: ").strip()
|
||||||
|
if query:
|
||||||
|
# Prepare LLM kwargs with thinking budget if specified
|
||||||
|
llm_kwargs = {}
|
||||||
|
if args.thinking_budget:
|
||||||
|
llm_kwargs["thinking_budget"] = args.thinking_budget
|
||||||
|
|
||||||
|
response = chat.ask(
|
||||||
|
query,
|
||||||
|
top_k=args.top_k,
|
||||||
|
complexity=args.complexity,
|
||||||
|
beam_width=args.beam_width,
|
||||||
|
prune_ratio=args.prune_ratio,
|
||||||
|
recompute_embeddings=args.recompute_embeddings,
|
||||||
|
pruning_strategy=args.pruning_strategy,
|
||||||
|
llm_kwargs=llm_kwargs,
|
||||||
|
)
|
||||||
|
print(f"LEANN: {response}")
|
||||||
|
|
||||||
async def run(self, args=None):
|
async def run(self, args=None):
|
||||||
parser = self.create_parser()
|
parser = self.create_parser()
|
||||||
|
|||||||
@@ -7,13 +7,11 @@ Preserves all optimization parameters to ensure performance
|
|||||||
import logging
|
import logging
|
||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
from typing import Any, Optional
|
from typing import Any
|
||||||
|
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import torch
|
import torch
|
||||||
|
|
||||||
from .settings import resolve_ollama_host, resolve_openai_api_key, resolve_openai_base_url
|
|
||||||
|
|
||||||
# Set up logger with proper level
|
# Set up logger with proper level
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
LOG_LEVEL = os.getenv("LEANN_LOG_LEVEL", "WARNING").upper()
|
LOG_LEVEL = os.getenv("LEANN_LOG_LEVEL", "WARNING").upper()
|
||||||
@@ -33,7 +31,6 @@ def compute_embeddings(
|
|||||||
adaptive_optimization: bool = True,
|
adaptive_optimization: bool = True,
|
||||||
manual_tokenize: bool = False,
|
manual_tokenize: bool = False,
|
||||||
max_length: int = 512,
|
max_length: int = 512,
|
||||||
provider_options: Optional[dict[str, Any]] = None,
|
|
||||||
) -> np.ndarray:
|
) -> np.ndarray:
|
||||||
"""
|
"""
|
||||||
Unified embedding computation entry point
|
Unified embedding computation entry point
|
||||||
@@ -49,8 +46,6 @@ def compute_embeddings(
|
|||||||
Returns:
|
Returns:
|
||||||
Normalized embeddings array, shape: (len(texts), embedding_dim)
|
Normalized embeddings array, shape: (len(texts), embedding_dim)
|
||||||
"""
|
"""
|
||||||
provider_options = provider_options or {}
|
|
||||||
|
|
||||||
if mode == "sentence-transformers":
|
if mode == "sentence-transformers":
|
||||||
return compute_embeddings_sentence_transformers(
|
return compute_embeddings_sentence_transformers(
|
||||||
texts,
|
texts,
|
||||||
@@ -62,21 +57,11 @@ def compute_embeddings(
|
|||||||
max_length=max_length,
|
max_length=max_length,
|
||||||
)
|
)
|
||||||
elif mode == "openai":
|
elif mode == "openai":
|
||||||
return compute_embeddings_openai(
|
return compute_embeddings_openai(texts, model_name)
|
||||||
texts,
|
|
||||||
model_name,
|
|
||||||
base_url=provider_options.get("base_url"),
|
|
||||||
api_key=provider_options.get("api_key"),
|
|
||||||
)
|
|
||||||
elif mode == "mlx":
|
elif mode == "mlx":
|
||||||
return compute_embeddings_mlx(texts, model_name)
|
return compute_embeddings_mlx(texts, model_name)
|
||||||
elif mode == "ollama":
|
elif mode == "ollama":
|
||||||
return compute_embeddings_ollama(
|
return compute_embeddings_ollama(texts, model_name, is_build=is_build)
|
||||||
texts,
|
|
||||||
model_name,
|
|
||||||
is_build=is_build,
|
|
||||||
host=provider_options.get("host"),
|
|
||||||
)
|
|
||||||
elif mode == "gemini":
|
elif mode == "gemini":
|
||||||
return compute_embeddings_gemini(texts, model_name, is_build=is_build)
|
return compute_embeddings_gemini(texts, model_name, is_build=is_build)
|
||||||
else:
|
else:
|
||||||
@@ -368,15 +353,12 @@ def compute_embeddings_sentence_transformers(
|
|||||||
return embeddings
|
return embeddings
|
||||||
|
|
||||||
|
|
||||||
def compute_embeddings_openai(
|
def compute_embeddings_openai(texts: list[str], model_name: str) -> np.ndarray:
|
||||||
texts: list[str],
|
|
||||||
model_name: str,
|
|
||||||
base_url: Optional[str] = None,
|
|
||||||
api_key: Optional[str] = None,
|
|
||||||
) -> np.ndarray:
|
|
||||||
# TODO: @yichuan-w add progress bar only in build mode
|
# TODO: @yichuan-w add progress bar only in build mode
|
||||||
"""Compute embeddings using OpenAI API"""
|
"""Compute embeddings using OpenAI API"""
|
||||||
try:
|
try:
|
||||||
|
import os
|
||||||
|
|
||||||
import openai
|
import openai
|
||||||
except ImportError as e:
|
except ImportError as e:
|
||||||
raise ImportError(f"OpenAI package not installed: {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."
|
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)
|
api_key = os.getenv("OPENAI_API_KEY")
|
||||||
resolved_api_key = resolve_openai_api_key(api_key)
|
if not api_key:
|
||||||
|
|
||||||
if not resolved_api_key:
|
|
||||||
raise RuntimeError("OPENAI_API_KEY environment variable not set")
|
raise RuntimeError("OPENAI_API_KEY environment variable not set")
|
||||||
|
|
||||||
# Cache OpenAI client
|
# Cache OpenAI client
|
||||||
cache_key = f"openai_client::{resolved_base_url}"
|
cache_key = "openai_client"
|
||||||
if cache_key in _model_cache:
|
if cache_key in _model_cache:
|
||||||
client = _model_cache[cache_key]
|
client = _model_cache[cache_key]
|
||||||
else:
|
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
|
_model_cache[cache_key] = client
|
||||||
logger.info("OpenAI client cached")
|
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(
|
def compute_embeddings_ollama(
|
||||||
texts: list[str],
|
texts: list[str], model_name: str, is_build: bool = False, host: str = "http://localhost:11434"
|
||||||
model_name: str,
|
|
||||||
is_build: bool = False,
|
|
||||||
host: Optional[str] = None,
|
|
||||||
) -> np.ndarray:
|
) -> np.ndarray:
|
||||||
"""
|
"""
|
||||||
Compute embeddings using Ollama API with simplified batch processing.
|
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
|
texts: List of texts to compute embeddings for
|
||||||
model_name: Ollama model name (e.g., "nomic-embed-text", "mxbai-embed-large")
|
model_name: Ollama model name (e.g., "nomic-embed-text", "mxbai-embed-large")
|
||||||
is_build: Whether this is a build operation (shows progress bar)
|
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:
|
Returns:
|
||||||
Normalized embeddings array, shape: (len(texts), embedding_dim)
|
Normalized embeddings array, shape: (len(texts), embedding_dim)
|
||||||
@@ -556,19 +533,17 @@ def compute_embeddings_ollama(
|
|||||||
if not texts:
|
if not texts:
|
||||||
raise ValueError("Cannot compute embeddings for empty text list")
|
raise ValueError("Cannot compute embeddings for empty text list")
|
||||||
|
|
||||||
resolved_host = resolve_ollama_host(host)
|
|
||||||
|
|
||||||
logger.info(
|
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
|
# Check if Ollama is running
|
||||||
try:
|
try:
|
||||||
response = requests.get(f"{resolved_host}/api/version", timeout=5)
|
response = requests.get(f"{host}/api/version", timeout=5)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
except requests.exceptions.ConnectionError:
|
except requests.exceptions.ConnectionError:
|
||||||
error_msg = (
|
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"
|
"Please ensure Ollama is running:\n"
|
||||||
" • macOS/Linux: ollama serve\n"
|
" • macOS/Linux: ollama serve\n"
|
||||||
" • Windows: Make sure Ollama is running in the system tray\n\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
|
# Check if model exists and provide helpful suggestions
|
||||||
try:
|
try:
|
||||||
response = requests.get(f"{resolved_host}/api/tags", timeout=5)
|
response = requests.get(f"{host}/api/tags", timeout=5)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
models = response.json()
|
models = response.json()
|
||||||
model_names = [model["name"] for model in models.get("models", [])]
|
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
|
# Verify the model supports embeddings by testing it
|
||||||
try:
|
try:
|
||||||
test_response = requests.post(
|
test_response = requests.post(
|
||||||
f"{resolved_host}/api/embeddings",
|
f"{host}/api/embeddings", json={"model": model_name, "prompt": "test"}, timeout=10
|
||||||
json={"model": model_name, "prompt": "test"},
|
|
||||||
timeout=10,
|
|
||||||
)
|
)
|
||||||
if test_response.status_code != 200:
|
if test_response.status_code != 200:
|
||||||
error_msg = (
|
error_msg = (
|
||||||
@@ -692,7 +665,7 @@ def compute_embeddings_ollama(
|
|||||||
while retry_count < max_retries:
|
while retry_count < max_retries:
|
||||||
try:
|
try:
|
||||||
response = requests.post(
|
response = requests.post(
|
||||||
f"{resolved_host}/api/embeddings",
|
f"{host}/api/embeddings",
|
||||||
json={"model": model_name, "prompt": truncated_text},
|
json={"model": model_name, "prompt": truncated_text},
|
||||||
timeout=30,
|
timeout=30,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -8,8 +8,6 @@ import time
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
from .settings import encode_provider_options
|
|
||||||
|
|
||||||
# Lightweight, self-contained server manager with no cross-process inspection
|
# Lightweight, self-contained server manager with no cross-process inspection
|
||||||
|
|
||||||
# Set up logging based on environment variable
|
# Set up logging based on environment variable
|
||||||
@@ -84,40 +82,16 @@ class EmbeddingServerManager:
|
|||||||
) -> tuple[bool, int]:
|
) -> tuple[bool, int]:
|
||||||
"""Start the embedding server."""
|
"""Start the embedding server."""
|
||||||
# passages_file may be present in kwargs for server CLI, but we don't need it here
|
# 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 this manager already has a live server, just reuse it
|
||||||
if (
|
if self.server_process and self.server_process.poll() is None and self.server_port:
|
||||||
self.server_process
|
|
||||||
and self.server_process.poll() is None
|
|
||||||
and self.server_port
|
|
||||||
and self._server_config == config_signature
|
|
||||||
):
|
|
||||||
logger.info("Reusing in-process server")
|
logger.info("Reusing in-process server")
|
||||||
return True, self.server_port
|
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
|
# For Colab environment, use a different strategy
|
||||||
if _is_colab_environment():
|
if _is_colab_environment():
|
||||||
logger.info("Detected Colab environment, using alternative startup strategy")
|
logger.info("Detected Colab environment, using alternative startup strategy")
|
||||||
return self._start_server_colab(
|
return self._start_server_colab(port, model_name, embedding_mode, **kwargs)
|
||||||
port,
|
|
||||||
model_name,
|
|
||||||
embedding_mode,
|
|
||||||
provider_options=provider_options,
|
|
||||||
**kwargs,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Always pick a fresh available port
|
# Always pick a fresh available port
|
||||||
try:
|
try:
|
||||||
@@ -127,21 +101,13 @@ class EmbeddingServerManager:
|
|||||||
return False, port
|
return False, port
|
||||||
|
|
||||||
# Start a new server
|
# Start a new server
|
||||||
return self._start_new_server(
|
return self._start_new_server(actual_port, model_name, embedding_mode, **kwargs)
|
||||||
actual_port,
|
|
||||||
model_name,
|
|
||||||
embedding_mode,
|
|
||||||
provider_options=provider_options,
|
|
||||||
config_signature=config_signature,
|
|
||||||
**kwargs,
|
|
||||||
)
|
|
||||||
|
|
||||||
def _start_server_colab(
|
def _start_server_colab(
|
||||||
self,
|
self,
|
||||||
port: int,
|
port: int,
|
||||||
model_name: str,
|
model_name: str,
|
||||||
embedding_mode: str = "sentence-transformers",
|
embedding_mode: str = "sentence-transformers",
|
||||||
provider_options: Optional[dict] = None,
|
|
||||||
**kwargs,
|
**kwargs,
|
||||||
) -> tuple[bool, int]:
|
) -> tuple[bool, int]:
|
||||||
"""Start server with Colab-specific configuration."""
|
"""Start server with Colab-specific configuration."""
|
||||||
@@ -159,20 +125,8 @@ class EmbeddingServerManager:
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
# In Colab, we'll use a more direct approach
|
# In Colab, we'll use a more direct approach
|
||||||
self._launch_server_process_colab(
|
self._launch_server_process_colab(command, actual_port)
|
||||||
command,
|
return self._wait_for_server_ready_colab(actual_port)
|
||||||
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
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Failed to start embedding server in Colab: {e}")
|
logger.error(f"Failed to start embedding server in Colab: {e}")
|
||||||
return False, actual_port
|
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
|
# Note: No compatibility check needed; manager is per-searcher and configs are stable per instance
|
||||||
|
|
||||||
def _start_new_server(
|
def _start_new_server(
|
||||||
self,
|
self, port: int, model_name: str, embedding_mode: str, **kwargs
|
||||||
port: int,
|
|
||||||
model_name: str,
|
|
||||||
embedding_mode: str,
|
|
||||||
provider_options: Optional[dict] = None,
|
|
||||||
config_signature: Optional[dict] = None,
|
|
||||||
**kwargs,
|
|
||||||
) -> tuple[bool, int]:
|
) -> tuple[bool, int]:
|
||||||
"""Start a new embedding server on the given port."""
|
"""Start a new embedding server on the given port."""
|
||||||
logger.info(f"Starting embedding server on port {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)
|
command = self._build_server_command(port, model_name, embedding_mode, **kwargs)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self._launch_server_process(
|
self._launch_server_process(command, port)
|
||||||
command,
|
return self._wait_for_server_ready(port)
|
||||||
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
|
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.error(f"Failed to start embedding server: {e}")
|
logger.error(f"Failed to start embedding server: {e}")
|
||||||
return False, port
|
return False, port
|
||||||
@@ -237,12 +173,7 @@ class EmbeddingServerManager:
|
|||||||
|
|
||||||
return command
|
return command
|
||||||
|
|
||||||
def _launch_server_process(
|
def _launch_server_process(self, command: list, port: int) -> None:
|
||||||
self,
|
|
||||||
command: list,
|
|
||||||
port: int,
|
|
||||||
provider_options: Optional[dict] = None,
|
|
||||||
) -> None:
|
|
||||||
"""Launch the server process."""
|
"""Launch the server process."""
|
||||||
project_root = Path(__file__).parent.parent.parent.parent.parent
|
project_root = Path(__file__).parent.parent.parent.parent.parent
|
||||||
logger.info(f"Command: {' '.join(command)}")
|
logger.info(f"Command: {' '.join(command)}")
|
||||||
@@ -262,20 +193,14 @@ class EmbeddingServerManager:
|
|||||||
|
|
||||||
# Start embedding server subprocess
|
# Start embedding server subprocess
|
||||||
logger.info(f"Starting server process with command: {' '.join(command)}")
|
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(
|
self.server_process = subprocess.Popen(
|
||||||
command,
|
command,
|
||||||
cwd=project_root,
|
cwd=project_root,
|
||||||
stdout=stdout_target,
|
stdout=stdout_target,
|
||||||
stderr=stderr_target,
|
stderr=stderr_target,
|
||||||
env=env,
|
|
||||||
)
|
)
|
||||||
self.server_port = port
|
self.server_port = port
|
||||||
# Record config for in-process reuse (best effort; refined later when ready)
|
# Record config for in-process reuse
|
||||||
try:
|
try:
|
||||||
self._server_config = {
|
self._server_config = {
|
||||||
"model_name": command[command.index("--model-name") + 1]
|
"model_name": command[command.index("--model-name") + 1]
|
||||||
@@ -287,14 +212,12 @@ class EmbeddingServerManager:
|
|||||||
"embedding_mode": command[command.index("--embedding-mode") + 1]
|
"embedding_mode": command[command.index("--embedding-mode") + 1]
|
||||||
if "--embedding-mode" in command
|
if "--embedding-mode" in command
|
||||||
else "sentence-transformers",
|
else "sentence-transformers",
|
||||||
"provider_options": provider_options or {},
|
|
||||||
}
|
}
|
||||||
except Exception:
|
except Exception:
|
||||||
self._server_config = {
|
self._server_config = {
|
||||||
"model_name": "",
|
"model_name": "",
|
||||||
"passages_file": "",
|
"passages_file": "",
|
||||||
"embedding_mode": "sentence-transformers",
|
"embedding_mode": "sentence-transformers",
|
||||||
"provider_options": provider_options or {},
|
|
||||||
}
|
}
|
||||||
logger.info(f"Server process started with PID: {self.server_process.pid}")
|
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
|
# Removed: cross-process adoption no longer supported
|
||||||
return
|
return
|
||||||
|
|
||||||
def _launch_server_process_colab(
|
def _launch_server_process_colab(self, command: list, port: int) -> None:
|
||||||
self,
|
|
||||||
command: list,
|
|
||||||
port: int,
|
|
||||||
provider_options: Optional[dict] = None,
|
|
||||||
) -> None:
|
|
||||||
"""Launch the server process with Colab-specific settings."""
|
"""Launch the server process with Colab-specific settings."""
|
||||||
logger.info(f"Colab Command: {' '.join(command)}")
|
logger.info(f"Colab Command: {' '.join(command)}")
|
||||||
|
|
||||||
# In Colab, we need to be more careful about process management
|
# 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(
|
self.server_process = subprocess.Popen(
|
||||||
command,
|
command,
|
||||||
stdout=subprocess.PIPE,
|
stdout=subprocess.PIPE,
|
||||||
stderr=subprocess.PIPE,
|
stderr=subprocess.PIPE,
|
||||||
text=True,
|
text=True,
|
||||||
env=env,
|
|
||||||
)
|
)
|
||||||
self.server_port = port
|
self.server_port = port
|
||||||
logger.info(f"Colab server process started with PID: {self.server_process.pid}")
|
logger.info(f"Colab server process started with PID: {self.server_process.pid}")
|
||||||
@@ -433,7 +345,6 @@ class EmbeddingServerManager:
|
|||||||
"model_name": "",
|
"model_name": "",
|
||||||
"passages_file": "",
|
"passages_file": "",
|
||||||
"embedding_mode": "sentence-transformers",
|
"embedding_mode": "sentence-transformers",
|
||||||
"provider_options": provider_options or {},
|
|
||||||
}
|
}
|
||||||
|
|
||||||
def _wait_for_server_ready_colab(self, port: int) -> tuple[bool, int]:
|
def _wait_for_server_ready_colab(self, port: int) -> tuple[bool, int]:
|
||||||
|
|||||||
@@ -41,7 +41,6 @@ class BaseSearcher(LeannBackendSearcherInterface, ABC):
|
|||||||
print("WARNING: embedding_model not found in meta.json. Recompute will fail.")
|
print("WARNING: embedding_model not found in meta.json. Recompute will fail.")
|
||||||
|
|
||||||
self.embedding_mode = self.meta.get("embedding_mode", "sentence-transformers")
|
self.embedding_mode = self.meta.get("embedding_mode", "sentence-transformers")
|
||||||
self.embedding_options = self.meta.get("embedding_options", {})
|
|
||||||
|
|
||||||
self.embedding_server_manager = EmbeddingServerManager(
|
self.embedding_server_manager = EmbeddingServerManager(
|
||||||
backend_module_name=backend_module_name,
|
backend_module_name=backend_module_name,
|
||||||
@@ -78,7 +77,6 @@ class BaseSearcher(LeannBackendSearcherInterface, ABC):
|
|||||||
passages_file=passages_source_file,
|
passages_file=passages_source_file,
|
||||||
distance_metric=distance_metric,
|
distance_metric=distance_metric,
|
||||||
enable_warmup=kwargs.get("enable_warmup", False),
|
enable_warmup=kwargs.get("enable_warmup", False),
|
||||||
provider_options=self.embedding_options,
|
|
||||||
)
|
)
|
||||||
if not server_started:
|
if not server_started:
|
||||||
raise RuntimeError(f"Failed to start embedding server on port {actual_port}")
|
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
|
from .embedding_compute import compute_embeddings
|
||||||
|
|
||||||
embedding_mode = self.meta.get("embedding_mode", "sentence-transformers")
|
embedding_mode = self.meta.get("embedding_mode", "sentence-transformers")
|
||||||
return compute_embeddings(
|
return compute_embeddings([query], self.embedding_model, embedding_mode)
|
||||||
[query],
|
|
||||||
self.embedding_model,
|
|
||||||
embedding_mode,
|
|
||||||
provider_options=self.embedding_options,
|
|
||||||
)
|
|
||||||
|
|
||||||
def _compute_embedding_via_server(self, chunks: list, zmq_port: int) -> np.ndarray:
|
def _compute_embedding_via_server(self, chunks: list, zmq_port: int) -> np.ndarray:
|
||||||
"""Compute embeddings using the ZMQ embedding server."""
|
"""Compute embeddings using the ZMQ embedding server."""
|
||||||
|
|||||||
@@ -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
|
|
||||||
@@ -104,7 +104,11 @@ astchunk = { path = "packages/astchunk-leann", editable = true }
|
|||||||
[tool.ruff]
|
[tool.ruff]
|
||||||
target-version = "py39"
|
target-version = "py39"
|
||||||
line-length = 100
|
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]
|
[tool.ruff.lint]
|
||||||
|
|||||||
@@ -1,14 +0,0 @@
|
|||||||
from leann.cli import LeannCLI
|
|
||||||
|
|
||||||
|
|
||||||
def test_cli_ask_accepts_positional_query(tmp_path, monkeypatch):
|
|
||||||
monkeypatch.chdir(tmp_path)
|
|
||||||
|
|
||||||
cli = LeannCLI()
|
|
||||||
parser = cli.create_parser()
|
|
||||||
|
|
||||||
args = parser.parse_args(["ask", "my-docs", "Where are prompts configured?"])
|
|
||||||
|
|
||||||
assert args.command == "ask"
|
|
||||||
assert args.index_name == "my-docs"
|
|
||||||
assert args.query == "Where are prompts configured?"
|
|
||||||
Reference in New Issue
Block a user