Fine-Tuning via Provider APIs

# Format training data as JSONL for the OpenAI fine-tuning API
# Each line contains a messages array with system, user, and assistant turns
import json
from typing import List, Dict

def prepare_openai_training_file(
 examples: List[Dict],
 output_path: str,
 validate: bool = True
) -> dict:
 """Prepare and validate a JSONL file for OpenAI fine-tuning."""
 stats = {"total": 0, "valid": 0, "errors": [], "token_estimates": []}

 with open(output_path, "w") as f:
 for i, example in enumerate(examples):
 stats["total"] += 1
 messages = example.get("messages", [])

 if validate:
 # Validate message structure
 if not messages:
 stats["errors"].append(f"Example {i}: empty messages")
 continue

 roles = [m["role"] for m in messages]

 # Must end with assistant message
 if roles[-1] != "assistant":
 stats["errors"].append(
 f"Example {i}: last message must be 'assistant'"
 )
 continue

 # Check for valid roles
 valid_roles = {"system", "user", "assistant"}
 invalid = set(roles) - valid_roles
 if invalid:
 stats["errors"].append(
 f"Example {i}: invalid roles {invalid}"
 )
 continue

 # Rough token estimate (4 chars per token)
 total_chars = sum(len(m["content"]) for m in messages)
 estimated_tokens = total_chars // 4
 stats["token_estimates"].append(estimated_tokens)

 f.write(json.dumps(example) + "\n")
 stats["valid"] += 1

 if stats["token_estimates"]:
 import numpy as np
 tokens = np.array(stats["token_estimates"])
 stats["token_summary"] = {
 "mean": int(tokens.mean()),
 "median": int(np.median(tokens)),
 "p95": int(np.percentile(tokens, 95)),
 "total_training_tokens": int(tokens.sum()),
 }

 print(f"Prepared {stats['valid']}/{stats['total']} examples")
 if stats["errors"]:
 print(f"Errors: {len(stats['errors'])}")
 for err in stats["errors"][:5]:
 print(f" {err}")

 return stats

# Example usage
examples = [
 {
 "messages": [
 {"role": "system", "content": "You are a helpful customer support agent."},
 {"role": "user", "content": "How do I reset my password?"},
 {"role": "assistant", "content": "To reset your password, go to Settings, "
 "then Security, and click 'Reset Password'. You will receive an email "
 "with a reset link within 5 minutes."}
 ]
 },
 # ... more examples
]

stats = prepare_openai_training_file(examples, "train.jsonl")

# Format training data as JSONL for the OpenAI fine-tuning API
# Each line contains a messages array with system, user, and assistant turns
from openai import OpenAI
import time

client = OpenAI() # Uses OPENAI_API_KEY env var

# Step 1: Upload training file
with open("train.jsonl", "rb") as f:
 training_file = client.files.create(file=f, purpose="fine-tune")
print(f"Uploaded file: {training_file.id}")

# Step 2: (Optional) Upload validation file
with open("val.jsonl", "rb") as f:
 validation_file = client.files.create(file=f, purpose="fine-tune")

# Step 3: Create fine-tuning job
job = client.fine_tuning.jobs.create(
 training_file=training_file.id,
 validation_file=validation_file.id,
 model="gpt-4o-mini-2024-07-18",
 hyperparameters={
 "n_epochs": 3, # Number of epochs
 "learning_rate_multiplier": 1.8, # Relative to default
 "batch_size": 4, # Auto if not specified
 },
 suffix="customer-support-v1", # Custom model name suffix
)
print(f"Job created: {job.id}")
print(f"Status: {job.status}")

