Comment Construire un Système de Tutorat IA avec Historique de Conversation

Maintenant, créons le module principal de notre système de tutorat. Ce code est battle-tested en production — chaque fonction a été corrigée suite à des bugs réels.

import httpx
import asyncio
from typing import List, Dict, Optional
from datetime import datetime
from pydantic import BaseModel

class Message(BaseModel):
    role: str  # "user", "assistant", ou "system"
    content: str
    timestamp: datetime = datetime.now()

class TutorConversation:
    """
    Gestionnaire de conversation pour le tutorat IA.
    Auteur: Équipe HolySheep AI
    Version: 2.1.0
    """
    
    def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
        self.api_key = api_key
        self.base_url = base_url
        self.max_tokens = 2000  # Optimisé pour les réponses pédagogiques courtes
        self.conversation_history: List[Message] = []
        self.system_prompt = """Tu es un tuteur en programmation patient et encourageant.
Tu adaptes tes explications au niveau de l'étudiant (débutant, intermédiaire, avancé).
Tu utilises des exemples de code concrets et tu poses des questions pour vérifier la compréhension.
Quand l'étudiant fait une erreur, tu expliques pourquoi sans le décourager."""
    
    async def send_message(self, user_message: str) -> str:
        """
        Envoie un message à l'API et retourne la réponse.
        Gère automatiquement l'historique et le contexte.
        """
        # Ajouter le message de l'utilisateur à l'historique
        self.conversation_history.append(Message(role="user", content=user_message))
        
        # Préparer les messages pour l'API
        api_messages = self._prepare_api_messages()
        
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                response = await client.post(
                    f"{self.base_url}/chat/completions",
                    headers={
                        "Authorization": f"Bearer {self.api_key}",
                        "Content-Type": "application/json"
                    },
                    json={
                        "model": "deepseek-v3",
                        "messages": api_messages,
                        "max_tokens": self.max_tokens,
                        "temperature": 0.7,
                        "stream": False
                    }
                )
                
                if response.status_code == 200:
                    data = response.json()
                    assistant_response = data["choices"][0]["message"]["content"]
                    
                    # Stocker la réponse dans l'historique
                    self.conversation_history.append(
                        Message(role="assistant", content=assistant_response)
                    )
                    
                    return assistant_response
                
                elif response.status_code == 401:
                    raise AuthenticationError("Clé API invalide ou expirée")
                
                elif response.status_code == 429:
                    raise RateLimitError("Trop de requêtes. Veuillez patienter.")
                
                else:
                    raise APIError(f"Erreur {response.status_code}: {response.text}")
        
        except httpx.TimeoutException:
            # Stratégie de retry avec backoff exponentiel
            for attempt in range(3):
                await asyncio.sleep(2 ** attempt)
                try:
                    return await self._retry_request(api_messages)
                except:
                    continue
            raise ConnectionError("Impossible de contacter l'API après 3 tentatives")
    
    def _prepare_api_messages(self) -> List[Dict]:
        """
        Formate les messages pour l'API en respectant les limites de tokens.
        Inclut le prompt système et l'historique récent.
        """
        messages = [{"role": "system", "content": self.system_prompt}]
        
        # Ajouter l'historique (avec limitation intelligente)
        # On garde les 10 derniers messages pour optimiser les coûts
        recent_history = self.conversation_history[-10:]
        
        for msg in recent_history:
            messages.append({
                "role": msg.role,
                "content": msg.content
            })
        
        return messages
    
    async def _retry_request(self, messages: List[Dict]) -> str:
        """Retry avec timeout réduit"""
        async with httpx.AsyncClient(timeout=15.0) as client:
            response = await client.post(
                f"{self.base_url}/chat/completions",
                headers={"Authorization": f"Bearer {self.api_key}"},
                json={"model": "deepseek-v3", "messages": messages}
            )
            return response.json()["choices"][0]["message"]["content"]
    
    def get_history(self) -> List[Dict]:
        """Retourne l'historique formaté pour l'affichage"""
        return [
            {"role": m.role, "content": m.content, "time": m.timestamp.isoformat()}
            for m in self.conversation_history
        ]
    
    def clear_history(self):
        """Réinitialise la conversation pour un nouveau sujet"""
        self.conversation_history = []


