"""EmpathRAG Core runtime. One guarded conversational RAG interface used by the demo and evaluation. """ from __future__ import annotations from dataclasses import asdict, dataclass import os from pathlib import Path from typing import Literal import sqlite3 import time from .ml_router import MLRouter from .output_guard import validate_output from .response_planner import ( CLARIFY, INTERNATIONAL_SOURCE_HINT, build_response_plan, classify_intl_topic, decide_stage, has_international_concern, render_crisis_response, render_intl_factual_offer, ) from .rephraser import RephraseResult, ResponseRephraser from .safety_policy import SafetyLevel, SafetyTriagePolicy from .service_graph import match_services from .v2_schema import SafetyTier, SupportRoute, classify_route, map_safety_level AudienceMode = Literal["student", "helping_friend"] BackendMode = Literal["demo_fast", "hybrid_ml", "real_llm"] @dataclass class EmpathRAGResult: response: str route_label: str safety_tier: str should_intercept: bool retrieved_sources: list[dict] recommended_action: str output_guard: dict trajectory_state: str latency_ms: dict classifier_confidence: dict retrieval_mode: str safety_precheck: dict safety_explanation: dict safety_reason: str escalation_reason: str retrieval_corpus: str emotion_name: str = "neutral" crisis: bool = False crisis_confidence: float = 0.0 retrieved_chunks: list[str] | None = None international_concern: bool = False conversation_stage: str = "offer" turn_index: int = 1 intl_topic: str = "" rephraser_provider: str = "deterministic" rephraser_used_llm: bool = False rephraser_latency_ms: float = 0.0 rephraser_last_error: str = "" def to_dict(self) -> dict: row = asdict(self) row["retrieved_chunks"] = row["retrieved_chunks"] or [] row["safety_level"] = row["safety_tier"] row["route"] = row["route_label"] row["latency_ms"] = self.latency_ms return row @dataclass class _TurnPlan: """Everything ``_plan_turn`` decides before the rephrase step. Shared between sync ``run_turn`` and ``run_turn_streaming`` so the pre-rephrase planning logic isn't duplicated across the two paths. """ message: str session_id: str audience_mode: str backend_mode: str turn_index: int template_response: str retrieved: list[dict] recommended_action: str route_label: str safety_tier: SafetyTier safety_reason: str stage: str intl_concern: bool intl_topic: str should_intercept: bool retrieval_mode: str latency: dict t_total: float stage1_level: str stage1_reason: str stage1_should_intercept: bool ml_prediction: object guardrail_info: dict escalation_reason: str class EmpathRAGCore: # Safety layers that can be selectively disabled for ablation evaluation. # See eval/run_ablation_eval.py for the harness that uses this. VALID_DISABLED_LAYERS = frozenset({ "stage1_precheck", # Stage-1 lexical safety policy precheck "output_guard", # validate_output at OFFER stage "rephrase_safety", # verify_rephrased_safety on LLM output "registry_filter", # resource registry + retrieval filter }) def __init__( self, curated_db_path: Path | str = Path("data/curated/indexes/metadata_curated.db"), retrieval_corpus: str = "curated_support", top_k: int = 5, router_model_dir: Path | str = Path("models/router"), ml_confidence_threshold: float = 0.35, use_model_guardrail: bool | None = None, compute_ig_on_intercept: bool | None = None, guardrail_threshold: float = 0.5, disable_layers: set[str] | None = None, ): self.disable_layers = set(disable_layers or set()) invalid = self.disable_layers - self.VALID_DISABLED_LAYERS if invalid: raise ValueError( f"Unknown disable_layers: {invalid}. " f"Valid: {sorted(self.VALID_DISABLED_LAYERS)}" ) self.curated_db_path = Path(curated_db_path) self.retrieval_corpus = "curated_support" if self.curated_db_path.exists() else retrieval_corpus self.top_k = top_k self.safety_policy = SafetyTriagePolicy() self.ml_router = MLRouter(Path(router_model_dir), min_confidence=ml_confidence_threshold) self.use_model_guardrail = _env_flag("EMPATHRAG_CORE_USE_GUARDRAIL") if use_model_guardrail is None else use_model_guardrail self.compute_ig_on_intercept = _env_flag("EMPATHRAG_CORE_COMPUTE_IG") if compute_ig_on_intercept is None else compute_ig_on_intercept self.guardrail_threshold = guardrail_threshold self.rephraser = ResponseRephraser() self._guardrail = None self._guardrail_error = "" self.tier_history: dict[str, list[str]] = {} self.locked_sessions: dict[str, str] = {} # Session-scoped sticky flags so context detected on one turn carries # forward (international concern, last specific route) instead of being # re-derived from each message in isolation. self.session_intl_flag: dict[str, bool] = {} # Turns since the user last mentioned F-1 / visa / international # status keywords. ISSS auto-surfacing decays after 2 silent turns # so the conversation can fully shift topic without ISSS hijacking # every subsequent response. self.session_turns_since_intl: dict[str, int] = {} self.session_last_specific_route: dict[str, str] = {} # Rolling per-session message log so the rephraser has continuity. # Only the last few user-assistant pairs are passed to the LLM; full # history is kept here for diagnostics. Each entry: {"role", "content"}. self.session_message_history: dict[str, list[dict]] = {} def reset_session(self, session_id: str | None = None) -> None: if session_id: self.tier_history.pop(session_id, None) self.locked_sessions.pop(session_id, None) self.session_intl_flag.pop(session_id, None) self.session_turns_since_intl.pop(session_id, None) self.session_last_specific_route.pop(session_id, None) self.session_message_history.pop(session_id, None) else: self.tier_history.clear() self.locked_sessions.clear() self.session_intl_flag.clear() self.session_turns_since_intl.clear() self.session_last_specific_route.clear() self.session_message_history.clear() # Input length cap: defends against (1) accidental wall-of-text pastes # ballooning LLM cost and planner regex matching, (2) deliberate DoS # via runaway-length payloads. Configurable via env so larger contexts # can be unlocked for evaluation runs that legitimately need them. MAX_USER_MESSAGE_CHARS = int(os.getenv("EMPATHRAG_MAX_USER_MESSAGE_CHARS", "2000")) def run_turn( self, message: str, session_id: str, audience_mode: AudienceMode = "student", resource_profile: str = "umd", backend_mode: BackendMode = "hybrid_ml", turn_index: int | None = None, ) -> EmpathRAGResult: plan = self._plan_turn(message, session_id, audience_mode, backend_mode, turn_index) if plan.should_intercept: return self._finalize_turn(plan, plan.template_response, None) rephrase_result = self.rephraser.rephrase( user_message=message, template_response=plan.template_response, retrieved_sources=plan.retrieved, recommended_action=plan.recommended_action, history=self._recent_history(session_id), skip_safety_check="rephrase_safety" in self.disable_layers, ) return self._finalize_turn(plan, rephrase_result.response, rephrase_result) def run_turn_streaming( self, message: str, session_id: str, audience_mode: AudienceMode = "student", resource_profile: str = "umd", backend_mode: BackendMode = "hybrid_ml", turn_index: int | None = None, ): """Generator variant of :meth:`run_turn`. Yields: * ``("token", accumulated_text)`` for each streamed chunk * ``("done", EmpathRAGResult)`` exactly once at the end Crisis interception, deterministic mode, and safety-rejected rephrases all collapse to a single deterministic ``("token", final_text)`` plus a ``("done", ...)``. Real Groq/Anthropic streaming only kicks in when ``EMPATHRAG_REPHRASER_ENABLED=1`` and the route is non-crisis. """ plan = self._plan_turn(message, session_id, audience_mode, backend_mode, turn_index) if plan.should_intercept: result = self._finalize_turn(plan, plan.template_response, None) yield ("token", result.response) yield ("done", result) return history = self._recent_history(session_id) if not self.rephraser.enabled: # Deterministic mode: no LLM, no real streaming. Hand the template # through and let the demo's word-chunk reveal layer fake-stream it # the way it always has. rephrase_result = self.rephraser.rephrase( user_message=message, template_response=plan.template_response, retrieved_sources=plan.retrieved, recommended_action=plan.recommended_action, history=history, skip_safety_check="rephrase_safety" in self.disable_layers, ) result = self._finalize_turn(plan, rephrase_result.response, rephrase_result) yield ("token", result.response) yield ("done", result) return accumulated = "" rephrase_result: RephraseResult | None = None for event in self.rephraser.rephrase_streaming( user_message=message, template_response=plan.template_response, retrieved_sources=plan.retrieved, recommended_action=plan.recommended_action, history=history, skip_safety_check="rephrase_safety" in self.disable_layers, ): if event[0] == "chunk": _, accumulated, _provider_name = event yield ("token", accumulated) elif event[0] == "final": _, rephrase_result = event # rephrase_streaming guarantees exactly one ("final", ...). assert rephrase_result is not None result = self._finalize_turn(plan, rephrase_result.response, rephrase_result) # If the output guard corrected the response or the streamed text was # safety-rejected and replaced with the deterministic fallback, the # bubble currently shows stale text. Yield one more correction. if result.response != accumulated: yield ("token", result.response) yield ("done", result) def _recent_history(self, session_id: str, max_messages: int = 4) -> list[dict]: """Return the last ``max_messages`` entries from this session's message history. Used to give the rephraser continuity across turns without leaking older context the planner has already moved past.""" hist = self.session_message_history.get(session_id, []) return hist[-max_messages:] if hist else [] # ------------------------------------------------------------------ # Internal: plan + finalize # ------------------------------------------------------------------ def _plan_turn( self, message: str, session_id: str, audience_mode: AudienceMode, backend_mode: BackendMode, turn_index: int | None, ) -> _TurnPlan: if turn_index is None: turn_index = len(self.tier_history.get(session_id, [])) + 1 t_total = time.perf_counter() latency: dict[str, float] = {} # Length cap: a very long input is either an accidental wall-of-text # paste or a DoS attempt. We truncate before the planner sees it (so # regex matching + downstream LLM cost stay bounded) and surface a # short clarify-style response if the truncation actually mattered. message_was_truncated = False if message and len(message) > self.MAX_USER_MESSAGE_CHARS: message = message[: self.MAX_USER_MESSAGE_CHARS] message_was_truncated = True t0 = time.perf_counter() if "stage1_precheck" in self.disable_layers: # Ablation: pretend the message passes safety. Used to measure # what fraction of escalation catches depend on the Stage-1 # lexical check vs. downstream layers (ML router, contextual # overrides, output guard). Never set in production. from .safety_policy import SafetyDecision, SafetyLevel as _SL stage1_decision = SafetyDecision( level=_SL.PASS, confidence=0.0, reason="stage1_disabled_ablation", should_intercept=False, ) else: stage1_decision = self.safety_policy.classify(message, confidence=0.0, model_flag=False) latency["stage1_precheck_ms"] = _elapsed_ms(t0) t0 = time.perf_counter() guardrail_info = self._run_optional_guardrail(message, skip_ig=True) if not stage1_decision.should_intercept else _guardrail_skipped("stage1_intercept") latency["model_guardrail_ms"] = _elapsed_ms(t0) if guardrail_info["available"] and guardrail_info["model_flag"]: safety_decision = self.safety_policy.classify( message, confidence=float(guardrail_info["confidence"]), model_flag=True, ) else: safety_decision = stage1_decision latency["hard_safety_ms"] = latency["stage1_precheck_ms"] wellbeing_request = _wellbeing_request(message) safety_tier = map_safety_level(safety_decision.level, wellbeing_request=wellbeing_request) safety_reason = safety_decision.reason safety_tier, safety_reason = _apply_contextual_safety_overrides( message, safety_tier, safety_reason, audience_mode ) rule_decision = classify_route(message, safety_tier, audience_mode=audience_mode) t0 = time.perf_counter() ml_prediction = self.ml_router.predict(message, rule_decision.route, safety_tier) latency["classifier_ms"] = _elapsed_ms(t0) route_label = ml_prediction.route_label if backend_mode in {"hybrid_ml", "real_llm"} else rule_decision.route.value if safety_tier == SafetyTier.IMMINENT_SAFETY: route_label = SupportRoute.PEER_HELPER.value if rule_decision.route == SupportRoute.PEER_HELPER else SupportRoute.CRISIS_IMMEDIATE.value else: safety_tier = SafetyTier(ml_prediction.safety_tier) if ml_prediction.used_ml else safety_tier escalation_reason = self._update_trajectory(session_id, safety_tier.value, message) if session_id in self.locked_sessions: safety_tier = SafetyTier.IMMINENT_SAFETY safety_reason = self.locked_sessions[session_id] should_intercept = safety_decision.should_intercept or safety_tier == SafetyTier.IMMINENT_SAFETY retrieval_mode = _retrieval_mode(backend_mode, should_intercept) if should_intercept and self.compute_ig_on_intercept and self.use_model_guardrail: t0 = time.perf_counter() guardrail_info = self._run_optional_guardrail(message, skip_ig=False) latency["integrated_gradients_ms"] = _elapsed_ms(t0) t0 = time.perf_counter() if "registry_filter" in self.disable_layers: # Ablation: no resource-registry filtering, no curated retrieval. # Planner falls back to generic templates with no named UMD # resources. Used to measure how much of the system's # resource-grounding depends on the registry layer. retrieved = [] else: retrieved = self._retrieve(message, route_label, safety_tier.value, audience_mode, should_intercept) latency["retrieval_ms"] = _elapsed_ms(t0) # Cross-cutting: when an F-1 / visa / international-status worry shows up # at any point in the session, keep the ISSS surface and the # international-aware framing for subsequent turns — but decay it so # ISSS doesn't hijack every later turn after the topic has shifted. intl_now = has_international_concern(message) if intl_now: self.session_intl_flag[session_id] = True self.session_turns_since_intl[session_id] = 0 else: self.session_turns_since_intl[session_id] = ( self.session_turns_since_intl.get(session_id, 0) + 1 ) intl_session = self.session_intl_flag.get(session_id, False) # "Active" means F-1 framing should drive retrieval + sub-topic logic # this turn. After 2 turns with no F-1 keyword the flag remains True # for diagnostics but stops dominating responses. intl_active = intl_session and self.session_turns_since_intl.get(session_id, 0) <= 2 if not should_intercept and intl_active: already_has_isss = any( (s.get("source_name") or "").lower().startswith("umd international") or (s.get("service_id") or "").startswith("umd_isss") for s in retrieved ) if not already_has_isss: retrieved = [INTERNATIONAL_SOURCE_HINT] + retrieved intl_topic = "" if should_intercept: template_response = render_crisis_response( route_label, audience_mode=audience_mode, user_message=message, ) recommended_action = _recommended_action(route_label, safety_tier.value) stage = "offer" else: response_plan = build_response_plan( message, route_label, safety_tier.value, retrieved, audience_mode, international_concern_override=intl_active, ) stage = decide_stage(message, route_label, safety_tier.value, turn_index) # Only classify F-1 sub-topic when the framing is actively in play. # After 2 silent turns, decayed; don't pin the planner to OPT/RCL/etc. intl_topic = classify_intl_topic(message) if intl_active else "" # Conversational utterances (greeting / goodbye / meta) route to # short purposeful templates BEFORE the regular minimal / # incomplete checks. These aren't support content — they're # conversation openers and closers. Treating them as OFFER # responses produces jarring "let me tell you about UMD CC" # replies to a simple "hi". conv_intent = ( "" if message_was_truncated else _conversational_intent(message) ) # Length-cap / minimal / incomplete short-circuits, in priority # order. Each picks a clarify-style template instead of trying # to OFFER a full route response on insufficient or excessive # input. Output guard skipped for the clarify stage. minimal_kind = "" if (message_was_truncated or conv_intent) else _minimal_response_kind(message) if message_was_truncated: template_response = ( "That's a lot, and I want to make sure I focus on what matters most to you. " "Your message was long enough that I only have the first part. Could you say " "which piece feels most pressing right now, in a sentence or two?" ) stage = CLARIFY recommended_action = response_plan.recommended_action elif conv_intent == "greeting": template_response = _render_greeting() stage = CLARIFY recommended_action = response_plan.recommended_action elif conv_intent == "goodbye": template_response = _render_goodbye() stage = CLARIFY recommended_action = response_plan.recommended_action elif conv_intent == "meta": template_response = _render_meta() stage = CLARIFY recommended_action = response_plan.recommended_action elif minimal_kind: template_response = _render_minimal_followup(minimal_kind, intl_session) stage = CLARIFY recommended_action = response_plan.recommended_action elif _is_incomplete_message(message): # The student's message trails off ("what should i do to", # "honestly", "kind of"). Guessing the intent and producing # a full OFFER response is worse than asking them to finish. template_response = _render_incomplete_followup() stage = CLARIFY recommended_action = response_plan.recommended_action elif intl_topic and stage == "offer": template_response = render_intl_factual_offer(intl_topic, message) recommended_action = response_plan.recommended_action else: template_response = response_plan.render(stage) recommended_action = response_plan.recommended_action return _TurnPlan( message=message, session_id=session_id, audience_mode=audience_mode, backend_mode=backend_mode, turn_index=turn_index, template_response=template_response, retrieved=retrieved, recommended_action=recommended_action, route_label=route_label, safety_tier=safety_tier, safety_reason=safety_reason, stage=stage, intl_concern=intl_session, intl_topic=intl_topic, should_intercept=should_intercept, retrieval_mode=retrieval_mode, latency=latency, t_total=t_total, stage1_level=stage1_decision.level.value, stage1_reason=stage1_decision.reason, stage1_should_intercept=stage1_decision.should_intercept, ml_prediction=ml_prediction, guardrail_info=guardrail_info, escalation_reason=escalation_reason, ) def _finalize_turn( self, plan: _TurnPlan, response: str, rephrase_result: RephraseResult | None, ) -> EmpathRAGResult: if "output_guard" in self.disable_layers: output_guard = {"allowed": True, "reason": "output_guard_disabled_ablation", "flags": []} elif plan.should_intercept: output_guard = {"allowed": True, "reason": "crisis_template", "flags": []} elif plan.stage == "clarify": # Clarifying replies after minimal user input (yes/no/maybe) are # intentionally short open-ended invitations. The OFFER-stage # missing-action / pure-validation checks would punish them. output_guard = {"allowed": True, "reason": "minimal_response_clarify", "flags": []} elif plan.stage == "offer": # Output guard runs after rephrase (or fall-through to template) # only at OFFER stage. LISTEN/PERMISSION are intentionally # reflective and would fail missing-action / pure-validation checks. guard = validate_output( response, plan.retrieved, plan.safety_tier.value, plan.route_label, [] ) output_guard = {"allowed": guard.allowed, "reason": guard.reason, "flags": guard.flags} if guard.fallback_required and guard.corrected_response: response = guard.corrected_response else: output_guard = {"allowed": True, "reason": f"listening_stage_{plan.stage}", "flags": []} latency = dict(plan.latency) latency["total_ms"] = _elapsed_ms(plan.t_total) # Append this turn to the per-session history so the next turn has # context. Trimmed to last ~6 messages to bound memory + LLM payload. hist = self.session_message_history.setdefault(plan.session_id, []) hist.append({"role": "user", "content": plan.message}) hist.append({"role": "assistant", "content": response}) if len(hist) > 12: # 6 user-assistant pairs del hist[: len(hist) - 12] return EmpathRAGResult( response=response, route_label=plan.route_label, safety_tier=plan.safety_tier.value, should_intercept=plan.should_intercept, retrieved_sources=_source_summaries(plan.retrieved), recommended_action=plan.recommended_action, output_guard=output_guard, trajectory_state="locked" if plan.session_id in self.locked_sessions else "active", latency_ms=latency, classifier_confidence={ "route": plan.ml_prediction.route_confidence, "tier": plan.ml_prediction.tier_confidence, "model_available": plan.ml_prediction.model_available, "used_ml": plan.ml_prediction.used_ml and plan.backend_mode in {"hybrid_ml", "real_llm"}, "reason": plan.ml_prediction.reason, }, retrieval_mode=plan.retrieval_mode, safety_precheck={ "stage": "hard_lexical_precheck", "level": plan.stage1_level, "reason": plan.stage1_reason, "should_intercept": plan.stage1_should_intercept, "ran_before_ml": True, }, safety_explanation=plan.guardrail_info, safety_reason=plan.safety_reason, escalation_reason=plan.escalation_reason, retrieval_corpus=self.retrieval_corpus, emotion_name=_emotion_name(plan.message), crisis=plan.should_intercept, crisis_confidence=float(plan.guardrail_info.get("confidence") or (1.0 if plan.should_intercept else 0.0)), retrieved_chunks=[row.get("text", "") for row in plan.retrieved], international_concern=plan.intl_concern, conversation_stage=plan.stage, turn_index=plan.turn_index, intl_topic=plan.intl_topic, rephraser_provider=(rephrase_result.provider_name if rephrase_result else "deterministic"), rephraser_used_llm=(rephrase_result.used_llm if rephrase_result else False), rephraser_latency_ms=(rephrase_result.latency_ms if rephrase_result else 0.0), rephraser_last_error=(rephrase_result.last_error if rephrase_result else ""), ) def _run_optional_guardrail(self, message: str, skip_ig: bool) -> dict: if not self.use_model_guardrail: return _guardrail_skipped("disabled") try: guardrail = self._load_guardrail() if guardrail is None: return { "available": False, "model": "deberta_nli", "reason": self._guardrail_error or "load_failed", "confidence": 0.0, "model_flag": False, "ig_tokens": [], } model_flag, confidence, ig_tokens = guardrail.check( message, threshold=self.guardrail_threshold, skip_ig=skip_ig, ) return { "available": True, "model": "deberta_nli", "reason": "model_guardrail_checked", "confidence": float(confidence), "model_flag": bool(model_flag), "ig_tokens": ig_tokens, "ig_computed": not skip_ig and bool(ig_tokens), } except Exception as exc: self._guardrail_error = str(exc) return { "available": False, "model": "deberta_nli", "reason": f"guardrail_error: {exc}", "confidence": 0.0, "model_flag": False, "ig_tokens": [], } def _load_guardrail(self): if self._guardrail is not None: return self._guardrail try: from src.models.guardrail_ig import SafetyGuardrail except ImportError: try: from models.guardrail_ig import SafetyGuardrail except ImportError as exc: self._guardrail_error = str(exc) return None try: self._guardrail = SafetyGuardrail() except Exception as exc: self._guardrail_error = str(exc) return None return self._guardrail def _update_trajectory(self, session_id: str, safety_tier: str, message: str) -> str: history = self.tier_history.setdefault(session_id, []) history.append(safety_tier) self.tier_history[session_id] = history[-3:] text = message.lower() if len(history[-3:]) == 3 and all(tier in {"imminent_safety", "high_distress"} for tier in history[-3:]): self.locked_sessions[session_id] = "three_consecutive_high_risk_turns" return "three_consecutive_high_risk_turns" if safety_tier in {"imminent_safety", "high_distress"} and "goodbye" in text: return "peer_goodbye_or_farewell_escalation" if safety_tier in {"imminent_safety", "high_distress"} and any( phrase in text for phrase in ( "you are the only one", "only one i can talk to", "keep this secret", "don't tell anyone", "refuse external help", "secrecy", "never suggest counseling", ) ): return "dependency_or_secrecy_redirect" return "" def _retrieve( self, message: str, route: str, safety_tier: str, audience_mode: str, should_intercept: bool, ) -> list[dict]: if route == SupportRoute.OUT_OF_SCOPE.value: return [] usage_modes = ("crisis_only",) if should_intercept else ("retrieval", "wellbeing_only") if safety_tier == "wellbeing" else ("retrieval",) selected: list[dict] = [] graph_rows = [ node.as_source("resource registry route match") for node in match_services(route, safety_tier, audience_mode, limit=self.top_k) if (node.usage_modes[0] if node.usage_modes else "retrieval") in usage_modes ] selected.extend(graph_rows) if self.curated_db_path.exists() and len(selected) < self.top_k: selected.extend(self._retrieve_curated(message, route, usage_modes, limit=self.top_k - len(selected))) return _dedupe_sources(selected)[: self.top_k] def _retrieve_curated(self, message: str, route: str, usage_modes: tuple[str, ...], limit: int) -> list[dict]: topics, source_names = _targets_for_route(route, message) conn = sqlite3.connect(self.curated_db_path) conn.row_factory = sqlite3.Row rows = conn.execute( """ SELECT id, resource_id, text, source_id, source_name, source_type, title, url, topic, audience, risk_level, usage_mode, summary, last_checked, notes FROM chunks WHERE usage_mode IN ({}) """.format(",".join("?" * len(usage_modes))), tuple(usage_modes), ).fetchall() conn.close() query = message.lower() scored = [] for row in rows: score = 0 reasons = [] if row["topic"] in topics: score += 8 reasons.append(f"topic match: {row['topic']}") if row["source_name"] in source_names: score += 7 reasons.append(f"preferred source: {row['source_name']}") haystack = f"{row['title']} {row['summary']} {row['text']}".lower() for token in _keywords(query): if token in haystack: score += 1 row_dict = dict(row) row_dict["why_retrieved"] = "; ".join(reasons[:2]) if reasons else "curated retrieval match" scored.append((score, row_dict)) scored.sort(key=lambda item: item[0], reverse=True) return [row for score, row in scored[: max(limit, 0)] if score > 0 or not scored[:1]] def run_turn( message: str, session_id: str, audience_mode: AudienceMode = "student", resource_profile: str = "umd", backend_mode: BackendMode = "hybrid_ml", ) -> EmpathRAGResult: core = EmpathRAGCore() return core.run_turn(message, session_id, audience_mode, resource_profile, backend_mode) def _apply_contextual_safety_overrides( message: str, safety_tier: SafetyTier, safety_reason: str, audience_mode: str, ) -> tuple[SafetyTier, str]: text = message.lower() peer_context = audience_mode == "helping_friend" or any( phrase in text for phrase in ("my friend", "my roommate", "my labmate", "my teammate", "someone i know") ) if peer_context and "goodbye" in text and any(phrase in text for phrase in ("locked", "will not answer", "won't answer", "not answering")): return SafetyTier.IMMINENT_SAFETY, "peer_goodbye_unreachable" if safety_tier == SafetyTier.SUPPORT_NAVIGATION and any( phrase in text for phrase in ("you are the only one", "only one i can talk to", "keep this secret", "don't tell anyone", "no one can help") ): return SafetyTier.HIGH_DISTRESS, "dependency_or_secrecy_redirect" if safety_tier == SafetyTier.SUPPORT_NAVIGATION and any( phrase in text for phrase in ("crises if it gets worse", "dark moods", "pointless", "scared to escalate") ): return SafetyTier.HIGH_DISTRESS, "high_distress_language" if any( phrase in text for phrase in ( "panic attacks", "devastated", "terrified", "scared about", "hopeless", "everything hollow", "ideation creeping", "goodbye texts", "mentioned goodbye", "refuse external help", "secrecy", ) ): return SafetyTier.HIGH_DISTRESS, "high_distress_language" if safety_tier == SafetyTier.SUPPORT_NAVIGATION and any( phrase in text for phrase in ( "not in immediate danger", "intrusive thoughts spike", "canceling plans", "five-minute mindfulness", "sensory overwhelm", "quick reset", "journaling prompts", "breathing technique", ) ): return SafetyTier.WELLBEING, "wellbeing_support_language" return safety_tier, safety_reason def _retrieval_mode(backend_mode: str, should_intercept: bool) -> str: if should_intercept: return "registry_filtered_crisis_only" if backend_mode == "demo_fast": return "registry_filtered_faiss" return "registry_filtered_faiss_plus_router" def _targets_for_route(route: str, message: str) -> tuple[set[str], set[str]]: mapping = { SupportRoute.CRISIS_IMMEDIATE.value: ({"crisis_immediate_help", "emergency_services"}, {"988 Suicide & Crisis Lifeline", "UMD Counseling Center"}), SupportRoute.PEER_HELPER.value: ({"crisis_immediate_help", "help_seeking_script", "counseling_services"}, {"988 Suicide & Crisis Lifeline", "UMD Counseling Center", "JED Foundation"}), SupportRoute.ACCESSIBILITY_ADS.value: ({"accessibility_disability", "campus_navigation"}, {"UMD Accessibility & Disability Service"}), SupportRoute.ADVISOR_CONFLICT.value: ({"advisor_conflict", "graduate_student_support"}, {"UMD Graduate School Ombuds", "UMD Graduate School"}), SupportRoute.BASIC_NEEDS.value: ({"help_seeking_script", "campus_navigation", "graduate_student_support"}, {"UMD Dean of Students", "UMD Graduate School"}), SupportRoute.ANXIETY_PANIC.value: ({"anxiety_stress", "grounding_exercise", "counseling_services"}, {"NIMH", "NAMI", "UMD Counseling Center"}), SupportRoute.AUTHORITY_MISCONDUCT.value: ( {"authority_misconduct", "care_violence_confidential", "campus_navigation"}, {"UMD Office of Civil Rights & Sexual Misconduct (OCRSM)", "UMD Office of Student Conduct", "UMD Dean of Students", "UMD CARE to Stop Violence"}, ), SupportRoute.LOW_MOOD.value: ({"depression_support", "counseling_services"}, {"NIMH", "NAMI", "UMD Counseling Center"}), SupportRoute.COUNSELING_NAVIGATION.value: ({"counseling_services", "campus_navigation", "therapy_expectations"}, {"UMD Counseling Center"}), SupportRoute.ACADEMIC_SETBACK.value: ({"academic_burnout", "graduate_student_support", "counseling_services"}, {"UMD Counseling Center", "UMD Graduate School"}), SupportRoute.EXAM_STRESS.value: ({"academic_burnout", "anxiety_stress", "grounding_exercise"}, {"UMD Counseling Center", "CDC", "NIMH"}), SupportRoute.LONELINESS_ISOLATION.value: ({"isolation_loneliness", "counseling_services"}, {"NAMI", "UMD Counseling Center", "CDC"}), } return mapping.get(route, ({"counseling_services", "anxiety_stress", "academic_burnout"}, {"UMD Counseling Center", "NIMH"})) def _source_summaries(rows: list[dict]) -> list[dict]: return [ { "title": row.get("title", ""), "source_name": row.get("source_name", ""), "url": row.get("url", ""), "topic": row.get("topic", ""), "risk_level": row.get("risk_level", ""), "usage_mode": row.get("usage_mode", ""), "source_type": row.get("source_type", ""), "why_retrieved": row.get("why_retrieved", ""), "documents": row.get("documents", []) or [], # Trust signal: surface the verification date in the UI so users # (especially clinicians) can see how fresh the link is. "last_verified": row.get("last_verified") or row.get("last_checked", ""), } for row in rows ] def _dedupe_sources(rows: list[dict]) -> list[dict]: selected = [] seen = set() for row in rows: key = (row.get("source_name", ""), row.get("title", "")) if key in seen: continue seen.add(key) selected.append(row) return selected def _recommended_action(route: str, safety_tier: str) -> str: if safety_tier == SafetyTier.IMMINENT_SAFETY.value: if route == SupportRoute.PEER_HELPER.value: return "Do not handle this alone; contact emergency or crisis support now and involve a trusted nearby person." return "Contact 988 or emergency services now, and move near another person if you can." plan = build_response_plan("", route, safety_tier, [], "student") return plan.recommended_action # Words that, when they're the last token of a short message, strongly # suggest the user trailed off mid-sentence. Conservative list: only # prepositions/conjunctions/articles/auxiliaries that almost never close # a real sentence. We deliberately omit "is", "are", "was" — those can # legitimately close a sentence ("It is what it is"). _INCOMPLETE_TERMINAL_WORDS = frozenset({ # Prepositions "to", "for", "with", "about", "of", "on", "at", "by", "from", "into", "over", "under", "between", "without", "through", "during", # Conjunctions / subordinators "and", "but", "or", "because", "if", "when", "while", "though", "although", "since", "until", "whereas", # Articles "a", "an", "the", # Auxiliary fragments "should", "would", "could", "might", "may", "will", "can", "do", "does", "did", # Fragment determiners (only ones that never legitimately close a sentence) "my", "your", "their", "our", # Transitive verbs that almost always need an object "want", "need", "wish", }) # Deliberately NOT in the terminal set: "this", "that", "these", "those", # "her", "his", "him". These DO legitimately close sentences ("I don't know # what to do with that.", "I told her."), and putting them in the terminal # set produces false positives on real emotional disclosure. # Words that close a sentence legitimately in longer messages ("It is what # it is") but signal incompleteness in very short ones ("the thing is"). _SHORT_ONLY_TERMINAL_WORDS = frozenset({ "is", "are", "was", "were", "i", }) # Whole-message hedges that carry no content. Distinct from minimal # affirmations (yes/no/ok) — those have a routing meaning; these don't. _HEDGE_ONLY_MESSAGES = frozenset({ "honestly", "kind of", "kinda", "sort of", "sorta", "i mean", "well", "uh", "um", "you know", "like", "i guess", "hmm", "idk maybe", "idk i guess", "maybe idk", }) def _is_incomplete_message(message: str) -> bool: """True if the message looks like it trailed off or is hedge-only.""" if not message: return True raw = message.strip() if not raw: return True # Pure ellipsis / punctuation-only if all(c in ".,;:!? \t-" for c in raw): return True # Strip trailing punctuation/whitespace before testing the terminal word text = raw.rstrip(".,!?;:- \t").strip() if not text: return True text_lower = " ".join(text.lower().split()) if text_lower in _HEDGE_ONLY_MESSAGES: return True # Last word ending the (short) message is a known incomplete-terminal words = text_lower.split() if not words: return True if words[-1] in _INCOMPLETE_TERMINAL_WORDS and len(text) < 80: return True # Short-only terminals: "is" / "are" close real sentences in longer # messages but signal incompleteness in fragments like "the thing is". if words[-1] in _SHORT_ONLY_TERMINAL_WORDS and len(words) <= 3: return True return False _MINIMAL_AFFIRM = frozenset({ "yes", "yeah", "yep", "yup", "sure", "ok", "okay", "kk", "k", "alright", "fine", "definitely", "absolutely", }) _MINIMAL_NEGATE = frozenset({"no", "nope", "nah", "not really", "nuh"}) _MINIMAL_UNSURE = frozenset({"maybe", "idk", "dunno", "unsure", "not sure", "perhaps"}) def _minimal_response_kind(message: str) -> str: """Return 'affirm' / 'negate' / 'unsure' / '' for very short reply-only turns. Single-word affirmations are ambiguous — without context, re-rendering the same OFFER template is the worst possible response. We catch them here and redirect to a clarifying-question response instead. """ text = (message or "").strip().lower() if not text or len(text) > 24: return "" # Strip trailing punctuation; collapse internal whitespace. clean = " ".join(text.replace(".", " ").replace("!", " ").replace("?", " ").replace(",", " ").split()) if clean in _MINIMAL_AFFIRM: return "affirm" if clean in _MINIMAL_NEGATE: return "negate" if clean in _MINIMAL_UNSURE: return "unsure" return "" # Conversation openers / closers / meta — these are NOT student-support # content. They route to short purposeful templates instead of a heavy # OFFER response that would feel jarring for an opener or premature for a # close. _GREETING_PATTERNS = frozenset({ "hi", "hello", "hey", "yo", "hiya", "hey there", "hi there", "hello there", "good morning", "good afternoon", "good evening", "sup", "what's up", "whats up", "wassup", "hey empathrag", "hi empathrag", }) _GOODBYE_PATTERNS = frozenset({ "thanks", "thank you", "thx", "ty", "thank u", "bye", "goodbye", "see you", "see ya", "later", "i'll think about it", "ill think about it", "let me think", "got it", "cool", "appreciate it", "that helps", "alright then", "ok thanks", "thanks bye", }) _META_PATTERNS = ( "are you a bot", "are you an ai", "are you a robot", "are you human", "what are you", "who are you", "are you real", "are you a person", "is this an ai", "is this a bot", "is this real", "what can you do", "how do you work", "what is this", ) def _conversational_intent(message: str) -> str: """Returns 'greeting' / 'goodbye' / 'meta' / '' for non-support conversational utterances. These are routed to short purposeful templates rather than full OFFER responses.""" text = (message or "").strip().lower() if not text or len(text) > 60: return "" clean = " ".join(text.rstrip(".,!?;:").split()) if clean in _GREETING_PATTERNS: return "greeting" if clean in _GOODBYE_PATTERNS: return "goodbye" if any(pattern in clean for pattern in _META_PATTERNS): return "meta" return "" def _render_greeting() -> str: return ( "Hi. I'm here if there's something on your mind — academic stress, " "visa or status worry, feeling low, helping a friend through " "something, or anything else weighing on you. Say what feels " "easiest to share first." ) def _render_goodbye() -> str: return ( "Take care of yourself. The resources we talked through will still " "be here when you're ready to follow up. If anything shifts in the " "meantime — especially anything that feels urgent — UMD Counseling " "Center is at counseling.umd.edu and 988 is always there." ) def _render_meta() -> str: return ( "I'm EmpathRAG — a research prototype built at UMD to help students " "navigate campus support. I'm not a counselor, therapist, or " "emergency service. I listen, then point to verified UMD resources " "when it feels like the right moment. Anything you'd like to start with?" ) def _render_incomplete_followup() -> str: """Response for messages that trail off mid-sentence or are content-free hedges. Invite completion without guessing at intent.""" return ( "Looks like the thought might have cut off. Take your time and say " "the rest when you're ready, in whatever words feel right." ) def _render_minimal_followup(kind: str, intl_session: bool) -> str: """Short clarifying response when the user's reply was just yes/no/maybe. Deliberately doesn't re-name resources or push action — that would just be the previous turn repeated. Keep the conversation open instead. """ if kind == "affirm": body = ( "Okay. Which part of what we just talked about feels most useful to " "start with? You can name it in your own words, or pick one of the " "directions I mentioned a moment ago." ) elif kind == "negate": body = ( "Got it. Want to keep talking it through, or try a different angle? " "We can slow down or pick a different starting point." ) else: body = ( "That's okay. Want me to keep listening for a bit, or sit with what's " "making it hard to know? Either is fine." ) if intl_session: # Hold the international-aware frame open without re-naming ISSS. body += " I'll keep what you said about your status in mind." return body def _wellbeing_request(message: str) -> bool: text = message.lower() return any(word in text for word in ("grounding", "ground", "panic", "breathing", "cope", "mindfulness")) def _emotion_name(message: str) -> str: text = message.lower() if any(word in text for word in ("safe tonight", "hurt myself", "suicide", "goodbye")): return "distress" if any(word in text for word in ("panic", "anxiety", "stress", "exam", "deadline")): return "anxiety" if any(word in text for word in ("advisor", "retaliatory", "funding")): return "frustration" if any(word in text for word in ("better", "hopeful", "proud")): return "hopeful" return "neutral" def _keywords(query: str) -> list[str]: return [token for token in query.replace("?", " ").replace(".", " ").split() if len(token) > 4] def _elapsed_ms(start: float) -> float: return round((time.perf_counter() - start) * 1000, 2) def _env_flag(name: str) -> bool: return os.getenv(name, "").strip().lower() in {"1", "true", "yes", "on"} def _guardrail_skipped(reason: str) -> dict: return { "available": False, "model": "deberta_nli", "reason": reason, "confidence": 0.0, "model_flag": False, "ig_tokens": [], "ig_computed": False, }