ElebobaandClaude Opus 5.5
51cc259002
CI / ruff · pytest (pull_request) Successful in 21s
Topic from engine or keyword classifier, section via integrations.drug_info,
sentence retrieval by keyword overlap (dosing sentences dropped), LLM only
rephrases the excerpt through gateway.phrase("med_answer") with forbidden-output
and added-content guards; dose questions and missing answers -> None + fallback text.
Refs #15
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
249 lines
12 KiB
Python
249 lines
12 KiB
Python
"""Answers to patient questions about their drug — ONLY from the ГРЛС instruction excerpt (SPEC §7 п.5).
|
||
|
||
Pipeline (deterministic except for the optional rephrasing step):
|
||
|
||
1. **Dose guard.** Any question about the dose ("можно увеличить дозу?") is never answered by the bot:
|
||
the personal dose is the doctor's decision -> `None` (caller says «уточните у врача»).
|
||
2. **Topic.** An explicit `topic` from the engine/rules wins (e.g. `quit_intent` -> "onset"); otherwise a
|
||
keyword classifier over the scrubbed question picks one of the supported section codes.
|
||
3. **Section.** The matching `DrugSection` is fetched via `integrations.drug_info` (lane B fills real
|
||
ГРЛС excerpts). No section for the topic -> `None`.
|
||
4. **Sentences.** The section is split into sentences; sentences with dosing regimens are dropped (we never
|
||
quote "до 200 мг" to a patient); the rest are scored by keyword overlap with the question, and the top
|
||
ones (kept in original order) form the excerpt. Best score 0 -> the excerpt has no answer -> `None`.
|
||
5. **Rephrasing.** The excerpt goes to `gateway.phrase("med_answer", ...)` with the excerpt as the base
|
||
meaning, so the gateway's forbidden-output filter and template fallback apply. On top of that we reject
|
||
rephrasings that add content not present in the excerpt (new numbers, mostly new words) and fall back to
|
||
the excerpt itself.
|
||
6. **Framing.** An intro ("По инструкции к препарату:") and a fixed doctor note from `config/texts/med_qa.yaml`
|
||
are added outside the LLM, so they are always present.
|
||
|
||
Logs never contain the question or the answer text — only codes.
|
||
"""
|
||
|
||
import logging
|
||
import re
|
||
from functools import lru_cache
|
||
from pathlib import Path
|
||
|
||
import yaml
|
||
from pydantic import BaseModel
|
||
|
||
from app.config import get_settings
|
||
from app.core.types import ScrubbedText
|
||
from app.integrations.base import DrugInfo, DrugSection, integrations
|
||
from app.llm.gateway import get_gateway, is_forbidden
|
||
|
||
log = logging.getLogger(__name__)
|
||
|
||
PHRASE_INTENT = "med_answer"
|
||
MAX_SENTENCES = 3 # the answer is meant to be 2–3 short sentences
|
||
MAX_EXCERPT_CHARS = 350
|
||
# Share of content-word stems in an LLM rephrasing that may be absent from the excerpt (and the question).
|
||
# Simplifying a sentence legitimately introduces some new words; a mostly new text means the model wrote
|
||
# its own answer instead of rephrasing the instruction, which SPEC §7 п.5 forbids.
|
||
MAX_NOVEL_STEM_SHARE = 0.5
|
||
STEM_LEN = 5
|
||
|
||
SECTIONS = ("side_effects", "onset", "alcohol", "discontinuation", "administration", "interactions", "overdose")
|
||
# Engine/rule codes accepted as `topic` (demo days 2, 5, 10) -> section code.
|
||
TOPIC_ALIASES = {
|
||
"nausea": "side_effects",
|
||
"side_effect": "side_effects",
|
||
"quit_intent": "onset",
|
||
"alcohol_interaction": "alcohol",
|
||
}
|
||
|
||
|
||
def _rx(*patterns: str) -> re.Pattern[str]:
|
||
return re.compile("|".join(patterns), re.IGNORECASE)
|
||
|
||
|
||
# Question -> topic. Order matters: the first match wins. Specific topics go first; "onset" precedes
|
||
# "discontinuation" because in the demo "от таблеток только хуже, брошу" is driven by the (delayed) effect.
|
||
TOPIC_CUES: tuple[tuple[str, re.Pattern[str]], ...] = (
|
||
("side_effects", _rx(r"тошн", r"ташн", r"мутит", r"рвот", r"побочк", r"побочн")),
|
||
("alcohol", _rx(r"алкогол", r"\bпив", r"выпи(ть|л|ва)", r"\bвин(о|а|ом)\b", r"водк", r"спиртн")),
|
||
("overdose", _rx(r"передоз", r"лишн\w* таблет", r"две таблетки сразу", r"больше,? чем назнач")),
|
||
("interactions", _rx(r"совмест", r"вместе с", r"други\w* (лекарств|препарат|таблет)", r"сочета")),
|
||
(
|
||
"onset",
|
||
_rx(r"подейств", r"когда (начн|будет|станет)", r"не помога", r"не действ", r"эффект", r"(только|стало) хуже"),
|
||
),
|
||
("discontinuation", _rx(r"брос", r"отмен", r"переста\w*", r"прекрат", r"отказ\w* от")),
|
||
("administration", _rx(r"как (правильно )?(принимать|пить)", r"(до|после|во время) еды", r"натощак")),
|
||
)
|
||
|
||
# Questions about the dose are always redirected to the doctor, whatever the instruction says.
|
||
DOSE_QUESTION = _rx(r"доз", r"\d+\s*мг", r"полтаблет", r"половин\w* таблет", r"(увелич|уменьш|сниз)\w* (приём|прием)")
|
||
|
||
# Sentences quoting a dosing regimen are never shown: the patient's dose is in the prescription, set by
|
||
# the doctor, and an instruction range ("до 200 мг/сут") read out of context reads like advice.
|
||
DOSING_SENTENCE = _rx(r"\d+\s*мг", r"доз\w*\s+(может|можно|следует|увелич|сниж|повыш|уменьш)", r"мг/сут")
|
||
|
||
# Section-level cues: a sentence that talks about the section's subject gets +1 even without word overlap,
|
||
# so an explicit topic with a vague question ("брошу") still finds e.g. the onset sentence.
|
||
SECTION_CUES: dict[str, re.Pattern[str]] = {
|
||
"side_effects": _rx(r"побочн", r"нежелательн", r"проход"),
|
||
"onset": _rx(r"эффект", r"недел", r"улучшени"),
|
||
"alcohol": _rx(r"алкогол", r"этанол", r"спиртн"),
|
||
"discontinuation": _rx(r"отмен", r"прекращ", r"синдром"),
|
||
"administration": _rx(r"еды", r"сутки", r"принима"),
|
||
"interactions": _rx(r"одновременн", r"совместн", r"ингибитор"),
|
||
"overdose": _rx(r"передозиров"),
|
||
}
|
||
|
||
# Symptom groups: prefixed verbs ("подташнивает") defeat prefix stemming, so a symptom named in the
|
||
# question is matched against sentences by group (+2 — the strongest relevance signal).
|
||
SYMPTOM_GROUPS: tuple[re.Pattern[str], ...] = (
|
||
_rx(r"тошн", r"ташн", r"мутит", r"рвот"),
|
||
_rx(r"\bсон\b", r"\bсна\b", r"бессонн", r"засып", r"уснуть", r"сонлив"),
|
||
_rx(r"головн\w* бол", r"голова болит", r"болит голова"),
|
||
_rx(r"головокруж", r"кружится"),
|
||
_rx(r"понос", r"диаре", r"желуд", r"живот"),
|
||
_rx(r"либидо", r"влечени", r"потенци", r"эякуляц"),
|
||
_rx(r"тревог", r"беспокойств", r"возбужден"),
|
||
_rx(r"пот(ливость|еет|ею)", r"потоотдел"),
|
||
)
|
||
|
||
_SENTENCE_SPLIT = re.compile(r"(?<=[.!?…])\s+")
|
||
_WORD = re.compile(r"[а-яёa-z]+", re.IGNORECASE)
|
||
_NUMBER = re.compile(r"\d+")
|
||
|
||
|
||
class MedAnswer(BaseModel):
|
||
text: str # patient-facing: intro + (rephrased) excerpt + doctor note
|
||
section: str # DrugSection.section code
|
||
source: str # ГРЛС reference from DrugSection.source
|
||
phrased_by_llm: bool # False when the excerpt is shown verbatim (mock, rejection, failure)
|
||
|
||
|
||
class MedQATexts(BaseModel):
|
||
fallback: str
|
||
intro: str
|
||
doctor_note: str
|
||
|
||
|
||
@lru_cache
|
||
def load_med_qa_texts(path: Path | None = None) -> MedQATexts:
|
||
path = path or get_settings().config_dir / "texts" / "med_qa.yaml"
|
||
return MedQATexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
|
||
|
||
|
||
def fallback_text() -> str:
|
||
"""«Уточните у врача» — what the caller sends when `med_qa` returns `None`."""
|
||
return load_med_qa_texts().fallback
|
||
|
||
|
||
def _norm(text: str) -> str:
|
||
return " ".join(text.lower().replace("ё", "е").split())
|
||
|
||
|
||
def _stems(text: str) -> set[str]:
|
||
# Crude Russian stemming: first STEM_LEN letters of words long enough to carry meaning (skips "не", "с").
|
||
return {w[:STEM_LEN] for w in _WORD.findall(_norm(text)) if len(w) >= 4}
|
||
|
||
|
||
def detect_topic(question: str) -> str | None:
|
||
text = _norm(question)
|
||
for topic, pattern in TOPIC_CUES:
|
||
if pattern.search(text):
|
||
return topic
|
||
return None
|
||
|
||
|
||
def _score(sentence: str, question: str, q_stems: set[str], section: str) -> int:
|
||
s = _norm(sentence)
|
||
score = sum(2 for g in SYMPTOM_GROUPS if g.search(question) and g.search(s))
|
||
cue = SECTION_CUES.get(section)
|
||
if cue is not None and cue.search(s):
|
||
score += 1
|
||
return score + len(q_stems & _stems(s))
|
||
|
||
|
||
def select_excerpt(section: DrugSection, question: str) -> str | None:
|
||
"""Most relevant sentences of the section (original order), or `None` if nothing is relevant."""
|
||
q = _norm(question)
|
||
q_stems = _stems(q)
|
||
sentences = [s.strip() for s in _SENTENCE_SPLIT.split(section.text.strip()) if s.strip()]
|
||
scored = [
|
||
(_score(s, q, q_stems, section.section), i, s)
|
||
for i, s in enumerate(sentences)
|
||
# Defence in depth: never surface a sentence the gateway would block if an LLM had written it.
|
||
if not DOSING_SENTENCE.search(s) and not is_forbidden(s)
|
||
]
|
||
relevant = [x for x in scored if x[0] > 0]
|
||
if not relevant:
|
||
return None
|
||
# Highest score first, ties by position; then restore reading order of the chosen sentences.
|
||
top = sorted(sorted(relevant, key=lambda x: (-x[0], x[1]))[:MAX_SENTENCES], key=lambda x: x[1])
|
||
return _trim(" ".join(s for _, _, s in top))
|
||
|
||
|
||
def _trim(text: str) -> str:
|
||
if len(text) <= MAX_EXCERPT_CHARS:
|
||
return text
|
||
cut = text[:MAX_EXCERPT_CHARS].rsplit(" ", 1)[0].rstrip(",;:")
|
||
return cut + "…"
|
||
|
||
|
||
def adds_content(answer: str, excerpt: str, question: str) -> bool:
|
||
"""True if a rephrasing introduces facts absent from the excerpt (numbers, mostly new words)."""
|
||
if not set(_NUMBER.findall(answer)) <= set(_NUMBER.findall(excerpt)):
|
||
return True # new numbers = new doses/terms/frequencies — never allowed
|
||
stems = _stems(answer)
|
||
if not stems:
|
||
return True
|
||
novel = stems - _stems(excerpt) - _stems(question)
|
||
return len(novel) / len(stems) > MAX_NOVEL_STEM_SHARE
|
||
|
||
|
||
async def _rephrase(excerpt: str, question: str, inn: str, topic: str) -> tuple[str, bool]:
|
||
text = await get_gateway().phrase(PHRASE_INTENT, {"excerpt": excerpt, "question": question, "drug": inn})
|
||
# The gateway returns the base (== excerpt, whitespace-normalised) on failure/filter; mock echoes it.
|
||
if " ".join(text.split()) == " ".join(excerpt.split()):
|
||
return excerpt, False
|
||
if adds_content(text, excerpt, question):
|
||
log.warning("med_qa rephrasing rejected: adds content topic=%s", topic)
|
||
return excerpt, False
|
||
return text, True
|
||
|
||
|
||
async def med_qa(
|
||
inn: str,
|
||
question: ScrubbedText,
|
||
*,
|
||
topic: str | None = None,
|
||
drug_info: DrugInfo | None = None,
|
||
) -> MedAnswer | None:
|
||
"""Answer about drug `inn` strictly from its ГРЛС excerpt; `None` -> caller sends `fallback_text()`."""
|
||
if DOSE_QUESTION.search(_norm(question)):
|
||
log.info("med_qa: dose question redirected to doctor")
|
||
return None
|
||
explicit = TOPIC_ALIASES.get(topic, topic) if topic else None
|
||
resolved = explicit if explicit in SECTIONS else detect_topic(question)
|
||
if resolved is None:
|
||
log.info("med_qa: topic not detected")
|
||
return None
|
||
try:
|
||
sections = await (drug_info or integrations.drug_info).sections(inn)
|
||
except Exception as exc:
|
||
log.warning("med_qa: drug_info failed error=%s", type(exc).__name__)
|
||
return None
|
||
section = next((s for s in sections if s.section == resolved), None)
|
||
if section is None:
|
||
log.info("med_qa: no section topic=%s", resolved)
|
||
return None
|
||
excerpt = select_excerpt(section, question)
|
||
if excerpt is None:
|
||
log.info("med_qa: no relevant sentence topic=%s", resolved)
|
||
return None
|
||
body, by_llm = await _rephrase(excerpt, question, inn, resolved)
|
||
texts = load_med_qa_texts()
|
||
return MedAnswer(
|
||
text=f"{texts.intro} {body}\n\n{texts.doctor_note}",
|
||
section=section.section,
|
||
source=section.source,
|
||
phrased_by_llm=by_llm,
|
||
)
|