oumi.core.synthesis

Contents

oumi.core.synthesis#

Submodules#

oumi.core.synthesis.attribute_formatter module#

class oumi.core.synthesis.attribute_formatter.AttributeFormatter(params: GeneralSynthesisParams)[source]#

Bases: object

Formats a sample using a format string.

Integrates information from permutable attributes to support formatting of placeholders in the format string (i.e. {attribute_id.value}).

format(sample: dict[str, str], format_string: str, missing_values_allowed: bool = False) → str[source]#

Format a sample using a format string.

Parameters:

  • sample – The sample to format.

  • format_string – The format string to use.

  • missing_values_allowed – If True, missing values are allowed in the sample.

Returns:

The formatted string.

oumi.core.synthesis.attribute_synthesizer module#

class oumi.core.synthesis.attribute_synthesizer.AttributeSynthesizer(params: GeneralSynthesisParams, inference_config: InferenceConfig)[source]#

Bases: object

Synthesizes values for a generated attribute based on the given samples.

Parameters:

  • params – The parameters for the attribute synthesizer.

  • inference_config – The configuration for the inference engine.

build_batch_conversations(samples: list[dict], generated_attribute: GeneratedAttribute) → list[Conversation][source]#

Build inference conversations from samples for batch operations.

cancel_batch(batch_id: str) → BatchInfo[source]#

Cancel a batch inference job.

Parameters:

batch_id – The batch ID returned from synthesize_batch().

Returns:

BatchInfo containing the updated job status.

Raises:

NotImplementedError – If the inference engine does not support batch.

get_batch_results(batch_id: str, samples: list[dict], generated_attribute: GeneratedAttribute) → list[dict[str, str]][source]#

Get results from a completed batch inference job.

Parameters:

  • batch_id – The batch ID returned from synthesize_batch().

  • samples – The original samples that were submitted for synthesis.

  • generated_attribute – The generated attribute configuration.

Returns:

A list of dictionaries, one for each sample, with the generated attribute value added to the dictionary (same format as synthesize()).

Raises:

  • NotImplementedError – If the inference engine does not support batch.

  • RuntimeError – If any items failed synthesis.

get_batch_results_partial(batch_id: str, samples: list[dict], generated_attribute: GeneratedAttribute) → SynthBatchResult[source]#

Get partial results from a completed batch inference job.

This method returns successful results alongside failure information. Parse failures from process_inference_results are also captured.

Parameters:

  • batch_id – The batch ID returned from synthesize_batch().

  • samples – The original samples that were submitted for synthesis.

  • generated_attribute – The generated attribute configuration.

Returns:

SynthBatchResult with successful outputs and failure info.

Raises:

NotImplementedError – If the inference engine does not support batch.

get_batch_status(batch_id: str) → BatchInfo[source]#

Get the status of a batch inference job.

Parameters:

batch_id – The batch ID returned from synthesize_batch().

Returns:

BatchInfo containing the job status and progress information.

Raises:

NotImplementedError – If the inference engine does not support batch.

process_inference_results(inference_results: list[Conversation], generated_attribute: GeneratedAttribute) → list[dict[str, str]][source]#

Extract and postprocess inference results.

Parameters:

  • inference_results – The inference results from the inference engine.

  • generated_attribute – The generated attribute configuration.

Returns:

A list of dictionaries with the processed attribute values.

synthesize(samples: list[dict], generated_attribute: GeneratedAttribute) → list[dict[str, str]][source]#

Synthesize a value for the generated attribute.

Order will be identical to the order of the samples.

Parameters:

  • samples – The samples to synthesize values for.

  • generated_attribute – The generated attribute to synthesize a value for.

Returns:

A list of dictionaries, one for each sample, with the generated attribute value added to the dictionary.

synthesize_batch(samples: list[dict], generated_attribute: GeneratedAttribute) → str[source]#

Submit a batch inference job for attribute synthesis.

Parameters:

  • samples – The samples to synthesize values for.

  • generated_attribute – The generated attribute to synthesize a value for.

Returns:

The batch ID that can be used with get_batch_status and get_batch_results.

Raises:

NotImplementedError – If the inference engine does not support batch.

