7693e97d36
docs/superpowers/specs/2026-09-22-llm-tagging-design.md lays out P35: a versioned hand-curated vocabulary, embedding shortlist + per-theme yes/no judgement with an evidence chunk on the largest live host, resumable docket-scoped runs with state in bib extra_json, write-back that replaces only llm: tags, and a golden-set gate before fan-out. Slice 1: src/llm/vocab/themes.yaml — 53 themes (conversion factor, practice expense, telehealth, care management, behavioral health, Part B drugs, MSSP, QPP, …), each with a label, one-sentence definition, letter-language synonyms and the FR section stems it was seeded from (dev/scripts/llm_vocab_candidates.py mines heading paragraphs from the CY2018+ rules' fr_anchors). llm.vocab.load() validates unique kebab-case slugs, non-empty definitions, integer version ≥ 1 and that retired slugs are not still active; Theme.card is the text embedded per theme; split_tags() separates llm: tags from hand tags for the write-back. stack llm vocab [--path] [--slug].
82 lines
2.9 KiB
Python
82 lines
2.9 KiB
Python
"""Seed a theme-vocabulary revision from Federal Register section headings (#574).
|
||
|
||
Walks ``fr_anchors`` for heading-like paragraphs ("B. Determination of
|
||
Practice Expense…") in every PFS rule whose rule year is at or after
|
||
``--since`` (default 2018 — the regulations.gov comment dockets start in
|
||
2017), stems the titles, and prints the stems ordered by how many rule
|
||
years carry them, with an example title each. The output is raw material
|
||
for a human editing ``src/llm/vocab/themes.yaml`` — it is not the vocab.
|
||
|
||
Usage:
|
||
uv run python dev/scripts/llm_vocab_candidates.py [--since 2018] [--top 200]
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import re
|
||
from collections import defaultdict
|
||
|
||
_HEADING = re.compile(r"^(?:[IVX]{1,4}|[A-Z]|\d{1,2}|[a-z])\. +([A-Z][^.]{6,120})$")
|
||
_STOP = re.compile(
|
||
r"\b(cy|20\d\d|proposed|final|rule|for|the|of|and|to|in|a|an|under|pfs|physician fee schedule)\b"
|
||
)
|
||
_SKIP = ("authority citation", "on page", "column", "paragraph")
|
||
|
||
|
||
def stem(title: str) -> str:
|
||
norm = re.sub(r"[^a-z0-9 ]", " ", title.lower())
|
||
norm = _STOP.sub(" ", norm)
|
||
return re.sub(r"\s+", " ", norm).strip()
|
||
|
||
|
||
def candidates(store, *, since: int) -> list[tuple[int, str, str]]:
|
||
"""``(n_rule_years, stem, example title)`` sorted by n desc."""
|
||
from pfs.descriptors import rule_year_of
|
||
|
||
con = store._con() # noqa: SLF001
|
||
years: dict[str, int] = {}
|
||
for key, title, published in con.execute(
|
||
"SELECT key, title, date_published FROM items WHERE item_type = 'rule'"
|
||
).fetchall():
|
||
# rule_year_of reads the payment year from the title, else falls
|
||
# back to publication year + 1 (PFS rules publish July–December).
|
||
years[key] = int(rule_year_of(title or "", published or ""))
|
||
recent = {k for k, y in years.items() if y and y >= since}
|
||
by: dict[str, set[int]] = defaultdict(set)
|
||
example: dict[str, str] = {}
|
||
for key, text in con.execute(
|
||
"SELECT item_key, text FROM fr_anchors WHERE length(text) < 140"
|
||
).fetchall():
|
||
if key not in recent:
|
||
continue
|
||
m = _HEADING.match((text or "").strip())
|
||
if not m:
|
||
continue
|
||
title = m.group(1).strip()
|
||
s = stem(title)
|
||
if len(s) < 5 or any(s.startswith(x) or x in s for x in _SKIP):
|
||
continue
|
||
by[s].add(years[key])
|
||
example.setdefault(s, title)
|
||
return sorted(
|
||
((len(v), k, example[k]) for k, v in by.items()), key=lambda r: (-r[0], r[1])
|
||
)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description=__doc__.split("\n")[0])
|
||
ap.add_argument("--since", type=int, default=2018)
|
||
ap.add_argument("--top", type=int, default=200)
|
||
args = ap.parse_args()
|
||
from conf import connect
|
||
|
||
store = connect.bib()
|
||
for n, s, ex in candidates(store, since=args.since)[: args.top]:
|
||
print(f"{n:3d} {ex}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|