ElebobaandClaude Opus 5.5
6a77831603
CI / ruff · pytest (pull_request) Successful in 1m30s
- docs/source/rag: PDF sources (KR PTSD 2023, KR depression 2024, meds module spec), pdf->txt converter (tables as markdown, page markers, hand-transcribed figures), word-level check against pdftotext; fixes HAM-D/MADRS fake headers and split rows - config/rag: corpus txt + sources.yaml (documents, audiences, retrieval params, synonyms) - app/llm/rag: Snowball stemmer, section-aware chunker with page citations, BM25 + synonym expansion + densest-passage reranking, extractive answer() with coverage gate, patient dose/prescribing filter, optional LLM rephrasing rejected when it adds content; CLI python -m app.llm.rag - tests: gold retrieval set (hit@3 >= 15/18), off-topic, safety, faithfulness - ADR-011 Refs #60 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
53 lines
2.2 KiB
Python
53 lines
2.2 KiB
Python
"""Manual check of the knowledge base on synthetic questions.
|
|
|
|
python -m app.llm.rag "через сколько начинают действовать антидепрессанты"
|
|
python -m app.llm.rag --audience doctor "синдром отмены антидепрессантов"
|
|
python -m app.llm.rag --search "ДПДГ при ПТСР" # raw retrieval: chunks, scores, passages
|
|
python -m app.llm.rag --stats # corpus summary
|
|
|
|
The question always goes through `privacy.scrub()` first; the CLI never bypasses it.
|
|
"""
|
|
|
|
import argparse
|
|
import asyncio
|
|
from collections import Counter
|
|
|
|
from app.core import privacy
|
|
from app.llm.gateway import get_gateway
|
|
from app.llm.rag.index import get_index
|
|
from app.llm.rag.qa import answer, fallback_text
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(prog="python -m app.llm.rag")
|
|
parser.add_argument("question", nargs="?")
|
|
parser.add_argument("--audience", choices=["patient", "doctor", "team"], default="patient")
|
|
parser.add_argument("--search", action="store_true", help="show retrieved chunks instead of an answer")
|
|
parser.add_argument("--top", type=int, default=5)
|
|
parser.add_argument("--stats", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
|
|
index = get_index()
|
|
if args.stats:
|
|
by_doc = Counter(c.doc.short for c in index.chunks)
|
|
print(f"chunks={len(index.chunks)}", ", ".join(f"{d}: {n}" for d, n in by_doc.items()))
|
|
return 0
|
|
if not args.question:
|
|
parser.error("question is required")
|
|
question = privacy.scrub(args.question)
|
|
|
|
if args.search:
|
|
for hit in index.search(question, audience=args.audience, top_k=args.top, min_score=0):
|
|
print(f"{hit.score:6.1f} (bm25 {hit.bm25:5.1f}) {hit.chunk.citation} {hit.chunk.section}")
|
|
print(f" > {hit.passage[:300]}")
|
|
return 0
|
|
|
|
print(f"provider={get_gateway().provider_name} audience={args.audience}\n")
|
|
reply = asyncio.run(answer(question, audience=args.audience, index=index))
|
|
print(reply.text if reply else fallback_text(args.audience))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|