Testing & évaluation d'applications LangChain : du test unitaire à la production

# Code sans tests déployé en production
chain = prompt | llm | parser

# 3 semaines plus tard...
# ❌ OpenAI met à jour GPT-4 → les réponses changent
# ❌ Un prompt change → regression non détectée
# ❌ Un cas edge → l'application crash
# ❌ Coûts explosent → pas de monitoring

pip install pytest pytest-mock langchain-openai

# app/chains.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

def create_translation_chain():
    prompt = ChatPromptTemplate.from_template(
        "Traduis ce texte en {language}:\n\n{text}"
    )
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    return prompt | llm | StrOutputParser()

# tests/test_chains.py
import pytest
from unittest.mock import Mock, patch
from langchain.schema import AIMessage
from app.chains import create_translation_chain

def test_translation_chain():
    # Mock du LLM
    with patch('app.chains.ChatOpenAI') as mock_llm_class:
        # Configurer le mock pour retourner une réponse
        mock_llm = Mock()
        mock_llm.invoke.return_value = AIMessage(content="Bonjour le monde")
        mock_llm_class.return_value = mock_llm

        # Créer et tester la chain
        chain = create_translation_chain()
        result = chain.invoke({
            "language": "français",
            "text": "Hello world"
        })

        # Vérifications
        assert result == "Bonjour le monde"
        mock_llm.invoke.assert_called_once()

# app/agents.py
from langchain.agents import create_react_agent, AgentExecutor
from langchain_openai import ChatOpenAI
from langchain.tools import Tool
from langchain.prompts import PromptTemplate

def search_tool(query: str) -> str:
    """Simule une recherche"""
    return f"Résultats pour: {query}"

def create_search_agent():
    llm = ChatOpenAI(model="gpt-4")
    tools = [
        Tool(
            name="search",
            func=search_tool,
            description="Recherche d'informations"
        )
    ]

    prompt = PromptTemplate.from_template(
        "Tu es un assistant de recherche.\n\n{input}\n\n{agent_scratchpad}"
    )

    agent = create_react_agent(llm, tools, prompt)
    return AgentExecutor(agent=agent, tools=tools, verbose=True)

# tests/test_agents.py
import pytest
from unittest.mock import Mock, patch
from langchain.schema import AgentAction, AgentFinish
from app.agents import create_search_agent

def test_search_agent():
    with patch('app.agents.ChatOpenAI') as mock_llm_class:
        # Mock du LLM pour simuler les étapes de l'agent
        mock_llm = Mock()

        # Étape 1: L'agent décide d'utiliser l'outil
        mock_llm.invoke.side_effect = [
            AgentAction(
                tool="search",
                tool_input="Python LangChain",
                log="Je vais chercher des infos"
            ),
            # Étape 2: L'agent termine
            AgentFinish(
                return_values={"output": "Voici les résultats"},
                log="Recherche terminée"
            )
        ]

        mock_llm_class.return_value = mock_llm

        agent_executor = create_search_agent()
        result = agent_executor.invoke({"input": "Recherche Python LangChain"})

        assert "output" in result
        assert mock_llm.invoke.call_count == 2

# app/parsers.py
from langchain.output_parsers import PydanticOutputParser
from pydantic import BaseModel, Field
from typing import List

class Recipe(BaseModel):
    name: str = Field(description="Nom de la recette")
    ingredients: List[str] = Field(description="Liste des ingrédients")
    steps: List[str] = Field(description="Étapes de préparation")

def create_recipe_parser():
    return PydanticOutputParser(pydantic_object=Recipe)

# tests/test_parsers.py
import pytest
from app.parsers import create_recipe_parser, Recipe

def test_recipe_parser_valid():
    parser = create_recipe_parser()

    # Output simulé du LLM
    llm_output = """
    {
        "name": "Crêpes",
        "ingredients": ["250g farine", "3 œufs", "500ml lait"],
        "steps": ["Mélanger", "Cuire"]
    }
    """

    result = parser.parse(llm_output)

    assert isinstance(result, Recipe)
    assert result.name == "Crêpes"
    assert len(result.ingredients) == 3
    assert len(result.steps) == 2

