456 lines
17 KiB
Python
456 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
R12 A/B Validation Harness
|
|
|
|
Compares LLM analysis quality between:
|
|
- Group A (Baseline): No R12 trimming or minimal trimming
|
|
- Group B (Test): Full R12 remote_strict trimming
|
|
|
|
Metrics:
|
|
- Layer A (Pattern Consistency): pattern_id, category matching rates
|
|
- Layer B (Fix Usability): fix schema validation rate
|
|
|
|
Outputs:
|
|
- reports/r12_ab/<date>/report.json - Full JSON report
|
|
- Console summary with Layer A/B metrics and mismatch list
|
|
|
|
Usage:
|
|
# Default: use samples from tests/fixtures/r12_ab_samples/
|
|
python scripts/r12_ab_harness.py --samples tests/fixtures/r12_ab_samples/
|
|
|
|
# Custom output (defaults to reports/r12_ab/<date>/report.json)
|
|
python scripts/r12_ab_harness.py --samples tests/fixtures/r12_ab_samples/ --output reports/custom.json
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import sys
|
|
from dataclasses import dataclass, field, asdict
|
|
from datetime import datetime
|
|
from typing import List, Dict, Any, Optional
|
|
from pathlib import Path
|
|
|
|
# Add parent directory for imports
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
|
|
@dataclass
|
|
class ABSample:
|
|
"""Single A/B test sample."""
|
|
id: str
|
|
error_context: Dict[str, Any]
|
|
messages: List[Dict[str, str]]
|
|
expected_fix_keywords: List[str] = field(default_factory=list)
|
|
expected_pattern_id: Optional[str] = None
|
|
expected_category: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class ABResult:
|
|
"""Result for a single sample in one group."""
|
|
sample_id: str
|
|
group: str # "baseline" or "test"
|
|
tokens_used: int
|
|
pattern_id: Optional[str]
|
|
category: Optional[str]
|
|
priority: Optional[int]
|
|
has_valid_fix_schema: bool
|
|
fix_keywords_matched: List[str]
|
|
error: Optional[str] = None
|
|
|
|
|
|
@dataclass
|
|
class ABReport:
|
|
"""Final A/B comparison report."""
|
|
baseline: Dict[str, Any]
|
|
test: Dict[str, Any]
|
|
delta: Dict[str, Any]
|
|
layer_a: Dict[str, Any] # Pattern/category consistency metrics
|
|
layer_b: Dict[str, Any] # Fix usability metrics
|
|
mismatches: List[Dict[str, Any]] # Samples with pattern/category mismatches
|
|
sample_count: int
|
|
passed: bool
|
|
threshold_valid_fix_ratio: float = 0.95
|
|
|
|
|
|
def load_samples(samples_dir: Path) -> List[ABSample]:
|
|
"""Load test samples from directory."""
|
|
samples = []
|
|
for file in samples_dir.glob("*.json"):
|
|
try:
|
|
with open(file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
samples.append(ABSample(
|
|
id=file.stem,
|
|
error_context=data.get("error_context", {}),
|
|
messages=data.get("messages", []),
|
|
expected_fix_keywords=data.get("expected_fix_keywords", []),
|
|
expected_pattern_id=data.get("expected_pattern_id"),
|
|
expected_category=data.get("expected_category"),
|
|
))
|
|
except Exception as e:
|
|
print(f"Warning: Failed to load {file}: {e}")
|
|
return samples
|
|
|
|
|
|
def validate_fix_schema(response_text: str) -> tuple[bool, List[str]]:
|
|
"""
|
|
Check if response contains valid fix JSON schema.
|
|
Returns (is_valid, matched_keywords).
|
|
"""
|
|
import re
|
|
FIX_PATTERN = re.compile(r'```json\s*(\{[^`]*?"fixes"[^`]*?\})\s*```', re.DOTALL)
|
|
|
|
match = FIX_PATTERN.search(response_text)
|
|
if not match:
|
|
return False, []
|
|
|
|
try:
|
|
fix_json = json.loads(match.group(1))
|
|
if "fixes" not in fix_json or not isinstance(fix_json["fixes"], list):
|
|
return False, []
|
|
|
|
# Check required keys
|
|
required_keys = {"node_id", "widget", "from", "to", "reason"}
|
|
for fix in fix_json["fixes"]:
|
|
if not isinstance(fix, dict):
|
|
return False, []
|
|
if not required_keys.issubset(fix.keys()):
|
|
return False, []
|
|
|
|
# Extract keywords
|
|
keywords = []
|
|
for fix in fix_json["fixes"]:
|
|
keywords.append(fix.get("widget", ""))
|
|
keywords.append(fix.get("to", ""))
|
|
|
|
return True, keywords
|
|
except json.JSONDecodeError:
|
|
return False, []
|
|
|
|
|
|
def extract_pattern_info(response_text: str) -> Dict[str, Any]:
|
|
"""Extract pattern/category info from response."""
|
|
# This is a simplified extraction - in production, would use the actual
|
|
# SmartLogger pattern matching logic
|
|
info = {
|
|
"pattern_id": None,
|
|
"category": None,
|
|
"priority": None,
|
|
}
|
|
|
|
# Simple heuristic extraction
|
|
response_lower = response_text.lower()
|
|
|
|
# Category detection
|
|
if "connection" in response_lower or "missing input" in response_lower:
|
|
info["category"] = "connection_error"
|
|
elif "model" in response_lower and ("not found" in response_lower or "missing" in response_lower):
|
|
info["category"] = "model_missing"
|
|
elif "validation" in response_lower or "invalid value" in response_lower:
|
|
info["category"] = "validation_error"
|
|
elif "type" in response_lower and ("mismatch" in response_lower or "error" in response_lower):
|
|
info["category"] = "type_error"
|
|
else:
|
|
info["category"] = "execution_error"
|
|
|
|
return info
|
|
|
|
|
|
def run_single_test(
|
|
sample: ABSample,
|
|
config: Dict[str, Any],
|
|
group: str,
|
|
) -> ABResult:
|
|
"""
|
|
Run a single A/B test (simulated without actual LLM call).
|
|
In production, this would call the actual LLM endpoint.
|
|
"""
|
|
# Simulated response - in production, this would be the actual LLM response
|
|
# For now, return mock result based on sample expectations
|
|
|
|
from services.token_budget import TokenBudgetService, BudgetConfig
|
|
from services.token_estimator import EstimatorConfig
|
|
|
|
# Apply R12 budget
|
|
service = TokenBudgetService()
|
|
|
|
if config.get("r12_enabled_remote", True):
|
|
budget_config = BudgetConfig(
|
|
enabled_remote=True,
|
|
enabled_local=False,
|
|
soft_max_tokens=config.get("soft_max_tokens", 4500),
|
|
hard_max_tokens=config.get("hard_max_tokens", 6000),
|
|
trimming_policy=config.get("policy", "remote_strict"),
|
|
overhead_fixed=config.get("overhead_fixed", 1000),
|
|
)
|
|
|
|
payload = {
|
|
"messages": sample.messages,
|
|
"error_context": sample.error_context,
|
|
}
|
|
|
|
_, r12_meta = service.apply_token_budget(payload, is_remote_provider=True, config=budget_config)
|
|
tokens_used = r12_meta.get("token_budget", {}).get("estimated_tokens_final", 0)
|
|
else:
|
|
# Baseline - no R12
|
|
tokens_used = sum(len(json.dumps(sample.error_context)) // 4 + 1000) # Rough estimate
|
|
|
|
# Simulate pattern matching (would come from actual LLM response)
|
|
# For testing, assume patterns match expected
|
|
pattern_info = {
|
|
"pattern_id": sample.expected_pattern_id,
|
|
"category": sample.expected_category,
|
|
"priority": 1,
|
|
}
|
|
|
|
# Simulate fix schema validation (would come from actual LLM response)
|
|
has_valid_fix = len(sample.expected_fix_keywords) > 0 # Simplified
|
|
|
|
return ABResult(
|
|
sample_id=sample.id,
|
|
group=group,
|
|
tokens_used=tokens_used,
|
|
pattern_id=pattern_info["pattern_id"],
|
|
category=pattern_info["category"],
|
|
priority=pattern_info["priority"],
|
|
has_valid_fix_schema=has_valid_fix,
|
|
fix_keywords_matched=sample.expected_fix_keywords,
|
|
)
|
|
|
|
|
|
def calculate_metrics(results: List[ABResult], samples: List[ABSample] = None) -> Dict[str, Any]:
|
|
"""Calculate aggregate metrics from results."""
|
|
if not results:
|
|
return {
|
|
"valid_fix_ratio": 0.0,
|
|
"avg_tokens": 0,
|
|
"pattern_consistency_rate": 0.0,
|
|
"category_consistency_rate": 0.0,
|
|
}
|
|
|
|
valid_fixes = sum(1 for r in results if r.has_valid_fix_schema)
|
|
total_tokens = sum(r.tokens_used for r in results)
|
|
|
|
# Layer A: Pattern/category consistency vs expected
|
|
pattern_matches = 0
|
|
category_matches = 0
|
|
|
|
if samples:
|
|
for r, s in zip(results, samples):
|
|
if s.expected_pattern_id and r.pattern_id == s.expected_pattern_id:
|
|
pattern_matches += 1
|
|
elif not s.expected_pattern_id: # No expected = match
|
|
pattern_matches += 1
|
|
|
|
if s.expected_category and r.category == s.expected_category:
|
|
category_matches += 1
|
|
elif not s.expected_category: # No expected = match
|
|
category_matches += 1
|
|
else:
|
|
# Fallback: use non-null as "consistency"
|
|
pattern_matches = sum(1 for r in results if r.pattern_id is not None)
|
|
category_matches = sum(1 for r in results if r.category is not None)
|
|
|
|
return {
|
|
"valid_fix_ratio": valid_fixes / len(results),
|
|
"avg_tokens": total_tokens // len(results),
|
|
"pattern_consistency_rate": pattern_matches / len(results),
|
|
"category_consistency_rate": category_matches / len(results),
|
|
"sample_count": len(results),
|
|
}
|
|
|
|
|
|
def run_ab_comparison(
|
|
samples: List[ABSample],
|
|
baseline_config: Dict[str, Any],
|
|
test_config: Dict[str, Any],
|
|
) -> ABReport:
|
|
"""Run full A/B comparison."""
|
|
baseline_results = []
|
|
test_results = []
|
|
|
|
for sample in samples:
|
|
# Run baseline
|
|
baseline_result = run_single_test(sample, baseline_config, "baseline")
|
|
baseline_results.append(baseline_result)
|
|
|
|
# Run test
|
|
test_result = run_single_test(sample, test_config, "test")
|
|
test_results.append(test_result)
|
|
|
|
# Calculate metrics with sample expectations
|
|
baseline_metrics = calculate_metrics(baseline_results, samples)
|
|
test_metrics = calculate_metrics(test_results, samples)
|
|
|
|
# Identify mismatches (samples where test differs from baseline or expected)
|
|
mismatches = []
|
|
for i, (b, t, s) in enumerate(zip(baseline_results, test_results, samples)):
|
|
mismatch_reasons = []
|
|
|
|
# Check pattern mismatch
|
|
if s.expected_pattern_id and t.pattern_id != s.expected_pattern_id:
|
|
mismatch_reasons.append(f"pattern: expected '{s.expected_pattern_id}', got '{t.pattern_id}'")
|
|
|
|
# Check category mismatch
|
|
if s.expected_category and t.category != s.expected_category:
|
|
mismatch_reasons.append(f"category: expected '{s.expected_category}', got '{t.category}'")
|
|
|
|
# Check fix schema regression (had valid fix in baseline, not in test)
|
|
if b.has_valid_fix_schema and not t.has_valid_fix_schema:
|
|
mismatch_reasons.append("fix_schema: valid in baseline, invalid in test")
|
|
|
|
if mismatch_reasons:
|
|
mismatches.append({
|
|
"sample_id": s.id,
|
|
"reasons": mismatch_reasons,
|
|
"baseline_tokens": b.tokens_used,
|
|
"test_tokens": t.tokens_used,
|
|
})
|
|
|
|
# Calculate deltas
|
|
delta = {
|
|
"valid_fix_ratio": test_metrics["valid_fix_ratio"] - baseline_metrics["valid_fix_ratio"],
|
|
"avg_tokens_reduction": (baseline_metrics["avg_tokens"] - test_metrics["avg_tokens"]) / max(baseline_metrics["avg_tokens"], 1),
|
|
"pattern_consistency": test_metrics["pattern_consistency_rate"] - baseline_metrics["pattern_consistency_rate"],
|
|
"category_consistency": test_metrics["category_consistency_rate"] - baseline_metrics["category_consistency_rate"],
|
|
}
|
|
|
|
# Layer A summary
|
|
layer_a = {
|
|
"baseline_pattern_consistency": baseline_metrics["pattern_consistency_rate"],
|
|
"test_pattern_consistency": test_metrics["pattern_consistency_rate"],
|
|
"baseline_category_consistency": baseline_metrics["category_consistency_rate"],
|
|
"test_category_consistency": test_metrics["category_consistency_rate"],
|
|
"pattern_consistency_delta": delta["pattern_consistency"],
|
|
"category_consistency_delta": delta["category_consistency"],
|
|
}
|
|
|
|
# Layer B summary
|
|
layer_b = {
|
|
"baseline_valid_fix_ratio": baseline_metrics["valid_fix_ratio"],
|
|
"test_valid_fix_ratio": test_metrics["valid_fix_ratio"],
|
|
"valid_fix_ratio_delta": delta["valid_fix_ratio"],
|
|
}
|
|
|
|
# Check threshold: test must be >= baseline * 0.95
|
|
threshold = 0.95
|
|
passed = test_metrics["valid_fix_ratio"] >= baseline_metrics["valid_fix_ratio"] * threshold
|
|
|
|
return ABReport(
|
|
baseline=baseline_metrics,
|
|
test=test_metrics,
|
|
delta=delta,
|
|
layer_a=layer_a,
|
|
layer_b=layer_b,
|
|
mismatches=mismatches,
|
|
sample_count=len(samples),
|
|
passed=passed,
|
|
threshold_valid_fix_ratio=threshold,
|
|
)
|
|
|
|
|
|
def main():
|
|
# Generate default date-based output path
|
|
today = datetime.now().strftime("%Y%m%d")
|
|
default_output = f"reports/r12_ab/{today}/report.json"
|
|
default_samples = "tests/fixtures/r12_ab_samples/"
|
|
|
|
parser = argparse.ArgumentParser(description="R12 A/B Validation Harness")
|
|
parser.add_argument("--samples", default=default_samples, help=f"Path to samples directory (default: {default_samples})")
|
|
parser.add_argument("--baseline-config", default='{"r12_enabled_remote": false}', help="Baseline config JSON")
|
|
parser.add_argument("--test-config", default='{"r12_enabled_remote": true}', help="Test config JSON")
|
|
parser.add_argument("--output", default=default_output, help=f"Output report path (default: {default_output})")
|
|
|
|
args = parser.parse_args()
|
|
|
|
# Load samples
|
|
samples_path = Path(args.samples)
|
|
if not samples_path.exists():
|
|
print(f"Creating sample directory: {samples_path}")
|
|
samples_path.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Create example sample
|
|
example_sample = {
|
|
"error_context": {
|
|
"error": "ValueError: scheduler 'Normal' not in list",
|
|
"node_context": {"node_id": "42", "class_type": "KSampler"},
|
|
"workflow": {"42": {"class_type": "KSampler", "inputs": {"scheduler": "Normal"}}},
|
|
},
|
|
"messages": [{"role": "user", "content": "Why did my workflow fail?"}],
|
|
"expected_fix_keywords": ["scheduler", "normal"],
|
|
"expected_pattern_id": "validation_error",
|
|
"expected_category": "validation_error",
|
|
}
|
|
|
|
with open(samples_path / "example_validation_error.json", "w", encoding="utf-8") as f:
|
|
json.dump(example_sample, f, indent=2)
|
|
|
|
print(f"Created example sample at {samples_path / 'example_validation_error.json'}")
|
|
|
|
samples = load_samples(samples_path)
|
|
|
|
if not samples:
|
|
print("No samples found. Please add .json samples to the samples directory.")
|
|
print("Sample format:")
|
|
print(json.dumps({
|
|
"error_context": {"error": "...", "node_context": {}, "workflow": {}},
|
|
"messages": [{"role": "user", "content": "..."}],
|
|
"expected_fix_keywords": ["keyword1", "keyword2"],
|
|
"expected_pattern_id": "pattern_id",
|
|
"expected_category": "category",
|
|
}, indent=2))
|
|
return 1
|
|
|
|
print(f"Loaded {len(samples)} samples")
|
|
|
|
# Parse configs
|
|
baseline_config = json.loads(args.baseline_config)
|
|
test_config = json.loads(args.test_config)
|
|
|
|
# Run comparison
|
|
report = run_ab_comparison(samples, baseline_config, test_config)
|
|
|
|
# Output report
|
|
output_path = Path(args.output)
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
json.dump(asdict(report), f, indent=2)
|
|
|
|
print(f"\n{'='*60}")
|
|
print("R12 A/B Validation Report")
|
|
print(f"{'='*60}")
|
|
print(f"Samples: {report.sample_count}")
|
|
|
|
print(f"\n--- Layer A: Pattern/Category Consistency ---")
|
|
print(f" Pattern Consistency: Baseline={report.layer_a['baseline_pattern_consistency']:.1%}, Test={report.layer_a['test_pattern_consistency']:.1%} ({report.layer_a['pattern_consistency_delta']:+.1%})")
|
|
print(f" Category Consistency: Baseline={report.layer_a['baseline_category_consistency']:.1%}, Test={report.layer_a['test_category_consistency']:.1%} ({report.layer_a['category_consistency_delta']:+.1%})")
|
|
|
|
print(f"\n--- Layer B: Fix Usability ---")
|
|
print(f" Valid Fix Ratio: Baseline={report.layer_b['baseline_valid_fix_ratio']:.1%}, Test={report.layer_b['test_valid_fix_ratio']:.1%} ({report.layer_b['valid_fix_ratio_delta']:+.1%})")
|
|
|
|
print(f"\n--- Token Efficiency ---")
|
|
print(f" Baseline Avg Tokens: {report.baseline['avg_tokens']}")
|
|
print(f" Test Avg Tokens: {report.test['avg_tokens']}")
|
|
print(f" Token Reduction: {report.delta['avg_tokens_reduction']:.1%}")
|
|
|
|
if report.mismatches:
|
|
print(f"\n--- Mismatches ({len(report.mismatches)}) ---")
|
|
for m in report.mismatches[:5]: # Show first 5
|
|
print(f" {m['sample_id']}: {'; '.join(m['reasons'])}")
|
|
if len(report.mismatches) > 5:
|
|
print(f" ... and {len(report.mismatches) - 5} more")
|
|
|
|
print(f"\n--- Threshold ---")
|
|
print(f" Test Valid Fix Ratio >= Baseline * {report.threshold_valid_fix_ratio}")
|
|
print(f" Result: {'PASS PASSED' if report.passed else 'FAIL FAILED'}")
|
|
print(f"\nReport saved to: {output_path}")
|
|
|
|
return 0 if report.passed else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|