Embedding Request Limits¶

BatchEmbeddings and AsyncBatchEmbeddings split cache batches into requests with at most 2,048 inputs, 8,192 tokens per input, and 300,000 total tokens. These defaults match the OpenAI Embeddings contract.

batch_size=None still enables adaptive batching. Positive sizes remain cache chunk sizes, and nonpositive sizes still select all unique inputs at the cache layer. In every mode the embedding layer may split that chunk further. These limits do not constrain generic BatchCache use. Splits retain input order, deduplication, and the existing concurrency bound; responses are ordered by their SDK index before caching.

Empty strings or a single input exceeding either token limit raise ValueError before sending any requests for the affected cache chunk. Text is never silently truncated. Earlier chunks in a long operation may already have run.

For a compatible provider with different limits, pass limits=EmbeddingLimits(...) to either core factory or constructor, pandas .ai / .aio embeddings() or embeddings_with_cache(), or Spark/DuckDB embeddings_udf(). All numeric limits must be positive. This option is not sent to the SDK as an API parameter.

from openaivec import BatchEmbeddings, EmbeddingLimits

limits = EmbeddingLimits(
    max_inputs=512,
    max_input_tokens=4096,
    max_request_tokens=100000,
    encoding_name="cl100k_base",
)
embedder = BatchEmbeddings.of(client, "my-embedding-deployment", limits=limits)

By default, token counting uses tiktoken's model encoding. Unknown deployment names use cl100k_base, shared by the supported OpenAI embedding models. Custom providers must specify the correct encoding and limits; these local checks do not replace provider quotas or rate limits.

openaivec.EmbeddingLimits `dataclass` ¶

EmbeddingLimits(
    max_inputs: int = 2048,
    max_input_tokens: int = 8192,
    max_request_tokens: int = 300000,
    encoding_name: str | None = None,
)

Hard request limits for the selected embedding provider.

Attributes:

Name	Type	Description
`max_inputs`	`int`	Maximum strings per request. Defaults to 2048.
`max_input_tokens`	`int`	Maximum tokens per string. Defaults to 8192.
`max_request_tokens`	`int`	Maximum aggregate tokens. Defaults to 300000.
`encoding_name`	`str \| None`	Explicit tiktoken encoding for a custom provider. None selects the model encoding, with cl100k_base for deployment aliases. All numeric limits must be positive.

openaivec.BatchEmbeddings `dataclass` ¶

BatchEmbeddings(
    client: OpenAI,
    model_name: str,
    cache: BatchCache[str, NDArray[float32]] = (
        lambda: BatchCache(
            batch_size=None,
            max_cache_size=DEFAULT_MANAGED_CACHE_SIZE,
        )
    )(),
    api_kwargs: dict[str, Any] = dict(),
    limits: EmbeddingLimits = EmbeddingLimits(),
    retry_policy: RetryPolicy | None = None,
)

Thin wrapper around the OpenAI embeddings endpoint (synchronous).

By default, requests are limited to 2,048 inputs and 300,000 total tokens, even with automatic or nonpositive batch sizes. Empty inputs and inputs over 8,192 tokens are rejected before sending the affected cache batch. Text is never truncated. Deployment aliases use the cl100k_base tokenizer shared by the supported OpenAI embedding models. Supply limits to override these provider defaults.

Attributes:

Name	Type	Description
`client`	`OpenAI`	Configured OpenAI client.
`model_name`	`str`	For Azure OpenAI, use your deployment name. For OpenAI, use the model name (e.g., `"text-embedding-3-small"`).
`cache`	`BatchCache[str, NDArray[float32]]`	Batching proxy for ordered, cached mapping. Library-managed instances use bounded retention by default.
`api_kwargs`	`dict[str, Any]`	Additional OpenAI API parameters stored at initialization.
`limits`	`EmbeddingLimits`	Provider request limits, independent of cache batch size.
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.

Methods:¶

of `classmethod` ¶

