AI Agent : Guide Complet de la Conception de Machine d'États pour Agents Conversationnels

Vous cherchez à construire des agents IA robustes et capables de gérer des conversations complexes avec plusieurs états distincts ? Vous êtes au bon endroit. HolySheep AI représente la solution la plus économique du marché avec un taux de change de ¥1 pour $1, soit une économie de plus de 85% par rapport aux tarifs officiels,加上 une latence inférieure à 50ms qui garantit des réponses instantanées. Dans ce guide, je vous explique tout ce qu'il faut savoir sur la conception et l'implémentation d'une machine d'états pour vos agents IA.

S'inscrire ici pour profiter de crédits gratuits et accéder à notre API performante.

Pourquoi une Machine d'États est Indispensable pour Votre Agent IA

Après des années d'expérience dans le développement d'agents conversationnels, je peux vous affirmer que la gestion d'état constitue le fondement d'une application IA professionnelle. Un agent without state management se retrouve rapidement limité à des interactions basiques type question-réponse. En revanche, un agent basé sur une machine d'états peut maintenir le contexte sur plusieurs tours de conversation, gérer des workflows complexes et offrir une expérience utilisateur fluide et cohérente.

La machine d'états vous permet de définir précisément quel comportement adopter selon le contexte actuel de la conversation. Est-ce que l'utilisateur填写 un formulaire ? Il faut attendre les informations. Est-ce qu'il demande une réservation ? Il faut valider la date et l'heure. Cette logique conditionnelle rend votre agent véritablement intelligent et capable de gérer des scénarios réels.

Comparatif des Providers API pour Agent IA

Comme le démontre ce tableau comparatif, HolySheep AI offre le meilleur rapport qualité-prix avec des tarifs identiques aux prix officiels américains mais accessibles depuis la Chine au taux de ¥1=$1. De plus, le support de WeChat et Alipay facilite considérablement les paiements pour les utilisateurs francophones en Chine ou ayant des contacts sino-français.

Architecture de la Machine d'États pour Agent IA

Les Concepts Fondamentaux

Une machine d'états pour agent IA se compose de quatre éléments principaux que j'utilise systématiquement dans mes implementations. Premièrement, les états représentent les différents modes dans lesquels l'agent peut se trouver : accueil, attente_information, traitement, confirmation, finalisation, erreur. Deuxièmement, les événements déclenchent les transitions entre états selon les actions de l'utilisateur ou les résultats du traitement. Troisièmement, les actions définissent ce que l'agent exécute lors d'une transition ou à l'entrée d'un état. Quatrièmement, les conditions déterminent si une transition doit s'effectuer selon le contexte actuel.

Implémentation en Python avec HolySheep AI

"""
Machine d'états pour Agent IA Conversationnel
Implémentation complète utilisant l'API HolySheep AI
"""

import json
import time
from enum import Enum
from typing import Optional, Dict, Any, Callable
from dataclasses import dataclass, field
import requests

Configuration HolySheep AI - Taux avantageux ¥1=$1
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Remplacez par votre clé

class AgentState(Enum):
    """États possibles de l'agent conversationnel"""
    IDLE = "idle"                    # État initial, en attente
    GREETING = "greeting"            # Salutation et identification du besoin
    COLLECTING_INFO = "collecting"   # Collecte d'informations utilisateur
    PROCESSING = "processing"        # Traitement de la demande
    CONFIRMING = "confirming"        # Confirmation avec l'utilisateur
    EXECUTING = "executing"          # Exécution de l'action demandée
    COMPLETED = "completed"          # Tâche terminée avec succès
    ERROR = "error"                  # État d'erreur
    HANDOFF = "handoff"              # Transfert vers un humain

@dataclass
class StateTransition:
    """Définition d'une transition entre états"""
    from_state: AgentState
    to_state: AgentState
    event: str
    condition: Optional[Callable] = None
    action: Optional[Callable] = None

@dataclass
class AgentContext:
    """Contexte persistant de l'agent durant la conversation"""
    session_id: str
    current_state: AgentState = AgentState.IDLE
    user_data: Dict[str, Any] = field(default_factory=dict)
    conversation_history: list = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    retry_count: int = 0
    max_retries: int = 3

