from typing import Dict, List GLOBAL_INTENTS = [ "restart", "new_topic", "complaint", "direct_support", "courses_info", "children_courses", "adults_courses", "new_student", "current_student", "unclear", ] STATE_ALLOWED_INTENTS: Dict[str, List[str]] = { "WAITING_MAIN_MENU": [ "new_student", "current_student", "courses_info", "complaint", "direct_support", ], "WAITING_USER_TYPE": [ "new_student", "current_student", ], "WAITING_AUDIENCE": [ "adults", "children", ], "WAITING_PRIOR_STUDY": [ "prior_study_yes", "prior_study_no", ], "WAITING_BEGINNER_SCHEDULE_CHOICE": [ "confirm_schedule_reviewed", "proceed_booking", "switch_to_prior_study_true", "switch_to_prior_study_false", "support_needed", ], "WAITING_PDF_102_CONFIRMATION": [ "confirm_pdf_reviewed", "switch_to_prior_study_true", "switch_to_prior_study_false", "support_needed", ], "WAITING_PLACEMENT_TEST_CONFIRMATION": [ "confirm_placement_test_reviewed", "switch_to_prior_study_true", "switch_to_prior_study_false", "support_needed", ], "WAITING_CURRENT_STUDENT_ACTION": [ "current_student_support", "current_student_next_level", ], "WAITING_SUPPORT_QUESTION": [ "support_question_text", ], "WAITING_LEVEL_SELECTION": [ "level_selected", "support_needed", ], "WAITING_PAYMENT_METHOD": [ "payment_method_selected", "support_needed", ], "WAITING_COMPLAINT_FORM": [ "complaint_form_submitted", ], "HANDOFF_DONE": [ "thanks", "courses_info", "children_courses", "adults_courses", "new_student", "current_student", "direct_support", "complaint", "restart", "new_topic", ], } INTENT_DESCRIPTIONS: Dict[str, str] = { "restart": "The user wants to restart the conversation from the beginning.", "new_topic": "The user explicitly wants to ask about something else or start a new topic.", "complaint": "The user is making or asking to make a complaint.", "direct_support": "The user wants to talk to customer support or service directly.", "courses_info": "The user is asking generally about available courses.", "children_courses": "The user is asking about children courses.", "adults_courses": "The user is asking about adult courses.", "new_student": "The user indicates they are a new student.", "current_student": "The user indicates they are a current student.", "adults": "Direct answer: adults.", "children": "Direct answer: children.", "prior_study_yes": "Direct answer: user studied German before.", "prior_study_no": "Direct answer: user did not study German before.", "confirm_schedule_reviewed": "The user confirms they reviewed the beginner schedule.", "proceed_booking": "The user wants to proceed with booking.", "switch_to_prior_study_true": "The user changes to the path where they studied German before.", "switch_to_prior_study_false": "The user changes to the beginner path.", "support_needed": "The user needs help or wants support within the current topic.", "confirm_pdf_reviewed": "The user confirms they reviewed the PDF/details file.", "confirm_placement_test_reviewed": "The user confirms they reviewed placement test details.", "current_student_support": "Current student wants support or has a question.", "current_student_next_level": "Current student wants to book the next level.", "support_question_text": "The message itself is the support question text.", "level_selected": "The user selected a course level.", "payment_method_selected": "The user selected a payment method.", "complaint_form_submitted": "The user says they submitted the complaint form.", "thanks": "The user is thanking the bot.", "unclear": "The message is unclear and should not be confidently routed.", } def get_allowed_intents_for_state(state: str) -> List[str]: state_specific = STATE_ALLOWED_INTENTS.get(state, []) final = [] for item in GLOBAL_INTENTS + state_specific: if item not in final: final.append(item) return final def _summarize_flow_data(flow_data: dict) -> str: if not flow_data: return "No extra context." fields = [] for key in [ "customer_type", "audience", "prior_study", "selected_level", "payment_method", "gender", ]: if key in flow_data and flow_data[key] is not None: fields.append(f"{key}={flow_data[key]}") return ", ".join(fields) if fields else "No extra context." def build_system_prompt(state: str, flow_data: dict, allowed_intents: List[str]) -> str: descriptions = [] for intent in allowed_intents: desc = INTENT_DESCRIPTIONS.get(intent, "") descriptions.append(f"- {intent}: {desc}") description_block = "\n".join(descriptions) context_summary = _summarize_flow_data(flow_data) return f""" You are a state-aware intent classifier for an Arabic WhatsApp bot. Your task: - Read the user's Arabic message. - Understand the CURRENT conversation state. - Choose exactly ONE intent label from the allowed list. - Return ONLY the label. - Do not explain. - Do not add punctuation. - Do not add extra words. Current state: {state} Known context: {context_summary} Allowed intent labels: {", ".join(allowed_intents)} Intent descriptions: {description_block} Decision rules: - If the message is a direct answer to the current question, choose the direct answer label. - If the user changes path but stays in the same broad topic, choose the appropriate state-switch label. - If the user changes to a new topic, choose the correct topic-switch label. - If you are not confident, return: unclear Return only one label from the allowed list. """.strip()