Optimization Run Tracking¶

This notebook covers the OptimizationTracker and OptimizationRun classes for persisting and analyzing optimization experiments. Learn how to:

1. Create and configure an OptimizationTracker
2. Log optimization runs
3. Query and compare runs
4. Export optimization history
5. Analyze improvement trends

---

Setup¶

In [ ]:

import sys
import tempfile
from datetime import datetime, timedelta
from pathlib import Path

sys.path.insert(0, str(Path.cwd().parent.parent.parent / "src"))

from karenina.integrations.gepa import (
    OptimizationRun,
    OptimizationTarget,
    OptimizationTracker,
)

# Create a temporary directory for the database
temp_dir = Path(tempfile.mkdtemp(prefix="gepa_tracking_"))
print(f"Using temp directory: {temp_dir}")

import sys import tempfile from datetime import datetime, timedelta from pathlib import Path sys.path.insert(0, str(Path.cwd().parent.parent.parent / "src")) from karenina.integrations.gepa import ( OptimizationRun, OptimizationTarget, OptimizationTracker, ) # Create a temporary directory for the database temp_dir = Path(tempfile.mkdtemp(prefix="gepa_tracking_")) print(f"Using temp directory: {temp_dir}")

---

Creating an OptimizationTracker¶

The tracker uses SQLite for persistent storage of optimization runs.

In [ ]:

# Create a tracker
db_path = temp_dir / "optimization_history.db"
tracker = OptimizationTracker(db_path)

print("Tracker created")
print(f"  Database path: {tracker.storage_path}")
print(f"  Database exists: {tracker.storage_path.exists()}")

# Create a tracker db_path = temp_dir / "optimization_history.db" tracker = OptimizationTracker(db_path) print("Tracker created") print(f" Database path: {tracker.storage_path}") print(f" Database exists: {tracker.storage_path.exists()}")

Production Usage¶

For production, use a persistent location:

In [ ]:

# Recommended: Use ~/.karenina for persistent storage
print("""
# Production tracker
tracker = OptimizationTracker("~/.karenina/optimization_history.db")

# Or project-specific
tracker = OptimizationTracker("./experiments/gepa_runs.db")
""")

# Recommended: Use ~/.karenina for persistent storage print(""" # Production tracker tracker = OptimizationTracker("~/.karenina/optimization_history.db") # Or project-specific tracker = OptimizationTracker("./experiments/gepa_runs.db") """)

---

OptimizationRun: Run Records¶

Each optimization run is recorded as an OptimizationRun object.

In [ ]:

# Create a sample optimization run
run1 = OptimizationRun(
    benchmark_name="AIME 2025",
    # What was optimized
    targets=[OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value],
    # Initial prompts
    seed_prompts={"answering_system_prompt": "You are a helpful math assistant."},
    # Optimized prompts (result)
    optimized_prompts={
        "answering_system_prompt": """You are an expert competition mathematician.
AIME answers are integers 0-999. Show your reasoning."""
    },
    # Scores
    train_score=0.75,
    val_score=0.70,
    test_score=0.68,
    improvement=0.166,  # 16.6% improvement
    # GEPA parameters
    reflection_model="anthropic/claude-haiku-4-5",
    metric_calls=75,
    # Trajectory info
    best_generation=8,
    total_generations=10,
)

print("OptimizationRun created:")
print(f"  Run ID: {run1.run_id}")
print(f"  Benchmark: {run1.benchmark_name}")
print(f"  Val Score: {run1.val_score:.2%}")
print(f"  Improvement: {run1.improvement:.2%}")

# Create a sample optimization run run1 = OptimizationRun( benchmark_name="AIME 2025", # What was optimized targets=[OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value], # Initial prompts seed_prompts={"answering_system_prompt": "You are a helpful math assistant."}, # Optimized prompts (result) optimized_prompts={ "answering_system_prompt": """You are an expert competition mathematician. AIME answers are integers 0-999. Show your reasoning.""" }, # Scores train_score=0.75, val_score=0.70, test_score=0.68, improvement=0.166, # 16.6% improvement # GEPA parameters reflection_model="anthropic/claude-haiku-4-5", metric_calls=75, # Trajectory info best_generation=8, total_generations=10, ) print("OptimizationRun created:") print(f" Run ID: {run1.run_id}") print(f" Benchmark: {run1.benchmark_name}") print(f" Val Score: {run1.val_score:.2%}") print(f" Improvement: {run1.improvement:.2%}")

