Evaluation-Driven Development for AI Systems: Complete Guide

test_cases = [
    {
        'id': 'refund_basic_01',
        'description': 'Customer within refund window requests refund',
        'input': 'Customer purchased item 5 days ago, wants refund, item unopened',
        'expected_output': 'Approve full refund',
        'acceptance_criteria': [
            'Mentions refund approval',
            'Specifies full amount',
            'References policy window'
        ],
        'priority': 'critical'
    },
    # ... 200 more
]

def generate_response(user_input: str, context: dict) -> str:
    """Generate response using LLM"""
    
    prompt = f"""You are a customer support agent.

Customer query: {user_input}

Context: {json.dumps(context)}

Policy: Refunds approved within 30 days for unopened items.

Response:"""
    
    response = llm.generate(prompt)
    return response

python -m eval_suite.run \
  --dataset spec/test_cases.json \
  --system customer_support \
  --output results/initial_run.json

Pass rate: 67% (134/200)
Critical failures: 8
Avg latency: 1,247ms

failures = [r for r in results if not r.passed]

# Group by failure type
failure_types = {}
for f in failures:
    reason = classify_failure_reason(f)
    if reason not in failure_types:
        failure_types[reason] = []
    failure_types[reason].append(f)

# Prioritize
for reason, fails in sorted(failure_types.items(), key=lambda x: -len(x[1])):
    print(f"{reason}: {len(fails)} failures")

Missing policy context: 23 failures
Incorrect date math: 15 failures
Wrong tone (too casual): 12 failures
Hallucinated information: 8 failures

# Iteration 1: Add policy context to prompt
prompt = f"""You are a customer support agent.

POLICY (follow exactly):
- Refunds approved within 30 days of purchase
- Item must be unopened
- Original packaging required
- Digital items non-refundable

Customer query: {user_input}
..."""

# Re-run evaluation
# Pass rate: 67% → 81%

# Iteration 2: Add date calculation tool
# Pass rate: 81% → 89%

# Iteration 3: Add tone instructions
# Pass rate: 89% → 94%

# Sample 10% of production traffic for eval
async def handle_request(query: str) -> str:
    response = generate_response(query)
    
    # Background eval (async)
    if random.random() < 0.10:
        asyncio.create_task(
            evaluate_and_log(query, response)
        )
    
    return response

# tests/test_calculator_agent.py
def test_agent_can_add_numbers():
    """Agent should use calculator tool for addition"""
    
    query = "What is 127 + 389?"
    
    response = agent.run(query)
    
    # Assertions
    assert response.tool_calls == [
        {'name': 'calculator', 'op': 'add', 'args': [127, 389]}
    ]
    assert "516" in response.text

def run(query: str) -> Response:
    prompt = f"""You have access to a calculator tool.

Query: {query}

Use calculator tool if needed."""
    
    # ... LLM call with tool definitions

def run(query: str) -> Response:
    prompt = f"""You have access to tools. Use them to answer accurately.

Available tools:
- calculator(op, args): Perform math operations

Query: {query}

Think step-by-step:
1. What operation is needed?
2. What are the exact numbers?
3. Call the tool
4. Return the result"""
    
    # ... improved prompting

def test_agent_handles_multiple_operations():
    """Agent should chain multiple calculator calls"""
    query = "What is (100 + 50) * 2?"
    response = agent.run(query)
    assert "300" in response.text

def test_agent_rejects_non_math_queries():
    """Agent should not hallucinate calculator for text queries"""
    query = "What is the capital of France?"
    response = agent.run(query)
    assert response.tool_calls == []
    assert "Paris" in response.text

# .git/hooks/pre-commit
#!/usr/bin/env python3
import subprocess
import sys

def run_smoke_tests():
    """Run critical tests only (~30 seconds)"""
    result = subprocess.run([
        'python', '-m', 'eval_suite.run',
        '--dataset', 'evals/smoke_test.json',  # 20 critical cases
        '--fast'
    ])
    return result.returncode == 0

if __name__ == '__main__':
    print("Running smoke tests...")
    if not run_smoke_tests():
        print("❌ Smoke tests failed. Commit blocked.")
        sys.exit(1)
    print("✅ Smoke tests passed.")
    sys.exit(0)

# .github/workflows/eval.yml
name: Evaluation Suite
on: [pull_request]

