Évaluation et testing des agents IA

# Même question, 3 exécutions différentes
agent.run("Résume cet article en 2 phrases")

# Réponse 1 : 47 mots, focus sur la méthodologie
# Réponse 2 : 31 mots, focus sur les résultats
# Réponse 3 : 53 mots, focus sur les implications

┌─────────────────────────────────────────────────┐
│          ÉVALUATION D'UN AGENT                  │
├─────────────────────────────────────────────────┤
│                                                 │
│  1️⃣ QUALITÉ     : Précision, pertinence        │
│  2️⃣ PERFORMANCE : Latence, throughput          │
│  3️⃣ COÛTS       : $/requête, tokens            │
│  4️⃣ FIABILITÉ   : Success rate, erreurs        │
│                                                 │
└─────────────────────────────────────────────────┘

import openai
from typing import Dict, List
from dataclasses import dataclass

@dataclass
class EvaluationResult:
    """Résultat d'évaluation"""
    accuracy: float
    relevance: float
    completeness: float
    factuality: float
    explanation: str

class AgentEvaluator:
    """Évaluateur d'agents IA"""

    def __init__(self, evaluator_model="gpt-4"):
        self.evaluator_model = evaluator_model

    def evaluate_response(
        self,
        query: str,
        agent_response: str,
        ground_truth: str = None
    ) -> EvaluationResult:
        """Évalue une réponse d'agent"""

        prompt = f"""Tu es un évaluateur expert d'agents IA.

Question : {query}
Réponse de l'agent : {agent_response}
{f'Réponse attendue : {ground_truth}' if ground_truth else ''}

Évalue cette réponse selon 4 critères (note de 0 à 1) :

1. **Accuracy** : La réponse est-elle correcte ?
2. **Relevance** : La réponse répond-elle à la question ?
3. **Completeness** : La réponse est-elle complète ?
4. **Factuality** : Les faits mentionnés sont-ils exacts ?

Réponds en JSON :
{{
    "accuracy": 0.0-1.0,
    "relevance": 0.0-1.0,
    "completeness": 0.0-1.0,
    "factuality": 0.0-1.0,
    "explanation": "Explication détaillée"
}}"""

        response = openai.chat.completions.create(
            model=self.evaluator_model,
            messages=[{"role": "user", "content": prompt}],
            response_format={"type": "json_object"}
        )

        import json
        result = json.loads(response.choices[0].message.content)

        return EvaluationResult(**result)

    def evaluate_agent(
        self,
        agent,
        test_cases: List[Dict]
    ) -> Dict:
        """Évalue un agent sur un ensemble de test cases"""

        results = []

        for i, test_case in enumerate(test_cases):
            print(f"\n🧪 Test {i+1}/{len(test_cases)}: {test_case['query']}")

            # Exécuter l'agent
            agent_response = agent.run(test_case["query"])

            # Évaluer
            evaluation = self.evaluate_response(
                query=test_case["query"],
                agent_response=agent_response,
                ground_truth=test_case.get("expected_answer")
            )

            results.append({
                "test_case": test_case,
                "response": agent_response,
                "evaluation": evaluation
            })

            print(f"  Accuracy: {evaluation.accuracy:.2f}")
            print(f"  Relevance: {evaluation.relevance:.2f}")
            print(f"  Completeness: {evaluation.completeness:.2f}")
            print(f"  Factuality: {evaluation.factuality:.2f}")

        # Calculer les moyennes
        avg_accuracy = sum(r["evaluation"].accuracy for r in results) / len(results)
        avg_relevance = sum(r["evaluation"].relevance for r in results) / len(results)
        avg_completeness = sum(r["evaluation"].completeness for r in results) / len(results)
        avg_factuality = sum(r["evaluation"].factuality for r in results) / len(results)

        return {
            "results": results,
            "summary": {
                "accuracy": avg_accuracy,
                "relevance": avg_relevance,
                "completeness": avg_completeness,
                "factuality": avg_factuality,
                "total_tests": len(test_cases)
            }
        }


# ========== EXEMPLE D'UTILISATION ==========