def test_recipe_parser_invalid():
    parser = create_recipe_parser()

    # Output invalide
    invalid_output = '{"name": "Test"}'  # ingredients manquants

    with pytest.raises(Exception):
        parser.parse(invalid_output)

# tests/test_memory.py
import pytest
from langchain.memory import ConversationBufferMemory

def test_conversation_memory():
    memory = ConversationBufferMemory()

    # Sauvegarder un contexte
    memory.save_context(
        {"input": "Bonjour, je m'appelle Alice"},
        {"output": "Bonjour Alice !"}
    )

    # Vérifier que la mémoire se souvient
    history = memory.load_memory_variables({})
    assert "Alice" in history["history"]

def test_memory_window():
    from langchain.memory import ConversationBufferWindowMemory

    memory = ConversationBufferWindowMemory(k=2)  # 2 derniers échanges

    # Ajouter 3 échanges
    for i in range(3):
        memory.save_context(
            {"input": f"Message {i}"},
            {"output": f"Réponse {i}"}
        )

    history = memory.load_memory_variables({})

    # Vérifier qu'on garde seulement les 2 derniers
    assert "Message 0" not in history["history"]
    assert "Message 1" in history["history"]
    assert "Message 2" in history["history"]

# tests/conftest.py
import pytest
from unittest.mock import Mock
from langchain.schema import AIMessage

@pytest.fixture
def mock_llm():
    """LLM mocké réutilisable"""
    llm = Mock()
    llm.invoke.return_value = AIMessage(content="Réponse mockée")
    return llm

@pytest.fixture
def sample_documents():
    """Documents de test"""
    from langchain.schema import Document
    return [
        Document(page_content="Document 1", metadata={"source": "test1.txt"}),
        Document(page_content="Document 2", metadata={"source": "test2.txt"}),
    ]

# tests/test_with_fixtures.py
def test_with_mock_llm(mock_llm):
    # Utiliser le mock depuis la fixture
    result = mock_llm.invoke("Test")
    assert result.content == "Réponse mockée"

def test_with_documents(sample_documents):
    assert len(sample_documents) == 2
    assert sample_documents[0].metadata["source"] == "test1.txt"

# tests/integration/conftest.py
import pytest
import os

@pytest.fixture(scope="session")
def real_llm():
    """LLM réel pour tests d'intégration"""
    from langchain_openai import ChatOpenAI

    # Utiliser un modèle moins cher pour les tests
    return ChatOpenAI(
        model="gpt-3.5-turbo",
        temperature=0,  # Déterminisme maximal
        max_tokens=100  # Limiter les coûts
    )

@pytest.fixture(scope="session")
def skip_if_no_api_key():
    """Skip les tests si pas de clé API"""
    if not os.getenv("OPENAI_API_KEY"):
        pytest.skip("OPENAI_API_KEY non définie")

# tests/integration/test_rag.py
import pytest
from langchain.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.schema import Document

@pytest.mark.integration
def test_rag_retrieval(real_llm, skip_if_no_api_key):
    # Créer un vectorstore de test
    documents = [
        Document(page_content="Paris est la capitale de la France."),
        Document(page_content="Berlin est la capitale de l'Allemagne."),
        Document(page_content="Rome est la capitale de l'Italie."),
    ]

    embeddings = OpenAIEmbeddings()
    vectorstore = FAISS.from_documents(documents, embeddings)

    # Créer la chain RAG
    qa_chain = RetrievalQA.from_chain_type(
        llm=real_llm,
        retriever=vectorstore.as_retriever(search_kwargs={"k": 1}),
        return_source_documents=True
    )

    # Tester la récupération et la génération
    result = qa_chain.invoke({"query": "Quelle est la capitale de la France ?"})

    assert "Paris" in result["result"]
    assert len(result["source_documents"]) == 1
    assert "Paris" in result["source_documents"][0].page_content

