ractogateway.finetune

RactoGateway fine-tuning module.

Provides a production-grade multimodal training pipeline for OpenAI, Google Gemini, and Anthropic Claude.

Quick start:

from ractogateway.finetune import (
    RactoDataset,
    RactoTrainingExample,
    RactoTrainingMessage,
    OpenAIFineTuner,
    GeminiFineTuner,
    AnthropicFineTuner,
)

class ractogateway.finetune.AnthropicFineTuner(api_key=None)[source]

Bases: object

Fine-tune Anthropic Claude models using the fine-tuning API.

Parameters:: api_key (str | None) – Anthropic API key. Falls back to the ANTHROPIC_API_KEY environment variable when not supplied.

Examples

End-to-end pipeline:

from ractogateway.finetune import RactoDataset, AnthropicFineTuner

ds = RactoDataset.from_pairs(
    [("Summarise this: …", "The text discusses…")],
    system="You are a concise summariser.",
)
tuner = AnthropicFineTuner()
model = tuner.run_pipeline(ds, model="claude-3-haiku-20240307")
print(model)   # "claude-3-haiku-20240307:ft:org-xxx:suffix:abc123"

upload_dataset(dataset)[source]

Upload dataset as an Anthropic training file.

Parameters:: dataset (RactoDataset) – The training examples to upload.
Return type:: str
Returns:: str – The Anthropic file ID used in create_job().

create_job(training_file, model='claude-3-haiku-20240307', *, validation_file=None, suffix=None, hyperparameters=None)[source]

Submit a fine-tuning job.

Parameters:

training_file (str) – File ID returned by upload_dataset().
model (str) – Base Claude model to fine-tune.
validation_file (str | None) – Optional validation file ID.
suffix (str | None) – Short label appended to the fine-tuned model name.
hyperparameters (dict[str, Any] | None) – Optional overrides, e.g. {"n_epochs": 3}.

Return type:

str

Returns:

str – The fine-tuning job ID.

get_status(job_id)[source]

Retrieve the current status of a fine-tuning job.

Return type:: dict[str, Any]
Returns:: dict – Keys: id, status, model, fine_tuned_model, created_at, finished_at, error.

list_jobs(limit=10)[source]

Return the most recent fine-tuning jobs (newest first).

Return type:: list[dict[str, Any]]

cancel_job(job_id)[source]

Cancel a running fine-tuning job.

Return type:: dict[str, Any]

wait_for_completion(job_id, *, poll_interval=60, verbose=True)[source]

Block until a fine-tuning job finishes.

Parameters:

job_id (str) – The job ID returned by create_job().
poll_interval (int) – Seconds between status-check API calls.
verbose (bool) – Print status lines to stdout.

Return type:

str

Returns:

str – Fine-tuned model name — pass directly to AnthropicDeveloperKit(model=...).

Raises:

RuntimeError – If the job ends in "failed" or "cancelled" state.

run_pipeline(dataset, model='claude-3-haiku-20240307', *, validation_dataset=None, suffix=None, hyperparameters=None, poll_interval=60, verbose=True)[source]

Validate → upload → train → wait in a single call.

Parameters:

dataset (RactoDataset) – Training examples.
model (str) – Base Claude model to fine-tune.
validation_dataset (RactoDataset | None) – Optional held-out validation set.
suffix (str | None) – Short label appended to the fine-tuned model name.
hyperparameters (dict[str, Any] | None) – Optional overrides, e.g. {"n_epochs": 3}.
poll_interval (int) – Seconds between status polls.
verbose (bool) – Print progress to stdout.

Return type:

str

Returns:

str – Fine-tuned model identifier — pass directly to AnthropicDeveloperKit(model=...).

Raises:

ValueError – If dataset validation fails.
RuntimeError – If the fine-tuning job fails remotely.

class ractogateway.finetune.GeminiFineTuner(api_key=None)[source]

Bases: object

Fine-tune Google Gemini models using the Generative AI tuning API.

Parameters:: api_key (str | None) – Google AI API key. Falls back to the GEMINI_API_KEY environment variable when not supplied.

Examples

End-to-end pipeline:

from ractogateway.finetune import RactoDataset, GeminiFineTuner

ds = RactoDataset.from_pairs(
    [("capital of France?", "Paris"), ("capital of Japan?", "Tokyo")],
)
tuner = GeminiFineTuner()
model_name = tuner.run_pipeline(
    ds,
    base_model="models/gemini-1.5-flash-001-tuning",
    display_name="geography-tutor",
)
print(model_name)  # "tunedModels/geography-tutor-abc123"

