AI Evals in CI/CD with GitHub Actions: Integration Guide

"""
.
├── .github/
│   └── workflows/
│       ├── eval-on-pr.yml          # Run on PRs
│       ├── eval-nightly.yml        # Comprehensive nightly
│       └── deploy-production.yml   # Deploy with gates
├── evals/
│   ├── __init__.py
│   ├── test_accuracy.py            # Accuracy evals
│   ├── test_hallucination.py       # Hallucination detection
│   ├── test_performance.py         # Latency benchmarks
│   └── datasets/
│       ├── accuracy_v1.jsonl       # Versioned eval data
│       └── edge_cases_v1.jsonl
├── src/
│   └── ai_system.py                # System under test
├── requirements.txt
└── pyproject.toml
"""

# evals/test_accuracy.py
import pytest
import json
from pathlib import Path
from typing import List, Dict, Any
from openai import AsyncOpenAI

client = AsyncOpenAI()

@pytest.fixture
def eval_dataset() -> List[Dict[str, Any]]:
    """Load evaluation dataset."""
    dataset_path = Path(__file__).parent / "datasets" / "accuracy_v1.jsonl"
    
    examples = []
    with dataset_path.open("r") as f:
        for line in f:
            examples.append(json.loads(line))
    
    return examples

@pytest.fixture
def baseline_metrics() -> Dict[str, float]:
    """Load baseline metrics from previous run."""
    baseline_path = Path(__file__).parent / "baselines" / "accuracy_baseline.json"
    
    if not baseline_path.exists():
        return {}
    
    with baseline_path.open("r") as f:
        return json.load(f)

@pytest.mark.asyncio
async def test_accuracy_threshold(eval_dataset, baseline_metrics):
    """Test accuracy meets or exceeds threshold."""
    from src.ai_system import AISystem
    
    system = AISystem()
    
    correct = 0
    total = len(eval_dataset)
    
    for example in eval_dataset:
        response = await system.query(example["input"])
        
        # Check if response matches expected output
        if _is_correct(response, example["expected_output"]):
            correct += 1
    
    accuracy = correct / total
    
    # Assert meets minimum threshold
    MIN_ACCURACY = 0.85
    assert accuracy >= MIN_ACCURACY, f"Accuracy {accuracy:.2%} below threshold {MIN_ACCURACY:.2%}"
    
    # Check for regression
    if baseline_metrics.get("accuracy"):
        baseline_accuracy = baseline_metrics["accuracy"]
        MAX_REGRESSION = 0.03  # 3% max drop
        
        assert accuracy >= baseline_accuracy - MAX_REGRESSION, (
            f"Accuracy regressed by {baseline_accuracy - accuracy:.2%} "
            f"(baseline: {baseline_accuracy:.2%}, current: {accuracy:.2%})"
        )
    
    print(f"✓ Accuracy: {accuracy:.2%} ({correct}/{total})")

def _is_correct(response: str, expected: str) -> bool:
    """Check if response is correct."""
    # Implement domain-specific correctness check
    return response.strip().lower() == expected.strip().lower()

@pytest.mark.asyncio
async def test_hallucination_rate(eval_dataset):
    """Test hallucination rate below threshold."""
    from src.ai_system import AISystem
    
    system = AISystem()
    
    hallucinations = 0
    total = len(eval_dataset)
    
    for example in eval_dataset:
        if "context" not in example:
            continue
        
        response = await system.query(
            example["input"],
            context=example["context"],
        )
        
        # Check if response is grounded in context
        is_grounded = await _check_grounding(response, example["context"])
        
        if not is_grounded:
            hallucinations += 1
    
    hallucination_rate = hallucinations / total
    
    MAX_HALLUCINATION_RATE = 0.02  # 2% max
    assert hallucination_rate <= MAX_HALLUCINATION_RATE, (
        f"Hallucination rate {hallucination_rate:.2%} exceeds threshold {MAX_HALLUCINATION_RATE:.2%}"
    )
    
    print(f"✓ Hallucination rate: {hallucination_rate:.2%} ({hallucinations}/{total})")

async def _check_grounding(response: str, context: str) -> bool:
    """Check if response is grounded in context."""
    prompt = f"""Is this response grounded in the context?

Response: {response}
Context: {context}

Return JSON: {{"grounded": true/false}}"""
    
    result = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        response_format={"type": "json_object"},
    )
    
    return json.loads(result.choices[0].message.content)["grounded"]

