mirror of
https://github.com/SuperClaude-Org/SuperClaude_Framework.git
synced 2025-12-29 16:16:08 +00:00
refactor: consolidate PM Agent optimization and pending changes
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>
This commit is contained in:
kazuki
309
scripts/ab_test_workflows.py
Executable file
309
scripts/ab_test_workflows.py
Executable file
@@ -0,0 +1,309 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user