create_job(dataset, base_model='models/gemini-1.5-flash-001-tuning', *, display_name='', epoch_count=5, batch_size=4, learning_rate=None)[source]

Start a Gemini supervised fine-tuning job.

Parameters:

dataset (RactoDataset) – Training examples. Each example must be a single-turn text pair (text_input / output). Examples with attachments or multi-turn conversations are not supported by this adapter — use Vertex AI for those.
base_model (str) – Tuning-enabled Gemini model identifier.
display_name (str) – Human-readable label for the tuned model.
epoch_count (int) – Number of training epochs.
batch_size (int) – Training batch size.
learning_rate (float | None) – Learning rate. None uses the provider default.

Return type:

Any

Returns:

google.generativeai.types.TunedModel (operation-like object) – Pass to wait_for_completion().

Raises:

ValueError – If the dataset fails validation, or if any examples are multimodal / multi-turn (unsupported by this adapter).

get_model(tuned_model_name)[source]

Retrieve metadata for a tuned model.

Parameters:: tuned_model_name (str) – Full tuned model name, e.g. "tunedModels/my-model-abc123".
Return type:: dict[str, Any]
Returns:: dict – Keys: name, display_name, state, base_model.

list_models()[source]

List all tuned models in this project.

Return type:: list[dict[str, Any]]

delete_model(tuned_model_name)[source]

Permanently delete a tuned model from your project.

Return type:: None

wait_for_completion(operation, *, poll_interval=60, verbose=True)[source]

Block until a tuning operation finishes.

Parameters:

operation (Any) – The object returned by create_job().
poll_interval (int) – Seconds between metadata checks.
verbose (bool) – Print progress to stdout.

Return type:

str

Returns:

str – Tuned model name (e.g. "tunedModels/my-model-abc123"). Pass directly to GoogleDeveloperKit(model=...).

Raises:

RuntimeError – If the tuning job ends in a failed state.

run_pipeline(dataset, base_model='models/gemini-1.5-flash-001-tuning', *, display_name='', epoch_count=5, batch_size=4, learning_rate=None, poll_interval=60, verbose=True)[source]

Validate → create → wait in a single call.

Parameters:

dataset (RactoDataset) – Text-pair training examples.
base_model (str) – Tuning-enabled Gemini model.
display_name (str) – Human-readable label for the tuned model.
epoch_count (int) – Training epochs.
batch_size (int) – Training batch size.
learning_rate (float | None) – Learning rate override.
poll_interval (int) – Seconds between status polls.
verbose (bool) – Print progress to stdout.

Return type:

str

Returns:

str – Tuned model name — pass to GoogleDeveloperKit(model=...).

class ractogateway.finetune.OpenAIFineTuner(api_key=None, *, base_url=None)[source]

Bases: object

Fine-tune OpenAI models using the fine-tuning API.

Parameters:

api_key (str | None) – OpenAI API key. Falls back to the OPENAI_API_KEY environment variable when not supplied.
base_url (str | None) – Optional custom base URL (Azure OpenAI, proxy, etc.).

Examples

End-to-end pipeline (simplest usage):

from ractogateway.finetune import RactoDataset, OpenAIFineTuner

ds = RactoDataset.from_pairs(
    [("What is Python?", "A high-level programming language.")],
    system="You are a Python tutor.",
)
tuner = OpenAIFineTuner()
model = tuner.run_pipeline(ds, model="gpt-4o-mini-2024-07-18")
print(model)   # "ft:gpt-4o-mini-2024-07-18:org::abc123"

upload_dataset(dataset)[source]

Upload dataset as an OpenAI training file.

Parameters:: dataset (RactoDataset) – The training examples to upload.
Return type:: str
Returns:: str – The OpenAI file ID (e.g. "file-abc123").

create_job(training_file, model='gpt-4o-mini-2024-07-18', *, validation_file=None, n_epochs='auto', batch_size='auto', learning_rate_multiplier='auto', suffix=None)[source]

Submit a fine-tuning job.

Parameters:

training_file (str) – File ID returned by upload_dataset().
model (str) – Base model to fine-tune.
validation_file (str | None) – Optional validation file ID (also produced by upload_dataset()).
n_epochs (int | str) – Training epochs.
batch_size (int | str) – Per-device batch size.
learning_rate_multiplier (float | str) – Scales the default learning rate.
suffix (str | None) – Custom label appended to the fine-tuned model name.

Return type:

str

Returns:

str – The fine-tuning job ID (e.g. "ftjob-abc123").

get_status(job_id)[source]

Retrieve the current status of a fine-tuning job.

