ZerdaZerda

OpenAIEmbedder

On this page

embeddings.openai_embedder.OpenAIEmbedder(BaseEmbedder) — OpenAI Embeddings API client.

Purpose

Wraps the openai Python SDK's client.embeddings.create() with request-level rate limiting, tiktoken-based token counting, cost tracking, retry-with-backoff, response-level caching, and optional embedding-dimension reduction (for text-embedding-3-* models).

Constructor

Python · 19 lines
OpenAIEmbedder(
    model_name: str = "text-embedding-3-small",
    api_key: Optional[str] = None,
    dimensions: Optional[int] = None,
    device: Optional[str] = None,
    normalize_embeddings: bool = True,
    batch_size: int = 100,
    max_length: Optional[int] = None,
    cache_embeddings: bool = True,
    show_progress: bool = False,
    timeout: int = 60,
    max_retries: int = 3,
    retry_delay: float = 1.0,
    enable_rate_limiting: bool = True,
    max_requests_per_minute: int = 3000,
    max_tokens_per_minute: int = 1_000_000,
    track_costs: bool = True,
    **kwargs,
)

Parameters

NameTypeDefaultNotes
model_namestr"text-embedding-3-small"Must be a key in MODEL_SPECS for accurate cost/dimension defaults; unknown names fall back to a generic 1536-dim spec with a warning
api_keyOptional[str]NoneFalls back to os.getenv('OPENAI_API_KEY'); raises ValueError if still unset
dimensionsOptional[int]NoneClamped to MODEL_SPECS[model]['max_dimensions'] if too high; raises ValueError if < 1
batch_sizeint100OpenAI's API supports up to 2048 per the module docstring, but no hard check enforces that ceiling here
cache_embeddingsboolTruePlain dict cache keyed by raw text
timeoutint60Passed to the OpenAI(timeout=...) client
max_retriesint3This class's own retry loop; the underlying OpenAI client is constructed with max_retries=0 so the SDK itself never retries independently
retry_delayfloat1.0Base delay for exponential backoff (retry_delay * 2**attempt)
enable_rate_limitingboolTrueEnables the module-level RateLimiter
max_requests_per_minuteint3000
max_tokens_per_minuteint1_000_000
track_costsboolTrueEnables self.usage_stats = UsageStats()

MODEL_SPECS (class attribute)

ModelMax/default dimensionsMax tokensCost per 1M tokensArabic support
text-embedding-3-large30728191$0.13excellent
text-embedding-3-small15368191$0.02excellent
text-embedding-ada-00215368191$0.10good

Raises

  • ValueError if no API key is resolvable.
  • ValueError if dimensions < 1.
  • ImportError if the openai package isn't installed.
  • ValueError from _validate_api_key() if the test request raises the SDK's APIError.

encode()

Python · 7 lines
encode(
    texts: Union[str, List[str]],
    show_progress_bar: bool = False,
    convert_to_numpy: bool = True,
    batch_size: Optional[int] = None,
    **kwargs,
) -> np.ndarray

Behavior

  1. Wraps a single string in a list; raises ValueError if the (post-filter) list is empty.
  2. Filters out empty/whitespace-only entries, logging a warning about how many were dropped.
  3. Splits each batch into cache hits and misses; only misses are sent to _process_batch().
  4. _process_batch() counts tokens via tiktoken (falls back to len(text)//4 if the tokenizer failed to load), applies rate limiting, and calls client.embeddings.create() with dimensions=self.dimensions for text-embedding-3-* models. Retries on RateLimitError / APIConnectionError with exponential backoff; does not retry on generic APIError (re-raises immediately).
  5. Merges cached and freshly-fetched embeddings back into original batch order.
  6. Stacks all batches, then L2-normalizes if normalize_embeddings=True (guarding against division by zero).
  7. Updates self._stats and logs a summary including cache hit rate, tokens used, and cost.

Returns

np.ndarray of shape (n, dimensions) — always 2D, even for single-string input (no unwrap step exists in this class, unlike ArabicEmbedder).

Raises

  • ValueError if all input texts are empty.
  • Propagates RuntimeError if all retries in _process_batch are exhausted, or re-raises the original SDK exception for non-retryable API errors.

Cost and usage tracking

estimate_cost(texts) -> Dict[str, Any]

Counts tokens via _count_tokens() (accurate tiktoken count, or len(text)//4 fallback) and multiplies by MODEL_SPECS[model]['cost_per_1m_tokens']. Does not call the API — purely a local estimate.

get_usage_stats() -> Dict[str, Any]

Merges self._stats with cache_hit_rate and (if track_costs) self.usage_stats.get_summary() (UsageStats: total tokens, requests, cost, elapsed time, requests-per-minute bucket count).

clear_cache(), get_model_info()

Standard cache-clear and model-info introspection; get_model_info() includes the full model_spec dict plus configured dimensions and tracking flags.

RateLimiter (module-level class)

Sliding-window limiter tracking request timestamps and per-token timestamps over the last 60 seconds, protected by a threading.RLock. wait_if_needed(estimated_tokens) sleeps as needed before allowing a new request through. Note: it appends one timestamp per estimated token to self.tokens, which is a memory/CPU cost proportional to token volume for high-throughput use.

UsageStats (dataclass)

Fields: total_tokens, total_requests, total_cost, start_time, requests_by_minute (a dict keyed by "YYYY-MM-DD HH:MM"). .get_summary() returns aggregated totals plus requests_per_minute (actually the count of distinct minute buckets seen, not a live rate).

Context manager

__enter__ returns self; __exit__ logs a session summary of tokens/requests/cost/elapsed time if track_costs=True.

Examples

Python · 13 lines
import os
from zerda.embeddings import OpenAIEmbedder

os.environ["OPENAI_API_KEY"] = "sk-..."

embedder = OpenAIEmbedder(model_name="text-embedding-3-large", dimensions=1024)
vectors = embedder.encode(["first document", "second document"])

cost_estimate = embedder.estimate_cost(["a fairly long piece of text to embed"])

with OpenAIEmbedder() as e:
    v = e.encode("hello")
# session summary logged on exit

Related pages