@pytest.mark.asyncio
async def test_latency_p95(eval_dataset):
    """Test P95 latency below threshold."""
    from src.ai_system import AISystem
    import time
    
    system = AISystem()
    
    latencies = []
    
    for example in eval_dataset[:50]:  # Sample for speed
        start = time.perf_counter()
        await system.query(example["input"])
        latency_ms = (time.perf_counter() - start) * 1000
        
        latencies.append(latency_ms)
    
    import numpy as np
    p95_latency = np.percentile(latencies, 95)
    
    MAX_P95_LATENCY = 2000  # 2 seconds
    assert p95_latency <= MAX_P95_LATENCY, (
        f"P95 latency {p95_latency:.0f}ms exceeds threshold {MAX_P95_LATENCY}ms"
    )
    
    print(f"✓ P95 latency: {p95_latency:.0f}ms")

# .github/workflows/eval-on-pr.yml
name: AI Evaluation on PR

on:
  pull_request:
    branches: [main, develop]
    paths:
      - 'src/**'
      - 'prompts/**'
      - 'evals/**'

env:
  OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
  EVAL_DATASET_VERSION: v1.0.0

jobs:
  run-evals:
    runs-on: ubuntu-latest
    timeout-minutes: 15
    
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          cache: 'pip'
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-asyncio pytest-html
      
      - name: Download eval datasets
        run: |
          # Download versioned datasets from storage
          aws s3 cp s3://my-bucket/eval-datasets/$EVAL_DATASET_VERSION/ evals/datasets/ --recursive
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
      
      - name: Run accuracy tests
        id: accuracy
        run: |
          pytest evals/test_accuracy.py -v --html=report.html --self-contained-html
        continue-on-error: true
      
      - name: Run hallucination tests
        id: hallucination
        run: |
          pytest evals/test_hallucination.py -v
        continue-on-error: true
      
      - name: Run performance tests
        id: performance
        run: |
          pytest evals/test_performance.py -v
        continue-on-error: true
      
      - name: Generate summary
        if: always()
        run: |
          echo "## AI Evaluation Results" >> $GITHUB_STEP_SUMMARY
          echo "" >> $GITHUB_STEP_SUMMARY
          
          if [ "${{ steps.accuracy.outcome }}" == "success" ]; then
            echo "✅ Accuracy tests passed" >> $GITHUB_STEP_SUMMARY
          else
            echo "❌ Accuracy tests failed" >> $GITHUB_STEP_SUMMARY
          fi
          
          if [ "${{ steps.hallucination.outcome }}" == "success" ]; then
            echo "✅ Hallucination tests passed" >> $GITHUB_STEP_SUMMARY
          else
            echo "❌ Hallucination tests failed" >> $GITHUB_STEP_SUMMARY
          fi
          
          if [ "${{ steps.performance.outcome }}" == "success" ]; then
            echo "✅ Performance tests passed" >> $GITHUB_STEP_SUMMARY
          else
            echo "❌ Performance tests failed" >> $GITHUB_STEP_SUMMARY
          fi
      
      - name: Upload test report
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: eval-report
          path: report.html
      
      - name: Comment on PR
        if: always() && github.event_name == 'pull_request'
        uses: actions/github-script@v7
        with:
          script: |
            const fs = require('fs');
            
            let comment = '## 🤖 AI Evaluation Results\n\n';
            comment += `**Accuracy**: ${{ steps.accuracy.outcome }}\n`;
            comment += `**Hallucination**: ${{ steps.hallucination.outcome }}\n`;
            comment += `**Performance**: ${{ steps.performance.outcome }}\n\n`;
            comment += '[View detailed report](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})\n';
            
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: comment
            });
      
      - name: Fail if any test failed
        if: steps.accuracy.outcome != 'success' || steps.hallucination.outcome != 'success' || steps.performance.outcome != 'success'
        run: exit 1

# .github/workflows/eval-nightly.yml
name: Nightly Comprehensive Evaluation

on:
  schedule:
    - cron: '0 2 * * *'  # 2 AM UTC daily
  workflow_dispatch:

env:
  OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}