# Step 4: Monitor training progress
def monitor_fine_tuning(job_id: str, poll_interval: int = 60):
 """Poll a fine-tuning job until completion."""
 while True:
 job = client.fine_tuning.jobs.retrieve(job_id)
 print(f"Status: {job.status}")

 # Check for events (training metrics)
 events = client.fine_tuning.jobs.list_events(
 fine_tuning_job_id=job_id, limit=5
 )
 for event in events.data:
 print(f" [{event.created_at}] {event.message}")

 if job.status in ("succeeded", "failed", "cancelled"):
 break

 time.sleep(poll_interval)

 if job.status == "succeeded":
 print(f"\nFine-tuned model: {job.fine_tuned_model}")
 return job.fine_tuned_model
 else:
 print(f"\nJob {job.status}: {job.error}")
 return None

model_name = monitor_fine_tuning(job.id)

# Step 5: Use the fine-tuned model (identical to standard API calls)
response = client.chat.completions.create(
 model="ft:gpt-4o-mini-2024-07-18:my-org:customer-support-v1:9abc123",
 messages=[
 {"role": "system", "content": "You are a helpful customer support agent."},
 {"role": "user", "content": "I can't find my order confirmation email."},
 ],
 temperature=0.7,
 max_tokens=200,
)

# Extract the generated message from the API response
print(response.choices[0].message.content)

# Upload training data to GCS and launch a Vertex AI SFT job
# Vertex AI fine-tunes Gemini models with LoRA adapter configuration
import vertexai
from vertexai.tuning import sft as vertex_sft
from google.cloud import storage

# Initialize Vertex AI
vertexai.init(project="my-project-id", location="us-central1")

# Step 1: Upload training data to GCS
# Vertex AI expects data in GCS (Google Cloud Storage)
# Format: JSONL with same messages structure as OpenAI

def upload_to_gcs(local_path: str, bucket_name: str, blob_name: str) -> str:
 """Upload training data to Google Cloud Storage."""
 client = storage.Client()
 bucket = client.bucket(bucket_name)
 blob = bucket.blob(blob_name)
 blob.upload_from_filename(local_path)
 gcs_uri = f"gs://{bucket_name}/{blob_name}"
 print(f"Uploaded to {gcs_uri}")
 return gcs_uri

train_uri = upload_to_gcs(
 "train.jsonl",
 "my-training-bucket",
 "fine-tuning/medical-qa/train.jsonl"
)
val_uri = upload_to_gcs(
 "val.jsonl",
 "my-training-bucket",
 "fine-tuning/medical-qa/val.jsonl"
)

# Step 2: Create supervised fine-tuning job
sft_tuning_job = vertex_sft.train(
 source_model="gemini-1.5-flash-002",
 train_dataset=train_uri,
 validation_dataset=val_uri,
 epochs=3,
 adapter_size=4, # LoRA rank (1, 4, 8, or 16)
 learning_rate_multiplier=1.0,
 tuned_model_display_name="medical-qa-gemini-v1",
)

# Step 3: Monitor (blocking call)
print(f"Job resource: {sft_tuning_job.resource_name}")

# Poll for completion
while not sft_tuning_job.has_ended:
 time.sleep(60)
 sft_tuning_job.refresh()
 print(f"State: {sft_tuning_job.state}")

# Step 4: Get the tuned model endpoint
tuned_model = sft_tuning_job.tuned_model_endpoint_name
print(f"Tuned model endpoint: {tuned_model}")

# Use the fine-tuned Gemini model for inference via Vertex AI
# The tuned model endpoint is returned by the completed SFT job
from vertexai.generative_models import GenerativeModel

# Load the fine-tuned model
model = GenerativeModel(
 model_name=tuned_model, # Endpoint from training
)

# Generate responses
response = model.generate_content(
 "Patient presents with recurring headaches and blurred vision. "
 "Suggest differential diagnoses.",
 generation_config={
 "temperature": 0.3,
 "max_output_tokens": 500,
 }
)

print(response.text)

# Fine-tune a Claude model via the Anthropic API
# Anthropic uses a different message format: system is a top-level field
import anthropic
import json

