"""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 ): # Both fields stay required. The formulary branches pediatric # dosing on BOTH — paracetamol prints an age band ("Trẻ em # 4-6 tuổi: 240 mg") *and* a weight rule ("10-50 kg: 15 # mg/kg") — so answering with only one of them would mean # picking a regimen the source does not let us pick. # # What changed on 2026-08-11 is the *question*, not the gate. # The fallback used to ask for age and weight every time, # including for whichever field the user had already given # (reproduced 5/5 live: "Bé 18 ký ...", "Bé nặng 18 kg ...", # "Trẻ 5 tuổi ..." all received the same sentence). The # effect was most visible when understanding.py did well: a # frame that parsed the weight and set needs_clarify=false # reaches this fallback, so a clearer question from the user # produced a more redundant question back. return AgentReply( "clarify", "missing_pediatric_age_or_weight", clarification=( frame.clarify_reason if frame.needs_clarify and frame.clarify_reason else _pediatric_clarify_question(frame) ), 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 = " và ".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 _pediatric_clarify_question(frame: QueryFrame) -> str: """Ask for the pediatric field that is actually missing. The caller still requires both age and weight; this only stops the question from asking for something the user already supplied in the very same sentence, which reads as not having been listened to and invites them to repeat themselves into the clarify-loop breaker. When a field IS known it is echoed back, so the user can see the value was received and correct it if the parse was wrong (e.g. a colloquial "18 ký" read as 18 kg). """ has_age = frame.age_text is not None has_weight = frame.weight_kg is not None if has_age and not has_weight: return f"Bé {frame.age_text} nặng bao nhiêu kg?" if has_weight and not has_age: return f"Bé nặng {_format_kg(frame.weight_kg)} kg, vậy bé bao nhiêu tuổi?" return "Bé bao nhiêu tuổi và cân nặng bao nhiêu kg?" def _format_kg(weight_kg: float) -> str: """Render a weight without a trailing '.0' on whole kilograms.""" return f"{weight_kg:g}" 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", "bé ", "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) + "."