💡 Comprehensive Examples & Use Cases¶

"""
CI/CD Integration Example
Automated LLM evaluation in continuous integration pipelines
"""

import os
import sys
import json
from typing import Dict, Any, List

class CIPipelineEvaluator:
    """
    Evaluation system designed for CI/CD pipelines
    Includes test result reporting, failure thresholds, and artifact generation
    """

    def __init__(self, config_path: str = "ci_evaluation_config.json"):
        self.config = self._load_config(config_path)
        self.registry = ModelRegistry()
        self.engine = ModelInferenceEngine(self.registry)
        self.test_results = []

    def _load_config(self, config_path: str) -> Dict[str, Any]:
        """Load CI configuration"""
        try:
            with open(config_path, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            return self._default_config()

    def _default_config(self) -> Dict[str, Any]:
        """Default CI configuration"""
        return {
            "models_to_test": ["gpt-3.5-turbo"],
            "test_suite": "regression",
            "accuracy_threshold": 0.85,
            "cost_threshold": 1.00,
            "timeout_threshold": 30.0,
            "generate_report": True,
            "fail_on_threshold": True
        }

    def run_regression_tests(self) -> Dict[str, Any]:
        """Run regression tests for all configured models"""

        print("🚀 Starting CI/CD LLM Evaluation Pipeline")
        print(f"Testing models: {self.config['models_to_test']}")

        # Load regression test suite
        test_cases = self._load_regression_tests()

        overall_results = {
            'pipeline_status': 'success',
            'model_results': {},
            'summary': {}
        }

        for model_name in self.config['models_to_test']:
            print(f"\n🤖 Testing model: {model_name}")

            try:
                # Register model from environment
                self._register_model_from_env(model_name)

                # Run evaluation
                results = self.engine.evaluate_model(model_name, test_cases)

                # Check thresholds
                threshold_results = self._check_thresholds(results, model_name)

                overall_results['model_results'][model_name] = {
                    'evaluation_results': results,
                    'threshold_checks': threshold_results,
                    'status': 'passed' if threshold_results['all_passed'] else 'failed'
                }

                if not threshold_results['all_passed']:
                    overall_results['pipeline_status'] = 'failed'

            except Exception as e:
                print(f"❌ Model {model_name} evaluation failed: {e}")
                overall_results['model_results'][model_name] = {
                    'status': 'error',
                    'error': str(e)
                }
                overall_results['pipeline_status'] = 'failed'

        # Generate summary
        overall_results['summary'] = self._generate_summary(overall_results)

        # Generate reports
        if self.config['generate_report']:
            self._generate_ci_report(overall_results)

        # Exit with appropriate code
        if self.config['fail_on_threshold'] and overall_results['pipeline_status'] == 'failed':
            print("❌ Pipeline failed due to threshold violations")
            sys.exit(1)

        print("✅ Pipeline completed successfully")
        return overall_results

    def _load_regression_tests(self) -> List[Dict]:
        """Load regression test cases"""
        test_file = f"tests/regression_{self.config['test_suite']}.json"

        try:
            with open(test_file, 'r') as f:
                return json.load(f)
        except FileNotFoundError:
            # Generate minimal test suite if not found
            generator = TestDatasetGenerator()
            return generator.generate_test_cases(
                {"domain": "regression", "required_capabilities": ["reasoning"]},
                count=20
            )

    def _register_model_from_env(self, model_name: str):
        """Register model using environment variables"""

        # Get configuration from environment
        api_key = os.getenv(f"{model_name.upper().replace('-', '_')}_API_KEY")
        if not api_key:
            raise ValueError(f"API key not found for {model_name}")

        # Basic model configuration
        config = {
            "provider": "openai" if "gpt" in model_name else "anthropic",
            "api_key": api_key,
            "capabilities": ["reasoning", "creativity"]
        }

        self.registry.register_model(model_name, config)

    def _check_thresholds(self, results: Dict, model_name: str) -> Dict[str, Any]:
        """Check if results meet quality thresholds"""

        metrics = results['aggregate_metrics']
        threshold_checks = {}

        # Accuracy threshold
        accuracy_passed = metrics['accuracy'] >= self.config['accuracy_threshold']
        threshold_checks['accuracy'] = {
            'value': metrics['accuracy'],
            'threshold': self.config['accuracy_threshold'],
            'passed': accuracy_passed
        }

        # Cost threshold
        cost_passed = metrics['total_cost'] <= self.config['cost_threshold']
        threshold_checks['cost'] = {
            'value': metrics['total_cost'],
            'threshold': self.config['cost_threshold'],
            'passed': cost_passed
        }

        # Timeout threshold
        timeout_passed = metrics['average_response_time'] <= self.config['timeout_threshold']
        threshold_checks['response_time'] = {
            'value': metrics['average_response_time'],
            'threshold': self.config['timeout_threshold'],
            'passed': timeout_passed
        }

        threshold_checks['all_passed'] = all(
            check['passed'] for check in threshold_checks.values()
        )

        # Print threshold results
        for metric, check in threshold_checks.items():
            if metric == 'all_passed':
                continue

            status = "✅" if check['passed'] else "❌"
            print(f"  {status} {metric}: {check['value']:.4f} "
                  f"(threshold: {check['threshold']:.4f})")

        return threshold_checks

    def _generate_ci_report(self, results: Dict[str, Any]):
        """Generate CI-friendly evaluation report"""

        # Create reports directory
        os.makedirs('reports', exist_ok=True)

        # Generate JSON report
        with open('reports/llm_evaluation_report.json', 'w') as f:
            json.dump(results, f, indent=2, default=str)

        # Generate markdown report
        markdown_report = self._create_markdown_report(results)
        with open('reports/llm_evaluation_report.md', 'w') as f:
            f.write(markdown_report)

        print("📊 Reports generated in ./reports/")

    def _create_markdown_report(self, results: Dict[str, Any]) -> str:
        """Create markdown evaluation report"""

        report = "# LLM Evaluation Report\n\n"
        report += f"**Pipeline Status**: {results['pipeline_status'].upper()}\n\n"

        report += "## Model Results\n\n"

        for model_name, model_results in results['model_results'].items():
            report += f"### {model_name}\n\n"

            if model_results['status'] == 'error':
                report += f"❌ **Error**: {model_results['error']}\n\n"
                continue

            eval_results = model_results['evaluation_results']
            metrics = eval_results['aggregate_metrics']

            report += f"- **Status**: {model_results['status'].upper()}\n"
            report += f"- **Accuracy**: {metrics['accuracy']:.1%}\n"
            report += f"- **Total Cost**: ${metrics['total_cost']:.4f}\n"
            report += f"- **Average Response Time**: {metrics['average_response_time']:.2f}s\n"
            report += f"- **Success Rate**: {metrics['success_rate']:.1%}\n\n"

            # Threshold checks
            threshold_checks = model_results['threshold_checks']
            report += "#### Threshold Checks\n\n"

            for metric, check in threshold_checks.items():
                if metric == 'all_passed':
                    continue

                status = "✅ Pass" if check['passed'] else "❌ Fail"
                report += f"- **{metric.title()}**: {status} "
                report += f"({check['value']:.4f} vs {check['threshold']:.4f})\n"

            report += "\n"

        return report

# CI/CD Usage Script
def main():
    """Main CI/CD evaluation script"""

    # Initialize evaluator
    evaluator = CIPipelineEvaluator()

    # Run regression tests
    results = evaluator.run_regression_tests()

    # Additional CI/CD specific actions
    if os.getenv('CI') == 'true':
        # Running in CI environment
        print("🔄 CI environment detected")

        # Set GitHub Actions outputs
        if os.getenv('GITHUB_ACTIONS') == 'true':
            with open(os.environ['GITHUB_OUTPUT'], 'a') as f:
                f.write(f"evaluation_status={results['pipeline_status']}\n")
                f.write(f"report_path=reports/llm_evaluation_report.md\n")

if __name__ == "__main__":
    main()