ZerdaZerda

cache.embedding_index

On this page

Overview

Two responsibilities, per the module docstring:

  1. EmbeddingProvider adapters — pluggable text-to-vector backends (OpenAI, HuggingFace, Ollama, Mock), each wrapped by an in-process LRU cache to avoid re-embedding identical text.
  2. EmbeddingIndex — a vector similarity search structure backed by FAISS when available, falling back to a pure-NumPy brute-force flat search otherwise.

Cosine similarity is used throughout; vectors are L2-normalized before indexing so that inner-product search is equivalent to cosine search.

_EmbeddingLRU (private)

A thread-safe LRU cache (OrderedDict-backed) keyed on sha256(text), used internally by CachedEmbedder to avoid re-embedding identical normalized queries. get() returns a defensive .copy() of the cached vector; put() evicts the least-recently-used entry (popitem(last=False)) once maxsize is reached.

BaseEmbeddingProvider

Python · 7 lines
class BaseEmbeddingProvider(ABC):
    @abstractmethod
    def encode(self, texts: List[str]) -> np.ndarray: ...
    async def aencode(self, texts: List[str]) -> np.ndarray: ...
    @property
    @abstractmethod
    def dimension(self) -> int: ...

The abstract contract every provider implements. aencode() has a concrete default implementation on the base class — it runs the subclass's synchronous encode() in the default thread-pool executor via loop.run_in_executor(None, self.encode, texts). No provider in this module overrides aencode(), so all four providers get async behavior "for free" via this thread-pool wrapper.

MockEmbeddingProvider

Python · 2 lines
class MockEmbeddingProvider(BaseEmbeddingProvider):
    def __init__(self, dim: int = 128) -> None

The default provider (EmbeddingProvider.MOCK), requiring no network access or API key — this is why the Quickstart examples run without configuration.

encode(self, texts: List[str]) -> np.ndarray

For each text: seeds a numpy.random.default_rng from sha256(text) % 2**31, draws a dim-length standard-normal vector, L2-normalizes it, and stacks all results into an (N, dim) array.

Determinism: identical input text always produces an identical vector (the seed is a pure function of the text), which is what makes the mock provider useful for reproducible tests — semantically similar text does not get similar vectors here (there is no actual semantic model), so real semantic-match behavior cannot be validated with this provider.

OpenAIEmbeddingProvider

Python · 2 lines
class OpenAIEmbeddingProvider(BaseEmbeddingProvider):
    def __init__(self, config: EmbeddingConfig) -> None

Attempts from zerda.embeddings import OpenAIEmbedder. If that import fails, raises ImportError("Install zerda[embeddings]: pip install zerda[embeddings]").

[!DANGER] zerda.embeddings is an external dependency not present in this codebase This module is referenced by name only — it does not ship with, and is not defined anywhere in, the analyzed source tree. Whether a package providing it actually exists, and under what name, cannot be determined from this repository. Until such a package is installed and importable, selecting EmbeddingProvider.OPENAI will always raise ImportError at CachedEmbedder construction time. See Known Issues.

encode() delegates directly to self._client.encode(texts) and returns its result as-is (no L2 normalization is applied at this layer, unlike the other providers — an inconsistency worth noting if you do get a working zerda.embeddings package installed). dimension reads self._client.embedding_dim. A static _l2_normalize helper is defined on the class but is not actually called from encode().

HuggingFaceEmbeddingProvider

Python · 2 lines
class HuggingFaceEmbeddingProvider(BaseEmbeddingProvider):
    def __init__(self, config: EmbeddingConfig) -> None

Same external-dependency situation: from zerda.embeddings import HuggingFaceEmbedder. encode() calls self._model.encode(texts, normalize_embeddings=True, show_progress_bar=False) and casts to float32. dimension reads self._model.embedding_dim, cached at construction.

OllamaEmbeddingProvider

Python · 2 lines
class OllamaEmbeddingProvider(BaseEmbeddingProvider):
    def __init__(self, config: EmbeddingConfig) -> None