synthesize_partial(samples: list[dict], generated_attribute: GeneratedAttribute, *, progress_path: str | None = None) → SynthPartialResult[source]#

Synthesize attribute values online, tolerating per-row failures.

Parameters:

  • samples – The samples to synthesize values for.

  • generated_attribute – The generated attribute to synthesize a value for.

  • progress_path – Path to a JSON file where inference progress counters are periodically written for external pollers.

Returns:

SynthPartialResult with successful outputs and failure info.

property total_cached_tokens: int#

Total cached tokens accumulated across all synthesize() calls.

property total_input_tokens: int#

Total input/prompt tokens accumulated across all synthesize() calls.

property total_output_tokens: int#

Total output/completion tokens accumulated across all synthesize() calls.

class oumi.core.synthesis.attribute_synthesizer.SynthBatchResult(successful: list[tuple[int, dict[str, str]]], failed_indices: list[int], error_messages: dict[int, str] = <factory>)[source]#

Bases: object

Result from partial batch synthesis, separating successes from failures.

error_messages: dict[int, str]#

Mapping of failed index to error message.

failed_indices: list[int]#

Indices of items that failed synthesis.

property has_failures: bool#

Return True if any items in the batch failed synthesis.

successful: list[tuple[int, dict[str, str]]]#

List of (original_index, attribute_dict) for successfully synthesized items.

class oumi.core.synthesis.attribute_synthesizer.SynthPartialResult(successful: list[tuple[int, dict[str, str]]], failures: dict[int, FailureDetail])[source]#

Bases: object

Result from partial online synthesis, separating successes from failures.

Returned by AttributeSynthesizer.synthesize_partial(). Unlike synthesize(), which raises if any row’s inference fails, this pairs each successfully synthesized attribute dict with its original sample index and reports inference and processing failures separately.

property error_messages: dict[int, str]#

Mapping of failed index to error message.

property failed_indices: list[int]#

Sorted indices of items that failed synthesis.

failures: dict[int, FailureDetail]#

Mapping of failed index to structured failure info.

property has_failures: bool#

Return True if any items failed synthesis.

successful: list[tuple[int, dict[str, str]]]#

List of (original_index, attribute_dict) for successfully synthesized items.

oumi.core.synthesis.attribute_transformation module#

class oumi.core.synthesis.attribute_transformation.AttributeTransformer(params: GeneralSynthesisParams)[source]#

Bases: object

Transforms attributes of a dataset plan to a particular format.

transform(samples: list[dict[str, Any]]) → list[dict[str, Any]][source]#

Transforms attributes of a dataset plan to a particular format.

Parameters:

  • samples – The samples to add transformed attributes to, using the values in

  • transformation. (each sample as the input to the)

Returns:

The samples with the transformed attributes added.

oumi.core.synthesis.conversation_synthesizer module#

class oumi.core.synthesis.conversation_synthesizer.ConversationSynthesizer(params: GeneralSynthesisParams, inference_config: InferenceConfig, environment_config: EnvironmentConfig | None = None)[source]#

Bases: object

Synthesizes a conversation.

Parameters:

  • params – The parameters for the conversation synthesizer.

  • inference_config – The configuration for the inference engine.

build_opening_turn_prompts(samples: list[dict], plans: list[str], multiturn_attribute: MultiTurnAttribute) → list[OpeningTurnPrompt][source]#

Parse plans and build the opening (turn 1, USER) generation prompts.

Inference-free stage after build_planner_prompts(): parses each plan into per-turn instructions and renders the opening-turn prompt.

Parameters:

  • samples – Augmented samples from build_planner_prompts (each carrying target_turns).

  • plans – Raw planner output strings, aligned 1:1 with samples.

  • multiturn_attribute – The multi-turn attribute defining conversation rules.

Returns:

One OpeningTurnPrompt per sample, in order.

build_planner_prompts(samples: list[dict], multiturn_attribute: MultiTurnAttribute) → list[PlannerPrompt][source]#

Attach grounding and build planner prompts, without inference.

Self-contained inference-free entrypoint for callers that drive the planner as a separate stage: prepares routers, attaches grounding, and renders one prompt per sample. target_turns is drawn randomly, so persist it if plans are inferred out-of-process.