jobs:
  comprehensive-eval:
    runs-on: ubuntu-latest
    timeout-minutes: 60
    
    steps:
      - name: Checkout code
        uses: actions/checkout@v4
      
      - name: Set up Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
      
      - name: Install dependencies
        run: pip install -r requirements.txt
      
      - name: Run full eval suite
        run: |
          pytest evals/ -v --html=nightly-report.html --self-contained-html
      
      - name: Upload metrics to monitoring
        run: |
          python scripts/upload_metrics.py nightly-report.html
        env:
          DATADOG_API_KEY: ${{ secrets.DATADOG_API_KEY }}
      
      - name: Alert on failure
        if: failure()
        uses: slackapi/slack-github-action@v1
        with:
          payload: |
            {
              "text": "🚨 Nightly AI evals failed!",
              "blocks": [
                {
                  "type": "section",
                  "text": {
                    "type": "mrkdwn",
                    "text": "Nightly evaluation suite failed. [View run](https://github.com/${{ github.repository }}/actions/runs/${{ github.run_id }})"
                  }
                }
              ]
            }
        env:
          SLACK_WEBHOOK_URL: ${{ secrets.SLACK_WEBHOOK_URL }}

# evals/test_comprehensive.py
import pytest
from typing import List, Dict, Any
import asyncio

class ComprehensiveEvalSuite:
    """Comprehensive evaluation test suite."""
    
    def __init__(self, system):
        self.system = system
    
    async def run_all_evals(
        self,
        eval_dataset: List[Dict[str, Any]],
    ) -> Dict[str, Any]:
        """Run all evaluation tests."""
        results = await asyncio.gather(
            self.eval_accuracy(eval_dataset),
            self.eval_hallucination(eval_dataset),
            self.eval_safety(eval_dataset),
            self.eval_consistency(eval_dataset),
            self.eval_latency(eval_dataset),
        )
        
        return {
            "accuracy": results[0],
            "hallucination": results[1],
            "safety": results[2],
            "consistency": results[3],
            "latency": results[4],
        }
    
    async def eval_accuracy(
        self,
        dataset: List[Dict[str, Any]],
    ) -> Dict[str, float]:
        """Evaluate accuracy metrics."""
        correct = 0
        total = len(dataset)
        
        for example in dataset:
            response = await self.system.query(example["input"])
            
            if self._is_correct(response, example["expected_output"]):
                correct += 1
        
        accuracy = correct / total
        
        return {
            "accuracy": accuracy,
            "correct": correct,
            "total": total,
        }
    
    async def eval_hallucination(
        self,
        dataset: List[Dict[str, Any]],
    ) -> Dict[str, float]:
        """Evaluate hallucination rate."""
        from evals.hallucination_detector import HallucinationDetector
        
        detector = HallucinationDetector()
        hallucinations = 0
        total = 0
        
        for example in dataset:
            if "context" not in example:
                continue
            
            response = await self.system.query(
                example["input"],
                context=example["context"],
            )
            
            result = await detector.detect(
                query=example["input"],
                response=response,
                context=example["context"],
            )
            
            if result.is_hallucination:
                hallucinations += 1
            
            total += 1
        
        rate = hallucinations / total if total > 0 else 0
        
        return {
            "hallucination_rate": rate,
            "hallucinations": hallucinations,
            "total": total,
        }
    
    async def eval_safety(
        self,
        dataset: List[Dict[str, Any]],
    ) -> Dict[str, float]:
        """Evaluate safety metrics."""
        unsafe = 0
        total = len(dataset)
        
        for example in dataset:
            if "safety_test" not in example:
                continue
            
            response = await self.system.query(example["input"])
            
            # Check if response is safe
            is_safe = await self._check_safety(response)
            
            if not is_safe:
                unsafe += 1
        
        safety_rate = 1 - (unsafe / total) if total > 0 else 1.0
        
        return {
            "safety_rate": safety_rate,
            "unsafe_responses": unsafe,
            "total": total,
        }
    
    async def eval_consistency(
        self,
        dataset: List[Dict[str, Any]],
    ) -> Dict[str, float]:
        """Evaluate response consistency."""
        from evals.consistency_checker import ConsistencyChecker
        
        checker = ConsistencyChecker()
        consistency_scores = []
        
        for example in dataset[:50]:  # Sample for speed
            score = await checker.check_consistency(
                query=example["input"],
                system=self.system,
            )
            consistency_scores.append(score)
        
        avg_consistency = sum(consistency_scores) / len(consistency_scores)
        
        return {
            "average_consistency": avg_consistency,
            "num_samples": len(consistency_scores),
        }
    
    async def eval_latency(
        self,
        dataset: List[Dict[str, Any]],
    ) -> Dict[str, float]:
        """Evaluate latency metrics."""
        import time
        import numpy as np
        
        latencies = []
        
        for example in dataset[:100]:  # Sample for speed
            start = time.perf_counter()
            await self.system.query(example["input"])
            latency_ms = (time.perf_counter() - start) * 1000
            
            latencies.append(latency_ms)
        
        return {
            "p50_latency_ms": float(np.percentile(latencies, 50)),
            "p95_latency_ms": float(np.percentile(latencies, 95)),
            "p99_latency_ms": float(np.percentile(latencies, 99)),
            "avg_latency_ms": float(np.mean(latencies)),
        }
    
    def _is_correct(self, response: str, expected: str) -> bool:
        """Check correctness."""
        # Domain-specific logic
        return True
    
    async def _check_safety(self, response: str) -> bool:
        """Check safety."""
        # Implement safety checks
        return True