# Test cases
test_cases = [
    {
        "query": "Quelle est la capitale de la France ?",
        "expected_answer": "Paris"
    },
    {
        "query": "Combien font 15 * 7 ?",
        "expected_answer": "105"
    },
    {
        "query": "Qui a écrit Les Misérables ?",
        "expected_answer": "Victor Hugo"
    }
]

# Évaluer
evaluator = AgentEvaluator()
evaluation_report = evaluator.evaluate_agent(my_agent, test_cases)

# Afficher le résumé
print(f"\n{'='*60}")
print(f"📊 RÉSUMÉ DE L'ÉVALUATION")
print(f"{'='*60}")
summary = evaluation_report["summary"]
print(f"Accuracy     : {summary['accuracy']:.1%}")
print(f"Relevance    : {summary['relevance']:.1%}")
print(f"Completeness : {summary['completeness']:.1%}")
print(f"Factuality   : {summary['factuality']:.1%}")
print(f"Total tests  : {summary['total_tests']}")

import time
import statistics
from typing import List

class PerformanceTracker:
    """Suivi des performances"""

    def __init__(self):
        self.latencies = []
        self.start_time = None
        self.end_time = None

    def __enter__(self):
        self.start_time = time.time()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.end_time = time.time()
        latency = self.end_time - self.start_time
        self.latencies.append(latency)

    def get_metrics(self) -> Dict:
        """Calcule les métriques de performance"""

        if not self.latencies:
            return {}

        sorted_latencies = sorted(self.latencies)
        n = len(sorted_latencies)

        return {
            "total_requests": n,
            "latency_mean": statistics.mean(self.latencies),
            "latency_median": statistics.median(self.latencies),
            "latency_p95": sorted_latencies[int(n * 0.95)],
            "latency_p99": sorted_latencies[int(n * 0.99)],
            "latency_min": min(self.latencies),
            "latency_max": max(self.latencies),
            "throughput": n / sum(self.latencies) if sum(self.latencies) > 0 else 0
        }

    def print_report(self):
        """Affiche un rapport de performance"""
        metrics = self.get_metrics()

        print(f"\n{'='*60}")
        print(f"⚡ RAPPORT DE PERFORMANCE")
        print(f"{'='*60}")
        print(f"Total requêtes : {metrics['total_requests']}")
        print(f"Latence moyenne : {metrics['latency_mean']:.3f}s")
        print(f"Latence médiane : {metrics['latency_median']:.3f}s")
        print(f"Latence P95     : {metrics['latency_p95']:.3f}s")
        print(f"Latence P99     : {metrics['latency_p99']:.3f}s")
        print(f"Latence min     : {metrics['latency_min']:.3f}s")
        print(f"Latence max     : {metrics['latency_max']:.3f}s")
        print(f"Throughput      : {metrics['throughput']:.2f} req/s")


# ========== UTILISATION ==========

tracker = PerformanceTracker()

# Tester sur 100 requêtes
for i in range(100):
    with tracker:
        agent.run(f"Question {i}")

# Rapport
tracker.print_report()

# Vérifier les cibles
metrics = tracker.get_metrics()
assert metrics["latency_p95"] < 2.0, "❌ P95 latency too high"
assert metrics["throughput"] > 1.0, "❌ Throughput too low"

