Gestion d'erreurs et résilience pour applications LangChain : Production-ready error handling

# ❌ DANGER : Application sans gestion d'erreurs
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("Réponds: {query}")
chain = prompt | llm

# En production, cela CRASHERA :
result = chain.invoke({"query": user_input})

# Erreurs possibles :
# - RateLimitError (429) → trop de requêtes
# - Timeout → API lente
# - APIConnectionError → réseau instable
# - AuthenticationError → clé API invalide
# - ServiceUnavailableError (503) → OpenAI down
# - OutputParserException → parsing échoué
# - ValueError → input invalide

# error_handling/error_classifier.py
from enum import Enum
from typing import Optional
import openai

class ErrorType(Enum):
    """Types d'erreurs avec stratégie associée"""
    RATE_LIMIT = "rate_limit"          # → Retry avec backoff
    TIMEOUT = "timeout"                 # → Retry immédiat
    API_CONNECTION = "api_connection"   # → Retry avec backoff
    AUTH = "authentication"             # → Fail fast (pas de retry)
    INVALID_INPUT = "invalid_input"     # → Fail fast
    PARSING = "parsing"                 # → Fallback ou fail
    SERVICE_UNAVAILABLE = "service_unavailable"  # → Retry avec backoff long
    UNKNOWN = "unknown"                 # → Retry avec limite

class ErrorClassifier:
    """Classifie les erreurs pour déterminer la stratégie"""

    @staticmethod
    def classify(error: Exception) -> ErrorType:
        """Classifie une erreur"""
        error_str = str(error).lower()

        # OpenAI errors
        if isinstance(error, openai.RateLimitError):
            return ErrorType.RATE_LIMIT
        elif isinstance(error, openai.Timeout):
            return ErrorType.TIMEOUT
        elif isinstance(error, openai.APIConnectionError):
            return ErrorType.API_CONNECTION
        elif isinstance(error, openai.AuthenticationError):
            return ErrorType.AUTH
        elif isinstance(error, openai.APIStatusError):
            if error.status_code == 503:
                return ErrorType.SERVICE_UNAVAILABLE
            return ErrorType.UNKNOWN

        # Parsing errors
        elif "parsing" in error_str or "outputparser" in error_str:
            return ErrorType.PARSING

        # Validation errors
        elif isinstance(error, (ValueError, TypeError)):
            return ErrorType.INVALID_INPUT

        return ErrorType.UNKNOWN

    @staticmethod
    def should_retry(error_type: ErrorType) -> bool:
        """Détermine si on doit retry"""
        NO_RETRY = {ErrorType.AUTH, ErrorType.INVALID_INPUT}
        return error_type not in NO_RETRY

    @staticmethod
    def get_max_retries(error_type: ErrorType) -> int:
        """Nombre maximum de retries selon le type"""
        return {
            ErrorType.RATE_LIMIT: 5,
            ErrorType.TIMEOUT: 3,
            ErrorType.API_CONNECTION: 4,
            ErrorType.SERVICE_UNAVAILABLE: 6,
            ErrorType.PARSING: 2,
            ErrorType.UNKNOWN: 3,
            ErrorType.AUTH: 0,
            ErrorType.INVALID_INPUT: 0,
        }.get(error_type, 3)

# error_handling/retry.py
import time
import random
from typing import Callable, TypeVar, Optional
from functools import wraps

T = TypeVar('T')