@pytest.mark.asyncio
async def test_comprehensive_eval_suite():
    """Run comprehensive eval suite with assertions."""
    from src.ai_system import AISystem
    import json
    
    system = AISystem()
    suite = ComprehensiveEvalSuite(system)
    
    # Load dataset
    dataset_path = Path(__file__).parent / "datasets" / "accuracy_v1.jsonl"
    dataset = []
    with dataset_path.open("r") as f:
        for line in f:
            dataset.append(json.loads(line))
    
    # Run all evals
    results = await suite.run_all_evals(dataset)
    
    # Assert thresholds
    assert results["accuracy"]["accuracy"] >= 0.85, "Accuracy below 85%"
    assert results["hallucination"]["hallucination_rate"] <= 0.02, "Hallucination rate above 2%"
    assert results["safety"]["safety_rate"] >= 0.98, "Safety rate below 98%"
    assert results["consistency"]["average_consistency"] >= 0.80, "Consistency below 80%"
    assert results["latency"]["p95_latency_ms"] <= 2000, "P95 latency above 2s"
    
    print("\n✓ All eval metrics passed:")
    print(f"  Accuracy: {results['accuracy']['accuracy']:.2%}")
    print(f"  Hallucination: {results['hallucination']['hallucination_rate']:.2%}")
    print(f"  Safety: {results['safety']['safety_rate']:.2%}")
    print(f"  Consistency: {results['consistency']['average_consistency']:.2%}")
    print(f"  P95 Latency: {results['latency']['p95_latency_ms']:.0f}ms")

# scripts/quality_gate.py
from typing import Dict, Any, List
from dataclasses import dataclass

@dataclass
class QualityThreshold:
    """Quality threshold definition."""
    metric: str
    operator: str  # >=, <=, ==
    threshold: float
    severity: str  # blocking, warning

class QualityGate:
    """Quality gate with configurable thresholds."""
    
    def __init__(self, thresholds: List[QualityThreshold]):
        self.thresholds = thresholds
    
    def evaluate(
        self,
        metrics: Dict[str, float],
    ) -> Dict[str, Any]:
        """Evaluate metrics against thresholds."""
        violations = []
        warnings = []
        
        for threshold in self.thresholds:
            metric_value = metrics.get(threshold.metric)
            
            if metric_value is None:
                continue
            
            # Check threshold
            passed = self._check_threshold(
                metric_value,
                threshold.operator,
                threshold.threshold,
            )
            
            if not passed:
                violation = {
                    "metric": threshold.metric,
                    "value": metric_value,
                    "threshold": threshold.threshold,
                    "operator": threshold.operator,
                }
                
                if threshold.severity == "blocking":
                    violations.append(violation)
                else:
                    warnings.append(violation)
        
        passed = len(violations) == 0
        
        return {
            "passed": passed,
            "violations": violations,
            "warnings": warnings,
        }
    
    def _check_threshold(
        self,
        value: float,
        operator: str,
        threshold: float,
    ) -> bool:
        """Check if value meets threshold."""
        if operator == ">=":
            return value >= threshold
        elif operator == "<=":
            return value <= threshold
        elif operator == "==":
            return abs(value - threshold) < 0.001
        else:
            raise ValueError(f"Unknown operator: {operator}")

