🚀 Run Module API Reference

The run module provides the main interface for executing and analyzing experiments.

Main user interface for running experiments.

This module provides the simple, intelligent interface that users interact with. No complex configuration needed - just run experiments and get insights.

class rexf.run.ExperimentRunner(storage_path='experiments.db', intelligent=True)[source]

Bases: object

Intelligent experiment runner with zero configuration.

This replaces the complex ExperimentRunner with a much simpler interface focused on user experience rather than architectural purity.

__init__(storage_path='experiments.db', intelligent=True)[source]: Initialize with minimal configuration.

single(experiment_func, **params)[source]

Run a single experiment with given parameters.

Parameters:

experiment_func (Callable) – Function decorated with @experiment
**params – Parameter values to use

Return type:

str

Returns:

Experiment run ID

best(metric=None, top=5)[source]

Get the best experiments by a metric.

Parameters:

metric (Optional[str]) – Metric to optimize for (auto-detected if None)
top (int) – Number of top experiments to return

Return type:

List[ExperimentData]

Returns:

List of best experiments

failed()[source]

Get all failed experiments.

Return type:: List[ExperimentData]

recent(hours=24)[source]

Get experiments from the last N hours.

Return type:: List[ExperimentData]

find(query)[source]

Find experiments using query expressions.

Supports queries like: - “accuracy > 0.9” - “status == ‘completed’” - “runtime < 60” - “param_learning_rate between 0.001 and 0.01”

Parameters:: query (str) – Query expression string
Return type:: List[ExperimentData]
Returns:: List of matching experiments

compare(experiments=None)[source]

Smart comparison of experiments with detailed analysis.

Parameters:: experiments (Optional[List[Union[str, ExperimentData]]]) – List of experiment IDs or ExperimentData objects. If None, compares recent experiments.
Return type:: None

insights(experiment_name=None)[source]

Generate comprehensive insights from experiments.

Parameters:: experiment_name (Optional[str]) – Optional experiment name to filter by
Return type:: Dict[str, Any]
Returns:: Dictionary of insights and recommendations

suggest(experiment_func=None, count=3, strategy='balanced', optimization_target=None)[source]

Suggest next experiments to run.

Parameters:

experiment_func – Function decorated with @experiment
count (int) – Number of suggestions to generate
strategy (str) – Suggestion strategy (“exploit”, “explore”, “balanced”)
optimization_target (Optional[str]) – Metric to optimize for

Return type:

Dict[str, Any]

Returns:

Dictionary with parameter suggestions and reasoning

auto_explore(experiment_func, strategy='random', budget=10, parameter_ranges=None, optimization_target=None)[source]

Automatically explore parameter space for an experiment.

Parameters:

experiment_func – Function decorated with @experiment
strategy (str) – Exploration strategy (“random”, “grid”, “adaptive”)
budget (int) – Number of experiments to run
parameter_ranges (Optional[Dict[str, Any]]) – Dict of parameter name -> range specification
optimization_target (Optional[str]) – Metric to optimize for

Return type:

List[str]

Returns:

List of experiment run IDs

parameter_space(experiment_name=None)[source]

Get parameter space summary for experiments.

Parameters:: experiment_name (Optional[str]) – Optional experiment name to filter by
Return type:: Dict[str, Any]
Returns:: Dictionary with parameter space analysis

metric_trends(metric_name, experiment_name=None)[source]

Get trends for a specific metric over time.

Parameters:

metric_name (str) – Name of the metric to analyze
experiment_name (Optional[str]) – Optional experiment name to filter by

Return type:

Dict[str, Any]

Returns:

Dictionary with trend analysis

query_help()[source]

Show help for query expressions.

Return type:: None

dashboard(host='localhost', port=8080, open_browser=True)[source]

Launch web dashboard for experiment visualization.

Parameters:

host (str) – Host to bind the server to
port (int) – Port to bind the server to
open_browser (bool) – Whether to automatically open the browser

Return type:

None

Returns:

None

close()[source]: Clean up resources.

rexf.run.single(experiment_func, **params)[source]

Run a single experiment. See ExperimentRunner.single().

Return type:: str

rexf.run.best(metric=None, top=5)[source]

Get best experiments. See ExperimentRunner.best().

Return type:: List[ExperimentData]

rexf.run.failed()[source]

