File size: 4,426 Bytes
56dc677 4a13628 56dc677 691fd14 95cb26e 56dc677 691fd14 56dc677 691fd14 56dc677 4a13628 56dc677 691fd14 56dc677 4a13628 56dc677 4a13628 56dc677 95cb26e 56dc677 691fd14 56dc677 95cb26e 56dc677 95cb26e 56dc677 4a13628 56dc677 691fd14 56dc677 691fd14 56dc677 |
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 |
import base64
import uuid
from typing import Optional, Dict, Any
from datetime import datetime
from models.audio import ChatbotResponse, UserMessage
class ChatbotService:
def __init__(self):
self.sessions: Dict[str, Dict[str, Any]] = {}
self.initialized = False
self._initialize_services()
def _initialize_services(self):
"""Initialize services"""
try:
from services.tts_service import SimpleTTSService
self.tts_service = SimpleTTSService()
print("✓ TTS service initialized")
except ImportError as e:
print(f"⚠️ TTS service not available: {e}")
self.tts_service = None
try:
from services.stt_service import STTService
self.stt_service = STTService()
print("✓ STT service initialized")
except ImportError as e:
print(f"⚠️ STT service not available: {e}")
self.stt_service = None
async def initialize(self):
"""Initialize the chatbot service"""
if self.stt_service:
await self.stt_service.initialize()
self.initialized = True
print("✓ Chatbot Service fully initialized")
async def process_user_message(self, user_message: UserMessage) -> ChatbotResponse:
# Update session
session = self._get_or_create_session(user_message.session_id)
# Process message based on type
if user_message.message_type == "audio" and self.stt_service:
text_input = await self.stt_service.transcribe_audio_base64(user_message.content)
elif user_message.message_type == "audio":
text_input = "[Voice message received]"
else:
text_input = user_message.content
# Add to conversation history
session["conversation_history"].append({
"role": "user",
"content": text_input,
"timestamp": user_message.timestamp
})
# Generate chatbot response
chatbot_text = await self._generate_chatbot_response(text_input, session)
# TTS: Convert response to audio
audio_base64 = None
if self.tts_service:
audio_base64 = await self.tts_service.text_to_speech_base64(chatbot_text)
# Create response
response = ChatbotResponse(
response_id=str(uuid.uuid4()),
text_response=chatbot_text,
audio_response=audio_base64,
session_id=user_message.session_id
)
# Add response to history
session["conversation_history"].append({
"role": "assistant",
"content": chatbot_text,
"audio_response": audio_base64,
"timestamp": response.timestamp
})
return response
async def _generate_chatbot_response(self, user_input: str, session: Dict[str, Any]) -> str:
"""Chatbot response generation logic"""
user_input_lower = user_input.lower()
if any(greet in user_input_lower for greet in ["hello", "hi", "hey"]):
return "Hello! I'm your voice assistant. How can I help you today?"
if "time" in user_input_lower:
return f"The current time is {datetime.now().strftime('%H:%M')}"
if "help" in user_input_lower:
return "I can process both text and voice messages. Try sending me a voice note!"
if "name" in user_input_lower:
return "I'm your AI voice assistant. I'm here to help with your questions!"
if "voice" in user_input_lower or "audio" in user_input_lower:
return "Yes! I support voice messages. You can speak to me and I'll respond with audio too!"
return f"I understand you said: '{user_input}'. How can I assist you further?"
def _get_or_create_session(self, session_id: str) -> Dict[str, Any]:
if session_id not in self.sessions:
self.sessions[session_id] = {
"conversation_history": [],
"created_at": datetime.now(),
"last_activity": datetime.now()
}
return self.sessions[session_id]
def get_session_history(self, session_id: str) -> Optional[Dict[str, Any]]:
return self.sessions.get(session_id) |