ElebobaandClaude Opus 5.5
6dda5f9b30
CI / ruff · pytest (pull_request) Successful in 2m10s
Free-text treatment and medication questions go to an LLM agent with read-only tools bound to the current patient: guideline search (RAG, doses stripped), ЕМИАС record mock (diagnoses, comorbidities, intolerances, other specialists' drugs, current prescription), drug info (composition, contraindications, interactions) and a deterministic patient-specific check. Providers: OpenRouter (temporary, synthetic data only) and an offline mock; GigaChat to follow. Scrubbed input only, forbidden-output filter, any failure falls back to the existing extractive RAG / instruction answers. Off unless OPENROUTER_API_KEY or CHAT_AGENT=mock. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
108 lines
4.5 KiB
Python
108 lines
4.5 KiB
Python
"""Questions about the condition and its treatment → knowledge base of clinical guidelines (#64, RAG #60).
|
|
|
|
menu «Вопрос о лечении» ──▶ kb.question (free text) ──▶ rag.answer: KR quote + source, or «уточните у врача»
|
|
free text in idle that looks like a question (veteran_daily `text:idle`, after the check-in signals and the
|
|
drug question) ──▶ rag.answer, or the usual support phrase when the KB has no answer
|
|
|
|
The engine has already scrubbed the text, checked consent and run the safety check (ACUTE never gets here).
|
|
Questions about the patient's own drug stay with `med_qa` (instruction only, SPEC §7 п.5); this flow answers
|
|
general questions from the guidelines, extractively and without doses (`app.llm.rag.qa`). Logs carry codes only.
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
|
|
import yaml
|
|
from pydantic import BaseModel, ConfigDict
|
|
|
|
from app.config import get_settings
|
|
from app.dialog.contracts import Event, Reply
|
|
from app.dialog.flows.core import menu_reply
|
|
from app.dialog.fsm import Ctx, Step, abandonable, handler_for, state, trigger
|
|
from app.llm import agent, rag
|
|
from app.llm.med_qa import DOSE_QUESTION
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
QUESTION_STATE = "kb.question"
|
|
# `scrub()` placeholders and the punctuation around them: "[NAME], что такое ПТСР?" starts with "что такое".
|
|
_PLACEHOLDERS = re.compile(r"^(?:\s*\[[A-Z_]+\][\s,.:;!-]*)+")
|
|
|
|
|
|
class KnowledgeTexts(BaseModel):
|
|
model_config = ConfigDict(extra="forbid", frozen=True)
|
|
|
|
prompt: str
|
|
dose_redirect: str
|
|
question_starts: list[str]
|
|
small_talk: list[str]
|
|
|
|
|
|
@lru_cache
|
|
def load_knowledge_texts(path: Path | None = None) -> KnowledgeTexts:
|
|
path = path or get_settings().config_dir / "texts" / "knowledge.yaml"
|
|
return KnowledgeTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
|
|
|
|
|
|
@lru_cache
|
|
def _question_start() -> re.Pattern[str]:
|
|
starts = sorted(load_knowledge_texts().question_starts, key=len, reverse=True)
|
|
return re.compile(r"^\s*(?:" + "|".join(re.escape(s) for s in starts) + r")\b", re.IGNORECASE)
|
|
|
|
|
|
@lru_cache
|
|
def _small_talk() -> re.Pattern[str]:
|
|
return re.compile(r"\b(?:" + "|".join(re.escape(s) for s in load_knowledge_texts().small_talk) + r")\b", re.I)
|
|
|
|
|
|
def looks_like_question(text: str) -> bool:
|
|
"""A "?" anywhere or a question word at the start ("что такое флэшбек", "можно ли ..."); small talk
|
|
("как дела?") is not a question for the knowledge base."""
|
|
text = _PLACEHOLDERS.sub("", text)
|
|
if _small_talk().search(text):
|
|
return False
|
|
return "?" in text or bool(_question_start().search(text))
|
|
|
|
|
|
async def knowledge_answer(ctx: Ctx, *, explicit: bool) -> str | None:
|
|
"""KB answer for `ctx.text`. No answer → «уточните у врача» if the patient asked via the menu (`explicit`),
|
|
`None` otherwise (the caller keeps its own reply, e.g. the support phrase)."""
|
|
if ctx.text is None:
|
|
return None
|
|
if DOSE_QUESTION.search(ctx.text.lower()):
|
|
log.info("kb: dose question redirected to doctor")
|
|
return load_knowledge_texts().dose_redirect
|
|
# Tool-using agent first (RAG + ЕМИАС record + drug-safety check); disabled/failed → extractive RAG below.
|
|
smart = await agent.answer_chat(ctx.token, ctx.text)
|
|
if smart is not None:
|
|
log.info("kb: agent answered provider=%s tools=%s", smart.provider, ",".join(smart.tools))
|
|
return agent.render(smart)
|
|
reply = await rag.answer(ctx.text, audience="patient")
|
|
if reply is not None:
|
|
log.info("kb: answered docs=%s", ",".join(c.doc for c in reply.citations))
|
|
return reply.text
|
|
log.info("kb: no answer explicit=%s", explicit)
|
|
return rag.fallback_text("patient") if explicit else None
|
|
|
|
|
|
@trigger("button:menu:kb_question")
|
|
async def _menu_question(ctx: Ctx, event: Event) -> Step:
|
|
return Step(replies=[Reply(text=load_knowledge_texts().prompt)], next_state=QUESTION_STATE)
|
|
|
|
|
|
@state(QUESTION_STATE)
|
|
async def _question(ctx: Ctx, event: Event) -> Step:
|
|
if event.type != "text" or ctx.text is None:
|
|
# A menu button or a reminder instead of the question: drop the question and let idle handle it.
|
|
step = Step(next_state="idle")
|
|
idle = handler_for("idle")
|
|
return step.then(await idle(ctx, event)) if idle is not None else step
|
|
text = await knowledge_answer(ctx, explicit=True)
|
|
return Step(replies=[menu_reply(ctx, text)], next_state="idle")
|
|
|
|
|
|
# A question asked yesterday must not swallow today's message.
|
|
abandonable("kb")
|