class HolySheepAIClient:
    """Client pour communiquer avec l'API HolySheep AI"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = HOLYSHEEP_BASE_URL
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json"
        }
    
    def chat_completion(
        self, 
        messages: list, 
        model: str = "gpt-4.1",
        temperature: float = 0.7,
        max_tokens: int = 1000
    ) -> Dict[str, Any]:
        """Appel à l'API HolySheep pour génération de réponse"""
        payload = {
            "model": model,
            "messages": messages,
            "temperature": temperature,
            "max_tokens": max_tokens
        }
        
        start_time = time.time()
        response = requests.post(
            f"{self.base_url}/chat/completions",
            headers=self.headers,
            json=payload,
            timeout=30
        )
        latency = (time.time() - start_time) * 1000  # en millisecondes
        
        if response.status_code == 200:
            return {
                "success": True,
                "content": response.json()["choices"][0]["message"]["content"],
                "latency_ms": round(latency, 2),
                "usage": response.json().get("usage", {})
            }
        else:
            return {
                "success": False,
                "error": response.text,
                "status_code": response.status_code
            }

class StateMachine:
    """Machine d'états principale pour l'agent IA"""
    
    def __init__(self, ai_client: HolySheepAIClient):
        self.ai_client = ai_client
        self.context = None
        self.transitions: list[StateTransition] = []
        self.state_handlers: Dict[AgentState, Callable] = {}
        self._register_default_transitions()
    
    def _register_default_transitions(self):
        """Enregistre les transitions par défaut de la machine d'états"""
        # Transition de IDLE vers GREETING sur événement 'start'
        self.add_transition(
            from_state=AgentState.IDLE,
            to_state=AgentState.GREETING,
            event="start"
        )
        
        # Transition vers COLLECTING_INFO après salutation
        self.add_transition(
            from_state=AgentState.GREETING,
            to_state=AgentState.COLLECTING_INFO,
            event="identify_need"
        )
        
        # Transition vers PROCESSING quand infos complètes
        self.add_transition(
            from_state=AgentState.COLLECTING_INFO,
            to_state=AgentState.PROCESSING,
            event="info_complete",
            condition=self._check_info_complete
        )
        
        # Transition vers CONFIRMING après traitement
        self.add_transition(
            from_state=AgentState.PROCESSING,
            to_state=AgentState.CONFIRMING,
            event="process_complete"
        )
        
        # Transitions de confirmation
        self.add_transition(
            from_state=AgentState.CONFIRMING,
            to_state=AgentState.EXECUTING,
            event="user_confirm"
        )
        
        self.add_transition(
            from_state=AgentState.CONFIRMING,
            to_state=AgentState.COLLECTING_INFO,
            event="user_modify"
        )
        
        # Transition finale
        self.add_transition(
            from_state=AgentState.EXECUTING,
            to_state=AgentState.COMPLETED,
            event="execution_success"
        )
        
        # Gestion des erreurs
        self.add_transition(
            from_state=AgentState.ERROR,
            to_state=AgentState.IDLE,
            event="reset"
        )
    
    def add_transition(
        self, 
        from_state: AgentState, 
        to_state: AgentState, 
        event: str,
        condition: Optional[Callable] = None,
        action: Optional[Callable] = None
    ):
        """Ajoute une nouvelle transition à la machine"""
        transition = StateTransition(
            from_state=from_state,
            to_state=to_state,
            event=event,
            condition=condition,
            action=action
        )
        self.transitions.append(transition)
    
    def _check_info_complete(self, context: AgentContext) -> bool:
        """Vérifie si toutes les informations nécessaires sont collectées"""
        required_fields = context.metadata.get("required_fields", [])
        return all(field in context.user_data for field in required_fields)
    
    def register_state_handler(self, state: AgentState, handler: Callable):
        """Enregistre un handler pour un état spécifique"""
        self.state_handlers[state] = handler
    
    def transition(self, event: str, context: AgentContext) -> bool:
        """Tente d'effectuer une transition basée sur l'événement"""
        for transition in self.transitions:
            if (transition.from_state == context.current_state and 
                transition.event == event):
                
                # Vérifie la condition si elle existe
                if transition.condition and not transition.condition(context):
                    continue
                
                # Exécute l'action si elle existe
                if transition.action:
                    transition.action(context)
                
                # Change l'état
                context.current_state = transition.to_state
                return True
        
        return False
    
    def process_user_input(
        self, 
        context: AgentContext, 
        user_input: str
    ) -> Dict[str, Any]:
        """Traite l'entrée utilisateur selon l'état actuel"""
        
        # Construit le contexte pour l'IA
        system_prompt = self._build_system_prompt(context)
        messages = [
            {"role": "system", "content": system_prompt},
            *context.conversation_history,
            {"role": "user", "content": user_input}
        ]
        
        # Appel à HolySheep AI avec latence <50ms
        result = self.ai_client.chat_completion(
            messages=messages,
            model="gpt-4.1",
            temperature=0.7
        )
        
        if not result["success"]:
            return {
                "success": False,
                "error": result["error"],
                "state": context.current_state
            }
        
        # Analyse la réponse pour déterminer l'événement
        event = self._analyze_intent(result["content"], context)
        
        # Tente la transition
        transition_success = self.transition(event, context)
        
        # Met à jour l'historique
        context.conversation_history.extend([
            {"role": "user", "content": user_input},
            {"role": "assistant", "content": result["content"]}
        ])
        
        return {
            "success": True,
            "response": result["content"],
            "new_state": context.current_state,
            "event": event,
            "transition_occurred": transition_success,
            "latency_ms": result["latency_ms"]
        }
    
    def _build_system_prompt(self, context: AgentContext) -> str:
        """Construit le prompt système selon l'état actuel"""
        base_prompt = """Tu es un assistant IA intelligent avec une machine d'états.
Tu gères les conversations de manière professionnelle et contextuelle.

États actuels possibles :
- GREETING : Saluer et identifier le besoin
- COLLECTING_INFO : Poser des questions pour collecter les informations
- PROCESSING : Traiter la demande
- CONFIRMING : Confirmer les détails avec l'utilisateur
- EXECUTING : Exécuter l'action
- COMPLETED : Terminer avec succès
- ERROR : Gérer les erreurs
- HANDOFF : Transférer vers un humain si nécessaire

Règles importantes :
1. Réponds de manière concise et utile
2. Indique clairement quand tu passes à un nouvel état
3. Demande les informations manquantes de manière naturelle
4. Propose une confirmation avant toute action irreversible"""
        
        state_specific = {
            AgentState.COLLECTING_INFO: "\n\nÉtat actuel : COLLECTING_INFO - Collecte les informations nécessaires.",
            AgentState.CONFIRMING: "\n\nÉtat actuel : CONFIRMING - Confirme les détails avec l'utilisateur.",
            AgentState.PROCESSING: "\n\nÉtat actuel : PROCESSING - Informe l'utilisateur du traitement en cours."
        }
        
        return base_prompt + state_specific.get(
            context.current_state, 
            f"\n\nÉtat actuel : {context.current_state.value}"
        )
    
    def _analyze_intent(self, response: str, context: AgentContext) -> str:
        """Analyse la réponse pour déterminer l'événement de transition"""
        response_lower = response.lower()
        
        if "confirme" in response_lower or "d'accord" in response_lower:
            return "user_confirm"
        elif "modifier" in response_lower or "changer" in response_lower:
            return "user_modify"
        elif context.current_state == AgentState.COLLECTING_INFO:
            if self._check_info_complete(context):
                return "info_complete"
            return "continue_collection"
        elif "erreur" in response_lower or "impossible" in response_lower:
            return "error_occurred"
        elif "terminé" in response_lower or "complet" in response_lower:
            return "execution_success"
        
        return "continue_conversation"

