Main Package API¶

The main openaivec package provides the core classes for AI-powered data processing.

Configuration¶

Client and model configuration helpers. These are the canonical entry point for setting up OpenAI / Azure OpenAI credentials, Fabric built-in models, and model names. Configuration is shared within the current Python process; Spark executors require their own setup. See the authentication guide.

openaivec.setup_fabric ¶

setup_fabric(
    *,
    responses_model: str = "gpt-5.1",
    embeddings_model: str = "text-embedding-ada-002",
    api_version: str = "2025-04-01-preview",
) -> None

Use Fabric's built-in models with runtime-managed authentication.

Explicitly replaces the default sync and async clients and model names for the current Python process. No Azure OpenAI resource, API key, or service principal secret is required. HTTP clients are created lazily. The async API requires the corresponding helper in the Fabric runtime.

Parameters:

Name	Type	Description	Default
`responses_model`	`str`	Built-in response model. Defaults to `gpt-5.1`.	`'gpt-5.1'`
`embeddings_model`	`str`	Built-in embedding model. Defaults to `text-embedding-ada-002`.	`'text-embedding-ada-002'`
`api_version`	`str`	Fabric-supported Azure OpenAI API version. Defaults to `2025-04-01-preview`.	`'2025-04-01-preview'`

Raises:

Type	Description
`RuntimeError`	Not running in a supported Fabric notebook runtime.
`ValueError`	A model name or API version is empty.

Example

import openaivec openaivec.setup_fabric() # doctest: +SKIP

Notes

Built-in models are a preview feature billed to Fabric capacity. This configures the notebook driver, not Spark executor authentication. Close resolved clients before replacing them with another setup.

Source code in src/openaivec/_provider.py

def setup_fabric(
    *,
    responses_model: str = "gpt-5.1",
    embeddings_model: str = "text-embedding-ada-002",
    api_version: str = "2025-04-01-preview",
) -> None:
    """Use Fabric's built-in models with runtime-managed authentication.

    Explicitly replaces the default sync and async clients and model names for
    the current Python process. No Azure OpenAI resource, API key, or service
    principal secret is required. HTTP clients are created lazily. The async
    API requires the corresponding helper in the Fabric runtime.

    Args:
        responses_model (str): Built-in response model. Defaults to ``gpt-5.1``.
        embeddings_model (str): Built-in embedding model. Defaults to
            ``text-embedding-ada-002``.
        api_version (str): Fabric-supported Azure OpenAI API version. Defaults
            to ``2025-04-01-preview``.

    Raises:
        RuntimeError: Not running in a supported Fabric notebook runtime.
        ValueError: A model name or API version is empty.

    Example:
        >>> import openaivec
        >>> openaivec.setup_fabric()  # doctest: +SKIP

    Notes:
        Built-in models are a preview feature billed to Fabric capacity. This
        configures the notebook driver, not Spark executor authentication.
        Close resolved clients before replacing them with another setup.
    """
    fabric.require_fabric_runtime()
    if not responses_model or not embeddings_model or not api_version:
        raise ValueError("Fabric model names and api_version must not be empty.")
    ensure_default_registrations()
    CONTAINER.register(OpenAI, lambda: fabric.provide_fabric_client(api_version=api_version))
    CONTAINER.register(AsyncOpenAI, lambda: fabric.provide_async_fabric_client(api_version=api_version))
    set_responses_model(responses_model)
    set_embeddings_model(embeddings_model)
    CONTAINER.register(
        SchemaInferer,
        lambda: SchemaInferer(client=get_client(), model_name=get_responses_model()),
    )

openaivec.set_client ¶

set_client(client: OpenAI) -> None

Register a custom OpenAI client.

Future schema inference resolves its provider again with the new client. Caller-owned clients are never closed by this function.

Parameters:

Name	Type	Description	Default
`client`	`OpenAI`	A pre-configured `openai.OpenAI` instance. To target Azure OpenAI, construct it with `base_url` ending in `/openai/v1/` and either an API key or a bearer-token provider callable as `api_key`.	required

Source code in src/openaivec/_provider.py

def set_client(client: OpenAI) -> None:
    """Register a custom ``OpenAI`` client.

    Future schema inference resolves its provider again with the new client.
    Caller-owned clients are never closed by this function.

    Args:
        client (OpenAI): A pre-configured ``openai.OpenAI`` instance. To target
            Azure OpenAI, construct it with ``base_url`` ending in ``/openai/v1/``
            and either an API key or a bearer-token provider callable as
            ``api_key``.
    """
    CONTAINER.register(OpenAI, lambda: client)
    CONTAINER._invalidate_instance(SchemaInferer)

openaivec.get_client ¶

get_client() -> OpenAI

Get the currently registered OpenAI client.

Returns:

Name	Type	Description
`OpenAI`	`OpenAI`	The registered client instance.

Source code in src/openaivec/_provider.py

def get_client() -> OpenAI:
    """Get the currently registered ``OpenAI`` client.

    Returns:
        OpenAI: The registered client instance.
    """
    return CONTAINER.resolve(OpenAI)

openaivec.set_async_client ¶

