File size: 4,648 Bytes
1d46e48 c059296 1d46e48 a71355d 1d46e48 a71355d 1d46e48 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 a71355d c059296 1d46e48 a71355d 1d46e48 a71355d 1d46e48 a71355d 1d46e48 a71355d 1d46e48 c059296 1d46e48 c059296 a71355d c059296 a71355d c059296 a71355d 1d46e48 a71355d 1d46e48 a71355d 1d46e48 a71355d 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 |
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 |