Spaces:
Runtime error
Runtime error
| import typing | |
| from tools import nlu_tool, scheduler, requests_store | |
| from app import propose_slots, notify_operator | |
| import yaml | |
| import os | |
| PROMPTS_PATH = 'prompts.yaml' | |
| def _load_prompts(): | |
| with open(PROMPTS_PATH, 'r') as f: | |
| txt = f.read() | |
| # If file contains a fenced code block (```yaml ... ```), strip fences | |
| stripped = txt | |
| lines = txt.splitlines() | |
| if lines and lines[0].strip().startswith("```") and lines[-1].strip().startswith("```"): | |
| stripped = "\n".join(lines[1:-1]) | |
| data = yaml.safe_load(stripped) | |
| # recursively search for collection_assistant | |
| def find_collection(obj): | |
| if isinstance(obj, dict): | |
| if 'collection_assistant' in obj: | |
| return obj['collection_assistant'] | |
| for v in obj.values(): | |
| found = find_collection(v) | |
| if found is not None: | |
| return found | |
| elif isinstance(obj, list): | |
| for item in obj: | |
| found = find_collection(item) | |
| if found is not None: | |
| return found | |
| return None | |
| found = find_collection(data) | |
| if found is not None: | |
| # If the value is a YAML block string (from |-) parse it into a dict | |
| if isinstance(found, str): | |
| try: | |
| inner = yaml.safe_load(found) | |
| if isinstance(inner, dict): | |
| return inner | |
| except Exception: | |
| pass | |
| return found | |
| raise KeyError('collection_assistant') | |
| def handle_message(message: str, context: dict = None) -> dict: | |
| """High-level handler that accepts a user message and performs: | |
| - NLU (intent + slots) | |
| - Minimal slot-filling for payment_commitment | |
| - Confirmation and request creation | |
| - Scheduling flow for human-operator requests | |
| Returns a dict with response text and any created request record. | |
| """ | |
| prompts = _load_prompts() | |
| nlu = nlu_tool.extract_intent_and_slots(message) | |
| intent = nlu.get('intent') | |
| slots = nlu.get('slots', {}) | |
| response = "" | |
| created = None | |
| if intent == 'payment_commitment': | |
| # find payment_commitment intent meta | |
| slot_meta = next((x for x in prompts.get('intents', []) if x.get('name') == 'payment_commitment'), None) | |
| # ensure required slots: customer_name, account_number, amount, date_by_when | |
| missing = [] | |
| for s in ['customer_name', 'account_number', 'amount', 'date_by_when']: | |
| if s not in slots or not slots[s]: | |
| missing.append(s) | |
| if missing: | |
| # return prompt for the first missing slot (simple) | |
| if slot_meta: | |
| slot_prompt = next((x.get('prompt') for x in slot_meta.get('slots', []) if x.get('id') == missing[0]), None) | |
| response = slot_prompt or f"Please provide {missing[0]}" | |
| else: | |
| response = f"Please provide {missing[0]}" | |
| return {'response': response, 'request': None} | |
| # All required slots present -> confirm and create request | |
| tpl = slot_meta.get('confirmation_template') if slot_meta else "Thank you. I recorded your commitment." | |
| response = tpl.format(customer_name=slots.get('customer_name', '[unknown]'), account_number=slots.get('account_number', '[unknown]'), amount=slots.get('amount'), date_by_when=slots.get('date_by_when')) | |
| handoff_tpl = slot_meta.get('handoff_payload_template') if slot_meta else None | |
| if handoff_tpl: | |
| handoff_json = handoff_tpl.format(customer_name=slots.get('customer_name', ''), account_number=slots.get('account_number', ''), amount=slots.get('amount', ''), date_by_when=slots.get('date_by_when', ''), contact_preference=slots.get('contact_preference', ''), nlu_confidence=nlu.get('nlu_confidence', 0)) | |
| else: | |
| handoff_json = str({'type': 'payment_commitment', 'slots': slots}) | |
| # persist | |
| created = requests_store.create_request({'raw_payload': handoff_json}) | |
| # notify operator stub | |
| notify_operator(f"HANDOFF: {handoff_json}") | |
| return {'response': response, 'request': created} | |
| if intent == 'request_human_operator': | |
| # find intent meta | |
| req_meta = next((x for x in prompts.get('intents', []) if x.get('name') == 'request_human_operator'), None) | |
| preferred = [] | |
| if 'preferred_windows' in slots and slots['preferred_windows']: | |
| preferred = [{'start': slots['preferred_windows'], 'end': slots['preferred_windows']}] | |
| candidates = propose_slots(preferred) | |
| if req_meta: | |
| response = req_meta.get('propose_slots_template', '').format(slot1_local=candidates[0] if len(candidates) > 0 else '', slot2_local=candidates[1] if len(candidates) > 1 else '', slot3_local=candidates[2] if len(candidates) > 2 else '') | |
| else: | |
| response = f"I can connect you at: {', '.join(candidates)}" | |
| return {'response': response, 'request': None} | |
| # fallback | |
| response = prompts.get('behaviors', {}).get('fallback', {}).get('prompt', "I didn't understand. Can you rephrase?") | |
| if not response: | |
| response = "I didn't understand. Can you rephrase?" | |
| return {'response': response, 'request': None} | |