chore: import upstream snapshot with attribution

turbovec-python/python/turbovec/__init__.py

@@ -0,0 +1,3 @@
+from ._turbovec import IdMapIndex, TurboQuantIndex
+
+__all__ = ["IdMapIndex", "TurboQuantIndex"]

turbovec-python/python/turbovec/_dedup.py

@@ -0,0 +1,71 @@
+"""Shared in-batch duplicate resolution for the framework integrations.
+
+Each upstream library resolves a repeated id *within a single write* its own
+way, and every turbovec wrapper must match its upstream to stay a true
+drop-in:
+
+- LangChain's ``InMemoryVectorStore`` overwrites on a repeated key  → KEEP_LAST
+- LlamaIndex rejects duplicate ``node_id`` in a batch               → REJECT
+- agno's LanceDb is append-only and keeps every row                 → KEEP_ALL
+- Haystack exposes a runtime ``DuplicatePolicy`` (FAIL/SKIP/OVERWRITE).
+  Its resolution is *stateful* (it dedups against the existing store as well
+  as the batch, with deferred issue-#89 removal), so it does not reduce to
+  the pure in-batch function here and keeps its own logic; this enum still
+  documents the mapping (OVERWRITE→KEEP_LAST, SKIP→KEEP_FIRST, FAIL→REJECT).
+
+The shared piece is the in-batch resolution only: given one key per item,
+return the indices to keep. Each wrapper still owns its key extraction and
+its cross-store upsert/removal.
+"""
+from __future__ import annotations
+
+import enum
+from typing import Hashable, List, Sequence
+
+
+class DuplicatePolicy(enum.Enum):
+    """How to resolve items that share a key within a single batch."""
+
+    KEEP_LAST = "keep_last"
+    """One item per key; the last occurrence wins (dict-overwrite semantics)."""
+
+    KEEP_FIRST = "keep_first"
+    """One item per key; the first occurrence wins."""
+
+    REJECT = "reject"
+    """Raise ``ValueError`` if any key repeats; otherwise keep everything."""
+
+    KEEP_ALL = "keep_all"
+    """No deduplication; items with duplicate keys all survive."""
+
+
+def resolve_duplicates(
+    keys: Sequence[Hashable], policy: DuplicatePolicy
+) -> List[int]:
+    """Return, in ascending order, the batch indices to keep under ``policy``.
+
+    The returned indices index into ``keys`` (and any parallel arrays the
+    caller holds). For KEEP_ALL and REJECT the result is ``0..len(keys)``;
+    for KEEP_LAST/KEEP_FIRST it collapses to one index per distinct key.
+
+    Raises:
+        ValueError: under REJECT, if any key occurs more than once.
+    """
+    if policy is DuplicatePolicy.KEEP_ALL:
+        return list(range(len(keys)))
+    if policy is DuplicatePolicy.REJECT:
+        seen: set = set()
+        for k in keys:
+            if k in seen:
+                raise ValueError(f"duplicate id in batch: {k!r}")
+            seen.add(k)
+        return list(range(len(keys)))
+    # KEEP_LAST / KEEP_FIRST collapse to one index per key.
+    chosen: dict = {}
+    for i, k in enumerate(keys):
+        if policy is DuplicatePolicy.KEEP_LAST or k not in chosen:
+            chosen[k] = i
+    return sorted(chosen.values())
+
+
+__all__ = ["DuplicatePolicy", "resolve_duplicates"]

turbovec-python/python/turbovec/_persist.py

@@ -0,0 +1,55 @@
+"""Shared persistence consistency checks for the framework integrations.
+
+Each wrapper persists two artifacts: the binary ``.tvim`` index and a JSON
+side-car holding the handle -> document/node/text payload maps. At query
+time the wrapper resolves an index-returned u64 handle through that side-car
+map. If the two files are out of sync — a partial copy, a stale backup, a
+hand-edited or tampered side-car — an index handle won't resolve and the
+wrapper would raise an opaque ``KeyError`` deep inside a query.
+
+``check_persisted_handles`` turns that into a clean ``ValueError`` at load
+time. ``IdMapIndex`` exposes only ``__len__`` and ``contains``; that's
+sufficient: if the side-car's handle set and the index have equal size and
+every side-car handle is present in the index, the two are a bijection (no
+index handle can be missing from the side-car).
+"""
+from __future__ import annotations
+
+from typing import Iterable
+
+
+def check_persisted_handles(index, handles: Iterable[int], *, what: str = "entry") -> None:
+    """Validate that the side-car's handle set matches the loaded index.
+
+    Args:
+        index: the loaded ``IdMapIndex`` (uses ``len`` and ``contains``).
+        handles: the u64 handles the side-car maps can resolve.
+        what: noun for error messages (e.g. "document", "node").
+
+    Raises:
+        ValueError: if the side-car has duplicate handles, a different count
+            than the index, or a handle the index doesn't contain.
+    """
+    handle_list = [int(h) for h in handles]
+    n_index = len(index)
+
+    if len(set(handle_list)) != len(handle_list):
+        raise ValueError(
+            f"persisted store is corrupt: duplicate {what} handles in the side-car"
+        )
+    if len(handle_list) != n_index:
+        raise ValueError(
+            f"persisted store is inconsistent with its index: side-car has "
+            f"{len(handle_list)} {what} handle(s) but the index holds {n_index}. "
+            f"The .tvim index and its JSON side-car are out of sync."
+        )
+    for h in handle_list:
+        if not index.contains(h):
+            raise ValueError(
+                f"persisted store is inconsistent with its index: {what} handle "
+                f"{h} is not present in the index. The .tvim index and its JSON "
+                f"side-car are out of sync."
+            )
+
+
+__all__ = ["check_persisted_handles"]

turbovec-python/python/turbovec/agno.py

