Schema Inference¶

SchemaInferer and AsyncSchemaInferer use the supplied client and model for every inference attempt. The asynchronous variant does not create or resolve a synchronous client and propagates cancellation to the active request.

max_retries=8 means at most eight total schema inference attempts, not eight additional retries. It must be at least one. Missing parsed output, SDK schema validation errors, and invalid generated models receive bounded correction feedback. Unrelated API or configuration errors propagate immediately.

The pandas series.ai.infer_schema() and await series.aio.infer_schema() methods forward API parameters such as store, timeout, temperature, and max_output_tokens to the configured client. These options are not schema input fields. Schema-less .aio.parse() also uses the configured asynchronous client for both inference and extraction. Provide an explicit schema to skip inference.

The pandas parse() / parse_with_cache() methods and Spark parse_udf() separate max_retries=8 (total schema inference attempts) from max_validation_retries=3 (additional extraction corrections). Set the latter to zero to disable extraction corrections. Neither option is forwarded as an OpenAI API parameter. store, timeout, and other API options apply to both inference and extraction; neither validation control changes transport retries. Use retry_policy to apply transport limits to both stages. Spark infer_schema() also accepts max_retries, retry_policy, and API options.

import openaivec
from openai import AsyncOpenAI

async def infer():
    async with AsyncOpenAI() as client:
        inferer = openaivec.AsyncSchemaInferer(client=client, model_name="gpt-4.1-mini")
        return await inferer.infer_schema(
            openaivec.SchemaInferenceInput(
                examples=["Order 42 has shipped"],
                instructions="Extract order status",
            ),
            max_retries=2,
            store=False,
        )

openaivec.SchemaInferer `dataclass` ¶

SchemaInferer(client: OpenAI, model_name: str)

High-level orchestrator for schema inference against the Responses API.

Responsibilities

Issue a structured parsing request with strict instructions.
Retry (up to max_retries) when the produced field list violates baseline structural rules (duplicate names, unsupported types, etc.).
Return a fully validated InferredSchema ready for dynamic model generation & downstream batch extraction.

The inferred schema intentionally avoids JSON Schema intermediates; the authoritative contract is the ordered FieldSpec list.

Attributes:

Name	Type	Description
`client`	`OpenAI`	OpenAI client for calling `responses.parse`.
`model_name`	`str`	Model / deployment identifier.

Methods:¶

infer_schema ¶

infer_schema(
    data: SchemaInferenceInput,
    *args: Any,
    max_retries: int = 8,
    retry_policy: RetryPolicy | None = None,
    **kwargs: Any,
) -> SchemaInferenceOutput

Infer a validated schema from representative examples.

Workflow: 1. Submit SchemaInferenceInput (JSON) + instructions via responses.parse requesting an InferredSchema object. 2. Attempt dynamic model build (parsed.build_model()) which performs recursive structural validation (names, types, enum/object specs) via the dynamic layer. 3. Retry (up to max_retries) on validation failure.

Parameters:

Name	Type	Description	Default
`data`	`SchemaInferenceInput`	Representative examples + instructions.	required
`*args`	`Any`	Positional passthrough to `client.responses.parse`.	`()`
`max_retries`	`int`	Attempts before surfacing the last validation error (must be >= 1). Defaults to 8.	`8`
`retry_policy`	`RetryPolicy \| None`	Transport policy. None preserves SDK settings.	`None`
`**kwargs`	`Any`	Keyword passthrough to `client.responses.parse`.	`{}`

Returns:

Name	Type	Description
`InferredSchema`	`SchemaInferenceOutput`	Fully validated schema (instructions, examples summary,
	`SchemaInferenceOutput`	ordered fields, extraction prompt).

Raises:

Type	Description
`ValueError`	Validation still fails after exhausting retries.

Source code in src/openaivec/_schema/infer.py

