ZerdaZerda

BaseEmbedder

On this page

embeddings.base_embedder.BaseEmbedder — abstract base class (abc.ABC) for all embedders in the package.

Purpose

Defines the interface every embedder must satisfy (encode(), embedding_dim) and provides shared, reusable functionality: device auto-detection, generic caching, similarity math, performance statistics, connection validation, and both sync and async context-manager support.

Constructor

Python · 10 lines
BaseEmbedder(
    model_name: str,
    device: Optional[str] = None,
    normalize_embeddings: bool = True,   # from EmbedderConfig().normalize_embeddings
    batch_size: int = 32,                # from EmbedderConfig().batch_size
    max_length: Optional[int] = 512,     # from EmbedderConfig().max_seq_length
    cache_embeddings: bool = False,      # from EmbedderConfig().enable_cache
    show_progress: bool = False,         # from EmbedderConfig().show_progress_bar
    **kwargs,
)

Since BaseEmbedder is abstract (defines abstract members encode and embedding_dim), it cannot be instantiated directly — attempting BaseEmbedder(...) raises TypeError.

Parameters

NameTypeDefaultNotes
model_namestrrequiredStored as self.model_name; no validation performed here
deviceOptional[str]NoneIf None, resolved via self._get_best_device()
normalize_embeddingsboolTrueStored; subclasses decide whether/how to use it
batch_sizeint32Stored; not validated (0 or negative would only fail downstream)
max_lengthOptional[int]512Stored as self.max_length
cache_embeddingsboolFalseIf True, initializes self._cache = {}
show_progressboolFalseStored as self.show_progress
**kwargsany—Stored as self.kwargs; not otherwise used by this class

Side effects

  • Initializes self._stats as a dict with keys total_encodings, total_texts, cache_hits, total_time, avg_time_per_text.
  • If cache_embeddings=True, creates self._cache = {} and logs "✅ Caching enabled".
  • Logs four INFO-level messages describing the class name, model, device, and batch size.

Abstract members

encode(texts, show_progress_bar=False, convert_to_numpy=True, **kwargs) -> np.ndarray

Must be implemented by every subclass. No default behavior exists at this level.

embedding_dim (property)

Must be implemented by every subclass; expected to return the output vector dimensionality as an int.

Key methods

encode_with_cache(texts, **kwargs) -> np.ndarray

Generic caching wrapper. If cache_embeddings=False, this is a passthrough to encode(). Otherwise, keys are MD5 hashes of the input text (_get_cache_key); for a list input, cached and non-cached texts are split, only non-cached texts are sent to encode(), and results are re-merged into original order.

  • Raises: nothing directly beyond what encode() itself may raise.
  • Note: as documented in FAQ → Known Issues, no shipped subclass currently routes its own encode() calls through this method — each implements separate, backend-specific caching instead.

similarity(text1, text2, metric='cosine') -> float

Accepts either raw text (auto-encoded via self.encode) or precomputed np.ndarray embeddings for either argument.

  • metric='cosine' and metric='dot' are implemented identically here — both compute np.dot(emb1, emb2) with no division by norms, meaning 'cosine' at this base-class level is actually a raw dot product, not true cosine similarity, unless the vectors are already unit-normalized (e.g. via normalize_embeddings=True upstream).
  • metric='euclidean' computes 1.0 / (1.0 + np.linalg.norm(emb1 - emb2)).
  • Raises ValueError for any other metric string.

batch_similarity(query, texts, top_k=None) -> Union[np.ndarray, List[tuple]]

Computes np.dot(text_embs, query_emb) between one query and many candidate texts (or precomputed embedding arrays).

  • Returns a plain np.ndarray of scores if top_k is None.
  • Returns a List[Tuple[int, float]] of (index, score) sorted descending if top_k is given.
  • Updates self._stats['total_time'] (or the dataclass equivalent) with elapsed wall time.

validate_connection(test_text="مرحباً Hello", detailed=False) -> Dict[str, Any]

Calls self.encode(test_text) and reports success/failure as a dict with keys success, reason, embedding_dim, and (if detailed=True) model_name, device, test_text_length, embedding_shape, embedding_norm, embedding_mean, embedding_std. Catches any exception from encode() and returns {"success": False, "reason": ..., "error_type": type(e).__name__} rather than propagating it — this method does not raise under normal operation.

get_model_info(), get_stats(), reset_stats(), clear_cache()

Return/reset introspection data. get_stats() branches on whether self._stats is a dict or an object (to accommodate HuggingFaceEmbedder's EmbeddingStats dataclass) — see Known Issues item 5.

timing(operation="encoding") (context manager)

Python · 2 lines
with embedder.timing("batch_encoding"):
    embeddings = embedder.encode(texts)

Logs f"⏱️ {operation}: {elapsed:.3f}s" at INFO level on exit. Does not suppress exceptions raised inside the with block.

cleanup()

Clears the cache (if enabled) and, if torch is importable and CUDA is available, calls torch.cuda.empty_cache(). Swallows ImportError silently. Also invoked automatically from __del__ (wrapped in a bare except: pass, so cleanup errors during garbage collection are always suppressed).

Async API

MethodBehavior
async aencode(texts, show_progress_bar=False, convert_to_numpy=True, **kwargs)await asyncio.to_thread(self.encode, texts, show_progress_bar, convert_to_numpy, **kwargs)
async aencode_with_cache(texts, **kwargs)await asyncio.to_thread(self.encode_with_cache, texts, **kwargs)
async abatch_similarity(query, candidates, **kwargs)await asyncio.to_thread(self.batch_similarity, query, candidates, **kwargs)
async __aenter__ / async __aexit__Enter returns self; exit calls self.cleanup() and returns False (exceptions are not suppressed)

These are inherited unmodified by every subclass — none override the async methods.

Thread safety

Not explicitly guaranteed. See FAQ → Is this thread-safe? for a full breakdown per subclass.

Related pages