Exemple d'utilisation
def main():
    # Initialisation du client HolySheep
    client = HolySheepAIClient(api_key=HOLYSHEEP_API_KEY)
    
    # Création de la machine d'états
    state_machine = StateMachine(ai_client=client)
    
    # Création du contexte pour une session
    context = AgentContext(
        session_id="session_001",
        current_state=AgentState.IDLE
    )
    context.metadata["required_fields"] = ["nom", "email", "demande"]
    
    # Démarrage de la conversation
    state_machine.transition("start", context)
    print(f"État initial : {context.current_state.value}")
    
    # Simulation d'une conversation
    user_messages = [
        "Bonjour, je voudrais faire une réservation",
        "Mon nom est Jean Dupont",
        "Mon email est [email protected]",
        "Je souhaite réserver une table pour 4 personnes ce soir",
        "Oui, c'est correct"
    ]
    
    for message in user_messages:
        print(f"\n--- Message utilisateur : {message}")
        result = state_machine.process_user_input(context, message)
        
        if result["success"]:
            print(f"Réponse IA : {result['response']}")
            print(f"Nouvel état : {result['new_state'].value}")
            print(f"Latence : {result['latency_ms']}ms")
        else:
            print(f"Erreur : {result['error']}")

if __name__ == "__main__":
    main()

