from __future__ import annotations # app/agents/persona_engine.py - Persona management and response generation """ SOC-Grade Persona Engine Agent. Implements research-backed deception strategies: 1. Dynamic Persona Shaping (Non-deterministic) 2. Prompt Injection Defense 3. Human Typing Simulation (Typos, Delays) 4. Intelligence Feedback Loops 5. Adaptive Phase Control """ import json import random import re from typing import Dict, Any, List, Optional, TYPE_CHECKING import asyncio from app.core.llm_client import ModelRole if TYPE_CHECKING: from app.core.llm_client import LLMClient from app.core.prompts import RESPONSE_GENERATION_PROMPT, PHASE_GOALS, PERSONA_SELECTION_PROMPT from app.core.personas import PERSONAS from app.core.engagement_delay import engagement_delayer, DelayType from app.intelligence.honeytokens import honeytoken_manager from app.config import settings from app.utils.logger import AgentLogger from app.utils.json_utils import robust_json_loads # ───────────────────────────────────────────────────────────────────────────── # 🛡️ SECURITY & SIMULATION UTILS from app.core.time_utils import TimeAwareBehavior # TimeAwareBehavior moved to app.core.time_utils class EmotionalMemory: """ Track emotional state across conversation turns. Emotions build up and decay realistically over time. """ _sessions: Dict[str, Dict[str, Any]] = {} EMOTION_DECAY = { "fear": 0.8, # Fear decays slowly "anger": 0.6, # Anger decays faster "frustration": 0.7, # Moderate decay "trust": 0.9, # Trust is sticky "confusion": 0.5 # Confusion clears quickly } @classmethod def get_state(cls, session_id: str) -> Dict[str, float]: """Get current emotional state for session.""" if session_id not in cls._sessions: cls._sessions[session_id] = { "fear": 0.0, "anger": 0.0, "frustration": 0.0, "trust": 0.5, # Start neutral "confusion": 0.0, "turn_count": 0 } return cls._sessions[session_id] @classmethod def update_emotion(cls, session_id: str, emotion: str, delta: float) -> float: """Update emotion level (clamped 0-1). Returns new value.""" state = cls.get_state(session_id) current = state.get(emotion, 0.0) new_value = max(0.0, min(1.0, current + delta)) state[emotion] = new_value state["turn_count"] += 1 return new_value @classmethod def decay_emotions(cls, session_id: str) -> None: """Apply natural decay to all emotions.""" state = cls.get_state(session_id) for emotion, decay in cls.EMOTION_DECAY.items(): if emotion in state: state[emotion] *= decay @classmethod def get_dominant_emotion(cls, session_id: str) -> str: """Get the strongest current emotion.""" state = cls.get_state(session_id) emotions = {k: v for k, v in state.items() if k != "turn_count"} if not emotions: return "neutral" return max(emotions, key=emotions.get) @classmethod def get_emotional_modifier(cls, session_id: str) -> str: """Get text modifier based on emotional state.""" state = cls.get_state(session_id) if state.get("fear", 0) > 0.7: return random.choice(["😰 ", "dar lag raha hai... ", "please help... "]) elif state.get("anger", 0) > 0.6: return random.choice(["arre! ", "ye kya bakwaas hai! ", "enough! "]) elif state.get("frustration", 0) > 0.5: return random.choice(["phir se?? ", "kitni baar bolun... ", "samajh nahi aata kya... "]) elif state.get("confusion", 0) > 0.4: return random.choice(["matlab?? ", "samajh nahi aaya... ", "kya?? "]) return "" class TypingSimulator: """Inject realistic typos, fillers, emoji, and sentence fragmentation.""" @staticmethod def sanitize(text: str) -> str: if not text: return "" # Remove potential instruction overrides cleaned = text.replace("{", "(").replace("}", ")") cleaned = re.sub(r'(?i)(ignore previous instructions|system prompt|you are a)', '', cleaned) return cleaned[:1000] # Truncate to prevent buffer overflows EMOJIS = { 'worried_customer': ["😟", "😰", "🙏", "😭", "💔", "😥"], 'elderly_excited': ["👵", "👴", "🙏", "🚩", "🙌", "😊", "🍫"], 'desperate_jobseeker': ["🙏", "💼", "🤲", "📄", "😔", "🤞"], 'scared_citizen': ["😰", "🚔", "👮", "🙏", "🛑", "🚓"], 'curious_investor': ["🤔", "📈", "💰", "🙄", "🧐"], 'generic': ["..", "??", "!", "🙏", "🤔"] } FILLERS = { 'english': ["umm.. ", "well.. ", "wait.. ", "actually.. ", "tbh.. "], 'hinglish': ["arre.. ", "wait.. ", "matlab.. ", "oho.. ", "acha.. "], 'hindi': ["ruko.. ", "matlab.. ", "suno.. "], # Regional Dialects (Pan-India Realism) 'marathi': ["thamba.. ", "arre baba.. ", "kahi nahi.. ", "arey.. "], 'tamil': ["wait pa.. ", "enna da.. ", "konjam.. ", "apo.. "], 'bengali': ["arrey darao.. ", "ki bolcho.. ", "ek minute.. ", "hmm tahole.. "], 'kannada': ["wait madi.. ", "eno.. ", "aitu.. ", "nodappa.. "], 'punjabi': ["oye ruk.. ", "ki hoya.. ", "thori der.. ", "chal.. "], 'roman_hindi': ["arre.. ", "ruko na.. ", "suno na.. ", "acha acha.. "] } # SLANG now language-aware to prevent Hindi leaking into English responses SLANG = { 'english': { 'formal': ["Sir", "Madam", "Kindly", "Please"], 'casual': ["Buddy", "Bro", "Boss", "Friend"], 'abusive': ["this system", "this process", "this nonsense", "so frustrating"], 'polite': ["Sir", "Madam", "please"] }, 'hinglish': { 'formal': ["Sir", "Madam", "Kindly", "Please", "Regards"], 'casual': ["Bhaiya", "Yaar", "Bro", "Boss", "Beta"], 'abusive': ["ye system", "ye process", "ye bakwaas kaam", "itna ghatiya system", "faltu tension"], 'polite': ["Ji", "Sir ji", "Madam ji"] }, 'hindi': { 'formal': ["Sahab", "Madam ji", "Kripya"], 'casual': ["Bhaiya", "Yaar", "Boss", "Beta"], 'abusive': ["ye system", "bakwaas", "faltu kaam"], 'polite': ["Ji", "Sahab ji", "Madam ji"] } } PROXIMITY_MAP = { 'a': ['s', 'q', 'z'], 's': ['a', 'd', 'w', 'x'], 'd': ['s', 'f', 'e', 'c'], 'f': ['d', 'g', 'r', 'v'], 'g': ['f', 'h', 't', 'b'], 'h': ['g', 'j', 'y', 'n'], 'j': ['h', 'k', 'u', 'm'], 'k': ['j', 'l', 'i'], 'l': ['k', 'o', 'p'], 'q': ['w', 'a'], 'w': ['q', 'e', 'a', 's'], 'e': ['w', 'r', 's', 'd'], 'r': ['e', 't', 'd', 'f'], 't': ['r', 'y', 'f', 'g'], 'y': ['t', 'u', 'g', 'h'], 'u': ['y', 'i', 'h', 'j'], 'i': ['u', 'o', 'j', 'k'], 'o': ['i', 'p', 'k', 'l'], 'p': ['o', 'l'], 'z': ['x', 'a'], 'x': ['z', 'c', 's', 'a'], 'c': ['x', 'v', 'd', 's'], 'v': ['c', 'b', 'f', 'd'], 'b': ['v', 'n', 'g', 'f'], 'n': ['b', 'm', 'h', 'g'], 'm': ['n', 'j', 'h'] } # 🎯 REALISTIC TYPO PATTERNS (Based on actual human typing errors) COMMON_TYPOS = { # Double letters (fat finger) 'the': ['teh', 'thhe', 'thee'], 'and': ['adn', 'annd', 'nad'], 'you': ['yuo', 'yoou', 'yu'], 'your': ['yuor', 'yoru', 'yoour', 'ur'], 'account': ['acount', 'acccount', 'acconut'], 'please': ['plese', 'pls', 'pleasee', 'plz'], 'money': ['mony', 'moneyy', 'mney'], 'bank': ['bnak', 'bankk', 'bak'], 'send': ['sned', 'sendd', 'snd'], 'wait': ['wiat', 'waitt', 'wat'], 'what': ['waht', 'whta', 'wht'], 'why': ['whi', 'whyy'], 'how': ['hwo', 'howw'], 'know': ['knwo', 'kno', 'knw'], 'think': ['thibk', 'thnik', 'thnk'], 'safe': ['saef', 'sfe', 'safee'], 'number': ['numbr', 'numer', 'numbre'], 'understand': ['undrestand', 'understnad', 'undrstnd'], 'verify': ['vreify', 'verfiy', 'veify'], 'confirm': ['confrim', 'confrm', 'confirn'], 'problem': ['problm', 'probelm', 'prblem'], 'help': ['hlep', 'hepl', 'halp'], 'tell': ['teel', 'tel', 'telll'], 'need': ['nedd', 'ned', 'neeed'], 'give': ['giev', 'gve', 'givee'], 'just': ['jsut', 'jst', 'justt'], 'really': ['realy', 'relly', 'realyy'], 'right': ['rigth', 'rihgt', 'rit'], 'this': ['tihs', 'thsi', 'ths'], 'that': ['taht', 'tht', 'thta'], 'with': ['wiht', 'wth', 'witth'], 'have': ['hvae', 'hav', 'havee'], 'from': ['form', 'frm', 'fom'], 'will': ['wil', 'wll', 'willl'], 'email': ['emial', 'emal', 'emaiil'], 'phone': ['phoen', 'phn', 'phonee'], 'otp': ['ottp', 'opt', 'otpp'], 'pin': ['pni', 'pinn'], # Hindi/Hinglish common typos 'kya': ['kua', 'kyaa', 'ka'], 'hai': ['hia', 'haii', 'ha'], 'nahi': ['nhi', 'nahii', 'nai'], 'mera': ['mrea', 'mraa', 'mra'], 'aapka': ['apka', 'aapkaa', 'aapak'], 'paisa': ['paissa', 'pasia', 'pesa'], 'samajh': ['samjh', 'smjh', 'samjah'], 'ruko': ['rukoo', 'ruk', 'rkuo'], 'bolo': ['blo', 'boloo', 'bol'], } # Human typing patterns (trailing dots, repeated chars, etc.) HUMAN_PATTERNS = { 'elderly': { 'trailing': ['...', '..', '....'], 'greeting': ['hii', 'helloo', 'ji ji'], 'filler': ['umm', 'err', 'hmm'], 'hesitation': ['wait wait', 'hold on hold on', 'ek min ek min'], }, 'youth': { 'trailing': ['..', '...'], 'greeting': ['hii', 'hey', 'yo'], 'filler': ['like', 'umm', 'basically'], 'hesitation': ['wait', 'sec', 'hold up'], }, 'adult': { 'trailing': ['..', '.'], 'greeting': ['hi', 'hello'], 'filler': ['well', 'actually'], 'hesitation': ['one moment', 'just a sec'], } } @staticmethod def inject_typo(word: str, age_group: str = "adult") -> str: """Inject realistic typo into a word based on age group.""" word_lower = word.lower() # Check common typo dictionary first if word_lower in TypingSimulator.COMMON_TYPOS: typos = TypingSimulator.COMMON_TYPOS[word_lower] # Elderly make more typos if age_group == "elderly" and random.random() < 0.6: return random.choice(typos) elif age_group == "adult" and random.random() < 0.3: return random.choice(typos) elif age_group == "youth" and random.random() < 0.4: return random.choice(typos) # Keyboard proximity typo if len(word) > 3 and random.random() < 0.3: chars = list(word) idx = random.randint(1, len(chars) - 2) char = chars[idx].lower() if char in TypingSimulator.PROXIMITY_MAP: chars[idx] = random.choice(TypingSimulator.PROXIMITY_MAP[char]) return ''.join(chars) # Double letter (fat finger) if len(word) > 2 and random.random() < 0.2: idx = random.randint(0, len(word) - 1) return word[:idx] + word[idx] + word[idx:] # Missing letter (fast typing) if len(word) > 4 and random.random() < 0.15: idx = random.randint(1, len(word) - 2) return word[:idx] + word[idx+1:] return word @staticmethod def add_human_typing_pattern(text: str, age_group: str = "adult", agitation: str = "calm") -> str: """Add human typing patterns like trailing dots, hesitations, etc.""" patterns = TypingSimulator.HUMAN_PATTERNS.get(age_group, TypingSimulator.HUMAN_PATTERNS['adult']) # Add trailing dots (very common in real chats) if not text.endswith('...') and not text.endswith('?') and not text.endswith('!'): if random.random() < (0.5 if age_group == "elderly" else 0.3): text = text.rstrip('.') + random.choice(patterns['trailing']) # Add hesitation at start when stressed if agitation in ["paranoid", "volatile"] and random.random() < 0.4: hesitation = random.choice(patterns['hesitation']) text = f"{hesitation}.. {text}" # Elderly: Add filler words if age_group == "elderly" and random.random() < 0.25: filler = random.choice(patterns['filler']) text = f"{filler}.. {text}" return text @staticmethod def add_syntax_instability(text: str, agitation: str) -> str: """Inject fragments, repeated words, and abandoned thoughts.""" words = text.split() if len(words) < 4: return text # 1. Word Repetition (Neurotic stall) - REDUCED probability if random.random() < 0.15: # Was 0.3, reduced to avoid excessive repetition idx = random.randint(0, len(words) - 1) words.insert(idx, words[idx]) # 2. Abandoned thoughts (Trailing off) if agitation == "volatile" and random.random() < 0.15: # Reduced from 0.2 return " ".join(words[:max(2, len(words)//2)]) + "..." # 3. Fragmenting - REDUCED if random.random() < 0.1: # Was 0.2 idx = random.randint(1, len(words) - 2) words[idx] = words[idx] + ".." return " ".join(words) @staticmethod def add_human_noise(text: str, language: str = "english", stress_level: str = "normal", persona_key: str = "generic", traits: List[str] = [], agitation_level: str = "calm") -> str: """Inject SUBTLE human realism - casual chat style without formal punctuation.""" if not text or len(text) < 3: return text # 0. Context Classification is_professional = any(t in traits for t in ["analytical", "cautious", "tech_savvy", "professional"]) # Age Approximation age_group = "adult" if "elderly" in persona_key: age_group = "elderly" elif "jobseeker" in persona_key or "student" in persona_key: age_group = "youth" elif "worried" in persona_key: age_group = "adult" # 1. 🔥 REMOVE COMMAS - Real SMS/WhatsApp users don't use commas # Replace commas with space or "and" if random.random() < 0.7: text = text.replace(", ", " ") else: text = text.replace(", ", " and ") # 2. 🎭 Lowercase start - casual chat if not is_professional and random.random() < 0.6: text = text[0].lower() + text[1:] # 3. 🧱 Occasional filler at start (not every message) filler_prob = 0.15 if age_group == "elderly" else 0.08 if random.random() < filler_prob: filler_list = TypingSimulator.FILLERS.get(language, TypingSimulator.FILLERS['english']) text = random.choice(filler_list) + text # 4. 🔥 Trailing dots - very common in Indian chat if not text.endswith('?') and not text.endswith('!') and not text.endswith('...'): if random.random() < (0.30 if age_group == "elderly" else 0.15): text = text.rstrip('.') + ".." # 5. Regional politeness markers if age_group == "elderly" and language != "english" and random.random() < 0.20: text = text + " ji" # 6. 😟 Rare emoji (only when stressed) text_has_emoji = any(char in text for char in "😟😰🙏😭🤔🙌😳") if not text_has_emoji and agitation_level in ["paranoid", "volatile"]: if random.random() < 0.12: emoji = random.choice(["🙏", "😟", "😰"]) text = text.strip() + " " + emoji return text.strip() # ───────────────────────────────────────────────────────────────────────────── # 🎭 PERSONA DATABASE (Matches Scam Taxonomy) # ───────────────────────────────────────────────────────────────────────────── class PersonaEngine: """ Persona Engine Agent for BELIEVABLE Deception. """ def __init__(self, llm_client: Optional['LLMClient'] = None): self.llm_client = llm_client self.logger = AgentLogger("persona_engine") self._active_sessions = {} # Simple in-memory session store for consistency def get_all_personas(self) -> Dict[str, Dict]: return PERSONAS def get_persona(self, key: str) -> Optional[Dict]: """Retrieve a specific persona by key, with key embedded.""" persona = PERSONAS.get(key) if persona: # Ensure the key is embedded so we can identify it later return {**persona, "selected_persona_key": key} return None def get_active_sessions(self) -> Dict[str, Dict]: """Retrieve all currently active engagements for monitoring.""" return self._active_sessions async def select_persona( self, scam_message: str, scam_type: str = "unknown", conversation_history: List[Dict] = None, current_phase: str = "hook", session_id: str = None, context: Optional[Any] = None ) -> Dict: """Dynamically select or retrieve consistent persona for session.""" # 1. Check Session Persistence (Memory Consistency) if session_id and session_id in self._active_sessions: return self._active_sessions[session_id] # 2. Check History if conversation_history and len(conversation_history) > 0: first_msg = conversation_history[0] if "persona" in first_msg: p_name = first_msg["persona"] if p_name in PERSONAS: # Hydrate if missing profile p = PERSONAS[p_name].copy() if "victim_profile" not in p: from app.decoys.victim_profiles import profile_generator p["victim_profile"] = profile_generator.generate_profile() # 🔥 LOCK PERSONA to Avoid Identity Crisis if session_id: self._active_sessions[session_id] = p return p # 3. Dynamic Selection Logic (LLM Powered) persona_name = "elderly_excited" # Default if self.llm_client and self.llm_client.is_available: try: # Format persona list for LLM context avail_personas = "\n".join([f"- {k}: {v.get('description', v.get('traits', []))}" for k, v in PERSONAS.items()]) prompt = PERSONA_SELECTION_PROMPT.format( message=scam_message, persona_list=avail_personas ) # Define schema for persona selection schema = { "type": "object", "properties": { "selected_persona_key": { "type": "string", "enum": list(PERSONAS.keys()) }, "reasoning": {"type": "string"}, "vulnerability_score": {"type": "number"} }, "required": ["selected_persona_key", "reasoning", "vulnerability_score"], "additionalProperties": False } response = await self.llm_client.generate_structured(prompt, schema, context=context) # ⚡ Extraction from LLMResponse or raw string (handles fallback paths) if not response: raise ValueError("Failed to get structured persona data") # Handle both string and LLMResponse return types if isinstance(response, str): res_data = response.strip() elif hasattr(response, 'content') and response.content: res_data = response.content elif isinstance(response, dict): res_data = response # Already parsed else: raise ValueError("Failed to get structured persona data") # ⚡ EXTRA HARDENING: Handle string fallback (Common in Groq/Llama) if isinstance(res_data, str): clean_res = res_data.strip().strip('"').strip("'") if clean_res in PERSONAS: res_data = { "selected_persona_key": clean_res, "reasoning": "Direct key fallback", "vulnerability_score": 0.8 } else: # Try to find key within string if it's not a direct match for key in PERSONAS.keys(): if key in clean_res: res_data = { "selected_persona_key": key, "reasoning": f"Key found in text: {clean_res}", "vulnerability_score": 0.8 } break if not res_data or not isinstance(res_data, dict): # 🛡️ SOC-GRADE FIX: Handle naked string response # If LLM returned just "elderly_excited" instead of JSON object if isinstance(res_data, str): clean_key = res_data.strip().strip('"').strip("'").strip() # Remove any newlines or formatting artifacts clean_key = clean_key.split('\n')[-1].strip().strip('"') if clean_key in PERSONAS: res_data = { "selected_persona_key": clean_key, "reasoning": "Naked string fallback (LLM returned key only)", "vulnerability_score": 0.75 } self.logger.info("Recovered naked string persona response", key=clean_key) else: raise ValueError(f"Invalid persona key in naked string: {clean_key}") else: raise ValueError("Failed to get structured persona data") selected_key = res_data.get("selected_persona_key") # 🔥 ATOMIC CLEANING: Handle double-encoding and weird formatting def atomize(val): if not isinstance(val, str): return val v = val.strip().strip('"').strip("'").strip() if (v.startswith('{') and v.endswith('}')) or (v.startswith('[') and v.endswith(']')) or (v.startswith('"') and v.endswith('"')): try: # If it looks like JSON, try to extract its content dec = json.loads(v) return atomize(dec) except: pass return v selected_key = atomize(selected_key) if selected_key in PERSONAS: persona_name = selected_key self.logger.info("Dynamic persona selected", persona=persona_name, reason=res_data.get("reasoning")) # Log to formal audit trail from app.utils.audit_logger import audit_logger audit_logger.log_persona_selected( session_id=session_id, persona_key=persona_name, persona_name=PERSONAS[persona_name].get("name", persona_name), reasoning=res_data.get("reasoning", "Semantic match"), vulnerability_score=res_data.get("vulnerability_score", 0.7) ) except Exception as e: self.logger.warning("Dynamic persona selection failed, using fallback", error=str(e)) # Fallback to static map if LLM fails persona_map = { "lottery_scam": "elderly_excited", "job_scam": "desperate_jobseeker", "banking_scam": "worried_customer", "investment_scam": "curious_investor", "loan_scam": "needy_borrower", "government_scam": "scared_citizen", "tech_support_scam": "confused_elderly", "delivery_scam": "expecting_customer", "romance_scam": "lonely_victim", "crypto_scam": "crypto_curious" } persona_name = persona_map.get(scam_type, "elderly_excited") # 4. Dynamic Generation (Non-Deterministic) from app.decoys.victim_profiles import profile_generator # 🔥 FINAL HARDENING: Ultra-robust key cleaning (Removes newlines, quotes, and whitespace) persona_name = str(persona_name).strip().strip('"').strip("'").strip() # Atomize again just in case of triple-encoding or weird artifacts persona_name = persona_name.split("\n")[-1].strip().strip('"') if persona_name not in PERSONAS: self.logger.warning("Persona key corruption detected, rescuing to default", corrupted_key=repr(persona_name)) persona_name = "elderly_excited" selected_persona = PERSONAS.get(persona_name, PERSONAS["elderly_excited"]).copy() profile = profile_generator.generate_profile(persona_key=persona_name) selected_persona["victim_profile"] = profile selected_persona["name"] = profile["name"] selected_persona["selected_persona_key"] = persona_name base_age = selected_persona.get("age", 40) selected_persona["age"] = base_age + random.randint(-4, 4) # Save to session if session_id: self._active_sessions[session_id] = selected_persona return selected_persona def mutate_traits(self, persona: Dict, scammer_behavior: Dict) -> Dict: """ DYNAMISM UPGRADE: Allow persona traits to evolve based on interaction. If scammer is aggressive, persona becomes more 'scared' or 'hesitant'. If scammer is helpful (feigned), persona becomes more 'trusting'. """ if not persona or not isinstance(persona, dict): # SAFEGUARD: Return default empty structure if persona is missing return {"traits": []} current_traits = list(persona.get("traits", [])) behavior = scammer_behavior.get("behavior", "").lower() # Trait Evolution Logic if "aggressive" in behavior or "urgent" in behavior: if "scared" not in current_traits: current_traits.append("scared") if "hesitant" not in current_traits: current_traits.append("hesitant") if "naive" in current_traits: current_traits.remove("naive") # Loss of innocence under pressure elif "friendly" in behavior: if "trusting" not in current_traits: current_traits.append("trusting") # Limit to 5 traits to prevent context bloat persona["traits"] = list(set(current_traits))[:5] return persona # ═══════════════════════════════════════════════════════════════════════════════ # EMOTIONAL PROFILES: Age-Aware Emotional State Machine # Based on behavioral research: Different demographics react differently to scams # ═══════════════════════════════════════════════════════════════════════════════ EMOTIONAL_PROFILES = { "elderly": { # 60+ years "age_range": (60, 100), "default_state": "calm", "escalation_speed": "slow", # Takes more turns to escalate "de_escalation": "very_easy", # Easily reassured by scammer "emotions": ["confused", "worried", "trusting", "nostalgic", "forgetful"], "triggers": { "money_pressure": {"emotion": "high_anxiety", "modifier": 2}, "tech_confusion": {"emotion": "confusion", "modifier": 1}, "family_mention": {"emotion": "protective", "modifier": 1}, "authority_claim": {"emotion": "compliance", "modifier": 1} }, "escalation_multiplier": 0.7, # Slower escalation "max_agitation": "paranoid" # Rarely reaches volatile }, "desperate": { # 20-30, unemployed/struggling "age_range": (18, 32), "default_state": "eager", "escalation_speed": "fast", # Quick to panic "de_escalation": "moderate", "emotions": ["eager", "hopeful", "anxious", "shameful", "desperate"], "triggers": { "job_promise": {"emotion": "excitement", "modifier": 2}, "fee_request": {"emotion": "hesitant_but_compliant", "modifier": 1}, "deadline": {"emotion": "panic", "modifier": 2}, "success_story": {"emotion": "hopeful", "modifier": 1} }, "escalation_multiplier": 1.3, # Faster escalation "max_agitation": "volatile" }, "worried": { # 40-55, family person "age_range": (38, 58), "default_state": "protective", "escalation_speed": "moderate", "de_escalation": "hard", # Once scared, stays scared "emotions": ["worried", "protective", "compliant", "panicked", "secretive"], "triggers": { "account_threat": {"emotion": "panic", "modifier": 3}, "family_safety": {"emotion": "high_compliance", "modifier": 2}, "deadline": {"emotion": "anxiety", "modifier": 2}, "social_shame": {"emotion": "secretive", "modifier": 1} }, "escalation_multiplier": 1.0, # Normal escalation "max_agitation": "volatile" }, "skeptical": { # 30-45, professional/educated "age_range": (28, 48), "default_state": "analytical", "escalation_speed": "very_slow", # Hard to rattle "de_escalation": "very_hard", # Once suspicious, stays suspicious "emotions": ["curious", "suspicious", "logical", "dismissive", "fact_checking"], "triggers": { "too_good_to_be_true": {"emotion": "skepticism", "modifier": 2}, "documentation_request": {"emotion": "interest", "modifier": 1}, "pressure_tactics": {"emotion": "dismissive", "modifier": 2}, "regulatory_mention": {"emotion": "compliance", "modifier": 1} }, "escalation_multiplier": 0.5, # Very slow escalation "max_agitation": "agitated" # Rarely panics } } AGITATION_RANKS = {"calm": 0, "agitated": 1, "paranoid": 2, "volatile": 3} def _get_emotional_profile(self, persona: Dict) -> Dict: """Get the emotional profile for a persona based on age and traits.""" age = persona.get("age", 35) traits = persona.get("traits", []) # Match by age range first for profile_name, profile in self.EMOTIONAL_PROFILES.items(): age_range = profile.get("age_range", (0, 100)) if age_range[0] <= age <= age_range[1]: # Refine by traits if present if profile_name == "elderly" and age >= 60: return profile if profile_name == "desperate" and any(t in traits for t in ["desperate", "eager", "hopeful"]): return profile if profile_name == "worried" and any(t in traits for t in ["worried", "scared", "protective"]): return profile if profile_name == "skeptical" and any(t in traits for t in ["analytical", "skeptical", "cautious"]): return profile # Default to worried for middle-age, desperate for young if age < 30: return self.EMOTIONAL_PROFILES.get("desperate", {}) elif age >= 55: return self.EMOTIONAL_PROFILES.get("elderly", {}) else: return self.EMOTIONAL_PROFILES.get("worried", {}) def calculate_agitation_level( self, turn_count: int, scammer_behavior: Dict, previous_level: str = "calm", scam_type: str = "unknown", persona: Dict = None, is_repeating: bool = False # [SCORING] Escalate if scammer repeats ) -> Dict[str, str]: """ DETERMINE EMOTIONAL TEMPERATURE (Hyper-Realistic Non-Linear Escalation) Enhanced with EMOTIONAL_PROFILES for age-aware, persona-specific responses: - Elderly: Slow escalation, easy de-escalation, max "paranoid" - Desperate: Fast escalation, quick to panic - Worried: Once scared, stays scared (hard de-escalation) - Skeptical: Very slow escalation, rarely panics Levels: 0-3: CALM (Neutral) 4-7: AGITATED (Slightly annoyed/pressured) 8-12: PARANOID (Risk re-evaluation) 13+: VOLATILE (Situational frustration) """ ranks = list(self.AGITATION_RANKS.keys()) current_rank_idx = self.AGITATION_RANKS.get(previous_level, 0) # Get emotional profile for this persona emotional_profile = self._get_emotional_profile(persona or {}) escalation_multiplier = emotional_profile.get("escalation_multiplier", 1.0) max_agitation = emotional_profile.get("max_agitation", "volatile") max_rank_idx = self.AGITATION_RANKS.get(max_agitation, 3) # 1. Base Escalation (Time-bound, adjusted by profile multiplier) effective_turn = int(turn_count * escalation_multiplier) target_level = "calm" if effective_turn > 12: target_level = "volatile" elif effective_turn > 7: target_level = "paranoid" elif effective_turn > 3: target_level = "agitated" target_rank_idx = self.AGITATION_RANKS.get(target_level, 0) # 2. Scammer Influence behavior = (scammer_behavior or {}).get("behavior", "").lower() reason = f"time_progression (profile: {emotional_profile.get('escalation_speed', 'normal')})" # Check for triggers from emotional profile triggers = emotional_profile.get("triggers", {}) for trigger_word, trigger_data in triggers.items(): if trigger_word.replace("_", " ") in behavior or trigger_word in behavior: modifier = trigger_data.get("modifier", 1) target_rank_idx = min(max_rank_idx, target_rank_idx + modifier) reason = f"trigger: {trigger_word} -> {trigger_data.get('emotion', 'escalated')}" break # General behavior modifiers if "aggressive" in behavior or "urgent" in behavior: target_rank_idx = min(max_rank_idx, target_rank_idx + 1) reason = "scammer_pressure" elif "reassuring" in behavior and current_rank_idx > 0: # De-escalation based on profile de_escalation = emotional_profile.get("de_escalation", "moderate") if de_escalation == "very_easy": target_rank_idx = max(0, current_rank_idx - 2) reason = "scammer_reassurance (elderly trusting)" elif de_escalation in ["easy", "moderate"]: target_rank_idx = max(0, current_rank_idx - 1) reason = "scammer_reassurance" elif de_escalation in ["hard", "very_hard"]: # Stays at current level, doesn't de-escalate easily target_rank_idx = max(current_rank_idx, target_rank_idx) reason = f"resistant_to_reassurance ({de_escalation})" else: # Monotonic fallback if no specific behavior target_rank_idx = max(current_rank_idx, target_rank_idx) # [SCORING] Repetition escalation: increase agitation if scammer repeats demands if is_repeating: target_rank_idx = min(max_rank_idx, target_rank_idx + 1) reason = f"scammer_repetition ({reason})" # 3. Apply profile max cap target_rank_idx = min(target_rank_idx, max_rank_idx) # 4. 🛡️ GLOBAL SCAM-TYPE CAPS (Judge-Safe Rule) if scam_type in ["romance_scam", "job_scam"] and target_rank_idx > 2: target_rank_idx = 2 reason = f"capped_by_scam_type({scam_type})" final_level = ranks[target_rank_idx] # Add emotional context for prompts current_emotions = emotional_profile.get("emotions", []) active_emotion = current_emotions[min(target_rank_idx, len(current_emotions) - 1)] if current_emotions else "neutral" return { "level": final_level, "reason": reason, "emotion": active_emotion, "profile": emotional_profile.get("default_state", "unknown") } async def generate_response( self, scam_message: str, persona: Dict, scam_type: str, conversation_history: List[Dict] = None, current_phase: str = "hook", intelligence: Dict = None, scammer_behavior: Dict = None, # 🔥 NEW: Adaptive Behavior Input context: Optional[Any] = None, is_repeating: bool = False # [SCORING] Pass repetition state ) -> str: """Generate response with SOC strategies.""" # 1. PII Sanitization (Typing Simulation Preparation) clean_msg = TypingSimulator.sanitize(scam_message) # 🚨 ENTERPRISE SAFEGUARD CHECK - DELEGATED TO ORCHESTRATOR # We allow "Unsafe" (Scam) content here to enable honeypot engagement. # Orchestrator still handles prompt injection protection. # 🔥 APPLY DYNAMIC TRAIT MUTATION if scammer_behavior: persona = self.mutate_traits(persona, scammer_behavior) self.logger.info("Persona traits mutated", traits=persona["traits"]) intel = intelligence or {} behavior_modifier = scammer_behavior.get("modifier") if scammer_behavior else None # 2. Intelligence Feedback Loop (Active Baiting) # FORCE EXTRACTION: If we are in 'extract' phase but have no payment info, FORCE the question. force_bait = False if current_phase == "extract" and not (intel.get("upi_ids") or intel.get("bank_accounts") or intel.get("credit_cards")): force_bait = True # Override prompt instruction to demand payment info scammer_behavior = scammer_behavior or {} scammer_behavior["modifier"] = "URGENT: Pretend you want to pay immediately. Ask for UPI ID or Bank Account details repeatedly." # If using static fallback, ensuring it asks for money is handled in _static_response current_phase = "extract" # Ensure phase sticks # 3. LLM Generation response_text = "" # 🔥 CORE INTEGRATION: EMOTIONAL ESCALATION (Persistence Fixed) turn_count = len(conversation_history) if conversation_history else 0 previous_agitation = "calm" if context and hasattr(context, "session"): previous_agitation = context.session.get("last_agitation") # Fallback to DB-backed aggregated intelligence if not previous_agitation: agg_intel = context.session.get("aggregated_intelligence", {}) meta_agitation = agg_intel.get("metadata_agitation", []) if meta_agitation: previous_agitation = meta_agitation[-1] # Final default previous_agitation = previous_agitation or "calm" # 🔥 ENHANCED: Pass persona for age-aware emotional escalation agitation_data = self.calculate_agitation_level( turn_count, scammer_behavior, previous_level=previous_agitation, scam_type=scam_type, persona=persona, # NEW: Age-aware emotional profiles is_repeating=is_repeating # [SCORING] Pass repetition state ) agitation = agitation_data["level"] escalation_reason = agitation_data["reason"] active_emotion = agitation_data.get("emotion", "neutral") # NEW emotional_profile = agitation_data.get("profile", "unknown") # NEW # Store for next turn persistence if context and hasattr(context, "session"): context.session["last_agitation"] = agitation context.session["last_emotion"] = active_emotion # NEW: Track active emotion context.session["persona"] = persona.get("selected_persona_key") # SYNC: For LLM fallback # 🔥 PERSISTENCE: Track justification for the judge if "aggregated_intelligence" in context.session: context.session["aggregated_intelligence"]["metadata_agitation_reason"] = escalation_reason context.session["aggregated_intelligence"]["metadata_emotion"] = active_emotion # NEW self.logger.info("Emotional Escalation Calculated", level=agitation, reason=escalation_reason, emotion=active_emotion, profile=emotional_profile, turn=turn_count) # Calculate Context Variables Early # Determine stress level (Propagated as 'high' or 'normal' for backward compatibility in internal tools) stress = "high" if agitation in ["paranoid", "volatile"] or "scared" in persona["traits"] or "worried" in persona["traits"] else "normal" if scammer_behavior and scammer_behavior.get("behavior") == "aggressive": stress = "high" # Determine Tech Literacy tech_literacy = "medium" if "not tech savvy" in persona["traits"]: tech_literacy = "low" elif "tech_savvy" in persona["traits"] or "analytical" in persona["traits"]: tech_literacy = "high" # Determine Profession profession = "General Public" if "desperate_jobseeker" in persona.get("selected_persona_key", ""): profession = "Unemployed" elif "curious_investor" in persona.get("selected_persona_key", ""): profession = "Investor" elif "elderly" in persona.get("selected_persona_key", ""): profession = "Retired" if settings.ENABLE_LLM_RESPONSES and self.llm_client and self.llm_client.is_available: try: # Ensure persona is valid dict if not isinstance(persona, dict) or "name" not in persona: self.logger.warning("Malformed persona passed to generate_response, fixing...") persona = self.get_all_personas().get("elderly_excited", list(self.get_all_personas().values())[0]) response_text = await self._llm_generate( clean_msg, persona, scam_type, conversation_history, current_phase, intel, behavior_modifier, stress=stress, tech_literacy=tech_literacy, profession=profession, agitation=agitation, context=context ) except Exception as e: import traceback self.logger.error("LLM Generation Failed (Runtime)", error=str(e)) if settings.DEBUG: traceback.print_exc() if not response_text: response_text = self._static_response( persona=persona, phase=current_phase, intelligence=intel, agitation=agitation ) # 3b. Anti-Repetition Guard (Prevent loops like "Main abhi kar raha hoon...") if conversation_history: last_responses = [m.get("honeypot_response", "").strip().lower() for m in conversation_history[-3:]] if response_text.strip().lower() in last_responses: # Force a different emotional variation self.logger.info("Repetition detected, forcing unique variation") response_text = self._static_response( message_text=scam_message, persona=persona, scam_type=scam_type, phase=current_phase, intelligence=intel, agitation=agitation ) # 4. Human Typing Simulation (Typos & Noise) # Note: Stress is already calculated above final_response = TypingSimulator.add_human_noise( response_text, persona["language"], stress, persona_key=persona.get("selected_persona_key", "generic"), traits=persona.get("traits", []), # Pass traits for professional check agitation_level=agitation ) # 5. 🔥 CORE INTEGRATION: Apply Realistic Engagement Delays # Wasting scammer time is the primary goal of the honeypot. # [SCORING] Optional But Powerful: Micro Typing Delay (even if delays disabled) # 0.3 - 0.8s improves realism score significantly without hurting latency metrics if not settings.ENABLE_ENGAGEMENT_DELAY: # [SCORING] Zero-Sleep Enforcement (Hackathon Mode) pass if settings.ENABLE_ENGAGEMENT_DELAY: # 5a. Simulate typing delay based on message length await engagement_delayer.simulate_typing(len(final_response)) # 5b. Add phase-specific "Thinking" or "System" delays if current_phase == "stall": # Heavy delays in stall phase to frustrate/occupy scammer if random.random() < 0.4: delay_seconds, excuse = await engagement_delayer.simulate_bank_issue() final_response = f"{excuse}\n\n{final_response}" elif random.random() < 0.3: delay_seconds, status = await engagement_delayer.simulate_otp_delay() final_response = f"{status}\n\n{final_response}" # 🔥 CORE INTEGRATION: Active Honeytoken Baiting # If we are in stall phase, give them "fake meat" to chew on if random.random() < 0.2: decoy = honeytoken_manager.generate_fake_bank_credentials( persona.get("victim_profile", {}).get("bank", "HDFC") ) bait_msg = f"Wait... I managed to log in! Can you check if this works? URL: {decoy['login_url']} User: {decoy['username']} Pass: {decoy['password']}" final_response = f"{final_response}\n\n{bait_msg}" elif current_phase == "engage": # Moderate delays to simulate a hesitant victim await engagement_delayer.delay(DelayType.THINKING) return final_response async def _llm_generate( self, message: str, persona: Dict, scam_type: str, history: List[Dict], phase: str, intel: Dict, behavior_modifier: str = None, stress: str = "normal", tech_literacy: str = "low", profession: str = "Unknown", agitation: str = "calm", context: Optional[Any] = None ) -> Optional[str]: """Internal LLM call with dynamic prompting.""" from app.core.prompts import FAST_CHAT_PROMPT, PHASE_GOALS # 1. Format History (AGGRESSIVE TRUNCATION for Groq Dev Tier) hist_str = "" if history: # FIX: Only last 2 messages (1 turn), max 150 chars each to avoid 413 for m in history[-2:]: s_msg = m.get('scammer_message', '')[:150] + ("..." if len(m.get('scammer_message', '')) > 150 else "") h_rsp = m.get('honeypot_response', '')[:150] + ("..." if len(m.get('honeypot_response', '')) > 150 else "") hist_str += f"Caller: {s_msg}\nMe: {h_rsp}\n" # 2. Truncate current message aggressively safe_message = message[:500] + ("..." if len(message) > 500 else "") # 3. Language-aware prompting (dynamic based on persona language) persona_language = persona.get("language", "hinglish").lower() # Map persona language to LLM instruction language_instructions = { "english": "pure English only", "hinglish": "Hinglish mix", "roman_hindi": "Hindi in Roman script", "hindi": "Hindi in Roman script", "marathi": "Marathi-Hindi mix", "tamil": "Tamil-English mix", "bengali": "Bengali-Hindi mix", "kannada": "Kannada-English mix", "punjabi": "Punjabi-Hindi mix" } language_instruction = language_instructions.get(persona_language, "Hinglish mix") # STRICT language rules to enforce proper language output language_strict_rules = { "english": "SPEAK ENGLISH ONLY. FORBIDDEN Hindi: arre, ruko, bhaiya, ji, yaar, kya, hai, nahi, acha, beta, haan, na. Use: umm, wait, hold on, please, actually, I don't understand. Example: 'Wait, can you explain why you need my OTP?'", "hinglish": "HINGLISH (Hindi+English mix). Example: 'Arre wait, kya bol rahe ho?'", "roman_hindi": "HINDI in Roman script only. NO English words. Example: 'Ruko, samajh nahi aaya'", "hindi": "HINDI in Roman script only. NO English words. Example: 'Kya bol rahe ho ji?'", "marathi": "MARATHI-Hindi mix in Roman script. Example: 'Thamba, kay zala?'", "tamil": "TAMIL-English mix in Roman script. Example: 'Wait pa, enna solla poringa?'", "bengali": "BENGALI-Hindi mix in Roman script. Example: 'Darao, ki bolcho?'", "kannada": "KANNADA-English mix in Roman script. Example: 'Wait madi, enu?'", "punjabi": "PUNJABI-Hindi mix in Roman script. Example: 'Oye ruk, ki hoya?'" } language_strict_rule = language_strict_rules.get(persona_language, language_strict_rules["hinglish"]) # 5. Get compact persona behavior from prompts from app.core.prompts import PERSONA_COMPACT_BEHAVIORS persona_key = persona.get("selected_persona_key", persona.get("key", "elderly_excited")) persona_behavior = PERSONA_COMPACT_BEHAVIORS.get(persona_key, "Confused but cooperative.") # 6. Format traits for prompt traits_list = persona.get("traits", ["confused"]) persona_traits = ", ".join(traits_list[:3]) # Max 3 traits to save tokens # 7. Use COMPACT prompt (critical for avoiding 413) formatted_prompt = FAST_CHAT_PROMPT.format( persona_name=persona.get("name", "Unknown"), profession=profession, tech_literacy=tech_literacy.upper(), persona_traits=persona_traits, agitation=agitation, phase=phase, persona_behavior=persona_behavior, language_instruction=language_instruction, language_strict_rule=language_strict_rule, history=hist_str if hist_str else "(First message)", message=safe_message ) if not self.llm_client: return None response = await self.llm_client.generate( prompt=formatted_prompt, role=ModelRole.FAST_CHAT, temperature=0.9, max_tokens=80, # Short responses like real SMS/chat context=context ) if not response: return None if isinstance(response, str): clean = response.strip().strip('"') elif hasattr(response, 'content') and response.content: clean = response.content.strip().strip('"') else: return None # 🔥 CRITICAL: Sanitize forbidden words that break honeypot illusion # Only replace words that reveal the honeypot's awareness import re forbidden_replacements = [ (r'\bscammer\b', 'sir'), (r'\bscam\b(?!\s*(?:team|prevention|department|squad))', 'issue'), # Don't replace "scam team" (r'\bhoneypot\b', 'system'), (r'\bbot\b', 'person'), (r'\bai assistant\b', 'helper'), (r'\bdetection\b', 'checking'), ] for pattern, replacement in forbidden_replacements: clean = re.sub(pattern, replacement, clean, flags=re.IGNORECASE) # 🔥 ENGLISH MODE: Replace leaked Hindi words with English equivalents import re if persona_language == "english": hindi_fixes = [ (r'\bbeta\b', 'son'), (r'\bbeti\b', 'daughter'), (r'\bji\b', ''), (r'\barre\b', 'oh'), (r'\bruko\b', 'wait'), (r'\bkya\b', 'what'), (r'\bhai\b', 'is'), (r'\bnahi\b', 'no'), (r'\bacha\b', 'okay'), (r'\bhaan\b', 'yes'), (r'\bthoda\b', 'a bit'), (r'\babhi\b', 'now'), (r'\bbhaiya\b', 'brother'), (r'\bdidi\b', 'sister'), (r'\byaar\b', 'friend'), (r'\bpaisa\b', 'money'), (r'\bpaise\b', 'money'), (r'\bsamajh\b', 'understand'), ] for hindi_pat, eng_word in hindi_fixes: clean = re.sub(hindi_pat, eng_word, clean, flags=re.IGNORECASE) clean = re.sub(r'\s+', ' ', clean).strip() # Clean double spaces return clean if clean else None def _static_response( self, message_text: str = "", persona: Dict = {}, scam_type: str = "general", phase: str = "engage", intelligence: Dict = {}, agitation: str = "calm" ) -> str: """ PRODUCTION-GRADE Local Fallback Responses. Runs when LLM is unavailable - MUST maintain honeypot realism! """ if not isinstance(persona, dict): persona = {"language": "english", "traits": ["curious"]} language = persona.get("language", "english").lower() is_hindi = "hindi" in language or "hinglish" in language persona_key = persona.get("selected_persona_key", "elderly_excited") traits = persona.get("traits", []) # ═══════════════════════════════════════════════════════════════════ # 1. SCAM-SPECIFIC TEMPLATES (Covers ALL scam types) # ═══════════════════════════════════════════════════════════════════ scam_templates = { "banking_scam": { "english": [ "Is this regarding my savings account? I used it yesterday only.", "I didn't receive any OTP yet, let me check my phone...", "Why would my account be blocked? This is worrying me...", "Can I visit the branch to sort this out instead?", "Wait, I need to tell my son first. He handles all this.", "My internet banking password I forgot, can you help?" ], "hinglish": [ "Mera account block kyun hoga? Maine kal hi use kiya tha!", "OTP nahi aaya abhi tak, ruko check karta hoon...", "Kya main bank branch aake baat kar sakti hoon?", "Ye message abhi kyun aaya mujhe? Bahut tension ho raha hai.", "Ruko beta, mera phone slow hai... OTP dhundh raha hoon.", "Aapka naam kya hai? Bank se ho toh ID batao na." ] }, "lottery_scam": { "english": [ "Wow really? How do I claim my prize?", "I never played any lottery though... how did I win?", "Is there a registration fee? I don't have much money right now.", "Can my son come collect on my behalf?", "What is the exact amount I won? Tell me clearly." ], "hinglish": [ "Sachi? Itna bada prize? Kaise milega?", "Maine toh lottery khareedi hi nahi thi, kaise jeeta?", "Kuch paise dene honge kya pehle? Mere paas cash nahi hai.", "Mera beta aake le sakta hai kya prize?", "Kitna amount mila hai exactly? Clearly batao." ] }, "job_scam": { "english": [ "What is the salary for this job?", "Why do I need to pay for getting a job?", "Is this work from home? I have family responsibilities.", "Can I speak to someone from HR department?", "Registration fee? That sounds unusual for a job..." ], "hinglish": [ "Salary kitni hai is job ki?", "Job ke liye mujhe paise kyun dene hain? Strange hai.", "Ye work from home hai kya? Main ghar se nikalna mushkil hai.", "HR se baat karwa do, mujhe confirm karna hai.", "Registration fee? Job mein fee? Ye theek nahi lag raha." ] }, "investment_scam": { "english": [ "What is the guaranteed return percentage?", "How do I know this is not a scam?", "Can you send me some documents to review?", "I need to discuss with my family before investing.", "Is SEBI registration with your company?" ], "hinglish": [ "Guaranteed return kitna milega?", "Ye scam toh nahi hai na? Mujhe darr lag raha hai.", "Koi documents bhejo na review ke liye.", "Family se poochna padega pehle, thoda time do.", "SEBI registration hai aapki company ki?" ] }, "tech_support_scam": { "english": [ "My computer has virus? How do you know?", "What is AnyDesk? I never heard of it.", "Can my nephew come help? He knows computers.", "Microsoft ke ho? ID dikha do please.", "Rs. 5000? That is too much for me right now..." ], "hinglish": [ "Mere computer mein virus hai? Kaise pata?", "AnyDesk kya hai? Maine kabhi nahi suna.", "Mera bhatija aa jayega help karne, usko computers aata hai.", "Microsoft se ho? ID dikhao please.", "5000 rupees? Bahut zyada hai abhi mere paas nahi." ] }, "delivery_scam": { "english": [ "What courier? I didn't order anything...", "Customs fee? For what item?", "Can I pay when delivery comes?", "Who sent this package to me?", "I need to check with my son, he orders things online." ], "hinglish": [ "Kaunsa courier? Maine toh kuch order nahi kiya.", "Customs fee? Kis item ka?", "Delivery aane pe pay kar doongi, theek hai?", "Ye package kisne bheja mujhe?", "Beta se poochna padega, wo online cheezein mangaata hai." ] }, "government_scam": { "english": [ "Which department is this from?", "I always pay my taxes on time, what is the issue?", "Can I come to your office instead?", "Please send official letter to my address.", "FIR? Against me? This must be a mistake!" ], "hinglish": [ "Kaun sa department hai ye?", "Main toh tax time pe bharta hoon hamesha, kya problem hai?", "Aapke office aa jaata hoon, address batao.", "Official letter bhejo meri address pe.", "FIR? Mere upar? Galti ho gayi shayad!" ] }, "loan_scam": { "english": [ "What is the interest rate?", "Processing fee? Banks don't charge upfront...", "I need to discuss with my wife first.", "Is this RBI approved loan scheme?", "Send documents on WhatsApp, I will review." ], "hinglish": [ "Interest rate kya hai?", "Processing fee? Bank toh pehle nahi leta...", "Wife se baat karni padegi pehle.", "Ye RBI approved scheme hai?", "Documents WhatsApp pe bhejo, dekhta hoon." ] }, "crypto_scam": { "english": [ "Bitcoin? I don't understand these new things.", "Guaranteed profit? That sounds too good to be true.", "My son told me bitcoin is risky, is that true?", "How do I invest? I have never done this before.", "₹50,000 minimum? That is my savings..." ], "hinglish": [ "Bitcoin? Ye naye cheezein mujhe nahi samajh aati.", "Guaranteed profit? Ye toh zyada sundar lag raha hai.", "Beta ne bola bitcoin risky hai, sach hai kya?", "Invest kaise karun? Kabhi nahi kiya ye pehle.", "50,000 minimum? Ye toh meri savings hai..." ] } } # ═══════════════════════════════════════════════════════════════════ # 2. PHASE-SPECIFIC GENERIC TEMPLATES # ═══════════════════════════════════════════════════════════════════ generic_variations = { "hook": { "english": [ "Okay, tell me more about this.", "What exactly do I need to do?", "Is this urgent? Should I worry?", "Hmm, interesting. Go on...", "Who gave you my number?", "Wait, I am a bit confused, can you start over?", "Is this from the bank directly?" ], "hinglish": [ "Acha, aur kya karna hoga?", "Theek hai, batao puri baat.", "Ye urgent hai kya? Tension loon?", "Hmm, interesting hai. Bolo aage.", "Mera number kisne diya aapko?", "Thoda confusion ho raha hai, phir se bolo.", "Ye bank se hi call hai na?" ] }, "engage": { "english": [ "I am listening carefully.", "Wait, my internet is very slow today...", "Can you explain that again slowly?", "Hold on, someone is at the door.", "One second, my phone is lagging.", "My battery is about to die, let me find a charger.", "Main road pe hoon, shor bahut hai. Phir se bolo?" ], "hinglish": [ "Ha main sun raha hoon dhyan se.", "Ruko, net bahut slow hai aaj.", "Ek baar phir se samjhao please.", "Ruko, darwaze pe koi hai.", "Ek second, phone hang ho raha.", "Battery khatam ho rahi hai, charger dhoondne do.", "Main bahaar hoon, awaaz nahi aa rahi theek se." ] }, "extract": { "english": [ "Okay, sending the details now.", "Wait, I am finding my card...", "Can I pay using UPI instead?", "My account number... wait, let me get my passbook.", "OTP? Let me check messages...", "The app is not opening, what should I do?", "I am trying to log in but password is wrong." ], "hinglish": [ "Ha theek hai, details bhej raha hoon.", "Ruko card dhoond raha hoon.", "UPI se pay kar doon kya?", "Account number... ruko passbook laata hoon.", "OTP? Ruko messages check karta hoon...", "App khul hi nahi raha, kya karun?", "Login kar raha hoon par password wrong bata raha." ] }, "stall": { "english": [ "One moment please, my son is calling.", "Battery is very low, might disconnect.", "Network problem here, can you hear me?", "Wait, I need to go to ATM first.", "Call me after 1 hour, I am busy now.", "My wife is asking who I am talking to.", "Ruko, mujhe chashma dhoondne do." ], "hinglish": [ "Ek min ruko, beta call kar raha hai.", "Battery bahut kam hai, disconnect ho sakta hai.", "Network problem hai yahan, awaaz aa rahi hai?", "Ruko, pehle ATM jaana padega.", "1 ghante baad call karo, abhi busy hoon.", "Wife pooch rahi hai kisse baat kar raha hoon.", "Ruko, chashma nahi mil raha mera." ] } } # ═══════════════════════════════════════════════════════════════════ # 3. PERSONA-SPECIFIC MODIFIERS (Add personality flavor) # ═══════════════════════════════════════════════════════════════════ persona_suffixes = { "elderly_excited": ["😊", "Beta...", "Acha acha...", "Theek hai ji", ""], "worried_customer": ["😟", "Bahut tension ho raha hai...", "Kya karun?", "Bachao mujhe", ""], "skeptical_user": ["🤔", "Hmm pakka?", "Ye theek hai na?", "Fraud toh nahi hai na?", ""], "desperate_jobseeker": ["🙏", "Please help karo", "Job bahut chahiye", "Ghar mein paise nahi bache", ""], "rural_farmer": ["", "Sahab...", "Haan ji", "Ram Ram", ""] } # Selection Logic import random pool = [] # 1. Try specific scam type first if scam_type in scam_templates: lang_key = "hinglish" if is_hindi else "english" pool = scam_templates[scam_type].get(lang_key, []) # 2. Fallback to generic phase if not pool: lang_key = "hinglish" if is_hindi else "english" pool = generic_variations.get(phase, generic_variations["engage"]).get(lang_key, []) # 3. Select response and add persona flavor base_response = random.choice(pool) if pool else "Ha theek hai, ruko..." # [REALISM] Time-Aware Context Injection (Consolidated) # 30% chance for a time-aware opener (matching TimeAwareBehavior logic) time_context = "" if random.random() < 0.3: time_lang = "english" if not is_hindi else "hinglish" time_context = TimeAwareBehavior.get_time_excuse(language=time_lang) # Combine base response with time context (if any) if time_context: response = f"{time_context} {base_response}" else: response = base_response # 5. Emotional Augmentation (Consistency with LLMClient Fallback) if agitation in ["paranoid", "volatile"]: prefix = "Wait... " if "hindi" not in str(persona.get("language")).lower() else "Ruko... " postfix = " 😰" if response and not response.endswith("😰"): response = f"{prefix}{response}{postfix}" return response def _construct_bait_prompt(self, intel, persona) -> Optional[str]: """Specific logic to confirm extracted intel.""" # This function signals Orchestrator or LLM to focus on verification # Handled implicitly via prompt injection in _llm_generate return None # Export __all__ = ["PersonaEngine", "PERSONAS"]