| name | adaptive-workflows |
| description | Self-learning workflow system that tracks what works best for your use cases. Records experiment results, suggests optimizations, creates custom templates, and builds a personal knowledge base. Use to learn from experience and optimize your LLM workflows over time. |
Adaptive Workflows
Build a self-learning system that gets better with every experiment you run.
Overview
Learn from experience:
- Experiment tracking - Record all experiments and results
- Pattern recognition - Identify what works best for your use cases
- Smart recommendations - Get suggestions based on past success
- Workflow templates - Create reusable templates from successful experiments
- A/B testing - Compare approaches systematically
- Knowledge base - Build your personal best practices library
- Continuous improvement - Workflows get better over time
Quick Start
Initialize Workflow Tracker
import json
from datetime import datetime
from pathlib import Path
class WorkflowTracker:
def __init__(self, storage_path="./workflows.json"):
self.storage_path = Path(storage_path)
self.experiments = self.load_experiments()
def load_experiments(self):
"""Load previous experiments"""
if self.storage_path.exists():
with open(self.storage_path, 'r') as f:
return json.load(f)
return []
def save_experiments(self):
"""Save experiments to disk"""
with open(self.storage_path, 'w') as f:
json.dump(self.experiments, f, indent=2)
def record_experiment(self, experiment: dict):
"""Record a new experiment"""
experiment['timestamp'] = datetime.now().isoformat()
experiment['id'] = len(self.experiments)
self.experiments.append(experiment)
self.save_experiments()
return experiment['id']
# Initialize
tracker = WorkflowTracker()
Record Your First Experiment
# After training
experiment = {
'task': 'medical_qa_finetuning',
'model': 'Llama-3.2-7B',
'dataset_size': 1000,
'hyperparameters': {
'learning_rate': 2e-4,
'lora_rank': 16,
'batch_size': 8,
'epochs': 3
},
'results': {
'final_loss': 0.42,
'training_time_hours': 2.5,
'cost_usd': 5.20,
'eval_accuracy': 0.89
},
'notes': 'Worked well, converged smoothly'
}
tracker.record_experiment(experiment)
Get Recommendations
# Get recommendations for similar task
recommendations = tracker.suggest_config(
task='medical_qa_finetuning',
constraints={'cost': 10, 'time_hours': 4}
)
print(f"Recommended learning rate: {recommendations['learning_rate']}")
print(f"Recommended rank: {recommendations['lora_rank']}")
Experiment Tracking
Complete Experiment Schema
experiment_schema = {
# Identifiers
'id': 'auto-generated',
'timestamp': 'ISO-8601',
'task': 'task_category', # e.g., 'medical_qa', 'code_generation'
# Model configuration
'model': {
'base_model': 'Llama-3.2-7B',
'quantization': '4bit',
'max_seq_length': 2048
},
# Dataset
'dataset': {
'name': 'medical_qa_v1',
'size': 1000,
'format': 'alpaca',
'quality_score': 0.85,
'source': 'synthetic'
},
# Hyperparameters
'hyperparameters': {
'learning_rate': 2e-4,
'lr_scheduler': 'cosine',
'warmup_ratio': 0.03,
'lora_rank': 16,
'lora_alpha': 16,
'batch_size': 8,
'gradient_accumulation': 1,
'epochs': 3,
'weight_decay': 0.01
},
# Results
'results': {
'final_loss': 0.42,
'best_loss': 0.38,
'eval_loss': 0.45,
'eval_accuracy': 0.89,
'training_time_hours': 2.5,
'cost_usd': 5.20,
'tokens_per_sec': 1200,
'peak_memory_gb': 18.5
},
# Deployment
'deployment': {
'format': 'GGUF',
'quantization': 'q4_k_m',
'deployment_platform': 'Ollama',
'inference_speed_tok_s': 50
},
# Evaluation
'evaluation': {
'test_accuracy': 0.87,
'human_eval_score': 4.2, # out of 5
'sample_outputs': ['...', '...'],
'issues_found': ['Occasional hallucination on rare conditions']
},
# Metadata
'tags': ['medical', 'qa', 'production'],
'success': True,
'notes': 'Worked well. Deployed to production.',
'git_commit': 'abc123'
}
Track Experiment
class ExperimentTracker(WorkflowTracker):
def track_training_run(
self,
task: str,
config: dict,
trainer, # SFTTrainer instance
dataset_info: dict
):
"""Automatically track a training run"""
experiment = {
'task': task,
'model': config['model_name'],
'dataset': dataset_info,
'hyperparameters': {
'learning_rate': config['learning_rate'],
'lora_rank': config['lora_rank'],
'lora_alpha': config['lora_alpha'],
'batch_size': config['batch_size'],
'epochs': config['num_epochs']
}
}
# Train
import time
start_time = time.time()
result = trainer.train()
training_time = (time.time() - start_time) / 3600 # hours
# Record results
experiment['results'] = {
'final_loss': result.training_loss,
'training_time_hours': training_time,
'cost_usd': self.estimate_cost(training_time, config['gpu_type'])
}
# Save
exp_id = self.record_experiment(experiment)
print(f"Experiment {exp_id} recorded")
return exp_id
def estimate_cost(self, hours: float, gpu_type: str) -> float:
"""Estimate training cost"""
gpu_costs = {
'RTX_4090': 0.40,
'A100': 1.50,
'H100': 3.00
}
return hours * gpu_costs.get(gpu_type, 1.0)
# Usage
tracker = ExperimentTracker()
exp_id = tracker.track_training_run(
task='medical_qa',
config={...},
trainer=trainer,
dataset_info={...}
)
Pattern Recognition
Identify Best Configurations
class PatternAnalyzer(ExperimentTracker):
def find_best_experiments(
self,
task: str,
metric: str = 'eval_accuracy',
top_k: int = 5
):
"""Find top performing experiments for a task"""
# Filter by task
task_experiments = [
exp for exp in self.experiments
if exp['task'] == task and 'results' in exp
]
# Sort by metric
sorted_exps = sorted(
task_experiments,
key=lambda x: x['results'].get(metric, 0),
reverse=True
)
return sorted_exps[:top_k]
def analyze_hyperparameter_impact(self, task: str, hyperparameter: str):
"""Analyze how a hyperparameter affects results"""
task_exps = [exp for exp in self.experiments if exp['task'] == task]
# Group by hyperparameter value
from collections import defaultdict
groups = defaultdict(list)
for exp in task_exps:
value = exp['hyperparameters'].get(hyperparameter)
if value and 'results' in exp:
groups[value].append(exp['results'].get('eval_accuracy', 0))
# Calculate averages
analysis = {
value: {
'count': len(scores),
'mean_accuracy': np.mean(scores),
'std': np.std(scores)
}
for value, scores in groups.items()
}
return analysis
# Usage
analyzer = PatternAnalyzer()
# Find best medical QA experiments
best = analyzer.find_best_experiments('medical_qa', top_k=3)
for exp in best:
print(f"Experiment {exp['id']}: {exp['results']['eval_accuracy']:.3f}")
# Analyze learning rate impact
lr_analysis = analyzer.analyze_hyperparameter_impact('medical_qa', 'learning_rate')
for lr, stats in lr_analysis.items():
print(f"LR {lr}: {stats['mean_accuracy']:.3f} ± {stats['std']:.3f}")
Detect Success Patterns
def detect_patterns(self, task: str):
"""Detect what makes experiments successful"""
successful = [
exp for exp in self.experiments
if exp['task'] == task
and exp['results'].get('eval_accuracy', 0) > 0.85
]
failed = [
exp for exp in self.experiments
if exp['task'] == task
and exp['results'].get('eval_accuracy', 0) < 0.70
]
patterns = {}
# Analyze each hyperparameter
for hyperparam in ['learning_rate', 'lora_rank', 'batch_size']:
success_values = [
exp['hyperparameters'][hyperparam]
for exp in successful
if hyperparam in exp['hyperparameters']
]
fail_values = [
exp['hyperparameters'][hyperparam]
for exp in failed
if hyperparam in exp['hyperparameters']
]
patterns[hyperparam] = {
'success_mean': np.mean(success_values),
'success_std': np.std(success_values),
'fail_mean': np.mean(fail_values),
'fail_std': np.std(fail_values),
'recommendation': 'Use higher value' if np.mean(success_values) > np.mean(fail_values) else 'Use lower value'
}
return patterns
Smart Recommendations
Configuration Recommender
class ConfigRecommender(PatternAnalyzer):
def suggest_config(
self,
task: str,
constraints: dict = None
):
"""Suggest optimal configuration based on past experiments"""
# Get successful experiments
successful_exps = [
exp for exp in self.experiments
if exp['task'] == task
and exp['results'].get('eval_accuracy', 0) > 0.80
]
if not successful_exps:
return self.get_default_config(task)
# Apply constraints
if constraints:
successful_exps = self.filter_by_constraints(
successful_exps,
constraints
)
if not successful_exps:
return self.get_default_config(task)
# Find best configuration
best_exp = max(
successful_exps,
key=lambda x: x['results'].get('eval_accuracy', 0)
)
# Extract configuration
recommended = {
'model': best_exp['model'],
'hyperparameters': best_exp['hyperparameters'],
'dataset': best_exp['dataset'],
'confidence': self.calculate_confidence(successful_exps),
'based_on_experiments': len(successful_exps),
'expected_accuracy': best_exp['results'].get('eval_accuracy'),
'expected_cost': best_exp['results'].get('cost_usd'),
'expected_time_hours': best_exp['results'].get('training_time_hours')
}
return recommended
def filter_by_constraints(self, experiments, constraints):
"""Filter experiments by constraints"""
filtered = experiments
if 'max_cost' in constraints:
filtered = [
exp for exp in filtered
if exp['results'].get('cost_usd', float('inf')) <= constraints['max_cost']
]
if 'max_time_hours' in constraints:
filtered = [
exp for exp in filtered
if exp['results'].get('training_time_hours', float('inf')) <= constraints['max_time_hours']
]
if 'min_accuracy' in constraints:
filtered = [
exp for exp in filtered
if exp['results'].get('eval_accuracy', 0) >= constraints['min_accuracy']
]
return filtered
def calculate_confidence(self, experiments):
"""Calculate confidence in recommendation"""
if len(experiments) >= 10:
return 'high'
elif len(experiments) >= 5:
return 'medium'
else:
return 'low'
# Usage
recommender = ConfigRecommender()
# Get recommendation with constraints
config = recommender.suggest_config(
task='medical_qa',
constraints={
'max_cost': 10,
'max_time_hours': 3,
'min_accuracy': 0.85
}
)
print(f"Recommended learning rate: {config['hyperparameters']['learning_rate']}")
print(f"Expected accuracy: {config['expected_accuracy']:.3f}")
print(f"Confidence: {config['confidence']}")
print(f"Based on {config['based_on_experiments']} similar experiments")
Automated Optimization Suggestions
def suggest_improvements(self, exp_id: int):
"""Suggest improvements for a specific experiment"""
exp = self.experiments[exp_id]
suggestions = []
# Compare to best experiments
best_exps = self.find_best_experiments(exp['task'], top_k=3)
if not best_exps:
return ["Not enough data for comparison"]
best_exp = best_exps[0]
current_accuracy = exp['results'].get('eval_accuracy', 0)
best_accuracy = best_exp['results'].get('eval_accuracy', 0)
if current_accuracy < best_accuracy:
# Compare hyperparameters
for hyperparam in ['learning_rate', 'lora_rank', 'batch_size']:
current_value = exp['hyperparameters'].get(hyperparam)
best_value = best_exp['hyperparameters'].get(hyperparam)
if current_value != best_value:
suggestions.append({
'parameter': hyperparam,
'current': current_value,
'suggested': best_value,
'reason': f'Top experiment uses {best_value}',
'expected_improvement': (best_accuracy - current_accuracy) * 100
})
# Check for common issues
if exp['results'].get('training_time_hours', 0) > 5:
suggestions.append({
'issue': 'slow_training',
'suggestion': 'Increase batch size or reduce gradient accumulation',
'impact': 'Faster training'
})
if exp['results'].get('final_loss') > exp['results'].get('best_loss', float('inf')) * 1.5:
suggestions.append({
'issue': 'unstable_training',
'suggestion': 'Reduce learning rate or add gradient clipping',
'impact': 'More stable convergence'
})
return suggestions
Workflow Templates
Create Template from Experiment
class WorkflowTemplateManager(ConfigRecommender):
def __init__(self, storage_path="./workflows.json"):
super().__init__(storage_path)
self.templates = self.load_templates()
def load_templates(self):
"""Load workflow templates"""
template_path = Path(self.storage_path).parent / "templates.json"
if template_path.exists():
with open(template_path, 'r') as f:
return json.load(f)
return []
def save_templates(self):
"""Save templates"""
template_path = Path(self.storage_path).parent / "templates.json"
with open(template_path, 'w') as f:
json.dump(self.templates, f, indent=2)
def create_template(
self,
name: str,
exp_id: int,
description: str = ""
):
"""Create a reusable template from successful experiment"""
exp = self.experiments[exp_id]
template = {
'name': name,
'description': description,
'task': exp['task'],
'created_from': exp_id,
'created_at': datetime.now().isoformat(),
'config': {
'model': exp['model'],
'dataset': exp['dataset'],
'hyperparameters': exp['hyperparameters']
},
'expected_results': exp['results'],
'usage_count': 0
}
self.templates.append(template)
self.save_templates()
return template
def use_template(self, template_name: str, overrides: dict = None):
"""Use a template with optional overrides"""
template = next(
(t for t in self.templates if t['name'] == template_name),
None
)
if not template:
raise ValueError(f"Template '{template_name}' not found")
# Start with template config
config = template['config'].copy()
# Apply overrides
if overrides:
for key, value in overrides.items():
if '.' in key: # Nested key like 'hyperparameters.learning_rate'
parts = key.split('.')
d = config
for part in parts[:-1]:
d = d[part]
d[parts[-1]] = value
else:
config[key] = value
# Update usage count
template['usage_count'] += 1
self.save_templates()
return config
def list_templates(self, task: str = None):
"""List available templates"""
templates = self.templates
if task:
templates = [t for t in templates if t['task'] == task]
return [
{
'name': t['name'],
'task': t['task'],
'description': t['description'],
'usage_count': t['usage_count'],
'expected_accuracy': t['expected_results'].get('eval_accuracy')
}
for t in templates
]
# Usage
manager = WorkflowTemplateManager()
# Create template from best experiment
template = manager.create_template(
name='medical_qa_standard',
exp_id=42, # ID of successful experiment
description='Standard configuration for medical Q&A fine-tuning'
)
# Use template
config = manager.use_template(
'medical_qa_standard',
overrides={'hyperparameters.learning_rate': 3e-4}
)
# List templates
templates = manager.list_templates(task='medical_qa')
for t in templates:
print(f"{t['name']}: {t['expected_accuracy']:.3f} accuracy (used {t['usage_count']} times)")
A/B Testing
Compare Configurations
class ABTester(WorkflowTemplateManager):
def setup_ab_test(
self,
task: str,
config_a: dict,
config_b: dict,
test_name: str,
num_runs: int = 3
):
"""Setup A/B test to compare two configurations"""
test = {
'name': test_name,
'task': task,
'config_a': config_a,
'config_b': config_b,
'num_runs': num_runs,
'completed_runs': 0,
'results_a': [],
'results_b': [],
'status': 'pending'
}
# Save test
tests_path = Path(self.storage_path).parent / "ab_tests.json"
tests = []
if tests_path.exists():
with open(tests_path, 'r') as f:
tests = json.load(f)
tests.append(test)
with open(tests_path, 'w') as f:
json.dump(tests, f, indent=2)
return test
def record_ab_result(
self,
test_name: str,
config: str, # 'a' or 'b'
results: dict
):
"""Record results from one run of A/B test"""
tests_path = Path(self.storage_path).parent / "ab_tests.json"
with open(tests_path, 'r') as f:
tests = json.load(f)
# Find test
test = next((t for t in tests if t['name'] == test_name), None)
if not test:
raise ValueError(f"Test '{test_name}' not found")
# Record result
if config == 'a':
test['results_a'].append(results)
else:
test['results_b'].append(results)
test['completed_runs'] += 1
# Check if complete
if (len(test['results_a']) >= test['num_runs'] and
len(test['results_b']) >= test['num_runs']):
test['status'] = 'complete'
test['analysis'] = self.analyze_ab_results(test)
# Save
with open(tests_path, 'w') as f:
json.dump(tests, f, indent=2)
return test
def analyze_ab_results(self, test: dict):
"""Analyze A/B test results"""
results_a = test['results_a']
results_b = test['results_b']
# Extract metrics
accuracy_a = [r['eval_accuracy'] for r in results_a]
accuracy_b = [r['eval_accuracy'] for r in results_b]
cost_a = [r['cost_usd'] for r in results_a]
cost_b = [r['cost_usd'] for r in results_b]
time_a = [r['training_time_hours'] for r in results_a]
time_b = [r['training_time_hours'] for r in results_b]
# Statistical analysis
from scipy import stats
# T-test for accuracy
t_stat, p_value = stats.ttest_ind(accuracy_a, accuracy_b)
analysis = {
'accuracy': {
'config_a_mean': np.mean(accuracy_a),
'config_a_std': np.std(accuracy_a),
'config_b_mean': np.mean(accuracy_b),
'config_b_std': np.std(accuracy_b),
'difference': np.mean(accuracy_b) - np.mean(accuracy_a),
'p_value': p_value,
'significant': p_value < 0.05
},
'cost': {
'config_a_mean': np.mean(cost_a),
'config_b_mean': np.mean(cost_b),
'difference': np.mean(cost_b) - np.mean(cost_a)
},
'time': {
'config_a_mean': np.mean(time_a),
'config_b_mean': np.mean(time_b),
'difference': np.mean(time_b) - np.mean(time_a)
},
'winner': 'b' if np.mean(accuracy_b) > np.mean(accuracy_a) else 'a',
'recommendation': self.generate_recommendation(
accuracy_a, accuracy_b,
cost_a, cost_b,
time_a, time_b
)
}
return analysis
def generate_recommendation(
self,
accuracy_a, accuracy_b,
cost_a, cost_b,
time_a, time_b
):
"""Generate recommendation from A/B test"""
acc_diff = np.mean(accuracy_b) - np.mean(accuracy_a)
cost_diff = np.mean(cost_b) - np.mean(cost_a)
time_diff = np.mean(time_b) - np.mean(time_a)
if acc_diff > 0.02: # B is 2% better
if cost_diff < 5: # And not much more expensive
return "Use Config B: significantly better accuracy at similar cost"
else:
return f"Use Config B if budget allows: {acc_diff*100:.1f}% better accuracy, ${cost_diff:.2f} more expensive"
elif acc_diff < -0.02: # A is better
return "Use Config A: better accuracy"
else: # Similar accuracy
if cost_diff < -2: # B is cheaper
return "Use Config B: similar accuracy, lower cost"
else:
return "Configs perform similarly, choose based on other factors"
# Usage
tester = ABTester()
# Setup A/B test
test = tester.setup_ab_test(
task='medical_qa',
config_a={'learning_rate': 2e-4, 'lora_rank': 16},
config_b={'learning_rate': 5e-4, 'lora_rank': 32},
test_name='lr_and_rank_test',
num_runs=3
)
# After each training run
tester.record_ab_result(
'lr_and_rank_test',
config='a',
results={'eval_accuracy': 0.87, 'cost_usd': 5.2, 'training_time_hours': 2.5}
)
# After all runs complete, get analysis
test = tester.get_test_results('lr_and_rank_test')
print(test['analysis']['recommendation'])
Knowledge Base
Build Personal Best Practices
class KnowledgeBase(ABTester):
def extract_insights(self):
"""Extract insights from all experiments"""
insights = {
'total_experiments': len(self.experiments),
'tasks': {},
'best_practices': [],
'common_issues': []
}
# Group by task
from collections import defaultdict
by_task = defaultdict(list)
for exp in self.experiments:
by_task[exp['task']].append(exp)
# Analyze each task
for task, exps in by_task.items():
successful = [e for e in exps if e['results'].get('eval_accuracy', 0) > 0.80]
if len(successful) < 3:
continue # Not enough data
insights['tasks'][task] = {
'total_experiments': len(exps),
'success_rate': len(successful) / len(exps),
'avg_accuracy': np.mean([e['results']['eval_accuracy'] for e in successful]),
'avg_cost': np.mean([e['results'].get('cost_usd', 0) for e in successful]),
'optimal_config': self.suggest_config(task)
}
# Extract best practices
insights['best_practices'] = self.extract_best_practices()
# Identify common issues
insights['common_issues'] = self.identify_common_issues()
return insights
def extract_best_practices(self):
"""Extract best practices from successful experiments"""
practices = []
# Analyze learning rate
lr_analysis = self.analyze_all_hyperparameter('learning_rate')
if lr_analysis:
practices.append({
'category': 'learning_rate',
'recommendation': f"Use learning rate around {lr_analysis['optimal']:.0e}",
'confidence': lr_analysis['confidence'],
'data_points': lr_analysis['count']
})
# Analyze LoRA rank
rank_analysis = self.analyze_all_hyperparameter('lora_rank')
if rank_analysis:
practices.append({
'category': 'lora_rank',
'recommendation': f"Use LoRA rank {rank_analysis['optimal']}",
'confidence': rank_analysis['confidence'],
'data_points': rank_analysis['count']
})
return practices
def identify_common_issues(self):
"""Identify common issues from failed experiments"""
issues = []
failed = [
exp for exp in self.experiments
if exp['results'].get('eval_accuracy', 1.0) < 0.70
]
if not failed:
return []
# Check for common patterns in failures
# High loss
high_loss = [
exp for exp in failed
if exp['results'].get('final_loss', 0) > 1.0
]
if len(high_loss) > len(failed) * 0.3:
issues.append({
'issue': 'High final loss in 30%+ of failed experiments',
'suggestion': 'Consider longer training or higher learning rate',
'affected_experiments': len(high_loss)
})
# Slow training
slow = [
exp for exp in failed
if exp['results'].get('training_time_hours', 0) > 5
]
if len(slow) > len(failed) * 0.3:
issues.append({
'issue': 'Slow training in failed experiments',
'suggestion': 'Optimize batch size and gradient checkpointing',
'affected_experiments': len(slow)
})
return issues
def generate_report(self):
"""Generate comprehensive report"""
insights = self.extract_insights()
report = f"""
# Adaptive Workflow Report
Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}
## Summary
- Total Experiments: {insights['total_experiments']}
- Tasks Tracked: {len(insights['tasks'])}
## Task Performance
"""
for task, stats in insights['tasks'].items():
report += f"""
### {task}
- Experiments: {stats['total_experiments']}
- Success Rate: {stats['success_rate']:.1%}
- Average Accuracy: {stats['avg_accuracy']:.3f}
- Average Cost: ${stats['avg_cost']:.2f}
**Recommended Configuration:**
```python
{json.dumps(stats['optimal_config']['hyperparameters'], indent=2)}
"""
report += "\n## Best Practices\n\n"
for practice in insights['best_practices']:
report += f"- **{practice['category']}**: {practice['recommendation']} "
report += f"(Confidence: {practice['confidence']}, based on {practice['data_points']} experiments)\n"
if insights['common_issues']:
report += "\n## Common Issues\n\n"
for issue in insights['common_issues']:
report += f"- **{issue['issue']}**: {issue['suggestion']}\n"
return report
Usage
kb = KnowledgeBase()
Generate report
report = kb.generate_report() print(report)
Save report
with open('workflow_report.md', 'w') as f: f.write(report)
## Integration with Other Skills
### Auto-Apply Learnings
```python
class AdaptiveTrainer(KnowledgeBase):
def smart_train(
self,
task: str,
dataset,
base_config: dict = None,
auto_optimize: bool = True
):
"""Train with automatic optimization based on past experience"""
# Get recommended config
if auto_optimize:
recommended = self.suggest_config(
task=task,
constraints=base_config.get('constraints', {})
)
print(f"Using optimized configuration based on {recommended['based_on_experiments']} past experiments")
config = recommended
else:
config = base_config or {}
# Setup model
model, tokenizer = FastLanguageModel.from_pretrained(
config['model'],
max_seq_length=config.get('max_seq_length', 2048),
load_in_4bit=True
)
# Setup LoRA
model = FastLanguageModel.get_peft_model(
model,
r=config['hyperparameters']['lora_rank'],
lora_alpha=config['hyperparameters']['lora_alpha'],
target_modules=["q_proj", "k_proj", "v_proj", "o_proj",
"gate_proj", "up_proj", "down_proj"]
)
# Setup training
trainer = SFTTrainer(
model=model,
tokenizer=tokenizer,
train_dataset=dataset,
args=TrainingArguments(
learning_rate=config['hyperparameters']['learning_rate'],
per_device_train_batch_size=config['hyperparameters']['batch_size'],
num_train_epochs=config['hyperparameters']['epochs'],
# ... other args
)
)
# Train and track
exp_id = self.track_training_run(
task=task,
config=config,
trainer=trainer,
dataset_info={'size': len(dataset)}
)
print(f"Training complete. Experiment ID: {exp_id}")
# Get suggestions for next time
suggestions = self.suggest_improvements(exp_id)
if suggestions:
print("\nSuggestions for next experiment:")
for s in suggestions:
print(f" - {s}")
return exp_id
# Usage
trainer = AdaptiveTrainer()
exp_id = trainer.smart_train(
task='medical_qa',
dataset=my_dataset,
auto_optimize=True # Use past learnings
)
Best Practices
1. Record Everything
# Always record experiments
# Even "failed" experiments provide valuable data
2. Be Consistent
# Use consistent task names and metrics
# Makes pattern recognition more effective
3. Regular Analysis
# Generate reports monthly
kb = KnowledgeBase()
report = kb.generate_report()
# Review and update templates
4. Version Control
# Keep workflow files in git
# Track evolution of your best practices
5. Start Simple
# Begin with basic tracking
# Add complexity as you gather data
Summary
Adaptive workflows enable continuous improvement:
- ✓ Track every experiment
- ✓ Analyze patterns in success/failure
- ✓ Get smart recommendations
- ✓ Create reusable templates
- ✓ Run A/B tests systematically
- ✓ Build personal knowledge base
- ✓ Improve with every iteration
The more you use it, the smarter it gets.
Start tracking today, and within 10-20 experiments, you'll have personalized best practices that work specifically for your use cases.