run_local_evaluation.py

import os
import json
from dotenv import load_dotenv
from azure.ai.evaluation import (
    RelevanceEvaluator,
    CoherenceEvaluator,
    GroundednessEvaluator,
    SimilarityEvaluator,
    F1ScoreEvaluator,
    evaluate
)
from azure.identity import DefaultAzureCredential, get_bearer_token_provider
from openai import AzureOpenAI

def analyze_results_with_model(aoai_client, model_deployment, model_results, scores_dict, winners_dict, overall_winner):
    """Use the model to analyze evaluation results and determine the best model"""

    # Prepare the comparison data as a structured summary
    summary_text = "Evaluation Results:\n\n"

    for eval_name in ['relevance', 'coherence', 'groundedness', 'similarity']:
        if eval_name in scores_dict:
            summary_text += f"{eval_name.capitalize()}:\n"
            for model_name, score in scores_dict[eval_name].items():
                summary_text += f"  - {model_name}: {score:.2f}\n"
            summary_text += "\n"

    summary_text += f"\nMetrics Won:\n"
    for model_name in winners_dict:
        if model_name != 'Tie':
            summary_text += f"  - {model_name}: {winners_dict[model_name]} metrics\n"

    summary_text += f"\nTies: {winners_dict['Tie']}"

    # Prepare prompt for the model
    system_prompt = """You are an AI model evaluation expert. Analyze the evaluation results and determine which model performed best overall.
Consider both average scores and consistency across metrics. Provide a single concise sentence in this exact format:
"The best model based on these evaluations is [MODEL_NAME] - ready for upgrade!"

Be definitive in your choice, even if the differences are small."""

    user_prompt = f"{summary_text}\n\nBased on these evaluation results, which model performed best?"

    try:
        response = aoai_client.chat.completions.create(
            model=model_deployment,
            messages=[
                {'role': 'system', 'content': system_prompt},
                {'role': 'user', 'content': user_prompt}
            ],
            temperature=0.0,
            max_completion_tokens=100,
            timeout=30
        )

        conclusion = response.choices[0].message.content.strip()
        return conclusion
    except Exception as ex:
        # Fallback to simple logic if model call fails
        return f"The best model based on these evaluations is {overall_winner} - ready for upgrade!"

