fix(llm): RAG citation pages, scrub placeholders, disease-aware document choice
CI / ruff · pytest (pull_request) Successful in 1m2s

- converter marks a paragraph glued across a page break with ⟦стр. N⟧; the chunker
  strips the marker and extends the chunk's page range (wrong-page citations 24 -> 3,
  the rest are scale-form table rows split by a page)
- [NAME]/[PHONE] placeholders from privacy.scrub() no longer count as unknown words
  (a scrubbed question used to fall below the coverage gate)
- a question naming one disease boosts its guideline in retrieval and passage choice
  ("ТМС при депрессии" was answered from the PTSD guideline); "антидепрессант" is not a cue

Refs #62

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-04 04:31:59 +03:00
co-authored by Claude Opus 5.5
parent 6c212d560c
commit fe29ebd629
8 changed files with 206 additions and 91 deletions
+61 -8
View File
@@ -9,7 +9,8 @@ The txt files come from `docs/source/rag/pdf_to_txt.py`: paragraphs are separate
* "Комментарии: ..." after a recommendation stays with the recommendation (the comment explains it);
* a line starting in lowercase continues the previous paragraph (cut by a page break the converter missed);
* tables are split by rows, every table piece repeats the table header, so a chunk is readable alone;
* each chunk keeps the page range it came from, for the citation "[КР ПТСР 2023, стр. 45]".
* each chunk keeps the page range it came from, for the citation "[КР ПТСР 2023, стр. 45]"; inline
"⟦стр. N⟧" markers (paragraph glued across a page break) are removed from the text but move the range.
"""
import re
@@ -25,6 +26,8 @@ from app.config import get_settings
Audience = str # "doctor" | "patient" | "team"
PAGE_MARKER = re.compile(r"^--- стр\. (\d+) ---$")
# A paragraph that flows onto the next page is glued by the converter; "⟦стр. N⟧" marks where page N begins.
INLINE_PAGE = re.compile(r"\s*⟦стр\. (\d+)⟧\s*")
TABLE_DIVIDER = re.compile(r"^\|(-+\|)+$")
# "2.1 Жалобы и анамнез", "3.1.2. Купирующая терапия ..." — a numbered subsection title.
SUBSECTION = re.compile(r"^(\d{1,2}(?:\.\d{1,2})+)\.?\s+([А-ЯЁA-Z].*)$") # not dates "28.07.2022 N ..."
@@ -77,6 +80,9 @@ class RetrievalConfig(BaseModel):
window_weight: float = 0.5
# section title prefix -> score multiplier, per audience ({"patient": {"Приложение В": 1.5}})
section_boost: dict[str, dict[str, float]] = Field(default_factory=dict)
# doc id -> regexes naming its disease; a question naming one disease favours that document
doc_cues: dict[str, list[str]] = Field(default_factory=dict)
doc_cue_boost: float = 1.0
class AnswerConfig(BaseModel):
@@ -120,8 +126,46 @@ class _Unit:
"""Indivisible piece of a chunk: a paragraph (with its comment) or one table row."""
text: str
page: int
page: int # page where the unit starts
table_header: str | None = None # "| a | b |\n|---|---|" for table rows
breaks: list[tuple[int, int]] = field(default_factory=list) # (offset in text, page starting there)
@property
def page_end(self) -> int:
return self.breaks[-1][1] if self.breaks else self.page
def page_at(self, offset: int) -> int:
page = self.page
for start, p in self.breaks:
if start <= offset:
page = p
return page
def append(self, line: str, page: int, sep: str) -> None:
"""Glue a continuation line; record a page break if it comes from a later page."""
base = len(self.text) + len(sep)
text, breaks = _strip_inline_pages(line)
if page > self.page_end:
self.breaks.append((base, page))
self.breaks.extend((base + off, p) for off, p in breaks)
self.text += sep + text
def _strip_inline_pages(line: str) -> tuple[str, list[tuple[int, int]]]:
"""Text without "⟦стр. N⟧" markers + (offset, page) of each marker in the cleaned text."""
out, breaks, pos = [], [], 0
for m in INLINE_PAGE.finditer(line):
out.append(line[pos : m.start()])
breaks.append((sum(len(x) for x in out) + 1, int(m.group(1))))
out.append(" ")
pos = m.end()
out.append(line[pos:])
return "".join(out), breaks
def _new_unit(line: str, page: int) -> _Unit:
text, breaks = _strip_inline_pages(line)
return _Unit(text, page, breaks=breaks)
@dataclass
@@ -185,7 +229,7 @@ def _sections(text: str, skip: list[str]) -> list[_Section]:
continue
if first.startswith("[Изображение"):
if not skipping: # transcribed figure: one unit, keeps its own line breaks
sections[-1].units.append(_Unit("\n".join(block[1:]) if len(block) > 1 else first, page))
sections[-1].units.append(_new_unit("\n".join(block[1:]) if len(block) > 1 else first, page))
continue
for j, line in enumerate(block):
title = heading(line)
@@ -199,11 +243,11 @@ def _sections(text: str, skip: list[str]) -> list[_Section]:
continue
units = sections[-1].units
if COMMENT.match(line) and units and units[-1].table_header is None:
units[-1].text += "\n" + line
units[-1].append(line, page, "\n")
elif line[0].islower() and units and units[-1].table_header is None:
units[-1].text += " " + line # paragraph continued after a page break or a figure
units[-1].append(line, page, " ") # paragraph continued after a page break or a figure
else:
units.append(_Unit(line, page))
units.append(_new_unit(line, page))
return [s for s in sections if s.units]
@@ -236,7 +280,16 @@ def _split_long(unit: _Unit, max_chars: int) -> list[_Unit]:
parts[-1] += " " + sentence
else:
parts.append(sentence)
return [_Unit(p, unit.page) for p in parts]
out: list[_Unit] = []
pos = 0
for part in parts:
start = unit.text.find(part[:40], pos)
start = pos if start < 0 else start
end = start + len(part) - 1
first, last = unit.page_at(start), unit.page_at(end)
out.append(_Unit(part, first, breaks=[(len(part) - 1, last)] if last != first else []))
pos = start + len(part)
return out
def _render(units: list[_Unit]) -> str:
@@ -280,7 +333,7 @@ def chunk_document(doc: DocumentMeta, text: str, cfg: ChunkingConfig) -> list[Ch
doc=doc,
section=section.title,
page_start=group[0].page,
page_end=group[-1].page,
page_end=max(u.page_end for u in group),
text=_render(group),
)
)