# tests/integration/test_agent_tools.py
import pytest
from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import Tool

@pytest.mark.integration
@pytest.mark.slow  # Marquer les tests lents
def test_agent_with_real_tools(real_llm):
    def calculator(expression: str) -> str:
        """Calcule une expression mathématique"""
        try:
            return str(eval(expression))
        except:
            return "Erreur de calcul"

    tools = [
        Tool(
            name="calculator",
            func=calculator,
            description="Calcule des expressions mathématiques"
        )
    ]

    from langchain.prompts import PromptTemplate
    prompt = PromptTemplate.from_template(
        "Résous ce problème: {input}\n\n{agent_scratchpad}"
    )

    agent = create_react_agent(real_llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, max_iterations=3)

    result = executor.invoke({"input": "Combien font 25 * 4 ?"})

    assert "100" in result["output"]

# pytest.ini
[pytest]
markers =
    unit: Tests unitaires rapides (avec mocks)
    integration: Tests d'intégration (vrais appels API)
    slow: Tests lents (> 5 secondes)
    expensive: Tests coûteux (nombreux appels LLM)

# Exécuter seulement les tests unitaires
pytest -m unit

# Exécuter les tests d'intégration
pytest -m integration

# Exclure les tests coûteux
pytest -m "not expensive"

# Tous les tests sauf les lents
pytest -m "not slow"

# evaluators/custom_evaluators.py
from langsmith.evaluation import EvaluationResult

def exactness_evaluator(run, example):
    """Évalue si la réponse est exactement celle attendue"""
    prediction = run.outputs.get("output", "")
    reference = example.outputs.get("output", "")

    return EvaluationResult(
        key="exactness",
        score=1.0 if prediction.strip() == reference.strip() else 0.0
    )

def contains_keyword_evaluator(run, example):
    """Vérifie que la réponse contient certains mots-clés"""
    prediction = run.outputs.get("output", "").lower()
    keywords = example.outputs.get("expected_keywords", [])

    matches = sum(1 for kw in keywords if kw.lower() in prediction)
    score = matches / len(keywords) if keywords else 0.0

    return EvaluationResult(
        key="keyword_match",
        score=score,
        comment=f"{matches}/{len(keywords)} mots-clés trouvés"
    )

def no_hallucination_evaluator(run, example):
    """Détecte les hallucinations basiques"""
    prediction = run.outputs.get("output", "")

    # Liste de patterns d'hallucination
    hallucination_patterns = [
        "en tant que modèle de langage",
        "je ne peux pas",
        "désolé mais",
        "il semblerait que",  # Hésitation
    ]

    has_hallucination = any(
        pattern in prediction.lower()
        for pattern in hallucination_patterns
    )

    return EvaluationResult(
        key="no_hallucination",
        score=0.0 if has_hallucination else 1.0,
        comment="Hallucination détectée" if has_hallucination else "OK"
    )

# evaluators/llm_evaluators.py
from langchain_openai import ChatOpenAI
from langsmith.evaluation import EvaluationResult

def create_relevance_evaluator():
    """Évalue la pertinence avec un LLM"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)

    def evaluate(run, example):
        query = example.inputs.get("query", "")
        prediction = run.outputs.get("output", "")

        evaluation_prompt = f"""
Sur une échelle de 0 à 10, évalue la pertinence de cette réponse.

Question: {query}
Réponse: {prediction}

Retourne SEULEMENT un nombre entre 0 et 10.
"""

        result = llm.invoke(evaluation_prompt)
        score = float(result.content.strip()) / 10.0

        return EvaluationResult(
            key="relevance",
            score=score,
            comment=f"Score de pertinence: {score*10}/10"
        )

    return evaluate

def create_factuality_evaluator():
    """Vérifie la factualité avec un LLM"""
    llm = ChatOpenAI(model="gpt-4", temperature=0)

    def evaluate(run, example):
        prediction = run.outputs.get("output", "")
        reference = example.outputs.get("reference", "")

        prompt = f"""
