""" PlotWeaver Voice Agent — Dialogue Manager ========================================== FSM for multi-turn Hausa conversations across 3 verticals. State lives in Gradio session state (dict) — no Redis needed in the Space. What changed vs the POC ----------------------- 1. Backend actions: data/terminal states may carry an ``"action"`` key naming a Backend method (see backend.py). The method runs on entry and merges language-neutral results into ``state.slots``. 2. Prompt templating: prompts may contain ``{slot}`` placeholders, filled from slots via ``render_prompt``. Confirmations now echo the real amount/recipient instead of saying "I'll send the money now" with no detail. """ from __future__ import annotations import logging from dataclasses import dataclass, field, asdict from enum import Enum from typing import Optional from backend import Backend, MockBackend logger = logging.getLogger("plotweaver.dialogue") _DEFAULT_BACKEND: Backend = MockBackend() class Vertical(str, Enum): BANK = "bank" TELECOM = "telecom" ECOMMERCE = "ecommerce" @dataclass class DialogueState: session_id: str vertical: str current_state: str = "greeting" slots: dict = field(default_factory=dict) turn_count: int = 0 escalate_to_human: bool = False history: list = field(default_factory=list) consecutive_unknowns: int = 0 state_before_fallback: Optional[str] = None def to_dict(self): return asdict(self) @classmethod def from_dict(cls, d): return cls(**d) if d else None SCENARIOS = { "bank": { "name": "PlotWeaver Bank", "states": { "greeting": { "ha": "Sannu! Wannan shine mataimakin banki na PlotWeaver. Yaya zan taimake ka yau? Za ka iya ce 'duba ma'auni', 'toshe kati', ko 'canjin kuɗi'.", "en": "Hello! This is the PlotWeaver banking assistant. How can I help you today? You can say 'check balance', 'block card', or 'transfer money'.", "expects": "intent", "transitions": {"check_balance": "ask_account_number", "block_card": "confirm_block_card", "transfer_money": "ask_recipient"}, }, "ask_account_number": { "ha": "Don Allah ka faɗi lambobin ƙarshe huɗu na asusunka.", "en": "Please say the last four digits of your account number.", "expects": "digits", "transitions": {"provide_digits": "return_balance"}, }, "return_balance": { "ha": "Ma'aunin asusunka shine Naira {balance}. Akwai wani abu?", "en": "Your account balance is {balance} Naira. Anything else?", "action": "get_balance", "expects": "yesno", "transitions": {"yes": "greeting", "no": "exit"}, }, "confirm_block_card": { "ha": "Don tabbatar, kana son toshe katinka? Ka ce 'i' ko 'a'a'.", "en": "To confirm, you want to block your card? Say 'yes' or 'no'.", "expects": "yesno", "transitions": {"yes": "card_blocked", "no": "greeting"}, }, "card_blocked": { "ha": "An toshe katinka (lamba {block_ref}). Sabon kati zai iso a cikin kwanaki {card_eta_days}. Ana juya ka ga wakili don tabbatarwa.", "en": "Your card is blocked (ref {block_ref}). A new card will arrive in {card_eta_days} days. Transferring you to an agent for confirmation.", "action": "block_card", "expects": None, "terminal": True, "escalate": True, }, "ask_recipient": { "ha": "Zuwa wa kake son turawa? Ka faɗi sunan mai karɓa.", "en": "Who do you want to transfer to? Say the recipient's name.", "expects": "name", "transitions": {"provide_name": "ask_amount"}, }, "ask_amount": { "ha": "Nawa kake son turawa, a Naira?", "en": "How much do you want to transfer, in Naira?", "expects": "amount", "transitions": {"provide_amount": "confirm_transfer"}, }, "confirm_transfer": { "ha": "Don tabbatarwa: zan tura Naira {amount} zuwa {name}. Ka ce 'i' don ci gaba.", "en": "To confirm: I'll transfer {amount} Naira to {name}. Say 'yes' to continue.", "expects": "yesno", "transitions": {"yes": "transfer_done", "no": "greeting"}, }, "transfer_done": { "ha": "An tura Naira {amount} zuwa {name}. Lambar tabbatarwa: {txn_ref}. Godiya da zabar PlotWeaver Bank.", "en": "Sent {amount} Naira to {name}. Reference: {txn_ref}. Thank you for choosing PlotWeaver Bank.", "action": "transfer", "expects": None, "terminal": True, }, }, }, "telecom": { "name": "PlotWeaver Telecom", "states": { "greeting": { "ha": "Sannu! Wannan shine PlotWeaver Telecom. Kana son 'saya airtime', 'saya bundle', ko 'yin korafi'?", "en": "Hello! This is PlotWeaver Telecom. Would you like to 'buy airtime', 'buy bundle', or 'file a complaint'?", "expects": "intent", "transitions": {"buy_airtime": "ask_airtime_amount", "buy_bundle": "ask_bundle_type", "complaint": "ask_complaint"}, }, "ask_airtime_amount": { "ha": "Nawa na airtime kake son saya? Misali, Naira ɗari ko dubu.", "en": "How much airtime? For example 100 or 1000 Naira.", "expects": "amount", "transitions": {"provide_amount": "airtime_done"}, }, "airtime_done": { "ha": "An kara Naira {amount} na airtime. Ma'aunin ka sabo shine Naira {airtime_balance}.", "en": "Loaded {amount} Naira airtime. Your new balance is {airtime_balance} Naira.", "action": "buy_airtime", "expects": None, "terminal": True, }, "ask_bundle_type": { "ha": "Wane irin bundle? Muna da 'rana', 'mako', ko 'wata'.", "en": "Which bundle type? 'day', 'week', or 'month'.", "expects": "bundle", "transitions": {"provide_bundle": "bundle_done"}, }, "bundle_done": { "ha": "An kunna bundle na '{bundle}'. Za ka iya yin amfani da shi yanzu.", "en": "Your '{bundle}' bundle is active. You can use it now.", "action": "buy_bundle", "expects": None, "terminal": True, }, "ask_complaint": { "ha": "Me ya faru? Ka bayyana matsalar da kake fuskanta.", "en": "What happened? Please describe the issue.", "expects": "text", "transitions": {"provide_text": "escalate"}, }, "escalate": { "ha": "Nagode. An buɗe lamba {ticket_id}. Zan juya ka ga wakili na mutum yanzu.", "en": "Thank you. Ticket {ticket_id} opened. I'll transfer you to a human agent now.", "action": "file_complaint", "expects": None, "terminal": True, "escalate": True, }, }, }, "ecommerce": { "name": "PlotWeaver Delivery", "states": { "greeting": { "ha": "Sannu! Wannan shine PlotWeaver Delivery. Kana son 'bincika oda', 'sake tsara lokaci', ko 'mayar da kaya'?", "en": "Hello! This is PlotWeaver Delivery. Would you like to 'check order', 'reschedule', or 'return'?", "expects": "intent", "transitions": {"check_order": "ask_order_id", "reschedule": "ask_order_id_reschedule", "return_item": "ask_order_id_return"}, }, "ask_order_id": { "ha": "Ka faɗi lambar oda naka.", "en": "Say your order number.", "expects": "digits", "transitions": {"provide_digits": "order_status"}, }, "order_status": { "ha": "Oda {order_id} yana kan hanya. Za a isar gobe da yamma.", "en": "Order {order_id} is on the way. It will be delivered tomorrow evening.", "action": "check_order", "expects": None, "terminal": True, }, "ask_order_id_reschedule": { "ha": "Ka faɗi lambar oda da kake son sake tsarawa.", "en": "Say the order number you want to reschedule.", "expects": "digits", "transitions": {"provide_digits": "ask_new_date"}, }, "ask_new_date": { "ha": "Wace rana kake so? Misali 'jumma'a' ko 'asabar'.", "en": "Which day? For example 'Friday' or 'Saturday'.", "expects": "date", "transitions": {"provide_date": "reschedule_done"}, }, "reschedule_done": { "ha": "An sake tsara isar oda {order_id} zuwa {date}. Za ka sami SMS na tabbatarwa.", "en": "Order {order_id} rescheduled to {date}. You'll receive a confirmation SMS.", "action": "reschedule", "expects": None, "terminal": True, }, "ask_order_id_return": { "ha": "Ka faɗi lambar oda da kake son mayarwa.", "en": "Say the order number you want to return.", "expects": "digits", "transitions": {"provide_digits": "return_reason"}, }, "return_reason": { "ha": "Me ya sa kake son mayarwa?", "en": "Why do you want to return it?", "expects": "text", "transitions": {"provide_reason": "return_done"}, }, "return_done": { "ha": "An karɓi buƙatarka (lamba {return_ref}). Wakili zai tattara kaya a gobe.", "en": "Your request is received (ref {return_ref}). An agent will collect the item tomorrow.", "action": "return_item", "expects": None, "terminal": True, }, }, }, } # Vertical-specific fallback prompts — spoken when the user asks something # out of scope. FALLBACK_PROMPTS = { "bank": { "ha": "Ban fahimci tambayarka ba. A nan, zan iya taimake ka da 'duba ma'auni', 'toshe kati', ko 'canjin kuɗi'. Don wasu tambayoyi, ka ce 'wakili' don yin magana da mutum.", "en": "I didn't understand your question. Here I can help with 'check balance', 'block card', or 'transfer money'. For other questions, say 'agent' to speak with a person.", }, "telecom": { "ha": "Ban fahimci tambayarka ba. Zan iya taimake ka da 'saya airtime', 'saya bundle', ko 'yin korafi'. Don wasu tambayoyi, ka ce 'wakili' don yin magana da mutum.", "en": "I didn't understand your question. I can help with 'buy airtime', 'buy bundle', or 'file a complaint'. For other questions, say 'agent' to speak with a person.", }, "ecommerce": { "ha": "Ban fahimci tambayarka ba. Zan iya taimake ka da 'bincika oda', 'sake tsara lokaci', ko 'mayar da kaya'. Don wasu tambayoyi, ka ce 'wakili' don yin magana da mutum.", "en": "I didn't understand your question. I can help with 'check order', 'reschedule delivery', or 'return an item'. For other questions, say 'agent' to speak with a person.", }, } # After this many consecutive 'unknown' intents, auto-escalate to human. MAX_CONSECUTIVE_UNKNOWNS = 2 class _SafeDict(dict): """format_map helper: leaves unknown placeholders intact instead of raising, so a missing slot surfaces as a visible literal during testing.""" def __missing__(self, key): return "{" + key + "}" def get_prompt(vertical: str, state_name: str) -> dict: """Raw prompt (may contain {slot} placeholders). Use render_prompt to fill.""" if state_name == "escalate_virtual": return {"ha": "Zan juya ka ga wakili na mutum yanzu. Ka jira ɗan lokaci.", "en": "I'll transfer you to a human agent now. Please hold."} if state_name == "exit": return {"ha": "Nagode. Sai watan.", "en": "Thank you. Goodbye."} if state_name == "fallback": return FALLBACK_PROMPTS.get(vertical, FALLBACK_PROMPTS["bank"]) s = SCENARIOS[vertical]["states"].get(state_name) if not s: return {"ha": "Ban fahimci abin da ka ce ba.", "en": "I didn't understand."} return {"ha": s["ha"], "en": s["en"]} def render_prompt(vertical: str, state_name: str, slots: dict) -> dict: """Prompt with {slot} placeholders interpolated from slots.""" p = get_prompt(vertical, state_name) safe = _SafeDict(slots or {}) return {k: v.format_map(safe) for k, v in p.items()} def get_expected_slot(vertical: str, state_name: str) -> Optional[str]: if state_name == "fallback": # Fallback accepts any intent — user might repeat, rephrase, or escalate return "intent" s = SCENARIOS[vertical]["states"].get(state_name) return s.get("expects") if s else None def _maybe_run_action(backend: Backend, vertical: str, state_name: str, state: "DialogueState") -> None: """If the target state declares an action, call the backend and merge the returned (language-neutral) values into slots. Never raises into the FSM.""" spec = SCENARIOS[vertical]["states"].get(state_name, {}) action = spec.get("action") if not action or backend is None: return method = getattr(backend, action, None) if method is None: logger.warning("Backend has no action %r", action) return try: result = method(dict(state.slots)) if result: state.slots.update(result) except Exception as e: logger.warning("Backend action %r failed: %s", action, e) def transition(state: DialogueState, intent: str, entities: dict, backend: Optional[Backend] = None) -> DialogueState: if backend is None: backend = _DEFAULT_BACKEND state.turn_count += 1 for k, v in entities.items(): state.slots[k] = v # Explicit human-agent request or too many turns → escalate if intent == "human_agent" or state.turn_count > 12: state.current_state = "escalate_virtual" state.escalate_to_human = True return state # Unknown intent handling: route to fallback, track consecutive count, # auto-escalate if user keeps asking out-of-scope things. if intent == "unknown": state.consecutive_unknowns += 1 if state.consecutive_unknowns >= MAX_CONSECUTIVE_UNKNOWNS: state.current_state = "escalate_virtual" state.escalate_to_human = True return state if state.current_state != "fallback": state.state_before_fallback = state.current_state state.current_state = "fallback" return state # Recognized intent → reset the unknown counter state.consecutive_unknowns = 0 # If we're in fallback and the user now says something recognized, # resume from the state we were in before falling back if state.current_state == "fallback" and state.state_before_fallback: resume_state = state.state_before_fallback state.state_before_fallback = None state.current_state = resume_state current = SCENARIOS[state.vertical]["states"].get(state.current_state) if not current: state.current_state = "greeting" current = SCENARIOS[state.vertical]["states"]["greeting"] # Try transition from the current state first next_state = current.get("transitions", {}).get(intent) # If current state has no transition for this intent, but GREETING does, # treat this as the user pivoting to a new top-level intent (e.g. midway # through balance check they say "transfer money" instead). Restart flow. if not next_state: greeting = SCENARIOS[state.vertical]["states"]["greeting"] pivot_state = greeting.get("transitions", {}).get(intent) if pivot_state: state.slots = {} # Reset slots when starting a new flow state.state_before_fallback = None next_state = pivot_state if next_state: state.current_state = next_state target = SCENARIOS[state.vertical]["states"].get(next_state, {}) # Run any backend action attached to the state we just entered, so its # results are in slots before the prompt is rendered. _maybe_run_action(backend, state.vertical, next_state, state) if target.get("escalate"): state.escalate_to_human = True else: # Intent recognized but neither current state nor greeting has a # transition for it. Route to fallback. state.consecutive_unknowns += 1 if state.consecutive_unknowns >= MAX_CONSECUTIVE_UNKNOWNS: state.current_state = "escalate_virtual" state.escalate_to_human = True return state if state.current_state != "fallback": state.state_before_fallback = state.current_state state.current_state = "fallback" return state