ElebobaandClaude Opus 5.5
2b13d54ed7
Daily check-in (mood → dose → sleep/nightmares → alcohol → free text with extract_checkin, buttons win over the LLM, progressive upsert of daily_checkins), medication reminder flow (taken / skip with reason / side effects → doses + daily_checkins), medication question menu via med_qa, med_info rule reactions from the drug instruction excerpt, «записаться раньше» request recorded for lane B, friendly intros for scheduled scales. actions.py: additive med_info handler hook. Refs #37 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
86 lines
4.4 KiB
Python
86 lines
4.4 KiB
Python
"""Execution of *dialog* actions returned by `MonitoringService.after_data_change()` (ADR-007).
|
||
|
||
The rules (lane B, `config/rules.yaml`) decide WHAT happens; this module only turns an already chosen action
|
||
into replies. `alert` / `enhanced_monitoring` are handled by lane B itself (persisted + doctor notified), so they
|
||
are ignored here. While a crisis is open only crisis-relevant actions (crisis, safety plan, risk screen) reach
|
||
the patient: no medication tips or visit offers in the middle of a crisis.
|
||
"""
|
||
|
||
from collections.abc import Awaitable, Callable, Iterable
|
||
from typing import Any
|
||
|
||
from app.core.rules import Action
|
||
from app.db.clinical import SafetyPlan
|
||
from app.dialog.contracts import Button, Reply
|
||
from app.dialog.crisis import crisis_open, open_crisis
|
||
from app.dialog.fsm import Ctx, Step
|
||
from app.dialog.scale_session import active_scale, add_pending
|
||
|
||
# med_info topic → phrase intent of `config/texts/phrases.yaml` (the LLM only rephrases the template).
|
||
_MED_INFO_INTENTS = {"delayed_effect": "quit_intent", "alcohol_interaction": "alcohol_info", "nausea": "nausea_info"}
|
||
_CRISIS_SAFE = {"crisis", "show_safety_plan", "schedule_scale"}
|
||
|
||
# Optional `med_info` answerer registered by a flow (`flows/meds.py`, #37): answers from the drug instruction
|
||
# (med_qa). Returning None falls back to the phrase template below.
|
||
MedInfoHandler = Callable[[Ctx, Action], Awaitable[Reply | None]]
|
||
_med_info_handler: MedInfoHandler | None = None
|
||
|
||
|
||
def register_med_info_handler(handler: MedInfoHandler | None) -> None:
|
||
global _med_info_handler
|
||
_med_info_handler = handler
|
||
|
||
|
||
def _plan_line(step: Any) -> str:
|
||
if isinstance(step, dict):
|
||
return ": ".join(str(v) for v in step.values() if v)
|
||
return str(step)
|
||
|
||
|
||
async def safety_plan_reply(ctx: Ctx, block: str | None = None) -> Reply:
|
||
"""The plan composed with the doctor (`safety_plans`), or the Stanley–Brown template from config."""
|
||
row = await ctx.session.get(SafetyPlan, ctx.token)
|
||
steps = [_plan_line(s) for s in row.steps] if row is not None and row.steps else ctx.texts.safety_plan.template
|
||
lines = [ctx.texts.safety_plan.title, *(f"{i}. {s}" for i, s in enumerate(steps, 1))]
|
||
if block == "means_restriction":
|
||
lines += ["", ctx.texts.safety_plan.means_restriction]
|
||
return Reply(text="\n".join(lines), kind="safety_plan")
|
||
|
||
|
||
async def execute(ctx: Ctx, actions: Iterable[Action]) -> Step:
|
||
step = Step()
|
||
shown_plan = False
|
||
for action in actions:
|
||
if crisis_open({**ctx.data, **step.data}) and action.type not in _CRISIS_SAFE:
|
||
continue
|
||
if action.type == "crisis":
|
||
if not crisis_open({**ctx.data, **step.data}):
|
||
step = step.then(open_crisis(ctx, categories=[], source=f"rule:{action.code}"))
|
||
elif action.type == "show_safety_plan" and not shown_plan:
|
||
shown_plan = True
|
||
step = step.then(Step(replies=[await safety_plan_reply(ctx, action.params.get("block"))]))
|
||
elif action.type == "schedule_scale" and action.params.get("scale"):
|
||
scale_id = str(action.params["scale"])
|
||
if active_scale(step.next_state or ctx.state) == scale_id:
|
||
continue # already being asked
|
||
data = {**ctx.data, **step.data}
|
||
pending = add_pending(data, scale_id, bool(action.params.get("scheduled", False)))
|
||
step = step.then(Step(data={"pending_scales": pending}))
|
||
elif action.type == "med_info":
|
||
reply = await _med_info_handler(ctx, action) if _med_info_handler is not None else None
|
||
if reply is None:
|
||
intent = _MED_INFO_INTENTS.get(str(action.params.get("topic")), "default")
|
||
reply = Reply(text=await ctx.gateway.phrase(intent, {}), kind="info")
|
||
step = step.then(Step(replies=[reply]))
|
||
elif action.type == "suggest_visit":
|
||
text = await ctx.gateway.phrase("book_visit_offer", {})
|
||
button = Button(label=ctx.texts.visit.button, data="menu:visit")
|
||
step = step.then(Step(replies=[Reply(text=text, buttons=[[button]], kind="info")]))
|
||
elif action.type == "reach_out":
|
||
topic = str(action.params.get("topic", "default"))
|
||
texts = ctx.texts.reach_out
|
||
text = {"silence": texts.silence, "missed_doses": texts.missed_doses}.get(topic, texts.default)
|
||
step = step.then(Step(replies=[Reply(text=text)]))
|
||
# alert / enhanced_monitoring: lane B persists and notifies; nothing to say to the patient.
|
||
return step
|