jobs:
  evaluate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Run full evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          python -m eval_suite.run \
            --dataset evals/full_suite.json \
            --output results/pr_${{ github.event.pull_request.number }}.json
      
      - name: Check regression
        run: |
          python scripts/check_regression.py \
            --current results/pr_${{ github.event.pull_request.number }}.json \
            --baseline results/production_baseline.json
      
      - name: Comment results
        uses: actions/github-script@v6
        with:
          script: |
            const fs = require('fs');
            const results = JSON.parse(fs.readFileSync('results/summary.json', 'utf8'));
            
            const comment = `## Evaluation Results
            
**Pass Rate:** ${results.pass_rate}
**Latency (p95):** ${results.p95_latency_ms}ms
**Regressions:** ${results.num_regressions}

${results.num_regressions > 0 ? '⚠️ Regressions detected!' : '✅ No regressions'}`;
            
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: comment
            });

# scripts/deployment_gate.py
def can_deploy() -> bool:
    """Check if system meets deployment criteria"""
    
    # Run full eval on staging
    results = run_evaluation(
        dataset='evals/production_suite.json',
        environment='staging'
    )
    
    # Check criteria
    checks = {
        'pass_rate >= 95%': results.pass_rate >= 0.95,
        'no critical failures': results.critical_failures == 0,
        'p95 latency < 2s': results.p95_latency_ms < 2000,
        'cost per request < $0.05': results.avg_cost_usd < 0.05
    }
    
    # Report
    print("Deployment Gate Checks:")
    for check, passed in checks.items():
        status = "✅" if passed else "❌"
        print(f"  {status} {check}")
    
    return all(checks.values())

if __name__ == '__main__':
    if can_deploy():
        print("\n✅ Deployment approved")
        sys.exit(0)
    else:
        print("\n❌ Deployment blocked")
        sys.exit(1)

class ProductionEvaluationMonitor:
    """Monitor production quality continuously"""
    
    def __init__(self, sampling_rate: float = 0.10):
        self.sampling_rate = sampling_rate
        self.metrics = RollingMetrics(window_seconds=3600)
    
    async def monitor_request(
        self,
        query: str,
        response: str,
        metadata: dict
    ):
        """Evaluate production request"""
        
        # Sample traffic
        if random.random() > self.sampling_rate:
            return
        
        # Async evaluation
        score = await self.evaluate(query, response)
        
        self.metrics.add(score)
        
        # Check for quality drop
        if self.metrics.mean() < 0.85:
            alert("Production quality below threshold!")

@dataclass
class QualityGate:
    """Deployment quality gate"""
    name: str
    threshold: float
    current_value: float
    
    @property
    def passed(self) -> bool:
        return self.current_value >= self.threshold

def evaluate_quality_gates(results: EvalResults) -> List[QualityGate]:
    """Evaluate all quality gates"""
    
    return [
        QualityGate(
            name="Overall Pass Rate",
            threshold=0.95,
            current_value=results.pass_rate
        ),
        QualityGate(
            name="Critical Tests",
            threshold=1.0,  # Must be 100%
            current_value=results.critical_pass_rate
        ),
        QualityGate(
            name="P95 Latency",
            threshold=2000.0,  # 2 seconds
            current_value=results.p95_latency_ms
        ),
        QualityGate(
            name="Cost Efficiency",
            threshold=0.05,  # $0.05 per request
            current_value=results.avg_cost_usd
        ),
        QualityGate(
            name="Hallucination Rate",
            threshold=0.02,  # <2%
            current_value=results.hallucination_rate
        )
    ]

def enforce_quality_gates(gates: List[QualityGate]) -> bool:
    """Check if all gates pass"""
    
    all_passed = all(g.passed for g in gates)
    
    print("\n=== Quality Gate Report ===")
    for gate in gates:
        status = "✅ PASS" if gate.passed else "❌ FAIL"
        print(f"{status} {gate.name}: {gate.current_value:.2f} (threshold: {gate.threshold:.2f})")
    
    print(f"\n{'✅ All gates passed' if all_passed else '❌ Some gates failed'}")
    print("=" * 30)
    
    return all_passed

def analyze_metric_correlation(
    leading_metrics: List[float],  # e.g., pass rates
    lagging_metrics: List[float]   # e.g., CSAT scores
) -> float:
    """Measure correlation between eval metrics and user satisfaction"""
    
    from scipy.stats import pearsonr
    
    correlation, p_value = pearsonr(leading_metrics, lagging_metrics)
    
    print(f"Correlation: {correlation:.2f}")
    print(f"P-value: {p_value:.4f}")
    print(f"Statistically significant: {p_value < 0.05}")
    
    return correlation

# Example
pass_rates = [0.95, 0.92, 0.97, 0.89, 0.94]
csat_scores = [0.88, 0.85, 0.90, 0.79, 0.87]