# Step 1: Prepare training data in Anthropic's format
training_examples = [
 {
 "system": "You are a medical triage assistant.",
 "messages": [
 {"role": "user", "content": "Patient reports chest pain radiating to left arm."},
 {"role": "assistant", "content": "Priority: URGENT. Recommend immediate ECG and "
 "troponin levels. Differential includes acute coronary syndrome, "
 "musculoskeletal pain, and anxiety. Escalate to attending physician."}
 ]
 },
 # ... more examples in the same format
]

# Write JSONL file
with open("anthropic_train.jsonl", "w") as f:
 for ex in training_examples:
 f.write(json.dumps(ex) + "\n")

# Step 2: Create the fine-tuning job via the Anthropic SDK
client = anthropic.Anthropic() # Uses ANTHROPIC_API_KEY env var

# Upload the training file
with open("anthropic_train.jsonl", "rb") as f:
 training_file = client.files.create(
 file=f,
 purpose="fine-tune"
 )

# Create the fine-tuning job
job = client.fine_tuning.jobs.create(
 model="claude-3-5-haiku-20241022",
 training_file=training_file.id,
 hyperparameters={
 "n_epochs": 4,
 "learning_rate": 1e-5,
 "batch_size": 8,
 },
)
print(f"Fine-tuning job created: {job.id}")
print(f"Status: {job.status}")

# Step 3: Poll for completion
import time
while job.status not in ("succeeded", "failed", "cancelled"):
 time.sleep(60)
 job = client.fine_tuning.jobs.retrieve(job.id)
 print(f"Status: {job.status}")

# Step 4: Use the fine-tuned model
if job.status == "succeeded":
 response = client.messages.create(
 model=job.fine_tuned_model,
 max_tokens=300,
 system="You are a medical triage assistant.",
 messages=[
 {"role": "user", "content": "Patient has sudden onset severe headache."}
 ],
 )
 print(response.content[0].text)

# Fine-tune an open-weight model on Together AI with full LoRA control
# Together supports Llama, Mistral, Qwen, and custom model uploads
import together
import json

# Step 1: Upload training data (same JSONL/ChatML format)
train_file = together.Files.upload(file="train.jsonl")
print(f"Uploaded: {train_file.id}")

# Step 2: Create a fine-tuning job with full hyperparameter control
job = together.Fine_tuning.create(
 training_file=train_file.id,
 model="meta-llama/Meta-Llama-3.1-8B-Instruct",
 n_epochs=3,
 learning_rate=2e-5,
 batch_size=8,
 lora=True,
 lora_r=16, # LoRA rank
 lora_alpha=32, # LoRA scaling factor
 lora_dropout=0.05,
 warmup_ratio=0.1,
 suffix="customer-support-v1",
)
print(f"Job: {job.id}, Status: {job.status}")

# Step 3: Poll for completion
import time
while job.status not in ("completed", "failed", "cancelled"):
 time.sleep(30)
 job = together.Fine_tuning.retrieve(job.id)
 print(f"Status: {job.status}")

# Step 4: Inference via Together's serverless endpoint
if job.status == "completed":
 response = together.Complete.create(
 model=job.output_name,
 prompt="<|user|>\nHow do I reset my password?\n<|assistant|>\n",
 max_tokens=200,
 temperature=0.7,
 )
 print(response.choices[0].text)

# Fine-tune a LoRA adapter on Fireworks AI with composable deployment
# Fireworks supports adapter composition: mix adapters at serving time
import fireworks.client as fw

# Configure the client
fw.client.api_key = "your-api-key"

# Step 1: Upload dataset
dataset = fw.Datasets.create(
 file=open("train.jsonl", "rb"),
 name="support-training-data",
)

# Step 2: Create fine-tuning job
job = fw.FineTuning.create(
 model="accounts/fireworks/models/llama-v3p1-8b-instruct",
 dataset=dataset.id,
 epochs=3,
 learning_rate=1e-4,
 lora_rank=16,
 max_seq_length=2048,
 output_model_name="support-adapter-v1",
)