# Usage in CI
if __name__ == "__main__":
    import sys
    import json
    
    # Load eval results
    with open("eval_results.json", "r") as f:
        metrics = json.load(f)
    
    # Define thresholds
    gate = QualityGate(thresholds=[
        QualityThreshold("accuracy", ">=", 0.85, "blocking"),
        QualityThreshold("hallucination_rate", "<=", 0.02, "blocking"),
        QualityThreshold("safety_rate", ">=", 0.98, "blocking"),
        QualityThreshold("p95_latency_ms", "<=", 2000, "warning"),
        QualityThreshold("average_consistency", ">=", 0.80, "warning"),
    ])
    
    # Evaluate
    result = gate.evaluate(metrics)
    
    if not result["passed"]:
        print("❌ Quality gate FAILED")
        print("\nBlocking violations:")
        for v in result["violations"]:
            print(f"  {v['metric']}: {v['value']:.4f} {v['operator']} {v['threshold']}")
        
        sys.exit(1)
    
    if result["warnings"]:
        print("⚠️  Quality gate PASSED with warnings")
        print("\nWarnings:")
        for w in result["warnings"]:
            print(f"  {w['metric']}: {w['value']:.4f} {w['operator']} {w['threshold']}")
    else:
        print("✅ Quality gate PASSED")
    
    sys.exit(0)

- name: Check quality gate
  run: |
    python scripts/quality_gate.py

# scripts/regression_detection.py
from typing import Dict, Any
import json
from pathlib import Path

class RegressionDetector:
    """Detect metric regressions against baseline."""
    
    def __init__(self, baseline_path: Path):
        self.baseline_path = baseline_path
        self.baseline_metrics = self._load_baseline()
    
    def _load_baseline(self) -> Dict[str, float]:
        """Load baseline metrics."""
        if not self.baseline_path.exists():
            return {}
        
        with self.baseline_path.open("r") as f:
            return json.load(f)
    
    def detect_regressions(
        self,
        current_metrics: Dict[str, float],
        max_regression: float = 0.03,  # 3% max drop
    ) -> Dict[str, Any]:
        """Detect regressions in metrics."""
        regressions = []
        improvements = []
        
        for metric, current_value in current_metrics.items():
            if metric not in self.baseline_metrics:
                continue
            
            baseline_value = self.baseline_metrics[metric]
            delta = current_value - baseline_value
            delta_pct = delta / baseline_value if baseline_value != 0 else 0
            
            # Determine if metric is "higher is better" or "lower is better"
            if metric in ["accuracy", "safety_rate", "consistency"]:
                # Higher is better
                if delta_pct < -max_regression:
                    regressions.append({
                        "metric": metric,
                        "baseline": baseline_value,
                        "current": current_value,
                        "delta": delta,
                        "delta_pct": delta_pct,
                    })
                elif delta_pct > 0.01:  # 1% improvement
                    improvements.append({
                        "metric": metric,
                        "baseline": baseline_value,
                        "current": current_value,
                        "delta": delta,
                        "delta_pct": delta_pct,
                    })
            
            elif metric in ["hallucination_rate", "p95_latency_ms"]:
                # Lower is better
                if delta_pct > max_regression:
                    regressions.append({
                        "metric": metric,
                        "baseline": baseline_value,
                        "current": current_value,
                        "delta": delta,
                        "delta_pct": delta_pct,
                    })
                elif delta_pct < -0.01:  # 1% improvement
                    improvements.append({
                        "metric": metric,
                        "baseline": baseline_value,
                        "current": current_value,
                        "delta": delta,
                        "delta_pct": delta_pct,
                    })
        
        return {
            "has_regressions": len(regressions) > 0,
            "regressions": regressions,
            "improvements": improvements,
        }
    
    def update_baseline(
        self,
        new_metrics: Dict[str, float],
    ) -> None:
        """Update baseline with new metrics."""
        with self.baseline_path.open("w") as f:
            json.dump(new_metrics, f, indent=2)
        
        print(f"✓ Updated baseline: {self.baseline_path}")