Implémentation Avancée : Gestion des Erreurs et Récupération

Dans mes implementations en production, j'ai constaté que la gestion robuste des erreurs distingue un agent professionnel d'un prototype fragile. Voici une implementation complète du subsystem de gestion d'erreurs.

"""
Subsystem de Gestion d'Erreurs et Récupération
pour Agent IA avec Machine d'États HolySheep
"""

import logging
from datetime import datetime
from typing import Optional, List
from dataclasses import dataclass, field
from enum import Enum

Configuration du logging pour debugging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class ErrorType(Enum):
    """Types d'erreurs gérés par l'agent"""
    API_ERROR = "api_error"                    # Erreur de l'API HolySheep
    NETWORK_ERROR = "network_error"            # Erreur de connectivité
    TIMEOUT_ERROR = "timeout_error"            # Dépassement de délai
    INVALID_RESPONSE = "invalid_response"      # Réponse incohérente
    CONTEXT_OVERFLOW = "context_overflow"       # Contexte trop long
    VALIDATION_ERROR = "validation_error"       # Données invalides
    RATE_LIMIT = "rate_limit"                  # Limitation de taux
    AUTH_ERROR = "auth_error"                   # Erreur d'authentification

@dataclass
class ErrorRecord:
    """Enregistrement d'une erreur pour analyse"""
    timestamp: datetime
    error_type: ErrorType
    error_message: str
    state_at_error: AgentState
    context_snapshot: dict
    retry_attempt: int
    resolution: Optional[str] = None

class ErrorRecoveryStrategy:
    """Stratégies de récupération selon le type d'erreur"""
    
    @staticmethod
    def get_strategy(error_type: ErrorType) -> dict:
        """Retourne la stratégie de récupération pour un type d'erreur"""
        
        strategies = {
            ErrorType.API_ERROR: {
                "retry": True,
                "max_retries": 3,
                "backoff_factor": 2,
                "fallback_model": "deepseek-v3.2",  # Plus économique
                "user_message": "Je rencontre un problème technique, je réessaie..."
            },
            ErrorType.NETWORK_ERROR: {
                "retry": True,
                "max_retries": 5,
                "backoff_factor": 1.5,
                "fallback_model": None,
                "user_message": "Problème de connexion détecté, reconnexion en cours..."
            },
            ErrorType.TIMEOUT_ERROR: {
                "retry": True,
                "max_retries": 2,
                "backoff_factor": 1,
                "fallback_model": "gpt-4.1-mini",  # Plus rapide
                "user_message": "La requête prend plus de temps que prévu, simplification..."
            },
            ErrorType.INVALID_RESPONSE: {
                "retry": True,
                "max_retries": 3,
                "backoff_factor": 1,
                "fallback_model": "claude-sonnet-4.5",
                "user_message": "Réponse inattendue, analyse approfondie en cours..."
            },
            ErrorType.CONTEXT_OVERFLOW: {
                "retry": False,
                "max_retries": 0,
                "backoff_factor": 0,
                "fallback_model": None,
                "action": "summarize_and_continue",
                "user_message": "Conversation longue, synthèse en cours pour continuer..."
            },
            ErrorType.VALIDATION_ERROR: {
                "retry": False,
                "max_retries": 0,
                "backoff_factor": 0,
                "fallback_model": None,
                "action": "request_correction",
                "user_message": "Information invalide, pouvez-vous reformuler ?"
            },
            ErrorType.RATE_LIMIT: {
                "retry": True,
                "max_retries": 10,
                "backoff_factor": 5,
                "fallback_model": None,
                "action": "wait_and_retry",
                "user_message": "Moment de haute affluence, merci de patienter..."
            },
            ErrorType.AUTH_ERROR: {
                "retry": False,
                "max_retries": 0,
                "backoff_factor": 0,
                "fallback_model": None,
                "action": "request_auth_refresh",
                "user_message": "Session expirée, reconnexion nécessaire."
            }
        }
        
        return strategies.get(error_type, strategies[ErrorType.API_ERROR])