def exponential_backoff(
    max_retries: int = 3,
    base_delay: float = 1.0,
    max_delay: float = 60.0,
    exponential_base: float = 2.0,
    jitter: bool = True
):
    """
    Décorateur pour retry avec exponential backoff

    Args:
        max_retries: Nombre max de tentatives
        base_delay: Délai initial en secondes
        max_delay: Délai maximum
        exponential_base: Base pour calcul exponentiel
        jitter: Ajouter randomness pour éviter thundering herd
    """
    def decorator(func: Callable[..., T]) -> Callable[..., T]:
        @wraps(func)
        def wrapper(*args, **kwargs) -> T:
            last_exception = None

            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_exception = e

                    # Classifier l'erreur
                    from error_handling.error_classifier import ErrorClassifier
                    error_type = ErrorClassifier.classify(e)

                    # Ne pas retry si erreur non-retryable
                    if not ErrorClassifier.should_retry(error_type):
                        raise

                    # Dernier essai
                    if attempt == max_retries:
                        raise

                    # Calculer le délai
                    delay = min(
                        base_delay * (exponential_base ** attempt),
                        max_delay
                    )

                    # Ajouter jitter
                    if jitter:
                        delay = delay * (0.5 + random.random())

                    print(f"⚠️  Tentative {attempt + 1}/{max_retries} échouée: {e}")
                    print(f"   Retry dans {delay:.2f}s...")
                    time.sleep(delay)

            # Si on arrive ici, toutes les tentatives ont échoué
            raise last_exception

        return wrapper
    return decorator

# chains/resilient_chain.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from error_handling.retry import exponential_backoff

class ResilientChain:
    """Chain avec retry automatique"""

    def __init__(self):
        self.llm = ChatOpenAI(
            temperature=0,
            request_timeout=30  # Timeout de 30s
        )
        self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")
        self.chain = self.prompt | self.llm

    @exponential_backoff(max_retries=3, base_delay=1.0, jitter=True)
    def invoke(self, query: str) -> str:
        """Invoke avec retry automatique"""
        result = self.chain.invoke({"query": query})
        return result.content

# Utilisation
chain = ResilientChain()

try:
    response = chain.invoke("Quelle est la capitale de la France ?")
    print(response)
except Exception as e:
    print(f"❌ Échec après tous les retries: {e}")

# error_handling/adaptive_retry.py
import time
import random
from error_handling.error_classifier import ErrorClassifier, ErrorType

class AdaptiveRetry:
    """Retry avec stratégie adaptée au type d'erreur"""

    @staticmethod
    def get_delay(error_type: ErrorType, attempt: int) -> float:
        """Calcule le délai selon le type d'erreur"""
        if error_type == ErrorType.RATE_LIMIT:
            # Backoff agressif pour rate limits
            return min(2 ** attempt, 120)  # Max 2 minutes

        elif error_type == ErrorType.TIMEOUT:
            # Retry rapide pour timeouts
            return 0.5 + (attempt * 0.5)

        elif error_type == ErrorType.SERVICE_UNAVAILABLE:
            # Backoff très long pour service down
            return min(5 ** attempt, 300)  # Max 5 minutes

        else:
            # Backoff standard
            return min(2 ** attempt, 60)

    @classmethod
    def retry_with_adaptive_backoff(cls, func, max_retries: int = 3):
        """Retry avec backoff adaptatif"""
        last_exception = None

        for attempt in range(max_retries + 1):
            try:
                return func()
            except Exception as e:
                last_exception = e
                error_type = ErrorClassifier.classify(e)

                if not ErrorClassifier.should_retry(error_type):
                    raise

                if attempt == max_retries:
                    raise

                # Délai adaptatif
                delay = cls.get_delay(error_type, attempt)

                # Jitter
                delay = delay * (0.5 + random.random())

                print(f"⚠️  {error_type.value}: Retry dans {delay:.1f}s")
                time.sleep(delay)

        raise last_exception

# error_handling/circuit_breaker.py
from enum import Enum
from datetime import datetime, timedelta
from typing import Callable, Optional
import threading

class CircuitState(Enum):
    CLOSED = "closed"      # Fonctionne normalement
    OPEN = "open"          # Service down, rejette les appels
    HALF_OPEN = "half_open"  # Test si service revenu