of(
    client: OpenAI,
    model_name: str,
    batch_size: int | None = None,
    *,
    limits: EmbeddingLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> BatchEmbeddings

Factory constructor.

Parameters:

Name	Type	Description	Default
`client`	`OpenAI`	OpenAI client.	required
`model_name`	`str`	For Azure OpenAI, use your deployment name. For OpenAI, use the model name.	required
`batch_size`	`int \| None`	Max unique inputs per API call. Defaults to None (automatic batch size optimization). Set to a positive integer for fixed batch size.	`None`
`limits`	`EmbeddingLimits \| None`	Provider-specific hard limits. None uses OpenAI defaults.	`None`
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.	`None`
`**api_kwargs`		Additional OpenAI API parameters (e.g., dimensions for text-embedding-3 models).	`{}`

Returns:

Name	Type	Description
`BatchEmbeddings`	`BatchEmbeddings`	Configured instance backed by a batching proxy.

Source code in src/openaivec/_embeddings.py

@classmethod
def of(
    cls,
    client: OpenAI,
    model_name: str,
    batch_size: int | None = None,
    *,
    limits: EmbeddingLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> "BatchEmbeddings":
    """Factory constructor.

    Args:
        client (OpenAI): OpenAI client.
        model_name (str): For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
        batch_size (int | None, optional): Max unique inputs per API call. Defaults to None
            (automatic batch size optimization). Set to a positive integer for fixed batch size.
        limits (EmbeddingLimits | None): Provider-specific hard limits. None uses OpenAI defaults.
        retry_policy (RetryPolicy | None): Transport limits. ``None`` preserves SDK retries.
        **api_kwargs: Additional OpenAI API parameters (e.g., dimensions for text-embedding-3 models).

    Returns:
        BatchEmbeddings: Configured instance backed by a batching proxy.
    """
    return cls(
        client=client,
        model_name=model_name,
        cache=BatchCache(batch_size=batch_size, max_cache_size=DEFAULT_MANAGED_CACHE_SIZE),
        api_kwargs=api_kwargs,
        limits=limits if limits is not None else EmbeddingLimits(),
        retry_policy=retry_policy,
    )

create ¶

create(inputs: list[str]) -> list[NDArray[np.float32]]

Generate embeddings for inputs using cached, ordered batching.

Parameters:

Name	Type	Description	Default
`inputs`	`list[str]`	Input strings. Duplicates allowed.	required

Returns:

Type	Description
`list[NDArray[float32]]`	list[NDArray[np.float32]]: Embedding vectors aligned to `inputs`.

Source code in src/openaivec/_embeddings.py

@observe(_LOGGER)
def create(self, inputs: list[str]) -> list[NDArray[np.float32]]:
    """Generate embeddings for inputs using cached, ordered batching.

    Args:
        inputs (list[str]): Input strings. Duplicates allowed.

    Returns:
        list[NDArray[np.float32]]: Embedding vectors aligned to ``inputs``.
    """
    return self.cache.map(inputs, self._embed_chunk)

openaivec.AsyncBatchEmbeddings `dataclass` ¶

AsyncBatchEmbeddings(
    client: AsyncOpenAI,
    model_name: str,
    cache: AsyncBatchCache[str, NDArray[float32]] = (
        lambda: AsyncBatchCache(
            batch_size=None,
            max_concurrency=8,
            max_cache_size=DEFAULT_MANAGED_CACHE_SIZE,
        )
    )(),
    api_kwargs: dict[str, Any] = dict(),
    limits: EmbeddingLimits = EmbeddingLimits(),
    retry_policy: RetryPolicy | None = None,
)

Thin wrapper around the OpenAI embeddings endpoint (asynchronous).

This class provides an asynchronous interface for generating embeddings using OpenAI models. It manages concurrency, handles rate limits automatically, and efficiently processes batches of inputs, including de-duplication.

Request limits and input validation match BatchEmbeddings. Splitting a cache batch does not increase concurrency; subrequests run sequentially within the existing cache worker.

Example

import asyncio
import numpy as np
from openai import AsyncOpenAI
from openaivec import AsyncBatchEmbeddings

# Assuming openai_async_client is an initialized AsyncOpenAI client
openai_async_client = AsyncOpenAI() # Replace with your actual client initialization

embedder = AsyncBatchEmbeddings.of(
    client=openai_async_client,
    model_name="text-embedding-3-small",
    batch_size=128,
    max_concurrency=8,
)
texts = ["This is the first document.", "This is the second document.", "This is the first document."]

# Asynchronous call
async def main():
    embeddings = await embedder.create(texts)
    # embeddings will be a list of numpy arrays (float32)
    # The embedding for the third text will be identical to the first
    # due to automatic de-duplication.
    print(f"Generated {len(embeddings)} embeddings.")
    print(f"Shape of first embedding: {embeddings[0].shape}")
    assert np.array_equal(embeddings[0], embeddings[2])

# Run the async function
asyncio.run(main())

Attributes:

Name	Type	Description
`client`	`AsyncOpenAI`	Configured OpenAI async client.
`model_name`	`str`	For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
`cache`	`AsyncBatchCache[str, NDArray[float32]]`	Async batching proxy. Library-managed instances use bounded retention by default.
`api_kwargs`	`dict`	Additional OpenAI API parameters stored at initialization.
`limits`	`EmbeddingLimits`	Provider request limits, independent of cache batch size.
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.

Methods:¶

of `classmethod` ¶

of(
    client: AsyncOpenAI,
    model_name: str,
    batch_size: int | None = None,
    max_concurrency: int = 8,
    *,
    limits: EmbeddingLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> AsyncBatchEmbeddings

Factory constructor.

Parameters:

Name	Type	Description	Default
`client`	`AsyncOpenAI`	OpenAI async client.	required
`model_name`	`str`	For Azure OpenAI, use your deployment name. For OpenAI, use the model name.	required
`batch_size`	`int \| None`	Max unique inputs per API call. Defaults to None (automatic batch size optimization). Set to a positive integer for fixed batch size.	`None`
`max_concurrency`	`int`	Max concurrent API calls. Defaults to 8.	`8`
`limits`	`EmbeddingLimits \| None`	Provider-specific hard limits. None uses OpenAI defaults.	`None`
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.	`None`
`**api_kwargs`		Additional OpenAI API parameters (e.g., dimensions for text-embedding-3 models).	`{}`

Returns:

Name	Type	Description
`AsyncBatchEmbeddings`	`AsyncBatchEmbeddings`	Configured instance with an async batching proxy.

Source code in src/openaivec/_embeddings.py

@classmethod
def of(
    cls,
    client: AsyncOpenAI,
    model_name: str,
    batch_size: int | None = None,
    max_concurrency: int = 8,
    *,
    limits: EmbeddingLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> "AsyncBatchEmbeddings":
    """Factory constructor.

    Args:
        client (AsyncOpenAI): OpenAI async client.
        model_name (str): For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
        batch_size (int | None, optional): Max unique inputs per API call. Defaults to None
            (automatic batch size optimization). Set to a positive integer for fixed batch size.
        max_concurrency (int, optional): Max concurrent API calls. Defaults to 8.
        limits (EmbeddingLimits | None): Provider-specific hard limits. None uses OpenAI defaults.
        retry_policy (RetryPolicy | None): Transport limits. ``None`` preserves SDK retries.
        **api_kwargs: Additional OpenAI API parameters (e.g., dimensions for text-embedding-3 models).

    Returns:
        AsyncBatchEmbeddings: Configured instance with an async batching proxy.
    """
    return cls(
        client=client,
        model_name=model_name,
        cache=AsyncBatchCache(
            batch_size=batch_size,
            max_concurrency=max_concurrency,
            max_cache_size=DEFAULT_MANAGED_CACHE_SIZE,
        ),
        api_kwargs=api_kwargs,
        limits=limits if limits is not None else EmbeddingLimits(),
        retry_policy=retry_policy,
    )

create `async` ¶

create(inputs: list[str]) -> list[NDArray[np.float32]]

Generate embeddings for inputs using proxy batching (async).

Parameters:

Name	Type	Description	Default
`inputs`	`list[str]`	Input strings. Duplicates allowed.	required

Returns:

Type	Description
`list[NDArray[float32]]`	list[NDArray[np.float32]]: Embedding vectors aligned to `inputs`.

Source code in src/openaivec/_embeddings.py

@observe(_LOGGER)
async def create(self, inputs: list[str]) -> list[NDArray[np.float32]]:
    """Generate embeddings for inputs using proxy batching (async).

    Args:
        inputs (list[str]): Input strings. Duplicates allowed.

    Returns:
        list[NDArray[np.float32]]: Embedding vectors aligned to ``inputs``.
    """
    return await self.cache.map(inputs, self._embed_chunk)  # type: ignore[arg-type]

Embedding Request Limits¶

openaivec.EmbeddingLimits dataclass ¶

openaivec.BatchEmbeddings dataclass ¶

Methods:¶

of classmethod ¶

create ¶

openaivec.AsyncBatchEmbeddings dataclass ¶

Methods:¶

of classmethod ¶

create async ¶

openaivec.EmbeddingLimits `dataclass` ¶

openaivec.BatchEmbeddings `dataclass` ¶

of `classmethod` ¶

openaivec.AsyncBatchEmbeddings `dataclass` ¶

of `classmethod` ¶

create `async` ¶