Run Fields¶

Field	Type	Description
run_id	str	Unique identifier (auto-generated)
timestamp	datetime	When the run occurred
benchmark_name	str	Name of the benchmark
targets	list[str]	What was optimized
seed_prompts	dict	Initial prompts
optimized_prompts	dict	Final optimized prompts
train_score	float	Training set score
val_score	float	Validation set score
test_score	float	Test set score (optional)
improvement	float	Relative improvement
reflection_model	str	GEPA reflection model
metric_calls	int	Number of evaluations
best_generation	int	Best generation number
total_generations	int	Total generations run
model_scores	dict	Per-model scores (for Pareto)

---

Logging Runs¶

Use log_run() to persist optimization runs.

In [ ]:

# Log the run
run_id = tracker.log_run(run1)
print(f"Logged run: {run_id}")

# Log the run run_id = tracker.log_run(run1) print(f"Logged run: {run_id}")

In [ ]:

# Create and log more runs for demonstration
runs_to_log = [
    OptimizationRun(
        benchmark_name="AIME 2025",
        targets=[OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value],
        seed_prompts={"answering_system_prompt": "Solve math problems."},
        optimized_prompts={"answering_system_prompt": "You are an AIME expert."},
        train_score=0.70,
        val_score=0.65,
        improvement=0.083,
        reflection_model="anthropic/claude-haiku-4-5",
        metric_calls=50,
        best_generation=5,
        total_generations=8,
        timestamp=datetime.now() - timedelta(hours=2),
    ),
    OptimizationRun(
        benchmark_name="AIME 2025",
        targets=[
            OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value,
            OptimizationTarget.PARSING_INSTRUCTIONS.value,
        ],
        seed_prompts={
            "answering_system_prompt": "You are a math tutor.",
            "parsing_instructions": "Extract the answer.",
        },
        optimized_prompts={
            "answering_system_prompt": "You are an AIME competition solver.",
            "parsing_instructions": "Find the integer 0-999 in the response.",
        },
        train_score=0.80,
        val_score=0.75,
        test_score=0.72,
        improvement=0.25,
        reflection_model="anthropic/claude-sonnet-4-5",
        metric_calls=100,
        best_generation=12,
        total_generations=15,
        timestamp=datetime.now() - timedelta(hours=1),
    ),
    OptimizationRun(
        benchmark_name="Math Benchmark",  # Different benchmark
        targets=[OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value],
        seed_prompts={"answering_system_prompt": "Calculate."},
        optimized_prompts={"answering_system_prompt": "Solve step by step."},
        train_score=0.85,
        val_score=0.82,
        improvement=0.37,
        reflection_model="anthropic/claude-haiku-4-5",
        metric_calls=60,
        best_generation=7,
        total_generations=10,
    ),
]

for run in runs_to_log:
    run_id = tracker.log_run(run)
    print(f"Logged: {run_id} ({run.benchmark_name}, val_score={run.val_score:.2%})")

# Create and log more runs for demonstration runs_to_log = [ OptimizationRun( benchmark_name="AIME 2025", targets=[OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value], seed_prompts={"answering_system_prompt": "Solve math problems."}, optimized_prompts={"answering_system_prompt": "You are an AIME expert."}, train_score=0.70, val_score=0.65, improvement=0.083, reflection_model="anthropic/claude-haiku-4-5", metric_calls=50, best_generation=5, total_generations=8, timestamp=datetime.now() - timedelta(hours=2), ), OptimizationRun( benchmark_name="AIME 2025", targets=[ OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value, OptimizationTarget.PARSING_INSTRUCTIONS.value, ], seed_prompts={ "answering_system_prompt": "You are a math tutor.", "parsing_instructions": "Extract the answer.", }, optimized_prompts={ "answering_system_prompt": "You are an AIME competition solver.", "parsing_instructions": "Find the integer 0-999 in the response.", }, train_score=0.80, val_score=0.75, test_score=0.72, improvement=0.25, reflection_model="anthropic/claude-sonnet-4-5", metric_calls=100, best_generation=12, total_generations=15, timestamp=datetime.now() - timedelta(hours=1), ), OptimizationRun( benchmark_name="Math Benchmark", # Different benchmark targets=[OptimizationTarget.ANSWERING_SYSTEM_PROMPT.value], seed_prompts={"answering_system_prompt": "Calculate."}, optimized_prompts={"answering_system_prompt": "Solve step by step."}, train_score=0.85, val_score=0.82, improvement=0.37, reflection_model="anthropic/claude-haiku-4-5", metric_calls=60, best_generation=7, total_generations=10, ), ] for run in runs_to_log: run_id = tracker.log_run(run) print(f"Logged: {run_id} ({run.benchmark_name}, val_score={run.val_score:.2%})")

