| |
| |
| |
| |
| |
| |
| |
|
|
| 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 |
|
|
| |
| 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()) |
|
|
| |
| |
| |
| |
|
|
| 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) |