Compare ces deux textes et détermine si la prédiction contient des informations factuellement incorrectes.

Référence (vérité terrain): {reference}
Prédiction: {prediction}

Réponds par:
- CORRECT si factuellement correct
- INCORRECT si contient des erreurs factuelles
- PARTIEL si partiellement correct
"""

        result = llm.invoke(prompt).content.strip().upper()

        score_map = {"CORRECT": 1.0, "PARTIEL": 0.5, "INCORRECT": 0.0}
        score = score_map.get(result, 0.5)

        return EvaluationResult(
            key="factuality",
            score=score,
            comment=f"Factualité: {result}"
        )

    return evaluate

# evaluators/performance_evaluators.py
from langsmith.evaluation import EvaluationResult

def latency_evaluator(threshold_seconds=3.0):
    """Vérifie que la latence est acceptable"""
    def evaluate(run, example):
        latency = run.end_time - run.start_time
        latency_seconds = latency.total_seconds()

        score = 1.0 if latency_seconds < threshold_seconds else 0.0

        return EvaluationResult(
            key="latency",
            score=score,
            comment=f"Latence: {latency_seconds:.2f}s (seuil: {threshold_seconds}s)"
        )

    return evaluate

def cost_evaluator(max_cost=0.05):
    """Vérifie que le coût par requête est acceptable"""
    def evaluate(run, example):
        # Extraire les tokens utilisés
        total_tokens = run.outputs.get("total_tokens", 0)

        # Coût estimé pour GPT-4 (exemple)
        cost_per_1k_tokens = 0.03
        estimated_cost = (total_tokens / 1000) * cost_per_1k_tokens

        score = 1.0 if estimated_cost < max_cost else 0.0

        return EvaluationResult(
            key="cost",
            score=score,
            comment=f"Coût estimé: ${estimated_cost:.4f} (max: ${max_cost})"
        )

    return evaluate

# scripts/create_test_dataset.py
from langsmith import Client

client = Client()

# Créer un dataset
dataset_name = "qa-evaluation-dataset"
dataset = client.create_dataset(dataset_name, description="Dataset pour tests Q&A")

# Ajouter des exemples
examples = [
    {
        "inputs": {"query": "Quelle est la capitale de la France ?"},
        "outputs": {"output": "Paris", "expected_keywords": ["Paris", "France"]}
    },
    {
        "inputs": {"query": "Qui a écrit Les Misérables ?"},
        "outputs": {"output": "Victor Hugo", "expected_keywords": ["Victor Hugo"]}
    },
    {
        "inputs": {"query": "Combien font 15 * 8 ?"},
        "outputs": {"output": "120", "expected_keywords": ["120"]}
    },
]

for example in examples:
    client.create_example(
        dataset_id=dataset.id,
        inputs=example["inputs"],
        outputs=example["outputs"]
    )

print(f"Dataset '{dataset_name}' créé avec {len(examples)} exemples")

# scripts/run_evaluation.py
from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

# Notre chain à évaluer
def create_qa_chain():
    prompt = ChatPromptTemplate.from_template("Réponds à cette question: {query}")
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    return prompt | llm | StrOutputParser()

# Évaluateurs
from evaluators.custom_evaluators import (
    exactness_evaluator,
    contains_keyword_evaluator,
    no_hallucination_evaluator
)

# Lancer l'évaluation
client = Client()

results = client.evaluate(
    create_qa_chain,
    data="qa-evaluation-dataset",
    evaluators=[
        exactness_evaluator,
        contains_keyword_evaluator,
        no_hallucination_evaluator
    ],
    experiment_prefix="qa-chain-v1"
)

# Afficher les résultats
print(f"Résultats: {results}")
print(f"Score moyen exactness: {results['results']['exactness']}")
print(f"Score moyen keywords: {results['results']['keyword_match']}")

# scripts/ab_test.py
from langsmith import Client
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# Version A : GPT-3.5
def create_chain_v1():
    prompt = ChatPromptTemplate.from_template("Réponds: {query}")
    llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
    return prompt | llm

# Version B : GPT-4
def create_chain_v2():
    prompt = ChatPromptTemplate.from_template("Réponds précisément: {query}")
    llm = ChatOpenAI(model="gpt-4", temperature=0)
    return prompt | llm

client = Client()

# Évaluer les deux versions
results_v1 = client.evaluate(
    create_chain_v1,
    data="qa-evaluation-dataset",
    evaluators=[exactness_evaluator],
    experiment_prefix="chain-v1-gpt35"
)

results_v2 = client.evaluate(
    create_chain_v2,
    data="qa-evaluation-dataset",
    evaluators=[exactness_evaluator],
    experiment_prefix="chain-v2-gpt4"
)

# Comparer
print("Version A (GPT-3.5):", results_v1['results'])
print("Version B (GPT-4):", results_v2['results'])
print(f"Amélioration: {results_v2['results']['exactness'] - results_v1['results']['exactness']}")

# tests/test_regression.py
import pytest
from langsmith import Client
import os

@pytest.mark.regression
def test_no_regression_from_baseline():
    """Vérifie qu'on n'a pas régressé par rapport à la baseline"""
    client = Client()

    # Récupérer le score de la baseline
    baseline_experiment = "qa-chain-baseline"
    baseline_results = client.read_project(project_name=baseline_experiment)
    baseline_score = baseline_results.feedback_stats.get("exactness", {}).get("avg", 0)

    # Exécuter l'évaluation actuelle
    from scripts.run_evaluation import create_qa_chain
    current_results = client.evaluate(
        create_qa_chain,
        data="qa-evaluation-dataset",
        evaluators=[exactness_evaluator],
        experiment_prefix="qa-chain-current"
    )

    current_score = current_results['results']['exactness']

    # Vérifier qu'on n'a pas régressé de plus de 5%
    regression_threshold = 0.05
    assert current_score >= (baseline_score - regression_threshold), \
        f"Régression détectée: {current_score} vs baseline {baseline_score}"

# tests/test_prompt_snapshots.py
import pytest
import json
import os

@pytest.fixture
def prompt_snapshot_file():
    return "tests/snapshots/prompts.json"

def test_prompt_unchanged(prompt_snapshot_file):
    """Vérifie que les prompts n'ont pas changé accidentellement"""
    from app.chains import create_translation_chain

    chain = create_translation_chain()
    current_prompt = chain.first.format(language="fr", text="test")

    # Charger le snapshot
    if os.path.exists(prompt_snapshot_file):
        with open(prompt_snapshot_file, 'r') as f:
            snapshots = json.load(f)
    else:
        snapshots = {}

    # Vérifier ou créer le snapshot
    if "translation_prompt" in snapshots:
        assert current_prompt == snapshots["translation_prompt"], \
            "Le prompt a changé ! Vérifier si c'est intentionnel."
    else:
        # Première exécution : créer le snapshot
        snapshots["translation_prompt"] = current_prompt
        os.makedirs(os.path.dirname(prompt_snapshot_file), exist_ok=True)
        with open(prompt_snapshot_file, 'w') as f:
            json.dump(snapshots, f, indent=2)
        pytest.skip("Snapshot créé, relancer le test")

