From 6424bfce5cfc3b8ebfed7a48972cc8989fe64871 Mon Sep 17 00:00:00 2001 From: Chaim Date: Tue, 21 Apr 2026 15:56:18 +0000 Subject: [PATCH] =?UTF-8?q?feat(kb):=20hierarchical=20heading=5Fpath=20for?= =?UTF-8?q?=20statutes=20(=D7=97=D7=9C=D7=A7>=D7=A4=D7=A8=D7=A7>=D7=A1?= =?UTF-8?q?=D7=99=D7=9E=D7=9F>=D7=A1=D7=A2=D7=99=D7=A3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Before: chunk_statute produced heading_path = section_ref = "ס' 195". After: heading_path walks the four Israeli-statute levels (part, chapter, sub-chapter, section) and renders them as "חלק ט׳ — ביטוח נכות > פרק ב׳ — ועדות רפואיות > סימן א׳ > ס' 195". section_ref stays the leaf so citations stay precise. - New _RE_PART / _RE_CHAPTER / _RE_SUBCHAPTER regexes with Hebrew ordinal support including multi-letter gershayim ("י״א", "ט״ז"). - _collect_markers emits all markers ordered by position; chunk_statute walks them tracking active_part/chapter/subchapter and attaches the full path to each section chunk. Refs Task Master #2 Co-Authored-By: Claude Opus 4.7 (1M context) --- api/services/kb/chunker.py | 117 +++++++++++++++++++++++++++++++------ 1 file changed, 99 insertions(+), 18 deletions(-) diff --git a/api/services/kb/chunker.py b/api/services/kb/chunker.py index cd3631f..1d5b90e 100644 --- a/api/services/kb/chunker.py +++ b/api/services/kb/chunker.py @@ -19,11 +19,31 @@ _TOKENS_PER_CHAR = 0.55 # and a batch is capped at 120K tokens. Keep chunks small so batches fit. _MAX_TOKENS_PER_CHUNK = 1500 -# A "section" in Israeli statutes: "סעיף 12", "תקנה 5", "5א.", "12(ב)" -_SECTION_RE = re.compile( - r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)?(\d+[א-ת]?\.?)\s*(?:\(([א-ת\d]+)\))?\s*[\.\:]", +# Hierarchy markers in Israeli statutes. Each regex captures the full heading line +# and is ordered from largest container to smallest. The section-level match +# (_RE_SECTION) also captures the section number and optional sub-letter. +# +# Hebrew ordinal numbers: "א", "ב", … "ט׳", "י״א", "י״ב", … "ט״ז", etc. +# Pattern: one or two Hebrew letters with optional gershayim/geresh between them, +# and optionally followed by a trailing geresh. +_HEB_ORD = r"[א-ת](?:[\"״][א-ת])?[׳'\"]?" + +_RE_PART = re.compile(rf"(?m)^\s*חלק\s+({_HEB_ORD})\s*[:\-–—]?\s*(.*?)\s*$") +_RE_CHAPTER = re.compile(rf"(?m)^\s*פרק\s+({_HEB_ORD})\s*[:\-–—]?\s*(.*?)\s*$") +_RE_SUBCHAPTER = re.compile(rf"(?m)^\s*סימן\s+({_HEB_ORD})\s*[:\-–—]?\s*(.*?)\s*$") +# Section header: "סעיף 12.", "12א.", "תקנה 5.", at start of a line, followed by +# optional sub-letter "(ב)" and then a dot / colon / dash or newline. +_RE_SECTION = re.compile( + r"(?m)^\s*(?:סעיף\s+|תקנה\s+|§\s*)?(\d+[א-ת]?)\s*(?:\(([א-ת\d]+)\))?\s*[\.:\-–—]" ) +_STATUTE_MARKERS = [ + ("part", _RE_PART), + ("chapter", _RE_CHAPTER), + ("subchapter", _RE_SUBCHAPTER), + ("section", _RE_SECTION), +] + # Header in circulars — markdown-ish headings, or numbered "נושא: ..." / "מס' N. ..." _HEADER_RE = re.compile( r"(?m)^\s*(?:#{1,4}\s+|מס['׳]\s*\d+\s*[\.\:]\s*|[A-Zא-ת][^\n]{0,80}\:\s*$)", @@ -83,34 +103,95 @@ def _split_oversized(chunks: list[Chunk]) -> list[Chunk]: return result +def _collect_markers(text: str) -> list[dict]: + """Collect all hierarchy markers from the text, sorted by position. + + Each marker carries: {level, start, end, number, title, heading}. + level ∈ {'chapter', 'subchapter', 'section'}. + """ + markers: list[dict] = [] + for level, rx in _STATUTE_MARKERS: + for m in rx.finditer(text): + g1 = m.group(1) if m.groups() else "" + if level == "section": + number = g1 + sub = m.group(2) if m.lastindex and m.lastindex >= 2 else None + ref = f"ס' {number}" + (f"({sub})" if sub else "") + heading = ref + title = "" + else: + prefix = {"part": "חלק", "chapter": "פרק", "subchapter": "סימן"}[level] + title = m.group(2).strip() if m.lastindex and m.lastindex >= 2 else "" + number = g1 + ref = f"{prefix} {number}" + heading = f"{ref}{(' — ' + title) if title else ''}" + markers.append({ + "level": level, + "start": m.start(), + "end": m.end(), + "number": number, + "ref": ref, + "heading": heading, + "title": title, + }) + markers.sort(key=lambda x: x["start"]) + return markers + + def chunk_statute(text: str) -> list[dict]: - """Law / regulation: one chunk per section. Preserves section numbering.""" + """Law / regulation: one chunk per section, with hierarchical heading_path. + + heading_path format: "חלק ט' — ביטוח נכות > פרק ב' > סימן א' > ס' 195(א)". + section_ref stays the leaf ("ס' 195(א)") for precise citation. + """ text = text.strip() if not text: return [] - matches = list(_SECTION_RE.finditer(text)) - if not matches: - # Fallback: treat the whole thing as one chunk. - return _as_dicts( - [Chunk(0, "", "", text, _count_tokens(text))] - ) + markers = _collect_markers(text) + section_markers = [m for m in markers if m["level"] == "section"] + if not section_markers: + return _as_dicts([Chunk(0, "", "", text, _count_tokens(text))]) + + # Walk markers in order, tracking the active part/chapter/subchapter. + active_part: dict | None = None + active_chapter: dict | None = None + active_subchapter: dict | None = None chunks: list[Chunk] = [] - for i, m in enumerate(matches): - start = m.start() - end = matches[i + 1].start() if i + 1 < len(matches) else len(text) + for i, m in enumerate(markers): + if m["level"] == "part": + active_part = m + active_chapter = None + active_subchapter = None + continue + if m["level"] == "chapter": + active_chapter = m + active_subchapter = None + continue + if m["level"] == "subchapter": + active_subchapter = m + continue + # section marker + start = m["start"] + end = markers[i + 1]["start"] if i + 1 < len(markers) else len(text) content = text[start:end].strip() if not content: continue - number = m.group(1).rstrip(".") - sub = m.group(2) - section_ref = f"ס' {number}" + (f"({sub})" if sub else "") + parts: list[str] = [] + if active_part: + parts.append(active_part["heading"]) + if active_chapter: + parts.append(active_chapter["heading"]) + if active_subchapter: + parts.append(active_subchapter["heading"]) + parts.append(m["ref"]) + heading_path = " > ".join(parts) chunks.append( Chunk( chunk_index=len(chunks), - heading_path=section_ref, - section_ref=section_ref, + heading_path=heading_path, + section_ref=m["ref"], content=content, token_count=_count_tokens(content), )