Parameters:

  • samples – The samples to plan conversations for.

  • multiturn_attribute – The multi-turn attribute defining conversation rules.

Returns:

One PlannerPrompt per input sample, in order.

build_seed_conversations(samples: list[dict], opening_turns: list[str], multiturn_attribute: MultiTurnAttribute) → list[SeedConversation][source]#

Build seed conversations for out-of-process multi-turn generation.

Inference-free stage after build_opening_turn_prompts(): pairs each opening user utterance with its seed conversation and generation state.

Parameters:

  • samples – Augmented samples (carrying target_turns and parsed_turn_plans) from build_opening_turn_prompts.

  • opening_turns – The opening user utterances, aligned 1:1 with samples.

  • multiturn_attribute – The multi-turn attribute defining conversation rules.

Returns:

One SeedConversation per sample, in order.

synthesize(samples: list[dict], multiturn_attributes: MultiTurnAttribute) → list[dict[str, dict | str] | None][source]#

Synthesize a multi-turn conversation.

Order will be identical to the order of the samples.

Parameters:

  • samples – The samples to synthesize values for.

  • multiturn_attributes – The multi-turn attribute defining conversation rules.

Returns:

  • a dictionary containing the conversation and plan, or

  • None when the synthesized conversation is filtered out.

Return type:

A list aligned to the input samples. Each entry is either

property total_cached_tokens: int#

Total cached tokens accumulated across all synthesize() calls.

property total_input_tokens: int#

Total input/prompt tokens accumulated across all synthesize() calls.

property total_output_tokens: int#

Total output/completion tokens accumulated across all synthesize() calls.

class oumi.core.synthesis.conversation_synthesizer.OpeningTurnPrompt(augmented_sample: dict, conversation: Conversation)[source]#

Bases: object

An augmented sample plus its opening-turn generation prompt.

augmented_sample: dict#
conversation: Conversation#
class oumi.core.synthesis.conversation_synthesizer.PlannerPrompt(augmented_sample: dict, conversation: Conversation)[source]#

Bases: object

A planner-prompt conversation and the augmented sample it was built from.

augmented_sample: dict#
conversation: Conversation#
class oumi.core.synthesis.conversation_synthesizer.SeedConversation(conversation: Conversation, generation_state: dict)[source]#

Bases: object

A seed conversation plus the generation_state a turn driver needs.

conversation: Conversation#
generation_state: dict#

oumi.core.synthesis.data_synthesizer module#

class oumi.core.synthesis.data_synthesizer.DataSynthesizer(generated_attributes: list[GeneratedAttribute], attribute_synthesizer: AttributeSynthesizer)[source]#

Bases: object

Synthesizes data using attributes from the dataset plan.

synthesize(dataset_plan_samples: list[dict[str, Any]]) → list[dict[str, Any]][source]#

Synthesize data using attributes from the dataset plan.

If there are multiple generated attributes, each will be synthesized sequentially, with their resulting output added to the dataset plan records.

oumi.core.synthesis.dataset_ingestion module#

class oumi.core.synthesis.dataset_ingestion.DatasetPath(path: str)[source]#

Bases: object

Path to a dataset in some storage location.

get_file_extension() → str[source]#

Get the file extension.

get_path_str() → str[source]#

Get the path.

get_storage_type() → DatasetStorageType[source]#

Get the storage type.

class oumi.core.synthesis.dataset_ingestion.DatasetReader[source]#

Bases: object

Reads a dataset from some storage location.

Supports: - HuggingFace - Local files (JSONL, CSV, TSV, Parquet, JSON, XLSX) - Glob patterns

read(data_source: DatasetSource) → list[dict][source]#

Read the data from the data path.

class oumi.core.synthesis.dataset_ingestion.DatasetStorageType(value)[source]#

Bases: Enum

Storage location for a dataset (local, HuggingFace, etc.).

HF = 'hf'#

HuggingFace

LOCAL = 'local'#

Local files

OUMI = 'oumi'#

Oumi

oumi.core.synthesis.dataset_planner module#

class oumi.core.synthesis.dataset_planner.DatasetPlanner(document_reader: DocumentReader | None = None, dataset_reader: DatasetReader | None = None)[source]#

