import uuid from typing import Dict, List, Optional from datetime import datetime import requests import json from config import GROQ_API_KEY, GROQ_CHAT_MODEL # Stockage en mémoire des conversations conversation_store: Dict[str, List[Dict]] = {} def generate_chat_response( user_input: str, conversation_id: Optional[str] = None, system_prompt: Optional[str] = None ) -> str: """ Génère une réponse de chatbot pour une entrée utilisateur """ try: # 1. Vérifier que l'input n'est pas vide if not user_input or not isinstance(user_input, str): raise ValueError("L'entrée utilisateur doit être une chaîne de caractères non vide") user_input = user_input.strip() if len(user_input) == 0: return "Je n'ai pas entendu ce que vous avez dit. Pouvez-vous répéter ?" # 2. Gérer la conversation if not conversation_id: conversation_id = str(uuid.uuid4()) # Initialiser la conversation si elle n'existe pas if conversation_id not in conversation_store: conversation_store[conversation_id] = [] # 3. Ajouter le message utilisateur à l'historique conversation_store[conversation_id].append({ "role": "user", "content": user_input, "timestamp": datetime.now().isoformat() }) # 4. Préparer le prompt système if not system_prompt: system_prompt = """Tu es un assistant vocal amical et utile. Tes réponses doivent être naturelles à l'oral, concises (max 2-3 phrases) et adaptées à une synthèse vocale. Sois courtois et serviable.""" # 5. Préparer les messages pour l'API Groq messages = [{"role": "system", "content": system_prompt}] # Ajouter l'historique (limité aux derniers messages) history = conversation_store[conversation_id][-6:] # Derniers 6 messages for msg in history: messages.append({"role": msg["role"], "content": msg["content"]}) # 6. Appeler l'API Groq Chat if not GROQ_API_KEY: # Fallback si pas de clé API response_text = f"Bonjour ! Vous avez dit : '{user_input}'. Je suis configuré pour répondre, mais l'API Groq n'est pas configurée." else: try: response_text = call_groq_chat_api(messages) except Exception as api_error: # Fallback en cas d'erreur API print(f"Erreur API Groq: {api_error}") response_text = f"D'accord, j'ai compris : {user_input}. Je suis un chatbot et je vous réponds." # 7. Ajouter la réponse à l'historique conversation_store[conversation_id].append({ "role": "assistant", "content": response_text, "timestamp": datetime.now().isoformat() }) # Limiter la taille de l'historique if len(conversation_store[conversation_id]) > 20: conversation_store[conversation_id] = conversation_store[conversation_id][-10:] return response_text except Exception as e: print(f"Erreur dans generate_chat_response: {e}") return f"Désolé, une erreur est survenue : {str(e)}" def call_groq_chat_api(messages: List[Dict]) -> str: """ Appelle l'API Groq Chat """ if not GROQ_API_KEY: raise RuntimeError("GROQ_API_KEY non configurée") url = "https://api.groq.com/openai/v1/chat/completions" headers = { "Authorization": f"Bearer {GROQ_API_KEY}", "Content-Type": "application/json" } payload = { "model": GROQ_CHAT_MODEL, "messages": messages, "temperature": 0.7, "max_tokens": 300, # Augmenté pour des réponses plus complètes "top_p": 0.9, "stream": False } try: response = requests.post(url, headers=headers, json=payload, timeout=30) response.raise_for_status() result = response.json() if "choices" not in result or len(result["choices"]) == 0: raise ValueError("Réponse invalide de l'API Groq") return result["choices"][0]["message"]["content"] except requests.exceptions.RequestException as e: raise Exception(f"Erreur de connexion à l'API Groq: {str(e)}") except KeyError as e: raise Exception(f"Format de réponse invalide: {str(e)}") def get_conversation_history(conversation_id: str) -> List[Dict]: """ Récupère l'historique d'une conversation """ return conversation_store.get(conversation_id, []) def clear_conversation(conversation_id: str): """ Efface une conversation """ if conversation_id in conversation_store: del conversation_store[conversation_id] return True return False