correlation = analyze_metric_correlation(pass_rates, csat_scores)
# Correlation: 0.94 (strong positive correlation)

class FailureCaptureSystem:
    """Capture production failures as test cases"""
    
    def log_failure(
        self,
        query: str,
        response: str,
        failure_type: str,
        user_feedback: Optional[str] = None
    ):
        """Log production failure"""
        
        failure = {
            'id': generate_id(),
            'timestamp': datetime.utcnow().isoformat(),
            'query': query,
            'response': response,
            'failure_type': failure_type,
            'user_feedback': user_feedback
        }
        
        # Save to database
        db.failures.insert_one(failure)
        
        # Auto-create test case
        self.create_test_case_from_failure(failure)
    
    def create_test_case_from_failure(self, failure: dict):
        """Convert failure to test case"""
        
        test_case = {
            'id': f"prod_failure_{failure['id']}",
            'category': 'production_failures',
            'input': failure['query'],
            'expected_output': self.determine_expected_output(failure),
            'tags': ['production', 'regression_prevention', failure['failure_type']],
            'priority': 'high',
            'metadata': {
                'source': 'production',
                'original_failure_id': failure['id'],
                'date_reported': failure['timestamp']
            }
        }
        
        # Add to eval suite
        eval_suite.add_test_case(test_case)
    
    def determine_expected_output(self, failure: dict) -> str:
        """Determine what correct output should have been"""
        
        if failure.get('user_feedback'):
            # User provided correction
            return failure['user_feedback']
        else:
            # Flag for manual review
            return "[NEEDS_MANUAL_REVIEW]"

def weekly_eval_review():
    """Review and update eval suite weekly"""
    
    # 1. Analyze new production failures
    new_failures = get_failures_since(days=7)
    print(f"New failures this week: {len(new_failures)}")
    
    # 2. Group by type
    failure_types = group_by(new_failures, key='failure_type')
    for ftype, failures in sorted(failure_types.items(), key=lambda x: -len(x[1])):
        print(f"  {ftype}: {len(failures)}")
    
    # 3. Add high-frequency failures to eval suite
    for ftype, failures in failure_types.items():
        if len(failures) >= 3:  # Threshold: 3+ occurrences
            # Add representative cases to eval suite
            for failure in failures[:2]:  # Add up to 2 examples
                add_to_eval_suite(failure)
    
    # 4. Remove stale test cases
    # Cases that have passed 100+ times in a row might be redundant
    stale_cases = find_always_passing_cases(min_runs=100)
    print(f"Stale cases (always passing): {len(stale_cases)}")
    # Review manually before removing

def run_ab_test(
    variant_a: System,
    variant_b: System,
    traffic_split: float = 0.5,
    duration_hours: int = 24
) -> Dict[str, any]:
    """Run A/B test with evaluation metrics"""
    
    results = {
        'variant_a': {'requests': 0, 'scores': [], 'latencies': []},
        'variant_b': {'requests': 0, 'scores': [], 'latencies': []}
    }
    
    start_time = time.time()
    
    while time.time() - start_time < duration_hours * 3600:
        # Production traffic
        query = get_next_request()
        
        # Route to variant
        variant = 'variant_a' if random.random() < traffic_split else 'variant_b'
        system = variant_a if variant == 'variant_a' else variant_b
        
        # Generate response
        start = time.time()
        response = system.generate(query)
        latency = (time.time() - start) * 1000
        
        # Evaluate (async)
        score = evaluate(query, response)
        
        # Record metrics
        results[variant]['requests'] += 1
        results[variant]['scores'].append(score)
        results[variant]['latencies'].append(latency)
    
    # Analyze results
    return {
        'variant_a': {
            'pass_rate': sum(1 for s in results['variant_a']['scores'] if s >= 0.7) / results['variant_a']['requests'],
            'avg_latency': np.mean(results['variant_a']['latencies']),
            'requests': results['variant_a']['requests']
        },
        'variant_b': {
            'pass_rate': sum(1 for s in results['variant_b']['scores'] if s >= 0.7) / results['variant_b']['requests'],
            'avg_latency': np.mean(results['variant_b']['latencies']),
            'requests': results['variant_b']['requests']
        }
    }

test_case = {
    'turns': [
        {'user': 'What are your hours?', 'expected': 'mention business hours'},
        {'user': 'Are you open on weekends?', 'expected': 'Saturday/Sunday hours'},
        {'user': 'Thanks', 'expected': 'acknowledge gratitude'}
    ]
}