class CostTracker:
    """Suivi des coûts"""

    PRICING = {
        "gpt-4": {"input": 0.03, "output": 0.06},  # $/1k tokens
        "gpt-3.5-turbo": {"input": 0.0015, "output": 0.002},
        "claude-3-opus": {"input": 0.015, "output": 0.075},
        "claude-3-sonnet": {"input": 0.003, "output": 0.015}
    }

    def __init__(self):
        self.calls = []

    def track_call(
        self,
        model: str,
        input_tokens: int,
        output_tokens: int
    ):
        """Enregistre un appel LLM"""

        pricing = self.PRICING.get(model, {"input": 0, "output": 0})

        cost = (
            (input_tokens / 1000) * pricing["input"] +
            (output_tokens / 1000) * pricing["output"]
        )

        self.calls.append({
            "model": model,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "cost": cost
        })

    def get_metrics(self) -> Dict:
        """Calcule les métriques de coût"""

        if not self.calls:
            return {}

        total_cost = sum(call["cost"] for call in self.calls)
        total_tokens = sum(
            call["input_tokens"] + call["output_tokens"]
            for call in self.calls
        )

        return {
            "total_calls": len(self.calls),
            "total_cost": total_cost,
            "cost_per_request": total_cost / len(self.calls),
            "total_tokens": total_tokens,
            "tokens_per_request": total_tokens / len(self.calls),
            "cost_per_1k_users": (total_cost / len(self.calls)) * 1000
        }

    def print_report(self):
        """Affiche un rapport de coûts"""
        metrics = self.get_metrics()

        print(f"\n{'='*60}")
        print(f"💰 RAPPORT DE COÛTS")
        print(f"{'='*60}")
        print(f"Total appels    : {metrics['total_calls']}")
        print(f"Coût total      : ${metrics['total_cost']:.4f}")
        print(f"Coût/requête    : ${metrics['cost_per_request']:.4f}")
        print(f"Total tokens    : {metrics['total_tokens']:,}")
        print(f"Tokens/requête  : {metrics['tokens_per_request']:.0f}")
        print(f"Coût/1k users   : ${metrics['cost_per_1k_users']:.2f}")


# ========== INTÉGRATION AVEC L'AGENT ==========

class CostTrackedAgent:
    """Agent avec tracking des coûts"""

    def __init__(self, base_agent):
        self.agent = base_agent
        self.cost_tracker = CostTracker()

    def run(self, query: str) -> str:
        """Exécute l'agent et track les coûts"""

        # Wrapper autour des appels OpenAI
        original_create = openai.chat.completions.create

        def tracked_create(*args, **kwargs):
            response = original_create(*args, **kwargs)

            # Extraire les métriques
            model = kwargs.get("model", "gpt-4")
            usage = response.usage

            self.cost_tracker.track_call(
                model=model,
                input_tokens=usage.prompt_tokens,
                output_tokens=usage.completion_tokens
            )

            return response

        # Patch temporaire
        openai.chat.completions.create = tracked_create

        try:
            result = self.agent.run(query)
        finally:
            # Restore
            openai.chat.completions.create = original_create

        return result

class ReliabilityTracker:
    """Suivi de la fiabilité"""

    def __init__(self):
        self.results = []

    def track_result(
        self,
        success: bool,
        error_type: str = None,
        has_hallucination: bool = False
    ):
        """Enregistre un résultat"""
        self.results.append({
            "success": success,
            "error_type": error_type,
            "has_hallucination": has_hallucination
        })

    def get_metrics(self) -> Dict:
        """Calcule les métriques de fiabilité"""

        if not self.results:
            return {}

        total = len(self.results)
        successes = sum(1 for r in self.results if r["success"])
        errors = sum(1 for r in self.results if not r["success"])
        timeouts = sum(1 for r in self.results if r["error_type"] == "timeout")
        hallucinations = sum(1 for r in self.results if r["has_hallucination"])

        return {
            "total_requests": total,
            "success_rate": successes / total,
            "error_rate": errors / total,
            "timeout_rate": timeouts / total,
            "hallucination_rate": hallucinations / total
        }

    def print_report(self):
        """Affiche un rapport de fiabilité"""
        metrics = self.get_metrics()

        print(f"\n{'='*60}")
        print(f"🔒 RAPPORT DE FIABILITÉ")
        print(f"{'='*60}")
        print(f"Total requêtes       : {metrics['total_requests']}")
        print(f"Success rate         : {metrics['success_rate']:.1%}")
        print(f"Error rate           : {metrics['error_rate']:.1%}")
        print(f"Timeout rate         : {metrics['timeout_rate']:.1%}")
        print(f"Hallucination rate   : {metrics['hallucination_rate']:.1%}")

import pytest
from typing import Callable

