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 # In-memory conversation storage conversation_store: Dict[str, List[Dict]] = {} def generate_chat_response( user_input: str, conversation_id: Optional[str] = None, system_prompt: Optional[str] = None ) -> str: """ Generate chatbot response for user input (English only) """ try: # 1. Validate input if not user_input or not isinstance(user_input, str): raise ValueError("User input must be a non-empty string") user_input = user_input.strip() if len(user_input) == 0: return "I didn't hear what you said. Can you repeat?" # 2. Handle conversation if not conversation_id: conversation_id = str(uuid.uuid4()) # Initialize conversation if it doesn't exist if conversation_id not in conversation_store: conversation_store[conversation_id] = [] # 3. Add user message to history conversation_store[conversation_id].append({ "role": "user", "content": user_input, "timestamp": datetime.now().isoformat() }) # 4. Prepare system prompt (English only) if not system_prompt: system_prompt = """You are a friendly and helpful English voice assistant. Respond in English only. Keep responses concise (2-3 sentences max), natural for speech, and helpful. Be polite and engaging.""" # 5. Prepare messages for Groq API messages = [{"role": "system", "content": system_prompt}] # Add conversation history (last 6 messages) history = conversation_store[conversation_id][-6:] for msg in history: messages.append({"role": msg["role"], "content": msg["content"]}) # 6. Call Groq Chat API if not GROQ_API_KEY: # Fallback if no API key response_text = f"Hello! You said: '{user_input}'. I'm a voice assistant configured to respond in English." else: try: response_text = call_groq_chat_api(messages) except Exception as api_error: print(f"Groq API error: {api_error}") response_text = f"I understand you said: {user_input}. How can I help you today?" # 7. Add response to history conversation_store[conversation_id].append({ "role": "assistant", "content": response_text, "timestamp": datetime.now().isoformat() }) # Limit history size 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"Error in generate_chat_response: {e}") return "Sorry, an error occurred. Can you please repeat?" def call_groq_chat_api(messages: List[Dict]) -> str: """ Call Groq Chat API """ if not GROQ_API_KEY: raise RuntimeError("GROQ_API_KEY is not configured") 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, "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("Invalid response from Groq API") return result["choices"][0]["message"]["content"] except requests.exceptions.RequestException as e: raise Exception(f"Groq API connection error: {str(e)}") except KeyError as e: raise Exception(f"Invalid response format: {str(e)}") def get_conversation_history(conversation_id: str) -> List[Dict]: """ Get conversation history """ return conversation_store.get(conversation_id, []) def clear_conversation(conversation_id: str) -> bool: """ Clear a conversation """ if conversation_id in conversation_store: del conversation_store[conversation_id] return True return False