|
|
import time |
|
|
import uuid |
|
|
from dataclasses import dataclass |
|
|
from typing import Dict, Any |
|
|
|
|
|
@dataclass |
|
|
class Turn: |
|
|
turn_id: str |
|
|
session_id: str |
|
|
user_id: str |
|
|
timestamp: float |
|
|
raw_input: Dict[str, Any] |
|
|
normalized_input: str = "" |
|
|
postprocessed_output: Dict[str, Any] = None |
|
|
|
|
|
def __post_init__(self): |
|
|
if self.turn_id is None: |
|
|
self.turn_id = str(uuid.uuid4()) |
|
|
if self.timestamp is None: |
|
|
self.timestamp = time.time() |
|
|
if self.postprocessed_output is None: |
|
|
self.postprocessed_output = {} |
|
|
|
|
|
class DialogueManager: |
|
|
def __init__(self, config, persistence, auth, voiceprint, asr): |
|
|
self.config = config |
|
|
self.persistence = persistence |
|
|
self.auth = auth |
|
|
self.voiceprint = voiceprint |
|
|
self.asr = asr |
|
|
|
|
|
def submit_turn(self, session_id, user_id, raw_input): |
|
|
turn = Turn( |
|
|
turn_id=str(uuid.uuid4()), |
|
|
session_id=session_id, |
|
|
user_id=user_id, |
|
|
timestamp=time.time(), |
|
|
raw_input=raw_input, |
|
|
normalized_input=raw_input.get("text", ""), |
|
|
postprocessed_output={"text": self.generate_response(raw_input.get("text", ""))} |
|
|
) |
|
|
|
|
|
|
|
|
self.persistence.save_turn(turn) |
|
|
|
|
|
return turn |
|
|
|
|
|
def generate_response(self, text): |
|
|
|
|
|
if not text.strip(): |
|
|
return "I'm here to help. What would you like to know?" |
|
|
|
|
|
|
|
|
if "hello" in text.lower(): |
|
|
return "Hello! How can I assist you today?" |
|
|
elif "help" in text.lower(): |
|
|
return "I'm Hive, your AI tutor. I can help with learning, answer questions, and remember our conversations." |
|
|
elif "thank" in text.lower(): |
|
|
return "You're welcome! Is there anything else I can help with?" |
|
|
else: |
|
|
return f"I understand you said: '{text}'. Let me think about how to best help you with that." |