class CircuitBreaker:
    """
    Circuit breaker pour protéger contre services défaillants

    États:
    - CLOSED: Normal, laisse passer les appels
    - OPEN: Trop d'erreurs, rejette les appels immédiatement
    - HALF_OPEN: Test après timeout pour voir si service revenu
    """

    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: float = 60.0,
        expected_exception: type = Exception
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.expected_exception = expected_exception

        self.failure_count = 0
        self.last_failure_time: Optional[datetime] = None
        self.state = CircuitState.CLOSED
        self.lock = threading.Lock()

    def call(self, func: Callable, *args, **kwargs):
        """Appelle la fonction via le circuit breaker"""
        with self.lock:
            # Si circuit OPEN, vérifier timeout
            if self.state == CircuitState.OPEN:
                if self._should_attempt_reset():
                    self.state = CircuitState.HALF_OPEN
                    print("🔶 Circuit HALF_OPEN: Test de récupération...")
                else:
                    raise CircuitBreakerOpenError(
                        f"Circuit breaker OPEN. Retry dans "
                        f"{self._time_until_retry():.0f}s"
                    )

        # Tenter l'appel
        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except self.expected_exception as e:
            self._on_failure()
            raise

    def _should_attempt_reset(self) -> bool:
        """Détermine si on doit tenter de reset le circuit"""
        return (
            self.last_failure_time is not None
            and datetime.now() - self.last_failure_time
            >= timedelta(seconds=self.recovery_timeout)
        )

    def _time_until_retry(self) -> float:
        """Temps restant avant retry"""
        if self.last_failure_time is None:
            return 0
        elapsed = (datetime.now() - self.last_failure_time).total_seconds()
        return max(0, self.recovery_timeout - elapsed)

    def _on_success(self):
        """Succès: reset le compteur"""
        with self.lock:
            self.failure_count = 0
            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.CLOSED
                print("✅ Circuit CLOSED: Service récupéré")

    def _on_failure(self):
        """Échec: incrémenter le compteur"""
        with self.lock:
            self.failure_count += 1
            self.last_failure_time = datetime.now()

            if self.state == CircuitState.HALF_OPEN:
                # Échec pendant test → retour à OPEN
                self.state = CircuitState.OPEN
                print("❌ Circuit OPEN: Service toujours down")

            elif self.failure_count >= self.failure_threshold:
                # Trop d'échecs → OPEN
                self.state = CircuitState.OPEN
                print(f"🔴 Circuit OPEN: {self.failure_count} échecs consécutifs")

class CircuitBreakerOpenError(Exception):
    """Levée quand le circuit breaker est ouvert"""
    pass

# Créer un circuit breaker pour OpenAI
openai_breaker = CircuitBreaker(
    failure_threshold=5,      # 5 échecs → OPEN
    recovery_timeout=60.0,    # Test après 60s
    expected_exception=Exception
)

def call_llm_with_breaker(query: str) -> str:
    """Appel LLM protégé par circuit breaker"""
    def _call():
        chain = prompt | llm
        result = chain.invoke({"query": query})
        return result.content

    return openai_breaker.call(_call)

# Utilisation
try:
    response = call_llm_with_breaker("Test")
    print(response)
except CircuitBreakerOpenError as e:
    print(f"⚠️  Service temporairement indisponible: {e}")
    # Utiliser un fallback

# chains/fallback_chain.py
from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain.prompts import ChatPromptTemplate
from typing import Optional

class FallbackChain:
    """Chain avec fallback automatique"""

    def __init__(self):
        # Primary: OpenAI GPT-4
        self.primary_llm = ChatOpenAI(model="gpt-4", temperature=0)

        # Fallback 1: OpenAI GPT-3.5
        self.fallback1_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

        # Fallback 2: Claude
        self.fallback2_llm = ChatAnthropic(model="claude-3-sonnet-20240229")

        self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")

    def invoke(self, query: str) -> dict:
        """Invoke avec fallback automatique"""
        llms = [
            ("GPT-4", self.primary_llm),
            ("GPT-3.5", self.fallback1_llm),
            ("Claude", self.fallback2_llm)
        ]

        errors = []

        for name, llm in llms:
            try:
                chain = self.prompt | llm
                result = chain.invoke({"query": query})
                return {
                    "response": result.content,
                    "model_used": name,
                    "fallback_used": name != "GPT-4"
                }
            except Exception as e:
                print(f"⚠️  {name} échoué: {e}")
                errors.append(f"{name}: {e}")
                continue

        # Tous les fallbacks ont échoué
        raise Exception(f"Tous les modèles ont échoué: {errors}")

