Create orchestrator.py

orchestrator.py

@@ -0,0 +1,514 @@
+"""
+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()}"