File size: 5,017 Bytes
1d46e48 c059296 1d46e48 c059296 1d46e48 c059296 1d46e48 c059296 1d46e48 c059296 1d46e48 c059296 1d46e48 c059296 1d46e48 c059296 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | 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 |