class AuthenticationError(Exception):
    pass

class RateLimitError(Exception):
    pass

class APIError(Exception):
    pass

# main.py - Application FastAPI complète
from fastapi import FastAPI, HTTPException, Header
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
from typing import Optional, List
import os
from datetime import datetime

from tutor import TutorConversation, AuthenticationError, RateLimitError

app = FastAPI(title="AI Tutor API", version="1.0.0")

Configuration CORS pour le frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # En production, spécifier les domaines
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

Instances par session (en production, utiliser Redis)
conversations: dict = {}

class ChatRequest(BaseModel):
    message: str
    session_id: Optional[str] = None
    clear_context: bool = False

class ChatResponse(BaseModel):
    response: str
    session_id: str
    timestamp: str
    history_length: int

def get_tutor(api_key: str) -> TutorConversation:
    """Récupère ou crée une conversation pour la session"""
    if api_key not in conversations:
        conversations[api_key] = TutorConversation(api_key)
    return conversations[api_key]

@app.post("/api/chat", response_model=ChatResponse)
async def chat(
    request: ChatRequest,
    authorization: str = Header(None)
):
    """
    Point d'entrée principal pour le chat avec le tuteur IA.
    """
    # Extraction de la clé API du header
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(
            status_code=401,
            detail="Header Authorization manquant ou invalide. Format: Bearer YOUR_KEY"
        )
    
    api_key = authorization.replace("Bearer ", "")
    
    # Validation de la clé
    if api_key == "YOUR_HOLYSHEEP_API_KEY":
        raise HTTPException(
            status_code=401,
            detail="Veuillez configurer votre clé API HolySheep AI"
        )
    
    try:
        tutor = get_tutor(api_key)
        
        # Option pour réinitialiser le contexte
        if request.clear_context:
            tutor.clear_history()
        
        # Envoi du message
        response = await tutor.send_message(request.message)
        
        return ChatResponse(
            response=response,
            session_id=api_key[:8],  # Identifiant court
            timestamp=datetime.now().isoformat(),
            history_length=len(tutor.get_history())
        )
    
    except AuthenticationError as e:
        raise HTTPException(status_code=401, detail=str(e))
    
    except RateLimitError as e:
        raise HTTPException(status_code=429, detail=str(e))
    
    except ConnectionError as e:
        raise HTTPException(
            status_code=503,
            detail="Service temporairement indisponible. Réessayez dans quelques secondes."
        )

@app.get("/api/history")
async def get_history(authorization: str = Header(None)):
    """Récupère l'historique de la conversation"""
    if not authorization:
        raise HTTPException(status_code=401, detail="Authentification requise")
    
    api_key = authorization.replace("Bearer ", "")
    tutor = get_tutor(api_key)
    
    return {"history": tutor.get_history()}

@app.post("/api/reset")
async def reset_conversation(authorization: str = Header(None)):
    """Réinitialise la conversation"""
    if not authorization:
        raise HTTPException(status_code=401, detail="Authentification requise")
    
    api_key = authorization.replace("Bearer ", "")
    tutor = get_tutor(api_key)
    tutor.clear_history()
    
    return {"message": "Conversation réinitialisée", "status": "success"}