Get failed experiments. See ExperimentRunner.failed().

Return type:: List[ExperimentData]

rexf.run.recent(hours=24)[source]

Get recent experiments. See ExperimentRunner.recent().

Return type:: List[ExperimentData]

rexf.run.find(query)[source]

Find experiments by query. See ExperimentRunner.find().

Return type:: List[ExperimentData]

rexf.run.compare(experiments=None)[source]

Compare experiments. See ExperimentRunner.compare().

Return type:: None

rexf.run.insights(experiment_name=None)[source]

Get experiment insights. See ExperimentRunner.insights().

Return type:: Dict[str, Any]

rexf.run.suggest(experiment_func=None, count=3, strategy='balanced', optimization_target=None)[source]

Get experiment suggestions. See ExperimentRunner.suggest().

Return type:: Dict[str, Any]

rexf.run.parameter_space(experiment_name=None)[source]

Get parameter space analysis. See ExperimentRunner.parameter_space().

Return type:: Dict[str, Any]

rexf.run.metric_trends(metric_name, experiment_name=None)[source]

Get metric trends. See ExperimentRunner.metric_trends().

Return type:: Dict[str, Any]

rexf.run.query_help()[source]

Show query help. See ExperimentRunner.query_help().

Return type:: None

rexf.run.auto_explore(experiment_func, strategy='random', budget=10, parameter_ranges=None, optimization_target=None)[source]

Auto-explore parameter space. See ExperimentRunner.auto_explore().

Return type:: List[str]

rexf.run.dashboard(host='localhost', port=8080, open_browser=True)[source]

Launch web dashboard. See ExperimentRunner.dashboard().

Return type:: None

Quick Reference

from rexf import run

# Execute experiments
run_id = run.single(my_experiment, param1=value1)
run_ids = run.auto_explore(my_experiment, strategy="random", budget=10)

# Retrieve results
experiments = run.all()
recent = run.recent(hours=6)
best = run.best(metric="accuracy", top=5)
found = run.find("accuracy > 0.9")

# Analysis and insights
insights = run.insights()
suggestions = run.suggest(my_experiment, count=5)
run.compare(best[:3])

# Interactive tools
run.dashboard()

Execution Functions

single()

Execute a single experiment with specified parameters.

rexf.run.single(experiment_func, **params)[source]

Run a single experiment. See ExperimentRunner.single().

Return type:: str

# Basic usage
run_id = run.single(my_experiment)

# With parameters
run_id = run.single(my_experiment, learning_rate=0.01, epochs=100)

# With keyword arguments
run_id = run.single(
    my_experiment,
    **{"learning_rate": 0.01, "batch_size": 32}
)

auto_explore()

Automatically explore parameter space using different strategies.

rexf.run.auto_explore(experiment_func, strategy='random', budget=10, parameter_ranges=None, optimization_target=None)[source]

Auto-explore parameter space. See ExperimentRunner.auto_explore().

Return type:: List[str]

# Random exploration
run_ids = run.auto_explore(
    my_experiment,
    strategy="random",
    budget=20,
    optimization_target="accuracy"
)

# Grid search
run_ids = run.auto_explore(
    my_experiment,
    strategy="grid",
    budget=15,
    parameter_ranges={
        "learning_rate": [0.001, 0.01, 0.1],
        "batch_size": [32, 64, 128]
    }
)

# Adaptive/Bayesian optimization
run_ids = run.auto_explore(
    my_experiment,
    strategy="adaptive",
    budget=25,
    optimization_target="accuracy"
)

Retrieval Functions

all()

Get all experiments from the database.

experiments = run.all()
print(f"Total experiments: {len(experiments)}")

recent()

Get experiments from recent time period.

rexf.run.recent(hours=24)[source]

Get recent experiments. See ExperimentRunner.recent().

Return type:: List[ExperimentData]

# Last 24 hours (default)
recent_experiments = run.recent()

# Last 6 hours
last_6h = run.recent(hours=6)

# Last 2 days
last_2d = run.recent(hours=48)

best()

Get top-performing experiments by a specific metric.

rexf.run.best(metric=None, top=5)[source]

Get best experiments. See ExperimentRunner.best().

Return type:: List[ExperimentData]

# Top 5 by accuracy (descending)
top_accuracy = run.best(metric="accuracy", top=5)

