Spaces:
Runtime error
Runtime error
File size: 5,269 Bytes
3696fe2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
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}
|