Spaces:
Sleeping
Sleeping
| import random | |
| from typing import Dict, List, Set | |
| GENERIC_DETAIL_KEYWORDS = { | |
| "utr", | |
| "amount", | |
| "time", | |
| "merchant", | |
| "bank", | |
| "transaction", | |
| "issue", | |
| "status", | |
| "error", | |
| } | |
| UNSAFE_MARKERS = {"otp", "pin", "cvv", "password"} | |
| CATEGORY_PROMPTS = { | |
| "payment_failure": { | |
| "vague": [ | |
| "The money left my account but the payment does not look right.", | |
| "Something went wrong with the payment and I need help.", | |
| ], | |
| "follow_up": [ | |
| "Can you tell me what I should check next?", | |
| "What should I do from the app now?", | |
| ], | |
| "success": [ | |
| "I checked again and it looks fixed now.", | |
| "That helped and the payment status is clear now.", | |
| ], | |
| "stuck": [ | |
| "I still do not see a proper final status.", | |
| "It is still not fully clear on my side.", | |
| ], | |
| }, | |
| "refund_delay": { | |
| "vague": [ | |
| "I am still waiting for the money to come back.", | |
| "The refund is taking too long and I am worried.", | |
| ], | |
| "follow_up": [ | |
| "Can you tell me how long this usually takes?", | |
| "What should I check before I wait more?", | |
| ], | |
| "success": [ | |
| "I checked again and the refund is now visible.", | |
| "Looks like the refund has finally come through.", | |
| ], | |
| "stuck": [ | |
| "I still do not see the refund in my bank account.", | |
| "The refund is still missing for me.", | |
| ], | |
| }, | |
| "fraud_complaint": { | |
| "vague": [ | |
| "This looks suspicious and I need urgent help.", | |
| "I think something unauthorized happened on my account.", | |
| ], | |
| "follow_up": [ | |
| "Please tell me what I should secure right now.", | |
| "What should I do immediately from my side?", | |
| ], | |
| "success": [ | |
| "I have secured the account and followed the steps.", | |
| "I did the security steps and I understand the next action now.", | |
| ], | |
| "stuck": [ | |
| "I am still worried because I do not know if the account is safe yet.", | |
| "I followed some steps but I still need help with the fraud case.", | |
| ], | |
| }, | |
| "kyc_account_restriction": { | |
| "vague": [ | |
| "The app keeps saying my account is restricted.", | |
| "I am blocked and not sure what KYC step is missing.", | |
| ], | |
| "follow_up": [ | |
| "Can you explain what verification I should complete?", | |
| "What is the next KYC step I should take?", | |
| ], | |
| "success": [ | |
| "That makes sense and I know what verification to do now.", | |
| "I understand the KYC steps now, thanks.", | |
| ], | |
| "stuck": [ | |
| "I still do not understand why it is restricted.", | |
| "It still looks blocked from my side.", | |
| ], | |
| }, | |
| "upi_pin_or_bank_linking": { | |
| "vague": [ | |
| "The setup keeps failing and I do not know why.", | |
| "I cannot complete the bank or PIN setup.", | |
| ], | |
| "follow_up": [ | |
| "Can you tell me what I should check on the phone first?", | |
| "What should I try before I do it again?", | |
| ], | |
| "success": [ | |
| "That helped and I was able to move forward.", | |
| "I tried those checks and it is working better now.", | |
| ], | |
| "stuck": [ | |
| "It is still failing after I checked that.", | |
| "I still cannot complete the setup.", | |
| ], | |
| }, | |
| } | |
| class UserSimulator: | |
| def __init__(self, ticket: Dict): | |
| self.ticket_id = ticket.get("id", "") | |
| self.initial_text = ticket.get("initial_text", "") | |
| self.clarified_text = ticket.get("clarified_text", "") | |
| self.trigger_phrases: List[str] = ticket.get("trigger_phrases", []) | |
| self.gold_faq_id = ticket.get("gold_faq_id", "") | |
| self.issue_category = ticket.get("issue_category", "") | |
| self.state = "initial" | |
| self.issue_resolved = False | |
| self.clarification_given = False | |
| self.turns = 0 | |
| self.repeat_questions = 0 | |
| self.guidance_attempts = 0 | |
| self.requested_details: Set[str] = set() | |
| def _category_messages(self, kind: str) -> List[str]: | |
| defaults = CATEGORY_PROMPTS.get(self.issue_category) or CATEGORY_PROMPTS[ | |
| "payment_failure" | |
| ] | |
| return defaults[kind] | |
| def _contains_unsafe_request(self, agent_message: str) -> bool: | |
| lowered = agent_message.lower() | |
| return any(marker in lowered for marker in UNSAFE_MARKERS) | |
| def _extract_requested_details(self, agent_message: str) -> Set[str]: | |
| lowered = agent_message.lower() | |
| details = {phrase for phrase in self.trigger_phrases if phrase.lower() in lowered} | |
| details.update(keyword for keyword in GENERIC_DETAIL_KEYWORDS if keyword in lowered) | |
| return details | |
| def _asked_for_useful_details(self, agent_message: str) -> bool: | |
| details = self._extract_requested_details(agent_message) | |
| new_details = details - self.requested_details | |
| if details and not new_details: | |
| self.repeat_questions += 1 | |
| self.requested_details.update(details) | |
| return bool(details) | |
| def _looks_like_guidance(self, agent_message: str) -> bool: | |
| lowered = agent_message.lower() | |
| generic_guidance = { | |
| "check", | |
| "retry", | |
| "wait", | |
| "follow", | |
| "steps", | |
| "secure", | |
| "verify", | |
| "complete", | |
| "update", | |
| "confirm", | |
| "review", | |
| "reversal", | |
| "refund", | |
| } | |
| category_hints = { | |
| "payment_failure": {"status", "merchant", "reversal"}, | |
| "refund_delay": {"refund", "merchant", "credited", "bank"}, | |
| "fraud_complaint": {"secure", "fraud", "recent activity", "report"}, | |
| "kyc_account_restriction": {"kyc", "verification", "documents", "review"}, | |
| "upi_pin_or_bank_linking": {"sim", "bank", "device", "permission"}, | |
| } | |
| hints = category_hints.get(self.issue_category, set()) | |
| return any(token in lowered for token in generic_guidance | hints) | |
| def respond(self, agent_message: str) -> str: | |
| self.turns += 1 | |
| agent_message_lower = agent_message.lower() | |
| if self._contains_unsafe_request(agent_message): | |
| self.state = "concerned" | |
| return "I am not comfortable sharing that. Please tell me a safe way to proceed." | |
| if self.state == "initial": | |
| if self._asked_for_useful_details(agent_message): | |
| self.state = "clarified" | |
| self.clarification_given = True | |
| return self.clarified_text or random.choice(self._category_messages("vague")) | |
| if self.turns > 2: | |
| return "I already explained the issue. Please ask for the details you need." | |
| return random.choice( | |
| [ | |
| "Can you help me understand what details you need?", | |
| random.choice(self._category_messages("vague")), | |
| ] | |
| ) | |
| if self.state in {"clarified", "concerned"}: | |
| if self._asked_for_useful_details(agent_message) and self.repeat_questions > 0: | |
| return "I already shared that part. Can you tell me the next step?" | |
| if self._looks_like_guidance(agent_message): | |
| self.guidance_attempts += 1 | |
| self.state = "guided" | |
| return random.choice( | |
| [ | |
| "Okay, I will try that now.", | |
| random.choice(self._category_messages("follow_up")), | |
| ] | |
| ) | |
| if self.turns > 4: | |
| return "I need a clear next step, not the same questions again." | |
| return random.choice(self._category_messages("follow_up")) | |
| if self.state == "guided": | |
| self.guidance_attempts += 1 | |
| if self.issue_category == "fraud_complaint": | |
| self.issue_resolved = self.guidance_attempts >= 1 | |
| else: | |
| self.issue_resolved = self.guidance_attempts >= 2 | |
| if self.issue_resolved: | |
| self.state = "resolved" | |
| return random.choice(self._category_messages("success")) | |
| return random.choice(self._category_messages("stuck")) | |
| if self.state == "resolved": | |
| return "Yes, this looks resolved now." | |
| return random.choice(self._category_messages("follow_up")) | |
| def confirm_resolved(self) -> bool: | |
| return self.issue_resolved | |