feat(llm): valuation evidence block, SSE payload and code-cited rule sources (refs P48)

src/llm/evidence.py

@@ -0,0 +1,168 @@
+"""Structured evidence for the chat: code valuations beside the excerpts.
+
+``valuation_evidence`` detects HCPCS/CPT codes (and registered families)
+in the question, reads their RVUs from the DuckDB read replica and
+attaches conversion factors from ``pfs.rules``. ``code_cited_sources``
+pulls rule chunks whose ``codes`` metadata mention those codes straight
+from pgvector so the answer can cite where CMS valued them. Both degrade
+to "no evidence" with a warning — the chat never fails because of them.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Sequence
+
+import duckdb
+from sqlalchemy import text
+
+from llm.config import LlmConfig
+from llm.links import as_source
+from pfs.families import detect_codes
+from pfs.valuation import ValuationRow, valuation
+
+log = logging.getLogger(__name__)
+
+_HEADER = (
+    "Valuation (authoritative for RVUs, conversion factors and payments; "
+    "national unadjusted, no GPCI):"
+)
+
+
+def _n(v: float | None, digits: int = 2, money: bool = False) -> str:
+    if v is None:
+        return "n/a"
+    s = f"{v:.{digits}f}"
+    return f"${s}" if money else s
+
+
+@dataclass(frozen=True)
+class ValuationEvidence:
+    codes: tuple[str, ...]
+    families: tuple[str, ...]
+    rows: tuple[ValuationRow, ...]
+    unpriced: tuple[str, ...]
+
+    def prompt_block(self) -> str:
+        lines = [_HEADER]
+        for r in self.rows:
+            lines.append(
+                f'{r.label} {r.code} "{r.description}" — status {r.status or "?"}; '
+                f"work {_n(r.work)}; PE {_n(r.pe_nf)} non-fac / {_n(r.pe_f)} fac; "
+                f"MP {_n(r.mp)}; total {_n(r.total_nf)} non-fac / {_n(r.total_f)} fac; "
+                f"CF {_n(r.cf, 4, money=True)} → payment {_n(r.pay_nf, money=True)} non-fac / "
+                f"{_n(r.pay_f, money=True)} fac"
+            )
+        if self.unpriced:
+            lines.append(
+                "Not priced in the indexed fee schedule: " + ", ".join(self.unpriced)
+            )
+        return "\n".join(lines)
+
+    def payload(self) -> dict[str, Any]:
+        provenance: list[dict[str, str]] = []
+        seen: set[str] = set()
+        for r in self.rows:
+            if r.label not in seen:
+                seen.add(r.label)
+                provenance.append(
+                    {
+                        "label": r.label,
+                        "vintage": r.vintage,
+                        "citation": r.citation,
+                        "url": r.url,
+                    }
+                )
+        return {
+            "type": "valuation",
+            "codes": list(self.codes),
+            "families": list(self.families),
+            "rows": [asdict(r) for r in self.rows],
+            "provenance": provenance,
+            "unpriced": list(self.unpriced),
+        }
+
+
+def _replica_path(cfg: LlmConfig) -> str:
+    p = Path(cfg.duckdb_replica)
+    if p.is_absolute():
+        return str(p)
+    from conf import ROOT
+
+    return str(ROOT / p)
+
+
+def valuation_evidence(question: str, cfg: LlmConfig) -> ValuationEvidence | None:
+    det = detect_codes(question)
+    if not det.codes:
+        return None
+    try:
+        con = duckdb.connect(_replica_path(cfg), read_only=True)
+    except Exception as e:  # noqa: BLE001 — duckdb raises several types
+        log.warning("valuation evidence skipped (%s): %s", cfg.duckdb_replica, e)
+        return None
+    try:
+        rows, unpriced = valuation(con, list(det.codes), years=cfg.valuation_years)
+    except Exception as e:  # noqa: BLE001
+        log.warning("valuation evidence skipped (query): %s", e)
+        return None
+    finally:
+        con.close()
+    return ValuationEvidence(det.codes, det.families, tuple(rows), tuple(unpriced))
+
+
+_CITED_SQL = text(
+    "SELECT e.document, e.cmetadata FROM langchain_pg_embedding e "
+    "JOIN langchain_pg_collection c ON c.uuid = e.collection_id "
+    "WHERE c.name = :collection "
+    "AND string_to_array(COALESCE(e.cmetadata->>'codes', ''), ' ') && CAST(:codes AS text[]) "
+    "ORDER BY e.cmetadata->>'date' DESC NULLS LAST, e.cmetadata->>'item_key', e.cmetadata->>'p_id' "
+    "LIMIT :lim"
+)
+
+
+def code_cited_sources(
+    engine: Any, codes: Sequence[str], *, per_code: int, collection: str = "rules"
+) -> list[dict]:
+    """Chunks in *collection* whose ``codes`` metadata mention any of
+    *codes* — at most *per_code* per code, newest rule first, deduped on
+    (item_key, p_id). Scores are 0.0: these are additive, not ranked."""
+    wanted = [c.upper() for c in codes]
+    try:
+        with engine.begin() as conn:
+            rows = conn.execute(
+                _CITED_SQL,
+                {
+                    "collection": collection,
+                    "codes": wanted,
+                    "lim": per_code * len(wanted) * 4,
+                },
+            ).fetchall()
+    except Exception as e:  # noqa: BLE001
+        log.warning("code-cited sources skipped: %s", e)
+        return []
+    out: list[dict] = []
+    seen: set[tuple[str, str]] = set()
+    count: dict[str, int] = {c: 0 for c in wanted}
+    for document, md in rows:
+        md = {k: str(v) for k, v in (md or {}).items()}
+        key = (md.get("item_key", ""), md.get("p_id", ""))
+        if key in seen:
+            continue
+        hit_codes = [
+            c for c in md.get("codes", "").split() if c in count and count[c] < per_code
+        ]
+        if not hit_codes:
+            continue
+        seen.add(key)
+        for c in hit_codes:
+            count[c] += 1
+        out.append(as_source(md, document, 0.0))
+    return out
+
+
+def merge_sources(retrieved: list[dict], extra: list[dict]) -> list[dict]:
+    labels = {s["label"] for s in retrieved}
+    return retrieved + [s for s in extra if s["label"] not in labels]

tests/llm/test_evidence.py

@@ -0,0 +1,226 @@
+"""llm.evidence — valuation prompt block, SSE payload, code-cited rule sources."""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock, patch
+
+from llm.config import LlmConfig
+from llm.evidence import (
+    ValuationEvidence,
+    code_cited_sources,
+    merge_sources,
+    valuation_evidence,
+)
+from pfs.valuation import ValuationRow
+
+CFG = LlmConfig(
+    ollama_hosts=("http://h1:11434",),
+    embed_model="e",
+    instruct_model="c",
+    embed_dim=768,
+    build_ann_index=False,
+    pg_host="x",
+    pg_port=5432,
+    pg_db="llm",
+    pg_user="llm",
+    duckdb_replica="/nonexistent/aco.ro.duckdb",
+    valuation_years=4,
+    code_cited_per_code=2,
+)
+
+
+def _row(**kw) -> ValuationRow:
+    base = dict(
+        code="G0556",
+        description="Adv prim care mgmt lvl 1",
+        vintage="CY2026 final",
+        year=2026,
+        proposed=False,
+        status="A",
+        work=0.25,
+        pe_nf=0.22,
+        pe_f=0.06,
+        mp=0.02,
+        total_nf=0.49,
+        total_f=0.33,
+        cf=33.4009,
+        pay_nf=16.37,
+        pay_f=11.02,
+        label="[PFS CY2026 Addendum B]",
+        citation="90 FR 49266",
+        url="https://www.federalregister.gov/citation/90-FR-49266",
+    )
+    base.update(kw)
+    return ValuationRow(**base)
+
+
+class TestEvidenceText:
+    def test_prompt_block_exact(self):
+        ev = ValuationEvidence(
+            codes=("G0556",), families=("APCM",), rows=(_row(),), unpriced=("Z9999",)
+        )
+        assert ev.prompt_block() == (
+            "Valuation (authoritative for RVUs, conversion factors and payments; "
+            "national unadjusted, no GPCI):\n"
+            '[PFS CY2026 Addendum B] G0556 "Adv prim care mgmt lvl 1" — status A; work 0.25; '
+            "PE 0.22 non-fac / 0.06 fac; MP 0.02; total 0.49 non-fac / 0.33 fac; "
+            "CF $33.4009 → payment $16.37 non-fac / $11.02 fac\n"
+            "Not priced in the indexed fee schedule: Z9999"
+        )
+
+    def test_prompt_block_null_components(self):
+        ev = ValuationEvidence(
+            ("G9999",),
+            (),
+            (
+                _row(
+                    code="G9999",
+                    status="I",
+                    work=None,
+                    pe_nf=None,
+                    pe_f=None,
+                    mp=None,
+                    total_nf=None,
+                    total_f=None,
+                    pay_nf=None,
+                    pay_f=None,
+                ),
+            ),
+            (),
+        )
+        line = ev.prompt_block().splitlines()[1]
+        assert (
+            "status I; work n/a; PE n/a non-fac / n/a fac" in line
+            and "payment n/a" in line
+        )
+
+    def test_payload_shape(self):
+        ev = ValuationEvidence(
+            ("G0556",),
+            ("APCM",),
+            (
+                _row(),
+                _row(
+                    year=2027,
+                    proposed=True,
+                    vintage="CY2027 proposed",
+                    label="[CY2027 NPRM Addendum B]",
+                    citation="91 FR 43842",
+                    url="https://www.federalregister.gov/citation/91-FR-43842",
+                ),
+            ),
+            (),
+        )
+        p = ev.payload()
+        assert (
+            p["type"] == "valuation"
+            and p["codes"] == ["G0556"]
+            and p["families"] == ["APCM"]
+        )
+        assert p["rows"][0]["pay_nf"] == 16.37 and p["rows"][1]["proposed"] is True
+        assert p["provenance"] == [
+            {
+                "label": "[PFS CY2026 Addendum B]",
+                "vintage": "CY2026 final",
+                "citation": "90 FR 49266",
+                "url": "https://www.federalregister.gov/citation/90-FR-49266",
+            },
+            {
+                "label": "[CY2027 NPRM Addendum B]",
+                "vintage": "CY2027 proposed",
+                "citation": "91 FR 43842",
+                "url": "https://www.federalregister.gov/citation/91-FR-43842",
+            },
+        ]
+        assert p["unpriced"] == []
+
+
+class TestValuationEvidence:
+    def test_none_when_no_codes(self):
+        assert (
+            valuation_evidence("what did commenters say about telehealth?", CFG) is None
+        )
+
+    @patch("llm.evidence.valuation", return_value=([_row()], []))
+    @patch("llm.evidence.duckdb.connect")
+    def test_detects_and_queries_replica_read_only(self, mock_connect, mock_val):
+        ev = valuation_evidence("How is APCM valued?", CFG)
+        mock_connect.assert_called_once_with(
+            "/nonexistent/aco.ro.duckdb", read_only=True
+        )
+        assert mock_val.call_args.args[1] == ["G0556", "G0557", "G0558"]
+        assert mock_val.call_args.kwargs == {"years": 4}
+        assert (
+            ev is not None
+            and ev.codes == ("G0556", "G0557", "G0558")
+            and ev.families == ("APCM",)
+        )
+        mock_connect.return_value.close.assert_called_once()
+
+    @patch("llm.evidence.duckdb.connect", side_effect=OSError("no replica"))
+    def test_unopenable_replica_yields_none(self, _c, caplog):
+        assert valuation_evidence("G0556?", CFG) is None
+        assert "valuation evidence skipped" in caplog.text
+
+
+def _engine(rows):
+    engine = MagicMock()
+    conn = engine.begin.return_value.__enter__.return_value
+    conn.execute.return_value.fetchall.return_value = rows
+    return engine, conn
+
+
+def _rule_md(p_id, codes, date="2024-12-09"):
+    return {
+        "kind": "rule",
+        "item_key": "R1",
+        "p_id": str(p_id),
+        "page": "97710",
+        "ordinal": "1",
+        "html_url": "https://www.federalregister.gov/d/2024-25382",
+        "fr_volume": "89",
+        "date": date,
+        "title": "CY2025 PFS final rule",
+        "codes": codes,
+    }
+
+
+class TestCodeCitedSources:
+    def test_sql_and_per_code_cap_and_dedupe(self):
+        engine, conn = _engine(
+            [
+                ("para one G0556", _rule_md(1, "G0556")),
+                ("para two G0556", _rule_md(2, "G0556")),
+                ("para three G0556", _rule_md(3, "G0556")),  # over the cap for G0556
+                (
+                    "para four G0557",
+                    _rule_md(4, "G0557 G0556"),
+                ),  # G0557 still under cap
+                ("para one again", _rule_md(1, "G0556")),  # duplicate (item_key, p_id)
+            ]
+        )
+        out = code_cited_sources(engine, ["G0556", "G0557"], per_code=2)
+        sql = str(conn.execute.call_args.args[0])
+        assert "string_to_array" in sql and "langchain_pg_collection" in sql
+        assert conn.execute.call_args.args[1]["collection"] == "rules"
+        assert conn.execute.call_args.args[1]["codes"] == ["G0556", "G0557"]
+        assert [s["snippet"] for s in out] == [
+            "para one G0556",
+            "para two G0556",
+            "para four G0557",
+        ]
+        assert all(s["kind"] == "rule" and s["score"] == 0.0 for s in out)
+
+    def test_engine_error_yields_empty(self, caplog):
+        engine = MagicMock()
+        engine.begin.side_effect = RuntimeError("pg down")
+        assert code_cited_sources(engine, ["G0556"], per_code=2) == []
+        assert "code-cited sources skipped" in caplog.text
+
+
+class TestMergeSources:
+    def test_keeps_order_and_dedupes_by_label(self):
+        a = [{"label": "X", "score": 0.9}, {"label": "Y", "score": 0.8}]
+        b = [{"label": "Y", "score": 0.0}, {"label": "Z", "score": 0.0}]
+        assert [s["label"] for s in merge_sources(a, b)] == ["X", "Y", "Z"]
+        assert merge_sources(a, b)[1]["score"] == 0.8