[!DANGER] This provider is not functional in the analyzed source __init__ imports from zerda.embeddings import OllamaEmbedder and constructs self._model = OllamaEmbedder(), then reads self._model.embedding_dim. It never sets self._httpx, self._url, or self._timeout. However, encode() references self._httpx.post(self._url, json={"model": self._model, ...}, timeout=self._timeout) — all three of these attributes are undefined, and self._model at that point is actually the OllamaEmbedder object (not a model-name string, contrary to what the json payload implies). Calling encode() on this provider will raise AttributeError. See Known Issues.

build_embedding_provider(config: EmbeddingConfig) -> BaseEmbeddingProvider

A match/case factory analogous to build_storage(), dispatching on config.provider to construct one of the four providers above. Raises ValueError for any other value (unreachable in practice, since EmbeddingProvider only has four members).

EmbeddingIndex

Python · 2 lines
class EmbeddingIndex:
    def __init__(self, dimension: int, max_size: int = 100_000, use_faiss: bool = True) -> None

Overview

A thread-safe approximate/exact nearest-neighbor index storing (entry_id, embedding) pairs, returning the top-k most similar entry IDs with cosine similarities on query. Backs CacheManager's semantic-search stage.

Constructor

If use_faiss=True, attempts import faiss and builds a faiss.IndexFlatIP(dimension) (inner-product index — mathematically equivalent to cosine similarity given L2-normalized inputs). If FAISS isn't installed, logs an info message and falls back to a NumPy-array implementation for the lifetime of the instance.

add(self, entry_id: str, vector: np.ndarray) -> None

Overview: inserts or replaces a vector for entry_id.

Internal Workflow:

  1. L2-normalize vector (_ensure_unit_norm).
  2. If entry_id already exists, remove it first (_remove_locked) to avoid duplicates.
  3. Add to the FAISS index or append to the NumPy _vectors matrix (np.vstack).
  4. Append entry_id to _id_map.
  5. If capacity (max_size) is now exceeded, trim the oldest entries via _trim_locked.

Complexity: O(d) for normalization (d = dimension); FAISS add is effectively O(d); the NumPy path's np.vstack is O(n·d) since it copies the entire existing matrix — this makes repeated add() calls on a large NumPy-backed index O(n) each, i.e. O(n²) total for n sequential inserts. See Known Issues → Performance notes for the practical implication.

Thread Safety: thread-safe (RLock).

remove(self, entry_id: str) -> None

Removes a vector by ID. For the NumPy backend this is O(n·d) (np.delete). For FAISS, since IndexFlatIP doesn't support in-place deletion, this triggers a full index rebuild (_rebuild_faiss_locked) from the remaining vectors — O(n·d) as well, but with the added constant-factor cost of reconstructing the FAISS index object itself.

search(self, vector: np.ndarray, top_k: int = 5, min_similarity: float = 0.0) -> List[Tuple[str, float]]

Overview: the primary read operation — find the top_k most similar stored vectors to vector, filtered to those with cosine similarity >= min_similarity.

Algorithm (NumPy fallback, _numpy_search): computes all similarities via a single matrix-vector product (self._vectors @ query), uses np.argpartition for an O(n) partial sort to find the top-k indices, then sorts just those k values descending.

