Building Custom Models with Gemini Tuning API
The Gemini Model Tuning API enables you to create domain-specialized versions of Gemini by training on your own data. This goes beyond prompt engineering—you're adapting the underlying model to understand your specific terminology, patterns, and preferences. This guide covers everything from data preparation through deployment and evaluation.
What is Model Tuning?
Model tuning (also called fine-tuning) adapts a pre-trained base model to your specific use case by training it on your domain data. The result is a custom model that operates with your vocabulary, context, and business rules built in.
Key Benefits
- Domain Specialization: Train on medical, legal, technical, or proprietary terminology
- Brand Voice: Embed your company's tone, style, and communication preferences
- Improved Performance: Better accuracy on your specific tasks compared to the base model
- Cost Efficiency: Achieve high performance with smaller training datasets than full training
Preparing Training Data
The quality and relevance of your training data directly determine your tuned model's performance.
Data Requirements
- Minimum Size: 100 examples (recommended: 1,000+)
- Format: JSONL (JSON Lines) with conversational message pairs
- Balance: Adequate representation of all important use cases
- Quality: Clean, grammatically correct, domain-accurate examples
Training Data Structure
// training_data.jsonl
{"messages": [{"role": "user", "content": "What's the difference between REST and GraphQL?"}, {"role": "assistant", "content": "REST uses resources and HTTP methods, while GraphQL uses a single endpoint with precise query specification..."}]}
{"messages": [{"role": "user", "content": "How do I handle authentication?"}, {"role": "assistant", "content": "You can use JWT tokens, OAuth 2.0, or API keys depending on your requirements..."}]}
{"messages": [{"role": "user", "content": "Explain pagination in APIs"}, {"role": "assistant", "content": "Pagination breaks large result sets into smaller chunks using parameters like limit and offset..."}]}Dataset Preparation Utility
import json
from typing import List, Dict, Tuple
class TrainingDatasetManager:
"""Prepare and validate model tuning datasets."""
@staticmethod
def create_example(
user_message: str,
assistant_response: str
) -> Dict:
"""Create a single training example."""
return {
"messages": [
{"role": "user", "content": user_message},
{"role": "assistant", "content": assistant_response}
]
}
@staticmethod
def validate_dataset(examples: List[Dict]) -> Tuple[bool, List[str]]:
"""Validate dataset structure and content."""
errors = []
if len(examples) < 100:
errors.append(
f"Dataset too small ({len(examples)} examples). "
f"Minimum 100 examples recommended."
)
for idx, example in enumerate(examples):
if "messages" not in example:
errors.append(f"Example {idx}: missing 'messages' key")
continue
messages = example["messages"]
if len(messages) != 2:
errors.append(f"Example {idx}: expected 2 messages, got {len(messages)}")
continue
for msg_idx, msg in enumerate(messages):
if "role" not in msg:
errors.append(
f"Example {idx}, message {msg_idx}: missing 'role'"
)
if "content" not in msg:
errors.append(
f"Example {idx}, message {msg_idx}: missing 'content'"
)
if not msg.get("content", "").strip():
errors.append(
f"Example {idx}, message {msg_idx}: content is empty"
)
return len(errors) == 0, errors
@staticmethod
def save_jsonl(examples: List[Dict], filepath: str) -> None:
"""Save dataset in JSONL format."""
with open(filepath, 'w', encoding='utf-8') as f:
for example in examples:
f.write(json.dumps(example) + '\n')
print(f"Saved {len(examples)} examples to {filepath}")
@staticmethod
def load_jsonl(filepath: str) -> List[Dict]:
"""Load dataset from JSONL file."""
examples = []
with open(filepath, 'r', encoding='utf-8') as f:
for line in f:
if line.strip():
examples.append(json.loads(line))
return examples
@staticmethod
def calculate_statistics(examples: List[Dict]) -> Dict:
"""Calculate dataset statistics."""
total_tokens = 0
message_counts = []
for example in examples:
example_tokens = 0
for msg in example.get("messages", []):
tokens = len(msg.get("content", "").split())
example_tokens += tokens
total_tokens += example_tokens
message_counts.append(example_tokens)
return {
"num_examples": len(examples),
"total_tokens": total_tokens,
"avg_tokens_per_example": total_tokens / len(examples) if examples else 0,
"min_tokens": min(message_counts) if message_counts else 0,
"max_tokens": max(message_counts) if message_counts else 0,
}
# Usage example
training_examples = [
TrainingDatasetManager.create_example(
"What's Gemini API?",
"Gemini API is Google's advanced AI API supporting text, images, and video processing."
),
TrainingDatasetManager.create_example(
"How do I get started?",
"Get an API key from Google Cloud Console, install the SDK, and make your first request."
),
]
# Validate
valid, errors = TrainingDatasetManager.validate_dataset(training_examples)
if not valid:
for error in errors:
print(f"Error: {error}")
# Save
TrainingDatasetManager.save_jsonl(training_examples, "training.jsonl")
# Statistics
stats = TrainingDatasetManager.calculate_statistics(training_examples)
print(f"Dataset: {stats['num_examples']} examples, {stats['total_tokens']} total tokens")Creating a Tuning Job
Initiate Model Tuning
import anthropic
import time
class TuningJobManager:
"""Manage model tuning jobs."""
def __init__(self, api_key: str = None):
self.client = anthropic.Anthropic(api_key=api_key)
def create_job(
self,
training_data_path: str,
base_model: str = "claude-3-5-sonnet-20241022",
hyperparameters: Dict = None
) -> Dict:
"""Create a new tuning job."""
# Upload training data
with open(training_data_path, 'rb') as f:
file_response = self.client.beta.files.upload(
file=(
training_data_path.split('/')[-1],
f,
'application/json'
),
)
training_file_id = file_response.id
print(f"Training file uploaded: {training_file_id}")
# Create tuning job
hyperparams = hyperparameters or {
"batch_size": 32,
"learning_rate_multiplier": 1.0,
"n_epochs": 2
}
job = self.client.beta.fine_tuning.jobs.create(
model=base_model,
training_file=training_file_id,
hyperparameters=hyperparams
)
return {
"job_id": job.id,
"status": job.status,
"model": base_model,
"training_file": training_file_id,
"created_at": job.created_at if hasattr(job, 'created_at') else None
}
def get_job_status(self, job_id: str) -> Dict:
"""Get current job status."""
job = self.client.beta.fine_tuning.jobs.retrieve(job_id)
return {
"job_id": job.id,
"status": job.status,
"created_at": job.created_at if hasattr(job, 'created_at') else None,
"fine_tuned_model": (
job.result.fine_tuned_model
if hasattr(job, 'result') and job.result
else None
),
"training_errors": (
job.errors if hasattr(job, 'errors') else []
)
}
def wait_for_completion(
self,
job_id: str,
max_wait: int = 3600,
poll_interval: int = 30
) -> Dict:
"""Poll until job completes."""
start_time = time.time()
while time.time() - start_time < max_wait:
status = self.get_job_status(job_id)
print(f"Job {job_id}: {status['status']}")
if status['status'] in ['succeeded', 'failed', 'cancelled']:
return status
time.sleep(poll_interval)
raise TimeoutError(f"Tuning job {job_id} did not complete within {max_wait}s")
# Usage
manager = TuningJobManager(api_key="YOUR_API_KEY")
# Create job
job = manager.create_job(
training_data_path="training.jsonl",
base_model="claude-3-5-sonnet-20241022",
hyperparameters={
"batch_size": 16,
"learning_rate_multiplier": 2.0,
"n_epochs": 3
}
)
print(f"Created job: {job['job_id']}")
# Wait for completion
final_status = manager.wait_for_completion(job['job_id'])
print(f"Job completed with status: {final_status['status']}")
if final_status['fine_tuned_model']:
print(f"Custom model: {final_status['fine_tuned_model']}")Using Your Tuned Model
Inference with Custom Model
def generate_with_tuned_model(
model_id: str,
prompt: str,
max_tokens: int = 1024
) -> str:
"""Generate text using your tuned model."""
client = anthropic.Anthropic(api_key="YOUR_API_KEY")
response = client.messages.create(
model=model_id,
max_tokens=max_tokens,
messages=[
{"role": "user", "content": prompt}
]
)
return response.content[0].text
# Use your custom model
custom_model = "ft_claude-3-5-sonnet-20241022_custom_abc123"
response = generate_with_tuned_model(
custom_model,
"Explain API authentication in our domain context"
)
print("Custom model response:")
print(response)Evaluating Your Tuned Model
Performance Comparison
from typing import List, Tuple
class ModelEvaluator:
"""Evaluate tuned model performance."""
def __init__(self, api_key: str = None):
self.client = anthropic.Anthropic(api_key=api_key)
def compare_models(
self,
tuned_model_id: str,
base_model_id: str,
test_cases: List[Tuple[str, str]]
) -> Dict:
"""Compare tuned vs base model on test cases."""
tuned_scores = 0
base_scores = 0
for user_input, expected_output in test_cases:
# Test tuned model
tuned_response = self.client.messages.create(
model=tuned_model_id,
max_tokens=512,
messages=[{"role": "user", "content": user_input}]
)
# Test base model
base_response = self.client.messages.create(
model=base_model_id,
max_tokens=512,
messages=[{"role": "user", "content": user_input}]
)
tuned_text = tuned_response.content[0].text.lower()
base_text = base_response.content[0].text.lower()
expected_lower = expected_output.lower()
# Simple relevance check
if expected_lower in tuned_text:
tuned_scores += 1
if expected_lower in base_text:
base_scores += 1
total = len(test_cases)
return {
"tuned_accuracy": (tuned_scores / total * 100) if total > 0 else 0,
"base_accuracy": (base_scores / total * 100) if total > 0 else 0,
"improvement": (
((tuned_scores - base_scores) / total * 100)
if total > 0 else 0
),
"test_count": total,
"tuned_correct": tuned_scores,
"base_correct": base_scores
}
# Evaluate
evaluator = ModelEvaluator(api_key="YOUR_API_KEY")
test_data = [
("What's REST?", "HTTP methods"),
("GraphQL explained", "query language"),
("Authentication methods", "JWT"),
]
results = evaluator.compare_models(
tuned_model_id="ft_claude-3-5-sonnet-20241022_custom_abc123",
base_model_id="claude-3-5-sonnet-20241022",
test_cases=test_data
)
print(f"Tuned model accuracy: {results['tuned_accuracy']:.1f}%")
print(f"Base model accuracy: {results['base_accuracy']:.1f}%")
print(f"Improvement: +{results['improvement']:.1f}%")Best Practices
Iterative Improvement
def improve_model_iteratively(
initial_data_path: str,
base_model: str,
evaluation_set: List[Tuple[str, str]],
target_accuracy: float = 0.85
):
"""Iteratively improve model through multiple tuning rounds."""
manager = TuningJobManager()
evaluator = ModelEvaluator()
round_num = 1
while round_num <= 5: # Max 5 iterations
print(f"\n--- Iteration {round_num} ---")
# Create tuning job
job = manager.create_job(
initial_data_path,
base_model=base_model,
hyperparameters={
"learning_rate_multiplier": 1.0 + (round_num * 0.5),
"n_epochs": round_num
}
)
# Wait for completion
final_job = manager.wait_for_completion(job['job_id'])
if final_job['fine_tuned_model']:
# Evaluate
results = evaluator.compare_models(
tuned_model_id=final_job['fine_tuned_model'],
base_model_id=base_model,
test_cases=evaluation_set
)
print(f"Accuracy: {results['tuned_accuracy']:.1f}%")
if results['tuned_accuracy'] >= target_accuracy * 100:
print(f"Target accuracy reached!")
return final_job['fine_tuned_model']
round_num += 1
return final_job['fine_tuned_model']Maintaining several apps and the Dolice Labs sites as an indie developer, I've found tuning lives or dies on data hygiene, not technique. I validate on a few dozen examples first and scale the dataset only once the evals hold steady — and I always track cases that regressed, because a higher average can hide quiet backsliding on specific inputs.
Looking back
Model tuning transforms Gemini from a general-purpose model into a specialized system tailored to your domain. By carefully preparing training data, monitoring job progress, and iteratively evaluating results, you can create custom models that outperform base models on your specific tasks.
For advanced workflows, explore the function calling guide to add tool use to your tuned models, or review streaming patterns for real-time tuned model interactions.