# .github/workflows/test.yml
name: Test LangChain Application

on:
  push:
    branches: [ main, develop ]
  pull_request:
    branches: [ main ]

jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest pytest-cov

      - name: Run unit tests
        run: pytest -m unit --cov=app --cov-report=xml

      - name: Upload coverage
        uses: codecov/codecov-action@v3

  integration-tests:
    runs-on: ubuntu-latest
    # Exécuter seulement si les unit tests passent
    needs: unit-tests
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run integration tests
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }}
        run: pytest -m integration --maxfail=3

  evaluation:
    runs-on: ubuntu-latest
    needs: integration-tests
    # Exécuter seulement sur main
    if: github.ref == 'refs/heads/main'
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Run LangSmith evaluation
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
          LANGCHAIN_API_KEY: ${{ secrets.LANGCHAIN_API_KEY }}
        run: python scripts/run_evaluation.py

      - name: Check for regressions
        run: pytest -m regression

# .pre-commit-config.yaml
repos:
  - repo: local
    hooks:
      - id: unit-tests
        name: Run unit tests
        entry: pytest -m unit
        language: system
        pass_filenames: false
        always_run: true

      - id: prompt-snapshot
        name: Check prompt snapshots
        entry: pytest tests/test_prompt_snapshots.py
        language: system
        pass_filenames: false
        files: app/.*\.py$

