# conversation_manager.py # Manages conversation sessions — memory within an active voice call. # # v2 — arabic_clarify removed # get_arabic_clarification_flag() removed from ConversationSession. # All callers (voice_pipeline._build_prompt, llm_server._build_prompt) # updated to not pass arabic_clarification to build_system_prompt. import time import uuid import logging from typing import List, Dict, Any, Optional from language_config import normalize_language from utils import needs70B from language_detector import SmartLanguageDetector from caller_info_extractor import CallerInfoCollector logger = logging.getLogger("conversation_manager") SESSION_TIMEOUT_SECONDS = 30 * 60 MAX_HISTORY_MESSAGES = 20 class ConversationSession: """Single conversation session tracking an active phone call instance.""" def __init__( self, session_id: str, company_name: str, agent_name: str, agent_gender: str = "male", custom_rules: Optional[List[str]] = None, document_context: str = "", caller_collection_fields: Optional[List[Dict[str, Any]]] = None, ): self.session_id = session_id self.company_name = company_name self.agent_name = agent_name self.agent_gender = agent_gender self.custom_rules = custom_rules or [] self.document_context = document_context self.caller_collection_fields: List[Dict[str, Any]] = caller_collection_fields or [] self.messages: List[Dict[str, str]] = [] self.language_detector = SmartLanguageDetector() self.caller_info = CallerInfoCollector(fields_spec=self.caller_collection_fields) self.created_at = time.time() self.last_active = time.time() self.turn_count = 0 self.greeting_streak = 0 # Database mapping properties self.company_id = None self.company_name_ur = None self.overview = None self.overview_ur = None self.agent_name_ur = None self.agent_role = None self.agent_role_ur = None self.phone_line_id = None self.line_token = None self.agent_config_id = None self.rag_enabled = False @classmethod def from_tenant_config(cls, config: Dict[str, Any], session_id: str) -> "ConversationSession": """Creates a ConversationSession from a database configuration dictionary.""" rules_raw = config.get("system_instruction_rules") or "" custom_rules = [] if isinstance(rules_raw, list): custom_rules = rules_raw elif isinstance(rules_raw, str): custom_rules = [r.strip() for r in rules_raw.split(".") if r.strip()] session = cls( session_id=session_id, company_name=config.get("company_name") or "Our Company", agent_name=config.get("agent_name") or "Sara", agent_gender=config.get("agent_gender") or "male", custom_rules=custom_rules, document_context="", caller_collection_fields=config.get("caller_details_to_collect") or [] ) session.company_id = config.get("company_id") session.company_name_ur = config.get("company_name_ur") session.overview = config.get("overview") session.overview_ur = config.get("overview_ur") session.agent_name_ur = config.get("agent_name_ur") session.agent_role = config.get("agent_role") session.agent_role_ur = config.get("agent_role_ur") session.phone_line_id = config.get("phone_line_id") session.line_token = config.get("line_token") session.agent_config_id = config.get("agent_config_id") session.rag_enabled = config.get("rag_enabled", False) return session def add_message(self, role: str, content: str) -> None: self.messages.append({"role": role, "content": content}) self.last_active = time.time() if len(self.messages) > MAX_HISTORY_MESSAGES: self.messages = self.messages[-MAX_HISTORY_MESSAGES:] def process_user_message(self, text: str) -> str: """ Process user message — update language detector and caller info. Returns the resolved language code string. """ self.turn_count += 1 self.last_active = time.time() language = normalize_language(self.language_detector.process_message(text)) self.caller_info.process_message(text, language) return language def classify_intent(self, text: str, lang_code: str) -> str: quick_exact = [ "yes", "no", "hello", "hi", "hey", "hold on", "wait", "stop", "ok", "okay", "thanks", "thank you", "جی", "ہاں", "نا", "شکریہ", "نہیں" ] normalized = (text or "").strip().lower() effective_lang = normalize_language(lang_code) if normalized in quick_exact or len(normalized) <= 6: return "fast" return "heavy" if needs70B(text, effective_lang) else "fast" def get_current_language(self) -> str: return normalize_language(self.language_detector.get_language()) # REMOVED: get_arabic_clarification_flag() # arabic_clarify tracking has been removed entirely (Urdu-only mode). # prompt_builder.py no longer accepts arabic_clarification parameter. # Re-enable alongside multi-language mode if needed. def is_expired(self) -> bool: return (time.time() - self.last_active) > SESSION_TIMEOUT_SECONDS def get_history_for_api(self) -> List[Dict[str, str]]: return self.messages.copy() def to_summary(self) -> Dict[str, Any]: return { "session_id": self.session_id, "company_id": self.company_id, "company": self.company_name, "agent": self.agent_name, "agent_gender": self.agent_gender, "language": self.get_current_language(), "turns": self.turn_count, "caller_info": self.caller_info.to_dict(), "duration_seconds": int(time.time() - self.created_at), } class ConversationManager: """Manages active voice-channel conversation tracks.""" def __init__(self): self.sessions: Dict[str, ConversationSession] = {} logger.info("ConversationManager tracking pool initialized successfully.") def create_session( self, company_name: str, agent_name: str, agent_gender: str = "male", custom_rules: List[str] = None, document_context: str = "", session_id: Optional[str] = None, caller_collection_fields: Optional[List[Dict[str, Any]]] = None, ) -> ConversationSession: sid = session_id or str(uuid.uuid4()) session = ConversationSession( session_id=sid, company_name=company_name, agent_name=agent_name, agent_gender=agent_gender, custom_rules=custom_rules, document_context=document_context, caller_collection_fields=caller_collection_fields, ) self.sessions[sid] = session logger.info(f"Session established: {sid} [{company_name} / Agent: {agent_name}]") return session def get_session(self, session_id: str) -> Optional[ConversationSession]: session = self.sessions.get(session_id) if session: if session.is_expired(): logger.info(f"Evicting expired communication stack: {session_id}") del self.sessions[session_id] return None return session return None def end_session(self, session_id: str) -> Optional[Dict[str, Any]]: session = self.sessions.pop(session_id, None) if session: summary = session.to_summary() logger.info(f"Session terminated via clean closure: {session_id} — Summary metrics: {summary}") return summary return None def cleanup_expired(self) -> None: expired = [ sid for sid, session in self.sessions.items() if session.is_expired() ] for sid in expired: logger.info(f"Garbage collection cycle purged expired session: {sid}") del self.sessions[sid] if expired: logger.info(f"Garbage collection cycle purged {len(expired)} expired session allocations.") def active_count(self) -> int: return len(self.sessions)