# Fastest experiments (ascending order)
fastest = run.best(metric="training_time", top=3, ascending=True)

# Best from specific experiment
best_ml = run.best(
    metric="f1_score",
    top=10,
    experiment_name="ml_experiment"
)

find()

Query experiments using expression strings.

rexf.run.find(query)[source]

Find experiments by query. See ExperimentRunner.find().

Return type:: List[ExperimentData]

# Simple comparisons
high_acc = run.find("accuracy > 0.9")
fast_runs = run.find("training_time < 60")
recent_good = run.find("accuracy > 0.8 and start_time > '2024-01-01'")

# Parameter queries (use param_ prefix)
low_lr = run.find("param_learning_rate < 0.01")
large_batches = run.find("param_batch_size >= 64")

# Range queries
moderate_acc = run.find("accuracy between 0.7 and 0.9")

# Status queries
completed = run.find("status == 'completed'")
failed = run.find("status == 'failed'")

Query Operators

Supported operators in find() expressions:

Comparison: >, >=, <, <=, ==, !=
Range: between (e.g., "value between 0.1 and 0.9")
Logical: and (or support coming soon)

Query Examples:

# Numeric comparisons
run.find("accuracy > 0.95")
run.find("loss <= 0.1")
run.find("epochs != 100")

# String comparisons
run.find("status == 'completed'")
run.find("param_optimizer == 'adam'")

# Range queries
run.find("accuracy between 0.8 and 0.95")
run.find("param_learning_rate between 0.001 and 0.1")

# Combined conditions
run.find("accuracy > 0.9 and training_time < 300")
run.find("param_batch_size >= 32 and status == 'completed'")

by_name()

Get experiments by experiment function name.

# Get all runs of specific experiment
ml_experiments = run.by_name("ml_hyperparameter_search")

# Get recent runs of specific experiment
recent_ml = run.by_name("ml_experiment", limit=10)

get_by_id()

Get a specific experiment by its run ID.

# Get specific experiment
experiment = run.get_by_id("abc123def456")

if experiment:
    print(f"Status: {experiment.status}")
    print(f"Metrics: {experiment.metrics}")
    print(f"Parameters: {experiment.parameters}")

Analysis Functions

insights()

Generate intelligent insights from experiment data.

rexf.run.insights(experiment_name=None)[source]

Get experiment insights. See ExperimentRunner.insights().

Return type:: Dict[str, Any]

# Overall insights
insights = run.insights()

# Insights for specific experiment
ml_insights = run.insights(experiment_name="ml_experiment")

# Access insight categories
summary = insights["summary"]
param_insights = insights["parameter_insights"]
performance = insights["performance_insights"]
correlations = insights["correlation_insights"]
anomalies = insights["anomaly_insights"]
recommendations = insights["recommendations"]

Example insight structure:

{
    "summary": {
        "total_experiments": 150,
        "success_rate": 0.92,
        "avg_accuracy": 0.847,
        "total_duration": 3600.5
    },
    "parameter_insights": {
        "learning_rate": {
            "impact_score": 0.85,
            "optimal_range": [0.001, 0.01],
            "correlation": 0.72
        }
    },
    "performance_insights": {
        "fastest_duration": 45.2,
        "slowest_duration": 320.1,
        "avg_duration": 120.5
    },
    "recommendations": [
        {
            "title": "Optimize learning rate",
            "description": "Try learning rates between 0.005 and 0.015",
            "priority": "high"
        }
    ]
}

suggest()

Get intelligent suggestions for next experiments.

rexf.run.suggest(experiment_func=None, count=3, strategy='balanced', optimization_target=None)[source]

Get experiment suggestions. See ExperimentRunner.suggest().

Return type:: Dict[str, Any]

# Basic suggestions
suggestions = run.suggest(my_experiment, count=5)

# Exploitation strategy (focus on best regions)
exploit_suggestions = run.suggest(
    my_experiment,
    count=3,
    strategy="exploit",
    optimization_target="accuracy"
)

# Exploration strategy (try new regions)
explore_suggestions = run.suggest(
    my_experiment,
    count=5,
    strategy="explore"
)

# Balanced strategy (mix of both)
balanced_suggestions = run.suggest(
    my_experiment,
    count=4,
    strategy="balanced",
    optimization_target="f1_score"
)