Bases: object

Planner for the dataset’s attributes for inference.

plan(synthesis_params: GeneralSynthesisParams, sample_count: int) → list[dict][source]#

Setup the dataset’s attributes for inference.

This function will create a list of dictionaries, with each dictionary representing a sample of the dataset with a particular attribute value for each attribute.

Source Types and Sampling Modes:

Example sources, Document sources, and Dataset sources support two modes:

  • Round-robin mode (default): When num_shots is None or 1, items are cycled sequentially across synthesis samples. Each item’s attributes are spread directly into the sample dict. Reference in templates using the field name (e.g., {field1}).

  • Dynamic sampling mode: When num_shots > 1, N items are randomly sampled per synthesis sample. Items are stored as a list under the source’s id. Reference in templates using bracket notation: {source_id[0].field1}.

Additionally:

  • Permutable attributes have their values sampled from a distribution.

  • Combination sampling overrides the distribution for particular attribute-value combinations.

The final list of dictionaries will be used to create a dataset.

Parameters:

  • synthesis_params – The synthesis parameters.

  • sample_count – The number of samples to plan.

Returns:

A list of dictionaries, each representing a sample of the dataset with the attribute values for each attribute.

oumi.core.synthesis.document_ingestion module#

class oumi.core.synthesis.document_ingestion.DocumentReader[source]#

Bases: object

Reader for documents.

read(document_path: str) → list[str][source]#

Read the document.

class oumi.core.synthesis.document_ingestion.DocumentSegmenter(params: DocumentSegmentationParams)[source]#

Bases: object

Segmenter for documents.

segment(document: str) → list[str][source]#

Segment the document.

segment_batch(documents: list[str]) → list[str][source]#

Segment multiple documents.

Segments will be returned as a flat list of segments.

oumi.core.synthesis.synthesis_pipeline module#

class oumi.core.synthesis.synthesis_pipeline.SynthesisPipeline(config: SynthesisConfig)[source]#

Bases: object

Pipeline for synthesizing a dataset.

synthesize() → list[dict[str, Any]][source]#

Synthesize a dataset.

oumi.core.synthesis.tool_router module#

Routing layer between LLM tool calls and environment-owned tools.

class oumi.core.synthesis.tool_router.ToolRouter(tool_specs: list[ToolDefinition], tools_by_id: dict[str, ToolParams], env_by_id: dict[str, BaseEnvironment], tool_to_env: dict[str, BaseEnvironment], env_params_by_id: dict[str, EnvironmentParams], tool_env_map: dict[str, str], on_env_built: Callable[[BaseEnvironment], None] | None)[source]#

Bases: object

Routes LLM tool calls to environment-owned tools.

Use for_sample() to obtain a per-sample clone whose envs have independent state; the synthesizer builds one clone per sample at batch entry so tool mutations don’t leak across samples.

env_by_id: dict[str, BaseEnvironment]#
env_params_by_id: dict[str, EnvironmentParams]#
for_sample() → ToolRouter[source]#

Return a router safe to use for one sample.

Envs whose requires_isolation() returns True are rebuilt via build_environment so their mutable state stays independent across samples; on_env_built re-runs on those fresh instances. Envs that don’t require isolation (e.g. DeterministicEnvironment and stateless SyntheticEnvironment) are shared with the parent router to avoid the per-sample build + inference-engine attach cost.

classmethod from_environment_config(env_config: EnvironmentConfig, on_env_built: Callable[[BaseEnvironment], None] | None = None) → ToolRouter[source]#

Build a router from an env config.

on_env_built: Callable[[BaseEnvironment], None] | None#
parse_and_validate_arguments(tool_id: str, raw_arguments: str) → dict[str, Any][source]#

Parse wire JSON args and validate against the tool’s parameters schema.

route_batch(calls: list[tuple[str, dict[str, Any]]]) → list[ToolResult][source]#

Dispatch a batch of (tool_id, args) pairs; preserves call order.

tool_env_map: dict[str, str]#
tool_specs: list[ToolDefinition]#
tool_to_env: dict[str, BaseEnvironment]#
tools_by_id: dict[str, ToolParams]#
Contents