set_async_client(client: AsyncOpenAI) -> None

Register a custom AsyncOpenAI client.

Parameters:

Name	Type	Description	Default
`client`	`AsyncOpenAI`	A pre-configured `openai.AsyncOpenAI` instance. See :func:`set_client` for the Azure configuration pattern.	required

Source code in src/openaivec/_provider.py

def set_async_client(client: AsyncOpenAI) -> None:
    """Register a custom ``AsyncOpenAI`` client.

    Args:
        client (AsyncOpenAI): A pre-configured ``openai.AsyncOpenAI`` instance.
            See :func:`set_client` for the Azure configuration pattern.
    """
    CONTAINER.register(AsyncOpenAI, lambda: client)

openaivec.get_async_client ¶

get_async_client() -> AsyncOpenAI

Get the currently registered AsyncOpenAI client.

Returns:

Name	Type	Description
`AsyncOpenAI`	`AsyncOpenAI`	The registered async client instance.

Source code in src/openaivec/_provider.py

def get_async_client() -> AsyncOpenAI:
    """Get the currently registered ``AsyncOpenAI`` client.

    Returns:
        AsyncOpenAI: The registered async client instance.
    """
    return CONTAINER.resolve(AsyncOpenAI)

openaivec.set_responses_model ¶

set_responses_model(name: str) -> None

Override the model used for text responses.

Future schema inference resolves its provider again with the new model. Existing response wrappers and UDFs retain their configured model.

Parameters:

Name	Type	Description	Default
`name`	`str`	Model or deployment name (e.g. `"gpt-6-luna"`).	required

Source code in src/openaivec/_provider.py

def set_responses_model(name: str) -> None:
    """Override the model used for text responses.

    Future schema inference resolves its provider again with the new model.
    Existing response wrappers and UDFs retain their configured model.

    Args:
        name (str): Model or deployment name (e.g. ``"gpt-6-luna"``).
    """
    CONTAINER.register(ResponsesModelName, lambda: ResponsesModelName(name))
    CONTAINER._invalidate_instance(SchemaInferer)

openaivec.get_responses_model ¶

get_responses_model() -> str

Get the currently registered model name for text responses.

Returns:

Name	Type	Description
`str`	`str`	The model name.

Source code in src/openaivec/_provider.py

def get_responses_model() -> str:
    """Get the currently registered model name for text responses.

    Returns:
        str: The model name.
    """
    return CONTAINER.resolve(ResponsesModelName).value

openaivec.set_embeddings_model ¶

set_embeddings_model(name: str) -> None

Override the model used for text embeddings.

Parameters:

Name	Type	Description	Default
`name`	`str`	Model or deployment name (e.g. `"text-embedding-3-small"`).	required

Source code in src/openaivec/_provider.py

def set_embeddings_model(name: str) -> None:
    """Override the model used for text embeddings.

    Args:
        name (str): Model or deployment name (e.g. ``"text-embedding-3-small"``).
    """
    CONTAINER.register(EmbeddingsModelName, lambda: EmbeddingsModelName(name))

openaivec.get_embeddings_model ¶

get_embeddings_model() -> str

Get the currently registered model name for text embeddings.

Returns:

Name	Type	Description
`str`	`str`	The model name.

Source code in src/openaivec/_provider.py

def get_embeddings_model() -> str:
    """Get the currently registered model name for text embeddings.

    Returns:
        str: The model name.
    """
    return CONTAINER.resolve(EmbeddingsModelName).value

Core Classes¶

All core functionality is accessible through the main package imports:

openaivec.BatchResponses `dataclass` ¶

BatchResponses(
    client: OpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    cache: BatchCache[str, ResponseFormat | None] = (
        lambda: BatchCache(
            batch_size=None,
            max_cache_size=DEFAULT_MANAGED_CACHE_SIZE,
        )
    )(),
    api_kwargs: dict[str, Any] = dict(),
    max_validation_retries: int = 3,
    multimodal: bool = False,
    retry_policy: RetryPolicy | None = None,
    limits: ResponseLimits = ResponseLimits(),
)

Bases: Generic[ResponseFormat]

Stateless façade that turns OpenAI's JSON‑mode API into a batched API.

This wrapper allows you to submit multiple user prompts in one JSON‑mode request and receive the answers in the original order.

Example

vector_llm = BatchResponses(
    client=openai_client,
    model_name="gpt-6-luna",
    system_message="You are a helpful assistant.",
    api_kwargs={"reasoning": {"effort": "none"}},
)
answers = vector_llm.parse(questions)

Attributes:

Name	Type	Description
`client`	`OpenAI`	Initialised OpenAI client.
`model_name`	`str`	For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
`system_message`	`str`	System prompt prepended to every request.
`response_format`	`type[ResponseFormat]`	Expected Pydantic model class or `str` for each assistant message.
`cache`	`BatchCache[str, ResponseFormat \| None]`	Order-preserving batching proxy with de-duplication and caching. Library-managed instances use bounded retention by default. Caller-provided caches must accept `None` because a response without parsed output is cached.
`max_validation_retries`	`int`	Number of retries when structured output fails local schema validation.
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.

