"""
voice_agent_standalone.py
──────────────────────────
AARA multilingual voice agent for Sahara Star Hotels.
CPU-only · HuggingFace Spaces
Key fixes vs v1:
• Whisper model defaults to faster-whisper-medium on CPU for better ASR quality
• LLM uses GGUF + llama-cpp-python (Q4_K_M ~330 MB, ~20 tok/s CPU)
with transformers as automatic fallback
• ResponseCache is SQLite-backed (atomic writes, concurrent-safe)
• TTSEngine can use lightweight Edge TTS with browser fallback
• enable_multi_pass_asr defaults to False
• llm_n_ctx reduced to 512 (enough for hotel turns)
• Silero VAD pre-screen before Whisper (saves a full inference on silence)
"""
from __future__ import annotations
import asyncio
import gc
import hashlib
import io
import json
import logging
import os
import re
import sqlite3
import subprocess
import tempfile
import threading
import time
import uuid
import wave
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Optional, cast
import numpy as np
from scipy import signal
# ─────────────────────────────────────────────
# Logging
# ─────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="[%(asctime)s] %(levelname)s %(name)s: %(message)s",
datefmt="%H:%M:%S",
)
logger = logging.getLogger("AARA")
# ─────────────────────────────────────────────
# Model defaults
# ─────────────────────────────────────────────
DEFAULT_WHISPER_MODEL = os.environ.get("AARA_WHISPER_MODEL", "Systran/faster-whisper-medium")
DEFAULT_QWEN_GGUF_REPO = "Qwen/Qwen2.5-0.5B-Instruct-GGUF"
DEFAULT_QWEN_GGUF_FILE = "qwen2.5-0.5b-instruct-q4_k_m.gguf"
DEFAULT_QWEN_CHAT_MODEL = "Qwen/Qwen2.5-0.5B-Instruct" # transformers fallback only
def _env_flag(name: str, default: bool) -> bool:
raw = os.environ.get(name)
if raw is None:
return default
return raw.strip().lower() not in {"0", "false", "no", "off"}
# ═══════════════════════════════════════════════════════════════════
# Configuration
# ═══════════════════════════════════════════════════════════════════
@dataclass
class AaraConfig:
"""Central configuration for all AARA components."""
# Paths
models_dir: Path = field(default_factory=lambda: Path(os.environ.get("AARA_MODELS_DIR", "models")))
db_path: str = "sahara_star.db"
reference_audio_dir: Path = field(default_factory=lambda: Path("reference_audio"))
output_audio_dir: Path = field(default_factory=lambda: Path("output_audio"))
log_dir: Path = field(default_factory=lambda: Path("logs"))
cache_dir: Path = field(default_factory=lambda: Path("cache"))
# Audio
sample_rate: int = 16000
channels: int = 1
silence_threshold: float = 0.010
silence_stop_sec: float = 0.5
min_speech_rms: float = 0.003
max_audio_length: float = 12.0
# Human voice detection
pitch_min_hz: float = 50.0
pitch_max_hz: float = 400.0
formant_min_hz: float = 200.0
formant_max_hz: float = 4000.0
min_voice_energy_ratio: float = 0.15
periodicity_threshold: float = 0.20
# ASR — use whisper-medium by default on CPU for better recognition quality
whisper_model_dir: str = field(default_factory=lambda: os.environ.get("AARA_WHISPER_MODEL_DIR", ""))
asr_confidence_threshold: float = 0.46 # More tolerant on CPU so fast/slow/energetic speech still reaches the agent
asr_confidence_rerun_threshold: float = 0.68
asr_language_prob_threshold: float = 0.50 # NEW: reject if Whisper's lang-detection prob < 50%
asr_min_word_count: int = 1 # NEW: reject single-word transcriptions with low conf
english_only_mode: bool = True # ENFORCED: English-only agent
asr_force_language: str = "en" # ENFORCED: English only
asr_retry_without_language_lock: bool = False # English only, no fallback
asr_min_trimmed_speech_sec: float = 0.20
asr_min_speech_ratio: float = 0.02
asr_vad_threshold: float = 0.24
asr_vad_min_speech_ms: int = 64
asr_vad_min_silence_ms: int = 80
asr_vad_speech_pad_ms: int = 320
asr_hallucination_silence_threshold: float = 0.8
asr_vad_rms_fallback_threshold: float = 0.014
asr_vad_duration_fallback_sec: float = 0.45
enable_multi_pass_asr: bool = False # FIXED: was True (2× latency on CPU)
enable_silero_vad: bool = True # Pre-screen audio before Whisper
enable_environment_adapt: bool = False # sounddevice not available on HF Spaces
# LLM
gguf_model_path: str = "" # Auto-detected if empty
planner_model_name: str = DEFAULT_QWEN_CHAT_MODEL # transformers fallback
planner_model_path: str = ""
llm_model_path: str = ""
llm_n_ctx: int = 768 # Optimized for English hotel context only
llm_n_batch: int = 64
llm_n_threads: int = 4
llm_max_tokens: int = 96 # Keep CPU responses short so fallbacks return quickly
llm_temperature: float = 0.35
llm_repeat_penalty: float = 1.08
llm_verbose: bool = False
# TTS — browser speech + server TTS (English only)
enable_server_tts: bool = True # Use server TTS for consistency
tts_voice_female: str = "en-IN-NeerjaNeural" # English (Indian)
tts_voice_male: str = "en-IN-PrabhatNeural" # English (Indian)
tts_voice_pace: float = 1.0
# Session
max_history_turns: int = 4 # Small GGUF model: keep only a few recent turns in context
unload_models_between_turns: bool = False
def __post_init__(self) -> None:
for d in (self.output_audio_dir, self.log_dir, self.cache_dir):
d.mkdir(parents=True, exist_ok=True)
if self.english_only_mode:
self.asr_force_language = "en"
self.asr_retry_without_language_lock = False
# Auto-detect Whisper
if not self.whisper_model_dir:
preferred_dirname = os.environ.get("AARA_WHISPER_MODEL_DIRNAME", "").strip()
candidate_names = [
preferred_dirname,
"whisper-large-v3-turbo",
"whisper-distil-large-v3",
"whisper-large-v3",
"whisper-medium",
"whisper-small",
"whisper-base",
]
for candidate in (self.models_dir / name for name in candidate_names if name):
if candidate.exists() and (candidate / "config.json").exists():
self.whisper_model_dir = str(candidate)
break
else:
self.whisper_model_dir = DEFAULT_WHISPER_MODEL
# Auto-detect GGUF model
if not self.gguf_model_path:
gguf_dir = self.models_dir / "qwen-gguf"
if gguf_dir.exists():
# Prefer Q4_K_M, then any .gguf
q4 = list(gguf_dir.glob("*q4_k_m*.gguf"))
any_gguf = list(gguf_dir.glob("*.gguf"))
candidates = q4 or any_gguf
if candidates:
self.gguf_model_path = str(candidates[0])
# Auto-detect transformers fallback model
if not self.planner_model_path:
for candidate_dir in (
self.models_dir / "qwen2.5-0.5b-instruct",
self.models_dir / "qwen2.5-0.5b",
):
if (candidate_dir / "config.json").exists() and (
list(candidate_dir.glob("*.safetensors")) or list(candidate_dir.glob("pytorch_model*"))
):
self.planner_model_path = str(candidate_dir)
break
if not self.llm_model_path:
self.llm_model_path = self.planner_model_path
# ═══════════════════════════════════════════════════════════════════
# Data structures
# ═══════════════════════════════════════════════════════════════════
@dataclass
class TranscriptionResult:
text: str
language: str
confidence: float
code_switched: bool = False
is_valid: bool = True
error: Optional[str] = None
@dataclass
class VoiceProfile:
key: str
display_name: str
tts_voice: str
reference_audio_path: Optional[str] = None
browser_keywords: tuple[str, ...] = ()
@dataclass
class SpeechActivityStats:
duration_sec: float
speech_sec: float
speech_ratio: float
rms: float
@dataclass
class ConversationTurn:
user_text: str
response_text: str
language: str
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
response_time_sec: float = 0.0
asr_confidence: float = 0.0
@dataclass
class SessionState:
history: list[ConversationTurn] = field(default_factory=list)
language_preference: str = "en"
voice_preference: str = "female"
custom_voice_path: Optional[str] = None
asr_confidence_scores: list[float] = field(default_factory=list)
error_log: list[str] = field(default_factory=list)
task_state: dict[str, Any] = field(default_factory=dict)
last_intent: str = ""
pending_clarification: str = ""
session_id: str = field(
default_factory=lambda: f"{datetime.now().strftime('%Y%m%d_%H%M%S')}-{uuid.uuid4().hex[:8]}"
)
def add_turn(self, turn: ConversationTurn) -> None:
self.history.append(turn)
# FIX: Only update language_preference when ASR is confident enough.
# Previously any turn (even conf=0.32) overwrote the session language,
# causing every subsequent turn to be hinted with the wrong language.
if turn.asr_confidence >= 0.65 or turn.language == "en":
self.language_preference = turn.language
self.asr_confidence_scores.append(turn.asr_confidence)
def get_history_for_llm(self, max_turns: int = 6) -> list[dict]:
recent = self.history[-max_turns:]
messages = []
for t in recent:
messages.append({"role": "user", "content": t.user_text})
messages.append({"role": "assistant", "content": t.response_text})
return messages
def clear(self) -> None:
self.history.clear()
self.asr_confidence_scores.clear()
self.error_log.clear()
self.task_state.clear()
self.last_intent = ""
self.pending_clarification = ""
@dataclass
class IntentResult:
intent: str
entities: dict[str, Any] = field(default_factory=dict)
confidence: float = 0.0
needs_clarification: bool = False
clarification_question: str = ""
refusal_reason: str = ""
source: str = "rule"
@dataclass
class TruthDecision:
action: str # answer | clarify | refuse
reason: str = ""
missing_fields: list[str] = field(default_factory=list)
@dataclass
class AgentResponse:
text: str
intent: str
action: str
source: str = "verified"
cache_key: str = ""
# ═══════════════════════════════════════════════════════════════════
# System prompts
# ═══════════════════════════════════════════════════════════════════
# Updated system prompts - LLM agent-based only
SYSTEM_PROMPT = """You are Aara, the warm and intelligent voice concierge for Sahara Star Hotels Mumbai.
Your role:
- Help guests with hotel bookings, room info, amenities, dining, services, and general assistance
- Provide verified information from the hotel database
- Answer general knowledge questions when needed
- Always be warm, helpful, and speak naturally
Rules:
1. For hotel-specific data (prices, room names, policies, availability), ONLY use verified database information
2. Never invent hotel facts, prices, room types, or booking details
3. For non-hotel questions, answer helpfully from general knowledge
4. If you lack hotel information, say: Let me have our front desk confirm that for you
5. Use natural spoken language - never use markdown, bullets, brackets, or placeholder text
6. Keep responses concise: max 3-4 short spoken sentences
7. Ask only one question at a time when you need clarification
8. Use conversation history to remember guest preferences
9. Be warm, guest-focused, and immediately helpful
"""
GENERAL_PROMPT = """You are Aara, Sahara Star Hotels' voice concierge.
Approach every query as an intelligent agent:
1. Identify what the guest needs
2. Use hotel database for verified facts about rooms, prices, services, policies
3. If hotel information is missing, tell the guest to contact the front desk
4. For general questions, answer helpfully with your general knowledge
5. Keep language natural, warm, and spoken
6. Use conversation context to provide personalized assistance
7. Never use markdown or special formatting
"""
MODEL_ONLY_PROMPT = """You are Aara, Sahara Star Hotels' intelligent voice concierge.
Context: The guest is conversing with you via voice. Be natural, warm, and conversational.
Guide:
- Use verified hotel database information for all hotel specifications
- Answer general knowledge questions when appropriate
- Keep responses to 2-3 short spoken sentences maximum
- Never invent hotel details, prices, or policies
- Ask one clarification question at a time if you need information
- Reference conversation history to understand guest continuity
- Be immediately helpful and direct
- Never use markdown, special characters, or formatting
"""
# English-only voice configuration
LANGUAGE_VOICE_MAP: dict[str, tuple[str, str]] = {
"en": ("en-IN-NeerjaNeural", "en-IN-PrabhatNeural"),
}
SUPPORTED_LANGUAGES: list[str] = ["en"]
PHASE_PROMPTS: dict[str, str] = {} # Removed: all responses are LLM-based
# ═══════════════════════════════════════════════════════════════════
# Helpers
# ═══════════════════════════════════════════════════════════════════
def _normalise_text(text: str) -> str:
return re.sub(r"\s+", " ", text.lower()).strip()
def _date_from_relative_expression(expression: str) -> Optional[str]:
expression = _normalise_text(expression)
today = datetime.now().date()
if "day after tomorrow" in expression:
return (today + timedelta(days=2)).isoformat()
if "tomorrow" in expression:
return (today + timedelta(days=1)).isoformat()
if "today" in expression:
return today.isoformat()
if "next week" in expression:
return (today + timedelta(days=7)).isoformat()
direct_date = re.search(r"\b(\d{4}-\d{2}-\d{2})\b", expression)
if direct_date:
return direct_date.group(1)
slash_date = re.search(r"\b(\d{1,2}/\d{1,2}/\d{4})\b", expression)
if slash_date:
try:
parsed = datetime.strptime(slash_date.group(1), "%d/%m/%Y")
return parsed.date().isoformat()
except ValueError:
return None
month_names = {
"jan": 1, "january": 1, "feb": 2, "february": 2, "mar": 3, "march": 3,
"apr": 4, "april": 4, "may": 5, "jun": 6, "june": 6, "jul": 7, "july": 7,
"aug": 8, "august": 8, "sep": 9, "sept": 9, "september": 9, "oct": 10, "october": 10,
"nov": 11, "november": 11, "dec": 12, "december": 12,
}
day_month = re.search(
r"\b(\d{1,2})(?:st|nd|rd|th)?\s+([a-zA-Z]+)(?:\s*,?\s*(\d{4}))?\b",
expression,
)
month_day = re.search(
r"\b([a-zA-Z]+)\s+(\d{1,2})(?:st|nd|rd|th)?(?:\s*,?\s*(\d{4}))?\b",
expression,
)
candidate = None
if day_month and month_names.get(day_month.group(2).lower()):
candidate = (
int(day_month.group(1)),
month_names[day_month.group(2).lower()],
int(day_month.group(3)) if day_month.group(3) else None,
)
elif month_day and month_names.get(month_day.group(1).lower()):
candidate = (
int(month_day.group(2)),
month_names[month_day.group(1).lower()],
int(month_day.group(3)) if month_day.group(3) else None,
)
if candidate:
day, month, year = candidate
try:
parsed = datetime(year or today.year, month, day).date()
if year is None and parsed < today:
parsed = datetime(today.year + 1, month, day).date()
return parsed.isoformat()
except ValueError:
return None
return None
def _extract_dates_from_text(text: str) -> list[str]:
normalised = _normalise_text(text)
dates: list[str] = []
seen: set[str] = set()
def _add_date(raw: str) -> None:
parsed = _date_from_relative_expression(raw)
if parsed and parsed not in seen:
seen.add(parsed)
dates.append(parsed)
range_patterns = [
r"\b(\d{1,2})(?:st|nd|rd|th)?\s*(?:to|-|through|till|until)\s*(\d{1,2})(?:st|nd|rd|th)?\s+([a-zA-Z]+)(?:\s*,?\s*(\d{4}))?\b",
r"\b([a-zA-Z]+)\s+(\d{1,2})(?:st|nd|rd|th)?\s*(?:to|-|through|till|until)\s*(\d{1,2})(?:st|nd|rd|th)?(?:\s*,?\s*(\d{4}))?\b",
r"\b(today|tomorrow|day after tomorrow)\s*(?:to|-|through|till|until)\s*(today|tomorrow|day after tomorrow)\b",
]
for match in re.finditer(range_patterns[0], normalised):
year_suffix = f" {match.group(4)}" if match.group(4) else ""
_add_date(f"{match.group(1)} {match.group(3)}{year_suffix}")
_add_date(f"{match.group(2)} {match.group(3)}{year_suffix}")
for match in re.finditer(range_patterns[1], normalised):
year_suffix = f" {match.group(4)}" if match.group(4) else ""
_add_date(f"{match.group(1)} {match.group(2)}{year_suffix}")
_add_date(f"{match.group(1)} {match.group(3)}{year_suffix}")
for match in re.finditer(range_patterns[2], normalised):
_add_date(match.group(1))
_add_date(match.group(2))
date_patterns = [
r"\b(?:today|tomorrow|day after tomorrow|next week|\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{4})\b",
r"\b\d{1,2}(?:st|nd|rd|th)?\s+[a-zA-Z]+(?:\s*,?\s*\d{4})?\b",
r"\b[a-zA-Z]+\s+\d{1,2}(?:st|nd|rd|th)?(?:\s*,?\s*\d{4})?\b",
]
for pattern in date_patterns:
for match in re.finditer(pattern, normalised):
_add_date(match.group(0))
return dates
def _extract_duration_days(expression: str) -> Optional[int]:
text = _normalise_text(expression)
numeric_match = re.search(r"\b(\d+)\s+(day|days|night|nights)\b", text)
if numeric_match:
return max(1, int(numeric_match.group(1)))
word_map = {
"one": 1, "two": 2, "three": 3, "four": 4, "five": 5,
"six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10,
}
word_match = re.search(r"\b(one|two|three|four|five|six|seven|eight|nine|ten)\s+(day|days|night|nights)\b", text)
if word_match:
return word_map[word_match.group(1)]
return None
def _coerce_future_booking_date(value: str) -> str:
try:
parsed = datetime.fromisoformat(str(value).split("T")[0]).date()
except Exception:
return value
today = datetime.now().date()
if parsed >= today:
return parsed.isoformat()
if parsed.year < today.year:
try:
parsed = parsed.replace(year=today.year)
except ValueError:
return value
if parsed < today:
try:
parsed = parsed.replace(year=today.year + 1)
except ValueError:
return value
return parsed.isoformat()
# ═══════════════════════════════════════════════════════════════════
# Response cache (SQLite-backed, concurrent-safe)
# ═══════════════════════════════════════════════════════════════════
class ResponseCache:
"""
SQLite-backed response cache.
FIXED: replaces the JSON file cache which was not safe under concurrent writes.
"""
def __init__(self, cache_dir: Path) -> None:
self.db_path = str(cache_dir / "response_cache.db")
self._lock = threading.Lock()
self._init_db()
def _init_db(self) -> None:
with self._connect() as conn:
conn.execute(
"CREATE TABLE IF NOT EXISTS response_cache "
"(key TEXT PRIMARY KEY, value TEXT, ts INTEGER)"
)
conn.execute(
"CREATE TABLE IF NOT EXISTS audio_cache "
"(key TEXT PRIMARY KEY, path TEXT, ts INTEGER)"
)
def _connect(self) -> sqlite3.Connection:
conn = sqlite3.connect(self.db_path, timeout=10, check_same_thread=False)
conn.execute("PRAGMA journal_mode=WAL")
return conn
def get_response(self, key: str) -> Optional[str]:
try:
with self._connect() as conn:
row = conn.execute(
"SELECT value FROM response_cache WHERE key=?", (key,)
).fetchone()
return row[0] if row else None
except Exception as exc:
logger.debug("Cache read error: %s", exc)
return None
def set_response(self, key: str, value: str) -> None:
try:
with self._connect() as conn:
conn.execute(
"INSERT OR REPLACE INTO response_cache (key, value, ts) VALUES (?,?,?)",
(key, value.strip(), int(time.time())),
)
except Exception as exc:
logger.warning("Cache write error: %s", exc)
def get_audio(self, key: str) -> Optional[str]:
try:
with self._connect() as conn:
row = conn.execute(
"SELECT path FROM audio_cache WHERE key=?", (key,)
).fetchone()
if row and Path(row[0]).exists():
return row[0]
except Exception:
pass
return None
def set_audio(self, key: str, value: str) -> None:
try:
with self._connect() as conn:
conn.execute(
"INSERT OR REPLACE INTO audio_cache (key, path, ts) VALUES (?,?,?)",
(key, value, int(time.time())),
)
except Exception as exc:
logger.warning("Audio cache write error: %s", exc)
def evict_old_audio(self, max_age_seconds: int = 3600) -> int:
"""Delete audio cache entries older than max_age_seconds. Returns count deleted."""
cutoff = int(time.time()) - max_age_seconds
try:
with self._connect() as conn:
rows = conn.execute(
"SELECT path FROM audio_cache WHERE ts < ?", (cutoff,)
).fetchall()
# The above code is a Python loop that iterates over a sequence of tuples, where each
# tuple contains a single element. The loop unpacks each tuple into the variable
# `path` and then executes the code block denoted by `
for (path,) in rows:
try:
Path(path).unlink(missing_ok=True)
except Exception:
pass
cursor = conn.execute(
"DELETE FROM audio_cache WHERE ts < ?", (cutoff,)
)
return cursor.rowcount
except Exception as exc:
logger.warning("Audio cache eviction error: %s", exc)
return 0
# ═══════════════════════════════════════════════════════════════════
# Intent extractor (rule-based, no model needed)
# ═══════════════════════════════════════════════════════════════════
class IntentExtractor:
def __init__(self, db: "HotelDatabase") -> None:
self.db = db
self._room_types: list[str] = []
self._restaurants: list[str] = []
self._service_categories: list[str] = []
@staticmethod
def _contains_phrase(text: str, phrase: str) -> bool:
return bool(re.search(rf"\b{re.escape(phrase)}\b", text))
@classmethod
def _contains_any_phrase(cls, text: str, phrases: list[str]) -> bool:
return any(cls._contains_phrase(text, phrase) for phrase in phrases)
def _refresh_catalogues(self) -> None:
try:
self._room_types = [
r.get("name", "").lower()
for r in self.db.get_room_types()
if r.get("name")
]
except Exception:
self._room_types = []
try:
self._restaurants = [
r.get("name", "").lower()
for r in self.db.execute_query(
"SELECT DISTINCT name FROM Restaurants WHERE is_active = 1"
)
if r.get("name")
]
except Exception:
self._restaurants = []
try:
self._service_categories = [
r.get("category", "").lower()
for r in self.db.execute_query(
"SELECT DISTINCT category FROM Services WHERE is_active = 1"
)
if r.get("category")
]
except Exception:
self._service_categories = []
@staticmethod
def _token_set(text: str) -> set[str]:
return set(re.findall(r"[a-z0-9]+", text))
def _match_room_type(self, normalised: str) -> Optional[str]:
if not normalised:
return None
text_tokens = self._token_set(normalised)
generic_tokens = {"room", "rooms", "suite", "suites", "view", "views", "with", "and"}
best_name: Optional[str] = None
best_score = 0.0
best_specificity = 0
for room_type in sorted(self._room_types, key=len, reverse=True):
if room_type and room_type in normalised:
return room_type
tokens = [token for token in self._token_set(room_type) if token not in generic_tokens]
if not tokens:
continue
matches = sum(1 for token in tokens if token in text_tokens)
if matches == 0:
continue
score = matches / len(tokens)
specificity = len(tokens)
if score > best_score or (abs(score - best_score) < 1e-6 and specificity > best_specificity):
best_name = room_type
best_score = score
best_specificity = specificity
if best_name and best_score >= 0.5:
return best_name
alias_map = {
"standard": "standard ac",
"deluxe": "deluxe room",
"club": "club room",
"junior": "junior suite",
"sahara": "sahara suite",
"presidential": "presidential suite",
}
for keyword, alias in alias_map.items():
if re.search(rf"\b{keyword}\b", normalised) and alias in self._room_types:
return alias
return None
@staticmethod
def _looks_like_emergency(normalised: str) -> bool:
emergency_patterns = [
r"\bfire\b",
r"\bmedical emergency\b",
r"\bemergency\b",
r"\bambulance\b",
r"\bbleeding\b",
r"\bheart attack\b",
r"\bhelp now\b",
r"\bdanger\b",
]
return any(re.search(pattern, normalised) for pattern in emergency_patterns)
@staticmethod
def _looks_like_complaint(normalised: str) -> bool:
explicit_terms = [
"complaint", "problem", "issue", "not working", "is not working", "isn't working",
"broken", "dirty", "smell", "smelly", "noisy", "water leak", "leak", "stain",
"bad service", "poor service",
]
if any(term in normalised for term in explicit_terms):
return True
issue_words = r"\b(no|not|broken|bad|issue|problem|complaint|dirty|smell|smelly|noisy|leak|urgent)\b"
hotel_assets = r"\b(ac|air conditioning|wifi|internet|tv|water|shower|bathroom|door|lock|room service|cleaning|towels?)\b"
return bool(re.search(issue_words, normalised) and re.search(hotel_assets, normalised))
def extract(self, text: str, language: str = "en") -> IntentResult:
self._refresh_catalogues()
normalised = _normalise_text(text.strip())
entities: dict[str, Any] = {}
if not normalised:
return IntentResult(
intent="unknown",
needs_clarification=True,
clarification_question="Please say that again.",
)
social_greeting_terms = ["hello", "hi", "hey"] # English only
social_gratitude_terms = ["thank you", "thanks"] # English only
end_terms = ["bye", "goodbye", "see you", "talk later", "that's all", "that is all"]
has_substantive = any(
t in normalised
for t in [
"book", "booking", "reserve", "reservation", "available", "availability",
"price", "pricing", "rate", "cost", "room type", "room types", "room", "suite",
"restaurant", "menu", "dine", "dining",
"service", "spa", "transfer", "laundry", "gym", "pool", "reservation status",
"my reservation", "booking status", "check in", "check-in", "check out",
"check-out", "phone", "contact", "call", "location", "address", "amenities",
"facilities", "complaint", "problem", "issue", "broken", "emergency",
]
)
if self._contains_any_phrase(normalised, end_terms) and not has_substantive:
return IntentResult(intent="end_conversation", confidence=0.98)
if self._contains_any_phrase(normalised, ["how are you", "good morning", "good evening", "who are you", "tell me a joke"]):
if not has_substantive:
return IntentResult(intent="general_chit_chat", confidence=0.98)
if self._contains_any_phrase(normalised, social_gratitude_terms) and not has_substantive:
return IntentResult(intent="gratitude", confidence=0.98)
if self._contains_any_phrase(normalised, social_greeting_terms) and not has_substantive:
return IntentResult(intent="greeting", confidence=0.98)
# Determine primary intent
intent = "general_knowledge"
room_type_terms = [
"room types", "types of rooms", "which rooms", "room categories",
"room category", "room options", "available room types",
]
if self._looks_like_emergency(normalised):
intent = "emergency"
elif self._looks_like_complaint(normalised):
intent = "hotel_complaint"
elif any(t in normalised for t in room_type_terms):
intent = "room_types"
elif any(t in normalised for t in ["book", "reserve", "reservation"]):
intent = "booking"
elif any(t in normalised for t in ["available", "availability", "vacancy"]):
intent = "availability"
elif any(t in normalised for t in ["price", "pricing", "rate", "cost"]):
intent = "pricing"
elif any(t in normalised for t in ["restaurant", "menu", "dine", "dining", "food"]):
intent = "restaurant"
elif any(t in normalised for t in ["service", "spa", "laundry", "gym", "pool"]):
intent = "services"
elif any(t in normalised for t in ["reservation status", "my reservation", "booking status"]):
intent = "reservation_status"
elif any(t in normalised for t in ["check in", "check-in", "arrival time"]):
intent = "faq_check_in"
elif any(t in normalised for t in ["check out", "check-out", "departure time"]):
intent = "faq_check_out"
elif any(t in normalised for t in ["phone", "contact", "call", "number", "email"]):
intent = "faq_contact"
elif any(t in normalised for t in ["location", "address", "directions", "how do i reach"]):
intent = "faq_location"
elif any(t in normalised for t in ["amenities", "facilities", "features"]):
intent = "faq_amenities"
# Entity extraction
email_match = re.search(r"[\w\.-]+@[\w\.-]+\.\w+", text)
if email_match:
entities["guest_email"] = email_match.group(0).strip()
phone_match = re.search(r"(?:\+?\d[\d\-\s]{7,}\d)", text)
if phone_match:
entities["guest_phone"] = phone_match.group(0).strip()
matched_room_type = self._match_room_type(normalised)
if matched_room_type:
entities["room_type"] = matched_room_type
parsed_dates = _extract_dates_from_text(normalised)
if parsed_dates:
entities["dates"] = parsed_dates
if len(parsed_dates) >= 1:
entities["check_in"] = parsed_dates[0]
if len(parsed_dates) >= 2:
entities["check_out"] = parsed_dates[1]
adults_match = re.search(r"(\d+)\s*(adults?|guests?|people|persons|members?)", normalised)
if adults_match:
entities["num_adults"] = int(adults_match.group(1))
else:
word_to_number = {
"one": 1, "two": 2, "three": 3, "four": 4, "five": 5,
"six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10,
}
word_match = re.search(
r"\b(one|two|three|four|five|six|seven|eight|nine|ten)\s*(adults?|guests?|people|persons|members?)\b",
normalised,
)
if word_match:
entities["num_adults"] = word_to_number.get(word_match.group(1), 0)
for restaurant in self._restaurants:
if restaurant and restaurant in normalised:
entities["restaurant_name"] = restaurant
break
for category in self._service_categories:
if category and category in normalised:
entities["service_category"] = category
break
if intent in {"hotel_complaint", "emergency"}:
entities["complaint_text"] = text.strip()
room_match = re.search(r"\broom\s*(?:no\.?|number)?\s*([0-9]{1,4}[a-z]?)\b", normalised)
if room_match:
entities["room_number"] = room_match.group(1)
guest_name_match = re.search(
r"(?:my name is|i am|this is)\s+([a-zA-Z][a-zA-Z\s\.\-]{1,60})",
text,
flags=re.IGNORECASE,
)
if guest_name_match:
entities["guest_name"] = guest_name_match.group(1).strip().rstrip(".?!")
confidence = 0.72
if intent in {"greeting", "gratitude", "end_conversation"}:
confidence = 0.98
elif intent in {"availability", "pricing", "room_types", "services", "restaurant", "reservation_status", "booking", "hotel_complaint", "emergency"}:
confidence = 0.84
elif intent in {"general_knowledge", "general_chit_chat"}:
confidence = 0.80
elif intent.startswith("faq_"):
confidence = 0.92
if entities:
confidence = min(0.99, confidence + 0.05)
return IntentResult(intent=intent, entities=entities, confidence=confidence)
# ═══════════════════════════════════════════════════════════════════
# Truth gate
# ═══════════════════════════════════════════════════════════════════
class TruthGate:
@staticmethod
def _booking_clarification(intent_name: str, missing_fields: list[str]) -> str:
if intent_name == "booking":
if "check_in" in missing_fields and "check_out" in missing_fields:
return "What dates would you like to stay?"
if "check_in" in missing_fields:
return "What is your check-in date?"
if "check_out" in missing_fields:
return "What is your check-out date?"
if "num_adults" in missing_fields:
return "How many guests will be staying?"
if "room_type" in missing_fields:
return "What room type would you prefer?"
if "guest_name" in missing_fields:
return "What name should I use for the booking?"
if "guest_email" in missing_fields:
return "What email address should I use for the booking?"
if intent_name in {"availability", "pricing"}:
if "check_in" in missing_fields and "check_out" in missing_fields:
return "What dates are you planning to stay?"
if "check_in" in missing_fields:
return "What is your check-in date?"
if "check_out" in missing_fields:
return "What is your check-out date?"
if intent_name == "reservation_status":
return "What email address should I use to look it up?"
if intent_name == "hotel_complaint":
return "What is your room number?"
return ""
def evaluate(self, intent: IntentResult) -> TruthDecision:
if intent.refusal_reason:
return TruthDecision(action="refuse", reason=intent.refusal_reason)
if intent.needs_clarification and intent.clarification_question:
return TruthDecision(action="clarify", reason=intent.clarification_question)
if intent.intent in {
"greeting", "gratitude", "end_conversation", "general_knowledge",
"general_chit_chat", "hybrid",
}:
return TruthDecision(action="answer")
if intent.intent == "unknown":
return TruthDecision(action="refuse", reason="Only verified hotel questions can be answered safely.")
if intent.intent == "emergency":
return TruthDecision(action="answer")
if intent.intent == "booking":
missing = [
f for f in ["check_in", "check_out", "num_adults", "room_type", "guest_name", "guest_email"]
if not intent.entities.get(f)
]
if missing:
return TruthDecision(action="clarify", reason=self._booking_clarification("booking", missing), missing_fields=missing)
return TruthDecision(action="answer")
if intent.intent in {"availability", "pricing"}:
missing = [f for f in ["check_in", "check_out"] if not intent.entities.get(f)]
if missing:
return TruthDecision(action="clarify", reason=self._booking_clarification(intent.intent, missing), missing_fields=missing)
return TruthDecision(action="answer")
if intent.intent == "room_types":
return TruthDecision(action="answer")
if intent.intent == "reservation_status":
if not intent.entities.get("guest_email"):
return TruthDecision(action="clarify", reason=self._booking_clarification("reservation_status", ["guest_email"]), missing_fields=["guest_email"])
return TruthDecision(action="answer")
if intent.intent in {
"services", "restaurant", "faq_check_in", "faq_check_out",
"faq_contact", "faq_location", "faq_amenities",
}:
return TruthDecision(action="answer")
if intent.intent == "hotel_complaint":
if not intent.entities.get("room_number"):
return TruthDecision(action="clarify", reason=self._booking_clarification("hotel_complaint", ["room_number"]), missing_fields=["room_number"])
return TruthDecision(action="answer")
return TruthDecision(action="refuse", reason="That request cannot be verified from trusted hotel data.")
# ═══════════════════════════════════════════════════════════════════
# Verified tool executor
# ═══════════════════════════════════════════════════════════════════
class VerifiedToolExecutor:
def __init__(self, db: "HotelDatabase") -> None:
self.db = db
def execute(self, intent: IntentResult) -> dict[str, Any]:
if intent.intent in {"faq_check_in", "faq_check_out", "faq_contact", "faq_location", "faq_amenities"}:
return {"profile": self.db.get_hotel_profile()}
if intent.intent == "availability":
return {"rooms": self.db.get_available_rooms(intent.entities["check_in"], intent.entities["check_out"], intent.entities.get("room_type"))}
if intent.intent == "pricing":
return {"room_types": self.db.get_room_types(intent.entities.get("room_type"))}
if intent.intent == "room_types":
return {"room_types": self.db.get_room_types(intent.entities.get("room_type"))}
if intent.intent == "services":
return {"services": self.db.get_services_by_category(intent.entities.get("service_category"))}
if intent.intent == "restaurant":
return {"menu": self.db.get_restaurant_menu(intent.entities.get("restaurant_name"))}
if intent.intent == "reservation_status":
return {"reservations": self.db.get_guest_reservations(intent.entities["guest_email"])}
if intent.intent == "booking":
return {"booking": self.db.create_booking(intent.entities)}
if intent.intent == "hotel_complaint":
complaint_text = intent.entities.get("complaint_text", "")
severity = "critical" if any(t in _normalise_text(complaint_text) for t in ["emergency", "fire", "medical", "urgent"]) else "normal"
complaint = self.db.log_complaint(
intent.entities.get("session_id", ""),
complaint_text,
room_number=intent.entities.get("room_number"),
severity=severity,
escalation_flag=1 if severity == "critical" else 0,
)
return {"complaint": complaint}
if intent.intent == "emergency":
complaint = self.db.log_complaint(
intent.entities.get("session_id", ""),
intent.entities.get("complaint_text", ""),
room_number=intent.entities.get("room_number"),
severity="critical",
escalation_flag=1,
)
return {"complaint": complaint}
return {"profile": self.db.get_hotel_profile()}
# ═══════════════════════════════════════════════════════════════════
# ASR — Multilingual, faster-whisper-medium
# ═══════════════════════════════════════════════════════════════════
class MultilingualASR:
"""
Speech-to-text using faster-whisper (medium by default on CPU).
FIXED: model defaults to 'medium' instead of the weaker 'small'.
FIXED: multi-pass ASR disabled by default.
ADDED: optional Silero VAD pre-screen.
"""
_HOTEL_PROMPT = (
"Natural spoken conversation in English or Indian-accented English. "
"The speaker may talk quickly, slowly, loudly, softly, or with background room noise. "
"Questions may be about hotel stays, travel, Mumbai, local information, or general knowledge. "
"Transcribe exactly what the speaker says. "
"Common words include Sahara Star, room booking, reservation, check in, check out, suite, "
"availability, tariff, breakfast, WiFi, airport shuttle, concierge, restaurant, spa, taxi, "
"weather, directions, nearby places, and thank you."
)
_HOTEL_HOTWORDS = (
"Sahara Star, Sahara Star Hotel, Mumbai, booking, reservation, check in, check out, "
"airport, concierge, room service, spa, gym, deluxe suite, weather, taxi, city, travel"
)
def __init__(self, config: AaraConfig) -> None:
self.cfg = config
self._model = None
self._vad_model = None
self._model_lock = threading.Lock()
def _load_model(self) -> None:
if self._model is not None:
return
try:
from faster_whisper import WhisperModel
except ImportError:
raise RuntimeError("faster-whisper not installed. Run: pip install faster-whisper")
source = self.cfg.whisper_model_dir or DEFAULT_WHISPER_MODEL
logger.info(" Loading Whisper from %s ...", source)
self._model = WhisperModel(
source,
device="cpu",
compute_type="int8",
cpu_threads=self.cfg.llm_n_threads,
num_workers=1,
)
logger.info(" ✅ Whisper loaded")
def _load_vad(self) -> None:
if self._vad_model is not None:
return
try:
from silero_vad import get_speech_timestamps, load_silero_vad
model = load_silero_vad(onnx=False)
self._vad_model = (model, get_speech_timestamps)
logger.info(" ✅ Silero VAD loaded")
except Exception as exc:
logger.warning(" Silero VAD unavailable (will skip VAD pre-screen): %s", exc)
self._vad_model = None
@staticmethod
def _looks_like_asr_artifact(text: str) -> bool:
lowered = _normalise_text(text)
if not lowered:
return False
alnum_chars = len(re.findall(r"[a-z0-9]", lowered))
if alnum_chars == 0 and len(lowered) >= 6:
return True
if lowered.count("_") >= 6 and alnum_chars <= 4:
return True
artifact_patterns = [
r"^transcribed by https?://",
r"\botter\.ai\b",
r"\bamara\.org\b",
r"\bsubtitles by\b",
r"\bcaption(s)? by\b",
r"\bthanks? for watching\b",
r"\bthank you for watching\b",
r"\bvisit our website\b",
r"\bwww\.[a-z0-9.-]+\.[a-z]{2,}\b",
]
return any(re.search(pattern, lowered) for pattern in artifact_patterns)
def _preprocess_audio(self, audio: np.ndarray) -> np.ndarray:
if audio.ndim > 1:
audio = audio.mean(axis=1)
audio = np.asarray(audio, dtype=np.float32).reshape(-1)
if audio.size == 0:
return audio
audio = np.nan_to_num(audio, nan=0.0, posinf=0.0, neginf=0.0)
audio = audio - float(np.mean(audio))
peak = float(np.max(np.abs(audio)))
if peak > 0:
audio = audio / peak
if audio.size >= 1024:
try:
nyquist = self.cfg.sample_rate / 2.0
low_cut = max(60.0, min(90.0, nyquist * 0.25))
high_cut = min(7600.0, nyquist * 0.95)
if high_cut > low_cut:
b, a = signal.butter(2, [low_cut / nyquist, high_cut / nyquist], btype="bandpass")
audio = signal.filtfilt(b, a, audio).astype(np.float32, copy=False)
except Exception:
pass
rms = float(np.sqrt(np.mean(np.square(audio), dtype=np.float64))) if audio.size else 0.0
if rms > 1e-6:
target_rms = 0.18
gain = float(np.clip(target_rms / rms, 0.75, 4.0))
audio = audio * gain
# Soft compression makes loud or aggressive speech less likely to clip while
# still lifting quieter speech enough for CPU VAD and Whisper.
audio = np.tanh(audio * 2.2) / np.tanh(2.2)
peak = float(np.max(np.abs(audio)))
if peak > 0:
audio = np.clip(audio / peak * 0.95, -1.0, 1.0)
return audio.astype(np.float32, copy=False)
def _prepare_vad_audio(self, audio: np.ndarray) -> np.ndarray:
if audio.ndim > 1:
audio = audio.mean(axis=1)
audio = np.asarray(audio, dtype=np.float32).reshape(-1)
if audio.size == 0:
return audio
audio = np.nan_to_num(audio, nan=0.0, posinf=0.0, neginf=0.0)
audio = audio - float(np.mean(audio))
rms = float(np.sqrt(np.mean(np.square(audio), dtype=np.float64))) if audio.size else 0.0
if rms > 1e-6:
vad_gain = float(np.clip(0.12 / rms, 1.0, 5.0))
audio = audio * vad_gain
audio = np.tanh(audio * 1.8) / np.tanh(1.8)
peak = float(np.max(np.abs(audio)))
if peak > 0:
audio = np.clip(audio / peak * 0.95, -1.0, 1.0)
return audio.astype(np.float32, copy=False)
def _get_speech_timestamps(self, audio: np.ndarray) -> Optional[list[dict[str, int]]]:
if self._vad_model is None:
return None
try:
import torch
model, get_speech_timestamps = self._vad_model
wav = torch.from_numpy(audio.astype(np.float32))
timestamps = get_speech_timestamps(
wav,
model,
sampling_rate=self.cfg.sample_rate,
threshold=self.cfg.asr_vad_threshold,
min_speech_duration_ms=self.cfg.asr_vad_min_speech_ms,
min_silence_duration_ms=self.cfg.asr_vad_min_silence_ms,
speech_pad_ms=self.cfg.asr_vad_speech_pad_ms,
)
return [dict(ts) for ts in timestamps]
except Exception as exc:
logger.debug(" Silero VAD timestamp extraction failed: %s", exc)
return None # Fail open — let Whisper decide
def _summarize_speech(
self,
audio: np.ndarray,
timestamps: Optional[list[dict[str, int]]] = None,
) -> SpeechActivityStats:
total_samples = max(1, int(audio.size))
duration_sec = total_samples / float(self.cfg.sample_rate)
rms = float(np.sqrt(np.mean(np.square(audio), dtype=np.float64))) if audio.size else 0.0
if not timestamps:
return SpeechActivityStats(duration_sec=duration_sec, speech_sec=duration_sec, speech_ratio=1.0, rms=rms)
speech_samples = 0
for ts in timestamps:
start = max(0, int(ts.get("start", 0)))
end = min(total_samples, int(ts.get("end", 0)))
if end > start:
speech_samples += end - start
speech_sec = speech_samples / float(self.cfg.sample_rate)
return SpeechActivityStats(
duration_sec=duration_sec,
speech_sec=speech_sec,
speech_ratio=(speech_samples / float(total_samples)) if total_samples else 0.0,
rms=rms,
)
def _trim_to_speech(self, audio: np.ndarray, timestamps: list[dict[str, int]]) -> np.ndarray:
if not timestamps:
return audio
merge_gap = int(self.cfg.sample_rate * 0.12)
merged: list[list[int]] = []
for ts in timestamps:
start = max(0, int(ts.get("start", 0)))
end = min(audio.size, int(ts.get("end", 0)))
if end <= start:
continue
if merged and start - merged[-1][1] <= merge_gap:
merged[-1][1] = max(merged[-1][1], end)
else:
merged.append([start, end])
if not merged:
return np.array([], dtype=np.float32)
chunks = [audio[start:end] for start, end in merged if end > start]
if not chunks:
return np.array([], dtype=np.float32)
trimmed = np.concatenate(chunks)
return trimmed.astype(np.float32, copy=False)
def warmup(self) -> None:
self._load_model()
if self.cfg.enable_silero_vad:
self._load_vad()
def _transcribe_raw(
self,
audio: np.ndarray,
language: Optional[str] = None,
*,
allow_forced_language: bool = True,
use_whisper_vad: bool = True,
) -> TranscriptionResult:
self._load_model()
max_val = float(np.max(np.abs(audio)))
if max_val > 1.0:
audio = audio / max_val * 0.95
# Explicit user/session hints win first; the force-language setting is a guardrail.
# Set cfg.asr_force_language = "" to re-enable pure auto-detection.
forced_language = self.cfg.asr_force_language.strip() if allow_forced_language else ""
if self.cfg.english_only_mode:
forced_language = "en"
effective_language = language or forced_language or None
segments, info = self._model.transcribe(
audio,
language=effective_language,
beam_size=1,
best_of=1,
temperature=0.0,
condition_on_previous_text=False,
vad_filter=use_whisper_vad,
vad_parameters={
"min_silence_duration_ms": self.cfg.asr_vad_min_silence_ms,
"speech_pad_ms": self.cfg.asr_vad_speech_pad_ms,
} if use_whisper_vad else None,
no_speech_threshold=0.60, # NEW: silence segments filtered out
log_prob_threshold=-1.0, # NEW: discard very low-probability segments
compression_ratio_threshold=2.2,
hallucination_silence_threshold=self.cfg.asr_hallucination_silence_threshold,
initial_prompt=self._HOTEL_PROMPT,
hotwords=self._HOTEL_HOTWORDS,
language_detection_threshold=max(self.cfg.asr_language_prob_threshold, 0.50),
language_detection_segments=2,
)
texts = []
avg_probs = []
no_speech_probs = []
for seg in segments:
texts.append(seg.text.strip())
avg_probs.append(seg.avg_logprob)
no_speech_probs.append(float(getattr(seg, "no_speech_prob", 0.0)))
full_text = " ".join(t for t in texts if t)
avg_logprob = float(np.mean(avg_probs)) if avg_probs else -2.0
confidence = float(np.clip(np.exp(avg_logprob), 0.0, 1.0))
if no_speech_probs:
confidence = float(np.clip(confidence * max(0.15, 1.0 - float(np.mean(no_speech_probs))), 0.0, 1.0))
detected_lang = info.language if hasattr(info, "language") else "en"
# FIX: Language-detection gate — if Whisper is uncertain AND no explicit
# language hint was given, fall back to "en" and penalise confidence.
# Skip this gate entirely when a language is forced (force lock already set effective_language).
lang_prob = float(getattr(info, "language_probability", 1.0))
if not forced_language and language is None and lang_prob < self.cfg.asr_language_prob_threshold:
detected_lang = "en"
confidence = confidence * lang_prob # punish uncertain detection
if self.cfg.english_only_mode:
detected_lang = "en"
code_switched = bool(
re.search(r"[\u0900-\u097F]", full_text) and re.search(r"[a-zA-Z]{3,}", full_text)
)
return TranscriptionResult(
text=full_text,
language=detected_lang,
confidence=confidence,
code_switched=code_switched,
is_valid=bool(full_text),
)
def transcribe(
self,
audio: np.ndarray,
language_hint: Optional[str] = None,
confidence_threshold: Optional[float] = None,
rerun_threshold: Optional[float] = None,
unload_models_between_turns: Optional[bool] = None,
request_id: str = "",
) -> TranscriptionResult:
if audio is None or len(audio) == 0:
return TranscriptionResult("", "en", 0.0, is_valid=False, error="Empty audio")
conf_threshold = confidence_threshold or self.cfg.asr_confidence_threshold
rerun_threshold = rerun_threshold or self.cfg.asr_confidence_rerun_threshold
unload = unload_models_between_turns if unload_models_between_turns is not None else self.cfg.unload_models_between_turns
vad_audio = self._prepare_vad_audio(audio)
audio = self._preprocess_audio(audio)
max_samples = int(self.cfg.max_audio_length * self.cfg.sample_rate)
if max_samples > 0 and audio.size > max_samples:
audio = audio[:max_samples]
vad_audio = vad_audio[:max_samples]
speech_stats = self._summarize_speech(vad_audio)
logger.info(
" [%s] Audio stats: len=%.2fs rms=%.4f",
request_id,
speech_stats.duration_sec,
speech_stats.rms,
)
timestamps: Optional[list[dict[str, int]]] = None
use_whisper_vad = True
if self.cfg.enable_silero_vad:
if self._vad_model is None:
self._load_vad()
timestamps = self._get_speech_timestamps(vad_audio)
if timestamps == []:
if (
speech_stats.rms >= self.cfg.asr_vad_rms_fallback_threshold
or (
speech_stats.duration_sec >= self.cfg.asr_vad_duration_fallback_sec
and speech_stats.rms >= max(self.cfg.min_speech_rms * 2.5, 0.008)
)
):
logger.info(
" [%s] VAD: empty timestamps but audio energy is strong enough; falling back to Whisper",
request_id,
)
timestamps = None
use_whisper_vad = False
else:
logger.info(" [%s] VAD: no speech detected — skipping Whisper", request_id)
return TranscriptionResult("", "en", 0.0, is_valid=False, error="No speech detected")
if timestamps:
speech_stats = self._summarize_speech(vad_audio, timestamps)
logger.info(
" [%s] VAD speech: %.2fs of %.2fs (%.0f%%)",
request_id,
speech_stats.speech_sec,
speech_stats.duration_sec,
speech_stats.speech_ratio * 100.0,
)
if (
speech_stats.speech_sec < self.cfg.asr_min_trimmed_speech_sec
and speech_stats.rms < max(self.cfg.min_speech_rms * 3.0, 0.018)
):
return TranscriptionResult(
"",
"en",
0.0,
is_valid=False,
error="I only caught a very short sound. Please say a full sentence.",
)
if (
speech_stats.speech_ratio < self.cfg.asr_min_speech_ratio
and speech_stats.speech_sec < 1.0
and speech_stats.rms < max(self.cfg.min_speech_rms * 3.0, 0.018)
):
return TranscriptionResult(
"",
"en",
0.0,
is_valid=False,
error="Please move a little closer to the microphone and try again.",
)
trimmed = self._trim_to_speech(audio, timestamps)
if trimmed.size:
audio = trimmed
use_whisper_vad = False
with self._model_lock:
try:
t0 = time.time()
logger.info(" [%s] ASR start (hint=%s)", request_id, language_hint or "auto")
result = self._transcribe_raw(
audio,
language_hint,
allow_forced_language=True,
use_whisper_vad=use_whisper_vad,
)
if (
self.cfg.asr_retry_without_language_lock
and (not result.text.strip() or result.confidence < rerun_threshold)
and (self.cfg.asr_force_language.strip() or language_hint)
):
retry = self._transcribe_raw(
audio,
None,
allow_forced_language=False,
use_whisper_vad=use_whisper_vad,
)
if retry.text.strip() and retry.confidence >= (result.confidence + 0.05):
logger.info(
" [%s] ASR retry selected: conf %.2f -> %.2f",
request_id,
result.confidence,
retry.confidence,
)
result = retry
elapsed = time.time() - t0
logger.info(
" [%s] ASR done: %r (lang=%s conf=%.2f %.1fs)",
request_id, result.text[:80], result.language, result.confidence, elapsed,
)
if result.text.strip() and self._looks_like_asr_artifact(result.text):
logger.warning(" [%s] ASR artifact rejected (%r)", request_id, result.text[:80])
result.is_valid = False
result.error = "Could not understand clearly. Please say that again."
return result
result.is_valid = bool(result.text.strip()) and result.confidence >= conf_threshold
# FIX: Minimum word-count gate — single-word results with borderline
# confidence are almost always misdetections (e.g. "Boquerum", "Stand review.")
word_count = len(result.text.split())
if result.is_valid and word_count <= self.cfg.asr_min_word_count and result.confidence < 0.80:
logger.warning(" [%s] ASR: single-word low-conf result rejected (%r conf=%.2f)", request_id, result.text, result.confidence)
result.is_valid = False
result.error = "Could not understand clearly. Please say a complete sentence."
except Exception as exc:
logger.error(" [%s] ASR error: %s", request_id, exc)
result = TranscriptionResult("", "en", 0.0, is_valid=False, error=str(exc))
finally:
if unload and self._model is not None:
self._model = None
gc.collect()
return result
# ═══════════════════════════════════════════════════════════════════
# LLM agent — GGUF first, transformers fallback
# ═══════════════════════════════════════════════════════════════════
class LLMAgent:
"""
FIXED: Tries GGUF (llama-cpp-python) before Transformers.
Q4_K_M GGUF is ~330 MB and runs at ~20 tok/s on CPU vs ~2 tok/s for float32.
"""
def __init__(self, config: AaraConfig, db: "HotelDatabase") -> None:
self.cfg = config
self.db = db
self.cache = ResponseCache(self.cfg.cache_dir)
self.intent_extractor = IntentExtractor(db)
self.truth_gate = TruthGate()
self.tool_executor = VerifiedToolExecutor(db)
self._llm = None
self._tokenizer = None
self._backend = "uninitialised"
self._lock = threading.Lock()
def warmup(self) -> None:
self._load_model()
# ── Model loading ─────────────────────────────────────────────
def _load_model(self) -> None:
if self._llm is not None:
return
# 1. Try GGUF via llama-cpp-python (preferred for CPU)
gguf_path = self._find_gguf()
if gguf_path:
self._load_gguf(gguf_path)
if self._llm is not None:
return
# 2. Fallback to transformers
self._load_transformers()
def _find_gguf(self) -> Optional[str]:
"""Find the best GGUF file on disk."""
# Explicit config path wins
if self.cfg.gguf_model_path and Path(self.cfg.gguf_model_path).exists():
return self.cfg.gguf_model_path
# Scan models dir
gguf_dir = self.cfg.models_dir / "qwen-gguf"
if gguf_dir.exists():
q4 = sorted(gguf_dir.glob("*q4_k_m*.gguf"))
any_gguf = sorted(gguf_dir.glob("*.gguf"))
candidates = q4 or any_gguf
if candidates:
return str(candidates[0])
# Recursive scan
for candidate in sorted(self.cfg.models_dir.rglob("*.gguf")):
return str(candidate)
return None
def _load_gguf(self, gguf_path: str) -> None:
try:
from llama_cpp import Llama
logger.info(" Loading GGUF from %s ...", Path(gguf_path).name)
self._llm = Llama(
model_path=gguf_path,
n_ctx=self.cfg.llm_n_ctx,
n_batch=self.cfg.llm_n_batch,
n_threads=self.cfg.llm_n_threads,
n_gpu_layers=0, # CPU-only
verbose=self.cfg.llm_verbose,
chat_format="chatml", # Works for Qwen, Mistral, Llama-3
)
self._backend = "gguf"
logger.info(" ✅ GGUF LLM loaded (%s, ctx=%d)", Path(gguf_path).name, self.cfg.llm_n_ctx)
except ImportError:
logger.warning(" llama-cpp-python not installed; trying transformers")
except Exception as exc:
logger.warning(" GGUF load failed: %s; trying transformers", exc)
def _load_transformers(self) -> None:
model_name = self.cfg.llm_model_path or self.cfg.planner_model_path or self.cfg.planner_model_name
if not model_name:
self._backend = "rules"
return
try:
import torch
from transformers import AutoModelForCausalLM, AutoTokenizer
logger.info(" Loading Transformers LLM from %s ...", model_name)
cache_dir = str(self.cfg.cache_dir / "hf_models")
self._tokenizer = AutoTokenizer.from_pretrained(
model_name, cache_dir=cache_dir, trust_remote_code=True
)
if self._tokenizer.pad_token_id is None and self._tokenizer.eos_token_id is not None:
self._tokenizer.pad_token = self._tokenizer.eos_token
self._llm = AutoModelForCausalLM.from_pretrained(
model_name,
cache_dir=cache_dir,
trust_remote_code=True,
torch_dtype=torch.float32,
)
self._llm.to("cpu").eval()
self._backend = "transformers"
logger.info(" ✅ Transformers LLM loaded (%s)", model_name)
except Exception as exc:
logger.warning(" Transformers load failed: %s — rule-based mode only", exc)
self._backend = "rules"
self._llm = None
self._tokenizer = None
def _unload_model(self) -> None:
self._llm = None
self._tokenizer = None
self._backend = "unloaded"
gc.collect()
# ── Inference ──────────────────────────────────────────────────
def _invoke_model(self, messages: list[dict], max_tokens: Optional[int] = None) -> str:
if self._llm is None:
self._load_model()
if self._llm is None:
raise RuntimeError("LLM backend unavailable")
tokens = max_tokens or self.cfg.llm_max_tokens
if self._backend == "transformers":
if self._tokenizer is None:
raise RuntimeError("Tokenizer unavailable")
import torch
if hasattr(self._tokenizer, "apply_chat_template"):
prompt_text = self._tokenizer.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
else:
lines = []
for m in messages:
lines.append(f"{m.get('role','user').upper()}: {m.get('content','')}")
lines.append("ASSISTANT:")
prompt_text = "\n".join(lines)
encoded = self._tokenizer(
prompt_text, return_tensors="pt", truncation=True, max_length=512
)
encoded = {k: v.to(self._llm.device) for k, v in encoded.items()}
with torch.inference_mode():
generated = self._llm.generate(
**encoded,
max_new_tokens=tokens,
do_sample=True,
temperature=self.cfg.llm_temperature,
top_p=0.95,
repetition_penalty=self.cfg.llm_repeat_penalty,
pad_token_id=self._tokenizer.pad_token_id or self._tokenizer.eos_token_id,
)
new_tokens = generated[0][encoded["input_ids"].shape[-1]:]
return self._tokenizer.decode(new_tokens, skip_special_tokens=True).strip()
# GGUF / llama-cpp path (also works for any create_chat_completion-compatible backend)
response = self._llm.create_chat_completion(
messages=messages,
temperature=self.cfg.llm_temperature,
max_tokens=tokens,
top_p=0.95,
repeat_penalty=self.cfg.llm_repeat_penalty,
)
return self._extract_response_text(response)
@staticmethod
def _extract_response_text(response: Any) -> str:
if isinstance(response, str):
return response.strip()
if isinstance(response, dict):
choices = response.get("choices")
if isinstance(choices, list) and choices:
msg = choices[0].get("message", {})
if isinstance(msg, dict):
content = msg.get("content")
if isinstance(content, str):
return content.strip()
text = choices[0].get("text")
if isinstance(text, str):
return text.strip()
return str(response).strip()
@staticmethod
def _strip_tool_tags(text: str) -> str:
text = re.sub(r".*?", "", text, flags=re.IGNORECASE | re.DOTALL)
text = text.replace("", "")
text = re.sub(r"\[(?:FETCH_DB|SAVE_DB|BOOKING_READY):[^\]]*\]", "", text, flags=re.IGNORECASE)
text = re.sub(r"\bBOOKING_COMPLETE:\s*[^\n]+", "", text, flags=re.IGNORECASE)
return re.sub(r"\s+", " ", text).strip()
@staticmethod
def _sanitize_spoken_text(text: str, *, max_questions: int = 1) -> str:
text = LLMAgent._strip_tool_tags(text)
text = re.sub(r"\*{1,2}|_{1,2}|`+|~+", "", text)
text = re.sub(r"^[\s\-*•\d\.\)]+", "", text, flags=re.MULTILINE)
text = text.replace("[", "").replace("]", "").replace("{", "").replace("}", "")
text = re.sub(r"\s+", " ", text).strip()
if max_questions >= 0 and text.count("?") > max_questions:
first_q = text.find("?")
prefix = text[: first_q + 1]
suffix = text[first_q + 1:].replace("?", ".")
text = f"{prefix}{suffix}"
text = re.sub(r"\s+([?.!,])", r"\1", text)
text = re.sub(r"([?.!,]){2,}", r"\1", text)
return text.strip()
def _build_cache_key(
self,
user_input: str,
transcription: TranscriptionResult,
history: list[dict],
session: Optional[SessionState] = None,
) -> str:
session_context = {}
if session is not None:
session_context = {
"last_intent": session.last_intent,
"pending_clarification": session.pending_clarification,
"task_state": session.task_state,
"voice_preference": session.voice_preference,
}
return hashlib.sha256(
json.dumps(
{
"language": transcription.language,
"user_input": user_input,
"history": history[-4:],
"session": session_context,
},
sort_keys=True, ensure_ascii=False,
default=str,
).encode()
).hexdigest()
# ── Response generation pipeline ──────────────────────────────
def _extract_follow_up_entities(
self,
text: str,
pending_intent: str,
existing_entities: Optional[dict[str, Any]] = None,
pending_fields: Optional[list[str]] = None,
) -> dict[str, Any]:
entities = dict(existing_entities or {})
extracted = self.intent_extractor.extract(text)
entities.update({k: v for k, v in extracted.entities.items() if v not in (None, "", [], {})})
normalised = _normalise_text(text)
existing_dates: list[str] = list(existing_entities.get("dates", [])) if isinstance((existing_entities or {}).get("dates"), list) else []
newly_parsed_dates = _extract_dates_from_text(normalised)
if pending_intent in {"booking", "availability", "pricing"}:
newly_parsed_dates = [_coerce_future_booking_date(date) for date in newly_parsed_dates]
pending_fields = list(pending_fields or [])
if newly_parsed_dates:
if len(newly_parsed_dates) >= 2:
entities["check_in"] = newly_parsed_dates[0]
entities["check_out"] = newly_parsed_dates[1]
elif len(newly_parsed_dates) == 1:
only_date = newly_parsed_dates[0]
if "check_out" in pending_fields and "check_in" not in pending_fields and existing_entities and existing_entities.get("check_in"):
entities["check_out"] = only_date
entities["check_in"] = str(existing_entities.get("check_in"))
elif "check_in" in pending_fields:
entities["check_in"] = only_date
elif not entities.get("check_in"):
entities["check_in"] = only_date
elif not entities.get("check_out") and entities.get("check_in") != only_date:
entities["check_out"] = only_date
merged_dates: list[str] = []
for date in existing_dates + newly_parsed_dates:
if date and date not in merged_dates:
merged_dates.append(date)
entities["dates"] = merged_dates
if entities.get("check_in") and entities.get("check_out"):
entities["dates"] = [str(entities["check_in"]), str(entities["check_out"])]
duration_days = _extract_duration_days(text)
if duration_days:
entities["stay_nights"] = duration_days
if entities.get("check_in") and not entities.get("check_out"):
try:
check_in = datetime.fromisoformat(str(entities["check_in"]).split("T")[0]).date()
entities["check_out"] = (check_in + timedelta(days=duration_days)).isoformat()
except Exception:
pass
if pending_intent in {"booking", "availability", "pricing"} and entities.get("check_in"):
entities["check_in"] = _coerce_future_booking_date(str(entities["check_in"]))
if pending_intent in {"booking", "availability", "pricing"} and entities.get("check_out"):
entities["check_out"] = _coerce_future_booking_date(str(entities["check_out"]))
if pending_intent == "booking" and "num_adults" not in entities:
adults_match = re.search(r"\b(\d+|one|two|three|four|five|six|seven|eight|nine|ten)\s+(members?|guests?|people|persons|adults?)\b", normalised)
if adults_match:
word = adults_match.group(1)
word_to_number = {
"one": 1, "two": 2, "three": 3, "four": 4, "five": 5,
"six": 6, "seven": 7, "eight": 8, "nine": 9, "ten": 10,
}
entities["num_adults"] = int(word) if word.isdigit() else word_to_number.get(word, 1)
return entities
def _merge_pending_intent(self, user_input: str, session: Optional[SessionState], intent: IntentResult) -> IntentResult:
if session is None:
return intent
pending_intent = str(session.task_state.get("pending_intent") or "").strip()
pending_fields = session.task_state.get("pending_fields") or []
stored_entities = session.task_state.get("collected_entities") or {}
if not pending_intent:
return intent
merged_entities = self._extract_follow_up_entities(
user_input,
pending_intent,
cast(dict[str, Any], stored_entities),
list(pending_fields),
)
merged_entities.update({k: v for k, v in intent.entities.items() if v not in (None, "", [], {})})
follow_up_text = _normalise_text(user_input)
is_short_follow_up = len(follow_up_text.split()) <= 6
looks_like_standalone_question = (
len(follow_up_text.split()) > 4
and re.search(
r"^(what|when|where|who|why|how|can|could|would|should|do|does|did|is|are|which|tell me|explain)\b",
follow_up_text,
)
)
explicit_other_intent = (
intent.intent not in {"general_knowledge", "general_chit_chat", "unknown", "hybrid"}
and intent.intent != pending_intent
)
should_use_pending = (
not explicit_other_intent
and (
(
intent.intent in {"general_knowledge", "general_chit_chat", "unknown", "hybrid"}
and not looks_like_standalone_question
)
or is_short_follow_up
or any(field in {"check_in", "check_out"} for field in pending_fields)
)
)
if not should_use_pending:
return intent
return IntentResult(
intent=pending_intent,
entities=merged_entities,
confidence=max(intent.confidence, 0.82),
source="session_follow_up",
)
@staticmethod
def _update_session_dialog_state(
session: Optional[SessionState],
intent: IntentResult,
decision: TruthDecision,
*,
preserve_existing_pending: bool = False,
previous_pending_state: Optional[dict[str, Any]] = None,
) -> None:
if session is None:
return
session.last_intent = intent.intent
if decision.action == "clarify":
session.pending_clarification = decision.reason
session.task_state["pending_intent"] = intent.intent
session.task_state["pending_fields"] = list(decision.missing_fields)
session.task_state["collected_entities"] = dict(intent.entities)
elif preserve_existing_pending and previous_pending_state and previous_pending_state.get("pending_intent"):
session.pending_clarification = str(previous_pending_state.get("pending_clarification") or "")
session.task_state["pending_intent"] = previous_pending_state.get("pending_intent")
session.task_state["pending_fields"] = list(previous_pending_state.get("pending_fields") or [])
session.task_state["collected_entities"] = dict(previous_pending_state.get("collected_entities") or {})
else:
session.pending_clarification = ""
session.task_state.pop("pending_intent", None)
session.task_state.pop("pending_fields", None)
session.task_state.pop("collected_entities", None)
@staticmethod
def _format_context_fields(fields: dict[str, Any]) -> str:
parts = []
for key, value in fields.items():
if value in (None, "", [], {}):
continue
if isinstance(value, list):
rendered = ", ".join(str(item) for item in value[:6])
elif isinstance(value, dict):
rendered = json.dumps(value, ensure_ascii=False, default=str)
else:
rendered = str(value)
parts.append(f"{key}={rendered}")
return ", ".join(parts) if parts else "none"
def _append_context_section(
self,
sections: list[str],
title: str,
value: Any,
*,
max_rows: int = 6,
) -> None:
rendered = self._compact_context_value(value, max_rows=max_rows)
if rendered:
sections.append(f"{title}: {rendered}")
@staticmethod
def _compact_row(row: dict[str, Any], *, max_fields: int = 5) -> str:
preferred_keys = [
"name", "room_type", "room_number", "confirmation_number", "reservation_id",
"check_in", "check_out", "status", "price_per_night", "weekend_price",
"total_price", "capacity", "bed_type", "view_type", "size_sqft",
"category", "price", "description", "location", "phone", "email",
"check_in_time", "check_out_time", "complaint_id", "severity",
]
parts: list[str] = []
seen: set[str] = set()
for key in preferred_keys + list(row.keys()):
if key in seen:
continue
seen.add(key)
value = row.get(key)
if value in (None, "", [], {}):
continue
parts.append(f"{key}={value}")
if len(parts) >= max_fields:
break
return ", ".join(parts)
def _compact_context_value(self, value: Any, *, max_rows: int = 3) -> str:
if value in (None, "", [], {}):
return ""
if isinstance(value, dict):
return self._compact_row(value)
if isinstance(value, list):
rows: list[str] = []
for item in value[:max_rows]:
if isinstance(item, dict):
rows.append(self._compact_row(item))
else:
rows.append(str(item))
if len(value) > max_rows:
rows.append(f"+{len(value) - max_rows} more")
return " | ".join(row for row in rows if row)
return str(value)
@staticmethod
def _trim_text_middle(text: str, max_chars: int) -> str:
if len(text) <= max_chars:
return text
head = max_chars // 2
tail = max_chars - head - 5
return f"{text[:head]}\n...\n{text[-tail:]}"
def _fit_messages_to_context(
self,
messages: list[dict[str, str]],
*,
max_tokens: Optional[int] = None,
) -> list[dict[str, str]]:
token_budget = max(256, self.cfg.llm_n_ctx - (max_tokens or self.cfg.llm_max_tokens) - 48)
char_budget = token_budget * 4
trimmed = [{"role": m.get("role", "user"), "content": str(m.get("content", "")).strip()} for m in messages]
while len(trimmed) > 2 and sum(len(m["content"]) for m in trimmed) > char_budget:
del trimmed[1]
if trimmed:
system_cap = max(900, int(char_budget * 0.62))
trimmed[0]["content"] = self._trim_text_middle(trimmed[0]["content"], system_cap)
if len(trimmed) >= 2:
user_cap = max(320, int(char_budget * 0.25))
trimmed[-1]["content"] = self._trim_text_middle(trimmed[-1]["content"], user_cap)
return trimmed
def _build_turn_system_prompt(
self,
*,
transcription: TranscriptionResult,
intent: IntentResult,
decision: TruthDecision,
context: str,
session: Optional[SessionState],
first_turn: bool,
booking_completed: bool,
) -> str:
concierge_name = "Raj" if (session and session.voice_preference == "male") else "Priya"
lines = [
MODEL_ONLY_PROMPT,
f"Today is {datetime.now().date().isoformat()}. Your current concierge name is {concierge_name}.",
]
if self.cfg.english_only_mode:
lines.append("English only mode is enabled. Always reply in English.")
phase_prompt = PHASE_PROMPTS.get(intent.intent)
if phase_prompt:
lines.append(phase_prompt)
if intent.intent == "end_conversation":
lines.append("Closing flow: thank the guest warmly and invite them to visit again.")
elif intent.intent == "gratitude":
lines.append("Thank-you flow: acknowledge warmly and keep it brief.")
elif intent.intent == "greeting":
lines.append("Greeting flow: greet warmly and invite the guest's request.")
elif decision.action == "clarify" and decision.reason:
lines.append(f"Next step: ask only this one question: {decision.reason}")
elif decision.action == "refuse":
lines.append(
"Out-of-scope flow: politely redirect the guest to bookings, rooms, dining, spa, local area, or other Sahara Star hotel help."
)
elif intent.intent in {"general_knowledge", "general_chit_chat", "hybrid", "unknown"}:
lines.append("Answer helpfully if the guest stays near travel or hotel help. Otherwise redirect back to Sahara Star assistance.")
else:
lines.append("Use the verified context directly and speak like a polished concierge, not a database.")
if booking_completed:
lines.append("A verified booking has just been completed. Confirm it clearly and end with a short thank-you.")
if intent.intent == "emergency":
lines.append("Say the team is being alerted right now and give the front desk emergency instruction immediately.")
if first_turn:
lines.append(f"First turn: greet the guest and introduce yourself as {concierge_name} before helping.")
if context:
lines.append(f"Verified context:\n{context}")
return "\n\n".join(lines)
@staticmethod
def _is_english_response(language: str) -> bool:
lowered = (language or "en").strip().lower()
return lowered in {"", "en", "en-in", "english"}
@staticmethod
def _spoken_date(value: Any) -> str:
text = str(value or "").strip()
if not text:
return ""
try:
parsed = datetime.fromisoformat(text.split("T")[0]).date()
return parsed.strftime("%B %d")
except Exception:
return text
@staticmethod
def _spoken_price(value: Any) -> str:
try:
amount = float(value)
return f"{amount:,.0f} rupees"
except Exception:
return str(value)
@staticmethod
def _join_items(items: list[str]) -> str:
cleaned = [item.strip() for item in items if item and item.strip()]
if not cleaned:
return ""
if len(cleaned) == 1:
return cleaned[0]
if len(cleaned) == 2:
return f"{cleaned[0]} and {cleaned[1]}"
return ", ".join(cleaned[:-1]) + f", and {cleaned[-1]}"
def _generate_fast_response(
self,
*,
transcription: TranscriptionResult,
session: Optional[SessionState],
intent: IntentResult,
decision: TruthDecision,
verified_data: dict[str, Any],
) -> Optional[tuple[str, str]]:
return None
def _run_model_attempts(
self,
base_messages: list[dict[str, str]],
*,
max_tokens: Optional[int] = None,
fallback: str,
) -> str:
attempt_messages = [base_messages]
attempt_messages.append(
base_messages[:-1] + [{
"role": "system",
"content": "Your previous draft was empty or weak. Reply now with one short, concrete spoken answer or one short clarification question."
}, base_messages[-1]]
)
try:
for messages in attempt_messages:
fitted_messages = self._fit_messages_to_context(messages, max_tokens=max_tokens)
response = self._invoke_model(fitted_messages, max_tokens=max_tokens or self.cfg.llm_max_tokens)
response = self._sanitize_spoken_text(response, max_questions=1)
if response:
return response
except Exception as exc:
logger.warning("LLM generation failed: %s", exc)
return fallback
def _resolve_turn_grounding(
self,
user_input: str,
session: Optional[SessionState] = None,
) -> tuple[IntentResult, TruthDecision, dict[str, Any]]:
previous_pending_state = None
if session is not None and session.task_state.get("pending_intent"):
previous_pending_state = {
"pending_intent": session.task_state.get("pending_intent"),
"pending_fields": list(session.task_state.get("pending_fields") or []),
"collected_entities": dict(session.task_state.get("collected_entities") or {}),
"pending_clarification": session.pending_clarification,
}
intent = self.intent_extractor.extract(user_input)
intent = self._merge_pending_intent(user_input, session, intent)
if intent.intent in {"hotel_complaint", "emergency"} and session is not None:
intent.entities.setdefault("session_id", session.session_id)
decision = self.truth_gate.evaluate(intent)
verified_data: dict[str, Any] = {}
actionable_hotel_intents = {
"availability", "pricing", "room_types", "services", "restaurant",
"reservation_status", "booking", "hotel_complaint", "emergency",
"faq_check_in", "faq_check_out", "faq_contact", "faq_location", "faq_amenities",
}
if intent.intent in actionable_hotel_intents:
try:
profile = self.db.get_hotel_profile()
if profile:
verified_data["profile"] = profile
except Exception:
pass
if intent.intent == "room_types":
try:
room_types = self.db.get_room_types(intent.entities.get("room_type"))
if room_types:
verified_data["room_types"] = room_types
except Exception:
pass
elif intent.intent == "booking" and not intent.entities.get("room_type"):
try:
room_types = self.db.get_room_types()
if room_types:
verified_data["room_types"] = room_types
except Exception:
pass
elif decision.action == "answer" and intent.intent not in {"greeting", "gratitude", "end_conversation", "general_knowledge", "general_chit_chat", "hybrid", "unknown"}:
try:
verified_data.update(self.tool_executor.execute(intent))
except Exception as exc:
logger.warning("Context tool execution failed for %s: %s", intent.intent, exc)
preserve_existing_pending = bool(
previous_pending_state
and decision.action == "answer"
and intent.intent != previous_pending_state.get("pending_intent")
and intent.intent not in {"greeting", "gratitude", "end_conversation", "emergency"}
and (
len(_normalise_text(user_input).split()) > 4
or intent.intent not in {"general_knowledge", "general_chit_chat", "hybrid", "unknown"}
)
)
self._update_session_dialog_state(
session,
intent,
decision,
preserve_existing_pending=preserve_existing_pending,
previous_pending_state=previous_pending_state,
)
if session is not None:
booking_payload = verified_data.get("booking")
if isinstance(booking_payload, dict) and booking_payload.get("confirmation_number") and not booking_payload.get("error"):
session.task_state["booking_completed"] = True
session.task_state["booking_result"] = dict(booking_payload)
elif intent.intent != "booking":
session.task_state.pop("booking_completed", None)
session.task_state.pop("booking_result", None)
complaint_payload = verified_data.get("complaint")
if isinstance(complaint_payload, dict) and complaint_payload.get("complaint_id"):
session.task_state["complaint_logged"] = True
if intent.intent == "emergency":
session.task_state["emergency_logged"] = True
elif intent.intent not in {"hotel_complaint", "emergency"}:
session.task_state.pop("complaint_logged", None)
session.task_state.pop("emergency_logged", None)
if intent.intent == "end_conversation":
session.task_state["conversation_should_close"] = True
elif intent.intent not in {"greeting", "gratitude"}:
session.task_state.pop("conversation_should_close", None)
return intent, decision, verified_data
def _build_context(
self,
user_input: str,
session: Optional[SessionState] = None,
*,
intent: Optional[IntentResult] = None,
decision: Optional[TruthDecision] = None,
verified_data: Optional[dict[str, Any]] = None,
) -> str:
if intent is None:
intent = self.intent_extractor.extract(user_input)
intent = self._merge_pending_intent(user_input, session, intent)
if decision is None:
decision = self.truth_gate.evaluate(intent)
verified_data = dict(verified_data or {})
sections: list[str] = [
f"Intent={intent.intent}; action={decision.action}; confidence={intent.confidence:.2f}; entities={self._format_context_fields(intent.entities)}"
]
if session is not None and (
session.last_intent or session.pending_clarification or session.task_state.get("pending_intent")
):
sections.append(
"State: "
f"last_intent={session.last_intent or 'none'}; "
f"pending_intent={session.task_state.get('pending_intent') or 'none'}; "
f"missing_fields={', '.join(session.task_state.get('pending_fields') or []) or 'none'}; "
f"pending_question={session.pending_clarification or 'none'}; "
f"collected={self._format_context_fields(cast(dict[str, Any], session.task_state.get('collected_entities') or {}))}"
)
self._append_context_section(sections, "Profile", verified_data.get("profile"), max_rows=1)
self._append_context_section(sections, "Room types", verified_data.get("room_types"), max_rows=3)
self._append_context_section(sections, "Available rooms", verified_data.get("rooms"), max_rows=3)
self._append_context_section(sections, "Services", verified_data.get("services"), max_rows=4)
self._append_context_section(sections, "Menu", verified_data.get("menu"), max_rows=4)
self._append_context_section(sections, "Reservations", verified_data.get("reservations"), max_rows=2)
self._append_context_section(sections, "Booking result", verified_data.get("booking"), max_rows=1)
self._append_context_section(sections, "Complaint", verified_data.get("complaint"), max_rows=1)
return "\n".join(section for section in sections if section).strip()
def _generate_model_response(
self,
user_input: str,
transcription: TranscriptionResult,
history: list[dict],
max_tokens: Optional[int] = None,
session: Optional[SessionState] = None,
*,
intent: Optional[IntentResult] = None,
decision: Optional[TruthDecision] = None,
verified_data: Optional[dict[str, Any]] = None,
) -> str:
if self._llm is None:
try:
self._load_model()
except Exception:
return "I can help with Sahara Star questions. What would you like to know?"
if self._llm is None or self._backend == "rules":
return "I can help with Sahara Star questions. What would you like to know?"
if intent is None or decision is None or verified_data is None:
intent, decision, verified_data = self._resolve_turn_grounding(user_input, session=session)
context = self._build_context(
user_input,
session=session,
intent=intent,
decision=decision,
verified_data=verified_data,
)
booking_completed = bool(isinstance(verified_data.get("booking"), dict) and verified_data.get("booking", {}).get("confirmation_number"))
first_turn = len(history) == 0 and not (session and session.task_state.get("opening_greeting_sent"))
system_prompt = self._build_turn_system_prompt(
transcription=transcription,
intent=intent,
decision=decision,
context=context,
session=session,
first_turn=first_turn,
booking_completed=booking_completed,
)
base_messages: list[dict] = [{"role": "system", "content": system_prompt}]
for m in history[-(self.cfg.max_history_turns):]:
if m.get("content"):
base_messages.append({"role": m.get("role", "user"), "content": m["content"]})
base_messages.append({
"role": "user",
"content": (
f"Today is {datetime.now().date().isoformat()}.\n"
f"Guest language: {transcription.language}\n"
f"Guest says: {user_input.strip()}\n"
"Reply naturally and directly with one concise spoken response."
),
})
if decision.action == "clarify" and decision.reason:
fallback = self._sanitize_spoken_text(decision.reason, max_questions=1) or "What would you like help with?"
elif decision.action == "refuse":
fallback = "I'm here to help with Sahara Star bookings, rooms, dining, and hotel services. What would you like help with?"
elif intent.intent == "end_conversation" or booking_completed:
fallback = "Thank you for visiting Sahara Star. We look forward to welcoming you again."
elif intent.intent == "emergency":
fallback = "Please stay calm. I'm alerting our team right now. You can also call the front desk immediately."
else:
fallback = "I can help with Sahara Star questions and general questions. What would you like to know?"
return self._run_model_attempts(base_messages, max_tokens=max_tokens, fallback=fallback)
def _generate_special_response(
self,
instruction: str,
*,
language: str = "en",
session: Optional[SessionState] = None,
max_tokens: Optional[int] = None,
include_history: bool = True,
fallback: str,
) -> str:
if self._llm is None:
try:
self._load_model()
except Exception:
return fallback
if self._llm is None or self._backend == "rules":
return fallback
concierge_name = "Raj" if (session and session.voice_preference == "male") else "Priya"
try:
profile = self.db.get_hotel_profile()
except Exception:
profile = {}
hotel_name = str(profile.get("name", "Sahara Star")).strip()
location = str(profile.get("location", "")).strip()
context_sections = [
f"Concierge name: {concierge_name}",
f"Hotel name: {hotel_name}",
]
if location:
context_sections.append(f"Hotel location: {location}")
if session is not None and (
session.last_intent or session.pending_clarification or session.task_state
):
context_sections.append(
"Conversation state: "
f"last_intent={session.last_intent or 'none'}, "
f"pending_question={session.pending_clarification or 'none'}, "
f"task_state={self._format_context_fields(cast(dict[str, Any], session.task_state or {}))}"
)
if self.cfg.english_only_mode:
language_sentence = "English only mode is enabled. Always reply in English."
system_prompt = (
f"{MODEL_ONLY_PROMPT}\n\n"
f"Today is {datetime.now().date().isoformat()}.\n"
f"{language_sentence}\n\n"
"Special task: produce one short spoken concierge reply for the current situation.\n"
f"Situation details: {instruction}\n\n"
"Available context:\n" + "\n".join(context_sections)
)
base_messages: list[dict[str, str]] = [{"role": "system", "content": system_prompt}]
if include_history and session is not None:
for message in session.get_history_for_llm(max_turns=min(4, self.cfg.max_history_turns)):
if message.get("content"):
base_messages.append({"role": message.get("role", "user"), "content": str(message["content"])})
base_messages.append({
"role": "user",
"content": (
f"Today is {datetime.now().date().isoformat()}.\n"
f"Guest language: {language}\n"
f"Situation: {instruction}\n"
"Reply naturally and directly. Never return an empty answer."
),
})
return self._run_model_attempts(base_messages, max_tokens=max_tokens, fallback=fallback)
def generate_greeting(self, language: str, voice_preference: str = "female", session: Optional[SessionState] = None) -> str:
if self.cfg.english_only_mode:
language = "en"
try:
profile = self.db.get_hotel_profile()
except Exception:
profile = {}
hotel_name = str(profile.get("name", "Sahara Star")).strip()
concierge_name = "Raj" if voice_preference == "male" else "Priya"
location = str(profile.get("location", "")).strip()
suffix = f" in {location}" if location else ""
fallback = f"Hello! Welcome to {hotel_name}{suffix}. I'm {concierge_name}, and I'm here to help. What can I do for you today?"
return self._generate_special_response(
f"Open the conversation. Welcome the guest to {hotel_name}{suffix}. Introduce yourself as {concierge_name}. Ask one short question about how you can help.",
language=language,
session=session,
include_history=False,
fallback=fallback,
)
def generate_no_input_prompt(self, language: str = "en", session: Optional[SessionState] = None) -> str:
if self.cfg.english_only_mode:
language = "en"
return self._generate_special_response(
"The guest did not say anything. Ask one short question about what they would like you to do or help with.",
language=language,
session=session,
include_history=True,
fallback="What would you like me to help you with today?",
)
def generate_unclear_prompt(self, language: str = "en", session: Optional[SessionState] = None) -> str:
if self.cfg.english_only_mode:
language = "en"
return self._generate_special_response(
"You heard audio from the guest but could not understand it clearly. In one short sentence, ask them to repeat what they need help with.",
language=language,
session=session,
include_history=True,
fallback="I didn't catch that clearly. What would you like help with?",
)
def generate_goodbye(self, language: str = "en", session: Optional[SessionState] = None, reason: str = "closing") -> str:
if self.cfg.english_only_mode:
language = "en"
reason_text = {
"booking_complete": "The booking is complete and the conversation is naturally ending.",
"silence": "The guest stayed silent after a follow-up prompt and the conversation is ending.",
"closing": "The conversation is ending now.",
}.get(reason, "The conversation is ending now.")
return self._generate_special_response(
f"{reason_text} Thank the guest warmly and invite them to visit again.",
language=language,
session=session,
include_history=True,
fallback="Thank you for visiting Sahara Star. We look forward to welcoming you again.",
)
def generate(
self,
user_input: str,
transcription: TranscriptionResult,
history: list[dict],
max_tokens: Optional[int] = None,
session: Optional[SessionState] = None,
request_id: str = "",
) -> str:
with self._lock:
try:
t0 = time.time()
intent, decision, verified_data = self._resolve_turn_grounding(user_input, session=session)
cache_key = self._build_cache_key(user_input, transcription, history, session=session)
cached = self.cache.get_response(cache_key)
if cached:
logger.info(" [%s] Response cache hit", request_id)
if session is not None:
session.task_state["last_route"] = "cache"
return cached
route = "model_grounded"
response = self._generate_model_response(
user_input,
transcription,
history,
max_tokens=max_tokens,
session=session,
intent=intent,
decision=decision,
verified_data=verified_data,
)
response = self._strip_tool_tags(response)
response = re.sub(r"\s+", " ", response).strip()
# Cap response length for TTS (browser SpeechSynthesis handles arbitrary length, but keep it concise)
if len(response) > 400:
sentences = re.split(r"(?<=[.!?])\s+", response)
capped = []
total = 0
for s in sentences:
if total + len(s) > 400:
break
capped.append(s)
total += len(s)
response = " ".join(capped) if capped else response[:400]
self.cache.set_response(cache_key, response)
if session is not None:
session.task_state["last_route"] = route
elapsed = time.time() - t0
logger.info(" [%s] Response done (%.1fs route=%s): %s...", request_id, elapsed, route, response[:80])
return response
except Exception as exc:
logger.error(" [%s] LLM error: %s", request_id, exc)
if session is not None:
session.task_state["last_route"] = "exception"
return "I can help with Sahara Star questions. What would you like to know?"
finally:
if self.cfg.unload_models_between_turns:
self._unload_model()
# ═══════════════════════════════════════════════════════════════════
# Hotel database (imported or minimal fallback)
# ═══════════════════════════════════════════════════════════════════
try:
from create_hotel_database import HotelDatabase, initialize_database
except ImportError:
class HotelDatabase: # type: ignore[no-redef]
"""Minimal inline fallback."""
def __init__(self, db_path: str = "sahara_star.db") -> None:
self.db_path = db_path
self._conn: Optional[sqlite3.Connection] = None
if not Path(db_path).exists():
self._bootstrap()
def _bootstrap(self) -> None:
conn = sqlite3.connect(self.db_path)
conn.execute("CREATE TABLE IF NOT EXISTS Hotels (hotel_id INTEGER PRIMARY KEY, name TEXT, location TEXT, check_in_time TEXT, check_out_time TEXT, phone TEXT, email TEXT, website TEXT)")
conn.execute("INSERT OR IGNORE INTO Hotels VALUES (1,'Sahara Star','Mumbai International Airport Area','2:00 PM','12:00 PM','+91-22-6698-9898','reservations@saharastar.com','www.saharastar.com')")
conn.commit()
conn.close()
def _connect(self) -> sqlite3.Connection:
if self._conn is None:
self._conn = sqlite3.connect(self.db_path, check_same_thread=False)
self._conn.row_factory = sqlite3.Row
return self._conn
def execute_query(self, sql: str, params: tuple = ()) -> list[dict]:
try:
cur = self._connect().execute(sql, params)
return [dict(r) for r in cur.fetchall()] or [{"message": "No results"}]
except Exception as e:
return [{"error": str(e)}]
def get_hotel_profile(self) -> dict:
rows = self.execute_query("SELECT * FROM Hotels WHERE hotel_id=1")
return rows[0] if rows else {}
def get_room_types(self, room_type: Optional[str] = None) -> list[dict]:
if room_type:
return self.execute_query("SELECT * FROM RoomTypes WHERE LOWER(name) LIKE ?", (f"%{room_type.lower()}%",))
return self.execute_query("SELECT * FROM RoomTypes WHERE is_active=1 LIMIT 10")
def get_available_rooms(self, check_in: str, check_out: str, room_type: Optional[str] = None) -> list[dict]:
return self.execute_query("SELECT * FROM Rooms LIMIT 5")
def get_services_by_category(self, category: Optional[str] = None) -> list[dict]:
if category:
return self.execute_query("SELECT * FROM Services WHERE LOWER(category) LIKE ? AND is_active=1 LIMIT 8", (f"%{category.lower()}%",))
return self.execute_query("SELECT * FROM Services WHERE is_active=1 LIMIT 8")
def get_restaurant_menu(self, restaurant_name: Optional[str] = None) -> list[dict]:
return self.execute_query("SELECT * FROM MenuItems LIMIT 8")
def get_guest_reservations(self, email: str) -> list[dict]:
return self.execute_query("SELECT * FROM Bookings WHERE guest_email=? ORDER BY created_at DESC LIMIT 3", (email,))
def create_booking(self, entities: dict) -> dict:
return {"error": "Booking requires the full hotel database setup."}
def log_complaint(self, session_id: str, complaint_text: str, room_number: Optional[str] = None, category: Optional[str] = None, severity: str = "normal", escalation_flag: int = 0, guest_id: Optional[str] = None) -> dict:
return {"complaint_id": uuid.uuid4().hex[:8], "severity": severity, "escalation_flag": escalation_flag}
def save_session_snapshot(self, *args: Any, **kwargs: Any) -> None:
pass
def log_guest_language_preference(self, *args: Any, **kwargs: Any) -> None:
pass
def get_session_snapshot(self, session_id: str) -> Optional[dict]:
return None
def format_results_for_llm(self, results: list[dict], max_rows: int = 10) -> str:
limited = results[:max_rows]
return "\n".join(", ".join(f"{k}: {v}" for k, v in r.items()) for r in limited)
def close(self) -> None:
if self._conn:
self._conn.close()
self._conn = None
# ═══════════════════════════════════════════════════════════════════
# Audio I/O (microphone — not used in HF Spaces browser mode)
# ═══════════════════════════════════════════════════════════════════
class AudioIO:
def __init__(self, config: AaraConfig) -> None:
self.cfg = config
def is_human_voice(self, audio: np.ndarray) -> bool:
if audio.ndim > 1:
audio = audio.mean(axis=1)
sr = self.cfg.sample_rate
fft = np.fft.rfft(audio)
freqs = np.fft.rfftfreq(len(audio), 1.0 / sr)
formant_mask = (freqs >= self.cfg.formant_min_hz) & (freqs <= self.cfg.formant_max_hz)
formant_energy = np.sum(np.abs(fft[formant_mask]) ** 2)
total_energy = np.sum(np.abs(fft) ** 2) + 1e-10
return (formant_energy / total_energy) >= self.cfg.min_voice_energy_ratio
@staticmethod
def save_audio(audio: np.ndarray, path: str, sample_rate: int = 16000) -> bool:
try:
import soundfile as sf
sf.write(path, audio, sample_rate)
return True
except Exception:
try:
with wave.open(path, "w") as wf:
wf.setnchannels(1)
wf.setsampwidth(2)
wf.setframerate(sample_rate)
wf.writeframes((audio * 32767).astype(np.int16).tobytes())
return True
except Exception:
return False
# ═══════════════════════════════════════════════════════════════════
# TTS
# ═══════════════════════════════════════════════════════════════════
class TTSEngine:
"""Lightweight server-side TTS with graceful fallback."""
def __init__(self, config: AaraConfig) -> None:
self.cfg = config
self._backend_checked = False
self._available = False
self._lock = threading.Lock()
def _ensure_backend(self) -> bool:
if self._backend_checked:
return self._available
self._backend_checked = True
try:
import edge_tts # noqa: F401
self._available = True
logger.info(" ✅ Edge TTS ready")
except Exception as exc:
logger.warning(" Edge TTS unavailable; browser speech will be used instead: %s", exc)
self._available = False
return self._available
def warmup(self) -> None:
self._ensure_backend()
@property
def is_available(self) -> bool:
return self._ensure_backend()
def synthesize(self, text: str, voice: VoiceProfile, request_id: str = "") -> Optional[str]:
spoken_text = re.sub(r"\s+", " ", (text or "")).strip()
if not spoken_text or not self._ensure_backend():
return None
filename = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{uuid.uuid4().hex[:8]}.mp3"
output_path = self.cfg.output_audio_dir / filename
rate_delta = int(round((self.cfg.tts_voice_pace - 1.0) * 100.0))
rate = f"{rate_delta:+d}%"
async def _render() -> None:
import edge_tts
communicate = edge_tts.Communicate(text=spoken_text, voice=voice.tts_voice, rate=rate)
await communicate.save(str(output_path))
try:
with self._lock:
asyncio.run(_render())
logger.info(" [%s] TTS done: %s (%s)", request_id, output_path.name, voice.key)
return str(output_path)
except Exception as exc:
logger.warning(" [%s] TTS failed, falling back to browser speech: %s", request_id, exc)
try:
if output_path.exists():
output_path.unlink()
except Exception:
pass
return None
# ═══════════════════════════════════════════════════════════════════
# Main orchestrator
# ═══════════════════════════════════════════════════════════════════
class AaraVoiceAgent:
"""Main AARA voice agent — coordinates ASR, LLM, DB, and session state."""
def __init__(self, config: Optional[AaraConfig] = None) -> None:
self.cfg = config or AaraConfig()
self.session = SessionState()
logger.info("🚀 Initialising AARA Voice Agent ...")
self.db = HotelDatabase(self.cfg.db_path)
self.audio_io = AudioIO(self.cfg)
self.asr = MultilingualASR(self.cfg)
self.llm = LLMAgent(self.cfg, self.db)
self.tts = TTSEngine(self.cfg) if self.cfg.enable_server_tts else None
logger.info(
"⚙️ Config: whisper=%s | gguf=%s | ctx=%d | unload=%s | server_tts=%s",
Path(self.cfg.whisper_model_dir).name if self.cfg.whisper_model_dir else "default",
Path(self.cfg.gguf_model_path).name if self.cfg.gguf_model_path else "none yet",
self.cfg.llm_n_ctx,
self.cfg.unload_models_between_turns,
bool(self.tts),
)
logger.info("✅ AARA initialised")
def get_voice_profile(self, preference: Optional[str], session: Optional[SessionState] = None) -> VoiceProfile:
normalized = (preference or (session.voice_preference if session else "female") or "female").strip().lower()
if normalized == "custom" and session and session.custom_voice_path:
custom_path = Path(session.custom_voice_path)
return VoiceProfile(
key="custom",
display_name="Custom Voice",
tts_voice=self.cfg.tts_voice_female,
reference_audio_path=str(custom_path) if custom_path.exists() else None,
)
if normalized == "male":
reference = self.cfg.reference_audio_dir / "male_reference.wav"
return VoiceProfile(
key="male",
display_name="Raj",
tts_voice=self.cfg.tts_voice_male,
reference_audio_path=str(reference) if reference.exists() else None,
browser_keywords=("male", "raj", "prabhat", "guy", "david", "mark", "ravi", "man"),
)
reference = self.cfg.reference_audio_dir / "female_reference.wav"
return VoiceProfile(
key="female",
display_name="Priya",
tts_voice=self.cfg.tts_voice_female,
reference_audio_path=str(reference) if reference.exists() else None,
browser_keywords=("female", "priya", "neerja", "woman", "samantha", "zira", "aria", "heera"),
)
def build_voice_output(
self,
text: str,
*,
preference: Optional[str] = None,
session: Optional[SessionState] = None,
request_id: str = "",
) -> tuple[VoiceProfile, Optional[str]]:
profile = self.get_voice_profile(preference, session=session)
audio_path = self.tts.synthesize(text, profile, request_id=request_id) if self.tts else None
return profile, audio_path
@staticmethod
def _repair_low_conf_transcription(text: str) -> str:
repaired = re.sub(r"\s+", " ", (text or "")).strip()
if not repaired:
return repaired
lowered = _normalise_text(repaired)
lowered = re.sub(r"\bbooker\b", "book", lowered)
lowered = re.sub(r"\bbooking room\b", "book a room", lowered)
lowered = re.sub(r"\bbook room\b", "book a room", lowered)
lowered = re.sub(r"\bbooked room\b", "book a room", lowered)
lowered = re.sub(r"\bcheck two members\b", "for two members", lowered)
lowered = re.sub(r"\btwo member\b", "two members", lowered)
lowered = re.sub(r"\bto members\b", "two members", lowered)
if "types of rooms" in lowered and "available" in lowered:
return "what types of rooms are available in the hotel?"
has_date_info = bool(_extract_dates_from_text(lowered))
has_guest_count = bool(
re.search(
r"\b(\d+|one|two|three|four|five|six|seven|eight|nine|ten)\s+(members?|guests?|people|persons|adults?)\b",
lowered,
)
)
if has_date_info or has_guest_count:
return lowered
has_room = bool(re.search(r"\b(room|suite)\b", lowered))
has_booking = bool(re.search(r"\b(book|booking|reserve|reservation)\b", lowered))
has_two_guests = bool(re.search(r"\b(two|2)\s+(members?|guests?|people|persons)\b", lowered))
if has_room and has_booking and has_two_guests:
return "book a room for two members"
if has_room and has_two_guests and not has_booking:
return "book a room for two members"
if has_room and has_booking and not has_two_guests:
return "book a room"
return lowered
@staticmethod
def _is_domain_salvageable(text: str, session: Optional[SessionState] = None) -> bool:
lowered = _normalise_text(text)
if not lowered:
return False
if MultilingualASR._looks_like_asr_artifact(lowered):
return False
word_count = len(lowered.split())
if len(lowered.split()) < 2:
if not session or not session.task_state.get("pending_intent"):
return False
patterns = (
r"\b(book|booking|reserve|reservation)\b",
r"\b(room|suite|deluxe|check in|check out)\b",
r"\b(one|two|three|four|1|2|3|4)\s+(members?|guests?|people|persons)\b",
r"\b(price|rate|tariff|available|availability)\b",
r"\b(today|tomorrow|next week|april|may|june|july|august|september|october|november|december|january|february|march)\b",
r"\b(day|days|night|nights)\b",
)
if any(re.search(pattern, lowered) for pattern in patterns):
return True
if re.search(r"^(what|when|where|who|why|how|can|could|would|should|do|does|did|is|are|tell me|explain)\b", lowered):
return True
if word_count >= 5 and re.search(r"\b(hotel|room|booking|reservation|stay|service|restaurant|menu|airport)\b", lowered):
return True
if session and session.task_state.get("pending_intent"):
return True
return False
def _should_rescue_very_low_confidence(self, text: str, session: Optional[SessionState] = None) -> bool:
lowered = _normalise_text(text)
if not lowered or MultilingualASR._looks_like_asr_artifact(lowered):
return False
extracted = self.llm.intent_extractor.extract(text)
strong_intents = {
"booking", "availability", "pricing", "room_types", "services", "restaurant",
"faq_check_in", "faq_check_out", "faq_contact", "faq_location", "faq_amenities",
}
if extracted.intent in strong_intents and extracted.confidence >= 0.84 and len(lowered.split()) >= 4:
return True
pending_intent = str(session.task_state.get("pending_intent") or "").strip() if session else ""
if pending_intent and len(lowered.split()) >= 2:
return True
return False
def _build_agent_turn_result(
self,
response_text: str,
*,
active_session: SessionState,
voice_preference: Optional[str],
request_id: str,
t_start: float,
transcription_text: str = "",
language: Optional[str] = None,
confidence: float = 0.0,
) -> dict[str, Any]:
result: dict[str, Any] = {
"transcription": transcription_text,
"language": language or active_session.language_preference,
"response": response_text,
"audio_path": None,
"voice_profile": None,
"voice_reference_path": None,
"confidence": confidence,
"rtt_seconds": round(time.time() - t_start, 2),
"error": None,
}
voice_profile, audio_path = self.build_voice_output(
response_text,
preference=voice_preference or active_session.voice_preference,
session=active_session,
request_id=request_id,
)
result["audio_path"] = audio_path
result["voice_profile"] = voice_profile.key
result["voice_reference_path"] = voice_profile.reference_audio_path
return result
def _handle_no_input_turn(
self,
*,
active_session: SessionState,
voice_preference: Optional[str],
language: Optional[str],
request_id: str,
t_start: float,
) -> dict[str, Any]:
current_language = language or active_session.language_preference or "en"
no_input_count = int(active_session.task_state.get("no_input_count", 0)) + 1
active_session.task_state["no_input_count"] = no_input_count
active_session.task_state["unclear_count"] = 0
should_close = bool(
active_session.task_state.get("booking_completed")
or active_session.task_state.get("conversation_should_close")
or no_input_count >= 2
)
if should_close:
reason = "booking_complete" if active_session.task_state.get("booking_completed") else "silence"
response_text = self.llm.generate_goodbye(
language=current_language,
session=active_session,
reason=reason,
)
active_session.task_state["conversation_closed"] = True
active_session.task_state["no_input_count"] = 0
else:
response_text = self.llm.generate_no_input_prompt(language=current_language, session=active_session)
active_session.task_state.pop("conversation_closed", None)
logger.info(" [%s] No-input follow-up: count=%d close=%s", request_id, no_input_count, should_close)
return self._build_agent_turn_result(
response_text,
active_session=active_session,
voice_preference=voice_preference,
request_id=request_id,
t_start=t_start,
transcription_text="",
language=current_language,
confidence=0.0,
)
def _handle_unclear_audio_turn(
self,
*,
active_session: SessionState,
voice_preference: Optional[str],
language: Optional[str],
request_id: str,
t_start: float,
confidence: float = 0.0,
) -> dict[str, Any]:
current_language = language or active_session.language_preference or "en"
unclear_count = int(active_session.task_state.get("unclear_count", 0)) + 1
active_session.task_state["unclear_count"] = unclear_count
active_session.task_state["no_input_count"] = 0
response_text = self.llm.generate_unclear_prompt(language=current_language, session=active_session)
logger.info(" [%s] Unclear-audio follow-up: count=%d", request_id, unclear_count)
return self._build_agent_turn_result(
response_text,
active_session=active_session,
voice_preference=voice_preference,
request_id=request_id,
t_start=t_start,
transcription_text="",
language=current_language,
confidence=confidence,
)
def process_turn(
self,
audio: Optional[np.ndarray] = None,
text_input: Optional[str] = None,
language_hint: Optional[str] = None,
status_callback: Optional[Any] = None,
play_audio: bool = False,
session: Optional[SessionState] = None,
voice_preference: Optional[str] = None,
asr_confidence_threshold: Optional[float] = None,
asr_confidence_rerun_threshold: Optional[float] = None,
llm_max_tokens: Optional[int] = None,
unload_models_between_turns: Optional[bool] = None,
request_id: str = "",
) -> dict[str, Any]:
t_start = time.time()
active_session = session or self.session
effective_conf_threshold = asr_confidence_threshold or self.cfg.asr_confidence_threshold
result: dict[str, Any] = {
"transcription": "",
"language": active_session.language_preference,
"response": "",
"audio_path": None,
"voice_profile": None,
"voice_reference_path": None,
"confidence": 0.0,
"rtt_seconds": 0.0,
"error": None,
}
if self.cfg.english_only_mode:
active_session.language_preference = "en"
log_language = "en" if self.cfg.english_only_mode else (language_hint or active_session.language_preference)
logger.info(
" [%s] Turn: mode=%s lang=%s voice=%s",
request_id,
"text" if text_input else "audio",
log_language,
voice_preference or active_session.voice_preference,
)
try:
if text_input:
transcription = TranscriptionResult(
text=text_input,
language="en" if self.cfg.english_only_mode else (language_hint or active_session.language_preference or "en"),
confidence=1.0,
is_valid=True,
)
else:
if audio is None:
result["error"] = "No audio provided."
return result
rerun_threshold = asr_confidence_rerun_threshold or self.cfg.asr_confidence_rerun_threshold
unload = unload_models_between_turns if unload_models_between_turns is not None else self.cfg.unload_models_between_turns
# Use session language as hint if not English
hint = "en" if self.cfg.english_only_mode else language_hint
if not hint and active_session.language_preference and active_session.language_preference != "en":
hint = active_session.language_preference
transcription = self.asr.transcribe(
audio,
language_hint=hint,
confidence_threshold=effective_conf_threshold,
rerun_threshold=rerun_threshold,
unload_models_between_turns=unload,
request_id=request_id,
)
if not text_input:
repaired_text = self._repair_low_conf_transcription(transcription.text)
if repaired_text and repaired_text != transcription.text:
logger.info(" [%s] ASR repaired: %r -> %r", request_id, transcription.text, repaired_text)
transcription.text = repaired_text
if transcription.text.strip() and MultilingualASR._looks_like_asr_artifact(transcription.text):
logger.info(" [%s] ASR artifact follow-up triggered: %r", request_id, transcription.text[:80])
return self._handle_unclear_audio_turn(
active_session=active_session,
voice_preference=voice_preference,
language=transcription.language or hint or active_session.language_preference,
request_id=request_id,
t_start=t_start,
confidence=transcription.confidence,
)
if not transcription.is_valid:
if not transcription.text:
lowered_error = (transcription.error or "").strip().lower()
if "no speech" in lowered_error or "empty audio" in lowered_error:
return self._handle_no_input_turn(
active_session=active_session,
voice_preference=voice_preference,
language=transcription.language or hint or active_session.language_preference,
request_id=request_id,
t_start=t_start,
)
return self._handle_unclear_audio_turn(
active_session=active_session,
voice_preference=voice_preference,
language=transcription.language or hint or active_session.language_preference,
request_id=request_id,
t_start=t_start,
confidence=transcription.confidence,
)
# Only block if confidence is extremely low (near-silence / pure noise)
if transcription.confidence < 0.05 and not self._should_rescue_very_low_confidence(transcription.text, active_session):
return self._handle_unclear_audio_turn(
active_session=active_session,
voice_preference=voice_preference,
language=transcription.language or hint or active_session.language_preference,
request_id=request_id,
t_start=t_start,
confidence=transcription.confidence,
)
# Otherwise accept the transcription text even if confidence is low
transcription.is_valid = True
result["transcription"] = transcription.text
if self.cfg.english_only_mode:
transcription.language = "en"
result["language"] = transcription.language
result["confidence"] = transcription.confidence
# Generate response
# FIX: Hard-reject extremely low-confidence transcriptions before they
# reach the LLM. Previously conf=0.46 "Boquerum" was passed through,
# causing the 0.5B model to hallucinate "a town in Bourgogne-Franche-Comté".
salvageable_low_conf = (
not text_input
and transcription.confidence < effective_conf_threshold
and self._is_domain_salvageable(transcription.text, active_session)
)
if salvageable_low_conf:
logger.info(
" [%s] ASR low confidence accepted for domain clarification: %r (%.2f)",
request_id,
transcription.text,
transcription.confidence,
)
if not text_input and transcription.confidence < effective_conf_threshold and not salvageable_low_conf:
return self._handle_unclear_audio_turn(
active_session=active_session,
voice_preference=voice_preference,
language=transcription.language or active_session.language_preference,
request_id=request_id,
t_start=t_start,
confidence=transcription.confidence,
)
active_session.task_state["no_input_count"] = 0
active_session.task_state["unclear_count"] = 0
active_session.task_state.pop("conversation_closed", None)
if self.cfg.english_only_mode:
active_session.language_preference = "en"
history = active_session.get_history_for_llm(max_turns=self.cfg.max_history_turns)
response_text = self.llm.generate(
transcription.text,
transcription,
history,
max_tokens=llm_max_tokens,
session=active_session,
request_id=request_id,
)
result["response"] = response_text
voice_profile, audio_path = self.build_voice_output(
response_text,
preference=voice_preference or active_session.voice_preference,
session=active_session,
request_id=request_id,
)
result["audio_path"] = audio_path
result["voice_profile"] = voice_profile.key
result["voice_reference_path"] = voice_profile.reference_audio_path
rtt = time.time() - t_start
result["rtt_seconds"] = round(rtt, 2)
active_session.add_turn(
ConversationTurn(
user_text=transcription.text,
response_text=response_text,
language=transcription.language,
response_time_sec=rtt,
asr_confidence=transcription.confidence,
)
)
# Persist session to DB (non-fatal if it fails)
try:
self.db.save_session_snapshot(
active_session.session_id,
language=transcription.language,
voice_preference=active_session.voice_preference,
turn_count=len(active_session.history),
last_intent=active_session.last_intent,
last_user_text=transcription.text,
state_json=json.dumps(active_session.task_state, ensure_ascii=False, default=str),
status="active",
)
except Exception as exc:
logger.debug(" [%s] Session snapshot skipped: %s", request_id, exc)
logger.info(" [%s] Turn done in %.1fs", request_id, rtt)
except Exception as exc:
logger.error(" [%s] Turn error: %s", request_id, exc, exc_info=True)
result["error"] = f"Internal error: {exc}"
return result
def chat(self, text: str, language: Optional[str] = None) -> str:
result = self.process_turn(text_input=text, language_hint=language)
return result.get("response", result.get("error", "No response"))
def set_voice_preference_for_session(self, preference: str, session: SessionState) -> None:
normalized = preference.strip().lower()
if normalized in ("female", "male"):
session.voice_preference = normalized
session.custom_voice_path = None
elif Path(preference).exists():
session.voice_preference = "custom"
session.custom_voice_path = preference
def set_language_for_session(self, language: str, session: SessionState) -> None:
if self.cfg.english_only_mode:
session.language_preference = "en"
return
normalized = language.strip().lower()
if normalized in SUPPORTED_LANGUAGES:
session.language_preference = normalized
def health_check(self) -> tuple[bool, dict]:
def _check_db() -> bool:
try:
rows = self.db.execute_query("SELECT 1 AS ok")
return bool(rows) and rows[0].get("ok") == 1
except Exception:
return False
def _check_pkg(name: str) -> bool:
import importlib
try:
importlib.import_module(name)
return True
except ImportError:
return False
def _ram_ok() -> bool:
try:
import psutil
return psutil.virtual_memory().available > 1_500_000_000
except Exception:
return True
checks = {
"database": _check_db(),
"faster_whisper": _check_pkg("faster_whisper"),
"llama_cpp": _check_pkg("llama_cpp"),
"transformers": _check_pkg("transformers"),
"silero_vad": _check_pkg("silero_vad"),
"edge_tts": _check_pkg("edge_tts"),
"asr_loaded": self.asr._model is not None,
"llm_loaded": self.llm._llm is not None,
"tts_ready": (self.tts.is_available if self.tts is not None else True),
"memory_ok": _ram_ok(),
"gguf_present": bool(self.cfg.gguf_model_path and Path(self.cfg.gguf_model_path).exists()),
"reference_audio": all(
(self.cfg.reference_audio_dir / name).exists()
for name in ("female_reference.wav", "male_reference.wav")
),
}
return all(checks.values()), checks
def shutdown(self) -> None:
logger.info("Shutting down AARA ...")
try:
self.asr._model = None
except Exception:
pass
try:
self.llm._unload_model()
except Exception:
pass
try:
self.db.close()
except Exception:
pass
gc.collect()
logger.info("✅ AARA shutdown complete")
def __repr__(self) -> str:
return f"AaraVoiceAgent(session={self.session.session_id}, turns={len(self.session.history)})"