class AgentTestSuite:
    """Suite de tests complète pour agents"""

    def __init__(self, agent):
        self.agent = agent
        self.quality_evaluator = AgentEvaluator()
        self.performance_tracker = PerformanceTracker()
        self.cost_tracker = CostTracker()
        self.reliability_tracker = ReliabilityTracker()

    def run_full_evaluation(self, test_cases: List[Dict]) -> Dict:
        """Évaluation complète"""

        print(f"\n{'='*60}")
        print(f"🧪 ÉVALUATION COMPLÈTE DE L'AGENT")
        print(f"{'='*60}")

        results = []

        for test_case in test_cases:
            result = self._run_single_test(test_case)
            results.append(result)

        # Rapports
        self.quality_evaluator.print_report()
        self.performance_tracker.print_report()
        self.cost_tracker.print_report()
        self.reliability_tracker.print_report()

        return {
            "results": results,
            "quality": self.quality_evaluator.get_metrics(),
            "performance": self.performance_tracker.get_metrics(),
            "cost": self.cost_tracker.get_metrics(),
            "reliability": self.reliability_tracker.get_metrics()
        }

    def _run_single_test(self, test_case: Dict) -> Dict:
        """Exécute un test unique"""

        query = test_case["query"]

        try:
            # Performance tracking
            with self.performance_tracker:
                response = self.agent.run(query)

            # Quality evaluation
            evaluation = self.quality_evaluator.evaluate_response(
                query=query,
                agent_response=response,
                ground_truth=test_case.get("expected_answer")
            )

            # Reliability tracking
            self.reliability_tracker.track_result(
                success=True,
                has_hallucination=evaluation.factuality < 0.7
            )

            return {
                "query": query,
                "response": response,
                "evaluation": evaluation,
                "success": True
            }

        except TimeoutError:
            self.reliability_tracker.track_result(
                success=False,
                error_type="timeout"
            )
            return {"query": query, "success": False, "error": "timeout"}

        except Exception as e:
            self.reliability_tracker.track_result(
                success=False,
                error_type=type(e).__name__
            )
            return {"query": query, "success": False, "error": str(e)}


# ========== TESTS PYTEST ==========

@pytest.fixture
def agent():
    """Fixture de l'agent"""
    return MyAgent()

@pytest.fixture
def test_suite(agent):
    """Fixture de la suite de tests"""
    return AgentTestSuite(agent)

def test_quality(test_suite):
    """Test de qualité"""
    test_cases = [
        {"query": "Quelle est la capitale de la France ?", "expected_answer": "Paris"},
        {"query": "Qui a écrit 1984 ?", "expected_answer": "George Orwell"}
    ]

    report = test_suite.run_full_evaluation(test_cases)

    assert report["quality"]["accuracy"] > 0.9, "Accuracy trop basse"
    assert report["quality"]["relevance"] > 0.9, "Relevance trop basse"

def test_performance(test_suite):
    """Test de performance"""
    test_cases = [{"query": f"Question {i}"} for i in range(100)]

    report = test_suite.run_full_evaluation(test_cases)

    assert report["performance"]["latency_p95"] < 2.0, "P95 latency trop élevée"
    assert report["performance"]["throughput"] > 1.0, "Throughput trop faible"

def test_cost(test_suite):
    """Test de coûts"""
    test_cases = [{"query": f"Question {i}"} for i in range(100)]

    report = test_suite.run_full_evaluation(test_cases)

    assert report["cost"]["cost_per_request"] < 0.10, "Coût par requête trop élevé"

def test_reliability(test_suite):
    """Test de fiabilité"""
    test_cases = [{"query": f"Question {i}"} for i in range(100)]

    report = test_suite.run_full_evaluation(test_cases)

    assert report["reliability"]["success_rate"] > 0.95, "Success rate trop faible"
    assert report["reliability"]["hallucination_rate"] < 0.10, "Trop d'hallucinations"

from datasets import load_dataset

def evaluate_on_hotpotqa(agent, num_samples=100):
    """Évalue sur HotpotQA"""

    dataset = load_dataset("hotpot_qa", "fullwiki", split="validation")
    samples = dataset.shuffle().select(range(num_samples))

    correct = 0

    for sample in samples:
        question = sample["question"]
        expected_answer = sample["answer"]

        agent_answer = agent.run(question)

        # Vérifier si la réponse est correcte
        if expected_answer.lower() in agent_answer.lower():
            correct += 1

    accuracy = correct / num_samples

    print(f"\n📊 HotpotQA Results:")
    print(f"Accuracy: {accuracy:.1%} ({correct}/{num_samples})")

    return accuracy

