feat(bot): LLM gateway with GigaChat and offline mock providers
CI / ruff · pytest (pull_request) Successful in 23s
CI / ruff · pytest (pull_request) Successful in 23s
Single entry point app/llm/gateway.py: extract_item, extract_checkin, risk_classify, phrase, summarize_for_doctor. Two-level design: LLMProvider (transport, GigaChat) and LLMTasks (PromptedTasks over a provider, or rule-based MockTasks). Shared policy in the gateway: assert_scrubbed backstop, confidence threshold 0.5, risk failure -> NONE, forbidden-output filter with config/texts/phrases.yaml fallbacks. Refs #7 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Claude Opus 5.5
+333
@@ -0,0 +1,333 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user