""" Orchestrator — Task Queue Dialogue Management =============================================== Replaces the one-intent-one-slot FSM. Fixes every issue in the feedback: P0 Compound requests → NLU decomposes; ALL tasks enter a queue; each is acknowledged and executed in order. Nothing is silently dropped. P0 Wrong destructive → destructive intents require BOTH intent confidence ≥ 0.75 AND explicit confirmation. Below threshold → clarifying question, never action. P1 Named entity ignored → slots from NLU prefill the task; only genuinely missing slots are asked. P1 Dead-end fallback → "unknown" triggers ONE clarifying question with candidate intents; second failure → human handoff WITH full context attached to the CRM ticket. P2 No balance guard → transfers are checked against balance before confirmation; insufficient funds is surfaced. P2 Number formatting → fmt_amount() used everywhere: "₦350,000". """ import re import uuid import logging from dataclasses import dataclass, field from datetime import datetime from typing import Optional from nlu import NLU, INTENT_SCHEMA logger = logging.getLogger(__name__) CONFIDENCE_GATE_DESTRUCTIVE = 0.75 # below this, never act on money/card intents CONFIDENCE_GATE_NORMAL = 0.50 MAX_CLARIFY_ATTEMPTS = 2 # then human handoff with context MAX_TURNS = 20 # ── Formatting (P2) ─────────────────────────────────────────────────────────── def fmt_amount(raw) -> str: """'350000' → '₦350,000' — single source of truth for money display.""" try: n = int(str(raw).replace(",", "").replace(".", "").replace("₦", "").strip()) return f"₦{n:,}" except (ValueError, TypeError): return f"₦{raw}" def amount_int(raw) -> Optional[int]: try: return int(str(raw).replace(",", "").replace(".", "").replace("₦", "").strip()) except (ValueError, TypeError): return None # ── Task model ──────────────────────────────────────────────────────────────── @dataclass class Task: task_id: str intent: str slots: dict confidence: float status: str = "pending" # pending | collecting | confirming | done | failed result: Optional[str] = None @property def required_slots(self) -> list: return INTENT_SCHEMA.get(self.intent, {}).get("slots", []) @property def missing_slots(self) -> list: # account_id is optional if we already verified identity this session return [s for s in self.required_slots if s not in self.slots] @property def is_destructive(self) -> bool: return INTENT_SCHEMA.get(self.intent, {}).get("destructive", False) @dataclass class Session: session_id: str = "" tasks: list = field(default_factory=list) # task queue active_task: Optional[Task] = None awaiting_slot: Optional[str] = None awaiting_confirmation: bool = False verified_account: Optional[str] = None # identity, session-scoped balance: Optional[int] = None # cached after lookup clarify_attempts: int = 0 turn: int = 0 escalated: bool = False history: list = field(default_factory=list) started_at: str = field(default_factory=lambda: datetime.utcnow().isoformat()) # ── Slot questions ──────────────────────────────────────────────────────────── SLOT_QUESTIONS = { "account_id": "Could you give me your account number, please?", "recipient": "Who would you like to send the money to?", "amount": "How much would you like to send?", "location": "Which city or area are you in?", "issue_desc": "Could you briefly describe the problem?", "order_id": "What is your order number?", "return_reason": "What is the reason for the return?", } class Orchestrator: def __init__(self, crm=None, nlu: Optional[NLU] = None): self.crm = crm self.nlu = nlu or NLU() def new_session(self) -> Session: return Session(session_id=str(uuid.uuid4())[:8]) # ── Main entry ──────────────────────────────────────────────────────────── def respond(self, english_text: str, hausa_text: str, session: Session) -> tuple[str, Session, bool]: session.turn += 1 session.history.append({"role": "user", "en": english_text, "ha": hausa_text, "turn": session.turn}) if session.turn >= MAX_TURNS: return self._escalate(session, "We've been talking a while — let me connect you to a colleague " "who can wrap this up quickly.") # 1. NLU with pending-slot context pending_intent = session.active_task.intent if session.active_task else None nlu_result = self.nlu.parse(english_text, pending_intent=pending_intent, pending_slot=session.awaiting_slot) tasks_found = nlu_result["tasks"] # 2. Route the parse response = self._handle_parse(tasks_found, english_text, session) # 3. Drive the queue until we need user input followup = self._drive_queue(session) if followup: response = (response + " " + followup).strip() if response else followup session.history.append({"role": "agent", "en": response, "ha": None, "turn": session.turn}) return response, session, session.escalated # ── Parse handling ──────────────────────────────────────────────────────── def _handle_parse(self, tasks_found: list, raw_text: str, session: Session) -> str: parts = [] for parsed in tasks_found: intent = parsed["intent"] conf = parsed["confidence"] slots = parsed["slots"] # ── Universal intents ──────────────────────────────────────────── if intent == "human_agent": return self._escalate(session, "Of course — connecting you to a human agent now. " f"Your reference is REF-{session.session_id}. " "They will see everything we discussed.")[0] if intent in ("cancel",): session.active_task = None session.awaiting_slot = None session.awaiting_confirmation = False session.tasks = [t for t in session.tasks if t.status == "done"] parts.append("Alright, I've cancelled that.") continue if intent == "goodbye": pending = [t for t in session.tasks if t.status in ("pending", "collecting", "confirming")] if pending: parts.append(f"Before you go — we still have " f"{len(pending)} pending request(s). " f"Say 'cancel' to drop them, or continue.") else: parts.append("Thank you for calling. Have a wonderful day!") continue if intent == "greeting" and session.turn <= 2: parts.append("Hello! I can help you check balances, send money, " "pay bills, track orders, or report an issue. " "What can I do for you?") continue # ── Confirmation answers for the active task ───────────────────── if session.awaiting_confirmation and intent in ( "confirmation_yes", "confirmation_no"): parts.append(self._handle_confirmation( intent == "confirmation_yes", session)) continue # ── Slot answer for the active task ────────────────────────────── if (session.awaiting_slot and session.active_task and intent in ("unknown", session.active_task.intent)): filled = self._try_fill_pending_slot(raw_text, slots, session) if filled: continue # queue driver will move it forward # ── Unknown → clarify, not dead-end (P1) ───────────────────────── if intent == "unknown" or conf < CONFIDENCE_GATE_NORMAL: parts.append(self._clarify_or_handoff(raw_text, session)) continue # ── New task → enqueue with prefilled slots (P0 + P1) ──────────── task = Task(task_id=str(uuid.uuid4())[:6], intent=intent, slots=dict(slots), confidence=conf) # identity carries over within the session if session.verified_account and "account_id" in task.required_slots: task.slots.setdefault("account_id", session.verified_account) session.tasks.append(task) session.clarify_attempts = 0 # Acknowledge compound requests explicitly (P0 #1) new_tasks = [t for t in session.tasks if t.status == "pending"] if len(new_tasks) > 1: names = ", then ".join(self._intent_label(t.intent) for t in new_tasks) parts.insert(0, f"Got it — I'll {names}, one at a time.") return " ".join(p for p in parts if p) # ── Queue driver ────────────────────────────────────────────────────────── def _drive_queue(self, session: Session) -> str: """ Advance the active task; when it completes, pull the next from the queue. Stops (returns a question) whenever user input is needed. """ out = [] while True: # promote next task if none active if session.active_task is None: nxt = next((t for t in session.tasks if t.status == "pending"), None) if nxt is None: break session.active_task = nxt nxt.status = "collecting" # identity verified earlier in this session carries forward — # never re-ask for the account number (P1) if session.verified_account and "account_id" in nxt.required_slots: nxt.slots.setdefault("account_id", session.verified_account) task = session.active_task # 1. Missing slots? Ask for exactly ONE (but never one we have) missing = task.missing_slots if missing: slot = missing[0] session.awaiting_slot = slot out.append(SLOT_QUESTIONS.get( slot, f"Could you provide the {slot.replace('_', ' ')}?")) return " ".join(out) # 2. Destructive → confidence gate + confirmation (P0 #2) if task.is_destructive and not session.awaiting_confirmation: if task.confidence < CONFIDENCE_GATE_DESTRUCTIVE: session.awaiting_confirmation = True task.status = "confirming" out.append(self._describe_action(task) + " — did I understand that correctly? " "Please say yes or no.") return " ".join(out) # balance guard before confirming a transfer (P2) guard_msg = self._balance_guard(task, session) if guard_msg: task.status = "failed" session.active_task = None session.awaiting_slot = None out.append(guard_msg) continue session.awaiting_confirmation = True task.status = "confirming" out.append(self._describe_action(task) + " Say yes to confirm or no to cancel.") return " ".join(out) # 3. Execute non-destructive task result = self._execute(task, session) task.status = "done" task.result = result out.append(result) session.active_task = None session.awaiting_slot = None # queue empty if out: remaining = [t for t in session.tasks if t.status == "pending"] if not remaining and not session.awaiting_confirmation: out.append("Is there anything else I can help you with?") return " ".join(out) # ── Confirmation handling ───────────────────────────────────────────────── def _handle_confirmation(self, confirmed: bool, session: Session) -> str: task = session.active_task session.awaiting_confirmation = False if task is None: return "" if confirmed: result = self._execute(task, session) task.status = "done" task.result = result session.active_task = None return result task.status = "failed" session.active_task = None return ("No problem, I've cancelled that. " "Just tell me if you'd like to do something else.") # ── Slot filling ────────────────────────────────────────────────────────── def _try_fill_pending_slot(self, raw_text: str, nlu_slots: dict, session: Session) -> bool: task = session.active_task slot = session.awaiting_slot if not task or not slot: return False value = nlu_slots.get(slot) # heuristic extraction for short answers if not value: t = raw_text.strip() if slot == "account_id": m = re.search(r'\b(\d{6,12})\b', t) value = m.group(1) if m else None elif slot == "amount": m = re.search(r'\b(\d{1,3}(?:[,\.]\d{3})*|\d+)\b', t) value = m.group(1) if m else None elif slot == "recipient": m = re.match(r'^(?:to\s+)?([a-zA-Z]{2,20})$', t) value = m.group(1) if m else None elif slot in ("issue_desc", "return_reason", "location"): # ANY non-trivial answer counts — "too small" is a valid # return reason (fixes P1 dead-end) value = t if len(t) >= 2 else None elif slot == "order_id": m = re.search(r'\b([a-zA-Z0-9\-]{4,20})\b', t) value = m.group(1) if m else None if value: task.slots[slot] = str(value) session.awaiting_slot = None session.clarify_attempts = 0 if slot == "account_id": session.verified_account = str(value) return True return False # ── Clarify → handoff (P1 dead-end fix) ────────────────────────────────── def _clarify_or_handoff(self, raw_text: str, session: Session) -> str: session.clarify_attempts += 1 if session.clarify_attempts >= MAX_CLARIFY_ATTEMPTS: return self._escalate(session, "I want to make sure you get proper help — connecting you " "to a human agent now. They'll see our whole conversation, " "so you won't have to repeat anything.")[0] return ("Sorry — just to be sure I get this right: are you asking " "about your balance, a transfer, a payment, an order, " "or something else? You can also say it in your own words.") # ── Execution ───────────────────────────────────────────────────────────── def _execute(self, task: Task, session: Session) -> str: i = task.intent s = task.slots if i == "balance_inquiry": bal = self._mock_balance(s.get("account_id", session.verified_account or "0")) session.balance = bal return (f"Your account balance is {fmt_amount(bal)} " f"as of {datetime.utcnow().strftime('%d %b %Y')}.") if i == "send_money": tx = f"TXN-{session.session_id}-{task.task_id.upper()}" amt = fmt_amount(s['amount']) if session.balance is not None: session.balance -= amount_int(s["amount"]) or 0 return (f"Done — {amt} sent to {s['recipient'].title()}. " f"Transaction reference: {tx}.") if i == "bill_payment": tx = f"TXN-{session.session_id}-{task.task_id.upper()}" return (f"Your payment of {fmt_amount(s['amount'])} has been " f"processed. Reference: {tx}.") if i == "block_card": return (f"Your card linked to account ending " f"…{s.get('account_id', '')[-4:]} is now blocked. " f"A replacement can be requested at any branch.") if i == "branch_info": loc = s.get("location", "your area") return (f"Our closest branch to {loc} is at 12 Ahmadu Bello Way — " f"open Monday to Friday, 8am to 4pm. You can pick up an " f"ATM card there with a valid ID.") if i == "card_request": return ("You can collect a new ATM card at any branch with a " "valid ID, or I can order one to be delivered — " "just say 'deliver my card'.") if i == "track_order": oid = s.get("order_id", "") return (f"Order {oid} is out for delivery and should arrive " f"within 2 business days.") if i == "return_item": ticket = self._crm_ticket(session, f"Return request — order {s.get('order_id','?')} — " f"reason: {s.get('return_reason','?')}") return (f"I've registered your return for order " f"{s.get('order_id','')} (reason: {s.get('return_reason','')}). " f"Ticket {ticket}. You'll receive a pickup label by SMS.") if i == "report_issue": ticket = self._crm_ticket(session, s.get("issue_desc", raw := "")) return (f"I've created support ticket {ticket}. " f"Our team will contact you within 24 hours.") return "Done." # ── Balance guard (P2) ──────────────────────────────────────────────────── def _balance_guard(self, task: Task, session: Session) -> Optional[str]: if task.intent != "send_money": return None amt = amount_int(task.slots.get("amount")) if amt is None: return None # look up balance if we haven't yet this session if session.balance is None and task.slots.get("account_id"): session.balance = self._mock_balance(task.slots["account_id"]) if session.balance is not None and amt > session.balance: return (f"I can't process that transfer: you asked to send " f"{fmt_amount(amt)} but your balance is " f"{fmt_amount(session.balance)}. " f"Would you like to send a smaller amount?") return None # ── Escalation with context (P1) ───────────────────────────────────────── def _escalate(self, session: Session, message: str) -> tuple[str, Session, bool]: session.escalated = True transcript = "\n".join( f"[{h['turn']}] {h['role']}: {h.get('en','')}" for h in session.history) self._crm_ticket(session, f"ESCALATION — full context attached:\n{transcript}", subject=f"Voice escalation {session.session_id}") pending = [t for t in session.tasks if t.status in ("pending", "collecting", "confirming")] if pending: message += (f" Note for the agent: {len(pending)} request(s) " f"still open ({', '.join(t.intent for t in pending)}).") return message, session, True # ── Helpers ─────────────────────────────────────────────────────────────── def _describe_action(self, task: Task) -> str: s = task.slots if task.intent == "send_money": return (f"You want to send {fmt_amount(s['amount'])} " f"to {s['recipient'].title()}.") if task.intent == "bill_payment": return f"You want to pay {fmt_amount(s['amount'])}." if task.intent == "block_card": return (f"You want to BLOCK the card on account " f"ending …{s.get('account_id','')[-4:]}.") return f"You want to {self._intent_label(task.intent)}." @staticmethod def _intent_label(intent: str) -> str: return { "balance_inquiry": "check your balance", "send_money": "make a transfer", "bill_payment": "pay a bill", "block_card": "block your card", "branch_info": "find a branch", "card_request": "get a card", "track_order": "track your order", "return_item": "process a return", "report_issue": "log your issue", }.get(intent, intent.replace("_", " ")) @staticmethod def _mock_balance(account_id: str) -> int: import hashlib seed = int(hashlib.md5(str(account_id).encode()).hexdigest()[:6], 16) return (seed % 400_000) + 50_000 def _crm_ticket(self, session: Session, description: str, subject: str = "") -> str: if self.crm: try: res = self.crm.create_ticket( subject=subject or f"Voice session {session.session_id}", description=description) return res["ticket_id"] except Exception as e: logger.warning(f"CRM ticket failed: {e}") return f"TKT-{session.session_id}-{str(uuid.uuid4())[:4].upper()}"