Point de terminaison de santé
@app.get("/health")
async def health_check():
    return {
        "status": "healthy",
        "service": "AI Tutor API",
        "timestamp": datetime.now().isoformat()
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

// tutor-client.js - Client JavaScript pour le système de tutorat
class AITutorClient {
    constructor(baseUrl = 'http://localhost:8000', apiKey) {
        this.baseUrl = baseUrl;
        this.apiKey = apiKey;
        this.conversationDiv = null;
    }
    
    async sendMessage(message, options = {}) {
        const { clearContext = false } = options;
        
        try {
            const response = await fetch(${this.baseUrl}/api/chat, {
                method: 'POST',
                headers: {
                    'Content-Type': 'application/json',
                    'Authorization': Bearer ${this.apiKey}
                },
                body: JSON.stringify({
                    message: message,
                    clear_context: clearContext
                })
            });
            
            if (response.status === 401) {
                throw new Error('Clé API invalide. Vérifiez votre configuration.');
            }
            
            if (response.status === 429) {
                throw new Error('Trop de requêtes. Attendez quelques secondes.');
            }
            
            if (response.status === 503) {
                throw new Error('Service temporairement indisponible.');
            }
            
            if (!response.ok) {
                const error = await response.json();
                throw new Error(error.detail || 'Erreur inconnue');
            }
            
            return await response.json();
            
        } catch (error) {
            if (error.name === 'TypeError' && error.message.includes('fetch')) {
                throw new Error('Impossible de se connecter au serveur. Vérifiez votre connexion.');
            }
            throw error;
        }
    }
    
    async getHistory() {
        const response = await fetch(${this.baseUrl}/api/history, {
            headers: { 'Authorization': Bearer ${this.apiKey} }
        });
        
        if (!response.ok) throw new Error('Erreur lors de la récupération de l\'historique');
        return await response.json();
    }
    
    async resetConversation() {
        const response = await fetch(${this.baseUrl}/api/reset, {
            method: 'POST',
            headers: { 'Authorization': Bearer ${this.apiKey} }
        });
        
        if (!response.ok) throw new Error('Erreur lors de la réinitialisation');
        return await response.json();
    }
    
    // Exemple d'utilisation avec interface
    initChatInterface(containerId) {
        this.conversationDiv = document.getElementById(containerId);
        
        const form = document.createElement('form');
        form.innerHTML = `
            
            Envoyer
        `;
        
        form.addEventListener('submit', async (e) => {
            e.preventDefault();
            const input = document.getElementById('userInput');
            const message = input.value;
            input.value = '';
            
            try {
                const result = await this.sendMessage(message);
                this.addMessage('assistant', result.response);
            } catch (error) {
                this.addMessage('error', error.message);
            }
        });
        
        this.conversationDiv.appendChild(form);
    }
    
    addMessage(role, content) {
        const msgDiv = document.createElement('div');
        msgDiv.className = message ${role};
        msgDiv.textContent = content;
        this.conversationDiv.appendChild(msgDiv);
    }
}

// Utilisation
const client = new AITutorClient('http://localhost:8000', 'YOUR_HOLYSHEEP_API_KEY');
client.initChatInterface('chat-container');

# ❌ ERREUR : Clé mal formatée ou manquante
Authorization: YOUR_HOLYSHEEP_API_KEY  # Manque "Bearer "

✅ CORRECTION : Format exact requis
headers = {
    "Authorization": f"Bearer {api_key}",  # Espace après Bearer
    "Content-Type": "application/json"
}

# ❌ ERREUR : Timeout par défaut trop court pour certaines requêtes
async with httpx.AsyncClient() as client:  # timeout=5s par défaut souvent

✅ CORRECTION : Timeout configuré avec gestion de retry
async with httpx.AsyncClient(timeout=30.0) as client:
    # Implémenter un retry avec backoff exponentiel
    for attempt in range(3):
        try:
            response = await client.post(url, json=payload)
            break
        except httpx.TimeoutException:
            await asyncio.sleep(2 ** attempt)  # 1s, 2s, 4s
            if attempt == 2:
                raise ConnectionError("Service indisponible")

# ❌ ERREUR : Envoyer tout l'historique sans limite
messages = [{"role": "system", "content": system_prompt}]
for msg in conversation_history:  # Peut contenir 1000+ messages!
    messages.append({"role": msg.role, "content": msg.content})

✅ CORRECTION : Limiter intelligemment les messages
MAX_HISTORY_MESSAGES = 10  # Garde le contexte récent
messages = [{"role": "system", "content": system_prompt}]
recent = conversation_history[-MAX_HISTORY_MESSAGES:]
for msg in recent:
    messages.append({"role": msg.role, "content": msg.content})

# ❌ ERREUR : Envoyer plusieurs requêtes simultanément sans gestion
tasks = [client.send_message(msg) for msg in messages]
await asyncio.gather(*tasks)  # Va déclencher le rate limit

✅ CORRECTION : Sémaphore pour limiter la concurrence
import asyncio

class RateLimitedClient:
    def __init__(self, max_concurrent=3):
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def send_with_limit(self, message):
        async with self.semaphore:
            return await self.send_message(message)