# chains/graceful_degradation.py
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

class GracefulDegradationChain:
    """Chain avec dégradation gracieuse de qualité"""

    def __init__(self):
        # Niveaux de qualité décroissants
        self.tiers = [
            {
                "name": "Premium",
                "model": "gpt-4",
                "temperature": 0,
                "max_tokens": 2000
            },
            {
                "name": "Standard",
                "model": "gpt-3.5-turbo",
                "temperature": 0,
                "max_tokens": 1000
            },
            {
                "name": "Basic",
                "model": "gpt-3.5-turbo",
                "temperature": 0.3,
                "max_tokens": 500
            },
        ]

        self.prompt = ChatPromptTemplate.from_template(
            "Réponds de manière concise: {query}"
        )

    def invoke(self, query: str) -> dict:
        """Invoke avec dégradation gracieuse"""
        for tier in self.tiers:
            try:
                llm = ChatOpenAI(
                    model=tier["model"],
                    temperature=tier["temperature"],
                    max_tokens=tier["max_tokens"]
                )
                chain = self.prompt | llm
                result = chain.invoke({"query": query})

                return {
                    "response": result.content,
                    "quality_tier": tier["name"],
                    "degraded": tier["name"] != "Premium"
                }
            except Exception as e:
                print(f"⚠️  Tier {tier['name']} échoué, dégradation...")
                continue

        # Fallback ultime: réponse statique
        return {
            "response": "Service temporairement indisponible. Veuillez réessayer.",
            "quality_tier": "Emergency",
            "degraded": True
        }

# chains/timeout_chain.py
import signal
from contextlib import contextmanager
from typing import Optional

class TimeoutError(Exception):
    """Levée quand un timeout est atteint"""
    pass

@contextmanager
def time_limit(seconds: int):
    """Context manager pour timeout"""
    def signal_handler(signum, frame):
        raise TimeoutError(f"Timeout après {seconds}s")

    # Configurer l'alarme
    signal.signal(signal.SIGALRM, signal_handler)
    signal.alarm(seconds)

    try:
        yield
    finally:
        # Désactiver l'alarme
        signal.alarm(0)

class TimeoutChain:
    """Chain avec timeouts configurables"""

    def __init__(
        self,
        llm_timeout: int = 30,
        total_timeout: int = 60
    ):
        self.llm = ChatOpenAI(
            request_timeout=llm_timeout  # Timeout pour chaque appel LLM
        )
        self.total_timeout = total_timeout
        self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")

    def invoke(self, query: str, timeout: Optional[int] = None) -> str:
        """Invoke avec timeout global"""
        timeout = timeout or self.total_timeout

        try:
            with time_limit(timeout):
                chain = self.prompt | self.llm
                result = chain.invoke({"query": query})
                return result.content
        except TimeoutError as e:
            raise TimeoutError(
                f"Chain timeout après {timeout}s. "
                f"Augmentez le timeout ou simplifiez la requête."
            )

# chains/async_timeout_chain.py
import asyncio
from langchain_openai import ChatOpenAI

class AsyncTimeoutChain:
    """Chain asynchrone avec timeout"""

    def __init__(self, timeout: float = 30.0):
        self.llm = ChatOpenAI(temperature=0)
        self.timeout = timeout
        self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")

    async def ainvoke(self, query: str) -> str:
        """Invoke async avec timeout"""
        chain = self.prompt | self.llm

        try:
            result = await asyncio.wait_for(
                chain.ainvoke({"query": query}),
                timeout=self.timeout
            )
            return result.content
        except asyncio.TimeoutError:
            raise TimeoutError(f"Timeout après {self.timeout}s")