Return type:: dict[str, Any]
Returns:: dict – Keys: id, status, model, fine_tuned_model, created_at, finished_at, trained_tokens, error.

list_jobs(limit=10)[source]

Return the most recent fine-tuning jobs (newest first).

Return type:: list[dict[str, Any]]

list_events(job_id, limit=20)[source]

Return recent training log events for a job.

Return type:: list[dict[str, Any]]

cancel_job(job_id)[source]

Cancel a running fine-tuning job.

Return type:: dict[str, Any]

wait_for_completion(job_id, *, poll_interval=30, verbose=True)[source]

Block until a fine-tuning job finishes.

Parameters:

job_id (str) – The job ID returned by create_job().
poll_interval (int) – Seconds between status-check API calls.
verbose (bool) – Print status lines to stdout.

Return type:

str

Returns:

str – The fine-tuned model name ready for use in OpenAILLMKit.

Raises:

RuntimeError – If the job ends in "failed" or "cancelled" state.

run_pipeline(dataset, model='gpt-4o-mini-2024-07-18', *, validation_dataset=None, n_epochs='auto', batch_size='auto', learning_rate_multiplier='auto', suffix=None, poll_interval=30, verbose=True)[source]

Validate → upload → train → wait in a single call.

This is the recommended entry-point for most use cases.

Parameters:

dataset (RactoDataset) – Training examples.
model (str) – Base model to fine-tune.
validation_dataset (RactoDataset | None) – Optional held-out validation set (uploaded separately).
n_epochs (int | str) – Training hyperparameters. Pass "auto" to let OpenAI decide.
batch_size (int | str) – Training hyperparameters. Pass "auto" to let OpenAI decide.
learning_rate_multiplier (float | str) – Training hyperparameters. Pass "auto" to let OpenAI decide.
suffix (str | None) – Short label appended to the fine-tuned model name.
poll_interval (int) – Seconds between status polls while waiting.
verbose (bool) – Print progress to stdout.

Return type:

str

Returns:

str – Fine-tuned model identifier — pass directly to OpenAIDeveloperKit(model=...):

kit = opd.OpenAIDeveloperKit(model=fine_tuned_model)

Raises:

ValueError – If dataset validation fails.
RuntimeError – If the fine-tuning job fails remotely.

class ractogateway.finetune.RactoDataset(examples=None)[source]

Bases: object

An ordered collection of RactoTrainingExample objects.

This is the central data container for building, validating, splitting, and exporting fine-tuning datasets for any supported LLM provider.

Parameters:: examples (list[RactoTrainingExample] | None) – Initial examples. An empty dataset is created when omitted.

Examples

Build from (user, assistant) pairs:

ds = RactoDataset.from_pairs(
    [
        ("What is Python?", "Python is a high-level programming language."),
        ("What is a list?", "A list is a mutable ordered sequence."),
    ],
    system="You are a Python tutor.",
)

Add multimodal examples manually:

ds.add(
    RactoTrainingExample.from_pair(
        user="Describe this image.",
        assistant="The image shows a flowchart with three decision nodes.",
        user_attachments=[RactoFile.from_path("diagram.png")],
    )
)

Export to JSONL for fine-tuning:

train_ds, val_ds = ds.split(0.8, seed=42)
train_ds.export_jsonl("train.jsonl", provider="openai")
val_ds.export_jsonl("val.jsonl", provider="openai")

add(example)[source]

Append a single training example.

Return type:: None

extend(examples)[source]

Append multiple training examples at once.

Return type:: None

classmethod from_pairs(pairs, *, system='')[source]

Build a text-only dataset from (user, assistant) pairs.

Parameters:

pairs (list[tuple[str, str]]) – Each tuple is (user_message, expected_assistant_response).
system (str) – Optional system prompt applied uniformly to every example.

Return type:

RactoDataset

classmethod from_jsonl(path, provider='openai')[source]

Load a JSONL dataset previously exported for provider.

Supports text-only OpenAI, Anthropic, and Gemini formats.

Parameters:

path (str | Path) – Path to the .jsonl file.
provider (str) – One of "openai", "anthropic", "gemini".

Return type:

RactoDataset

shuffle(seed=None)[source]

Return a new dataset with examples in random order.

Parameters:: seed (int | None) – Optional random seed for reproducibility.
Return type:: RactoDataset

split(train_ratio=0.8, *, seed=None)[source]

Split into train and validation datasets.

Parameters:

train_ratio (float) – Fraction of examples for the training split. Must be between 0 and 1 (exclusive).
seed (int | None) – Optional random seed for reproducible shuffling.

