Merge pull request 'feat(bot): med_qa — ответы о препарате только по выдержкам ГРЛС' (#48) from feat/15-med-qa into dev
CI / ruff · pytest (push) Successful in 25s

This commit was merged in pull request #48.
This commit is contained in:
2026-10-03 15:33:55 +03:00
4 changed files with 491 additions and 0 deletions
+248
View File
@@ -0,0 +1,248 @@
"""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,
)
+12
View File
@@ -0,0 +1,12 @@
# Тексты ответов о препарате (med_qa, SPEC §7 п.5). Зона clinical вычитывает формулировки.
# Сам ответ берётся ТОЛЬКО из выдержки инструкции ГРЛС; здесь — лишь обрамление и запасной текст.
# Правила: без советов по дозам, без рекомендаций препаратов, решение об изменениях — за врачом.
# Если в инструкции нет ответа (нет раздела / нет подходящего фрагмента / вопрос о дозе).
fallback: "В инструкции к препарату нет ответа на этот вопрос. Уточните, пожалуйста, у вашего врача."
# Вступление перед выдержкой — пациент должен понимать, что это текст инструкции, а не совет бота.
intro: "По инструкции к препарату:"
# Всегда добавляется в конце ответа (не проходит через LLM).
doctor_note: "Не меняйте схему приёма и не прекращайте его самостоятельно — любые изменения решает ваш врач."
+2
View File
@@ -19,6 +19,8 @@ phrases:
sleep_support: "Спасибо, что рассказали про сон. Я отмечу это, и ваш врач увидит, как вы спите." sleep_support: "Спасибо, что рассказали про сон. Я отмечу это, и ваш врач увидит, как вы спите."
book_visit_offer: "Предлагаю записаться к врачу пораньше. Выбрать удобное время можно кнопкой ниже." book_visit_offer: "Предлагаю записаться к врачу пораньше. Выбрать удобное время можно кнопкой ниже."
doctor_notified: "Я передал информацию вашему врачу. Он свяжется с вами." doctor_notified: "Я передал информацию вашему врачу. Он свяжется с вами."
# med_qa: базовый смысл = выдержка из инструкции ГРЛС; LLM только упрощает её, ничего не добавляя.
med_answer: "{excerpt}"
# Шаблон саммари для врача (только по структурированным данным). # Шаблон саммари для врача (только по структурированным данным).
summary: summary:
+229
View File
@@ -0,0 +1,229 @@
"""med_qa: answers about the drug only from the instruction excerpt (SPEC §7 п.5), demo days 2/5/10."""
import pytest
from app.core.types import ScrubbedText
from app.integrations.base import DrugSection
from app.llm import med_qa as med_qa_module
from app.llm.gateway import LLMGateway, is_forbidden
from app.llm.med_qa import MedAnswer, adds_content, detect_topic, fallback_text, med_qa, select_excerpt
from app.llm.mock import MockTasks
S = ScrubbedText
SRC = "ГРЛС: тестовая выдержка (синтетика)"
# Synthetic, simplified test excerpts — NOT real ГРЛС data (lane B owns data/meds/grls).
FAKE_SERTRALINE = {
"side_effects": (
"Наиболее частые нежелательные реакции в начале терапии: тошнота, диарея, сухость во рту. "
"Тошнота обычно возникает в первые дни приёма и проходит в течение 1–2 недель. "
"Для уменьшения тошноты препарат принимают после еды. "
"Возможны нарушения сна и головная боль."
),
"onset": (
"Терапевтический эффект развивается постепенно. "
"Улучшение обычно отмечается через 2–4 недели регулярного приёма. "
"Суточная доза может увеличиваться врачом до 200 мг."
),
"alcohol": (
"Совместный приём с алкоголем не рекомендуется. "
"Алкоголь может ухудшать сон и настроение и усиливать побочные эффекты."
),
"discontinuation": (
"Резкое прекращение приёма может вызвать синдром отмены: головокружение, тревогу, нарушения сна. "
"Отмену проводят постепенно под контролем врача."
),
}
class FakeDrugInfo:
def __init__(self, sections: dict[str, str] | None = None) -> None:
self._sections = FAKE_SERTRALINE if sections is None else sections
self.calls: list[str] = []
async def sections(self, inn: str) -> list[DrugSection]:
self.calls.append(inn)
if inn != "сертралин":
return []
return [DrugSection(inn=inn, section=k, text=v, source=SRC) for k, v in self._sections.items()]
class BrokenDrugInfo:
async def sections(self, inn: str) -> list[DrugSection]:
raise RuntimeError("grls down")
class FixedPhraseTasks(MockTasks):
"""Simulates an LLM that returns a fixed rephrasing."""
name = "fake-llm"
def __init__(self, reply: str) -> None:
self.reply = reply
async def phrase(self, intent: str, payload: ScrubbedText, base_text: str) -> str | None:
return self.reply
@pytest.fixture
def info() -> FakeDrugInfo:
return FakeDrugInfo()
@pytest.fixture
def mock_gateway(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(med_qa_module, "get_gateway", lambda: LLMGateway(MockTasks()))
def use_llm(monkeypatch: pytest.MonkeyPatch, reply: str) -> None:
monkeypatch.setattr(med_qa_module, "get_gateway", lambda: LLMGateway(FixedPhraseTasks(reply)))
def assert_safe(answer: MedAnswer) -> None:
assert not is_forbidden(answer.text)
assert "мг" not in answer.text # never quotes dosing regimens
assert "врач" in answer.text # doctor note always present
# --- topic detection --------------------------------------------------------------------------------
@pytest.mark.parametrize(
("question", "topic"),
[
("подташнивает с утра", "side_effects"),
("опять тошнит", "side_effects"),
("тошнота не проходит", "side_effects"),
("от таблеток только хуже, брошу", "onset"),
("когда подействует?", "onset"),
("без пива не могу уснуть", "alcohol"),
("можно ли выпить вина?", "alcohol"),
("хочу бросить таблетки", "discontinuation"),
("как принимать, до еды или после еды?", "administration"),
("какая сегодня погода", None),
],
)
def test_detect_topic(question: str, topic: str | None) -> None:
assert detect_topic(question) == topic
async def test_explicit_topic_wins(info: FakeDrugInfo, mock_gateway: None) -> None:
# The classifier would say "discontinuation"; the engine decided the answer should be about onset.
answer = await med_qa("сертралин", S("брошу"), topic="onset", drug_info=info)
assert answer is not None and answer.section == "onset"
async def test_engine_alias_topic(info: FakeDrugInfo, mock_gateway: None) -> None:
answer = await med_qa("сертралин", S("ну и что"), topic="alcohol_interaction", drug_info=info)
assert answer is not None and answer.section == "alcohol"
# --- demo days ---------------------------------------------------------------------------------------
async def test_day2_nausea(info: FakeDrugInfo, mock_gateway: None) -> None:
answer = await med_qa("сертралин", S("подташнивает с утра"), drug_info=info)
assert answer is not None
assert answer.section == "side_effects" and answer.source == SRC
assert "1–2 недель" in answer.text and "после еды" in answer.text
assert answer.phrased_by_llm is False # mock echoes the excerpt
assert_safe(answer)
async def test_day5_quit(info: FakeDrugInfo, mock_gateway: None) -> None:
answer = await med_qa("сертралин", S("от таблеток только хуже, брошу"), topic="quit_intent", drug_info=info)
assert answer is not None and answer.section == "onset"
assert "2–4 недели" in answer.text
assert "200" not in answer.text # dosing sentence is dropped
assert "не прекращайте" in answer.text
assert_safe(answer)
async def test_day10_alcohol(info: FakeDrugInfo, mock_gateway: None) -> None:
answer = await med_qa("сертралин", S("без пива не могу уснуть"), drug_info=info)
assert answer is not None and answer.section == "alcohol"
assert "не рекомендуется" in answer.text
assert_safe(answer)
# --- retrieval -----------------------------------------------------------------------------------------
def test_select_excerpt_picks_relevant_sentences() -> None:
section = DrugSection(inn="сертралин", section="side_effects", text=FAKE_SERTRALINE["side_effects"], source=SRC)
excerpt = select_excerpt(section, "подташнивает с утра")
assert excerpt is not None
assert "тошнот" in excerpt.lower()
assert "головная боль" not in excerpt
def test_select_excerpt_none_when_not_relevant() -> None:
section = DrugSection(inn="сертралин", section="side_effects", text="Возможна сухость во рту.", source=SRC)
assert select_excerpt(section, "болит колено") is None
async def test_missing_section_returns_none(mock_gateway: None) -> None:
info = FakeDrugInfo({"side_effects": FAKE_SERTRALINE["side_effects"]})
assert await med_qa("сертралин", S("можно пиво?"), drug_info=info) is None
assert "врач" in fallback_text()
async def test_unknown_drug_and_unknown_topic(info: FakeDrugInfo, mock_gateway: None) -> None:
assert await med_qa("флуоксетин", S("тошнит"), drug_info=info) is None
assert await med_qa("сертралин", S("какая погода"), drug_info=info) is None
async def test_default_registry_has_no_data(mock_gateway: None) -> None:
# Before lane B (#14) registers real excerpts the default NoDrugInfo returns nothing.
assert await med_qa("сертралин", S("тошнит")) is None
async def test_drug_info_failure_returns_none(mock_gateway: None) -> None:
assert await med_qa("сертралин", S("тошнит"), drug_info=BrokenDrugInfo()) is None
@pytest.mark.parametrize("question", ["можно увеличить дозу?", "можно пить 100 мг?", "можно пить полтаблетки?"])
async def test_dose_questions_go_to_doctor(info: FakeDrugInfo, mock_gateway: None, question: str) -> None:
assert await med_qa("сертралин", S(question), topic="administration", drug_info=info) is None
assert info.calls == []
# --- LLM rephrasing guard --------------------------------------------------------------------------
async def test_forbidden_llm_output_falls_back_to_excerpt(info: FakeDrugInfo, monkeypatch: pytest.MonkeyPatch) -> None:
use_llm(monkeypatch, "Тошнота пройдёт. Уменьшите дозу сертралина вдвое.")
answer = await med_qa("сертралин", S("подташнивает с утра"), drug_info=info)
assert answer is not None and answer.phrased_by_llm is False
assert "Тошнота обычно возникает" in answer.text
assert_safe(answer)
async def test_llm_adding_numbers_is_rejected(info: FakeDrugInfo, monkeypatch: pytest.MonkeyPatch) -> None:
use_llm(monkeypatch, "Тошнота обычно проходит за 3 дня.")
answer = await med_qa("сертралин", S("подташнивает с утра"), drug_info=info)
assert answer is not None and answer.phrased_by_llm is False
assert "3 дня" not in answer.text
async def test_llm_inventing_content_is_rejected(info: FakeDrugInfo, monkeypatch: pytest.MonkeyPatch) -> None:
use_llm(monkeypatch, "Попейте имбирный чай, погуляйте вечером и съешьте кусочек лимона.")
answer = await med_qa("сертралин", S("подташнивает с утра"), drug_info=info)
assert answer is not None and answer.phrased_by_llm is False
assert "имбир" not in answer.text
async def test_faithful_rephrasing_is_accepted(info: FakeDrugInfo, monkeypatch: pytest.MonkeyPatch) -> None:
reply = "Тошнота обычно бывает в первые дни приёма и проходит за 1–2 недели. Препарат принимают после еды."
use_llm(monkeypatch, reply)
answer = await med_qa("сертралин", S("подташнивает с утра"), drug_info=info)
assert answer is not None and answer.phrased_by_llm is True
assert reply in answer.text
assert_safe(answer)
def test_adds_content() -> None:
excerpt = "Тошнота обычно проходит в течение 1–2 недель."
assert adds_content("Тошнота проходит за 5 недель.", excerpt, "тошнит") is True
assert adds_content("Тошнота обычно проходит в течение 1–2 недель.", excerpt, "тошнит") is False
assert adds_content("", excerpt, "тошнит") is True