def infer_schema(
    self,
    data: SchemaInferenceInput,
    *args: Any,
    max_retries: int = 8,
    retry_policy: RetryPolicy | None = None,
    **kwargs: Any,
) -> SchemaInferenceOutput:
    """Infer a validated schema from representative examples.

      Workflow:
            1. Submit ``SchemaInferenceInput`` (JSON) + instructions via
                ``responses.parse`` requesting an ``InferredSchema`` object.
            2. Attempt dynamic model build (``parsed.build_model()``) which performs recursive
                structural validation (names, types, enum/object specs) via the dynamic layer.
            3. Retry (up to ``max_retries``) on validation failure.

    Args:
        data (SchemaInferenceInput): Representative examples + instructions.
        *args: Positional passthrough to ``client.responses.parse``.
        max_retries (int, optional): Attempts before surfacing the last validation error
            (must be >= 1). Defaults to 8.
        retry_policy (RetryPolicy | None): Transport policy. None preserves SDK settings.
        **kwargs: Keyword passthrough to ``client.responses.parse``.

    Returns:
        InferredSchema: Fully validated schema (instructions, examples summary,
        ordered fields, extraction prompt).

    Raises:
        ValueError: Validation still fails after exhausting retries.
    """
    if max_retries < 1:
        raise ValueError("max_retries must be >= 1")

    last_err: ValueError | None = None
    previous_errors: list[str] = []
    input_json = data.model_dump_json()
    deadline = retry_deadline(retry_policy)
    for _ in range(max_retries):
        try:
            response: ParsedResponse[SchemaInferenceOutput] = call_with_retry(
                self.client,
                retry_policy,
                lambda client, options: client.responses.parse(
                    model=self.model_name,
                    instructions=_schema_instructions(previous_errors),
                    input=input_json,
                    text_format=SchemaInferenceOutput,
                    *args,
                    **options,
                ),
                kwargs,
                deadline=deadline,
            )
        except ValidationError as error:
            last_err = error
        else:
            try:
                return _validated_schema(response.output_parsed)
            except ValueError as error:
                last_err = error
        previous_errors.append(str(last_err))
    raise ValueError(f"Schema validation failed after {max_retries} attempts. Last error: {last_err}") from last_err

openaivec.AsyncSchemaInferer `dataclass` ¶

AsyncSchemaInferer(client: AsyncOpenAI, model_name: str)

Infer schemas using a configured asynchronous Responses client.

Attributes:

Name	Type	Description
`client`	`AsyncOpenAI`	Client used for every inference attempt.
`model_name`	`str`	Model or deployment identifier.

Methods:¶

infer_schema `async` ¶

infer_schema(
    data: SchemaInferenceInput,
    *args: Any,
    max_retries: int = 8,
    retry_policy: RetryPolicy | None = None,
    **kwargs: Any,
) -> SchemaInferenceOutput

Infer and validate a schema without resolving a synchronous client.

Parameters:

Name	Type	Description	Default
`data`	`SchemaInferenceInput`	Representative examples and instructions.	required
`*args`	`Any`	Positional passthrough to `client.responses.parse`.	`()`
`max_retries`	`int`	Maximum inference attempts, at least 1. Defaults to 8, matching `SchemaInferer`.	`8`
`retry_policy`	`RetryPolicy \| None`	Transport policy. None preserves SDK settings.	`None`
`**kwargs`	`Any`	Keyword passthrough to `client.responses.parse`.	`{}`

Returns:

Name	Type	Description
`SchemaInferenceOutput`	`SchemaInferenceOutput`	Validated schema and extraction prompt.

Raises:

Type	Description
`ValueError`	Invalid attempt limit or exhausted schema validation.

Source code in src/openaivec/_schema/infer.py

async def infer_schema(
    self,
    data: SchemaInferenceInput,
    *args: Any,
    max_retries: int = 8,
    retry_policy: RetryPolicy | None = None,
    **kwargs: Any,
) -> SchemaInferenceOutput:
    """Infer and validate a schema without resolving a synchronous client.

    Args:
        data (SchemaInferenceInput): Representative examples and instructions.
        *args: Positional passthrough to ``client.responses.parse``.
        max_retries (int, optional): Maximum inference attempts, at least 1.
            Defaults to 8, matching ``SchemaInferer``.
        retry_policy (RetryPolicy | None): Transport policy. None preserves SDK settings.
        **kwargs: Keyword passthrough to ``client.responses.parse``.

    Returns:
        SchemaInferenceOutput: Validated schema and extraction prompt.

    Raises:
        ValueError: Invalid attempt limit or exhausted schema validation.
    """
    if max_retries < 1:
        raise ValueError("max_retries must be >= 1")
    last_err: ValueError | None = None
    previous_errors: list[str] = []
    input_json = data.model_dump_json()
    deadline = retry_deadline(retry_policy)
    for _ in range(max_retries):
        try:
            response: ParsedResponse[SchemaInferenceOutput] = await call_with_retry_async(
                self.client,
                retry_policy,
                lambda client, options: client.responses.parse(
                    model=self.model_name,
                    instructions=_schema_instructions(previous_errors),
                    input=input_json,
                    text_format=SchemaInferenceOutput,
                    *args,
                    **options,
                ),
                kwargs,
                deadline=deadline,
            )
        except ValidationError as error:
            last_err = error
        else:
            try:
                return _validated_schema(response.output_parsed)
            except ValueError as error:
                last_err = error
        previous_errors.append(str(last_err))
    raise ValueError(f"Schema validation failed after {max_retries} attempts. Last error: {last_err}") from last_err

