Main Package API¶

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

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: BatchingMapProxy[str, ResponseFormat] = (
        lambda: BatchingMapProxy(batch_size=None)
    )(),
    api_kwargs: dict[
        str, int | float | str | bool
    ] = dict(),
)

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‑4o‑mini",
    system_message="You are a helpful assistant."
)
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`	`BatchingMapProxy[str, ResponseFormat]`	Order‑preserving batching proxy with de‑duplication and caching.

Notes

Internally the work is delegated to two helpers:

_predict_chunk – fragments the workload and restores ordering.
_request_llm – performs a single OpenAI API call.

Functions¶

of `classmethod` ¶

of(
    client: OpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    batch_size: int | 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`
`**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,
    **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.
        **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=BatchingMapProxy(batch_size=batch_size),
        api_kwargs=api_kwargs,
    )

of_task `classmethod` ¶

of_task(
    client: OpenAI,
    model_name: str,
    task: PreparedTask[ResponseFormat],
    batch_size: int | None = None,
) -> 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`

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
) -> "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.

    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=BatchingMapProxy(batch_size=batch_size),
        api_kwargs=task.api_kwargs,
    )

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`.

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``.
    """
    return self.cache.map(inputs, self._predict_chunk)  # type: ignore[return-value]

openaivec.AsyncBatchResponses `dataclass` ¶

AsyncBatchResponses(
    client: AsyncOpenAI,
    model_name: str,
    system_message: str,
    response_format: type[ResponseFormat] = str,
    cache: AsyncBatchingMapProxy[str, ResponseFormat] = (
        lambda: AsyncBatchingMapProxy(
            batch_size=None, max_concurrency=8
        )
    )(),
    api_kwargs: dict[
        str, int | float | str | bool
    ] = dict(),
)

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-4.1-mini",
    system_message="You are a helpful assistant.",
    batch_size=64,
    max_concurrency=5,
)
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`	`AsyncBatchingMapProxy[str, ResponseFormat]`	Async batching proxy with de‑duplication and concurrency control.

Functions¶

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,
    **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`
`**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,
    **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.
        **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=AsyncBatchingMapProxy(batch_size=batch_size, max_concurrency=max_concurrency),
        api_kwargs=api_kwargs,
    )

of_task `classmethod` ¶

of_task(
    client: AsyncOpenAI,
    model_name: str,
    task: PreparedTask[ResponseFormat],
    batch_size: int | None = None,
    max_concurrency: int = 8,
) -> 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`

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,
) -> "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.

    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=AsyncBatchingMapProxy(batch_size=batch_size, max_concurrency=max_concurrency),
        api_kwargs=task.api_kwargs,
    )

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`.

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``.
    """
    return await self.cache.map(inputs, self._predict_chunk)  # type: ignore[return-value]

openaivec.BatchEmbeddings `dataclass` ¶

BatchEmbeddings(
    client: OpenAI,
    model_name: str,
    cache: BatchingMapProxy[str, NDArray[float32]] = (
        lambda: BatchingMapProxy(batch_size=None)
    )(),
    api_kwargs: dict[
        str, int | float | str | bool
    ] = dict(),
)

Thin wrapper around the OpenAI embeddings endpoint (synchronous).

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`	`BatchingMapProxy[str, NDArray[float32]]`	Batching proxy for ordered, cached mapping.
`api_kwargs`	`dict[str, Any]`	Additional OpenAI API parameters stored at initialization.

Functions¶

of `classmethod` ¶

of(
    client: OpenAI,
    model_name: str,
    batch_size: int | 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`
`**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, **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.
        **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=BatchingMapProxy(batch_size=batch_size),
        api_kwargs=api_kwargs,
    )

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: AsyncBatchingMapProxy[str, NDArray[float32]] = (
        lambda: AsyncBatchingMapProxy(
            batch_size=None, max_concurrency=8
        )
    )(),
    api_kwargs: dict[
        str, int | float | str | bool
    ] = dict(),
)

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.

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`	`AsyncBatchingMapProxy[str, NDArray[float32]]`	Async batching proxy.
`api_kwargs`	`dict`	Additional OpenAI API parameters stored at initialization.

Functions¶

of `classmethod` ¶

of(
    client: AsyncOpenAI,
    model_name: str,
    batch_size: int | None = None,
    max_concurrency: int = 8,
    **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`
`**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,
    **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.
        **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=AsyncBatchingMapProxy(batch_size=batch_size, max_concurrency=max_concurrency),
        api_kwargs=api_kwargs,
    )

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=[])

Functions¶

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_API_KEY, AZURE_OPENAI_BASE_URL, AZURE_OPENAI_API_VERSION

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-4.1-mini`.	`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_API_KEY``, ``AZURE_OPENAI_BASE_URL``, ``AZURE_OPENAI_API_VERSION``

    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-4.1-mini``.
        **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¶

Core Classes¶

openaivec.BatchResponses dataclass ¶

Functions¶

of classmethod ¶

of_task classmethod ¶

parse ¶

openaivec.AsyncBatchResponses dataclass ¶

Functions¶

of classmethod ¶

of_task classmethod ¶

parse async ¶

openaivec.BatchEmbeddings dataclass ¶

Functions¶

of classmethod ¶

create ¶

openaivec.AsyncBatchEmbeddings dataclass ¶

Functions¶

of classmethod ¶

create async ¶

Prompt Building¶

openaivec.FewShotPromptBuilder ¶

Functions¶

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.BatchEmbeddings `dataclass` ¶

of `classmethod` ¶

openaivec.AsyncBatchEmbeddings `dataclass` ¶

of `classmethod` ¶

create `async` ¶

of `classmethod` ¶

of_empty `classmethod` ¶