Return type:

tuple[RactoDataset, RactoDataset]

Returns:

tuple[RactoDataset, RactoDataset] – (train_dataset, validation_dataset)

validate(provider='openai')[source]

Check examples for common formatting errors.

Parameters:: provider (str) – Provider to validate against ("openai", "anthropic", or "gemini").
Return type:: list[str]
Returns:: list[str] – A list of human-readable error strings. An empty list means the dataset is ready to use.

to_jsonl_string(provider='openai')[source]

Serialize all examples to a JSONL string for provider.

Parameters:: provider (str) – One of "openai" / "generic", "anthropic", "gemini".
Return type:: str

export_jsonl(path, provider='openai', *, overwrite=False)[source]

Write the dataset to a .jsonl file on disk.

Parameters:

path (str | Path) – Destination file path.
provider (str) – One of "openai", "anthropic", "gemini".
overwrite (bool) – When False (default), raise FileExistsError if the file already exists.

Return type:

Path

Returns:

Path – The resolved path of the written file.

summary()[source]

Return brief statistics about the dataset.

Return type:: dict[str, Any]
Returns:: dict – Keys: examples, total_messages, avg_turns_per_example, multimodal_examples.

class ractogateway.finetune.RactoTrainingExample(messages)[source]

Bases: object

A complete conversation used as one training record.

Parameters:

messages (list[RactoTrainingMessage]) –

Ordered turns. Typical shapes:

Single-turn : [user, assistant]
With system : [system, user, assistant]
Multi-turn : [system, user, assistant, user, assistant, …]

Examples

>>> ex = RactoTrainingExample.from_pair(
...     user="What is 2 + 2?",
...     assistant="4",
...     system="You are a maths tutor.",
... )

>>> # Multimodal example (image + question)
>>> ex = RactoTrainingExample.from_pair(
...     user="Describe this chart.",
...     assistant="The chart shows monthly revenue for Q4 2024.",
...     user_attachments=[RactoFile.from_path("chart.png")],
... )

classmethod from_pair(user, assistant, *, system='', user_attachments=None)[source]

Create a single-turn (prompt → completion) training example.

Parameters:

user (str) – The user prompt.
assistant (str) – The desired model response.
system (str) – Optional system prompt prepended to the conversation.
user_attachments (list[RactoFile] | None) – Images or other files attached to the user turn.

Return type:

RactoTrainingExample

classmethod from_conversation(turns)[source]

Build from a list of (role, content) tuples.

Parameters:: turns (list[tuple[Literal['system', 'user', 'assistant'], str]]) – E.g. [("system", "…"), ("user", "…"), ("assistant", "…")]
Return type:: RactoTrainingExample

to_openai_dict()[source]

Serialize to OpenAI fine-tuning JSONL record.

Output format:

{"messages": [{"role": "system", "content": "…"}, …]}

Return type:: dict[str, Any]

to_anthropic_dict()[source]

Serialize to Anthropic fine-tuning JSONL record.

Output format:

{"system": "…", "messages": [{"role": "user", …}, …]}

The system key is only present when a system message exists.

Return type:: dict[str, Any]

to_gemini_dict()[source]

Serialize to Gemini tuning record.

For text-only single-turn examples (most common) the output is:

{"text_input": "…", "output": "…"}

For multimodal or multi-turn examples the Vertex AI contents format is used:

{"contents": [{"role": "user", "parts": […]}, …]}

Return type:: dict[str, Any]

class ractogateway.finetune.RactoTrainingMessage(role, content, attachments=<factory>)[source]

Bases: object

One conversational turn inside a training example.

Parameters:

role (Literal['system', 'user', 'assistant']) – Speaker role.
content (str) – Text content of the message.
attachments (list[RactoFile]) – Optional images / PDFs for multimodal training examples. Use RactoFile.from_path() or RactoFile.from_bytes().

role: Literal['system', 'user', 'assistant']

content: str

attachments: list[RactoFile]

to_openai()[source]

Return an OpenAI-compatible message dict.

Text-only messages produce {"role": ..., "content": str}. Messages with attachments produce a content-block list: {"role": ..., "content": [image_url_block, ..., text_block]}.

Return type:: dict[str, Any]

to_anthropic()[source]

Return an Anthropic-compatible message dict.

System messages should be lifted to the top-level system field — RactoTrainingExample.to_anthropic_dict() handles this automatically.

Return type:: dict[str, Any]

to_gemini_parts()[source]

Return a list of Gemini content parts (text + inline_data).

Return type:: list[dict[str, Any]]