Files
duocthu/apps/ai-service/rag/agent.py
T

584 lines
27 KiB
Python

"""The new RAG orchestrator — LLM understanding in, grounded answer out.
Replaces the old front-end wholesale:
- `CatalogDrugResolver` (fuzzy) + `SectionResolver` (keyword) -> `understanding.py`
- `ConversationalLoopService` + `conversation.py` (Focus / Summariser / manual
follow-up inheritance) -> the LLM reads a plain turn history and resolves
"thuốc đó" / "còn liều thì sao" itself.
What is deliberately KEPT because it is the safety spine, not the brittle part:
- `RetrievalService.retrieve_framed` (Qdrant section/overview retrieval, whole
section, provenance, quarantine `VERIFY_PDF`),
- `GroundedAnswerService.answer_from_result` (`grounding.verify` + entailment
on every generated claim; a configured generator that fails abstains rather
than degrading to a raw source dump).
This module owns routing only; it states no medical fact of its own.
"""
from __future__ import annotations
import logging
import time
from dataclasses import dataclass, replace
from typing import Protocol
from .answer import AnswerBlock, AnswerPlan, Citation, GroundedAnswerService
from .budget import RequestBudget
from .models import EvidenceDecision, RetrievalResult
from .policy import looks_non_human
from .service import RetrievalService
from .understanding import QueryFrame, QueryUnderstander
logger = logging.getLogger(__name__)
TUONG_TAC = "tuong_tac_thuoc"
HISTORY_TURNS = 6
# F-08: a normal answerable agent turn makes 3 sequential Bedrock calls
# (understand, generate, entailment). Input-field sufficiency is enforced by
# the structured state machine; generation still fails closed when retrieved
# evidence is insufficient. Defaults are sized with headroom, so ordinary
# traffic never trips the budget — it exists for the pathological case
# (a stuck/slow call, or an unexpectedly long retry chain), not to shave
# time off the common path.
MAX_WALL_CLOCK_MS = 40_000
MAX_LLM_CALLS_PER_TURN = 8
# Found live 2026-08-07 (50-question hand-typed browser audit): the
# understanding LLM can get stuck re-asking the same (or a near-identical)
# clarify question forever — reproduced 3 times independently, one case
# never converged after 5 real answered turns. `understanding.py`'s prior-
# frame merge (F-11) fixes most of the underlying cause, but a code-level
# backstop is still needed: nothing bounds a model that just keeps deciding
# needs_clarify=true no matter what. After this many CONSECUTIVE clarify
# turns on the same conversation, force a hard stop instead of asking again.
MAX_CONSECUTIVE_CLARIFY = 4
class ConversationStore(Protocol):
"""Durable, cross-worker alternative to the in-process history dict —
the gap ADR 0008 names as still open. Satisfied by
`adapters.postgres.PostgresConversationStore`; entirely optional — with
none configured, `RagAgent` behaves exactly as before this existed."""
def recent(self, conversation_id: str, limit: int) -> list[str]: ...
def append(self, conversation_id: str, line: str) -> None: ...
class AutocompleteSource(Protocol):
"""As-you-type suggestion, kept deterministic and independent of the LLM
understander — a prefix match needs no model call."""
def complete(self, prefix: str, k: int) -> list[str]: ...
@dataclass(frozen=True)
class AgentReply:
decision: str # answerable | abstain | clarify | verify_pdf
reason: str
answer: str | None = None
clarification: str | None = None
citations: tuple[Citation, ...] = ()
drugs: tuple[str, ...] = ()
turn_type: str = ""
generated: bool = False
quick_replies: tuple[str, ...] = ()
blocks: tuple[AnswerBlock, ...] = ()
answer_mode: str = "concise"
plan: AnswerPlan | None = None
class RagAgent:
def __init__(
self,
understander: QueryUnderstander,
retrieval: RetrievalService,
answers: GroundedAnswerService,
autocomplete: AutocompleteSource | None = None,
history_turns: int = HISTORY_TURNS,
max_wall_clock_ms: int = MAX_WALL_CLOCK_MS,
max_llm_calls_per_turn: int = MAX_LLM_CALLS_PER_TURN,
store: ConversationStore | None = None,
) -> None:
self._understander = understander
self._retrieval = retrieval
self._answers = answers
self._autocomplete = autocomplete
self._history_turns = history_turns
self._max_wall_clock_ms = max_wall_clock_ms
self._max_llm_calls_per_turn = max_llm_calls_per_turn
self._store = store
self._history: dict[str, list[str]] = {}
# In-process only, same durability caveat as `_history` (ADR 0008's
# named gap) — lost on restart, not shared across workers.
self._last_frame: dict[str, QueryFrame] = {}
self._clarify_streak: dict[str, int] = {}
def complete(self, prefix: str, k: int = 8) -> list[str]:
"""Display names matching a typed prefix, for input autocomplete."""
if self._autocomplete is None:
return []
return [_display_name(drug_id) for drug_id in self._autocomplete.complete(prefix, k)]
def handle(self, turn: str, conversation_id: str | None = None) -> AgentReply:
# F-08: one budget per turn, threaded through every LLM call this
# turn makes (understand, then whatever `_route` reaches).
t0 = time.monotonic()
budget = RequestBudget.start(self._max_wall_clock_ms, self._max_llm_calls_per_turn)
history = self._get_history(conversation_id)
t1 = time.monotonic()
prior_frame = self._last_frame.get(conversation_id) if conversation_id else None
frame = self._understander.understand(
turn, tuple(history), budget=budget, prior_frame=prior_frame
)
t2 = time.monotonic()
reply = self._route(turn, frame, budget)
reply = self._enforce_clarify_circuit_breaker(conversation_id, reply)
t3 = time.monotonic()
if conversation_id is not None:
self._remember(conversation_id, turn, reply)
# A clarify can originate downstream of understanding (the dose
# route invariant or evidence sufficiency). Persist that as an
# open frame too; otherwise the next short reply sees a prior
# frame marked complete and the structured merge cannot inherit
# the drug/population the user already supplied.
remembered_frame = frame
if reply.decision == "clarify" and reply.clarification:
remembered_frame = replace(
frame,
needs_clarify=True,
clarify_reason=reply.clarification,
quick_replies=reply.quick_replies,
)
self._last_frame[conversation_id] = remembered_frame
t4 = time.monotonic()
# Temporary instrumentation (2026-08-07): added specifically to
# pinpoint a live, reproduced-in-browser case of the FIRST LLM call
# already reporting the F-08 budget exhausted — i.e. >20s elapsed
# before even one Bedrock call was attempted, with no code between
# `RequestBudget.start()` and that first `budget.require()` that
# should plausibly take anywhere near that long. Logs unconditionally
# (not just on the slow path) so a normal turn's timing is on record
# too, for comparison.
# .warning, not .info: uvicorn's default logging config only wires
# handlers onto its own "uvicorn"/"uvicorn.access" loggers, not the
# root logger, so a plain .info() here would silently go nowhere —
# confirmed by `understanding.py`'s existing warning-level log
# already showing up in the same server output this session.
logger.warning(
"handle() timing: history=%.2fs understand=%.2fs route=%.2fs "
"remember=%.2fs total=%.2fs",
t1 - t0, t2 - t1, t3 - t2, t4 - t3, t4 - t0,
)
return reply
def _enforce_clarify_circuit_breaker(
self, conversation_id: str | None, reply: AgentReply
) -> AgentReply:
"""Hard stop after `MAX_CONSECUTIVE_CLARIFY` clarify turns in a row.
Every other failure mode in this file degrades to a bounded, honest
abstain — this is the one path that previously had no bound at all:
a model that keeps deciding needs_clarify=true has no natural exit,
and the user has no way out except abandoning the conversation. Any
non-clarify decision (answered, or a different abstain reason) resets
the streak — this only fires on genuinely consecutive clarifies.
"""
if conversation_id is None:
return reply
if reply.decision != "clarify":
self._clarify_streak.pop(conversation_id, None)
return reply
streak = self._clarify_streak.get(conversation_id, 0) + 1
if streak >= MAX_CONSECUTIVE_CLARIFY:
self._clarify_streak.pop(conversation_id, None)
return AgentReply(
"abstain", "clarify_loop_exhausted",
answer=(
"Hệ thống chưa xác định đủ thông tin sau nhiều lần hỏi lại. "
"Anh/chị vui lòng gõ lại TOÀN BỘ câu hỏi trong một tin nhắn "
"đầy đủ (tên thuốc, đối tượng, cân nặng/tuổi nếu có, đường "
"dùng), hoặc bấm \"Tạo phiên tra cứu mới\" để bắt đầu lại."
),
turn_type=reply.turn_type,
)
self._clarify_streak[conversation_id] = streak
return reply
def _get_history(self, conversation_id: str | None) -> list[str]:
if conversation_id is None:
return []
if self._store is not None:
try:
return self._store.recent(conversation_id, self._history_turns * 2)
except Exception:
# Fail-open (F-09's precedent): a store outage means this
# turn is understood fresh, with no memory of earlier ones —
# worse UX, not a 500. `routers/rag.py` wraps
# `PostgresTraceRepository.save()` the same bare way for the
# same reason.
return []
return self._history.get(conversation_id, [])
def _route(self, turn: str, frame: QueryFrame, budget: RequestBudget) -> AgentReply:
tt = frame.turn_type
section_overview = _is_section_overview(turn, frame)
if section_overview and not frame.section_overview:
frame = replace(frame, section_overview=True)
# Deterministic scope guard precedes every conversational clarify. A
# non-human dose must abstain, never ask which attribute/route and make
# an out-of-scope request look recoverable.
if looks_non_human(turn):
return AgentReply(
"abstain", "out_of_scope",
answer="Nội dung này nằm ngoài phần chuyên luận thuốc của Dược thư "
"(có thể thuộc phần hướng dẫn chung/phụ lục chưa được đưa vào). "
"Tôi chưa có dữ liệu để trả lời chính xác.",
turn_type=tt)
# Dosing is a small state machine, not an unconstrained model opinion.
# The LLM extracts the fields and can phrase/populate a useful initial
# clarify; code decides which core fields are actually required. Live
# testing caught the model asking an adult's weight repeatedly even
# after the user supplied a route, while previously skipping route and
# dumping oral + rectal regimens together.
if tt == "dosing_calc" and frame.drugs and not section_overview:
if frame.population is None:
return AgentReply(
"clarify", "missing_population",
clarification=(
frame.clarify_reason
if frame.needs_clarify and frame.clarify_reason
else "Anh/chị muốn tra liều cho người lớn hay trẻ em?"
),
drugs=frame.drugs, turn_type=tt,
quick_replies=(
frame.quick_replies
if frame.needs_clarify else ()
),
)
if frame.population in {"tre_em", "tre_so_sinh"} and (
frame.age_text is None or frame.weight_kg is None
):
return AgentReply(
"clarify", "missing_pediatric_age_or_weight",
clarification=(
frame.clarify_reason
if frame.needs_clarify and frame.clarify_reason
else "Bé bao nhiêu tuổi và cân nặng bao nhiêu kg?"
),
drugs=frame.drugs, turn_type=tt,
quick_replies=(
frame.quick_replies
if frame.needs_clarify else ()
),
)
# Route is intentionally not a universal required slot. Retrieval
# and the answer contract decide from the actual evidence whether
# omitting it is harmless (one applicable route -> answer now) or
# materially ambiguous (several routes -> model clarification and
# model-proposed quick replies). This prevents chip funnels for a
# question that was already precise enough to answer.
if (
frame.needs_clarify
and frame.clarify_reason
and tt != "dosing_calc"
and not section_overview
):
# `system_error` set means this isn't a real clarify at all — the
# understanding call itself failed (provider outage, malformed
# output) and failed closed to this same shape. Surface the real
# reason instead of the generic "needs_more_info" so a technical
# failure is distinguishable from an ordinary question back, both
# in the API response and in `/metrics`/traces (found live
# 2026-08-07: these were indistinguishable, which is why a real
# outage looked identical to normal clarify traffic).
if frame.system_error:
return AgentReply(
"abstain", frame.system_error,
answer=frame.clarify_reason,
drugs=frame.drugs, turn_type=tt,
)
return AgentReply(
"clarify", "needs_more_info",
clarification=frame.clarify_reason,
drugs=frame.drugs, turn_type=tt,
quick_replies=frame.quick_replies,
)
if tt == "drug_attribute" and frame.drugs and frame.attribute is None:
return AgentReply(
"clarify", "missing_attribute",
clarification=(
"Anh/chị muốn tra nội dung nào của thuốc này "
"(chỉ định, chống chỉ định, thận trọng, tác dụng phụ…)?"
),
drugs=frame.drugs, turn_type=tt,
)
if tt == "smalltalk":
return AgentReply(
"answerable", "smalltalk",
answer="Chào anh/chị! Em là trợ lý tra cứu Dược thư Quốc gia Việt Nam "
"2018, sẵn sàng hỗ trợ tra liều dùng, chống chỉ định, tương tác "
"thuốc... Anh/chị đang cần tra thuốc nào ạ?",
turn_type=tt)
if tt == "out_of_scope":
return AgentReply(
"abstain", "out_of_scope",
answer="Nội dung này nằm ngoài phần chuyên luận thuốc của Dược thư "
"(có thể thuộc phần hướng dẫn chung/phụ lục chưa được đưa vào). "
"Tôi chưa có dữ liệu để trả lời chính xác.",
turn_type=tt)
if not frame.drugs:
if frame.unknown_drugs:
names = ", ".join(frame.unknown_drugs)
return AgentReply(
"abstain", "drug_not_in_formulary",
answer=f"Không tìm thấy \"{names}\" trong Dược thư Quốc gia Việt Nam.",
turn_type=tt)
if tt == "symptom_to_drug":
if frame.indication:
return self._symptom_to_drug(turn, frame, budget)
return AgentReply(
"clarify", "no_indication",
clarification="Anh/chị mô tả triệu chứng hoặc chỉ định cần tra giúp "
"em với?",
turn_type=tt)
return AgentReply(
"clarify", "no_drug",
clarification="Anh/chị muốn tra thuốc nào?", turn_type=tt)
if tt == "interaction" and len(frame.drugs) >= 2:
return self._interaction(turn, frame, budget)
# drug_attribute / drug_overview / dosing_calc / fallback: one drug + section
return self._single_drug(turn, frame, budget)
def _single_drug(self, turn: str, frame: QueryFrame, budget: RequestBudget) -> AgentReply:
query = _synthesize_query(frame.standalone_query or turn, frame)
# `dosing_calc` semantically names the dosage section even when the
# understanding model leaves the separate `attribute` field null.
# Passing null here falls into an overview retrieval and was observed
# pulling interactions/precautions into a plain adult-dose answer.
section_key = (
"lieu_luong_va_cach_dung"
if frame.turn_type == "dosing_calc"
else frame.attribute
)
result = self._retrieval.retrieve_framed(
frame.drugs[0], section_key, query,
is_overview=frame.turn_type == "drug_overview",
)
return self._grounded(query, result, frame, budget=budget)
def _interaction(self, turn: str, frame: QueryFrame, budget: RequestBudget) -> AgentReply:
"""Gather the interaction section of each named drug and synthesise.
Absence of a match is stated as "not found in each drug's interaction
section", never as "safe" — the answer layer's grounding still applies.
A quarantined table/formula in EITHER drug's interaction section
forces the whole combined answer to VERIFY_PDF via
`RetrievalService.decide` — the same policy the single-drug path
already applies to a section with quarantined content. Previously
this only kept `part.decision == ANSWERABLE` parts, so a quarantined
drug's evidence (and the "table exists, verify PDF" notice it must
produce per the quarantine contract) was silently dropped instead of
surfaced; a confident interaction answer could omit exactly the
unverified contraindication table it should have flagged. Generating
a synthesis claim from one verified and one unverified source is not
safer than generating from either alone, so both now block
generation the same way.
"""
evidences = []
query = _synthesize_query(frame.standalone_query or turn, frame)
for drug_id in frame.drugs:
part = self._retrieval.retrieve_framed(drug_id, TUONG_TAC, query)
if part.decision in (EvidenceDecision.ANSWERABLE, EvidenceDecision.VERIFY_PDF):
evidences.extend(part.evidence)
if not evidences:
listed = "".join(frame.drugs)
return AgentReply(
"abstain", "no_interaction_evidence",
answer=f"Không tìm thấy mục tương tác thuốc cho {listed} trong Dược "
"thư. Điều này KHÔNG có nghĩa là an toàn khi phối hợp.",
drugs=frame.drugs, turn_type=frame.turn_type)
combined = self._retrieval.decide(tuple(evidences))
return self._grounded(query, combined, frame, budget=budget)
def _symptom_to_drug(
self, turn: str, frame: QueryFrame, budget: RequestBudget
) -> AgentReply:
"""Reverse lookup: a symptom/indication -> which drugs' `chi_dinh`
actually names it. A factual list from the formulary, not a
treatment ranking or recommendation — no drug is preferred over
another here, only cited as indicated ([[feedback_no_recommendation_gate]]:
this audience is doctors/pharmacists, a lookup like this is normal use).
Absence is stated plainly, never as "no such drug exists" — the
formulary may simply not name this indication under any monograph.
"""
result = self._retrieval.retrieve_by_indication(frame.indication)
if result.decision == EvidenceDecision.ABSTAIN:
return AgentReply(
"abstain", result.reason,
answer=f"Không tìm thấy thuốc nào trong Dược thư Quốc gia Việt Nam ghi "
f"nhận chỉ định cho \"{frame.indication}\". Điều này KHÔNG có "
"nghĩa là không có thuốc điều trị — vui lòng tra theo tên thuốc "
"cụ thể nếu đã biết.",
turn_type=frame.turn_type)
# `matched_doc_id` is always `{drug_id}__chi_dinh__{part_index}` — the
# drugs actually found, not `frame.drugs` (empty by construction for
# this turn_type; the router only reaches here with no named drug).
matched_drugs = tuple(dict.fromkeys(
evidence.matched_doc_id.split("__")[0] for evidence in result.evidence
))
return self._grounded(
turn, result, frame, drugs=matched_drugs, list_mode=True, budget=budget
)
def _grounded(
self, turn: str, result: RetrievalResult, frame: QueryFrame,
drugs: tuple[str, ...] | None = None, list_mode: bool = False,
budget: RequestBudget | None = None,
) -> AgentReply:
ga = self._answers.answer_from_result(
turn, result, list_mode=list_mode, budget=budget, prechecked=True
)
decision = ga.result.decision.value
if ga.clarification is not None:
decision = "clarify"
reason = "needs_more_info" if ga.clarification is not None else ga.result.reason
return AgentReply(
decision=decision,
reason=reason,
answer=ga.answer,
clarification=ga.clarification,
citations=ga.citations,
drugs=drugs if drugs is not None else frame.drugs,
turn_type=frame.turn_type,
generated=ga.generated,
quick_replies=ga.quick_replies,
blocks=ga.blocks,
answer_mode=ga.answer_mode,
plan=ga.plan,
)
def _remember(self, conversation_id: str, turn: str, reply: AgentReply) -> None:
lines = [f"Người dùng: {turn}"]
spoken = reply.answer or reply.clarification
if spoken:
lines.append(f"Trợ lý: {spoken[:300]}")
if self._store is not None:
try:
for line in lines:
self._store.append(conversation_id, line)
except Exception:
# Fail-open: this turn's memory is lost, not the response
# already computed and about to be returned to the caller.
pass
return
history = self._history.setdefault(conversation_id, [])
history.extend(lines)
# Keep only the recent window; the LLM re-reads it every turn. Only
# needed for the in-process dict — the store path windows at READ
# time instead (`recent(..., limit)`), so old rows just sit unused
# rather than needing a delete on every turn.
excess = len(history) - self._history_turns * 2
if excess > 0:
del history[:excess]
def _display_name(drug_id: str) -> str:
"""A readable display name from a drug id ('paracetamol_acetaminophen')."""
return drug_id.replace("_", " ").title()
def _is_section_overview(turn: str, frame: QueryFrame) -> bool:
"""Separate a handbook survey from a patient-specific decision.
The model supplies the first-class flag, while the narrow lexical
backstop makes a clear section lookup deterministic. A personal target
always wins: broad words inside a patient-specific dose question must not
suppress a necessary clarification.
"""
text = turn.casefold()
personal_cues = (
"cho tôi", "tôi đang", "bệnh nhân này", "ca này", "", "trẻ ",
"tuổi", "cân nặng", " kg", "suy thận", "suy gan", "nên dùng liều nào",
)
if any(cue in text for cue in personal_cues):
return False
overview_cues = (
"dược thư hướng dẫn", "những ", "các ", "toàn bộ", "tất cả",
"trình bày", "theo nhóm", "theo tần suất", "nếu có", "tổng quan",
)
return frame.section_overview or any(cue in text for cue in overview_cues)
_POPULATION_LABELS = {
"tre_em": "trẻ em",
"tre_so_sinh": "trẻ sơ sinh",
"nguoi_lon": "người lớn",
"nguoi_cao_tuoi": "người cao tuổi",
"phu_nu_co_thai": "phụ nữ có thai",
"phu_nu_cho_con_bu": "phụ nữ cho con bú",
"suy_than": "suy thận",
"suy_gan": "suy gan",
}
_ROUTE_LABELS = {
"uong": "uống",
"tiem_tinh_mach": "tiêm tĩnh mạch",
"tiem_bap": "tiêm bắp",
"tiem_duoi_da": "tiêm dưới da",
"dat_truc_trang": "đặt trực tràng",
"boi_ngoai_da": "bôi ngoài da",
"nho_mat": "nhỏ mắt",
"nho_mui": "nhỏ mũi",
"khac": "khác",
}
def _synthesize_query(turn: str, frame: QueryFrame) -> str:
"""Fold the structured context `understanding.py` resolved — possibly
across several turns — into one self-contained question.
`GroundedAnswerService.answer_from_result` has no conversation history of
its own; the `query` string it receives IS the entire context its
generation LLM call sees (and the legacy sufficiency call, if enabled).
Passing the bare current
turn loses everything resolved earlier: a reply like "Uống" answering a
route question three turns into a dose conversation would reach
generation as just "Uống", indistinguishable from a user who typed
nothing else — the two P0s the 2026-08-06 audit named (population/
weight/age/route extracted but discarded downstream) are exactly this
gap. Redundant when the turn is already self-contained (a fresh
single-shot question re-states its own population/route, so this just
repeats it) — harmless, since omission is the failure mode, not
repetition.
"""
parts = [turn]
if frame.section_overview:
parts.append(
"Phạm vi yêu cầu: tra cứu tổng quan toàn mục; trình bày các nhánh "
"trong Dược thư với nhãn điều kiện, không chọn một phác đồ cho một người bệnh"
)
if frame.population:
parts.append(f"Đối tượng: {_POPULATION_LABELS.get(frame.population, frame.population)}")
if frame.age_text:
parts.append(f"Tuổi: {frame.age_text}")
if frame.weight_kg is not None:
parts.append(f"Cân nặng: {frame.weight_kg:g} kg")
if frame.route:
parts.append(f"Đường dùng: {_ROUTE_LABELS.get(frame.route, frame.route)}")
if frame.indication:
parts.append(f"Chỉ định/triệu chứng: {frame.indication}")
if len(parts) == 1:
return turn
return ". ".join(parts) + "."