# Step 3: Deploy the adapter (composable with base model)
# Fireworks serves the adapter on top of the base model at inference time
deployment = fw.Deployments.create(
 model=job.output_model,
 min_replicas=1,
 max_replicas=4, # auto-scales based on traffic
)

# Step 4: Inference (adapter applied at serving time)
response = fw.ChatCompletion.create(
 model=f"accounts/my-org/models/support-adapter-v1",
 messages=[
 {"role": "user", "content": "I can't find my order confirmation."}
 ],
 temperature=0.7,
)
print(response.choices[0].message.content)

# Compare fine-tuning costs across API providers and self-hosted options
# Estimates training cost, monthly inference cost, and annual totals
from dataclasses import dataclass

@dataclass
class FineTuningCostEstimate:
 """Compare fine-tuning costs across providers."""

 # Dataset parameters
 num_examples: int = 10_000
 avg_tokens_per_example: int = 500
 num_epochs: int = 3

 # Inference parameters (monthly)
 monthly_requests: int = 100_000
 avg_input_tokens: int = 300
 avg_output_tokens: int = 150

 def openai_cost(self, model: str = "gpt-4o-mini") -> dict:
 """Estimate OpenAI fine-tuning + inference costs."""
 # Training pricing (per 1M tokens)
 training_prices = {
 "gpt-4o-mini": {"train": 3.00},
 "gpt-4o": {"train": 25.00},
 }
 # Inference pricing (per 1M tokens, fine-tuned models)
 inference_prices = {
 "gpt-4o-mini": {"input": 0.30, "output": 1.20},
 "gpt-4o": {"input": 3.75, "output": 15.00},
 }

 train_price = training_prices[model]
 infer_price = inference_prices[model]

 # Training cost
 total_training_tokens = (
 self.num_examples * self.avg_tokens_per_example * self.num_epochs
 )
 training_cost = (total_training_tokens / 1_000_000) * train_price["train"]

 # Monthly inference cost
 monthly_input_tokens = self.monthly_requests * self.avg_input_tokens
 monthly_output_tokens = self.monthly_requests * self.avg_output_tokens
 monthly_inference = (
 (monthly_input_tokens / 1_000_000) * infer_price["input"] +
 (monthly_output_tokens / 1_000_000) * infer_price["output"]
 )

 return {
 "provider": f"OpenAI ({model})",
 "training_cost": f"${training_cost:.2f}",
 "monthly_inference": f"${monthly_{inference}:.2f}",
 "annual_{total}": f"${training_cost + monthly_inference * 12:.2f}",
 "training_tokens": f"{total_training_tokens:,}",
 }

 def self_hosted_cost(self, gpu_hourly: float = 2.50) -> dict:
 """Estimate self-hosted fine-tuning costs."""
 # Rough estimate: ~10K tokens/second on A100
 total_training_tokens = (
 self.num_examples * self.avg_tokens_per_example * self.num_epochs
 )
 training_hours = total_training_tokens / (10_000 * 3600)
 training_cost = training_hours * gpu_hourly

 # Serving: dedicated GPU instance
 serving_monthly = gpu_hourly * 24 * 30 # Always-on

 return {
 "provider": "Self-hosted (A100)",
 "training_cost": f"${training_{cost}:.2f}",
 "monthly_{inference}": f"${serving_monthly:.2f}",
 "annual_total": f"${training_cost + serving_monthly * 12:.2f}",
 "training_tokens": f"{total_training_tokens:,}",
 }

# Compare costs
estimator = FineTuningCostEstimate(
 num_examples=10_000,
 monthly_requests=100_000,
)

for result in [
 estimator.openai_cost("gpt-4o-mini"),
 estimator.openai_cost("gpt-4o"),
 estimator.self_hosted_cost(),
]:
 print(f"\n{result['provider']}:")
 for k, v in result.items():
 if k != "provider":
 print(f" {k}: {v}")