def load_evaluation_data(responses_file, ground_truth_file):
    """Load model responses and merge with ground truth"""
    # Load ground truth
    ground_truth_dict = {}
    with open(ground_truth_file, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                data = json.loads(line)
                ground_truth_dict[data['query']] = data['ground_truth']

    # Load responses and merge
    eval_data = []
    with open(responses_file, 'r', encoding='utf-8') as f:
        for line in f:
            if line.strip():
                data = json.loads(line)
                query = data['query']
                eval_data.append({
                    'query': query,
                    'response': data['response'],
                    'ground_truth': ground_truth_dict.get(query, ''),
                    'context': ground_truth_dict.get(query, '')  # For groundedness
                })

    return eval_data

def run_evaluation(data, model_config, model_name):
    """Run evaluations using azure-ai-evaluation library"""

    print(f"\n{'='*60}")
    print(f"Evaluating {model_name}")
    print('='*60)
    print(f"Test cases: {len(data)}")

    try:
        # Write data to temp file (evaluate expects a file path)
        temp_file = f"temp_eval_{model_name.replace('.', '_').replace('-', '_')}.jsonl"
        with open(temp_file, 'w', encoding='utf-8') as f:
            for item in data:
                f.write(json.dumps(item) + '\n')
        print(f"[OK] Data written to {temp_file}")

        # Initialize evaluators
        print("\nInitializing evaluators...")

        # LLM-based evaluators (need model config)
        relevance_eval = RelevanceEvaluator(model_config)
        coherence_eval = CoherenceEvaluator(model_config)
        groundedness_eval = GroundednessEvaluator(model_config)
        similarity_eval = SimilarityEvaluator(model_config)

        # Code-based evaluator (no model needed)
        f1_eval = F1ScoreEvaluator()

        print("[OK] Evaluators initialized")

        # Run evaluation
        print("\nRunning evaluation...")
        result = evaluate(
            data=temp_file,
            evaluators={
                "relevance": relevance_eval,
                "coherence": coherence_eval,
                "groundedness": groundedness_eval,
                "similarity": similarity_eval,
                "f1_score": f1_eval,
            },
        )

        print("[OK] Evaluation completed")

        # Display results
        print(f"\nResults for {model_name}:")
        print("-" * 60)

        metrics = result['metrics']

        # Print aggregate metrics
        for metric_name, metric_value in metrics.items():
            if isinstance(metric_value, (int, float)):
                print(f"{metric_name}: {metric_value:.3f}")

        # Extract evaluator stats from metrics
        evaluator_stats = {}
        for evaluator_name in ['relevance', 'coherence', 'groundedness', 'similarity']:
            # Get average score from metrics
            score_key = f"{evaluator_name}.{evaluator_name}"
            if score_key in metrics:
                avg_score = metrics[score_key]
                # Typically, scores >= 3 are considered passing (on 1-5 scale)
                # For binary_aggregate, 1.0 = 100% pass rate
                binary_key = f"{evaluator_name}.binary_aggregate"
                pass_rate = metrics.get(binary_key, 0.0)

                evaluator_stats[evaluator_name] = {
                    'avg_score': avg_score,
                    'pass_rate': pass_rate,
                }

        print("\nDetailed Results:")
        print("-" * 60)
        for eval_name, stats in evaluator_stats.items():
            print(f"\n{eval_name.upper()}:")
            print(f"  Average Score: {stats['avg_score']:.2f}")
            print(f"  Pass Rate: {stats['pass_rate']:.1%}")

        return {
            'model': model_name,
            'metrics': metrics,
            'stats': evaluator_stats,
            'result': result
        }

    except Exception as e:
        print(f"\n[ERROR] Evaluation failed: {e}")
        import traceback
        traceback.print_exc()
        return None

if __name__ == "__main__":
    # Load environment variables
    load_dotenv()

    print("="*60)
    print("Azure AI Local Evaluation")
    print("="*60)

    # Get Azure OpenAI configuration for LLM evaluators
    azure_endpoint = os.environ["AZURE_OPENAI_ENDPOINT"]
    # Use gpt-4o for evaluation (gpt-5.1 doesn't support max_tokens parameter)
    azure_deployment = "gpt-4o"

    print(f"\nAzure OpenAI Endpoint: {azure_endpoint}")
    print(f"Evaluator Model: {azure_deployment}")

    # Create Azure OpenAI client for result analysis
    aoai_api_version = '2024-10-21'
    credential = DefaultAzureCredential()
    token_provider = get_bearer_token_provider(
        credential, "https://cognitiveservices.azure.com/.default"
    )
    aoai_client = AzureOpenAI(
        api_version=aoai_api_version,
        azure_endpoint=azure_endpoint,
        azure_ad_token_provider=token_provider
    )

    # Model configuration for evaluators
    model_config = {
        "azure_endpoint": azure_endpoint,
        "azure_deployment": azure_deployment,
        "api_version": "2024-10-21",
    }

    # Load data for all models
    print("\nLoading evaluation data...")

    models_to_evaluate = [
        ("GPT-4o", "model_responses_gpt-4o.jsonl"),
        ("GPT-5.1", "model_responses_gpt_5_1.jsonl"),
        ("Model-Router", "model_responses_model-router.jsonl")
    ]

    model_data = {}
    for model_name, file_path in models_to_evaluate:
        try:
            data = load_evaluation_data(file_path, "evaluate_test_data.jsonl")
            if len(data) > 7:
                model_data[model_name] = data
                print(f"[OK] {model_name}: {len(data)} test cases")
            else:
                print(f"[SKIP] {model_name}: Only {len(data)} test cases (minimum 8 required)")
        except FileNotFoundError:
            print(f"[SKIP] {model_name}: File {file_path} not found")

    # Run evaluations for all loaded models
    print("\nRunning evaluations...")
    model_results = {}
    for model_name, data in model_data.items():
        results = run_evaluation(data, model_config, model_name)
        if results:
            model_results[model_name] = results

    # Comparison summary
    print("\n" + "="*100)
    print("COMPARISON SUMMARY")
    print("="*100)

    if len(model_results) >= 2:
        # Print comparison table
        model_names = list(model_results.keys())

        # Build header dynamically based on number of models
        header_parts = ["Metric"]
        for model_name in model_names:
            header_parts.append(f"{model_name} Score")
        header_parts.append("Winner")

        # Calculate column widths
        metric_width = 15
        score_width = 18
        winner_width = 20

        print("\n" + "-"*100)
        header = f"{header_parts[0]:<{metric_width}}"
        for i in range(1, len(model_names) + 1):
            header += f" {header_parts[i]:<{score_width}}"
        header += f" {header_parts[-1]:<{winner_width}}"
        print(header)
        print("-"*100)

        winners = {model_name: 0 for model_name in model_names}
        winners['Tie'] = 0

        # Store all scores for model analysis
        scores_dict = {}

        # Compare each evaluator
        for eval_name in ['relevance', 'coherence', 'groundedness', 'similarity']:
            # Check if all models have this metric
            if all(eval_name in model_results[m]['stats'] for m in model_names):
                scores = {m: model_results[m]['stats'][eval_name]['avg_score'] for m in model_names}

                # Store scores for model analysis
                scores_dict[eval_name] = scores

                # Determine winner
                max_score = max(scores.values())
                winners_list = [m for m, s in scores.items() if s == max_score]

                if len(winners_list) == 1:
                    winner = winners_list[0]
                    winners[winner] += 1
                else:
                    winner = "Tie"
                    winners['Tie'] += 1

                # Build row
                row = f"{eval_name.capitalize():<{metric_width}}"
                for model_name in model_names:
                    row += f" {scores[model_name]:<{score_width}.2f}"
                row += f" {winner:<{winner_width}}"
                print(row)

        print("-"*100)

        # Overall scores
        totals = {m: sum(s['avg_score'] for s in model_results[m]['stats'].values()) for m in model_names}
        max_total = max(totals.values())
        overall_winners = [m for m, t in totals.items() if t == max_total]

        if len(overall_winners) == 1:
            overall_winner = overall_winners[0]
        else:
            overall_winner = "Tie"

        # Print TOTAL row
        total_row = f"{'TOTAL':<{metric_width}}"
        for model_name in model_names:
            total_row += f" {totals[model_name]:<{score_width}.2f}"
        total_row += f" {overall_winner:<{winner_width}}"
        print(total_row)
        print("="*100)

        # Use model to analyze results and generate conclusion
        print("\nSUMMARY:")
        conclusion = analyze_results_with_model(aoai_client, azure_deployment, model_results, scores_dict, winners, overall_winner)
        print(f"{conclusion}")

        # Pass rate table
        print("\n" + "-"*100)
        pass_header = f"{'Metric':<{metric_width}}"
        for model_name in model_names:
            pass_header += f" {model_name + ' Pass':<{score_width}}"
        pass_header += f" {'Best Pass Rate':<{winner_width}}"
        print(pass_header)
        print("-"*100)

        for eval_name in ['relevance', 'coherence', 'groundedness', 'similarity']:
            if all(eval_name in model_results[m]['stats'] for m in model_names):
                pass_rates = {m: model_results[m]['stats'][eval_name]['pass_rate'] for m in model_names}

                max_pass = max(pass_rates.values())
                best_models = [m for m, p in pass_rates.items() if p == max_pass]

                if len(best_models) == 1:
                    best = best_models[0]
                else:
                    best = "Tie"

                # Build row
                pass_row = f"{eval_name.capitalize():<{metric_width}}"
                for model_name in model_names:
                    pass_row += f" {pass_rates[model_name]:<{score_width}.1%}"
                pass_row += f" {best:<{winner_width}}"
                print(pass_row)

        print("-"*100)
    else:
        print("\n[FAILED] Need at least 2 models to compare. Check logs above.")