mirror of
https://github.com/SuperClaude-Org/SuperClaude_Framework.git
synced 2025-12-29 16:16:08 +00:00
a4ffe52724
PM Agent optimization (already committed separately): - superclaude/commands/pm.md: 1652→14 lines - superclaude/agents/pm-agent.md: 735→429 lines - docs/agents/pm-agent-guide.md: new guide file Other pending changes: - setup: framework_docs, mcp, logger, remove ui.py - superclaude: __main__, cli/app, cli/commands/install - tests: test_ui updates - scripts: workflow metrics analysis tools - docs/memory: session state updates 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
310 lines
10 KiB
Python
Executable File
310 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
A/B Testing Framework for Workflow Variants
|
|
|
|
Compares two workflow variants with statistical significance testing.
|
|
|
|
Usage:
|
|
python scripts/ab_test_workflows.py \\
|
|
--variant-a progressive_v3_layer2 \\
|
|
--variant-b experimental_eager_layer3 \\
|
|
--metric tokens_used
|
|
"""
|
|
|
|
import json
|
|
import argparse
|
|
from pathlib import Path
|
|
from typing import Dict, List, Tuple
|
|
import statistics
|
|
from scipy import stats
|
|
|
|
|
|
class ABTestAnalyzer:
|
|
"""A/B testing framework for workflow optimization"""
|
|
|
|
def __init__(self, metrics_file: Path):
|
|
self.metrics_file = metrics_file
|
|
self.metrics: List[Dict] = []
|
|
self._load_metrics()
|
|
|
|
def _load_metrics(self):
|
|
"""Load metrics from JSONL file"""
|
|
if not self.metrics_file.exists():
|
|
print(f"Error: {self.metrics_file} not found")
|
|
return
|
|
|
|
with open(self.metrics_file, 'r') as f:
|
|
for line in f:
|
|
if line.strip():
|
|
self.metrics.append(json.loads(line))
|
|
|
|
def get_variant_metrics(self, workflow_id: str) -> List[Dict]:
|
|
"""Get all metrics for a specific workflow variant"""
|
|
return [m for m in self.metrics if m['workflow_id'] == workflow_id]
|
|
|
|
def extract_metric_values(self, metrics: List[Dict], metric: str) -> List[float]:
|
|
"""Extract specific metric values from metrics list"""
|
|
values = []
|
|
for m in metrics:
|
|
if metric in m:
|
|
value = m[metric]
|
|
# Handle boolean metrics
|
|
if isinstance(value, bool):
|
|
value = 1.0 if value else 0.0
|
|
values.append(float(value))
|
|
return values
|
|
|
|
def calculate_statistics(self, values: List[float]) -> Dict:
|
|
"""Calculate statistical measures"""
|
|
if not values:
|
|
return {
|
|
'count': 0,
|
|
'mean': 0,
|
|
'median': 0,
|
|
'stdev': 0,
|
|
'min': 0,
|
|
'max': 0
|
|
}
|
|
|
|
return {
|
|
'count': len(values),
|
|
'mean': statistics.mean(values),
|
|
'median': statistics.median(values),
|
|
'stdev': statistics.stdev(values) if len(values) > 1 else 0,
|
|
'min': min(values),
|
|
'max': max(values)
|
|
}
|
|
|
|
def perform_ttest(
|
|
self,
|
|
variant_a_values: List[float],
|
|
variant_b_values: List[float]
|
|
) -> Tuple[float, float]:
|
|
"""
|
|
Perform independent t-test between two variants.
|
|
|
|
Returns:
|
|
(t_statistic, p_value)
|
|
"""
|
|
if len(variant_a_values) < 2 or len(variant_b_values) < 2:
|
|
return 0.0, 1.0 # Not enough data
|
|
|
|
t_stat, p_value = stats.ttest_ind(variant_a_values, variant_b_values)
|
|
return t_stat, p_value
|
|
|
|
def determine_winner(
|
|
self,
|
|
variant_a_stats: Dict,
|
|
variant_b_stats: Dict,
|
|
p_value: float,
|
|
metric: str,
|
|
lower_is_better: bool = True
|
|
) -> str:
|
|
"""
|
|
Determine winning variant based on statistics.
|
|
|
|
Args:
|
|
variant_a_stats: Statistics for variant A
|
|
variant_b_stats: Statistics for variant B
|
|
p_value: Statistical significance (p-value)
|
|
metric: Metric being compared
|
|
lower_is_better: True if lower values are better (e.g., tokens_used)
|
|
|
|
Returns:
|
|
Winner description
|
|
"""
|
|
# Require statistical significance (p < 0.05)
|
|
if p_value >= 0.05:
|
|
return "No significant difference (p ≥ 0.05)"
|
|
|
|
# Require minimum sample size (20 trials per variant)
|
|
if variant_a_stats['count'] < 20 or variant_b_stats['count'] < 20:
|
|
return f"Insufficient data (need 20 trials, have {variant_a_stats['count']}/{variant_b_stats['count']})"
|
|
|
|
# Compare means
|
|
a_mean = variant_a_stats['mean']
|
|
b_mean = variant_b_stats['mean']
|
|
|
|
if lower_is_better:
|
|
if a_mean < b_mean:
|
|
improvement = ((b_mean - a_mean) / b_mean) * 100
|
|
return f"Variant A wins ({improvement:.1f}% better)"
|
|
else:
|
|
improvement = ((a_mean - b_mean) / a_mean) * 100
|
|
return f"Variant B wins ({improvement:.1f}% better)"
|
|
else:
|
|
if a_mean > b_mean:
|
|
improvement = ((a_mean - b_mean) / b_mean) * 100
|
|
return f"Variant A wins ({improvement:.1f}% better)"
|
|
else:
|
|
improvement = ((b_mean - a_mean) / a_mean) * 100
|
|
return f"Variant B wins ({improvement:.1f}% better)"
|
|
|
|
def generate_recommendation(
|
|
self,
|
|
winner: str,
|
|
variant_a_stats: Dict,
|
|
variant_b_stats: Dict,
|
|
p_value: float
|
|
) -> str:
|
|
"""Generate actionable recommendation"""
|
|
if "No significant difference" in winner:
|
|
return "⚖️ Keep current workflow (no improvement detected)"
|
|
|
|
if "Insufficient data" in winner:
|
|
return "📊 Continue testing (need more trials)"
|
|
|
|
if "Variant A wins" in winner:
|
|
return "✅ Keep Variant A as standard (statistically better)"
|
|
|
|
if "Variant B wins" in winner:
|
|
if variant_b_stats['mean'] > variant_a_stats['mean'] * 0.8: # At least 20% better
|
|
return "🚀 Promote Variant B to standard (significant improvement)"
|
|
else:
|
|
return "⚠️ Marginal improvement - continue testing before promotion"
|
|
|
|
return "🤔 Manual review recommended"
|
|
|
|
def compare_variants(
|
|
self,
|
|
variant_a_id: str,
|
|
variant_b_id: str,
|
|
metric: str = 'tokens_used',
|
|
lower_is_better: bool = True
|
|
) -> str:
|
|
"""
|
|
Compare two workflow variants on a specific metric.
|
|
|
|
Args:
|
|
variant_a_id: Workflow ID for variant A
|
|
variant_b_id: Workflow ID for variant B
|
|
metric: Metric to compare (default: tokens_used)
|
|
lower_is_better: True if lower values are better
|
|
|
|
Returns:
|
|
Comparison report
|
|
"""
|
|
# Get metrics for each variant
|
|
variant_a_metrics = self.get_variant_metrics(variant_a_id)
|
|
variant_b_metrics = self.get_variant_metrics(variant_b_id)
|
|
|
|
if not variant_a_metrics:
|
|
return f"Error: No data for variant A ({variant_a_id})"
|
|
if not variant_b_metrics:
|
|
return f"Error: No data for variant B ({variant_b_id})"
|
|
|
|
# Extract metric values
|
|
a_values = self.extract_metric_values(variant_a_metrics, metric)
|
|
b_values = self.extract_metric_values(variant_b_metrics, metric)
|
|
|
|
# Calculate statistics
|
|
a_stats = self.calculate_statistics(a_values)
|
|
b_stats = self.calculate_statistics(b_values)
|
|
|
|
# Perform t-test
|
|
t_stat, p_value = self.perform_ttest(a_values, b_values)
|
|
|
|
# Determine winner
|
|
winner = self.determine_winner(a_stats, b_stats, p_value, metric, lower_is_better)
|
|
|
|
# Generate recommendation
|
|
recommendation = self.generate_recommendation(winner, a_stats, b_stats, p_value)
|
|
|
|
# Format report
|
|
report = []
|
|
report.append("=" * 80)
|
|
report.append("A/B TEST COMPARISON REPORT")
|
|
report.append("=" * 80)
|
|
report.append("")
|
|
report.append(f"Metric: {metric}")
|
|
report.append(f"Better: {'Lower' if lower_is_better else 'Higher'} values")
|
|
report.append("")
|
|
|
|
report.append(f"## Variant A: {variant_a_id}")
|
|
report.append(f" Trials: {a_stats['count']}")
|
|
report.append(f" Mean: {a_stats['mean']:.2f}")
|
|
report.append(f" Median: {a_stats['median']:.2f}")
|
|
report.append(f" Std Dev: {a_stats['stdev']:.2f}")
|
|
report.append(f" Range: {a_stats['min']:.2f} - {a_stats['max']:.2f}")
|
|
report.append("")
|
|
|
|
report.append(f"## Variant B: {variant_b_id}")
|
|
report.append(f" Trials: {b_stats['count']}")
|
|
report.append(f" Mean: {b_stats['mean']:.2f}")
|
|
report.append(f" Median: {b_stats['median']:.2f}")
|
|
report.append(f" Std Dev: {b_stats['stdev']:.2f}")
|
|
report.append(f" Range: {b_stats['min']:.2f} - {b_stats['max']:.2f}")
|
|
report.append("")
|
|
|
|
report.append("## Statistical Significance")
|
|
report.append(f" t-statistic: {t_stat:.4f}")
|
|
report.append(f" p-value: {p_value:.4f}")
|
|
if p_value < 0.01:
|
|
report.append(" Significance: *** (p < 0.01) - Highly significant")
|
|
elif p_value < 0.05:
|
|
report.append(" Significance: ** (p < 0.05) - Significant")
|
|
elif p_value < 0.10:
|
|
report.append(" Significance: * (p < 0.10) - Marginally significant")
|
|
else:
|
|
report.append(" Significance: n.s. (p ≥ 0.10) - Not significant")
|
|
report.append("")
|
|
|
|
report.append(f"## Result: {winner}")
|
|
report.append(f"## Recommendation: {recommendation}")
|
|
report.append("")
|
|
report.append("=" * 80)
|
|
|
|
return "\n".join(report)
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="A/B test workflow variants")
|
|
parser.add_argument(
|
|
'--variant-a',
|
|
required=True,
|
|
help='Workflow ID for variant A'
|
|
)
|
|
parser.add_argument(
|
|
'--variant-b',
|
|
required=True,
|
|
help='Workflow ID for variant B'
|
|
)
|
|
parser.add_argument(
|
|
'--metric',
|
|
default='tokens_used',
|
|
help='Metric to compare (default: tokens_used)'
|
|
)
|
|
parser.add_argument(
|
|
'--higher-is-better',
|
|
action='store_true',
|
|
help='Higher values are better (default: lower is better)'
|
|
)
|
|
parser.add_argument(
|
|
'--output',
|
|
help='Output file (default: stdout)'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Find metrics file
|
|
metrics_file = Path('docs/memory/workflow_metrics.jsonl')
|
|
|
|
analyzer = ABTestAnalyzer(metrics_file)
|
|
report = analyzer.compare_variants(
|
|
args.variant_a,
|
|
args.variant_b,
|
|
args.metric,
|
|
lower_is_better=not args.higher_is_better
|
|
)
|
|
|
|
if args.output:
|
|
with open(args.output, 'w') as f:
|
|
f.write(report)
|
|
print(f"Report written to {args.output}")
|
|
else:
|
|
print(report)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|