class ErrorHandler:
    """Gestionnaire centralisé des erreurs pour l'agent"""
    
    def __init__(self, state_machine: StateMachine):
        self.state_machine = state_machine
        self.error_history: List[ErrorRecord] = []
        self.recovery_strategy = ErrorRecoveryStrategy()
    
    def handle_error(
        self,
        error: Exception,
        error_type: ErrorType,
        context: AgentContext
    ) -> dict:
        """Gère une erreur selon le type et le contexte"""
        
        # Enregistre l'erreur
        error_record = ErrorRecord(
            timestamp=datetime.now(),
            error_type=error_type,
            error_message=str(error),
            state_at_error=context.current_state,
            context_snapshot=context.user_data.copy(),
            retry_attempt=context.retry_count
        )
        self.error_history.append(error_record)
        
        logger.error(f"Erreur détectée : {error_type.value} - {error}")
        
        # Récupère la stratégie de récupération
        strategy = self.recovery_strategy.get_strategy(error_type)
        
        # Bascule en état d'erreur
        context.current_state = AgentState.ERROR
        
        # Vérifie si on peut réessayer
        if strategy["retry"] and context.retry_count < strategy["max_retries"]:
            return self._retry_with_backoff(context, strategy, error_record)
        else:
            return self._handle_irrecoverable(context, strategy, error_record)
    
    def _retry_with_backoff(
        self,
        context: AgentContext,
        strategy: dict,
        error_record: ErrorRecord
    ) -> dict:
        """Effectue une nouvelle tentative avec backoff exponentiel"""
        
        import time
        import random
        
        # Calcule le délai avec backoff
        delay = strategy["backoff_factor"] * (2 ** context.retry_count)
        delay += random.uniform(0, 1)  # Ajoute du jitter
        delay = min(delay, 60)  # Maximum 60 secondes
        
        logger.info(f"Tentative {context.retry_count + 1}, attente {delay:.2f}s")
        time.sleep(delay)
        
        context.retry_count += 1
        
        return {
            "action": "retry",
            "delay_seconds": delay,
            "retry_attempt": context.retry_count,
            "fallback_model": strategy.get("fallback_model"),
            "user_message": strategy["user_message"]
        }
    
    def _handle_irrecoverable(
        self,
        context: AgentContext,
        strategy: dict,
        error_record: ErrorRecord
    ) -> dict:
        """Gère les erreurs non récupérables"""
        
        action = strategy.get("action", "handoff")
        
        if action == "summarize_and_continue":
            # Réduit le contexte et continue
            return self._summarize_context(context)
        elif action == "request_correction":
            # Demande une correction à l'utilisateur
            return {
                "action": "request_correction",
                "user_message": strategy["user_message"]
            }
        elif action == "request_auth_refresh":
            # Demande une reconnexion
            return {
                "action": "request_auth_refresh",
                "user_message": strategy["user_message"]
            }
        else:
            # Transfert vers un humain
            return self._initiate_handoff(context, error_record)
    
    def _summarize_context(self, context: AgentContext) -> dict:
        """Résumé le contexte pour éviter le dépassement"""
        
        # Construit le résumé
        summary_prompt = """Résume cette conversation en conservant les informations essentielles :
        
Contexte actuel :
- État : {state}
- Données collectées : {data}

Conversation récente :
{history}

Fournis un résumé structuré avec :
1. Informations clés de l'utilisateur
2. Progression de la conversation
3. Objectif actuel

Sois concis mais complet.""".format(
            state=context.current_state.value,
            data=context.user_data,
            history="\n".join([
                f"{m['role']}: {m['content'][:100]}..." 
                for m in context.conversation_history[-6:]
            ])
        )
        
        # Appel API pour générer le résumé
        response = self.state_machine.ai_client.chat_completion(
            messages=[{"role": "user", "content": summary_prompt}],
            model="gpt-4.1",
            max_tokens=500
        )
        
        if response["success"]:
            summary = response["content"]
            # Remplace l'historique par le résumé
            context.conversation_history = [
                {"role": "system", "content": f"RESUME_CONVERSATION: {summary}"}
            ]
            
            return {
                "action": "continue_with_summary",
                "summary": summary,
                "user_message": "J'ai résumé notre conversation pour continuer efficacement."
            }
        else:
            return self._initiate_handoff(context, None)
    
    def _initiate_handoff(
        self,
        context: AgentContext,
        error_record: Optional[ErrorRecord]
    ) -> dict:
        """Initie le transfert vers un agent humain"""
        
        context.current_state = AgentState.HANDOFF
        
        # Génère le rapport de transfert
        handoff_report = f"""
=== RAPPORT DE TRANSFERT ===
Session : {context.session_id}
Heure : {datetime.now().isoformat()}
État au transfert : {error_record.state_at_error.value if error_record else 'N/A'}
Dernière erreur : {error_record.error_message if error_record else 'N/A'}

Données collectées :
{json.dumps(context.user_data, indent=2, ensure_ascii=False)}

Conversation récente :
{chr(10).join([
    f"{m['role']}: {m['content']}" 
    for m in context.conversation_history[-10:]
])}
============================
"""
        
        logger.warning(f"Transfert vers humain:\n{handoff_report}")
        
        return {
            "action": "handoff",
            "handoff_report": handoff_report,
            "user_message": "Je vous connecte avec un de nos conseillers pour résoudre ce problème."
        }
    
    def get_error_statistics(self) -> dict:
        """Retourne des statistiques sur les erreurs"""
        if not self.error_history:
            return {"total_errors": 0}
        
        stats = {
            "total_errors": len(self.error_history),
            "by_type": {},
            "recovery_rate": 0,
            "average_retries": 0
        }
        
        recovered = 0
        total_retries = 0
        
        for error in self.error_history:
            type_name = error.error_type.value
            stats["by_type"][type_name] = stats["by_type"].get(type_name, 0) + 1
            
            if error.resolution:
                recovered += 1
            total_retries += error.retry_attempt
        
        stats["recovery_rate"] = (recovered / len(self.error_history)) * 100
        stats["average_retries"] = total_retries / len(self.error_history)
        
        return stats