Notes

Internally the work is delegated to two helpers:

_predict_chunk – fragments the workload and restores ordering.
_request_llm – issues OpenAI API calls and retries with validation feedback when needed.

Methods:¶

of `classmethod` ¶

of(
    client: OpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    batch_size: int | None = None,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> BatchResponses

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
`system_message`	`str`	System prompt for the model.	required
`response_format`	`type[ResponseFormat]`	Expected output type. Defaults to `str`.	`str`
`batch_size`	`int \| None`	Max unique prompts per API call. Defaults to None (automatic batch size optimization). Set to a positive integer for fixed batch size.	`None`
`max_validation_retries`	`int`	Retry count when structured output fails local schema validation. Defaults to 3.	`3`
`multimodal`	`bool`	When `True`, file paths and URLs in inputs are sent as multimodal content. Defaults to `False`.	`False`
`limits`	`ResponseLimits \| None`	Text request context and item limits.	`None`
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.	`None`
`**api_kwargs`		Additional OpenAI API parameters (temperature, top_p, etc.).	`{}`

Returns:

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

Source code in src/openaivec/_responses.py

@classmethod
def of(
    cls,
    client: OpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    batch_size: int | None = None,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> "BatchResponses":
    """Factory constructor.

    Args:
        client (OpenAI): OpenAI client.
        model_name (str): For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
        system_message (str): System prompt for the model.
        response_format (type[ResponseFormat], optional): Expected output type. Defaults to ``str``.
        batch_size (int | None, optional): Max unique prompts per API call. Defaults to None
            (automatic batch size optimization). Set to a positive integer for fixed batch size.
        max_validation_retries (int, optional): Retry count when structured output fails local
            schema validation. Defaults to 3.
        multimodal (bool, optional): When ``True``, file paths and URLs in
            inputs are sent as multimodal content. Defaults to ``False``.
        limits (ResponseLimits | None): Text request context and item limits.
        retry_policy (RetryPolicy | None): Transport limits. ``None`` preserves SDK retries.
        **api_kwargs: Additional OpenAI API parameters (temperature, top_p, etc.).

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

of_task `classmethod` ¶

of_task(
    client: OpenAI,
    model_name: str,
    task: PreparedTask[ResponseFormat],
    batch_size: int | None = None,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> BatchResponses

Factory from a PreparedTask.

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
`task`	`PreparedTask`	Prepared task with instructions and response format.	required
`batch_size`	`int \| None`	Max unique prompts per API call. Defaults to None (automatic batch size optimization). Set to a positive integer for fixed batch size.	`None`
`max_validation_retries`	`int`	Retry count when structured output fails local schema validation. Defaults to 3.	`3`
`multimodal`	`bool`	When `True`, file paths and URLs in inputs are sent as multimodal content. Defaults to `False`.	`False`
`limits`	`ResponseLimits \| None`	Text request context and item limits.	`None`
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.	`None`
`**api_kwargs`		Additional OpenAI API parameters forwarded to the Responses API.	`{}`

Returns:

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

Source code in src/openaivec/_responses.py

@classmethod
def of_task(
    cls,
    client: OpenAI,
    model_name: str,
    task: PreparedTask[ResponseFormat],
    batch_size: int | None = None,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> "BatchResponses":
    """Factory from a PreparedTask.

    Args:
        client (OpenAI): OpenAI client.
        model_name (str): For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
        task (PreparedTask): Prepared task with instructions and response format.
        batch_size (int | None, optional): Max unique prompts per API call. Defaults to None
            (automatic batch size optimization). Set to a positive integer for fixed batch size.
        max_validation_retries (int, optional): Retry count when structured output fails local
            schema validation. Defaults to 3.
        multimodal (bool, optional): When ``True``, file paths and URLs in
            inputs are sent as multimodal content. Defaults to ``False``.
        limits (ResponseLimits | None): Text request context and item limits.
        retry_policy (RetryPolicy | None): Transport limits. ``None`` preserves SDK retries.
        **api_kwargs: Additional OpenAI API parameters forwarded to the Responses API.

    Returns:
        BatchResponses: Configured instance backed by a batching proxy.
    """
    return cls(
        client=client,
        model_name=model_name,
        system_message=task.instructions,
        response_format=task.response_format,
        cache=BatchCache(batch_size=batch_size, max_cache_size=DEFAULT_MANAGED_CACHE_SIZE),
        api_kwargs=api_kwargs,
        max_validation_retries=max_validation_retries,
        multimodal=multimodal,
        limits=limits if limits is not None else ResponseLimits(),
        retry_policy=retry_policy,
    )

parse ¶

parse(inputs: list[str]) -> list[ResponseFormat | None]

Batched predict.

Parameters:

Name	Type	Description	Default
`inputs`	`list[str]`	Prompts that require responses. Duplicates are de‑duplicated.	required

Returns:

Type	Description
`list[ResponseFormat \| None]`	list[ResponseFormat \| None]: Assistant responses aligned to `inputs`. An absent parsed result is returned as `None` and cached.

Source code in src/openaivec/_responses.py

@observe(_LOGGER)
def parse(self, inputs: list[str]) -> list[ResponseFormat | None]:
    """Batched predict.

    Args:
        inputs (list[str]): Prompts that require responses. Duplicates are de‑duplicated.

    Returns:
        list[ResponseFormat | None]: Assistant responses aligned to ``inputs``.
            An absent parsed result is returned as ``None`` and cached.
    """
    keys = [_cache_key_for_input(value, multimodal=self.multimodal) for value in inputs]
    results = self.cache.map(keys, self._predict_chunk)
    if self.multimodal:
        _ensure_files_unchanged(keys)
    return results

openaivec.AsyncBatchResponses `dataclass` ¶

AsyncBatchResponses(
    client: AsyncOpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    cache: AsyncBatchCache[str, ResponseFormat | None] = (
        lambda: AsyncBatchCache(
            batch_size=None,
            max_concurrency=8,
            max_cache_size=DEFAULT_MANAGED_CACHE_SIZE,
        )
    )(),
    api_kwargs: dict[str, Any] = dict(),
    max_validation_retries: int = 3,
    multimodal: bool = False,
    retry_policy: RetryPolicy | None = None,
    limits: ResponseLimits = ResponseLimits(),
)

Bases: Generic[ResponseFormat]

Stateless façade that turns OpenAI's JSON-mode API into a batched API (Async version).

This wrapper allows you to submit multiple user prompts in one JSON-mode request and receive the answers in the original order asynchronously. It also controls the maximum number of concurrent requests to the OpenAI API.

Example

import asyncio
from openai import AsyncOpenAI
from openaivec import AsyncBatchResponses

openai_async_client = AsyncOpenAI()  # initialize your client

vector_llm = AsyncBatchResponses.of(
    client=openai_async_client,
    model_name="gpt-6-luna",
    system_message="You are a helpful assistant.",
    batch_size=64,
    max_concurrency=5,
    reasoning={"effort": "none"},
)
questions = [
    "What is the capital of France?",
    "Explain quantum physics simply.",
]

async def main():
    answers = await vector_llm.parse(questions)
    print(answers)

asyncio.run(main())

Attributes:

Name	Type	Description
`client`	`AsyncOpenAI`	Initialised OpenAI async client.
`model_name`	`str`	For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
`system_message`	`str`	System prompt prepended to every request.
`response_format`	`type[ResponseFormat]`	Expected Pydantic model class or `str` for each assistant message.
`cache`	`AsyncBatchCache[str, ResponseFormat \| None]`	Async batching proxy with de-duplication and concurrency control. Library-managed instances use bounded retention by default. Caller-provided caches must accept `None` because a response without parsed output is cached.
`max_validation_retries`	`int`	Number of retries when structured output fails local schema validation.
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.

Methods:¶

of `classmethod` ¶

of(
    client: AsyncOpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    batch_size: int | None = None,
    max_concurrency: int = 8,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> AsyncBatchResponses

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
`system_message`	`str`	System prompt.	required
`response_format`	`type[ResponseFormat]`	Expected output type. Defaults to `str`.	`str`
`batch_size`	`int \| None`	Max unique prompts 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`
`max_validation_retries`	`int`	Retry count when structured output fails local schema validation. Defaults to 3.	`3`
`multimodal`	`bool`	When `True`, file paths and URLs in inputs are sent as multimodal content. Defaults to `False`.	`False`
`limits`	`ResponseLimits \| None`	Text request context and item limits.	`None`
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.	`None`
`**api_kwargs`		Additional OpenAI API parameters (temperature, top_p, etc.).	`{}`

Returns:

Name	Type	Description
`AsyncBatchResponses`	`AsyncBatchResponses`	Configured instance backed by an async batching proxy.

Source code in src/openaivec/_responses.py

@classmethod
def of(
    cls,
    client: AsyncOpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    batch_size: int | None = None,
    max_concurrency: int = 8,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> "AsyncBatchResponses":
    """Factory constructor.

    Args:
        client (AsyncOpenAI): OpenAI async client.
        model_name (str): For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
        system_message (str): System prompt.
        response_format (type[ResponseFormat], optional): Expected output type. Defaults to ``str``.
        batch_size (int | None, optional): Max unique prompts 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.
        max_validation_retries (int, optional): Retry count when structured output fails local
            schema validation. Defaults to 3.
        multimodal (bool, optional): When ``True``, file paths and URLs in
            inputs are sent as multimodal content. Defaults to ``False``.
        limits (ResponseLimits | None): Text request context and item limits.
        retry_policy (RetryPolicy | None): Transport limits. ``None`` preserves SDK retries.
        **api_kwargs: Additional OpenAI API parameters (temperature, top_p, etc.).

    Returns:
        AsyncBatchResponses: Configured instance backed by an async batching proxy.
    """
    return cls(
        client=client,
        model_name=model_name,
        system_message=system_message,
        response_format=response_format,
        cache=AsyncBatchCache(
            batch_size=batch_size,
            max_concurrency=max_concurrency,
            max_cache_size=DEFAULT_MANAGED_CACHE_SIZE,
        ),
        api_kwargs=api_kwargs,
        max_validation_retries=max_validation_retries,
        multimodal=multimodal,
        limits=limits if limits is not None else ResponseLimits(),
        retry_policy=retry_policy,
    )

of_task `classmethod` ¶

of_task(
    client: AsyncOpenAI,
    model_name: str,
    task: PreparedTask[ResponseFormat],
    batch_size: int | None = None,
    max_concurrency: int = 8,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> AsyncBatchResponses

Factory from a PreparedTask.

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
`task`	`PreparedTask`	Prepared task with instructions and response format.	required
`batch_size`	`int \| None`	Max unique prompts 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`
`max_validation_retries`	`int`	Retry count when structured output fails local schema validation. Defaults to 3.	`3`
`multimodal`	`bool`	When `True`, file paths and URLs in inputs are sent as multimodal content. Defaults to `False`.	`False`
`limits`	`ResponseLimits \| None`	Text request context and item limits.	`None`
`retry_policy`	`RetryPolicy \| None`	Transport limits. `None` preserves SDK retries.	`None`
`**api_kwargs`		Additional OpenAI API parameters forwarded to the Responses API.	`{}`

Returns:

Name	Type	Description
`AsyncBatchResponses`	`AsyncBatchResponses`	Configured instance backed by an async batching proxy.

Source code in src/openaivec/_responses.py

@classmethod
def of_task(
    cls,
    client: AsyncOpenAI,
    model_name: str,
    task: PreparedTask[ResponseFormat],
    batch_size: int | None = None,
    max_concurrency: int = 8,
    max_validation_retries: int = 3,
    multimodal: bool = False,
    *,
    limits: ResponseLimits | None = None,
    retry_policy: RetryPolicy | None = None,
    **api_kwargs,
) -> "AsyncBatchResponses":
    """Factory from a PreparedTask.

    Args:
        client (AsyncOpenAI): OpenAI async client.
        model_name (str): For Azure OpenAI, use your deployment name. For OpenAI, use the model name.
        task (PreparedTask): Prepared task with instructions and response format.
        batch_size (int | None, optional): Max unique prompts 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.
        max_validation_retries (int, optional): Retry count when structured output fails local
            schema validation. Defaults to 3.
        multimodal (bool, optional): When ``True``, file paths and URLs in
            inputs are sent as multimodal content. Defaults to ``False``.
        limits (ResponseLimits | None): Text request context and item limits.
        retry_policy (RetryPolicy | None): Transport limits. ``None`` preserves SDK retries.
        **api_kwargs: Additional OpenAI API parameters forwarded to the Responses API.

    Returns:
        AsyncBatchResponses: Configured instance backed by an async batching proxy.
    """
    return cls(
        client=client,
        model_name=model_name,
        system_message=task.instructions,
        response_format=task.response_format,
        cache=AsyncBatchCache(
            batch_size=batch_size,
            max_concurrency=max_concurrency,
            max_cache_size=DEFAULT_MANAGED_CACHE_SIZE,
        ),
        api_kwargs=api_kwargs,
        max_validation_retries=max_validation_retries,
        multimodal=multimodal,
        limits=limits if limits is not None else ResponseLimits(),
        retry_policy=retry_policy,
    )

parse `async` ¶

parse(inputs: list[str]) -> list[ResponseFormat | None]

Batched predict (async).

Parameters:

Name	Type	Description	Default
`inputs`	`list[str]`	Prompts that require responses. Duplicates are de‑duplicated.	required

Returns:

Type	Description
`list[ResponseFormat \| None]`	list[ResponseFormat \| None]: Assistant responses aligned to `inputs`. An absent parsed result is returned as `None` and cached.

Source code in src/openaivec/_responses.py

@observe(_LOGGER)
async def parse(self, inputs: list[str]) -> list[ResponseFormat | None]:
    """Batched predict (async).

    Args:
        inputs (list[str]): Prompts that require responses. Duplicates are de‑duplicated.

    Returns:
        list[ResponseFormat | None]: Assistant responses aligned to ``inputs``.
            An absent parsed result is returned as ``None`` and cached.
    """
    keys = [_cache_key_for_input(value, multimodal=self.multimodal) for value in inputs]
    results = await self.cache.map(keys, self._predict_chunk)
    if self.multimodal:
        _ensure_files_unchanged(keys)
    return results

Responses request budgets¶

Text Responses requests are planned against a conservative 128,000-token context estimate and 128 items per request, independently of cache batch_size. Each request includes the serialized input envelope, vectorized instructions, generated structured-output schema, 256 expected output tokens per item, and a 256-token validation-feedback allowance when retries are on. batch_size=None remains adaptive; fixed sizes and nonpositive sizes also respect the request budget. An oversized single text input raises ValueError before any requests for that cache chunk are sent. Results preserve IDs, order, and cache deduplication across request splits.

Check the actual context window and output requirements of your model or Azure deployment. The default is not model metadata and token counts are estimates, not a guarantee against server-side request limits. Override the limits and tokenizer when deployment metadata is unavailable:

from openaivec import BatchResponses, ResponseLimits

responses = BatchResponses.of(
    client,
    "my-deployment",
    "Summarize each item.",
    batch_size=64,
    limits=ResponseLimits(
        max_request_tokens=32000,
        max_inputs=32,
        expected_output_tokens_per_item=512,
        encoding_name="o200k_base",
    ),
)

The same limits option works on AsyncBatchResponses.of, both of_task factories, and direct constructors. max_output_tokens (if supplied to the API) is also reserved when larger than the per-item allowance. The planner applies only to batched text inputs, not individual multimodal requests.

Multimodal files and URLs¶

Set multimodal=True to classify local images and binary documents, or image and document URLs, by their file extension. Signed URL query parameters are preserved in the request but are not used to classify the media type. Audio (.mp3 and .wav) is rejected by the Responses API wrapper.

Local text files are read into the text batch; images are inlined. Local binary documents are uploaded to the Files API as temporary, request-owned files and deleted when the request finishes, including failed requests. Validation correction attempts reuse the same upload before it is deleted. The caller remains responsible for any files uploaded directly through the OpenAI client or through the internal builder's standalone build() method. The cache identifies local files by their contents as well as path, so a changed file does not reuse an earlier result.

openaivec.ResponseLimits `dataclass` ¶

ResponseLimits(
    max_request_tokens: int = 128000,
    max_inputs: int = 128,
    expected_output_tokens_per_item: int = 256,
    validation_feedback_tokens: int = 256,
    encoding_name: str | None = None,
)

Estimated per-request context budget for batched text Responses.

max_request_tokens is the model/deployment context limit, not an API parameter. Override it and encoding_name for custom Azure deployments. Output and validation allowances are estimates; choose values appropriate for the task and model. A correction whose actual feedback exceeds the configured budget fails before a request is sent. Multimodal requests use a separate path.

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]

Prompt Building¶

openaivec.FewShotPromptBuilder ¶

FewShotPromptBuilder()

Builder for creating few-shot prompts with validation.

Usage

builder = (FewShotPromptBuilder() .purpose("Your task description") .example("input1", "output1") # At least one required .example("input2", "output2") .build())

Note

Both .purpose() and at least one .example() call are required before calling .build(), .improve(), or .get_object().

Initialize an empty FewShotPromptBuilder.

Note

You must call .purpose() and at least one .example() before building.

Source code in src/openaivec/_prompt.py

def __init__(self):
    """Initialize an empty FewShotPromptBuilder.

    Note:
        You must call .purpose() and at least one .example() before building.
    """
    self._prompt = FewShotPrompt(purpose="", cautions=[], examples=[])

Methods:¶

of `classmethod` ¶

of(prompt: FewShotPrompt) -> FewShotPromptBuilder

Create a builder pre‑populated with an existing prompt.

Parameters:

Name	Type	Description	Default
`prompt`	`FewShotPrompt`	The prompt to start from.	required

Returns:

Name	Type	Description
`FewShotPromptBuilder`	`FewShotPromptBuilder`	A new builder instance.

Source code in src/openaivec/_prompt.py

@classmethod
def of(cls, prompt: FewShotPrompt) -> "FewShotPromptBuilder":
    """Create a builder pre‑populated with an existing prompt.

    Args:
        prompt (FewShotPrompt): The prompt to start from.

    Returns:
        FewShotPromptBuilder: A new builder instance.
    """
    builder = cls()
    builder._prompt = prompt
    return builder

of_empty `classmethod` ¶

of_empty() -> FewShotPromptBuilder

Create a builder.

Returns:

Name	Type	Description
`FewShotPromptBuilder`	`FewShotPromptBuilder`	A new builder instance with an empty prompt.

Source code in src/openaivec/_prompt.py

@classmethod
def of_empty(cls) -> "FewShotPromptBuilder":
    """Create a builder.

    Returns:
        FewShotPromptBuilder: A new builder instance with an empty prompt.
    """
    return cls.of(FewShotPrompt(purpose="", cautions=[], examples=[]))

purpose ¶

purpose(purpose: str) -> FewShotPromptBuilder

Set the purpose of the prompt.

Parameters:

Name	Type	Description	Default
`purpose`	`str`	A concise statement describing the prompt’s goal.	required

Returns:

Name	Type	Description
`FewShotPromptBuilder`	`FewShotPromptBuilder`	The current builder instance (for chaining).

Source code in src/openaivec/_prompt.py

def purpose(self, purpose: str) -> "FewShotPromptBuilder":
    """Set the purpose of the prompt.

    Args:
        purpose (str): A concise statement describing the prompt’s goal.

    Returns:
        FewShotPromptBuilder: The current builder instance (for chaining).
    """
    self._prompt.purpose = purpose
    return self

caution ¶

caution(caution: str) -> FewShotPromptBuilder

Append a cautionary note to the prompt.

Parameters:

Name	Type	Description	Default
`caution`	`str`	A caution or edge‑case description.	required

Returns:

Name	Type	Description
`FewShotPromptBuilder`	`FewShotPromptBuilder`	The current builder instance.

Source code in src/openaivec/_prompt.py

def caution(self, caution: str) -> "FewShotPromptBuilder":
    """Append a cautionary note to the prompt.

    Args:
        caution (str): A caution or edge‑case description.

    Returns:
        FewShotPromptBuilder: The current builder instance.
    """
    if self._prompt.cautions is None:
        self._prompt.cautions = []
    self._prompt.cautions.append(caution)
    return self

example ¶

example(
    input_value: str | BaseModel,
    output_value: str | BaseModel,
) -> FewShotPromptBuilder

Add a single input/output example.

At least one example is required before calling .build(), .improve(), or .get_object().

Parameters:

Name	Type	Description	Default
`input_value`	`str \| BaseModel`	Example input; if a Pydantic model is provided it is serialised to JSON.	required
`output_value`	`str \| BaseModel`	Expected output; serialised if needed.	required

Returns:

Name	Type	Description
`FewShotPromptBuilder`	`FewShotPromptBuilder`	The current builder instance.

Source code in src/openaivec/_prompt.py

def example(
    self,
    input_value: str | BaseModel,
    output_value: str | BaseModel,
) -> "FewShotPromptBuilder":
    """Add a single input/output example.

    At least one example is required before calling .build(), .improve(), or .get_object().

    Args:
        input_value (str | BaseModel): Example input; if a Pydantic model is
            provided it is serialised to JSON.
        output_value (str | BaseModel): Expected output; serialised if needed.

    Returns:
        FewShotPromptBuilder: The current builder instance.
    """
    if self._prompt.examples is None:
        self._prompt.examples = []

    input_string = input_value if isinstance(input_value, str) else input_value.model_dump_json()
    output_string = output_value if isinstance(output_value, str) else output_value.model_dump_json()
    self._prompt.examples.append(Example(input=input_string, output=output_string))
    return self

improve ¶

improve(
    client: OpenAI | None = None,
    model_name: str | None = None,
    **api_kwargs,
) -> FewShotPromptBuilder

Iteratively refine the prompt using an LLM.

The method calls a single LLM request that returns multiple editing steps and stores each step for inspection.

When client is None, automatically creates a client using environment variables: - For OpenAI: OPENAI_API_KEY - For Azure OpenAI: AZURE_OPENAI_BASE_URL with AZURE_OPENAI_API_KEY or Entra ID (DefaultAzureCredential)

Parameters:

Name	Type	Description	Default
`client`	`OpenAI \| None`	Configured OpenAI client. If None, uses DI container with environment variables.	`None`
`model_name`	`str \| None`	Model identifier. If None, uses default `gpt-6-luna`.	`None`
`**api_kwargs`		Additional OpenAI API parameters (temperature, top_p, etc.).	`{}`

Returns:

Name	Type	Description
`FewShotPromptBuilder`	`FewShotPromptBuilder`	The current builder instance containing the refined prompt and iteration history.

Raises:

Type	Description
`ValueError`	If the prompt is not valid (missing purpose or examples).

Source code in src/openaivec/_prompt.py

def improve(
    self,
    client: OpenAI | None = None,
    model_name: str | None = None,
    **api_kwargs,
) -> "FewShotPromptBuilder":
    """Iteratively refine the prompt using an LLM.

    The method calls a single LLM request that returns multiple
    editing steps and stores each step for inspection.

    When client is None, automatically creates a client using environment variables:
    - For OpenAI: ``OPENAI_API_KEY``
    - For Azure OpenAI: ``AZURE_OPENAI_BASE_URL`` with ``AZURE_OPENAI_API_KEY``
      or Entra ID (``DefaultAzureCredential``)

    Args:
        client (OpenAI | None): Configured OpenAI client. If None, uses DI container with environment variables.
        model_name (str | None): Model identifier. If None, uses default ``gpt-6-luna``.
        **api_kwargs: Additional OpenAI API parameters (temperature, top_p, etc.).

    Returns:
        FewShotPromptBuilder: The current builder instance containing the refined prompt and iteration history.

    Raises:
        ValueError: If the prompt is not valid (missing purpose or examples).
    """
    # Validate before making API call to provide early feedback
    self._validate()

    _client = client or CONTAINER.resolve(OpenAI)
    _model_name = model_name or CONTAINER.resolve(ResponsesModelName).value

    response: ParsedResponse[Response] = _client.responses.parse(
        model=_model_name,
        instructions=_PROMPT,
        input=Request(prompt=self._prompt).model_dump_json(),
        text_format=Response,
        **api_kwargs,
    )

    # keep the original prompt
    self._steps = [Step(id=0, analysis="Original Prompt", prompt=self._prompt)]

    # add the histories
    if response.output_parsed:
        for step in response.output_parsed.iterations:
            self._steps.append(step)

    # set the final prompt
    self._prompt = self._steps[-1].prompt

    # Validate the improved prompt to ensure examples weren't removed by LLM
    try:
        self._validate()
    except ValueError as e:
        _logger.warning(f"LLM produced invalid prompt during improve(): {e}")
        # Restore original prompt if LLM produced invalid result
        self._prompt = self._steps[0].prompt
        raise ValueError(
            f"LLM improvement failed to maintain required fields: {e}. "
            "This may indicate an issue with the improvement instructions or model behavior."
        )

    return self

explain ¶

explain() -> FewShotPromptBuilder

Pretty‑print the diff of each improvement iteration.

Returns:

Name	Type	Description
`FewShotPromptBuilder`	`FewShotPromptBuilder`	The current builder instance.

Source code in src/openaivec/_prompt.py

def explain(self) -> "FewShotPromptBuilder":
    """Pretty‑print the diff of each improvement iteration.

    Returns:
        FewShotPromptBuilder: The current builder instance.
    """
    if not hasattr(self, "_steps") or not self._steps:
        print("No improvement steps available. Call improve() first.")
        return self

    for previous, current in zip(self._steps, self._steps[1:]):
        print(f"=== Iteration {current.id} ===\n")
        print(f"Instruction: {current.analysis}")
        diff = difflib.unified_diff(
            _render_prompt(previous.prompt).splitlines(),
            _render_prompt(current.prompt).splitlines(),
            fromfile="before",
            tofile="after",
            lineterm="",
        )
        for line in diff:
            print(line)
    return self

get_object ¶

get_object() -> FewShotPrompt

Return the underlying FewShotPrompt object.

Returns:

Name	Type	Description
`FewShotPrompt`	`FewShotPrompt`	The validated prompt object.

Source code in src/openaivec/_prompt.py

def get_object(self) -> FewShotPrompt:
    """Return the underlying FewShotPrompt object.

    Returns:
        FewShotPrompt: The validated prompt object.
    """
    self._validate()
    return self._prompt

build ¶

build() -> str

Build and return the prompt as XML.

Returns:

Name	Type	Description
`str`	`str`	XML representation of the prompt.

Source code in src/openaivec/_prompt.py

def build(self) -> str:
    """Build and return the prompt as XML.

    Returns:
        str: XML representation of the prompt.
    """
    self._validate()
    return self.build_xml()

build_json ¶

build_json(**kwargs) -> str

Build and return the prompt as a JSON string.

Parameters:

Name	Type	Description	Default
`**kwargs`		Keyword arguments forwarded to Pydantic's `model_dump_json`. Common options include `indent`, `include`, `exclude`, `by_alias`, `exclude_unset`, `exclude_defaults`, `exclude_none`.	`{}`

Returns:

Name	Type	Description
`str`	`str`	JSON representation of the prompt.

Source code in src/openaivec/_prompt.py

def build_json(self, **kwargs) -> str:
    """Build and return the prompt as a JSON string.

    Args:
        **kwargs: Keyword arguments forwarded to Pydantic's ``model_dump_json``.
            Common options include ``indent``, ``include``, ``exclude``,
            ``by_alias``, ``exclude_unset``, ``exclude_defaults``, ``exclude_none``.

    Returns:
        str: JSON representation of the prompt.
    """
    self._validate()
    return self._prompt.model_dump_json(**kwargs)

build_xml ¶

build_xml() -> str

Alias for meth:build for explicit XML generation.

Returns:

Name	Type	Description
`str`	`str`	XML representation of the prompt.

Source code in src/openaivec/_prompt.py

def build_xml(self) -> str:
    """Alias for :py:meth:`build` for explicit XML generation.

    Returns:
        str: XML representation of the prompt.
    """
    self._validate()
    return _render_prompt(self._prompt)

Main Package API¶

Configuration¶

openaivec.setup_fabric ¶

openaivec.set_client ¶

openaivec.get_client ¶

openaivec.set_async_client ¶

openaivec.get_async_client ¶

openaivec.set_responses_model ¶

openaivec.get_responses_model ¶

openaivec.set_embeddings_model ¶

openaivec.get_embeddings_model ¶

Core Classes¶

openaivec.BatchResponses dataclass ¶

Methods:¶

of classmethod ¶

of_task classmethod ¶

parse ¶

openaivec.AsyncBatchResponses dataclass ¶

Methods:¶

of classmethod ¶

of_task classmethod ¶

parse async ¶

Responses request budgets¶

Multimodal files and URLs¶

openaivec.ResponseLimits dataclass ¶

openaivec.BatchEmbeddings dataclass ¶

Methods:¶

of classmethod ¶

create ¶

openaivec.AsyncBatchEmbeddings dataclass ¶

Methods:¶

of classmethod ¶

create async ¶

Prompt Building¶

openaivec.FewShotPromptBuilder ¶

Methods:¶

of classmethod ¶

of_empty classmethod ¶

purpose ¶

caution ¶

example ¶

improve ¶

explain ¶

get_object ¶

build ¶

build_json ¶

build_xml ¶

openaivec.BatchResponses `dataclass` ¶

of `classmethod` ¶

of_task `classmethod` ¶

openaivec.AsyncBatchResponses `dataclass` ¶

of `classmethod` ¶

of_task `classmethod` ¶

parse `async` ¶

openaivec.ResponseLimits `dataclass` ¶

openaivec.BatchEmbeddings `dataclass` ¶

of `classmethod` ¶

openaivec.AsyncBatchEmbeddings `dataclass` ¶

of `classmethod` ¶

create `async` ¶

of `classmethod` ¶

of_empty `classmethod` ¶