6a6396b929
The valuation table now captions itself as national unadjusted amounts, titles the status and CF cells with their notes, lists the unpaid-status reasons under the table, and renders a citation with no URL as a span instead of an <a href=""> that reloads the chat. The prompt block carries the same two notes. Markdown chunks scan their heading, so a section titled with a code stays findable by it past the first chunk.
187 lines
6.9 KiB
Python
187 lines
6.9 KiB
Python
"""llm.chunk — markdown-aware chunking with deterministic ids."""
|
|
|
|
import pytest
|
|
|
|
from llm.chunk import Doc, chunk_doc, content_hash
|
|
|
|
FRONTMATTER_DOC = """---
|
|
comment_id: CMS-2019-0111-0042
|
|
docket_id: CMS-2019-0111
|
|
---
|
|
|
|
# Re: CY 2020 PFS Proposed Rule
|
|
|
|
We object to the E/M consolidation.
|
|
|
|
## Telehealth
|
|
|
|
Originating-site rules should be relaxed.
|
|
"""
|
|
|
|
|
|
def _doc(text, key="K1"):
|
|
return Doc(key=key, text=text, metadata={"docket": "CMS-2019-0111"})
|
|
|
|
|
|
class TestChunkDoc:
|
|
def test_strips_yaml_frontmatter(self):
|
|
chunks = chunk_doc(_doc(FRONTMATTER_DOC))
|
|
assert "comment_id:" not in chunks[0].text
|
|
assert chunks[0].text.startswith("# Re:")
|
|
|
|
def test_deterministic_ids(self):
|
|
a = chunk_doc(_doc(FRONTMATTER_DOC))
|
|
b = chunk_doc(_doc(FRONTMATTER_DOC))
|
|
assert [c.id for c in a] == [c.id for c in b]
|
|
h = content_hash(FRONTMATTER_DOC)
|
|
assert a[0].id == f"K1:{h[:12]}:0000"
|
|
|
|
def test_metadata_carries_item_key_and_seq(self):
|
|
chunks = chunk_doc(_doc(FRONTMATTER_DOC))
|
|
assert chunks[0].metadata["item_key"] == "K1"
|
|
assert chunks[0].metadata["docket"] == "CMS-2019-0111"
|
|
assert chunks[0].metadata["seq"] == "0"
|
|
|
|
def test_splits_on_headings_before_size(self):
|
|
text = "# A\n\n" + "para. " * 100 + "\n\n# B\n\nshort."
|
|
chunks = chunk_doc(_doc(text), target_chars=300)
|
|
assert all(len(c.text) <= 300 + 200 for c in chunks)
|
|
assert any(c.text.lstrip().startswith("# B") for c in chunks)
|
|
|
|
def test_long_paragraph_hard_wrapped_with_overlap(self):
|
|
text = "x" * 5000
|
|
chunks = chunk_doc(_doc(text), target_chars=2000, overlap_chars=200)
|
|
assert len(chunks) == 3
|
|
assert chunks[1].text[:200] == chunks[0].text[-200:]
|
|
|
|
def test_empty_and_whitespace_yield_nothing(self):
|
|
assert chunk_doc(_doc("")) == []
|
|
assert chunk_doc(_doc(" \n\n ")) == []
|
|
|
|
def test_overlap_gte_target_raises(self):
|
|
with pytest.raises(ValueError, match="overlap_chars must be smaller"):
|
|
chunk_doc(_doc("some text"), target_chars=100, overlap_chars=100)
|
|
|
|
|
|
class TestControlCharStripping:
|
|
def test_strips_nul_and_control_chars(self):
|
|
doc = _doc("Clean text\x00 with\x0c a NUL\x07 and formfeed.")
|
|
chunks = chunk_doc(doc)
|
|
joined = "".join(c.text for c in chunks)
|
|
assert "\x00" not in joined
|
|
assert "\x0c" not in joined
|
|
assert "\x07" not in joined
|
|
assert "Clean text with a NUL and formfeed." in joined
|
|
|
|
def test_keeps_tab_newline_return(self):
|
|
doc = _doc("line1\n\nline2\twith tab")
|
|
joined = "".join(c.text for c in chunk_doc(doc))
|
|
assert "\t" in joined and "line2" in joined
|
|
|
|
|
|
# ── section metadata + paragraph-anchored rule chunks ────────────────
|
|
|
|
from llm.chunk import Paragraph # noqa: E402
|
|
|
|
|
|
def _para(p_id, page, ordinal, text):
|
|
return Paragraph(p_id=p_id, page=page, ordinal=ordinal, text=text)
|
|
|
|
|
|
class TestSectionMetadata:
|
|
def test_heading_text_recorded_per_chunk(self):
|
|
doc = Doc(
|
|
key="K",
|
|
text="intro para\n\n## attachment_1.pdf\n\nbody one\n\n## attachment_2.docx\n\nbody two",
|
|
metadata={},
|
|
)
|
|
chunks = chunk_doc(doc, target_chars=60, overlap_chars=5)
|
|
assert [c.metadata["section"] for c in chunks] == [
|
|
"",
|
|
"attachment_1.pdf",
|
|
"attachment_2.docx",
|
|
]
|
|
|
|
def test_no_heading_is_empty_section(self):
|
|
(c,) = chunk_doc(Doc(key="K", text="plain", metadata={}))
|
|
assert c.metadata["section"] == ""
|
|
|
|
|
|
class TestParagraphChunks:
|
|
def test_packs_whole_paragraphs_and_stamps_anchor_metadata(self):
|
|
paras = (
|
|
_para(10, 100, 1, "A" * 30),
|
|
_para(11, 100, 2, "B" * 30),
|
|
_para(12, 101, 1, "C" * 30),
|
|
)
|
|
doc = Doc(
|
|
key="R",
|
|
text="\n\n".join(p.text for p in paras),
|
|
metadata={"kind": "rule"},
|
|
paragraphs=paras,
|
|
)
|
|
chunks = chunk_doc(doc, target_chars=70, overlap_chars=5)
|
|
assert [c.text for c in chunks] == ["A" * 30 + "\n\n" + "B" * 30, "C" * 30]
|
|
assert chunks[0].metadata["p_id"] == "10"
|
|
assert chunks[0].metadata["p_id_last"] == "11"
|
|
assert chunks[0].metadata["page"] == "100"
|
|
assert chunks[0].metadata["ordinal"] == "1"
|
|
assert chunks[1].metadata["p_id"] == "12"
|
|
assert chunks[1].metadata["page"] == "101"
|
|
assert chunks[0].metadata["kind"] == "rule"
|
|
assert chunks[0].metadata["item_key"] == "R"
|
|
assert chunks[0].id.endswith(":0000") and chunks[1].id.endswith(":0001")
|
|
|
|
def test_oversized_paragraph_is_wrapped_but_keeps_its_anchor(self):
|
|
paras = (_para(5, 200, 3, "X" * 100),)
|
|
doc = Doc(key="R", text=paras[0].text, metadata={}, paragraphs=paras)
|
|
chunks = chunk_doc(doc, target_chars=40, overlap_chars=10)
|
|
assert len(chunks) == 3 # windows 0-40, 30-70, 60-100
|
|
assert {c.metadata["p_id"] for c in chunks} == {"5"}
|
|
assert chunks[1].text[:10] == chunks[0].text[-10:]
|
|
|
|
def test_empty_paragraph_text_skipped(self):
|
|
paras = (_para(1, 1, 1, " "), _para(2, 1, 2, "real"))
|
|
doc = Doc(key="R", text="real", metadata={}, paragraphs=paras)
|
|
(c,) = chunk_doc(doc)
|
|
assert c.metadata["p_id"] == "2"
|
|
|
|
def test_ids_are_deterministic(self):
|
|
paras = (_para(1, 1, 1, "same"),)
|
|
a = chunk_doc(Doc(key="R", text="same", metadata={}, paragraphs=paras))
|
|
b = chunk_doc(Doc(key="R", text="same", metadata={}, paragraphs=paras))
|
|
assert [c.id for c in a] == [c.id for c in b]
|
|
|
|
|
|
class TestCodesMetadata:
|
|
def test_codes_stamped_per_chunk(self):
|
|
text = "We propose G0556 and G0557.\n\n## Other\n\nNo codes here.\n\n## More\n\n99490 applies."
|
|
chunks = chunk_doc(_doc(text), target_chars=60, overlap_chars=10)
|
|
found = {c.metadata["codes"] for c in chunks}
|
|
assert "G0556 G0557" in found
|
|
assert "99490" in found
|
|
assert "" in found
|
|
|
|
def test_heading_codes_stamped_on_every_piece_of_the_section(self):
|
|
"""A section headed by the code it discusses ("## G0556 — APCM")
|
|
splits into several chunks; the ones past the first never repeat
|
|
the code, and used to be unfindable by it."""
|
|
text = "## G0556 — APCM\n\n" + "We propose to value the service. " * 20
|
|
chunks = chunk_doc(_doc(text), target_chars=120, overlap_chars=10)
|
|
assert len(chunks) > 1
|
|
assert all(c.metadata["codes"] == "G0556" for c in chunks)
|
|
|
|
def test_rule_paragraph_chunks_get_codes(self):
|
|
from llm.chunk import Paragraph
|
|
|
|
doc = Doc(
|
|
key="R1",
|
|
text="x",
|
|
metadata={"kind": "rule"},
|
|
paragraphs=(
|
|
Paragraph(1, 100, 1, "APCM code G0556 is valued at 0.25 work RVUs."),
|
|
),
|
|
)
|
|
(chunk,) = chunk_doc(doc)
|
|
assert chunk.metadata["codes"] == "G0556"
|