def evaluate_on_fever(agent, num_samples=100):
    """Évalue sur FEVER"""

    dataset = load_dataset("fever", "v1.0", split="paper_test")
    samples = dataset.shuffle().select(range(num_samples))

    correct = 0

    for sample in samples:
        claim = sample["claim"]
        expected_label = sample["label"]  # SUPPORTS, REFUTES, NOT ENOUGH INFO

        prompt = f"Cette affirmation est-elle vraie, fausse, ou impossible à vérifier ? '{claim}'"
        agent_response = agent.run(prompt)

        # Parser la réponse de l'agent
        agent_label = parse_label(agent_response)

        if agent_label == expected_label:
            correct += 1

    accuracy = correct / num_samples

    print(f"\n📊 FEVER Results:")
    print(f"Accuracy: {accuracy:.1%} ({correct}/{num_samples})")

    return accuracy

import random
from datetime import datetime, timedelta

class ABTest:
    """Framework d'A/B testing pour agents"""

    def __init__(self, variant_a, variant_b, split=0.5):
        self.variant_a = variant_a
        self.variant_b = variant_b
        self.split = split

        self.results_a = []
        self.results_b = []

    def run(self, query: str, user_id: str = None) -> str:
        """Exécute un agent selon l'A/B split"""

        # Déterminer la variante
        if user_id:
            # Consistent assignment basé sur user_id
            variant = "a" if hash(user_id) % 2 == 0 else "b"
        else:
            # Random assignment
            variant = "a" if random.random() < self.split else "b"

        # Exécuter
        start_time = datetime.now()

        if variant == "a":
            response = self.variant_a.run(query)
            self.results_a.append({
                "query": query,
                "response": response,
                "latency": (datetime.now() - start_time).total_seconds()
            })
        else:
            response = self.variant_b.run(query)
            self.results_b.append({
                "query": query,
                "response": response,
                "latency": (datetime.now() - start_time).total_seconds()
            })

        return response

    def analyze(self) -> Dict:
        """Analyse les résultats de l'A/B test"""

        import scipy.stats as stats

        # Latences
        latencies_a = [r["latency"] for r in self.results_a]
        latencies_b = [r["latency"] for r in self.results_b]

        # T-test pour latence
        t_stat, p_value = stats.ttest_ind(latencies_a, latencies_b)

        print(f"\n{'='*60}")
        print(f"📊 RÉSULTATS A/B TEST")
        print(f"{'='*60}")

        print(f"\n🅰️ Variant A:")
        print(f"  Samples: {len(self.results_a)}")
        print(f"  Latence moyenne: {statistics.mean(latencies_a):.3f}s")
        print(f"  Latence P95: {sorted(latencies_a)[int(len(latencies_a)*0.95)]:.3f}s")

        print(f"\n🅱️ Variant B:")
        print(f"  Samples: {len(self.results_b)}")
        print(f"  Latence moyenne: {statistics.mean(latencies_b):.3f}s")
        print(f"  Latence P95: {sorted(latencies_b)[int(len(latencies_b)*0.95)]:.3f}s")

        print(f"\n📈 Statistiques:")
        print(f"  t-statistic: {t_stat:.4f}")
        print(f"  p-value: {p_value:.4f}")

        if p_value < 0.05:
            winner = "A" if statistics.mean(latencies_a) < statistics.mean(latencies_b) else "B"
            print(f"\n✅ Résultat significatif : Variant {winner} gagne (p < 0.05)")
        else:
            print(f"\n⚠️ Pas de différence significative (p > 0.05)")

        return {
            "variant_a": {
                "samples": len(self.results_a),
                "latency_mean": statistics.mean(latencies_a)
            },
            "variant_b": {
                "samples": len(self.results_b),
                "latency_mean": statistics.mean(latencies_b)
            },
            "p_value": p_value,
            "significant": p_value < 0.05
        }


# ========== UTILISATION ==========

# Variantes
agent_v1 = MyAgent(model="gpt-3.5-turbo")  # Version actuelle
agent_v2 = MyAgent(model="gpt-4")          # Nouvelle version

# A/B test
ab_test = ABTest(variant_a=agent_v1, variant_b=agent_v2)