# Installation
pip install pre-commit
pre-commit install

# Les tests unitaires s'exécutent avant chaque commit
git commit -m "Update chain"  # → tests automatiques

# benchmarks/benchmark_suite.py
import time
from typing import List, Dict
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

class ChainBenchmark:
    def __init__(self, chain, test_cases: List[Dict]):
        self.chain = chain
        self.test_cases = test_cases

    def run(self):
        results = {
            "total_time": 0,
            "avg_time": 0,
            "min_time": float('inf'),
            "max_time": 0,
            "successes": 0,
            "failures": 0,
            "details": []
        }

        for test_case in self.test_cases:
            start = time.time()
            try:
                output = self.chain.invoke(test_case["input"])
                elapsed = time.time() - start

                results["successes"] += 1
                results["total_time"] += elapsed
                results["min_time"] = min(results["min_time"], elapsed)
                results["max_time"] = max(results["max_time"], elapsed)

                results["details"].append({
                    "input": test_case["input"],
                    "output": output,
                    "time": elapsed,
                    "status": "success"
                })
            except Exception as e:
                results["failures"] += 1
                results["details"].append({
                    "input": test_case["input"],
                    "error": str(e),
                    "status": "failure"
                })

        results["avg_time"] = results["total_time"] / len(self.test_cases)
        return results

# Utilisation
if __name__ == "__main__":
    prompt = ChatPromptTemplate.from_template("Réponds: {query}")
    llm = ChatOpenAI(model="gpt-3.5-turbo")
    chain = prompt | llm

    test_cases = [
        {"input": {"query": "Quelle est la capitale de la France ?"}},
        {"input": {"query": "Combien font 2+2 ?"}},
        {"input": {"query": "Qui a peint la Joconde ?"}},
    ]

    benchmark = ChainBenchmark(chain, test_cases)
    results = benchmark.run()

    print(f"Temps moyen: {results['avg_time']:.2f}s")
    print(f"Temps min: {results['min_time']:.2f}s")
    print(f"Temps max: {results['max_time']:.2f}s")
    print(f"Taux de réussite: {results['successes']}/{len(test_cases)}")

# app/monitoring.py
from langsmith import Client
from datetime import datetime, timedelta