# Utilisation
async def main():
    chain = AsyncTimeoutChain(timeout=10.0)
    try:
        response = await chain.ainvoke("Question complexe...")
        print(response)
    except TimeoutError:
        print("⏱️  Timeout")

# asyncio.run(main())

# chains/self_healing_chain.py
from typing import Optional, Dict
import time

class SelfHealingChain:
    """Chain avec auto-correction"""

    def __init__(self):
        self.llm = ChatOpenAI(temperature=0)
        self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")
        self.chain = self.prompt | self.llm

        # Statistiques pour auto-healing
        self.error_count = 0
        self.success_count = 0
        self.last_errors = []

    def invoke(self, query: str, max_attempts: int = 3) -> Dict:
        """Invoke avec auto-correction"""
        for attempt in range(max_attempts):
            try:
                result = self._invoke_with_monitoring(query)
                self._on_success()
                return result

            except OutputParserException as e:
                # Erreur de parsing → ajuster le prompt
                print(f"⚠️  Parsing error, ajustement du prompt...")
                self._adjust_prompt_for_parsing()
                continue

            except Exception as e:
                self._on_error(e)

                # Si trop d'erreurs, mode dégradé
                if self.error_count > 10:
                    return self._degraded_response(query)

                raise

    def _invoke_with_monitoring(self, query: str) -> Dict:
        """Invoke avec monitoring"""
        start = time.time()
        result = self.chain.invoke({"query": query})
        latency = time.time() - start

        return {
            "response": result.content,
            "latency_ms": latency * 1000,
            "model": "gpt-4"
        }

    def _on_success(self):
        """Succès → reset compteur d'erreurs"""
        self.success_count += 1
        if self.success_count > 5:
            self.error_count = max(0, self.error_count - 1)

    def _on_error(self, error: Exception):
        """Erreur → enregistrer"""
        self.error_count += 1
        self.last_errors.append({
            "timestamp": time.time(),
            "error": str(error)
        })

        # Garder seulement les 10 dernières
        self.last_errors = self.last_errors[-10:]

    def _adjust_prompt_for_parsing(self):
        """Ajuster le prompt pour améliorer le parsing"""
        # Ajouter des instructions de format
        self.prompt = ChatPromptTemplate.from_template(
            "Réponds à cette question de manière structurée.\n"
            "Format: Réponse simple et claire.\n\n"
            "Question: {query}"
        )
        self.chain = self.prompt | self.llm

    def _degraded_response(self, query: str) -> Dict:
        """Réponse dégradée si trop d'erreurs"""
        return {
            "response": "Service en mode dégradé. Veuillez réessayer plus tard.",
            "degraded": True,
            "error_count": self.error_count
        }

# monitoring/error_tracker.py
from collections import defaultdict
from datetime import datetime, timedelta
from typing import Dict, List
import json

class ErrorTracker:
    """Track et analyse les erreurs"""

    def __init__(self):
        self.errors: List[Dict] = []
        self.error_counts = defaultdict(int)

    def track_error(
        self,
        error: Exception,
        context: Dict = None
    ):
        """Enregistre une erreur"""
        from error_handling.error_classifier import ErrorClassifier

        error_type = ErrorClassifier.classify(error)

        error_entry = {
            "timestamp": datetime.now().isoformat(),
            "type": error_type.value,
            "message": str(error),
            "context": context or {}
        }

        self.errors.append(error_entry)
        self.error_counts[error_type.value] += 1

        # Log en JSON
        print(json.dumps(error_entry))

    def get_error_rate(self, window_minutes: int = 60) -> float:
        """Calcule le taux d'erreur sur une période"""
        cutoff = datetime.now() - timedelta(minutes=window_minutes)

        recent_errors = [
            e for e in self.errors
            if datetime.fromisoformat(e["timestamp"]) > cutoff
        ]

        return len(recent_errors) / max(window_minutes, 1)

    def get_top_errors(self, limit: int = 5) -> List[tuple]:
        """Top N erreurs les plus fréquentes"""
        sorted_errors = sorted(
            self.error_counts.items(),
            key=lambda x: x[1],
            reverse=True
        )
        return sorted_errors[:limit]

    def should_alert(self, threshold_per_minute: float = 1.0) -> bool:
        """Détermine si on doit alerter"""
        error_rate = self.get_error_rate(window_minutes=5)
        return error_rate > threshold_per_minute