Algorithm (FAISS path): delegates to faiss.IndexFlatIP.search, which performs an exhaustive inner-product scan (flat index — no approximation, despite FAISS's broader capability for approximate search structures) and returns distances/indices directly.

Returns: an empty list immediately if the index is empty (not self._id_map), without allocating a query vector at all.

Complexity:

  • NumPy fallback: O(n·d) for the similarity computation, O(n) for argpartition.
  • FAISS IndexFlatIP: also O(n·d) (it's an exact, brute-force index — the "Flat" in the name means no clustering/quantization), but with FAISS's optimized (typically vectorized/BLAS-backed) implementation, which is usually faster in wall-clock terms than the NumPy path for the same n.

__len__(self) -> int

Returns len(self._id_map).

clear(self) -> None

Empties _id_map and _vectors; if using FAISS, rebuilds an empty IndexFlatIP.

Private helpers

MethodPurpose
_numpy_search(query, k, min_sim)Brute-force cosine search over self._vectors
_remove_locked(entry_id)Removes one entry from _id_map and the vector store (FAISS: triggers rebuild)
_trim_locked(n)Removes the n oldest entries (position 0 onward — a FIFO-style overflow policy)
_rebuild_faiss_locked()Reconstructs the FAISS index from _vectors/_id_map
_ensure_unit_norm(vec) (static)L2-normalizes a vector; returns it unchanged if its norm is 0

Thread Safety

All public mutating/reading methods acquire self._lock (RLock).

Examples

Python · 13 lines
import numpy as np
from zerda.cache.embedding_index import EmbeddingIndex

idx = EmbeddingIndex(dimension=128, max_size=10_000)

idx.add("doc-1", np.random.rand(128).astype(np.float32))
idx.add("doc-2", np.random.rand(128).astype(np.float32))

results = idx.search(np.random.rand(128).astype(np.float32), top_k=1)
print(results)   # [(entry_id, similarity)]

idx.remove("doc-1")
print(len(idx))   # 1

CachedEmbedder

Python · 2 lines
class CachedEmbedder:
    def __init__(self, config: EmbeddingConfig) -> None

Overview

Wraps a BaseEmbeddingProvider with an in-process LRU (via _EmbeddingLRU, enabled when config.cache_embeddings is True), async batch processing, and an asyncio.Lock used to serialize concurrent calls to the same instance. This is the class CacheManager actually holds (self._embedder) — it never talks to a raw BaseEmbeddingProvider directly.

dimension (property)

Delegates to the wrapped provider's dimension.

async embed_single(self, text: str) -> np.ndarray

Convenience wrapper: calls embed_batch([text]) and returns element 0.

async embed_batch(self, texts: List[str]) -> np.ndarray

Overview

The main entry point — embeds a list of texts, using the LRU cache for any text seen before and batching the rest through the provider.

Internal Workflow

  1. Return an empty (0, dimension) array immediately if texts is empty.
  2. For each text, check the LRU (if enabled); cache hits go straight into a cached dict keyed by original index; misses are collected into todo_idx/todo_texts.
  3. For the texts that need embedding, iterate in chunks of config.batch_size (default 64), calling provider.aembed(batch) for each chunk.

[!DANGER] CachedEmbedder calls provider.aembed, but BaseEmbeddingProvider defines aencode Every embedding provider in this module implements encode()/inherits aencode() from BaseEmbeddingProvider — there is no aembed method defined anywhere on BaseEmbeddingProvider or any of its four subclasses in the analyzed source. CachedEmbedder.embed_batch() calling await self._provider.aembed(batch) will raise AttributeError for every provider, including the default MockEmbeddingProvider. This means CacheManager.get()/put() — which both call embed_single/embed_batch internally — cannot currently complete successfully with any embedding provider, including the mock one, as shipped in this source tree. See Known Issues for the full analysis and a suggested workaround.

  1. On a per-chunk exception, logs the error and substitutes a zero-vector batch so the pipeline doesn't crash outright (this exception handling is intact regardless of the aembed/aencode naming issue above).
  2. Populates the LRU (if enabled) and the cached dict for every newly-embedded text.
  3. Returns np.stack([cached[i] for i in range(len(texts))], axis=0) — reassembled in the original input order.

Complexity

O(n) in the number of texts for LRU lookups; embedding cost itself is delegated to the provider and is typically the dominant cost (network round-trip for real providers).

Thread Safety / Async Safety

The asyncio.Lock (self._lock) is declared on the instance but is not actually acquired anywhere inside embed_single/embed_batch in the analyzed source — see Known Issues for the implication (concurrent embed_batch calls on the same CachedEmbedder are not serialized by this lock as the field's name might suggest; CacheManager separately protects identical in-flight queries via its own _InflightCoalescer, which is a different mechanism).

Examples

Python · 10 lines
import asyncio
from zerda.cache import CachedEmbedder
from zerda.cache import EmbeddingConfig

async def main():
    embedder = CachedEmbedder(EmbeddingConfig())   # MOCK provider by default
    vec = await embedder.embed_single("hello world")
    print(vec.shape)

asyncio.run(main())

[!WARNING] The example above will raise AttributeError as shipped Per the note under embed_batch() above, this call currently fails inside the library itself ('MockEmbeddingProvider' object has no attribute 'aembed') regardless of which provider is configured. It is shown here because it represents the intended public API surface, not because it currently runs successfully against this exact source tree.

Related pages