class ProductionMonitor:
    def __init__(self, project_name: str):
        self.client = Client()
        self.project_name = project_name

    def get_daily_metrics(self):
        """Récupère les métriques des dernières 24h"""
        end_time = datetime.now()
        start_time = end_time - timedelta(days=1)

        runs = self.client.list_runs(
            project_name=self.project_name,
            start_time=start_time,
            end_time=end_time
        )

        total_runs = 0
        total_latency = 0
        errors = 0

        for run in runs:
            total_runs += 1
            if run.end_time and run.start_time:
                latency = (run.end_time - run.start_time).total_seconds()
                total_latency += latency

            if run.error:
                errors += 1

        return {
            "total_runs": total_runs,
            "avg_latency": total_latency / total_runs if total_runs > 0 else 0,
            "error_rate": errors / total_runs if total_runs > 0 else 0,
            "errors": errors
        }

    def alert_if_degraded(self, threshold_latency=3.0, threshold_error_rate=0.05):
        """Alerte si les métriques se dégradent"""
        metrics = self.get_daily_metrics()

        alerts = []

        if metrics["avg_latency"] > threshold_latency:
            alerts.append(f"⚠️ Latence élevée: {metrics['avg_latency']:.2f}s")

        if metrics["error_rate"] > threshold_error_rate:
            alerts.append(f"⚠️ Taux d'erreur élevé: {metrics['error_rate']*100:.1f}%")

        return alerts

# Script de monitoring quotidien
if __name__ == "__main__":
    monitor = ProductionMonitor("my-production-app")
    alerts = monitor.alert_if_degraded()

    if alerts:
        print("🚨 ALERTES DÉTECTÉES:")
        for alert in alerts:
            print(alert)

        # Envoyer notification Slack
        import requests
        webhook_url = os.getenv("SLACK_WEBHOOK_URL")
        if webhook_url:
            requests.post(webhook_url, json={
                "text": "🚨 Alertes LangChain Production",
                "blocks": [
                    {
                        "type": "section",
                        "text": {"type": "mrkdwn", "text": "\n".join(alerts)}
                    }
                ]
            })

        # Ou envoyer email avec SendGrid/SMTP
        email = os.getenv("ALERT_EMAIL")
        if email:
            import smtplib
            from email.message import EmailMessage

            msg = EmailMessage()
            msg["Subject"] = "🚨 Alertes LangChain Production"
            msg["From"] = "[email protected]"
            msg["To"] = email
            msg.set_content("\n".join(alerts))

            # Avec SMTP (Gmail, SendGrid, etc.)
            # smtp = smtplib.SMTP("smtp.gmail.com", 587)
            # smtp.starttls()
            # smtp.login(os.getenv("SMTP_USER"), os.getenv("SMTP_PASSWORD"))
            # smtp.send_message(msg)
            # smtp.quit()
    else:
        print("✅ Tout va bien")

        /\
       /  \  Évaluations production (peu, lentes, coûteuses)
      /____\
     /      \  Tests d'intégration (moyens, réels)
    /________\
   /          \  Tests unitaires (nombreux, rapides, mocks)
  /__________\

# ✅ BON : Temperature à 0 pour tests
llm = ChatOpenAI(model="gpt-4", temperature=0)

# ❌ MAUVAIS : Temperature élevée = résultats variables
llm = ChatOpenAI(model="gpt-4", temperature=0.9)

# ✅ BON : Chaque test est isolé
@pytest.fixture
def clean_vectorstore():
    vectorstore = create_vectorstore()
    yield vectorstore
    vectorstore.delete_collection()  # Cleanup

# ❌ MAUVAIS : Tests partagent l'état
global_vectorstore = create_vectorstore()  # Réutilisé partout

# ✅ BON : Limiter les tokens pour tests
llm = ChatOpenAI(max_tokens=50)  # Réponses courtes

# ✅ BON : Utiliser des petits datasets
test_docs = documents[:10]  # Pas les 10 000 documents

# ❌ MAUVAIS : Tests lents
llm = ChatOpenAI(max_tokens=2000)  # Trop long pour tests

# tests/test_edge_cases.py
def test_empty_input(qa_chain):
    result = qa_chain.invoke({"query": ""})
    assert result is not None  # Gère l'input vide

def test_very_long_input(qa_chain):
    long_query = "test " * 10000  # Input très long
    result = qa_chain.invoke({"query": long_query})
    # Vérifie que ça ne crash pas

def test_special_characters(qa_chain):
    result = qa_chain.invoke({"query": "Test <script>alert('xss')</script>"})
    assert "<script>" not in result  # Pas d'injection