class CircuitBreaker:
    """Pattern Circuit Breaker pour éviter les cascading failures"""
    
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        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 = "CLOSED"  # CLOSED, OPEN, HALF_OPEN
    
    def call(self, func, *args, **kwargs):
        """Exécute la fonction avec protection circuit breaker"""
        
        if self.state == "OPEN":
            if self._should_attempt_reset():
                self.state = "HALF_OPEN"
            else:
                raise Exception("Circuit Breaker OPEN - requête bloquée")
        
        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:
        """Vérifie si on doit tenter une réinitialisation"""
        if not self.last_failure_time:
            return False
        elapsed = (datetime.now() - self.last_failure_time).total_seconds()
        return elapsed >= self.recovery_timeout
    
    def _on_success(self):
        """Gère le succès"""
        self.failure_count = 0
        self.state = "CLOSED"
    
    def _on_failure(self):
        """Gère l'échec"""
        self.failure_count += 1
        self.last_failure_time = datetime.now()
        
        if self.failure_count >= self.failure_threshold:
            self.state = "OPEN"
            logger.warning(f"Circuit Breaker OPEN après {self.failure_count} échecs")

Intégration complète avec monitoring
class AgentWithErrorHandling:
    """Agent IA complet avec gestion d'erreurs et monitoring"""
    
    def __init__(self, api_key: str):
        self.client = HolySheepAIClient(api_key)
        self.state_machine = StateMachine(self.client)
        self.error_handler = ErrorHandler(self.state_machine)
        self.circuit_breaker = CircuitBreaker(
            failure_threshold=5,
            recovery_timeout=60
        )
        
        # Configuration du monitoring
        self.metrics = {
            "total_requests": 0,
            "successful_requests": 0,
            "failed_requests": 0,
            "average_latency": 0,
            "total_cost": 0
        }
    
    def process_message(self, session_id: str, user_message: str) -> dict:
        """Traite un message avec gestion complète des erreurs"""
        
        self.metrics["total_requests"] += 1
        
        # Récupère ou crée le contexte
        context = self._get_or_create_context(session_id)
        
        try:
            # Utilise le circuit breaker
            result = self.circuit_breaker.call(
                self.state_machine.process_user_input,
                context,
                user_message
            )
            
            if result["success"]:
                self.metrics["successful_requests"] += 1
                # Met à jour les métriques de latence
                self._update_latency_metrics(result.get("latency_ms", 0))
                # Met à jour les coûts (estimation)
                self._update_cost_metrics(result.get("usage", {}))
            
            return result
            
        except Exception as e:
            self.metrics["failed_requests"] += 1
            
            # Détermine le type d'erreur
            error_type = self._classify_error(e)
            
            # Gère l'erreur
            error_result = self.error_handler.handle_error(e, error_type, context)
            
            return {
                "success": False,
                "error_type": error_type.value,
                "error_message": str(e),
                "recovered": error_result["action"] != "handoff",
                **error_result
            }
    
    def _classify_error(self, error: Exception) -> ErrorType:
        """Classification automatique du type d'erreur"""
        error_str = str(error).lower()
        
        if "401" in error_str or "403" in error_str or "unauthorized" in error_str:
            return ErrorType.AUTH_ERROR
        elif "429" in error_str or "rate limit" in error_str:
            return ErrorType.RATE_LIMIT
        elif "timeout" in error_str or "timed out" in error_str:
            return ErrorType.TIMEOUT_ERROR
        elif "network" in error_str or "connection" in error_str:
            return ErrorType.NETWORK_ERROR
        elif "invalid" in error_str or "malformed" in error_str:
            return ErrorType.INVALID_RESPONSE
        else:
            return ErrorType.API_ERROR
    
    def _get_or_create_context(self, session_id: str) -> AgentContext:
        """Récupère ou crée un contexte de session"""
        # Implémentation simplifiée - en production, utiliser Redis ou DB
        if not hasattr(self, '_contexts'):
            self._contexts = {}
        
        if session_id not in self._contexts:
            self._contexts[session_id] = AgentContext(
                session_id=session_id,
                current_state=AgentState.IDLE
            )
        
        return self._contexts[session_id]
    
    def _update_latency_metrics(self, latency_ms: float):
        """Met à jour les métriques de latence"""
        current_avg = self.metrics["average_latency"]
        total = self.metrics["total_requests"]
        self.metrics["average_latency"] = (
            (current_avg * (total - 1) + latency_ms) / total
        )
    
    def _update_cost_metrics(self, usage: dict):
        """Met à jour les métriques de coût"""
        # Estimation basée sur les tokens utilisés
        prompt_tokens = usage.get("prompt_tokens", 0)
        completion_tokens = usage.get("completion_tokens", 0)
        total_tokens = prompt_tokens + completion_tokens
        
        # Coût au million de
Ressources connexes
📚 Tutoriels API IA
💰 Voir les tarifs
📖 Documentation
🚀 Inscription gratuite
Articles connexes
GPT-5.5 Streaming API : Implémenter l'Effet Typewriter en Te
Vector Database et API IA : Guide Complet pour Débutants en 
Dify 工作流 API 调用错误处理最佳实践