# Usage
if __name__ == "__main__":
    import sys
    
    detector = RegressionDetector(
        baseline_path=Path("evals/baselines/production_baseline.json")
    )
    
    # Load current metrics
    with open("eval_results.json", "r") as f:
        current_metrics = json.load(f)
    
    # Detect regressions
    result = detector.detect_regressions(current_metrics)
    
    if result["has_regressions"]:
        print("❌ REGRESSIONS DETECTED")
        print("\nRegressions:")
        for r in result["regressions"]:
            print(f"  {r['metric']}: {r['baseline']:.4f} → {r['current']:.4f} ({r['delta_pct']:+.1%})")
        
        sys.exit(1)
    
    if result["improvements"]:
        print("✅ No regressions (improvements detected)")
        print("\nImprovements:")
        for i in result["improvements"]:
            print(f"  {i['metric']}: {i['baseline']:.4f} → {i['current']:.4f} ({i['delta_pct']:+.1%})")
    else:
        print("✅ No regressions")
    
    # Update baseline on main branch
    if sys.argv[1] == "--update-baseline":
        detector.update_baseline(current_metrics)

# evals/test_performance.py
import pytest
import time
import asyncio
from typing import List
import numpy as np

@pytest.mark.asyncio
async def test_latency_benchmarks():
    """Benchmark latency metrics."""
    from src.ai_system import AISystem
    
    system = AISystem()
    
    # Warm up
    await system.query("Hello")
    
    # Benchmark
    latencies = []
    num_requests = 100
    
    for i in range(num_requests):
        start = time.perf_counter()
        await system.query(f"Test query {i}")
        latency_ms = (time.perf_counter() - start) * 1000
        latencies.append(latency_ms)
    
    # Calculate percentiles
    p50 = np.percentile(latencies, 50)
    p95 = np.percentile(latencies, 95)
    p99 = np.percentile(latencies, 99)
    avg = np.mean(latencies)
    
    print(f"\nLatency benchmarks ({num_requests} requests):")
    print(f"  P50: {p50:.0f}ms")
    print(f"  P95: {p95:.0f}ms")
    print(f"  P99: {p99:.0f}ms")
    print(f"  Avg: {avg:.0f}ms")
    
    # Assert thresholds
    assert p50 <= 1000, f"P50 latency {p50:.0f}ms exceeds 1s"
    assert p95 <= 2000, f"P95 latency {p95:.0f}ms exceeds 2s"
    assert p99 <= 3000, f"P99 latency {p99:.0f}ms exceeds 3s"

@pytest.mark.asyncio
async def test_throughput_benchmark():
    """Benchmark throughput."""
    from src.ai_system import AISystem
    
    system = AISystem()
    
    num_requests = 50
    concurrency = 10
    
    async def worker(request_id: int):
        await system.query(f"Request {request_id}")
    
    start = time.perf_counter()
    
    # Run concurrent requests
    tasks = [worker(i) for i in range(num_requests)]
    await asyncio.gather(*tasks)
    
    duration = time.perf_counter() - start
    throughput = num_requests / duration
    
    print(f"\nThroughput benchmark:")
    print(f"  Requests: {num_requests}")
    print(f"  Duration: {duration:.2f}s")
    print(f"  Throughput: {throughput:.1f} req/s")
    
    # Assert minimum throughput
    MIN_THROUGHPUT = 5  # req/s
    assert throughput >= MIN_THROUGHPUT, (
        f"Throughput {throughput:.1f} req/s below minimum {MIN_THROUGHPUT} req/s"
    )

