ElebobaandClaude Opus 5.5
b4731fb1d6
CI / ruff · pytest (pull_request) Successful in 22s
Freeze interfaces between lanes A (bot), B (backend), C (dashboard) so three developers work in parallel without touching each other's files: dialog Event/Reply + engine registry with echo stub, integration protocols with in-memory defaults, MonitoringService with noop stub, /api/v1 schemas with fixture-backed routes, frozen main.py wiring, config/modules.yaml + due_scales(), DAR-5, tzdata, Andrey trajectory, lanes in WORKFLOW, ADR-010. Refs #34 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""Deterministic offline `LLMTasks` (ADR-004): keyword rules over Russian text.
|
||
|
||
Works at the task level (not by parsing prompts): it receives the same arguments as `PromptedTasks`
|
||
and returns the same pydantic models, so the gateway policy (thresholds, filters, fallbacks) is
|
||
identical for both providers. It must cover every free-text reply in docs/product/01_demo_persona.md.
|
||
|
||
These are rough heuristics for the demo, not a clinical NLP model. The risk rules here are an
|
||
*auxiliary* signal only; the primary safety signal is the psychiatrist's lexicon in `app.core.safety`.
|
||
"""
|
||
|
||
import re
|
||
|
||
from app.core.types import ScrubbedText
|
||
from app.llm.schemas import CheckinFacts, ItemScore, RiskVerdict, SideEffect
|
||
|
||
|
||
def _norm(text: str) -> str:
|
||
return " ".join(text.lower().replace("ё", "е").split())
|
||
|
||
|
||
def _any(patterns: tuple[str, ...], text: str) -> str | None:
|
||
for p in patterns:
|
||
m = re.search(p, text)
|
||
if m:
|
||
return m.group(0)
|
||
return None
|
||
|
||
|
||
# --- extract_item ----------------------------------------------------------------------------------
|
||
# Answers are mapped to a severity fraction of the item maximum, following the PHQ-9 frequency anchors
|
||
# (0 = not at all, 1 = several days, 2 = more than half the days, 3 = nearly every day). The fraction is
|
||
# then scaled to the item range, so PCL-5 (0..4) items reuse the same cues: round(fraction * max_score).
|
||
_SEVERITY_CUES: tuple[tuple[float, tuple[str, ...]], ...] = (
|
||
(
|
||
1.0,
|
||
(
|
||
r"почти кажд\w+ (день|ночь|дня|ночи)",
|
||
r"кажд\w+ (день|ночь)",
|
||
r"постоянно",
|
||
r"вс[её] время",
|
||
r"ежедневно",
|
||
r"почти не сплю",
|
||
r"совсем не (сплю|могу)",
|
||
r"очень (плохо|сильно|тяжело)",
|
||
r"ужасно",
|
||
r"невыносимо",
|
||
r"крайне",
|
||
),
|
||
),
|
||
(
|
||
2 / 3,
|
||
(
|
||
r"часто",
|
||
r"больше половины",
|
||
r"большую часть",
|
||
r"плохо",
|
||
r"сильно",
|
||
r"тяжело",
|
||
r"много раз",
|
||
r"не могу",
|
||
r"нет сил",
|
||
r"ничего не хочется",
|
||
),
|
||
),
|
||
(
|
||
1 / 3,
|
||
(
|
||
r"иногда",
|
||
r"несколько (дней|раз|ночей)",
|
||
r"пару раз",
|
||
r"время от времени",
|
||
r"бывает",
|
||
r"изредка",
|
||
r"немного",
|
||
r"слегка",
|
||
r"не очень",
|
||
r"чуть-чуть",
|
||
),
|
||
),
|
||
(
|
||
0.0,
|
||
(
|
||
r"\bнет\b",
|
||
r"не было",
|
||
r"ни разу",
|
||
r"совсем не беспоко",
|
||
r"не беспоко",
|
||
r"нормально",
|
||
r"\bхорошо\b",
|
||
r"отлично",
|
||
r"в порядке",
|
||
),
|
||
),
|
||
)
|
||
_UNSURE = (r"не знаю", r"сложно сказать", r"трудно сказать", r"не помню", r"затрудняюсь")
|
||
_NUMERIC_ANSWER = re.compile(r"^(?:на\s*)?(\d)(?:\s*балл\w*)?[.!]?$")
|
||
|
||
|
||
def _extract_item(scale_item_id: str, max_score: int, answer: str) -> ItemScore:
|
||
text = _norm(answer)
|
||
if _any(_UNSURE, text):
|
||
return ItemScore(item=scale_item_id, score=0, confidence=0.2, evidence=text[:60])
|
||
numeric = _NUMERIC_ANSWER.match(text)
|
||
if numeric and int(numeric.group(1)) <= max_score:
|
||
return ItemScore(item=scale_item_id, score=int(numeric.group(1)), confidence=0.9, evidence=text[:60])
|
||
# Highest matched severity wins: "сплю плохо, кошмары почти каждую ночь" -> nearly every day.
|
||
for fraction, cues in _SEVERITY_CUES:
|
||
hit = _any(cues, text)
|
||
if hit:
|
||
return ItemScore(item=scale_item_id, score=round(fraction * max_score), confidence=0.8, evidence=hit)
|
||
# No frequency/intensity cue -> low confidence, the gateway turns it into a clarifying question.
|
||
return ItemScore(item=scale_item_id, score=0, confidence=0.3, evidence="")
|
||
|
||
|
||
# --- extract_checkin -------------------------------------------------------------------------------
|
||
_SIDE_EFFECT_CUES: tuple[tuple[SideEffect, tuple[str, ...]], ...] = (
|
||
("nausea", (r"тошн", r"мутит", r"рвот", r"подташнива")),
|
||
# Only explicit inability to sleep counts as the insomnia side effect; "сплю плохо" alone is a sleep
|
||
# symptom (scale item / sleep bucket), otherwise veterans would get a side-effect flag every day.
|
||
("insomnia", (r"не могу (у|за)снуть", r"(у|за)снуть не могу", r"бессонниц", r"не спится")),
|
||
("headache", (r"голова болит", r"болит голова", r"головн\w* бол", r"мигрен")),
|
||
("dizziness", (r"кружится голова", r"голова кружится", r"головокруж")),
|
||
("sexual", (r"либидо", r"влечени", r"потенци", r"эрекц", r"оргазм")),
|
||
("agitation", (r"беспокойств", r"не нахожу себе места", r"взвинчен", r"трясет", r"дерга")),
|
||
)
|
||
_GENERIC_SIDE_EFFECT = (r"от (таблет\w*|лекарств\w*|препарат\w*) (только )?хуже", r"побочк", r"побочн")
|
||
|
||
_QUIT = (
|
||
r"брош\w*\b(?! (пить|курить))",
|
||
r"бросить(?! (пить|курить))",
|
||
r"бросаю(?! (пить|курить))",
|
||
r"перестан\w* (их |его )?(принимать|пить)",
|
||
r"не буду (их |его |больше )*(принимать|пить (таблет|лекарств|препарат))",
|
||
r"решил\w* не принимать",
|
||
r"хочу отказаться",
|
||
r"откажусь от",
|
||
)
|
||
_MED_TAKEN_NO = (
|
||
r"не принял",
|
||
r"не принимал",
|
||
r"не пил (таблет|лекарств|препарат)",
|
||
r"пропустил",
|
||
r"забыл\w* (принять|выпить|про таблет)",
|
||
r"решил\w* не принимать",
|
||
)
|
||
_MED_TAKEN_YES = (r"\bпринял\w*", r"выпил\w* (таблет|лекарств|препарат)", r"приняла")
|
||
|
||
# "Не могу уснуть без пива" is drinking, not abstinence — checked before the "без алкоголя" rule.
|
||
_ALCOHOL_DEPENDENT = (r"(не могу|не получается|не усну)\D{0,20}без (алкогол|пив|вин|водк|выпивк)",)
|
||
_ALCOHOL_NONE = (
|
||
r"не пил\b(?! (таблет|лекарств|препарат))",
|
||
r"без алкогол",
|
||
r"алкогол\w* (нет|не было)",
|
||
r"не выпивал",
|
||
r"трезв",
|
||
)
|
||
_ALCOHOL_MORE = (
|
||
r"напил",
|
||
r"много (пива|выпил|алкогол)",
|
||
r"(три|четыре|пять|несколько|много) (бутыл|банок|бокал|рюм)",
|
||
r"бутылк\w* водки",
|
||
r"литр",
|
||
r"пол-?литр",
|
||
r"запой",
|
||
)
|
||
_ALCOHOL_SOME = (
|
||
r"\bпив",
|
||
r"\bвин(о|ца)\b",
|
||
r"(бокал|выпил)\w* вина",
|
||
r"водк",
|
||
r"коньяк",
|
||
r"виски",
|
||
r"бокал",
|
||
r"рюмк",
|
||
r"выпил\w* (пару|одну|бутылку|банку|немного)",
|
||
r"пару (бутыл|банок)",
|
||
r"алкогол",
|
||
r"выпива",
|
||
)
|
||
|
||
_NIGHTMARES_NO = (r"кошмар\w* (нет|не было|не снил)", r"без кошмар")
|
||
_NIGHTMARES_YES = (r"кошмар", r"страшн\w* сн", r"плох\w* сн", r"снится (война|бой|фронт)")
|
||
|
||
_MED_WORDS = r"(таблет|препарат|лекарств|сертралин|доз|пить их|принимать)"
|
||
_MOOD = (
|
||
re.compile(r"настроени\w*\D{0,12}?(\d{1,2})"),
|
||
re.compile(r"(\d{1,2})\s*(?:/|из)\s*10"),
|
||
)
|
||
_SLEEP_HOURS = re.compile(r"(?:спал\w*|сплю|сна)\D{0,10}?(\d{1,2})\s*(?:час|ч\b)|(\d{1,2})\s*час\w* сна")
|
||
_SLEEP_SHORT = (r"почти не спал", r"\bне спал", r"пару часов", r"часа два", r"глаз не сомкнул")
|
||
|
||
|
||
def _sleep_bucket(hours: int) -> str:
|
||
if hours < 4:
|
||
return "<4"
|
||
if hours < 6:
|
||
return "4-6"
|
||
if hours <= 8:
|
||
return "6-8"
|
||
return ">8"
|
||
|
||
|
||
def _extract_checkin(raw: str) -> CheckinFacts:
|
||
text = _norm(raw)
|
||
|
||
mood = None
|
||
for pattern in _MOOD:
|
||
m = pattern.search(text)
|
||
if m and int(m.group(1)) <= 10:
|
||
mood = int(m.group(1))
|
||
break
|
||
|
||
side_effects: list[SideEffect] = [code for code, cues in _SIDE_EFFECT_CUES if _any(cues, text)]
|
||
if not side_effects and _any(_GENERIC_SIDE_EFFECT, text):
|
||
side_effects.append("other")
|
||
|
||
quit_intent = _any(_QUIT, text) is not None
|
||
if _any(_MED_TAKEN_NO, text):
|
||
med_taken: bool | None = False
|
||
elif _any(_MED_TAKEN_YES, text):
|
||
med_taken = True
|
||
else:
|
||
med_taken = None
|
||
|
||
# "Не могу уснуть без пива" mentions drinking without a quantity: record the minimal bucket ("1-2")
|
||
# so the alcohol-days rule still sees a drinking day; the button answer can refine it.
|
||
if _any(_ALCOHOL_DEPENDENT, text):
|
||
alcohol: str | None = "1-2"
|
||
elif _any(_ALCOHOL_NONE, text):
|
||
alcohol = "none"
|
||
elif _any(_ALCOHOL_MORE, text):
|
||
alcohol = "more"
|
||
elif _any(_ALCOHOL_SOME, text):
|
||
alcohol = "1-2"
|
||
else:
|
||
alcohol = None
|
||
|
||
sleep = None
|
||
m = _SLEEP_HOURS.search(text)
|
||
if m:
|
||
sleep = _sleep_bucket(int(m.group(1) or m.group(2)))
|
||
elif _any(_SLEEP_SHORT, text):
|
||
sleep = "<4"
|
||
|
||
if _any(_NIGHTMARES_NO, text):
|
||
nightmares: bool | None = False
|
||
elif _any(_NIGHTMARES_YES, text):
|
||
nightmares = True
|
||
else:
|
||
nightmares = None
|
||
|
||
med_question = "?" in text and re.search(_MED_WORDS, text) is not None
|
||
|
||
signals = (mood, med_taken, alcohol, sleep, nightmares)
|
||
found = any(x is not None for x in signals) or bool(side_effects) or quit_intent or med_question
|
||
return CheckinFacts(
|
||
mood=mood,
|
||
med_taken=med_taken,
|
||
side_effects=side_effects,
|
||
quit_intent=quit_intent,
|
||
alcohol=alcohol,
|
||
sleep_hours_bucket=sleep,
|
||
nightmares=nightmares,
|
||
med_question=med_question,
|
||
confidence=0.85 if found else 0.3,
|
||
)
|
||
|
||
|
||
# --- risk_classify ---------------------------------------------------------------------------------
|
||
_RISK_ACUTE = (
|
||
r"покончить с собой",
|
||
r"убить себя",
|
||
r"убью себя",
|
||
r"суицид",
|
||
r"самоубийств",
|
||
r"повеситься",
|
||
r"вскрыть вены",
|
||
r"не хочу жить",
|
||
r"не хочется жить",
|
||
r"жить не хочу",
|
||
r"наглотаться таблет",
|
||
r"выпил\w* все таблетки",
|
||
r"прощайте",
|
||
r"порезать себя",
|
||
)
|
||
_RISK_POSSIBLE = (
|
||
r"бессмысленн",
|
||
r"нет смысла",
|
||
r"смысла нет",
|
||
r"устал\w* от всего",
|
||
r"безнадежн",
|
||
r"никому не нужен",
|
||
r"никому не нужна",
|
||
r"хочу исчезнуть",
|
||
r"лучше бы меня не было",
|
||
r"заснуть и не проснуться",
|
||
r"не проснуться",
|
||
r"всем будет лучше без меня",
|
||
r"нет сил жить",
|
||
r"тупик",
|
||
)
|
||
|
||
|
||
def _risk(raw: str) -> RiskVerdict:
|
||
text = _norm(raw)
|
||
if _any(_RISK_ACUTE, text):
|
||
return RiskVerdict(level="acute", confidence=0.9)
|
||
if _any(_RISK_POSSIBLE, text):
|
||
return RiskVerdict(level="possible", confidence=0.7)
|
||
return RiskVerdict(level="none", confidence=0.6)
|
||
|
||
|
||
class MockTasks:
|
||
name = "mock"
|
||
|
||
async def extract_item(
|
||
self, scale_item_id: str, item_text: str, max_score: int, answer: ScrubbedText
|
||
) -> ItemScore | None:
|
||
return _extract_item(scale_item_id, max_score, answer)
|
||
|
||
async def extract_checkin(self, text: ScrubbedText) -> CheckinFacts | None:
|
||
return _extract_checkin(text)
|
||
|
||
async def risk_classify(self, text: ScrubbedText) -> RiskVerdict | None:
|
||
return _risk(text)
|
||
|
||
async def phrase(self, intent: str, payload: ScrubbedText, base_text: str) -> str | None:
|
||
return base_text
|
||
|
||
async def summarize_for_doctor(self, structured: ScrubbedText, base_text: str) -> str | None:
|
||
return base_text
|