---

Retrieving Runs¶

Get a Specific Run¶

In [ ]:

# Get run by ID
retrieved = tracker.get_run(run1.run_id)

if retrieved:
    print("Retrieved run:")
    print(f"  ID: {retrieved.run_id}")
    print(f"  Benchmark: {retrieved.benchmark_name}")
    print(f"  Val Score: {retrieved.val_score:.2%}")
    print(f"  Timestamp: {retrieved.timestamp}")

# Get run by ID retrieved = tracker.get_run(run1.run_id) if retrieved: print("Retrieved run:") print(f" ID: {retrieved.run_id}") print(f" Benchmark: {retrieved.benchmark_name}") print(f" Val Score: {retrieved.val_score:.2%}") print(f" Timestamp: {retrieved.timestamp}")

Get Best Run for a Benchmark¶

In [ ]:

# Get best by validation score
best_by_val = tracker.get_best_run("AIME 2025", metric="val_score")

if best_by_val:
    print("Best AIME 2025 run (by val_score):")
    print(f"  ID: {best_by_val.run_id}")
    print(f"  Val Score: {best_by_val.val_score:.2%}")
    print(f"  Improvement: {best_by_val.improvement:.2%}")

# Get best by validation score best_by_val = tracker.get_best_run("AIME 2025", metric="val_score") if best_by_val: print("Best AIME 2025 run (by val_score):") print(f" ID: {best_by_val.run_id}") print(f" Val Score: {best_by_val.val_score:.2%}") print(f" Improvement: {best_by_val.improvement:.2%}")

In [ ]:

# Get best by improvement
best_by_imp = tracker.get_best_run("AIME 2025", metric="improvement")

if best_by_imp:
    print("Best AIME 2025 run (by improvement):")
    print(f"  ID: {best_by_imp.run_id}")
    print(f"  Val Score: {best_by_imp.val_score:.2%}")
    print(f"  Improvement: {best_by_imp.improvement:.2%}")

# Get best by improvement best_by_imp = tracker.get_best_run("AIME 2025", metric="improvement") if best_by_imp: print("Best AIME 2025 run (by improvement):") print(f" ID: {best_by_imp.run_id}") print(f" Val Score: {best_by_imp.val_score:.2%}") print(f" Improvement: {best_by_imp.improvement:.2%}")

List Runs¶

In [ ]:

# List all runs
all_runs = tracker.list_runs()

print(f"All runs ({len(all_runs)}):")
for run in all_runs:
    print(f"  {run.run_id}: {run.benchmark_name} | val={run.val_score:.2%} | imp={run.improvement:.2%}")

# List all runs all_runs = tracker.list_runs() print(f"All runs ({len(all_runs)}):") for run in all_runs: print(f" {run.run_id}: {run.benchmark_name} | val={run.val_score:.2%} | imp={run.improvement:.2%}")

In [ ]:

# List runs for specific benchmark
aime_runs = tracker.list_runs(benchmark_name="AIME 2025")

print(f"AIME 2025 runs ({len(aime_runs)}):")
for run in aime_runs:
    print(f"  {run.run_id}: val={run.val_score:.2%} | imp={run.improvement:.2%}")

# List runs for specific benchmark aime_runs = tracker.list_runs(benchmark_name="AIME 2025") print(f"AIME 2025 runs ({len(aime_runs)}):") for run in aime_runs: print(f" {run.run_id}: val={run.val_score:.2%} | imp={run.improvement:.2%}")

In [ ]:

# Pagination
first_two = tracker.list_runs(limit=2, offset=0)
next_two = tracker.list_runs(limit=2, offset=2)