openaivec.SchemaInferenceInput ¶

Bases: BaseModel

Input payload for schema inference.

Attributes:

Name	Type	Description
`examples`	`list[str]`	Representative sample texts restricted to the in‑scope distribution (exclude outliers / noise). Size should be minimal yet sufficient to surface recurring patterns.
`instructions`	`str`	Plain language description of downstream usage (analytics, filtering, enrichment, feature engineering, etc.). Guides field relevance & exclusion of outcome labels.

openaivec.SchemaInferenceOutput ¶

Bases: BaseModel

Result of a schema inference round.

Contains the normalized instructions, objective examples_summary, the root hierarchical object_spec contract, and the canonical reusable inference_prompt. The prompt MUST be fully derivable from the other components (no new unstated facts) to preserve traceability.

Attributes:

Name	Type	Description
`instructions`	`str`	Unambiguous restatement of the user's objective.
`examples_summary`	`str`	Neutral description of structural / semantic patterns observed in the examples.
`examples_instructions_alignment`	`str`	Mapping from instructions facets to concrete recurring evidence (or explicit gaps) anchoring extraction scope.
`object_spec`	`ObjectSpec`	Root `ObjectSpec` (UpperCamelCase name) whose `fields` recursively define the extraction schema.
`inference_prompt`	`str`	Canonical instructions enforcing exact field names, hierarchy, and types (no additions/removals/renames).

Attributes¶

model `property` ¶

model: type[BaseModel]

Dynamically materialized Pydantic model for the inferred schema.

Equivalent to calling :meth:build_model each access (not cached).

Returns:

Type	Description
`type[BaseModel]`	type[BaseModel]: Fresh model type reflecting `fields` ordering.

task `property` ¶

task: PreparedTask[BaseModel]

PreparedTask integrating the schema's extraction prompt & model.

Returns:

Name	Type	Description
`PreparedTask`	`PreparedTask[BaseModel]`	Ready for batched structured extraction calls.

Methods:¶

load `classmethod` ¶

load(path: str) -> SchemaInferenceOutput

Load an inferred schema from a JSON file.

Parameters:

Name	Type	Description	Default
`path`	`str`	Path to a UTF‑8 JSON document previously produced via `save`.	required

Returns:

Name	Type	Description
`InferredSchema`	`SchemaInferenceOutput`	Reconstructed instance.

Source code in src/openaivec/_schema/infer.py

@classmethod
def load(cls, path: str) -> "SchemaInferenceOutput":
    """Load an inferred schema from a JSON file.

    Args:
        path (str): Path to a UTF‑8 JSON document previously produced via ``save``.

    Returns:
        InferredSchema: Reconstructed instance.
    """
    with open(path, "r", encoding="utf-8") as f:
        return cls.model_validate_json(f.read())

save ¶

save(path: str) -> None

Persist this inferred schema as pretty‑printed JSON.

Parameters:

Name	Type	Description	Default
`path`	`str`	Destination filesystem path.	required

Source code in src/openaivec/_schema/infer.py

def save(self, path: str) -> None:
    """Persist this inferred schema as pretty‑printed JSON.

    Args:
        path (str): Destination filesystem path.
    """
    with open(path, "w", encoding="utf-8") as f:
        f.write(self.model_dump_json(indent=2))

Schema Inference¶

openaivec.SchemaInferer dataclass ¶

Methods:¶

infer_schema ¶

openaivec.AsyncSchemaInferer dataclass ¶

Methods:¶

infer_schema async ¶

openaivec.SchemaInferenceInput ¶

openaivec.SchemaInferenceOutput ¶

Attributes¶

model property ¶

task property ¶

Methods:¶

load classmethod ¶

save ¶

openaivec.SchemaInferer `dataclass` ¶

openaivec.AsyncSchemaInferer `dataclass` ¶

infer_schema `async` ¶

model `property` ¶

task `property` ¶

load `classmethod` ¶