# Simuler du trafic
for i in range(1000):
    user_id = f"user_{i % 100}"  # 100 utilisateurs
    query = f"Question {i}"
    ab_test.run(query, user_id=user_id)

# Analyser
ab_test.analyze()

import prometheus_client as prom

class ProductionMonitor:
    """Monitoring production avec Prometheus"""

    def __init__(self):
        # Métriques Prometheus
        self.request_counter = prom.Counter(
            "agent_requests_total",
            "Total number of agent requests"
        )

        self.latency_histogram = prom.Histogram(
            "agent_latency_seconds",
            "Agent response latency"
        )

        self.error_counter = prom.Counter(
            "agent_errors_total",
            "Total number of agent errors",
            ["error_type"]
        )

        self.quality_gauge = prom.Gauge(
            "agent_quality_score",
            "Agent quality score (0-1)"
        )

    def track_request(self, agent, query: str):
        """Track une requête"""

        self.request_counter.inc()

        start = time.time()

        try:
            response = agent.run(query)
            latency = time.time() - start

            self.latency_histogram.observe(latency)

            # Évaluer la qualité (async)
            quality_score = self._evaluate_quality_async(query, response)
            self.quality_gauge.set(quality_score)

            return response

        except Exception as e:
            self.error_counter.labels(error_type=type(e).__name__).inc()
            raise

    def _evaluate_quality_async(self, query: str, response: str) -> float:
        """Évalue la qualité de manière asynchrone"""
        # Utiliser un evaluator rapide ou sample
        # Pour éviter de ralentir les requêtes
        return 0.85  # Placeholder

class RegressionTester:
    """Détecte les régressions entre versions"""

    def __init__(self, baseline_agent, new_agent):
        self.baseline = baseline_agent
        self.new_agent = new_agent

    def test_regression(self, test_cases: List[Dict]) -> Dict:
        """Compare deux versions d'agent"""

        print(f"\n{'='*60}")
        print(f"🔄 TEST DE RÉGRESSION")
        print(f"{'='*60}")

        regressions = []
        improvements = []
        unchanged = []

        for test_case in test_cases:
            query = test_case["query"]

            # Baseline
            baseline_response = self.baseline.run(query)
            baseline_score = self._score_response(query, baseline_response)

            # New version
            new_response = self.new_agent.run(query)
            new_score = self._score_response(query, new_response)

            # Comparer
            diff = new_score - baseline_score

            if diff < -0.1:
                regressions.append({
                    "query": query,
                    "baseline_score": baseline_score,
                    "new_score": new_score,
                    "diff": diff
                })
            elif diff > 0.1:
                improvements.append({
                    "query": query,
                    "baseline_score": baseline_score,
                    "new_score": new_score,
                    "diff": diff
                })
            else:
                unchanged.append({"query": query})

        print(f"\n✅ Améliorations : {len(improvements)}")
        print(f"⚠️ Régressions   : {len(regressions)}")
        print(f"➖ Inchangés     : {len(unchanged)}")

        if regressions:
            print(f"\n⚠️ RÉGRESSIONS DÉTECTÉES:")
            for reg in regressions[:5]:
                print(f"  • {reg['query']}")
                print(f"    Baseline: {reg['baseline_score']:.2f} → New: {reg['new_score']:.2f}")

        return {
            "regressions": regressions,
            "improvements": improvements,
            "unchanged": unchanged
        }

    def _score_response(self, query: str, response: str) -> float:
        """Score une réponse (0-1)"""
        # Utiliser un evaluator
        evaluator = AgentEvaluator()
        result = evaluator.evaluate_response(query, response)
        return (result.accuracy + result.relevance) / 2

from langsmith import Client

client = Client()

# Créer un dataset
dataset = client.create_dataset("customer_support_qa")
client.create_examples(
    dataset_id=dataset.id,
    inputs=[{"question": "..."}],
    outputs=[{"answer": "..."}]
)

# Évaluer
results = client.run_on_dataset(
    dataset_name="customer_support_qa",
    llm_or_chain=my_agent
)

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy

results = evaluate(
    dataset=test_dataset,
    metrics=[faithfulness, answer_relevancy]
)

print(results.to_pandas())