@@ -0,0 +1,788 @@
+"""Agno VectorDb backed by turbovec's quantized index.
+
+Install with: ``pip install turbovec[agno]``.
+
+Implements Agno's ``VectorDb`` interface and matches the public surface
+of ``agno.vectordb.lancedb.LanceDb`` (the closest in-tree single-machine
+backend) so this can be swapped in wherever ``LanceDb`` is used.
+"""
+
+from __future__ import annotations
+
+import json
+from hashlib import md5
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional, Set, Union
+
+import numpy as np
+
+from ._turbovec import IdMapIndex
+
+try:
+    from agno.knowledge.document import Document
+    from agno.knowledge.embedder import Embedder
+    from agno.knowledge.reranker.base import Reranker
+    from agno.vectordb.base import VectorDb
+    from agno.vectordb.distance import Distance
+    from agno.vectordb.search import SearchType
+except ImportError as exc:
+    raise ImportError(
+        "agno is required to use turbovec.agno. "
+        "Install with: pip install turbovec[agno]"
+    ) from exc
+
+
+_INDEX_FILENAME = "index.tvim"
+_STORE_FILENAME = "docstore.json"
+# Bump when docstore.json shape changes; loader refuses unknown versions.
+_DOCSTORE_SCHEMA_VERSION = 1
+
+
+class TurboQuantVectorDb(VectorDb):
+    """Agno VectorDb backed by a :class:`IdMapIndex`.
+
+    Vectors are quantized to 2-4 bits per dimension. The public surface
+    mirrors ``agno.vectordb.lancedb.LanceDb`` so this is a drop-in
+    replacement wherever a single-machine LanceDb is used. Search-time
+    filtering is resolved to an allowlist *before* scoring (kernel-level)
+    rather than via post-filtering, so selective filters return up to
+    ``limit`` results from the filtered set instead of fewer.
+
+    Example::
+
+        from agno.knowledge.embedder.openai import OpenAIEmbedder
+        from turbovec.agno import TurboQuantVectorDb
+
+        vector_db = TurboQuantVectorDb(embedder=OpenAIEmbedder())
+        vector_db.create()
+        # ... use as a normal Agno VectorDb ...
+    """
+
+    def __init__(
+        self,
+        *,
+        id: Optional[str] = None,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        similarity_threshold: Optional[float] = None,
+        embedder: Optional[Embedder] = None,
+        bit_width: int = 4,
+        search_type: SearchType = SearchType.vector,
+        distance: Distance = Distance.cosine,
+        reranker: Optional[Reranker] = None,
+        path: Optional[str] = None,
+    ) -> None:
+        """
+        :param embedder: Required. Agno embedder used to encode documents
+            and queries. ``embedder.dimensions`` must be set — it's the
+            sole source of truth for the underlying quantized index's
+            dimensionality.
+        :param bit_width: Quantization width (2 or 4).
+        :param search_type: Only :class:`SearchType.vector` is supported;
+            other values raise :class:`ValueError`. (Keyword/hybrid search
+            would require an external BM25/lexical index.)
+        :param distance: Only :class:`Distance.cosine` is supported.
+            turbovec stores unit-normalized vectors, so the kernel's raw
+            score is cosine similarity directly.
+        :param reranker: Optional Agno reranker applied to the result set
+            after vector retrieval.
+        :param path: Optional directory for save/load persistence. When
+            given to the constructor, :meth:`create` loads existing data
+            from this path if present.
+        """
+        super().__init__(
+            id=id,
+            name=name,
+            description=description,
+            similarity_threshold=similarity_threshold,
+        )
+        if embedder is None:
+            raise ValueError(
+                "`embedder` is required; turbovec needs the embedder's "
+                "`dimensions` to size the underlying index."
+            )
+        if embedder.dimensions is None:
+            raise ValueError("Embedder.dimensions must be set.")
+        if bit_width not in (2, 4):
+            raise ValueError(f"bit_width must be 2 or 4, got {bit_width}")
+        if search_type != SearchType.vector:
+            raise ValueError(
+                f"TurboQuantVectorDb only supports search_type=SearchType.vector; "
+                f"got {search_type}. Use LanceDb / Chroma / etc. for keyword "
+                f"or hybrid search."
+            )
+        if distance != Distance.cosine:
+            raise ValueError(
+                f"TurboQuantVectorDb only supports distance=Distance.cosine; "
+                f"got {distance}. turbovec stores unit-normalized vectors."
+            )
+
+        self.embedder: Embedder = embedder
+        self.dimensions: int = embedder.dimensions
+        self.bit_width = bit_width
+        self.search_type = search_type
+        self.distance = distance
+        self.reranker = reranker
+        self.path: Optional[str] = path
+
+        # Lazy: the underlying IdMapIndex is created by `create()`, not
+        # in __init__. This matches LanceDb's `exists()` contract: a
+        # freshly-constructed store doesn't "exist" until `create()` is
+        # called, and `drop()` returns it to that state.
+        self._index: Optional[IdMapIndex] = None
+        # str doc_id -> set of u64 handles. One-to-many: agno's derived
+        # doc_id is NOT unique (two documents with identical content, or a
+        # repeated explicit doc.id within a batch, derive the same id), and
+        # LanceDb keeps every such row. Mapping one doc_id to a single handle
+        # silently orphaned the earlier vectors — present in search and the
+        # index count but unreachable by id, so undeletable (issue #104).
+        self._str_to_u64: Dict[str, Set[int]] = {}
+        # u64 handle -> stored payload (mirrors LanceDb's "payload" shape)
+        self._u64_to_doc: Dict[int, Dict[str, Any]] = {}
+        # u64 handle assignment counter
+        self._next_u64: int = 0
+        # Auxiliary indexes for O(1) protocol queries
+        self._content_hashes: Set[str] = set()
+        self._name_to_ids: Dict[str, Set[str]] = {}
+
+    # ---- handle allocation ------------------------------------------------
+
+    def _issue_handle(self) -> int:
+        self._next_u64 += 1
+        return self._next_u64
+
+    # ---- VectorDb protocol: lifecycle ------------------------------------
+
+    def create(self) -> None:
+        """Create the underlying index if it doesn't already exist.
+        Idempotent — calling on an already-created store is a no-op.
+
+        If ``path`` was set on the constructor and a previous save exists
+        under it, ``create()`` loads that save; otherwise it instantiates
+        a fresh empty index sized to ``embedder.dimensions``.
+        """
+        if self._index is not None:
+            return
+        # Try loading from path first if one was set; fall through to a
+        # fresh index if the path doesn't contain a previous save.
+        if self.path is not None and Path(self.path).is_dir():
+            try:
+                self._load_from(Path(self.path))
+                return
+            except FileNotFoundError:
+                pass
+        self._index = IdMapIndex(self.dimensions, self.bit_width)
+
+    async def async_create(self) -> None:
+        self.create()
+
+    def drop(self) -> None:
+        """Drop the underlying index. After this call ``exists()`` returns
+        ``False`` until ``create()`` is called again — matches LanceDb's
+        contract where ``drop()`` removes the table entirely."""
+        self._index = None
+        self._str_to_u64.clear()
+        self._u64_to_doc.clear()
+        self._next_u64 = 0
+        self._content_hashes.clear()
+        self._name_to_ids.clear()
+
+    async def async_drop(self) -> None:
+        self.drop()
+
+    def exists(self) -> bool:
+        """True iff the underlying index has been created via ``create()``
+        and not subsequently dropped. Matches LanceDb's
+        "table-exists-in-connection" semantic; *does not* mean
+        "has any documents" — call ``get_count()`` for that."""
+        return self._index is not None
+
+    async def async_exists(self) -> bool:
+        return self.exists()
+
+    def delete(self) -> bool:
+        """Returns ``False``. The Agno protocol declares this abstract
+        method but LanceDb (the drop-in reference) unconditionally
+        returns False — actual destruction goes through ``drop()``."""
+        return False
+
+    def optimize(self) -> None:
+        """No-op. The underlying quantized index doesn't have a
+        post-write optimization step. Matches LanceDb's ``optimize()``
+        which is also a no-op."""
+        return None
+
+    def get_count(self) -> int:
+        """Number of documents currently stored."""
+        if self._index is None:
+            return 0
+        return len(self._index)
+
+    async def async_get_count(self) -> int:
+        return self.get_count()
+
+    # ---- VectorDb protocol: existence checks ------------------------------
+
+    def name_exists(self, name: str) -> bool:
+        if self._index is None:
+            return False
+        return name in self._name_to_ids
+
+    async def async_name_exists(self, name: str) -> bool:
+        # LanceDb raises NotImplementedError here; we have a trivial sync
+        # backing call, so we return the real answer. Intentional deviation.
+        return self.name_exists(name)
+
+    def id_exists(self, id: str) -> bool:
+        if self._index is None:
+            return False
+        return id in self._str_to_u64
+
+    def content_hash_exists(self, content_hash: str) -> bool:
+        if self._index is None:
+            return False
+        return content_hash in self._content_hashes
+
+    # ---- VectorDb protocol: insert / upsert -------------------------------
+
+    @staticmethod
+    def _derive_doc_id(doc: Document, content_hash: str, cleaned_content: str) -> str:
+        """Match LanceDb's id-derivation contract so the same doc with the
+        same content_hash produces the same stable doc_id across stores."""
+        base_id = doc.id or md5(cleaned_content.encode()).hexdigest()
+        return md5(f"{base_id}_{content_hash}".encode()).hexdigest()
+
+    def _embed_missing(self, documents: List[Document]) -> None:
+        """Populate embeddings on any documents that don't have one. Uses
+        the embedder's batch path when available."""
+        to_embed = [
+            doc
+            for doc in documents
+            if doc.embedding is None
+            or (isinstance(doc.embedding, list) and len(doc.embedding) == 0)
+        ]
+        if not to_embed:
+            return
+        if (
+            getattr(self.embedder, "enable_batch", False)
+            and hasattr(self.embedder, "get_embeddings_batch_and_usage")
+        ):
+            contents = [doc.content for doc in to_embed]
+            embeddings, usages = self.embedder.get_embeddings_batch_and_usage(contents)
+            for j, doc in enumerate(to_embed):
+                if j < len(embeddings):
+                    doc.embedding = embeddings[j]
+                    doc.usage = usages[j] if j < len(usages) else None
+        else:
+            for doc in to_embed:
+                doc.embed(embedder=self.embedder)
+
+    async def _embed_missing_async(self, documents: List[Document]) -> None:
+        to_embed = [
+            doc
+            for doc in documents
+            if doc.embedding is None
+            or (isinstance(doc.embedding, list) and len(doc.embedding) == 0)
+        ]
+        if not to_embed:
+            return
+        if (
+            getattr(self.embedder, "enable_batch", False)
+            and hasattr(self.embedder, "async_get_embeddings_batch_and_usage")
+        ):
+            contents = [doc.content for doc in to_embed]
+            embeddings, usages = await self.embedder.async_get_embeddings_batch_and_usage(
+                contents
+            )
+            for j, doc in enumerate(to_embed):
+                if j < len(embeddings):
+                    doc.embedding = embeddings[j]
+                    doc.usage = usages[j] if j < len(usages) else None
+        else:
+            # Embedder has no async batch path — fall back to sync.
+            self._embed_missing(to_embed)
+
+    def insert(
+        self,
+        content_hash: str,
+        documents: List[Document],
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        if not documents:
+            return
+        if self._index is None:
+            # Match LanceDb's "table not initialized" handling: do not
+            # silently auto-create. Callers must invoke create() first.
+            raise RuntimeError(
+                "TurboQuantVectorDb not initialized — call create() before insert()."
+            )
+
+        # Merge `filters` into each document's metadata (matches LanceDb).
+        if filters:
+            for doc in documents:
+                meta = dict(doc.meta_data) if doc.meta_data else {}
+                meta.update(filters)
+                doc.meta_data = meta
+
+        self._embed_missing(documents)
+
+        # Raise on any document that still lacks an embedding rather than
+        # silently dropping — silent drops mask data-pipeline bugs.
+        missing = [doc for doc in documents if not doc.embedding]
+        if missing:
+            ids = [doc.id or "<no id>" for doc in missing]
+            raise ValueError(
+                f"failed to embed {len(missing)} document(s): {ids}"
+            )
+
+        # Batch the entire `documents` list into a single add_with_ids call.
+        # Per-document inserts would invalidate the SIMD-blocked cache
+        # between every doc.
+        vectors = np.asarray([doc.embedding for doc in documents], dtype=np.float32)
+        if vectors.ndim != 2:
+            raise ValueError(
+                f"expected 2D embedding batch, got {vectors.ndim}D"
+            )
+        if vectors.shape[1] != self.dimensions:
+            raise ValueError(
+                f"embedding dim {vectors.shape[1]} does not match "
+                f"index dim {self.dimensions}"
+            )
+        if not vectors.flags["C_CONTIGUOUS"]:
+            vectors = np.ascontiguousarray(vectors)
+
+        handles = np.array(
+            [self._issue_handle() for _ in documents], dtype=np.uint64
+        )
+        self._index.add_with_ids(vectors, handles)
+
+        for doc, handle in zip(documents, handles):
+            cleaned = doc.content.replace("\x00", "�") if doc.content else ""
+            doc_id = self._derive_doc_id(doc, content_hash, cleaned)
+            h = int(handle)
+            self._str_to_u64.setdefault(doc_id, set()).add(h)
+            self._u64_to_doc[h] = {
+                "id": doc_id,
+                "name": doc.name,
+                "content": cleaned,
+                "meta_data": dict(doc.meta_data) if doc.meta_data else {},
+                "usage": doc.usage,
+                "content_id": doc.content_id,
+                "content_hash": content_hash,
+            }
+            self._content_hashes.add(content_hash)
+            if doc.name:
+                self._name_to_ids.setdefault(doc.name, set()).add(doc_id)
+
+    async def async_insert(
+        self,
+        content_hash: str,
+        documents: List[Document],
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        if not documents:
+            return
+        await self._embed_missing_async(documents)
+        # Now every doc should have an embedding; insert delegates to sync.
+        self.insert(content_hash, documents, filters)
+
+    def upsert_available(self) -> bool:
+        return True
+
+    def upsert(
+        self,
+        content_hash: str,
+        documents: List[Document],
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        # Match LanceDb's semantic: replace all documents previously
+        # stored under this content_hash with the incoming batch. Not
+        # "replace by derived doc_id" — that's a different contract.
+        #
+        # Capture the existing generation's handles, run the insert, and
+        # only then drop the old vectors — so a failed insert (dim
+        # mismatch, non-finite embeddings) never destroys the data being
+        # replaced (issue #89). We delete by captured handle rather than
+        # re-querying by content_hash, because insert() re-derives ids
+        # under the SAME content_hash and would otherwise clobber the
+        # just-inserted rows.
+        old_handles = self._handles_for_content_hash(content_hash)
+        self.insert(content_hash, documents, filters)
+        for handle in old_handles:
+            self._remove_handle(handle)
+
+    async def async_upsert(
+        self,
+        content_hash: str,
+        documents: List[Document],
+        filters: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        old_handles = self._handles_for_content_hash(content_hash)
+        await self.async_insert(content_hash, documents, filters)
+        for handle in old_handles:
+            self._remove_handle(handle)
+
+    def _handles_for_content_hash(self, content_hash: str) -> List[int]:
+        """Internal handles of every document currently stored under this
+        content_hash. Used by upsert to defer removal of the previous
+        generation until the replacement add has succeeded (issue #89)."""
+        return [
+            handle
+            for handle, data in self._u64_to_doc.items()
+            if data.get("content_hash") == content_hash
+        ]
+
+    def _remove_handle(self, handle: int) -> None:
+        """Remove a single vector by its internal handle, leaving other
+        handles intact — including ones that share this document's derived
+        id (two distinct documents can map to the same doc_id, matching
+        LanceDb). Cleans the id, name, and content_hash side-indexes only
+        where no surviving handle still needs them."""
+        if self._index is None:
+            return
+        data = self._u64_to_doc.pop(handle, None)
+        if data is None:
+            return
+        self._index.remove(handle)
+        doc_id = data.get("id")
+        # Drop just this handle from the id's handle set; remove the id
+        # entirely only once no handle remains under it.
+        if doc_id is not None:
+            handles = self._str_to_u64.get(doc_id)
+            if handles is not None:
+                handles.discard(handle)
+                if not handles:
+                    del self._str_to_u64[doc_id]
+        # Drop the name->id link only if no surviving handle keeps that
+        # (name, id) pair. The derived doc_id excludes `name`, so two docs
+        # with different names can share an id — matching on id alone would
+        # leave a stale name entry when the last handle for this name goes.
+        name = data.get("name")
+        if name and name in self._name_to_ids:
+            if not any(
+                d.get("id") == doc_id and d.get("name") == name
+                for d in self._u64_to_doc.values()
+            ):
+                self._name_to_ids[name].discard(doc_id)
+                if not self._name_to_ids[name]:
+                    del self._name_to_ids[name]
+        # Drop the content_hash only if no surviving doc carries it.
+        ch = data.get("content_hash")
+        if ch and not any(
+            d.get("content_hash") == ch for d in self._u64_to_doc.values()
+        ):
+            self._content_hashes.discard(ch)
+
+    # ---- VectorDb protocol: search ----------------------------------------
+
+    def _resolve_filter_to_handles(
+        self, filters: Optional[Union[Dict[str, Any], List[Any]]]
+    ) -> Optional[List[int]]:
+        """Convert a dict filter into the list of internal u64 handles
+        whose document's ``meta_data`` matches every key/value pair (AND).
+        Returns ``None`` when no filter was supplied — caller should run
+        an unfiltered search. Returns ``[]`` to mean "no matches".
+
+        Matches LanceDb's dict-filter semantics (exact equality, AND of
+        keys). ``FilterExpr``-style list filters are not yet supported
+        in LanceDb itself, so we silently ignore them here too with a
+        debug log.
+        """
+        if filters is None:
+            return None
+        if isinstance(filters, list):
+            # LanceDb logs a warning and ignores. Mirror that — the
+            # alternative is to error and break callers that pass an
+            # accidental list.
+            return None
+        if not isinstance(filters, dict) or not filters:
+            return None
+        items = list(filters.items())
+        return [
+            handle
+            for handle, data in self._u64_to_doc.items()
+            if all((data.get("meta_data") or {}).get(k) == v for k, v in items)
+        ]
+
+    def _scaled_similarity(self, raw: float) -> float:
+        """Map cosine similarity in ``[-1, 1]`` to ``[0, 1]``. Clamped to
+        absorb the small overshoot caused by quantization noise."""
+        return max(0.0, min(1.0, (raw + 1.0) / 2.0))
+
+    def _build_results(
+        self, scores: np.ndarray, handles: np.ndarray
+    ) -> List[Document]:
+        results: List[Document] = []
+        threshold = self.similarity_threshold
+        for raw_score, handle in zip(scores[0], handles[0]):
+            doc_data = self._u64_to_doc.get(int(handle))
+            if doc_data is None:
+                continue
+            similarity = self._scaled_similarity(float(raw_score))
+            if threshold is not None and similarity < threshold:
+                continue
+            results.append(
+                Document(
+                    id=doc_data["id"],
+                    name=doc_data.get("name"),
+                    content=doc_data.get("content", ""),
+                    meta_data=dict(doc_data.get("meta_data") or {}),
+                    usage=doc_data.get("usage"),
+                    content_id=doc_data.get("content_id"),
+                    # Match LanceDb._build_search_results: thread the
+                    # store's embedder through so downstream code can call
+                    # `doc.embed()` / `doc.async_embed()` on a retrieved
+                    # hit without explicitly passing the embedder back in.
+                    # Without this, `doc.embed()` raises
+                    # "No embedder provided".
+                    embedder=self.embedder,
+                )
+            )
+        return results
+
+    def search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: Optional[Union[Dict[str, Any], List[Any]]] = None,
+    ) -> List[Document]:
+        # An empty query string usually indicates an upstream bug
+        # (uninitialised variable, failed prompt construction). LanceDb
+        # short-circuits this to [] rather than searching with a hash-
+        # derived embedding of "", which would return arbitrary garbage.
+        if not query:
+            return []
+        if self._index is None or len(self._index) == 0:
+            return []
+
+        query_embedding = self.embedder.get_embedding(query)
+        if query_embedding is None:
+            return []
+        qvec = np.asarray(query_embedding, dtype=np.float32)
+        if qvec.ndim == 1:
+            qvec = qvec[None, :]
+        if not qvec.flags["C_CONTIGUOUS"]:
+            qvec = np.ascontiguousarray(qvec)
+
+        allowed_handles = self._resolve_filter_to_handles(filters)
+        if allowed_handles is None:
+            # Unfiltered.
+            k = min(limit, len(self._index))
+            scores, handles = self._index.search(qvec, k)
+        else:
+            if not allowed_handles:
+                return []
+            allowlist = np.asarray(allowed_handles, dtype=np.uint64)
+            scores, handles = self._index.search(qvec, limit, allowlist=allowlist)
+
+        results = self._build_results(scores, handles)
+        if self.reranker is not None and results:
+            results = self.reranker.rerank(query=query, documents=results)
+        return results
+
+    async def async_search(
+        self,
+        query: str,
+        limit: int = 5,
+        filters: Optional[Union[Dict[str, Any], List[Any]]] = None,
+    ) -> List[Document]:
+        if not query:
+            return []
+        if self._index is None or len(self._index) == 0:
+            return []
+
+        if hasattr(self.embedder, "async_get_embedding"):
+            query_embedding = await self.embedder.async_get_embedding(query)
+        else:
+            query_embedding = self.embedder.get_embedding(query)
+        if query_embedding is None:
+            return []
+        qvec = np.asarray(query_embedding, dtype=np.float32)
+        if qvec.ndim == 1:
+            qvec = qvec[None, :]
+        if not qvec.flags["C_CONTIGUOUS"]:
+            qvec = np.ascontiguousarray(qvec)
+
+        allowed_handles = self._resolve_filter_to_handles(filters)
+        if allowed_handles is None:
+            k = min(limit, len(self._index))
+            scores, handles = self._index.search(qvec, k)
+        else:
+            if not allowed_handles:
+                return []
+            allowlist = np.asarray(allowed_handles, dtype=np.uint64)
+            scores, handles = self._index.search(qvec, limit, allowlist=allowlist)
+
+        results = self._build_results(scores, handles)
+        if self.reranker is not None and results:
+            results = self.reranker.rerank(query=query, documents=results)
+        return results
+
+    def get_supported_search_types(self) -> List[SearchType]:
+        # Only vector. Keyword and hybrid would require an external BM25
+        # / lexical index that turbovec doesn't ship. Return shape
+        # mirrors LanceDb: a list of SearchType enum members (not their
+        # `.value` strings).
+        return [SearchType.vector]
+
+    # ---- VectorDb protocol: delete ----------------------------------------
+
+    def delete_by_id(self, id: str) -> bool:
+        if self._index is None:
+            return False
+        handles = self._str_to_u64.get(id)
+        if not handles:
+            return False
+        # Remove every vector sharing this id — a non-unique derived doc_id
+        # can map to several handles. _remove_handle maintains the id, name,
+        # and content_hash side-indexes per handle.
+        for handle in list(handles):
+            self._remove_handle(handle)
+        return True
+
+    def delete_by_name(self, name: str) -> bool:
+        if self._index is None:
+            return False
+        # Remove exactly the handles whose stored name matches. Delegating to
+        # delete_by_id would key on the derived doc_id, which excludes `name`,
+        # so it would also delete a differently-named doc that happens to
+        # share the id. LanceDb deletes rows matching the predicate directly.
+        handles = [h for h, d in self._u64_to_doc.items() if d.get("name") == name]
+        for handle in handles:
+            self._remove_handle(handle)
+        return bool(handles)
+
+    def delete_by_metadata(self, metadata: Dict[str, Any]) -> bool:
+        if self._index is None:
+            return False
+        items = list(metadata.items())
+        # Remove the matching handles directly (see delete_by_name): the
+        # derived doc_id ignores metadata, so delete_by_id would over-delete
+        # distinct docs that collide on the id.
+        handles = [
+            h
+            for h, data in self._u64_to_doc.items()
+            if all((data.get("meta_data") or {}).get(k) == v for k, v in items)
+        ]
+        for handle in handles:
+            self._remove_handle(handle)
+        return bool(handles)
+
+    def delete_by_content_id(self, content_id: str) -> bool:
+        if self._index is None:
+            return False
+        # Remove the matching handles directly (see delete_by_name): the
+        # derived doc_id ignores content_id, so delete_by_id would over-delete
+        # distinct docs that collide on the id.
+        handles = [
+            h
+            for h, data in self._u64_to_doc.items()
+            if data.get("content_id") == content_id
+        ]
+        for handle in handles:
+            self._remove_handle(handle)
+        return bool(handles)
+
+    def update_metadata(self, content_id: str, metadata: Dict[str, Any]) -> None:
+        """Merge ``metadata`` into both ``meta_data`` and the ``filters``
+        payload field of every document whose ``content_id`` matches.
+        Mirrors LanceDb's update_metadata semantic which writes to both
+        fields (used by callers that pass filter-style restrictions at
+        retrieval time)."""
+        if self._index is None:
+            return
+        for data in self._u64_to_doc.values():
+            if data.get("content_id") == content_id:
+                meta = dict(data.get("meta_data") or {})
+                meta.update(metadata)
+                data["meta_data"] = meta
+                filters = data.get("filters")
+                if isinstance(filters, dict):
+                    filters = dict(filters)
+                    filters.update(metadata)
+                    data["filters"] = filters
+                else:
+                    data["filters"] = dict(metadata)
+
+    # ---- Persistence (JSON side-car) --------------------------------------
+
+    def save(self, folder_path: Optional[str] = None) -> None:
+        """Persist the quantized index plus a JSON side-car to disk. Pass
+        ``folder_path`` to override the constructor's ``path=``.
+
+        Writes two files under ``folder_path``:
+          - ``index.tvim`` — the :class:`IdMapIndex` payload.
+          - ``docstore.json`` — JSON-encoded document text, metadata, and
+            id maps. Side-car carries a ``schema_version`` field; loaders
+            reject unknown versions rather than silently misinterpreting
+            bytes.
+        """
+        path = folder_path if folder_path is not None else self.path
+        if path is None:
+            raise ValueError(
+                "No path to save to. Pass `folder_path=` here or set "
+                "`path=` on the constructor."
+            )
+        if self._index is None:
+            raise RuntimeError(
+                "TurboQuantVectorDb has no index to save — call create() first."
+            )
+        folder = Path(path)
+        folder.mkdir(parents=True, exist_ok=True)
+        self._index.write(str(folder / _INDEX_FILENAME))
+        payload = {
+            "schema_version": _DOCSTORE_SCHEMA_VERSION,
+            # Round-trip int handles via list-of-pairs (JSON keys must be
+            # strings, but our handles are ints).
+            "u64_to_doc": [[h, d] for h, d in self._u64_to_doc.items()],
+            "next_u64": self._next_u64,
+            "bit_width": self.bit_width,
+            "dimensions": self.dimensions,
+        }
+        with open(folder / _STORE_FILENAME, "w") as f:
+            json.dump(payload, f)
+
+    def _load_from(self, folder: Path) -> None:
+        side_car = folder / _STORE_FILENAME
+        index_file = folder / _INDEX_FILENAME
+        if not side_car.exists() or not index_file.exists():
+            raise FileNotFoundError(
+                f"missing one of {_STORE_FILENAME}/{_INDEX_FILENAME} under {folder}"
+            )
+        with open(side_car) as f:
+            state = json.load(f)
+        version = state.get("schema_version", 0)
+        if version != _DOCSTORE_SCHEMA_VERSION:
+            raise ValueError(
+                f"{_STORE_FILENAME} has schema_version {version}; this "
+                f"turbovec expects {_DOCSTORE_SCHEMA_VERSION}"
+            )
+        if state.get("dimensions") != self.dimensions:
+            raise ValueError(
+                f"persisted dimensions={state.get('dimensions')} does not "
+                f"match this store's embedder dimensions={self.dimensions}"
+            )
+
+        self._index = IdMapIndex.load(str(index_file))
+        self._u64_to_doc = {int(h): d for h, d in state["u64_to_doc"]}
+        self._next_u64 = int(state["next_u64"])
+
+        # Rebuild reverse indexes from the loaded payload. doc_id is
+        # non-unique, so accumulate handles into a set per id rather than a
+        # dict comprehension (which would drop all but the last handle and
+        # re-orphan the very vectors issue #104 fixed).
+        self._str_to_u64 = {}
+        for handle, data in self._u64_to_doc.items():
+            self._str_to_u64.setdefault(data["id"], set()).add(handle)
+        self._content_hashes = set()
+        self._name_to_ids = {}
+        for data in self._u64_to_doc.values():
+            ch = data.get("content_hash")
+            if ch:
+                self._content_hashes.add(ch)
+            name = data.get("name")
+            if name:
+                self._name_to_ids.setdefault(name, set()).add(data["id"])
+
+
+__all__ = ["TurboQuantVectorDb"]

turbovec-python/python/turbovec/haystack.py

@@ -0,0 +1,747 @@
+"""Haystack DocumentStore backed by turbovec's quantized index.
+
+Install with: ``pip install turbovec[haystack]``.
+
+Implements the Haystack 2.x ``DocumentStore`` protocol and mirrors most
+of ``InMemoryDocumentStore``'s public surface (write/filter/delete,
+``embedding_retrieval``, ``save_to_disk``/``load_from_disk``, pipeline
+``to_dict``/``from_dict``). BM25 (sparse-text) retrieval is not
+implemented — wire an ``InMemoryBM25Retriever`` against a separate
+store if you need keyword search alongside vector search. The
+quantized index discards full-precision embeddings after compression —
+callers that rely on ``Document.embedding`` after retrieval will see
+``None``.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import math
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Literal, Optional, Tuple
+
+import numpy as np
+
+from ._persist import check_persisted_handles
+from ._turbovec import IdMapIndex
+
+try:
+    from haystack import Document
+    from haystack.dataclasses import ByteStream
+    from haystack.dataclasses.sparse_embedding import SparseEmbedding
+    from haystack.document_stores.errors import DuplicateDocumentError
+    from haystack.document_stores.types import DuplicatePolicy
+    from haystack.utils.filters import document_matches_filter
+except ImportError as exc:
+    raise ImportError(
+        "haystack-ai is required to use turbovec.haystack. "
+        "Install with: pip install turbovec[haystack]"
+    ) from exc
+
+
+class TurboQuantDocumentStore:
+    """Haystack DocumentStore backed by a :class:`~turbovec.IdMapIndex`.
+
+    Vectors are quantized to 2–4 bits per dimension. Full-precision
+    embeddings are dropped after quantization — callers requesting
+    ``return_embedding=True`` on retrieval will see ``None`` on the
+    returned documents' ``embedding`` field regardless of the flag.
+
+    Example::
+
+        from turbovec.haystack import TurboQuantDocumentStore
+        from haystack import Document
+
+        store = TurboQuantDocumentStore(dim=1536, bit_width=4)
+        store.write_documents([
+            Document(content="...", embedding=[...], meta={"source": "a"}),
+            ...
+        ])
+        results = store.embedding_retrieval(query_embedding=[...], top_k=5)
+    """
+
+    def __init__(
+        self,
+        dim: Optional[int] = None,
+        bit_width: int = 4,
+        *,
+        embedding_similarity_function: Literal["dot_product", "cosine"] = "cosine",
+        async_executor: Optional[ThreadPoolExecutor] = None,
+        return_embedding: bool = False,
+    ) -> None:
+        """
+        :param dim: Vector dimensionality. When omitted, the underlying
+            quantized index is created lazily by ``IdMapIndex`` itself on
+            the first ``write_documents`` call — matches the no-``dim``
+            ergonomics of ``InMemoryDocumentStore``.
+        :param bit_width: Quantization width per coordinate (2 or 4).
+        :param embedding_similarity_function: ``"cosine"`` (default) or
+            ``"dot_product"``. Used to choose the ``scale_score`` formula
+            during retrieval. Defaults to ``"cosine"`` because turbovec
+            stores unit-normalized vectors.
+        :param async_executor: Optional executor for the ``*_async``
+            methods. If omitted, a single-threaded executor is created
+            and cleaned up on instance destruction.
+        :param return_embedding: Whether retrieval methods should leave
+            the ``embedding`` field populated on returned Documents.
+            turbovec never has the full-precision embedding available, so
+            this is always ``None`` either way; the flag is accepted for
+            API parity with ``InMemoryDocumentStore``.
+        """
+        self._bit_width = bit_width
+        self.embedding_similarity_function = embedding_similarity_function
+        self.return_embedding = return_embedding
+        # IdMapIndex itself supports lazy construction — pass dim through
+        # and let it handle eager vs lazy. No per-store lazy wrapping.
+        self._index = IdMapIndex(dim, bit_width)
+        # Haystack doc_id (str) -> u64 handle
+        self._str_to_u64: Dict[str, int] = {}
+        # u64 handle -> stored doc data {id, content, meta}
+        self._u64_to_doc: Dict[int, Dict[str, Any]] = {}
+        # Counter for assigning u64 handles. Starts at 0; each new
+        # handle is `_next_u64 + 1`, then we bump. Plain int so pickle
+        # can round-trip it directly.
+        self._next_u64: int = 0
+
+        # Executor lifecycle mirrors InMemoryDocumentStore: own one when
+        # the caller didn't pass one in, and shut it down in __del__.
+        self._owns_executor = async_executor is None
+        self.executor = async_executor or ThreadPoolExecutor(
+            thread_name_prefix=f"async-turbovec-docstore-executor-{id(self)}",
+            max_workers=1,
+        )
+
+    def __del__(self) -> None:
+        if (
+            hasattr(self, "_owns_executor")
+            and self._owns_executor
+            and hasattr(self, "executor")
+        ):
+            self.executor.shutdown(wait=True)
+
+    def shutdown(self) -> None:
+        """Explicitly shut down the async executor if this store owns it."""
+        if self._owns_executor:
+            self.executor.shutdown(wait=True)
+
+    def _issue_handle(self) -> int:
+        self._next_u64 += 1
+        return self._next_u64
+
+    @property
+    def storage(self) -> Dict[str, Document]:
+        """Map of ``doc_id -> Document`` for the currently stored documents.
+
+        Documents are reconstructed on every access; the
+        ``embedding`` field is always ``None``.
+        """
+        return {data["id"]: self._reconstruct(data) for data in self._u64_to_doc.values()}
+
+    # ---- DocumentStore protocol ---------------------------------------
+
+    def count_documents(self) -> int:
+        return len(self._str_to_u64)
+
+    def filter_documents(
+        self, filters: Optional[Dict[str, Any]] = None
+    ) -> List[Document]:
+        if filters:
+            self._validate_filters(filters)
+            docs = [
+                self._reconstruct(data)
+                for data in self._u64_to_doc.values()
+                if document_matches_filter(filters=filters, document=self._reconstruct(data))
+            ]
+        else:
+            docs = [self._reconstruct(data) for data in self._u64_to_doc.values()]
+        # `return_embedding` is informational here — we never have the
+        # full-precision embedding to begin with. Kept for parity.
+        return docs
+
+    def write_documents(
+        self,
+        documents: List[Document],
+        policy: DuplicatePolicy = DuplicatePolicy.NONE,
+    ) -> int:
+        # Match InMemoryDocumentStore's input-shape validation rather
+        # than letting a bad input AttributeError on `.embedding`.
+        if (
+            not isinstance(documents, Iterable)
+            or isinstance(documents, str)
+            or any(not isinstance(doc, Document) for doc in documents)
+        ):
+            raise ValueError("Please provide a list of Documents.")
+
+        if policy == DuplicatePolicy.NONE:
+            policy = DuplicatePolicy.FAIL
+
+        # First pass: validate and resolve duplicates according to policy.
+        # Duplicates are resolved against the batch-so-far as well as the
+        # existing store: InMemoryDocumentStore writes into its dict as it
+        # iterates, so a repeated id *within a single call* is resolved the
+        # same way a cross-call repeat would be. Without tracking the batch,
+        # every duplicate row still gets its own vector while _str_to_u64
+        # keeps only the last handle, orphaning the earlier vectors.
+        to_write: List[Document] = []
+        batch_pos: Dict[str, int] = {}  # doc.id -> index into to_write
+        to_remove: List[str] = []  # existing ids to drop, deferred past add
+        written = len(documents)
+        for doc in documents:
+            if doc.embedding is None:
+                raise ValueError(
+                    f"Document {doc.id!r} has no embedding. "
+                    "TurboQuantDocumentStore only stores documents with precomputed "
+                    "embeddings — run an embedder component before writing."
+                )
+            present = doc.id in self._str_to_u64 or doc.id in batch_pos
+            if policy != DuplicatePolicy.OVERWRITE and present:
+                if policy == DuplicatePolicy.FAIL:
+                    raise DuplicateDocumentError(
+                        f"ID '{doc.id}' already exists in the document store."
+                    )
+                if policy == DuplicatePolicy.SKIP:
+                    written -= 1
+                    continue
+            if policy == DuplicatePolicy.OVERWRITE:
+                if doc.id in self._str_to_u64:
+                    # Defer the removal until after the add succeeds so a
+                    # failed validation/add never destroys existing data
+                    # (issue #89).
+                    to_remove.append(doc.id)
+                if doc.id in batch_pos:
+                    # Last write wins: replace the earlier queued document
+                    # in place rather than appending a second vector.
+                    to_write[batch_pos[doc.id]] = doc
+                    continue
+            batch_pos[doc.id] = len(to_write)
+            to_write.append(doc)
+
+        if not to_write:
+            return written
+
+        vectors = np.asarray(
+            [doc.embedding for doc in to_write], dtype=np.float32
+        )
+        if vectors.ndim != 2:
+            raise ValueError(
+                f"expected 2D embedding batch, got {vectors.ndim}D"
+            )
+        # IdMapIndex.add_with_ids handles both eager (dim must match) and
+        # lazy (locks dim on first call) cases. Surface its mismatch
+        # panic as a clean ValueError for parity with previous behaviour.
+        existing_dim = self._index.dim
+        if existing_dim is not None and vectors.shape[1] != existing_dim:
+            raise ValueError(
+                f"embedding dim {vectors.shape[1]} does not match store dim {existing_dim}"
+            )
+        if not vectors.flags["C_CONTIGUOUS"]:
+            vectors = np.ascontiguousarray(vectors)
+
+        handles = np.array(
+            [self._issue_handle() for _ in to_write], dtype=np.uint64
+        )
+        self._index.add_with_ids(vectors, handles)
+
+        # The add succeeded — now it's safe to drop the old vectors for any
+        # overwritten ids. Done before the mapping loop below so _remove_one
+        # resolves the old handle, not the one we're about to assign.
+        for doc_id in to_remove:
+            self._remove_one(doc_id)
+
+        for doc, handle in zip(to_write, handles):
+            h = int(handle)
+            self._str_to_u64[doc.id] = h
+            self._u64_to_doc[h] = {
+                "id": doc.id,
+                "content": doc.content,
+                "meta": dict(doc.meta),
+                "blob": doc.blob,
+                "sparse_embedding": doc.sparse_embedding,
+            }
+        return written
+
+    def delete_documents(self, document_ids: List[str]) -> None:
+        # Haystack's protocol says silently ignore missing ids.
+        for doc_id in document_ids:
+            self._remove_one(doc_id)
+
+    # ---- Utility methods (InMemoryDocumentStore parity) ---------------
+
+    def delete_all_documents(self) -> None:
+        """Delete every document in the store."""
+        for doc_id in list(self._str_to_u64.keys()):
+            self._remove_one(doc_id)
+
+    def update_by_filter(
+        self, filters: Dict[str, Any], meta: Dict[str, Any]
+    ) -> int:
+        """Update metadata on every document matching ``filters``.
+
+        The new ``meta`` is merged into each matching document's existing
+        metadata. Embeddings are not touched — we never had them at full
+        precision anyway. Returns the number of documents updated.
+        """
+        self._validate_filters(filters)
+        updated = 0
+        for data in self._u64_to_doc.values():
+            if document_matches_filter(filters=filters, document=self._reconstruct(data)):
+                data["meta"].update(meta)
+                updated += 1
+        return updated
+
+    def delete_by_filter(self, filters: Dict[str, Any]) -> int:
+        """Delete every document matching ``filters``. Returns the count."""
+        self._validate_filters(filters)
+        matching_ids = [
+            data["id"]
+            for data in self._u64_to_doc.values()
+            if document_matches_filter(filters=filters, document=self._reconstruct(data))
+        ]
+        for doc_id in matching_ids:
+            self._remove_one(doc_id)
+        return len(matching_ids)
+
+    def count_documents_by_filter(self, filters: Dict[str, Any]) -> int:
+        if filters:
+            self._validate_filters(filters)
+            return sum(
+                1
+                for data in self._u64_to_doc.values()
+                if document_matches_filter(filters=filters, document=self._reconstruct(data))
+            )
+        return self.count_documents()
+
+    def count_unique_metadata_by_filter(
+        self, filters: Dict[str, Any], metadata_fields: List[str]
+    ) -> Dict[str, int]:
+        if filters:
+            self._validate_filters(filters)
+            docs_meta = [
+                data["meta"]
+                for data in self._u64_to_doc.values()
+                if document_matches_filter(filters=filters, document=self._reconstruct(data))
+            ]
+        else:
+            docs_meta = [data["meta"] for data in self._u64_to_doc.values()]
+
+        result: Dict[str, int] = {}
+        for field in metadata_fields:
+            key = field.removeprefix("meta.") if field.startswith("meta.") else field
+            values = {meta.get(key) for meta in docs_meta if key in meta and meta[key] is not None}
+            result[key] = len(values)
+        return result
+
+    def get_metadata_fields_info(self) -> Dict[str, Dict[str, str]]:
+        type_map: Dict[str, str] = {}
+        for data in self._u64_to_doc.values():
+            for key, value in data["meta"].items():
+                if value is None:
+                    continue
+                if isinstance(value, bool):
+                    type_map[key] = "boolean"
+                elif isinstance(value, int):
+                    type_map[key] = "int"
+                elif isinstance(value, float):
+                    type_map[key] = "float"
+                else:
+                    type_map[key] = "keyword"
+        return {k: {"type": v} for k, v in type_map.items()}
+
+    def get_metadata_field_min_max(self, metadata_field: str) -> Dict[str, Any]:
+        key = (
+            metadata_field.removeprefix("meta.")
+            if metadata_field.startswith("meta.")
+            else metadata_field
+        )
+        values = [
+            data["meta"][key]
+            for data in self._u64_to_doc.values()
+            if key in data["meta"]
+            and data["meta"][key] is not None
+            and isinstance(data["meta"][key], (int, float, str))
+        ]
+        if not values:
+            return {"min": None, "max": None}
+        try:
+            return {"min": min(values), "max": max(values)}
+        except TypeError:
+            return {"min": None, "max": None}
+
+    def get_metadata_field_unique_values(
+        self, metadata_field: str, search_term: Optional[str] = None
+    ) -> Tuple[List[str], int]:
+        key = (
+            metadata_field.removeprefix("meta.")
+            if metadata_field.startswith("meta.")
+            else metadata_field
+        )
+        if search_term:
+            docs_data = [
+                data
+                for data in self._u64_to_doc.values()
+                if data["content"] and search_term.lower() in data["content"].lower()
+            ]
+        else:
+            docs_data = list(self._u64_to_doc.values())
+        values = sorted(
+            {
+                str(data["meta"][key])
+                for data in docs_data
+                if key in data["meta"] and data["meta"][key] is not None
+            },
+            key=str,
+        )
+        return values, len(values)
+
+    @staticmethod
+    def _validate_filters(filters: Optional[Dict[str, Any]]) -> None:
+        # Match InMemoryDocumentStore (document_store.py:504-509): a
+        # filter dict must have a top-level "operator" (simple comparison
+        # or logical) or "conditions" (compound). A bare "field" without
+        # an operator is malformed and the reference rejects it; we do too.
+        if (
+            filters
+            and "operator" not in filters
+            and "conditions" not in filters
+        ):
+            raise ValueError(
+                "Invalid filter syntax. See https://docs.haystack.deepset.ai/docs/metadata-filtering for details."
+            )
+
+    # ---- Retrieval (not in core protocol but expected) ----------------
+
+    def embedding_retrieval(
+        self,
+        query_embedding: List[float],
+        filters: Optional[Dict[str, Any]] = None,
+        top_k: int = 10,
+        scale_score: bool = False,
+        return_embedding: Optional[bool] = None,
+    ) -> List[Document]:
+        """Return the ``top_k`` documents most similar to ``query_embedding``.
+
+        ``return_embedding=None`` (default) honours the store-level
+        ``return_embedding`` set in the constructor. turbovec never has
+        the full-precision embedding either way — the parameter is here
+        for API parity with ``InMemoryDocumentStore``.
+
+        ``filters`` are resolved to an allowlist before scoring, so the
+        kernel never wastes work on non-matching documents and the result
+        count is always ``min(top_k, n_matches)`` rather than ``< top_k``
+        when the filter is selective.
+        """
+        # `return_embedding` is accepted but we never have the full
+        # embedding to populate; left as-is for signature parity.
+        _ = return_embedding  # noqa: F841
+
+        if self.count_documents() == 0:
+            return []
+
+        qvec = np.asarray(query_embedding, dtype=np.float32)
+        if qvec.ndim == 1:
+            qvec = qvec[None, :]
+        # By this point n_documents > 0, so the index has a committed dim.
+        expected_dim = self._index.dim
+        if qvec.shape[1] != expected_dim:
+            raise ValueError(
+                f"query_embedding dim {qvec.shape[1]} does not match store dim {expected_dim}"
+            )
+        if not qvec.flags["C_CONTIGUOUS"]:
+            qvec = np.ascontiguousarray(qvec)
+
+        if filters is None:
+            fetch_k = min(top_k, self.count_documents())
+            scores, handles = self._index.search(qvec, fetch_k)
+        else:
+            self._validate_filters(filters)
+            # Resolve filter → handle allowlist by walking the in-memory
+            # doc table once. This is the same O(N) cost as the old
+            # post-filter pass, just moved upfront so the kernel can score
+            # only matching vectors.
+            allowed_handles = [
+                handle
+                for handle, data in self._u64_to_doc.items()
+                if document_matches_filter(filters, self._reconstruct(data))
+            ]
+            if not allowed_handles:
+                return []
+            allowlist = np.asarray(allowed_handles, dtype=np.uint64)
+            scores, handles = self._index.search(qvec, top_k, allowlist=allowlist)
+
+        out: List[Document] = []
+        for score, handle in zip(scores[0], handles[0]):
+            data = self._u64_to_doc[int(handle)]
+            out.append(self._reconstruct(data, score=float(score), scale_score=scale_score))
+        return out
+
+    # ---- Async variants ----------------------------------------------
+
+    async def count_documents_async(self) -> int:
+        return self.count_documents()
+
+    async def filter_documents_async(
+        self, filters: Optional[Dict[str, Any]] = None
+    ) -> List[Document]:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor, lambda: self.filter_documents(filters=filters)
+        )
+
+    async def write_documents_async(
+        self,
+        documents: List[Document],
+        policy: DuplicatePolicy = DuplicatePolicy.NONE,
+    ) -> int:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor, lambda: self.write_documents(documents=documents, policy=policy)
+        )
+
+    async def delete_documents_async(self, document_ids: List[str]) -> None:
+        await asyncio.get_running_loop().run_in_executor(
+            self.executor, lambda: self.delete_documents(document_ids=document_ids)
+        )
+
+    async def delete_all_documents_async(self) -> None:
+        await asyncio.get_running_loop().run_in_executor(
+            self.executor, self.delete_all_documents
+        )
+
+    async def update_by_filter_async(
+        self, filters: Dict[str, Any], meta: Dict[str, Any]
+    ) -> int:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor, lambda: self.update_by_filter(filters=filters, meta=meta)
+        )
+
+    async def count_documents_by_filter_async(self, filters: Dict[str, Any]) -> int:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor, lambda: self.count_documents_by_filter(filters=filters)
+        )
+
+    async def count_unique_metadata_by_filter_async(
+        self, filters: Dict[str, Any], metadata_fields: List[str]
+    ) -> Dict[str, int]:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor,
+            lambda: self.count_unique_metadata_by_filter(
+                filters=filters, metadata_fields=metadata_fields
+            ),
+        )
+
+    async def get_metadata_fields_info_async(self) -> Dict[str, Dict[str, str]]:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor, self.get_metadata_fields_info
+        )
+
+    async def get_metadata_field_min_max_async(
+        self, metadata_field: str
+    ) -> Dict[str, Any]:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor,
+            lambda: self.get_metadata_field_min_max(metadata_field=metadata_field),
+        )
+
+    async def get_metadata_field_unique_values_async(
+        self, metadata_field: str, search_term: Optional[str] = None
+    ) -> Tuple[List[str], int]:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor,
+            lambda: self.get_metadata_field_unique_values(
+                metadata_field=metadata_field, search_term=search_term
+            ),
+        )
+
+    async def embedding_retrieval_async(
+        self,
+        query_embedding: List[float],
+        filters: Optional[Dict[str, Any]] = None,
+        top_k: int = 10,
+        scale_score: bool = False,
+        return_embedding: Optional[bool] = None,
+    ) -> List[Document]:
+        return await asyncio.get_running_loop().run_in_executor(
+            self.executor,
+            lambda: self.embedding_retrieval(
+                query_embedding=query_embedding,
+                filters=filters,
+                top_k=top_k,
+                scale_score=scale_score,
+                return_embedding=return_embedding,
+            ),
+        )
+
+    # ---- Serialization (Pipeline to_dict / from_dict) -----------------
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "type": f"{self.__class__.__module__}.{self.__class__.__name__}",
+            "init_parameters": {
+                # `_index.dim` is None on a lazy uncommitted store and an
+                # int once an add has locked the dim — both round-trip cleanly.
+                "dim": self._index.dim,
+                "bit_width": self._bit_width,
+                "embedding_similarity_function": self.embedding_similarity_function,
+                "return_embedding": self.return_embedding,
+            },
+        }
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "TurboQuantDocumentStore":
+        params = data.get("init_parameters", {})
+        return cls(**params)
+
+    # ---- Persistence -------------------------------------------------
+
+    # Side-car schema. Bump when the on-disk shape changes; loader
+    # accepts the current version plus any older versions whose missing
+    # fields we know how to reconstruct (currently v1, written before
+    # blob / sparse_embedding round-trip was added — both default to None
+    # on load).
+    _DOCSTORE_SCHEMA_VERSION = 2
+    _DOCSTORE_SCHEMA_COMPAT = (1, 2)
+
+    def save_to_disk(self, folder_path: str | Path) -> None:
+        """Persist the quantized index plus the Haystack side-car to disk.
+
+        Writes into ``folder_path``:
+          - ``index.tvim`` — the :class:`IdMapIndex` payload. On a lazy
+            store that has never seen a write the file encodes the
+            uncommitted state via a ``dim=0`` sentinel.
+          - ``docstore.json`` — the str-id ↔ Document mapping and store
+            init parameters, JSON-encoded. Document metadata must be
+            JSON-serializable (the same constraint as
+            ``InMemoryDocumentStore.save_to_disk``).
+        """
+        folder = Path(folder_path)
+        folder.mkdir(parents=True, exist_ok=True)
+        self._index.write(str(folder / "index.tvim"))
+        # Keys in `_u64_to_doc` are ints (u64 handles); JSON object keys
+        # must be strings. Serialize as a list of [handle, data] pairs
+        # so we don't lose type fidelity on the round-trip.
+        payload = {
+            "schema_version": self._DOCSTORE_SCHEMA_VERSION,
+            "u64_to_doc": [
+                [h, self._serialize_doc_data(d)] for h, d in self._u64_to_doc.items()
+            ],
+            "next_u64": self._next_u64,
+            "bit_width": self._bit_width,
+            "embedding_similarity_function": self.embedding_similarity_function,
+            "return_embedding": self.return_embedding,
+        }
+        with open(folder / "docstore.json", "w") as f:
+            json.dump(payload, f)
+
+    @classmethod
+    def load_from_disk(
+        cls,
+        folder_path: str | Path,
+    ) -> "TurboQuantDocumentStore":
+        """Reload a store from a folder previously written by
+        :meth:`save_to_disk`. Safe to call on any path — the side-car is
+        plain JSON, never pickle, so there's no deserialization-of-code
+        risk."""
+        folder = Path(folder_path)
+        with open(folder / "docstore.json") as f:
+            state = json.load(f)
+        version = state.get("schema_version", 0)
+        if version not in cls._DOCSTORE_SCHEMA_COMPAT:
+            raise ValueError(
+                f"docstore.json has schema version {version}; "
+                f"this turbovec accepts versions {list(cls._DOCSTORE_SCHEMA_COMPAT)}"
+            )
+        store = cls(
+            bit_width=state["bit_width"],
+            embedding_similarity_function=state.get(
+                "embedding_similarity_function", "cosine"
+            ),
+            return_embedding=state.get("return_embedding", False),
+        )
+        # Reload the index — it carries dim internally (None for lazy
+        # uncommitted, int otherwise).
+        store._index = IdMapIndex.load(str(folder / "index.tvim"))
+        # Reconstruct {int handle: doc data} from the list-of-pairs form.
+        # `_deserialize_doc_data` is shape-tolerant: v1 entries lack the
+        # `blob` / `sparse_embedding` keys and come back with both set to
+        # None, which matches their original on-write state.
+        store._u64_to_doc = {
+            int(h): cls._deserialize_doc_data(d) for h, d in state["u64_to_doc"]
+        }
+        store._next_u64 = state["next_u64"]
+        # Rebuild str_to_u64 from the reloaded doc table.
+        store._str_to_u64 = {
+            data["id"]: handle for handle, data in store._u64_to_doc.items()
+        }
+        check_persisted_handles(store._index, store._u64_to_doc.keys(), what="document")
+        return store
+
+    # ---- Internals ----------------------------------------------------
+
+    def _remove_one(self, doc_id: str) -> bool:
+        handle = self._str_to_u64.pop(doc_id, None)
+        if handle is None:
+            return False
+        del self._u64_to_doc[handle]
+        self._index.remove(handle)
+        return True
+
+    def _reconstruct(
+        self,
+        data: Dict[str, Any],
+        score: Optional[float] = None,
+        scale_score: bool = False,
+    ) -> Document:
+        if score is not None and scale_score:
+            # Match Haystack's InMemoryDocumentStore._compute_query_embedding_similarity_scores
+            # (document_store.py:818-822): different formula per similarity
+            # function. turbovec uses unit-normalized vectors so the cosine
+            # branch is the natural default.
+            if self.embedding_similarity_function == "dot_product":
+                score = 1.0 / (1.0 + math.exp(-score / 100.0))
+            elif self.embedding_similarity_function == "cosine":
+                # Clamp to the exact cosine range before rescaling. Cauchy–Schwarz
+                # bounds the true cosine in [-1, 1], but the LUT scoring kernel's
+                # float-precision noise can land slightly outside that range on
+                # near-identical document/query pairs (e.g. a self-query under the
+                # length-renormalized estimator produces ~1.00016). Clamping
+                # preserves the [0, 1] contract for ``scale_score=True`` consumers.
+                score = (max(-1.0, min(1.0, score)) + 1.0) / 2.0
+        return Document(
+            id=data["id"],
+            content=data["content"],
+            meta=dict(data["meta"]),
+            blob=data.get("blob"),
+            sparse_embedding=data.get("sparse_embedding"),
+            score=score,
+        )
+
+    @staticmethod
+    def _serialize_doc_data(data: Dict[str, Any]) -> Dict[str, Any]:
+        # blob is a ByteStream; sparse_embedding is a SparseEmbedding.
+        # Both have a JSON-safe to_dict() form.
+        blob = data.get("blob")
+        sparse = data.get("sparse_embedding")
+        return {
+            "id": data["id"],
+            "content": data["content"],
+            "meta": data["meta"],
+            "blob": blob.to_dict() if blob is not None else None,
+            "sparse_embedding": sparse.to_dict() if sparse is not None else None,
+        }
+
+    @staticmethod
+    def _deserialize_doc_data(d: Dict[str, Any]) -> Dict[str, Any]:
+        blob = d.get("blob")
+        sparse = d.get("sparse_embedding")
+        return {
+            "id": d["id"],
+            "content": d["content"],
+            "meta": d["meta"],
+            "blob": ByteStream.from_dict(blob) if blob is not None else None,
+            "sparse_embedding": (
+                SparseEmbedding.from_dict(sparse) if sparse is not None else None
+            ),
+        }
+
+
+__all__ = ["TurboQuantDocumentStore"]

turbovec-python/python/turbovec/langchain.py

@@ -0,0 +1,558 @@
+"""LangChain VectorStore backed by turbovec's quantized index.
+
+Install with: ``pip install turbovec[langchain]``.
+
+The public surface mirrors langchain_core's in-tree ``InMemoryVectorStore``
+so this store can be swapped in wherever the in-memory store is used.
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from pathlib import Path
+from typing import Any, Callable, Iterable, Sequence
+
+import numpy as np
+
+from ._dedup import DuplicatePolicy, resolve_duplicates
+from ._persist import check_persisted_handles
+from ._turbovec import IdMapIndex
+
+try:
+    from langchain_core.documents import Document
+    from langchain_core.embeddings import Embeddings
+    from langchain_core.vectorstores import VectorStore
+except ImportError as exc:
+    raise ImportError(
+        "langchain-core is required to use turbovec.langchain. "
+        "Install with: pip install turbovec[langchain]"
+    ) from exc
+
+
+_INDEX_FILENAME = "index.tvim"
+_STORE_FILENAME = "docstore.json"
+# Bump when the docstore.json shape changes; loader refuses to deserialize
+# unknown versions.
+_DOCSTORE_SCHEMA_VERSION = 1
+
+
+class TurboQuantVectorStore(VectorStore):
+    """LangChain VectorStore backed by a :class:`IdMapIndex`.
+
+    Vectors are quantized to 2–4 bits per dimension. A side-car dictionary
+    holds the original text and metadata keyed by document id. Deletion
+    is supported in O(1) per id via the underlying :class:`IdMapIndex`.
+    """
+
+    def __init__(
+        self,
+        embedding: Embeddings,
+        index: IdMapIndex | None = None,
+        *,
+        bit_width: int = 4,
+        docs: dict[str, tuple[str, dict[str, Any]]] | None = None,
+        str_to_u64: dict[str, int] | None = None,
+        next_u64: int = 0,
+    ) -> None:
+        """
+        :param embedding: LangChain ``Embeddings`` instance used to encode
+            documents and queries.
+        :param index: Optional pre-built :class:`IdMapIndex`. When omitted,
+            a lazy ``IdMapIndex`` is created — it commits to a dim on the
+            first add and lets us match the no-arg constructor pattern of
+            langchain_core's ``InMemoryVectorStore``.
+        :param bit_width: Quantization width (2 or 4) used when the index
+            is created from scratch. Ignored if ``index`` is supplied.
+        """
+        self._embedding = embedding
+        # IdMapIndex itself supports lazy construction now — no per-store
+        # lazy wrapping needed. When `index` is None we create a lazy
+        # IdMapIndex(dim=None, bit_width) and let it handle the rest.
+        self._index = index if index is not None else IdMapIndex(bit_width=bit_width)
+        self._docs: dict[str, tuple[str, dict[str, Any]]] = docs if docs is not None else {}
+        self._str_to_u64: dict[str, int] = str_to_u64 if str_to_u64 is not None else {}
+        # Reverse map (u64 handle → str id) kept in sync so search results
+        # can translate handles back to LangChain document ids.
+        self._u64_to_str: dict[int, str] = {
+            handle: sid for sid, handle in self._str_to_u64.items()
+        }
+        self._next_u64: int = next_u64
+
+    def _issue_handle(self) -> int:
+        self._next_u64 += 1
+        return self._next_u64
+
+    @property
+    def embeddings(self) -> Embeddings:
+        return self._embedding
+
+    # ---- Relevance score normalization --------------------------------
+
+    def _select_relevance_score_fn(self) -> Callable[[float], float]:
+        # turbovec returns the raw inner product of unit-normalized vectors —
+        # ideally cosine similarity in [-1, 1]. Quantization noise can
+        # push that very slightly outside the bounds, so clamp after
+        # mapping to LangChain's [0, 1] relevance scale via (sim + 1) / 2.
+        return lambda sim: max(0.0, min(1.0, (sim + 1.0) / 2.0))
+
+    # ---- Write path ---------------------------------------------------
+
+    def add_texts(
+        self,
+        texts: Iterable[str],
+        metadatas: list[dict] | None = None,
+        ids: list[str] | None = None,
+        **_: Any,
+    ) -> list[str]:
+        texts_list = list(texts)
+        if not texts_list:
+            return []
+        if metadatas is None:
+            metadatas = [{} for _ in texts_list]
+        if ids is None:
+            ids = [str(uuid.uuid4()) for _ in texts_list]
+        if len(metadatas) != len(texts_list) or len(ids) != len(texts_list):
+            raise ValueError("texts, metadatas, and ids must all have the same length")
+
+        vectors = np.asarray(self._embedding.embed_documents(texts_list), dtype=np.float32)
+        return self._store_texts_and_vectors(texts_list, vectors, metadatas, ids)
+
+    async def aadd_texts(
+        self,
+        texts: Iterable[str],
+        metadatas: list[dict] | None = None,
+        ids: list[str] | None = None,
+        **_: Any,
+    ) -> list[str]:
+        texts_list = list(texts)
+        if not texts_list:
+            return []
+        if metadatas is None:
+            metadatas = [{} for _ in texts_list]
+        if ids is None:
+            ids = [str(uuid.uuid4()) for _ in texts_list]
+        if len(metadatas) != len(texts_list) or len(ids) != len(texts_list):
+            raise ValueError("texts, metadatas, and ids must all have the same length")
+
+        vectors = np.asarray(
+            await self._embedding.aembed_documents(texts_list), dtype=np.float32
+        )
+        return self._store_texts_and_vectors(texts_list, vectors, metadatas, ids)
+
+    def add_documents(
+        self,
+        documents: list[Document],
+        ids: list[str] | None = None,
+        **kwargs: Any,
+    ) -> list[str]:
+        # Override the base class default which drops the entire `ids` array
+        # if any Document has a None id. The reference InMemoryVectorStore
+        # falls back per-document so partial ids are honoured.
+        texts = [doc.page_content for doc in documents]
+        metadatas = [doc.metadata for doc in documents]
+        if ids is None:
+            ids = [doc.id or str(uuid.uuid4()) for doc in documents]
+        return self.add_texts(texts=texts, metadatas=metadatas, ids=ids, **kwargs)
+
+    async def aadd_documents(
+        self,
+        documents: list[Document],
+        ids: list[str] | None = None,
+        **kwargs: Any,
+    ) -> list[str]:
+        texts = [doc.page_content for doc in documents]
+        metadatas = [doc.metadata for doc in documents]
+        if ids is None:
+            ids = [doc.id or str(uuid.uuid4()) for doc in documents]
+        return await self.aadd_texts(
+            texts=texts, metadatas=metadatas, ids=ids, **kwargs
+        )
+
+    def _store_texts_and_vectors(
+        self,
+        texts_list: list[str],
+        vectors: np.ndarray,
+        metadatas: list[dict],
+        ids: list[str],
+    ) -> list[str]:
+        if vectors.ndim != 2:
+            raise ValueError(f"expected 2D embedding batch, got {vectors.ndim}D")
+
+        # Dedup intra-batch duplicate ids, keeping the last occurrence —
+        # matches InMemoryVectorStore, whose dict store silently overwrites
+        # on a repeated id. Without this every row is added to the index but
+        # _str_to_u64 keeps only the last handle per id, orphaning the
+        # earlier vectors. The returned id list still mirrors the input
+        # (one entry per input text), as the reference does.
+        result_ids = ids
+        keep = resolve_duplicates(ids, DuplicatePolicy.KEEP_LAST)
+        if len(keep) != len(ids):
+            ids = [ids[i] for i in keep]
+            texts_list = [texts_list[i] for i in keep]
+            metadatas = [metadatas[i] for i in keep]
+            vectors = vectors[keep]
+
+        # Validate before mutating any existing data. IdMapIndex.add_with_ids
+        # handles both eager (dim must match) and lazy (locks dim on first
+        # call) cases. Pre-check the eager case so we surface a clean
+        # ValueError rather than a Rust panic.
+        existing_dim = self._index.dim
+        if existing_dim is not None and vectors.shape[1] != existing_dim:
+            raise ValueError(
+                f"embedding dimension {vectors.shape[1]} does not match index dim {existing_dim}"
+            )
+        if not vectors.flags["C_CONTIGUOUS"]:
+            vectors = np.ascontiguousarray(vectors)
+
+        handles = np.array(
+            [self._issue_handle() for _ in texts_list], dtype=np.uint64
+        )
+        # Add first; if encoding rejects the batch (e.g. non-finite values)
+        # this raises before any existing data is touched. Only once the add
+        # has succeeded do we remove the old vectors for colliding ids, so a
+        # failed upsert never destroys existing data (issue #89). Handles are
+        # freshly issued, so the old and new vectors coexist until the delete.
+        self._index.add_with_ids(vectors, handles)
+
+        # Upsert: any id that already existed is removed so the re-added
+        # vector wins. Matches LangChain user expectation that `add_texts`
+        # with an existing id updates in place.
+        duplicates = [i for i in ids if i in self._str_to_u64]
+        if duplicates:
+            self.delete(duplicates)
+
+        for id_, text, meta, handle in zip(ids, texts_list, metadatas, handles):
+            h = int(handle)
+            self._str_to_u64[id_] = h
+            self._u64_to_str[h] = id_
+            self._docs[id_] = (text, dict(meta))
+        return result_ids
+
+    # ---- Read path (similarity search) --------------------------------
+
+    def similarity_search(
+        self,
+        query: str,
+        k: int = 4,
+        filter: dict[str, Any] | Callable[[Document], bool] | None = None,
+        **_: Any,
+    ) -> list[Document]:
+        return [
+            doc
+            for doc, _score in self.similarity_search_with_score(query, k=k, filter=filter)
+        ]
+
+    async def asimilarity_search(
+        self,
+        query: str,
+        k: int = 4,
+        filter: dict[str, Any] | Callable[[Document], bool] | None = None,
+        **_: Any,
+    ) -> list[Document]:
+        return [
+            doc
+            for doc, _score in await self.asimilarity_search_with_score(
+                query, k=k, filter=filter
+            )
+        ]
+
+    def similarity_search_with_score(
+        self,
+        query: str,
+        k: int = 4,
+        filter: dict[str, Any] | Callable[[Document], bool] | None = None,
+        **_: Any,
+    ) -> list[tuple[Document, float]]:
+        qvec = np.asarray(self._embedding.embed_query(query), dtype=np.float32)
+        return self._search_vector(qvec, k, filter=filter)
+
+    async def asimilarity_search_with_score(
+        self,
+        query: str,
+        k: int = 4,
+        filter: dict[str, Any] | Callable[[Document], bool] | None = None,
+        **_: Any,
+    ) -> list[tuple[Document, float]]:
+        qvec = np.asarray(
+            await self._embedding.aembed_query(query), dtype=np.float32
+        )
+        return self._search_vector(qvec, k, filter=filter)
+
+    def similarity_search_by_vector(
+        self,
+        embedding: list[float],
+        k: int = 4,
+        filter: dict[str, Any] | Callable[[Document], bool] | None = None,
+        **_: Any,
+    ) -> list[Document]:
+        qvec = np.asarray(embedding, dtype=np.float32)
+        return [doc for doc, _score in self._search_vector(qvec, k, filter=filter)]
+
+    async def asimilarity_search_by_vector(
+        self,
+        embedding: list[float],
+        k: int = 4,
+        filter: dict[str, Any] | Callable[[Document], bool] | None = None,
+        **_: Any,
+    ) -> list[Document]:
+        # The search itself is sync (no embedding step). Delegate.
+        return self.similarity_search_by_vector(embedding, k=k, filter=filter)
+
+    def _search_vector(
+        self,
+        qvec: np.ndarray,
+        k: int,
+        filter: dict[str, Any] | Callable[[Document], bool] | None = None,
+    ) -> list[tuple[Document, float]]:
+        if qvec.ndim == 1:
+            qvec = qvec[None, :]
+        if not qvec.flags["C_CONTIGUOUS"]:
+            qvec = np.ascontiguousarray(qvec)
+        # IdMapIndex handles the lazy-uncommitted case internally (returns
+        # empty search results). A len-zero check covers both that and
+        # the eager-but-empty case.
+        if len(self._index) == 0:
+            return []
+
+        if filter is None:
+            search_k = min(k, len(self._index))
+            scores, handles = self._index.search(qvec, search_k)
+        else:
+            predicate = self._compile_filter(filter)
+            allowed_handles = [
+                self._str_to_u64[sid]
+                for sid, (text, meta) in self._docs.items()
+                if predicate(Document(id=sid, page_content=text, metadata=dict(meta)))
+            ]
+            if not allowed_handles:
+                return []
+            allowlist = np.asarray(allowed_handles, dtype=np.uint64)
+            scores, handles = self._index.search(qvec, k, allowlist=allowlist)
+
+        results: list[tuple[Document, float]] = []
+        for score, handle in zip(scores[0], handles[0]):
+            sid = self._u64_to_str[int(handle)]
+            text, meta = self._docs[sid]
+            results.append(
+                (Document(id=sid, page_content=text, metadata=dict(meta)), float(score))
+            )
+        return results
+
+    @staticmethod
+    def _compile_filter(
+        filter: dict[str, Any] | Callable[[Document], bool],
+    ) -> Callable[[Document], bool]:
+        # Match the in-tree InMemoryVectorStore convention: callable filters
+        # receive a Document, not a metadata dict
+        # (langchain_core/vectorstores/in_memory.py).
+        if callable(filter):
+            return filter
+        if isinstance(filter, dict):
+            items = list(filter.items())
+            return lambda doc: all(doc.metadata.get(k) == v for k, v in items)
+        raise TypeError(
+            "filter must be a dict of metadata key/value pairs or a callable "
+            f"taking a Document, got {type(filter).__name__}"
+        )
+
+    # ---- Max marginal relevance ---------------------------------------
+    #
+    # MMR requires the full-precision vector of every candidate to compute
+    # pairwise diversity scores. turbovec discards full vectors after
+    # quantization (that's the point), so we can't faithfully implement
+    # MMR. Raise loudly with a useful message rather than silently fall
+    # back to the base class's bare NotImplementedError.
+
+    _MMR_MSG = (
+        "TurboQuantVectorStore does not support max-marginal-relevance "
+        "search because the underlying quantized index discards "
+        "full-precision vectors after compression. MMR requires the "
+        "original embedding for every candidate to compute pairwise "
+        "diversity. Use `similarity_search` / `similarity_search_with_score` "
+        "instead, or maintain a parallel store with full-precision "
+        "embeddings if you need MMR specifically."
+    )
+
+    def max_marginal_relevance_search(
+        self,
+        query: str,
+        k: int = 4,
+        fetch_k: int = 20,
+        lambda_mult: float = 0.5,
+        **kwargs: Any,
+    ) -> list[Document]:
+        raise NotImplementedError(self._MMR_MSG)
+
+    def max_marginal_relevance_search_by_vector(
+        self,
+        embedding: list[float],
+        k: int = 4,
+        fetch_k: int = 20,
+        lambda_mult: float = 0.5,
+        *,
+        filter: Callable[[Document], bool] | None = None,
+        **kwargs: Any,
+    ) -> list[Document]:
+        raise NotImplementedError(self._MMR_MSG)
+
+    async def amax_marginal_relevance_search(
+        self,
+        query: str,
+        k: int = 4,
+        fetch_k: int = 20,
+        lambda_mult: float = 0.5,
+        **kwargs: Any,
+    ) -> list[Document]:
+        raise NotImplementedError(self._MMR_MSG)
+
+    # ---- Get / delete -------------------------------------------------
+
+    def get_by_ids(self, ids: Sequence[str], /) -> list[Document]:
+        """Return Documents for the given ids. Missing ids are silently skipped
+        (matches the InMemoryVectorStore reference)."""
+        out: list[Document] = []
+        for sid in ids:
+            if sid in self._docs:
+                text, meta = self._docs[sid]
+                out.append(Document(id=sid, page_content=text, metadata=dict(meta)))
+        return out
+
+    async def aget_by_ids(self, ids: Sequence[str], /) -> list[Document]:
+        return self.get_by_ids(ids)
+
+    def delete(self, ids: list[str] | None = None, **_: Any) -> None:
+        """Remove documents by id. Missing ids are silently skipped — matches
+        the InMemoryVectorStore reference (which also accepts ``ids=None``
+        as a no-op)."""
+        if not ids:
+            return
+        for sid in ids:
+            handle = self._str_to_u64.pop(sid, None)
+            if handle is None:
+                continue
+            self._u64_to_str.pop(handle, None)
+            self._docs.pop(sid, None)
+            self._index.remove(handle)
+
+    async def adelete(self, ids: list[str] | None = None, **_: Any) -> None:
+        self.delete(ids)
+
+    # ---- Construction helpers -----------------------------------------
+
+    @classmethod
+    def from_texts(
+        cls,
+        texts: list[str],
+        embedding: Embeddings,
+        metadatas: list[dict] | None = None,
+        *,
+        bit_width: int = 4,
+        ids: list[str] | None = None,
+        **_: Any,
+    ) -> "TurboQuantVectorStore":
+        # The underlying index is created lazily on the first `add_texts`
+        # call, picking up `dim` from the first batch of embeddings — same
+        # no-`dim` ergonomics as InMemoryVectorStore.
+        store = cls(embedding=embedding, bit_width=bit_width)
+        if texts:
+            store.add_texts(texts, metadatas=metadatas, ids=ids)
+        return store
+
+    @classmethod
+    async def afrom_texts(
+        cls,
+        texts: list[str],
+        embedding: Embeddings,
+        metadatas: list[dict] | None = None,
+        *,
+        bit_width: int = 4,
+        ids: list[str] | None = None,
+        **_: Any,
+    ) -> "TurboQuantVectorStore":
+        store = cls(embedding=embedding, bit_width=bit_width)
+        if texts:
+            await store.aadd_texts(texts, metadatas=metadatas, ids=ids)
+        return store
+
+    # ---- Persistence --------------------------------------------------
+    #
+    # Method names match the InMemoryVectorStore reference (`dump`/`load`),
+    # but the on-disk layout is a folder containing the binary index file
+    # plus a JSON side-car (we can't embed the binary Rust index in a
+    # single JSON file the way the reference does with its raw-vector
+    # store).
+
+    def dump(self, folder_path: str | Path) -> None:
+        """Persist the quantized index plus the side-car to disk.
+
+        ``folder_path`` is a directory; turbovec writes ``index.tvim``
+        and ``docstore.json`` inside it. Document metadata must be
+        JSON-serializable (same constraint as ``InMemoryVectorStore``).
+        A lazy uncommitted index encodes its state via the index file's
+        own ``dim = 0`` sentinel; no special-case handling needed here.
+        """
+        folder = Path(folder_path)
+        folder.mkdir(parents=True, exist_ok=True)
+        self._index.write(str(folder / _INDEX_FILENAME))
+        # `_docs` stores tuples `(text, metadata)` — JSON would drop the
+        # tuple-ness on round-trip, so serialize each entry as an explicit
+        # `{"text": ..., "metadata": ...}` dict.
+        docs_payload = {
+            sid: {"text": text, "metadata": meta}
+            for sid, (text, meta) in self._docs.items()
+        }
+        payload = {
+            "schema_version": _DOCSTORE_SCHEMA_VERSION,
+            "docs": docs_payload,
+            "str_to_u64": self._str_to_u64,
+            "next_u64": self._next_u64,
+            # Pull bit_width off the live index — same value whether
+            # the index was constructed eagerly or lazily.
+            "bit_width": self._index.bit_width,
+        }
+        with open(folder / _STORE_FILENAME, "w") as f:
+            json.dump(payload, f)
+
+    @classmethod
+    def load(
+        cls,
+        folder_path: str | Path,
+        embedding: Embeddings,
+    ) -> "TurboQuantVectorStore":
+        """Reload a store from a folder previously written by :meth:`dump`.
+        Safe to call on any path — the side-car is plain JSON, never
+        pickle, so there's no deserialization-of-code risk."""
+        folder = Path(folder_path)
+        with open(folder / _STORE_FILENAME) as f:
+            state = json.load(f)
+        version = state.get("schema_version", 0)
+        if version != _DOCSTORE_SCHEMA_VERSION:
+            raise ValueError(
+                f"docstore.json has schema version {version}; "
+                f"this turbovec expects version {_DOCSTORE_SCHEMA_VERSION}"
+            )
+        # IdMapIndex.load handles the dim=0 (lazy-uncommitted) sentinel
+        # internally and reconstructs the index in the right state.
+        index = IdMapIndex.load(str(folder / _INDEX_FILENAME))
+        # Rehydrate `_docs` from the explicit `{"text", "metadata"}` form
+        # back into the internal tuple representation.
+        docs = {
+            sid: (entry["text"], entry["metadata"])
+            for sid, entry in state["docs"].items()
+        }
+        # JSON object keys are strings; the str_to_u64 values are already
+        # ints in the payload, just need to confirm.
+        str_to_u64 = {sid: int(h) for sid, h in state["str_to_u64"].items()}
+        check_persisted_handles(index, str_to_u64.values(), what="document")
+        return cls(
+            embedding=embedding,
+            index=index,
+            bit_width=state.get("bit_width", 4),
+            docs=docs,
+            str_to_u64=str_to_u64,
+            next_u64=int(state["next_u64"]),
+        )
+
+
+__all__ = ["TurboQuantVectorStore"]

turbovec-python/python/turbovec/llama_index.py

@@ -0,0 +1,678 @@
+"""LlamaIndex VectorStore backed by turbovec's quantized index.
+
+Install with: ``pip install turbovec[llama-index]``.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Any, List, Optional, Sequence
+
+import numpy as np
+
+from ._dedup import DuplicatePolicy, resolve_duplicates
+from ._persist import check_persisted_handles
+from ._turbovec import IdMapIndex
+
+try:
+    from llama_index.core.bridge.pydantic import PrivateAttr
+    from llama_index.core.schema import (
+        BaseNode,
+        NodeRelationship,
+        RelatedNodeInfo,
+        TextNode,
+    )
+    from llama_index.core.vector_stores.types import (
+        BasePydanticVectorStore,
+        FilterCondition,
+        FilterOperator,
+        MetadataFilter,
+        MetadataFilters,
+        VectorStoreQuery,
+        VectorStoreQueryMode,
+        VectorStoreQueryResult,
+    )
+    from llama_index.core.vector_stores.utils import (
+        metadata_dict_to_node,
+        node_to_metadata_dict,
+    )
+except ImportError as exc:
+    raise ImportError(
+        "llama-index-core is required to use turbovec.llama_index. "
+        "Install with: pip install turbovec[llama-index]"
+    ) from exc
+
+
+# Persistence layout: a single user-facing ``persist_path`` (the file the
+# framework asks us to write) is split into two files by extension —
+# ``{base}.tvim`` for the binary IdMapIndex and ``{base}.nodes.json`` for
+# the node side-car. Both live next to each other so the layout fits the
+# directory-of-namespaced-files pattern used by StorageContext.
+_INDEX_EXT = ".tvim"
+_STORE_EXT = ".nodes.json"
+# Filename template used by SimpleVectorStore for namespace lookup —
+# mirrored here so `from_persist_dir` works the same way.
+_NAMESPACE_SEP = "__"
+_DEFAULT_PERSIST_FNAME = "vector_store.json"
+_DEFAULT_VECTOR_STORE = "default"
+# Bump when the nodes.json shape changes; loader accepts the current
+# version plus any older versions whose missing fields we know how to
+# reconstruct (currently v1, written before full-node round-trip was
+# added — v1 entries are reconstructed as bare TextNodes with only
+# text + metadata + SOURCE relationship, matching the original
+# lossy behaviour rather than failing to load).
+_NODES_SCHEMA_VERSION = 2
+_NODES_SCHEMA_COMPAT = (1, 2)
+
+
+def _split_persist_base(persist_path: str | Path) -> Path:
+    """Strip the framework-provided extension off `persist_path` so the
+    binary index and JSON side-car can sit next to each other under a
+    shared base. We then append our own extensions in persist / load."""
+    p = Path(persist_path)
+    # Use the path without its suffix so both .tvim and .nodes.json share
+    # a base. If the input has no suffix (e.g. a bare folder-like name),
+    # use it as-is.
+    return p.with_suffix("") if p.suffix else p
+
+
+class TurboQuantVectorStore(BasePydanticVectorStore):
+    """LlamaIndex VectorStore backed by a :class:`IdMapIndex`.
+
+    Vectors are quantized to 2–4 bits per dimension. A side-car dictionary
+    holds node text and metadata keyed by ``node_id``. Supports ``delete``
+    (by ``ref_doc_id``, removing every node with that ref) and
+    ``delete_nodes`` (by ``node_id``) — both O(1) per node.
+    """
+
+    stores_text: bool = True
+    is_embedding_query: bool = True
+    flat_metadata: bool = False
+
+    _index: Any = PrivateAttr()
+    _nodes: dict[str, dict[str, Any]] = PrivateAttr()
+    _node_id_to_u64: dict[str, int] = PrivateAttr()
+    _u64_to_node_id: dict[int, str] = PrivateAttr()
+    _next_u64: int = PrivateAttr()
+
+    def __init__(self, index: IdMapIndex | None = None, *, bit_width: int = 4, **kwargs: Any) -> None:
+        """Construct the vector store.
+
+        :param index: Optional pre-built :class:`IdMapIndex`. When omitted,
+            a lazy ``IdMapIndex`` is created — it commits to a dim on the
+            first add and lets callers use the no-arg construction pattern
+            common to LlamaIndex's other vector stores (e.g. via
+            ``StorageContext.from_defaults(vector_store=TurboQuantVectorStore())``).
+        :param bit_width: Quantization width used when constructing the
+            lazy index. Ignored if ``index`` is supplied.
+        """
+        super().__init__(**kwargs)
+        # IdMapIndex itself supports lazy construction now — no per-store
+        # lazy wrapping needed.
+        self._index = index if index is not None else IdMapIndex(bit_width=bit_width)
+        self._nodes = {}
+        self._node_id_to_u64 = {}
+        self._u64_to_node_id = {}
+        self._next_u64 = 0
+
+    def _issue_handle(self) -> int:
+        self._next_u64 += 1
+        return self._next_u64
+
+    @classmethod
+    def class_name(cls) -> str:
+        return "TurboQuantVectorStore"
+
+    @classmethod
+    def from_params(cls, dim: int | None = None, bit_width: int = 4) -> "TurboQuantVectorStore":
+        """Build a store with a known ``dim`` (eager) or lazy when ``dim``
+        is omitted."""
+        return cls(index=IdMapIndex(dim, bit_width))
+
+    @property
+    def client(self) -> IdMapIndex:
+        return self._index
+
+    def add(self, nodes: list[BaseNode], **_: Any) -> list[str]:
+        if not nodes:
+            return []
+
+        # Reject intra-batch duplicates loudly. Letting them through would
+        # leave the index with N vectors but only the last node_id mapped
+        # back to one of them — the earlier handles become orphans that
+        # `query` later resolves through the duplicate node_id, returning
+        # the second node's payload attached to the first node's vector.
+        # Caller's job to deduplicate before calling add.
+        node_ids = [n.node_id for n in nodes]
+        try:
+            resolve_duplicates(node_ids, DuplicatePolicy.REJECT)
+        except ValueError:
+            seen: set[str] = set()
+            dup = next(nid for nid in node_ids if nid in seen or seen.add(nid))
+            raise ValueError(
+                f"duplicate node_id {dup!r} appears multiple times "
+                "in the input batch; deduplicate before calling add()"
+            ) from None
+
+        embeddings = [node.get_embedding() for node in nodes]
+        vectors = np.asarray(embeddings, dtype=np.float32)
+        if vectors.ndim != 2:
+            raise ValueError(
+                f"expected 2D embedding batch, got {vectors.ndim}D"
+            )
+        # IdMapIndex.add_with_ids handles eager (dim must match) and lazy
+        # (locks dim on first add) — pre-check the eager case so we
+        # surface a clean ValueError rather than a Rust panic.
+        existing_dim = self._index.dim
+        if existing_dim is not None and vectors.shape[1] != existing_dim:
+            raise ValueError(
+                f"node embedding dim {vectors.shape[1]} does not match index dim {existing_dim}"
+            )
+        if not vectors.flags["C_CONTIGUOUS"]:
+            vectors = np.ascontiguousarray(vectors)
+
+        handles = np.array([self._issue_handle() for _ in nodes], dtype=np.uint64)
+        # Add first; if validation above or encoding here (e.g. non-finite
+        # values) rejects the batch, it raises before any existing data is
+        # touched. Only after the add succeeds do we remove the old entries
+        # for colliding node_ids, so a failed upsert never destroys existing
+        # data (issue #89). Handles are freshly issued, so the old and new
+        # vectors coexist until the delete.
+        self._index.add_with_ids(vectors, handles)
+
+        # Upsert-like: if a node_id is already present in the STORE, remove
+        # the old entry so the new embedding wins.
+        duplicates = [n.node_id for n in nodes if n.node_id in self._node_id_to_u64]
+        for node_id in duplicates:
+            self._remove_node_by_id(node_id)
+
+        ids: list[str] = []
+        for node, handle in zip(nodes, handles):
+            h = int(handle)
+            nid = node.node_id
+            self._node_id_to_u64[nid] = h
+            self._u64_to_node_id[h] = nid
+            # `metadata` and `ref_doc_id` are kept at top level for fast
+            # filter / doc-id lookup (queries hit these on every hit;
+            # parsing _node_content per hit would be wasteful). `node_dict`
+            # is the framework's canonical metadata representation
+            # (`_node_content` + `_node_type` + original metadata keys),
+            # which `metadata_dict_to_node` reconstructs into a full
+            # BaseNode — preserving relationships (PREVIOUS / NEXT /
+            # PARENT / CHILD), excluded_*_metadata_keys, template fields,
+            # start/end_char_idx and mimetype on retrieval. The narrow
+            # `{text, metadata, ref_doc_id}` schema we used to keep lost
+            # all of those silently.
+            self._nodes[nid] = {
+                "metadata": dict(node.metadata),
+                "ref_doc_id": node.ref_doc_id,
+                "node_dict": node_to_metadata_dict(
+                    node, remove_text=False, flat_metadata=False
+                ),
+            }
+            ids.append(nid)
+        return ids
+
+    def delete(self, ref_doc_id: str, **_: Any) -> None:
+        """Delete every node whose ``ref_doc_id`` matches."""
+        matching = [
+            nid for nid, data in self._nodes.items() if data.get("ref_doc_id") == ref_doc_id
+        ]
+        for nid in matching:
+            self._remove_node_by_id(nid)
+
+    def delete_nodes(
+        self,
+        node_ids: Optional[List[str]] = None,
+        filters: Optional[MetadataFilters] = None,
+        **_: Any,
+    ) -> None:
+        """Delete every node matching ``node_ids`` and/or ``filters``. Both
+        constraints intersect when supplied. Missing node_ids are ignored.
+        Matches the signature and semantics of ``SimpleVectorStore.delete_nodes``.
+        """
+        if not node_ids and filters is None:
+            return
+        candidates = list(self._nodes.items())
+        if node_ids is not None:
+            node_id_set = set(node_ids)
+            candidates = [(nid, data) for nid, data in candidates if nid in node_id_set]
+        if filters is not None:
+            candidates = [
+                (nid, data)
+                for nid, data in candidates
+                if self._filters_match(data["metadata"], filters)
+            ]
+        for nid, _data in candidates:
+            self._remove_node_by_id(nid)
+
+    def clear(self) -> None:
+        """Drop every node from the store and reset to a fresh lazy index.
+
+        The new index keeps the same ``bit_width`` so subsequent adds
+        commit a new ``dim`` lazily.
+        """
+        bw = self._index.bit_width
+        self._index = IdMapIndex(bit_width=bw)
+        self._nodes = {}
+        self._node_id_to_u64 = {}
+        self._u64_to_node_id = {}
+        self._next_u64 = 0
+
+    def get(self, text_id: str) -> List[float]:
+        """LlamaIndex's protocol expects this to return the full-precision
+        embedding for a given node id. turbovec discards full-precision
+        embeddings after quantization, so we raise loudly with an
+        explanation rather than return a lossy reconstruction or zeroes.
+        """
+        raise NotImplementedError(
+            "TurboQuantVectorStore.get(text_id) cannot return the original "
+            "embedding because turbovec quantizes vectors to 2-4 bits per "
+            "dimension and discards full precision after encoding. Keep a "
+            "parallel docstore if you need the raw embedding."
+        )
+
+    def get_nodes(
+        self,
+        node_ids: Optional[List[str]] = None,
+        filters: Optional[MetadataFilters] = None,
+    ) -> List[BaseNode]:
+        """Return the nodes matching ``node_ids`` and/or ``filters``. Both
+        constraints intersect when supplied; missing node_ids are
+        silently skipped.
+
+        Unlike ``SimpleVectorStore`` (which raises NotImplementedError
+        here because it doesn't store nodes), turbovec keeps node text
+        and metadata in a side-car so this can return populated
+        ``TextNode`` objects directly.
+        """
+        candidates = list(self._nodes.items())
+        if node_ids is not None:
+            node_id_set = set(node_ids)
+            candidates = [(nid, data) for nid, data in candidates if nid in node_id_set]
+        if filters is not None:
+            candidates = [
+                (nid, data)
+                for nid, data in candidates
+                if self._filters_match(data["metadata"], filters)
+            ]
+        return [self._reconstruct_node(nid, data) for nid, data in candidates]
+
+    @staticmethod
+    def _reconstruct_node(nid: str, data: dict[str, Any]) -> BaseNode:
+        # v2 entries carry `node_dict` — round-trip via the framework's
+        # own helper so we get the full BaseNode subclass back
+        # (TextNode / IndexNode / ImageNode) with every field populated.
+        if "node_dict" in data:
+            return metadata_dict_to_node(data["node_dict"])
+        # v1 fallback: stores that were persisted before the full-node
+        # round-trip landed only have {text, metadata, ref_doc_id}.
+        # Reconstruct the minimum-fidelity TextNode they used to produce
+        # so old on-disk stores keep loading without manual migration.
+        node = TextNode(
+            id_=nid,
+            text=data["text"],
+            metadata=dict(data["metadata"]),
+        )
+        if data.get("ref_doc_id") is not None:
+            node.relationships[NodeRelationship.SOURCE] = RelatedNodeInfo(
+                node_id=data["ref_doc_id"]
+            )
+        return node
+
+    def _remove_node_by_id(self, node_id: str) -> bool:
+        handle = self._node_id_to_u64.pop(node_id, None)
+        if handle is None:
+            return False
+        self._u64_to_node_id.pop(handle, None)
+        self._nodes.pop(node_id, None)
+        self._index.remove(handle)
+        return True
+
+    def _resolve_allowed_handles(
+        self,
+        filters: MetadataFilters | None,
+        node_ids: list[str] | None,
+        doc_ids: list[str] | None,
+    ) -> list[int]:
+        """Resolve ``query.filters``, ``query.node_ids`` and ``query.doc_ids``
+        to the list of internal u64 handles that satisfy the filter. Empty
+        list means no node matches.
+
+        Semantics (matching the SimpleVectorStore reference where applicable):
+          - ``node_ids``: filter by node_id (set membership).
+          - ``doc_ids``: filter by ``ref_doc_id`` only (source document id).
+          - ``filters``: apply metadata filters.
+        All three intersect when more than one is supplied.
+        """
+        candidates = self._nodes.items()
+
+        if node_ids:
+            node_id_set = set(node_ids)
+            candidates = [(nid, data) for nid, data in candidates if nid in node_id_set]
+
+        if doc_ids:
+            doc_id_set = set(doc_ids)
+            candidates = [
+                (nid, data)
+                for nid, data in candidates
+                if data.get("ref_doc_id") in doc_id_set
+            ]
+
+        if filters is None:
+            return [self._node_id_to_u64[nid] for nid, _ in candidates]
+
+        return [
+            self._node_id_to_u64[nid]
+            for nid, data in candidates
+            if self._filters_match(data["metadata"], filters)
+        ]
+
+    @classmethod
+    def _filters_match(
+        cls, metadata: dict[str, Any], filters: MetadataFilters
+    ) -> bool:
+        condition = getattr(filters, "condition", None) or FilterCondition.AND
+        results: list[bool] = []
+        for f in filters.filters:
+            if isinstance(f, MetadataFilters):
+                results.append(cls._filters_match(metadata, f))
+            else:
+                results.append(cls._single_filter_match(metadata, f))
+        if condition == FilterCondition.AND:
+            return all(results) if results else True
+        if condition == FilterCondition.OR:
+            return any(results) if results else True
+        if condition == FilterCondition.NOT:
+            # Reference semantics (`build_metadata_filter_fn`,
+            # `utils.py:187-189`): NOT matches when none of the inner
+            # filters match. Empty inner list trivially satisfies NOT.
+            return not any(results)
+        raise NotImplementedError(
+            f"filter condition {condition!r} not supported by TurboQuantVectorStore"
+        )
+
+    @staticmethod
+    def _single_filter_match(metadata: dict[str, Any], f: MetadataFilter) -> bool:
+        # Semantics mirror SimpleVectorStore's _build_metadata_filter_fn
+        # (llama_index/core/vector_stores/simple.py) so that filtered
+        # results agree with the in-tree reference store.
+        op = f.operator
+        target = f.value
+        value = metadata.get(f.key)
+
+        # IS_EMPTY is the only operator that treats a missing key as a hit.
+        if op == FilterOperator.IS_EMPTY:
+            return value is None or value == "" or value == []
+
+        # Every other operator returns False when the key is absent — this
+        # matches the reference implementation (notably NE returns False on
+        # missing, not True).
+        if value is None:
+            return False
+
+        if op == FilterOperator.EQ:
+            return value == target
+        if op == FilterOperator.NE:
+            return value != target
+        if op == FilterOperator.GT:
+            return value > target
+        if op == FilterOperator.LT:
+            return value < target
+        if op == FilterOperator.GTE:
+            return value >= target
+        if op == FilterOperator.LTE:
+            return value <= target
+        if op == FilterOperator.IN:
+            return value in target
+        if op == FilterOperator.NIN:
+            return value not in target
+        if op == FilterOperator.CONTAINS:
+            return target in value
+        if op == FilterOperator.TEXT_MATCH:
+            # Reference (`utils.py:138-144`): case-SENSITIVE substring,
+            # both sides must be strings. Previous turbovec impl
+            # lowercased both sides — a silent semantic divergence that
+            # caused our results to disagree with SimpleVectorStore on
+            # mixed-case keys.
+            if isinstance(target, str) and isinstance(value, str):
+                return target in value
+            raise TypeError(
+                "Both metadata value and filter value must be strings "
+                "for the TEXT_MATCH operator"
+            )
+        if op == FilterOperator.TEXT_MATCH_INSENSITIVE:
+            if isinstance(target, str) and isinstance(value, str):
+                return target.lower() in value.lower()
+            raise TypeError(
+                "Both metadata value and filter value must be strings "
+                "for the TEXT_MATCH_INSENSITIVE operator"
+            )
+        if op == FilterOperator.ALL:
+            # Reference (`utils.py:152-153`): every element of `target`
+            # must be present in the metadata value (which is typically
+            # a list — tag-set matching).
+            return all(t in value for t in target)
+        if op == FilterOperator.ANY:
+            return any(t in value for t in target)
+        raise NotImplementedError(
+            f"filter operator {op!r} not supported by TurboQuantVectorStore"
+        )
+
+    def query(self, query: VectorStoreQuery, **_: Any) -> VectorStoreQueryResult:
+        # MMR / SVM / LINEAR_REGRESSION / HYBRID etc. all need access to
+        # full-precision vectors (for pairwise diversity, learned scoring,
+        # or sparse-dense fusion). turbovec discards full precision after
+        # quantization, so any non-DEFAULT mode is unsupportable here.
+        # Raise loudly instead of silently treating it as DEFAULT, which
+        # the previous impl did and which let callers think they were
+        # getting e.g. MMR diversity when they were not.
+        if query.mode != VectorStoreQueryMode.DEFAULT:
+            raise NotImplementedError(
+                f"TurboQuantVectorStore does not support query mode "
+                f"{query.mode!r}. Only VectorStoreQueryMode.DEFAULT is "
+                "supported — MMR / SVM / hybrid modes need access to "
+                "full-precision vectors which turbovec discards after "
+                "quantization. Maintain a parallel store with full vectors "
+                "if you need a non-default scoring mode."
+            )
+        if query.query_embedding is None:
+            raise ValueError(
+                "TurboQuantVectorStore requires a pre-computed query_embedding "
+                "(is_embedding_query=True)."
+            )
+        qvec = np.asarray(query.query_embedding, dtype=np.float32)
+        if qvec.ndim == 1:
+            qvec = qvec[None, :]
+        if not qvec.flags["C_CONTIGUOUS"]:
+            qvec = np.ascontiguousarray(qvec)
+
+        if len(self._index) == 0:
+            return VectorStoreQueryResult(nodes=[], similarities=[], ids=[])
+
+        has_filters = (
+            query.filters is not None
+            or bool(query.node_ids)
+            or bool(query.doc_ids)
+        )
+        if not has_filters:
+            k = min(query.similarity_top_k, len(self._index))
+            scores, handles = self._index.search(qvec, k)
+        else:
+            allowed_handles = self._resolve_allowed_handles(
+                query.filters, query.node_ids, query.doc_ids
+            )
+            if not allowed_handles:
+                return VectorStoreQueryResult(nodes=[], similarities=[], ids=[])
+            allowlist = np.asarray(allowed_handles, dtype=np.uint64)
+            scores, handles = self._index.search(
+                qvec, query.similarity_top_k, allowlist=allowlist
+            )
+
+        result_nodes: list[TextNode] = []
+        similarities: list[float] = []
+        ids: list[str] = []
+        for score, handle in zip(scores[0], handles[0]):
+            nid = self._u64_to_node_id[int(handle)]
+            data = self._nodes[nid]
+            result_nodes.append(self._reconstruct_node(nid, data))
+            similarities.append(float(score))
+            ids.append(nid)
+
+        return VectorStoreQueryResult(nodes=result_nodes, similarities=similarities, ids=ids)
+
+    # ---- Async overrides --------------------------------------------------
+    #
+    # The base class provides default async impls that delegate to sync via
+    # `return self.<sync>(...)`. We override them explicitly so the signature
+    # is visible on the class and an autodoc tool / IDE doesn't make
+    # callers chase the abstract base class for the documentation.
+
+    async def async_add(
+        self, nodes: Sequence[BaseNode], **kwargs: Any
+    ) -> List[str]:
+        return self.add(list(nodes), **kwargs)
+
+    async def adelete(self, ref_doc_id: str, **kwargs: Any) -> None:
+        self.delete(ref_doc_id, **kwargs)
+
+    async def adelete_nodes(
+        self,
+        node_ids: Optional[List[str]] = None,
+        filters: Optional[MetadataFilters] = None,
+        **kwargs: Any,
+    ) -> None:
+        self.delete_nodes(node_ids=node_ids, filters=filters, **kwargs)
+
+    async def aclear(self) -> None:
+        self.clear()
+
+    async def aquery(
+        self, query: VectorStoreQuery, **kwargs: Any
+    ) -> VectorStoreQueryResult:
+        return self.query(query, **kwargs)
+
+    async def aget_nodes(
+        self,
+        node_ids: Optional[List[str]] = None,
+        filters: Optional[MetadataFilters] = None,
+    ) -> List[BaseNode]:
+        return self.get_nodes(node_ids=node_ids, filters=filters)
+
+    # ---- Config serialization ---------------------------------------------
+
+    def to_dict(self, **_: Any) -> dict[str, Any]:
+        """Serialize the store's *configuration* (not its data) so a
+        fresh instance can be reconstructed via ``from_dict``. Mirrors
+        the contract of ``SimpleVectorStore.to_dict`` — config-only;
+        node data round-trips through ``persist`` / ``from_persist_path``.
+        """
+        return {
+            "bit_width": self._index.bit_width,
+            "dim": self._index.dim,  # may be None (lazy uncommitted)
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any], **_: Any) -> "TurboQuantVectorStore":
+        """Construct an empty store from a config dict produced by
+        ``to_dict``. To restore data, use ``from_persist_path``."""
+        dim = data.get("dim")
+        bit_width = data.get("bit_width", 4)
+        return cls(index=IdMapIndex(dim, bit_width))
+
+    def persist(self, persist_path: str, fs: Any = None) -> None:
+        """Persist the store. ``persist_path`` is treated as a path *stem*:
+        the binary index goes to ``{stem}.tvim`` and the node side-car to
+        ``{stem}.nodes.json``. Any extension on ``persist_path`` (e.g.
+        ``.json`` from a StorageContext default) is replaced.
+
+        This matches the layout assumed by ``StorageContext.persist`` —
+        which calls us with ``persist_path = {persist_dir}/{namespace}__vector_store.json`` —
+        and lets multiple namespaced stores coexist in the same directory.
+
+        Node metadata must be JSON-serializable (same constraint as
+        ``SimpleVectorStore``). ``fs`` (fsspec) is not yet supported;
+        pass a local path.
+        """
+        if fs is not None:
+            raise NotImplementedError(
+                "fsspec filesystems are not supported yet; pass a local path."
+            )
+        base = _split_persist_base(persist_path)
+        base.parent.mkdir(parents=True, exist_ok=True)
+        self._index.write(str(base.with_suffix(_INDEX_EXT)))
+        payload = {
+            "schema_version": _NODES_SCHEMA_VERSION,
+            "nodes": self._nodes,
+            # JSON object keys must be strings; round-trip int keys via
+            # an explicit list of [node_id, handle] pairs to preserve
+            # type fidelity.
+            "node_id_to_u64": list(self._node_id_to_u64.items()),
+            "next_u64": self._next_u64,
+        }
+        with open(base.with_suffix(_STORE_EXT), "w") as f:
+            json.dump(payload, f)
+
+    @classmethod
+    def from_persist_path(
+        cls,
+        persist_path: str,
+        fs: Any = None,
+    ) -> "TurboQuantVectorStore":
+        """Load a previously-persisted store. ``persist_path`` is the same
+        path that was passed to :meth:`persist` (extension is ignored;
+        ``{stem}.tvim`` and ``{stem}.nodes.json`` are read).
+
+        Safe to call on any path — the side-car is plain JSON, never
+        pickle, so there's no deserialization-of-code risk.
+        """
+        if fs is not None:
+            raise NotImplementedError(
+                "fsspec filesystems are not supported yet; pass a local path."
+            )
+        base = _split_persist_base(persist_path)
+        index = IdMapIndex.load(str(base.with_suffix(_INDEX_EXT)))
+        with open(base.with_suffix(_STORE_EXT)) as f:
+            state = json.load(f)
+        version = state.get("schema_version", 0)
+        if version not in _NODES_SCHEMA_COMPAT:
+            raise ValueError(
+                f"{_STORE_EXT.lstrip('.')} has schema version {version}; "
+                f"this turbovec accepts versions {list(_NODES_SCHEMA_COMPAT)}"
+            )
+        store = cls(index=index)
+        # v1 entries lack `node_dict` and reconstruct as narrow TextNodes;
+        # v2 entries carry it and reconstruct with full BaseNode fidelity.
+        # `_reconstruct_node` dispatches on shape, so we just load the
+        # dict as-is.
+        store._nodes = state["nodes"]
+        # Reconstruct {node_id: int handle} from the list-of-pairs form.
+        store._node_id_to_u64 = {nid: int(h) for nid, h in state["node_id_to_u64"]}
+        store._u64_to_node_id = {h: nid for nid, h in store._node_id_to_u64.items()}
+        store._next_u64 = int(state["next_u64"])
+        check_persisted_handles(index, store._u64_to_node_id.keys(), what="node")
+        return store
+
+    @classmethod
+    def from_persist_dir(
+        cls,
+        persist_dir: str,
+        namespace: str = _DEFAULT_VECTOR_STORE,
+        fs: Any = None,
+    ) -> "TurboQuantVectorStore":
+        """Load a store from a ``StorageContext``-style persist directory.
+
+        Builds the namespaced filename
+        ``{persist_dir}/{namespace}__vector_store.json`` and forwards to
+        :meth:`from_persist_path`. The ``.json`` suffix is conventional —
+        our actual on-disk files use ``.tvim`` and ``.nodes.json``
+        extensions derived from the same stem.
+        """
+        persist_fname = f"{namespace}{_NAMESPACE_SEP}{_DEFAULT_PERSIST_FNAME}"
+        persist_path = os.path.join(persist_dir, persist_fname)
+        return cls.from_persist_path(persist_path, fs=fs)
+
+
+__all__ = ["TurboQuantVectorStore"]