print(f"First 2: {[r.run_id for r in first_two]}")
print(f"Next 2: {[r.run_id for r in next_two]}")

# Pagination first_two = tracker.list_runs(limit=2, offset=0) next_two = tracker.list_runs(limit=2, offset=2) print(f"First 2: {[r.run_id for r in first_two]}") print(f"Next 2: {[r.run_id for r in next_two]}")

---

Comparing Runs¶

In [ ]:

# Compare specific runs
run_ids = [r.run_id for r in aime_runs[:3]]
comparison = tracker.compare_runs(run_ids)

print("Run Comparison:")
print(f"  Runs compared: {len(comparison['runs'])}")
print(f"  Best by val_score: {comparison['best']['val_score']}")
print(f"  Best by improvement: {comparison['best']['improvement']}")

print("\nMetrics:")
print(f"  Val scores: {comparison['metrics']['val_score']}")
print(f"  Improvements: {comparison['metrics']['improvement']}")

# Compare specific runs run_ids = [r.run_id for r in aime_runs[:3]] comparison = tracker.compare_runs(run_ids) print("Run Comparison:") print(f" Runs compared: {len(comparison['runs'])}") print(f" Best by val_score: {comparison['best']['val_score']}") print(f" Best by improvement: {comparison['best']['improvement']}") print("\nMetrics:") print(f" Val scores: {comparison['metrics']['val_score']}") print(f" Improvements: {comparison['metrics']['improvement']}")

---

Improvement Trends¶

In [ ]:

# Get improvement trend for a benchmark
trend = tracker.get_improvement_trend("AIME 2025", limit=10)

print("AIME 2025 Improvement Trend:")
for entry in trend:
    print(f"  {entry['timestamp'][:16]}: val={entry['val_score']:.2%}, imp={entry['improvement']:.2%}")

# Get improvement trend for a benchmark trend = tracker.get_improvement_trend("AIME 2025", limit=10) print("AIME 2025 Improvement Trend:") for entry in trend: print(f" {entry['timestamp'][:16]}: val={entry['val_score']:.2%}, imp={entry['improvement']:.2%}")

---

Exporting History¶

In [ ]:

# Export as JSON
json_export = tracker.export_history(format="json", benchmark_name="AIME 2025")

print("JSON Export (first 500 chars):")
print(json_export[:500])

# Export as JSON json_export = tracker.export_history(format="json", benchmark_name="AIME 2025") print("JSON Export (first 500 chars):") print(json_export[:500])

In [ ]:

# Export as CSV
csv_export = tracker.export_history(format="csv")

print("CSV Export:")
for line in csv_export.split("\n"):
    print(line)

# Export as CSV csv_export = tracker.export_history(format="csv") print("CSV Export:") for line in csv_export.split("\n"): print(line)

---

Deleting Runs¶

In [ ]:

# Delete a run
if all_runs:
    run_to_delete = all_runs[-1].run_id
    deleted = tracker.delete_run(run_to_delete)
    print(f"Deleted run {run_to_delete}: {deleted}")

    # Verify deletion
    remaining = tracker.list_runs()
    print(f"Remaining runs: {len(remaining)}")

# Delete a run if all_runs: run_to_delete = all_runs[-1].run_id deleted = tracker.delete_run(run_to_delete) print(f"Deleted run {run_to_delete}: {deleted}") # Verify deletion remaining = tracker.list_runs() print(f"Remaining runs: {len(remaining)}")

---

Cleanup¶

In [ ]:

# Clean up temp directory
import shutil

shutil.rmtree(temp_dir, ignore_errors=True)
print(f"Cleaned up: {temp_dir}")

# Clean up temp directory import shutil shutil.rmtree(temp_dir, ignore_errors=True) print(f"Cleaned up: {temp_dir}")

---

Summary¶

Method	Purpose
log_run()	Persist an optimization run
get_run()	Retrieve a specific run
get_best_run()	Get best run for a benchmark
list_runs()	List runs with filtering
compare_runs()	Compare multiple runs
get_improvement_trend()	Analyze trends over time
export_history()	Export as JSON/CSV
delete_run()	Remove a run

Next Steps¶

• 08_export_and_reuse.ipynb - Export optimized prompts for reuse