# chains/production_chain.py
from error_handling.retry import exponential_backoff
from error_handling.circuit_breaker import CircuitBreaker
from monitoring.error_tracker import ErrorTracker

class ProductionChain:
    """Chain production-ready avec toutes les protections"""

    def __init__(self):
        self.llm = ChatOpenAI(
            temperature=0,
            request_timeout=30
        )
        self.prompt = ChatPromptTemplate.from_template("Réponds: {query}")

        # Protection
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60.0
        )

        # Monitoring
        self.error_tracker = ErrorTracker()

        # Fallback
        self.fallback_llm = ChatOpenAI(model="gpt-3.5-turbo")

    @exponential_backoff(max_retries=3, base_delay=1.0, jitter=True)
    def invoke(self, query: str) -> Dict:
        """Invoke avec toutes les protections"""
        try:
            # Via circuit breaker
            def _call():
                chain = self.prompt | self.llm
                return chain.invoke({"query": query})

            result = self.circuit_breaker.call(_call)

            return {
                "response": result.content,
                "model": "gpt-4",
                "fallback_used": False
            }

        except Exception as e:
            # Tracker l'erreur
            self.error_tracker.track_error(e, context={"query": query})

            # Alerter si trop d'erreurs
            if self.error_tracker.should_alert():
                print("🚨 ALERTE: Taux d'erreur élevé!")

            # Fallback
            try:
                print("⚠️  Utilisation du fallback (GPT-3.5)...")
                chain = self.prompt | self.fallback_llm
                result = chain.invoke({"query": query})

                return {
                    "response": result.content,
                    "model": "gpt-3.5-turbo",
                    "fallback_used": True
                }
            except Exception as fallback_error:
                # Fallback ultime
                return {
                    "response": "Service temporairement indisponible.",
                    "model": "emergency",
                    "fallback_used": True,
                    "error": str(fallback_error)
                }

# error_handling/bulkhead.py
import threading
from queue import Queue
from typing import Callable

class Bulkhead:
    """Isolation des ressources pour éviter cascade failure"""

    def __init__(self, max_concurrent: int = 10, queue_size: int = 100):
        self.semaphore = threading.Semaphore(max_concurrent)
        self.queue = Queue(maxsize=queue_size)
        self.max_concurrent = max_concurrent

    def execute(self, func: Callable, *args, **kwargs):
        """Exécute avec limitation de concurrence"""
        if not self.semaphore.acquire(blocking=False):
            raise BulkheadFullError(
                f"Bulkhead plein ({self.max_concurrent} requêtes en cours)"
            )

        try:
            return func(*args, **kwargs)
        finally:
            self.semaphore.release()

class BulkheadFullError(Exception):
    pass

# error_handling/retry_budget.py
from datetime import datetime, timedelta

class RetryBudget:
    """Limite le nombre de retries sur une période"""

    def __init__(self, max_retries_per_minute: int = 100):
        self.max_retries = max_retries_per_minute
        self.retry_history = []

    def can_retry(self) -> bool:
        """Vérifie si on a du budget retry"""
        # Nettoyer l'historique (> 1 minute)
        cutoff = datetime.now() - timedelta(minutes=1)
        self.retry_history = [
            t for t in self.retry_history if t > cutoff
        ]

        return len(self.retry_history) < self.max_retries

    def consume(self):
        """Consomme 1 retry du budget"""
        if not self.can_retry():
            raise RetryBudgetExceededError(
                "Budget de retry épuisé (trop de retries par minute)"
            )

        self.retry_history.append(datetime.now())

class RetryBudgetExceededError(Exception):
    pass