""" conversation_manager.py ------------------------- A state-machine chatbot that: 1. Gets informed consent and explains the tool is a screening aid, not a diagnostic instrument. 2. Builds light rapport (small talk / open question) before screening. 3. Weaves PHQ-9, GAD-7, and PSS-10 items into natural conversational turns (not a rigid form) — each question is asked in plain language, the user's free-text answer is analyzed with local NLP AND mapped to the instrument's 0-3 / 0-4 response scale via lightweight heuristics. 4. Runs a safety/crisis check on every single user turn — before anything else happens with that turn. 5. Ends with open reflection questions, then produces a ScreeningResult. NOTE on response mapping: mapping free text ("yeah, most days honestly") to a validated Likert value is inherently approximate. This module uses transparent keyword heuristics and always re-confirms with the user when uncertain, rather than silently guessing. A production system should strongly consider a light confirmation UI step ("So that's 'more than half the days' — correct?") which this implementation includes. """ import re import uuid from typing import Dict, List, Optional, Tuple from . import questionnaires as q from .models import SessionState, SessionStage, Speaker from .nlp_analysis import TextAnalyzer from .safety import CrisisDetector from .risk_engine import RiskEngine, ScreeningResult CONSENT_TEXT = ( "Hi, I'm MindScreen AI. I'm a research/educational chatbot that can walk " "through some questions used in wellbeing screening (covering mood, " "anxiety, and stress). \n\n" "A few important things before we start:\n" " • I am not a licensed clinician and this is not a diagnosis — just a " "screening aid.\n" " • Your responses will be used for research purposes as described in " "the study information you were given.\n" " • If at any point you mention thoughts of self-harm or being in " "danger, I'll pause the screening and share crisis resources instead.\n\n" "Would you like to continue? (yes/no)" ) REFLECTION_PROMPTS = [ "Before we wrap up — is there anything about how you've been feeling " "lately that we haven't touched on yet?", "Thanks for sharing all that. How are you feeling right now, having " "talked through this?", ] _YES_RE = re.compile(r"\b(yes|yeah|yep|sure|ok|okay|continue|i agree|agreed)\b", re.I) _NO_RE = re.compile(r"\b(no|nope|not now|stop|decline)\b", re.I) # Heuristic keyword -> scale value mapping, checked in order (0..3 scale, and # separately 0..4 scale for PSS-10). Order matters: more specific phrases first. _SCALE_0_3_KEYWORDS: List[Tuple[re.Pattern, int]] = [ (re.compile(r"\bnearly every day|almost every day|every single day|constantly\b", re.I), 3), (re.compile(r"\bmore than half|most days|frequently|a lot lately\b", re.I), 2), (re.compile(r"\bseveral days|sometimes|a few times|now and then|occasionally\b", re.I), 1), (re.compile(r"\bnot at all|never|none|not really|no\b", re.I), 0), ] _SCALE_0_4_KEYWORDS: List[Tuple[re.Pattern, int]] = [ (re.compile(r"\bvery often|constantly|all the time\b", re.I), 4), (re.compile(r"\bfairly often|quite often|often\b", re.I), 3), (re.compile(r"\bsometimes|occasionally\b", re.I), 2), (re.compile(r"\balmost never|rarely|hardly ever\b", re.I), 1), (re.compile(r"\bnever|not at all|none\b", re.I), 0), ] def _map_free_text_to_scale(text: str, keyword_table: List[Tuple[re.Pattern, int]]) -> Optional[int]: for pattern, value in keyword_table: if pattern.search(text): return value return None class ConversationManager: def __init__(self, participant_id: str, risk_engine: Optional[RiskEngine] = None): self.state = SessionState(session_id=str(uuid.uuid4()), participant_id=participant_id) self.analyzer = TextAnalyzer() self.safety = CrisisDetector() self.risk_engine = risk_engine or RiskEngine() self._text_features_log = [] # List[TextFeatures] across the whole session # ------------------------------------------------------------------ # def start(self) -> str: self.state.add_message(Speaker.BOT, CONSENT_TEXT) return CONSENT_TEXT # ------------------------------------------------------------------ # def handle_user_turn(self, user_text: str) -> Dict: """Process one user message. Returns a dict: {"bot_message": str, "stage": SessionStage, "safety_triggered": bool, "result": Optional[ScreeningResult]} """ features = self.analyzer.analyze(user_text) self._text_features_log.append(features) self.state.add_message(Speaker.USER, user_text, nlp_features=features.to_dict()) # Safety check happens on EVERY turn, before anything else. safety_result = self.safety.check_text(features) if safety_result.triggered: self.state.stage = SessionStage.SAFETY_PAUSED self.state.safety_paused = True bot_msg = safety_result.message self.state.add_message(Speaker.BOT, bot_msg) return { "bot_message": bot_msg, "stage": self.state.stage, "safety_triggered": True, "result": None, } if self.state.stage == SessionStage.SAFETY_PAUSED: # Once paused, we do not resume automated screening in this demo. bot_msg = ( "I've paused the automated screening for your safety. If " "you'd like to talk to a professional now, please use one of " "the resources shared above. I'm here if you want to talk " "about something else." ) self.state.add_message(Speaker.BOT, bot_msg) return {"bot_message": bot_msg, "stage": self.state.stage, "safety_triggered": False, "result": None} handler = { SessionStage.INTRO_CONSENT: self._handle_consent, SessionStage.RAPPORT_BUILDING: self._handle_rapport, SessionStage.PHQ9: lambda t: self._handle_questionnaire_item( t, q.PHQ9_ITEMS, self.state.phq9_responses, _SCALE_0_3_KEYWORDS, q.FREQUENCY_LABELS_0_3, SessionStage.GAD7, "GAD7 intro"), SessionStage.GAD7: lambda t: self._handle_questionnaire_item( t, q.GAD7_ITEMS, self.state.gad7_responses, _SCALE_0_3_KEYWORDS, q.FREQUENCY_LABELS_0_3, SessionStage.PSS10, "PSS10 intro"), SessionStage.PSS10: lambda t: self._handle_questionnaire_item( t, q.PSS10_ITEMS, self.state.pss10_responses, _SCALE_0_4_KEYWORDS, q.FREQUENCY_LABELS_0_4, SessionStage.OPEN_REFLECTION, "reflection intro"), SessionStage.OPEN_REFLECTION: self._handle_reflection, }.get(self.state.stage) bot_msg = handler(user_text) if handler else "Thanks — this session is complete." self.state.add_message(Speaker.BOT, bot_msg) result = None if self.state.stage == SessionStage.COMPLETE: result = self._finalize() return {"bot_message": bot_msg, "stage": self.state.stage, "safety_triggered": False, "result": result} # ------------------------------------------------------------------ # def _handle_consent(self, user_text: str) -> str: if _YES_RE.search(user_text) and not _NO_RE.search(user_text): self.state.consent_given = True self.state.stage = SessionStage.RAPPORT_BUILDING return ( "Great, thank you. Before we get into specifics — how has " "your week been going overall?" ) self.state.stage = SessionStage.COMPLETE return "No problem — we won't continue with screening. Take care." def _handle_rapport(self, user_text: str) -> str: self.state.stage = SessionStage.PHQ9 self.state.current_item_index = 0 return self._phrase_item_question(q.PHQ9_ITEMS[0], first=True) def _handle_questionnaire_item( self, user_text: str, items: List[str], responses: List[int], keyword_table, labels: Dict[int, str], next_stage: SessionStage, _unused: str, ) -> str: value = _map_free_text_to_scale(user_text, keyword_table) if value is None: # Ask for clarification rather than guessing silently. options = ", ".join(f"'{label}'" for label in labels.values()) return ( "Just to make sure I capture that accurately — would you say " f"that's been: {options}?" ) responses.append(value) idx = len(responses) if idx < len(items): return self._phrase_item_question(items[idx], first=False) else: self.state.stage = next_stage self.state.current_item_index = 0 if next_stage == SessionStage.GAD7: return ( "Thanks for being open about that. Now shifting gears a " "little — I'd like to ask about anxiety and worry. " + self._phrase_item_question(q.GAD7_ITEMS[0], first=True) ) elif next_stage == SessionStage.PSS10: return ( "Appreciate that. Last section — this is about how " "stressed or in-control you've been feeling. " + self._phrase_item_question(q.PSS10_ITEMS[0], first=True) ) elif next_stage == SessionStage.OPEN_REFLECTION: return REFLECTION_PROMPTS[0] return "Thanks." def _phrase_item_question(self, item_text: str, first: bool) -> str: lead_in = "So, " if not first else "" return f"{lead_in}over the last two weeks, how often have you been bothered by: {item_text.lower()}?" def _handle_reflection(self, user_text: str) -> str: # One extra reflection question, then finish. if self.state.current_item_index == 0: self.state.current_item_index = 1 return REFLECTION_PROMPTS[1] self.state.stage = SessionStage.COMPLETE return ( "Thank you for taking the time to talk this through with me. " "I'm putting together your screening summary now — remember, " "this is a screening aid, not a diagnosis. Please share the " "summary with a licensed professional if you'd like to discuss " "it further." ) # ------------------------------------------------------------------ # def _finalize(self) -> Optional[ScreeningResult]: from datetime import datetime, timezone self.state.completed_at = datetime.now(timezone.utc) if (len(self.state.phq9_responses) < 9 or len(self.state.gad7_responses) < 7 or len(self.state.pss10_responses) < 10): return None # incomplete questionnaire data — cannot score reliably phq9_result = q.score_phq9(self.state.phq9_responses) gad7_result = q.score_gad7(self.state.gad7_responses) pss10_result = q.score_pss10(self.state.pss10_responses) # Safety net: even if free-text never tripped the crisis detector, # always re-check the PHQ-9 item 9 response directly. item9_check = self.safety.check_phq9_item9(self.state.phq9_responses[q.PHQ9_SUICIDE_ITEM_INDEX]) if item9_check.triggered: self.state.stage = SessionStage.SAFETY_PAUSED self.state.safety_paused = True nlp_summary = self.analyzer.aggregate(self._text_features_log) return self.risk_engine.assess( phq9=phq9_result, gad7=gad7_result, pss10=pss10_result, nlp_summary=nlp_summary, items_answered=self.state.total_items_answered(), items_total=self.state.total_items_expected(), )