# Use suggestions
for suggestion in suggestions["suggestions"]:
    print(f"Try: {suggestion['parameters']}")
    print(f"Reason: {suggestion['reasoning']}")
    print(f"Expected improvement: {suggestion['expected_improvement']}")

compare()

Compare multiple experiments side-by-side.

rexf.run.compare(experiments=None)[source]

Compare experiments. See ExperimentRunner.compare().

Return type:: None

# Compare best experiments
best_experiments = run.best(metric="accuracy", top=3)
run.compare(best_experiments)

# Compare specific experiments by ID
run.compare(["run_id_1", "run_id_2", "run_id_3"])

# Compare from query results
high_accuracy = run.find("accuracy > 0.9")
run.compare(high_accuracy[:5])

The comparison displays:

Parameter differences between experiments
Metric comparisons with statistical analysis
Performance analysis
Recommendations based on differences

Interactive Functions

dashboard()

Launch the interactive web dashboard.

rexf.run.dashboard(host='localhost', port=8080, open_browser=True)[source]

Launch web dashboard. See ExperimentRunner.dashboard().

Return type:: None

# Launch with defaults (localhost:8080)
run.dashboard()

# Custom host and port
run.dashboard(host="0.0.0.0", port=9000)

# Don't open browser automatically
run.dashboard(open_browser=False)

query_help()

Get help with query syntax and suggestions.

rexf.run.query_help()[source]

Show query help. See ExperimentRunner.query_help().

Return type:: None

# Get query suggestions
suggestions = run.query_help()

print("Suggested queries:")
for suggestion in suggestions:
    print(f"  {suggestion}")

# Example output:
# accuracy > 0.9
# param_learning_rate < 0.01
# training_time between 60 and 300

Utility Functions

summary()

Get a quick summary of all experiments.

summary = run.summary()

print(f"Total experiments: {summary['total']}")
print(f"Successful: {summary['successful']}")
print(f"Failed: {summary['failed']}")
print(f"Average duration: {summary['avg_duration']:.2f}s")

stats()

Get detailed statistics about experiments.

stats = run.stats()

# Experiment counts by name
by_name = stats["by_experiment_name"]

# Success/failure rates
success_rate = stats["success_rate"]

# Performance statistics
performance = stats["performance_stats"]

Advanced Usage

Working with ExperimentRunner

For advanced use cases, you can work directly with the ExperimentRunner:

from rexf.run import ExperimentRunner

# Create custom runner with specific storage path
runner = ExperimentRunner(storage_path="custom_experiments.db")

# Use custom runner
run_id = runner.single(my_experiment, param1=value1)

# Get insights from custom runner
insights = runner.insights()

# Clean up
runner.close()

Batch Operations

For efficient batch processing:

# Process multiple parameter sets efficiently
param_sets = [
    {"learning_rate": 0.01, "batch_size": 32},
    {"learning_rate": 0.005, "batch_size": 64},
    {"learning_rate": 0.001, "batch_size": 128},
]

run_ids = []
for params in param_sets:
    run_id = run.single(my_experiment, **params)
    run_ids.append(run_id)

# Analyze batch results
batch_experiments = [run.get_by_id(rid) for rid in run_ids]
run.compare(batch_experiments)

Error Handling

The run module gracefully handles various error conditions:

# Experiment function that might fail
@experiment
def risky_experiment(fail_probability=0.1):
    import random
    if random.random() < fail_probability:
        raise ValueError("Simulated failure")
    return {"success": True}

# Failed experiments still return run_ids
run_id = run.single(risky_experiment, fail_probability=0.5)

# Check if experiment failed
experiment = run.get_by_id(run_id)
if experiment.status == "failed":
    error_msg = experiment.metadata.get("error", "Unknown error")
    print(f"Experiment failed: {error_msg}")

# Query failed experiments
failed_experiments = run.find("status == 'failed'")
print(f"Found {len(failed_experiments)} failed experiments")

Performance Tips

Use batch operations for multiple experiments
Limit query results with run.find() for large datasets
Close runners when using custom ExperimentRunner instances
Use specific queries instead of retrieving all experiments

# Good: Specific query
recent_good = run.find("accuracy > 0.9 and start_time > '2024-01-01'")

# Less efficient: Get all then filter
all_experiments = run.all()
recent_good = [exp for exp in all_experiments
               if exp.metrics.get("accuracy", 0) > 0.9]