@pytest.mark.asyncio
async def test_cost_estimation():
    """Estimate cost per request."""
    from src.ai_system import AISystem
    
    system = AISystem()
    
    # Track token usage
    total_input_tokens = 0
    total_output_tokens = 0
    num_requests = 20
    
    for i in range(num_requests):
        response = await system.query(f"Test query {i}")
        
        # Extract token usage (implementation-specific)
        if hasattr(response, "usage"):
            total_input_tokens += response.usage.prompt_tokens
            total_output_tokens += response.usage.completion_tokens
    
    # Calculate cost (example pricing)
    INPUT_COST_PER_1M = 2.50  # $2.50 per 1M input tokens
    OUTPUT_COST_PER_1M = 10.00  # $10 per 1M output tokens
    
    avg_input_tokens = total_input_tokens / num_requests
    avg_output_tokens = total_output_tokens / num_requests
    
    cost_per_request = (
        (avg_input_tokens / 1_000_000) * INPUT_COST_PER_1M +
        (avg_output_tokens / 1_000_000) * OUTPUT_COST_PER_1M
    )
    
    print(f"\nCost estimation:")
    print(f"  Avg input tokens: {avg_input_tokens:.0f}")
    print(f"  Avg output tokens: {avg_output_tokens:.0f}")
    print(f"  Cost per request: ${cost_per_request:.4f}")
    
    # Assert cost threshold
    MAX_COST_PER_REQUEST = 0.05  # $0.05
    assert cost_per_request <= MAX_COST_PER_REQUEST, (
        f"Cost per request ${cost_per_request:.4f} exceeds ${MAX_COST_PER_REQUEST}"
    )

# evals/cost_manager.py
from typing import Dict, Any, Optional
import os

class EvalCostManager:
    """Manage evaluation costs."""
    
    def __init__(self, max_cost_usd: float = 10.0):
        self.max_cost_usd = max_cost_usd
        self.current_cost = 0.0
    
    def sample_dataset(
        self,
        dataset: list,
        is_pr: bool = True,
    ) -> list:
        """Sample dataset based on context."""
        if is_pr:
            # Quick eval on PRs—sample 20%
            sample_size = max(int(len(dataset) * 0.2), 50)
            return dataset[:sample_size]
        else:
            # Full eval on main/nightly
            return dataset
    
    def should_run_expensive_eval(self) -> bool:
        """Check if expensive eval should run."""
        # Only run expensive evals on main branch or nightly
        is_main = os.getenv("GITHUB_REF") == "refs/heads/main"
        is_nightly = os.getenv("GITHUB_EVENT_NAME") == "schedule"
        
        return is_main or is_nightly
    
    def estimate_cost(
        self,
        num_requests: int,
        avg_tokens: int = 1000,
    ) -> float:
        """Estimate eval cost."""
        COST_PER_1K_TOKENS = 0.01  # Approximate
        
        return (num_requests * avg_tokens * COST_PER_1K_TOKENS) / 1000

# Usage in tests
def pytest_configure(config):
    """Pytest hook to configure eval runs."""
    cost_manager = EvalCostManager(max_cost_usd=10.0)
    
    # Store in config for tests to access
    config.cost_manager = cost_manager

def pytest_collection_modifyitems(config, items):
    """Pytest hook to skip expensive tests on PRs."""
    cost_manager = config.cost_manager
    
    if not cost_manager.should_run_expensive_eval():
        skip_expensive = pytest.mark.skip(reason="Expensive eval—skipping on PR")
        
        for item in items:
            if "expensive" in item.keywords:
                item.add_marker(skip_expensive)

# .github/workflows/deploy-production.yml
name: Deploy to Production

on:
  push:
    branches: [main]

jobs:
  eval-and-deploy:
    runs-on: ubuntu-latest
    
    steps:
      - name: Checkout
        uses: actions/checkout@v4
      
      - name: Run full eval suite
        run: |
          pip install -r requirements.txt
          pytest evals/ -v --json-report --json-report-file=eval_results.json
      
      - name: Check quality gate
        run: |
          python scripts/quality_gate.py
      
      - name: Detect regressions
        run: |
          python scripts/regression_detection.py
      
      - name: Deploy to staging
        run: |
          # Deploy to staging environment
          aws ecs update-service --cluster staging --service ai-system --force-new-deployment
      
      - name: Run smoke tests on staging
        run: |
          pytest evals/test_smoke.py --env=staging
      
      - name: Deploy to production
        if: success()
        run: |
          # Deploy to production
          aws ecs update-service --cluster production --service ai-system --force-new-deployment
      
      - name: Update baseline
        if: success()
        run: |
          python scripts/regression_detection.py --update-baseline
          git config user.name "github-actions[bot]"
          git config user.email "github-actions[bot]@users.noreply.github.com"
          git add evals/baselines/
          git commit -m "Update eval baselines [skip ci]"
          git push