From 93a9404663529949b07beb52aa64e65dc1768d08 Mon Sep 17 00:00:00 2001 From: Chaim Date: Sun, 28 Jun 2026 21:57:38 +0000 Subject: [PATCH] feat(learning): grow style-exemplars from every final (P1 #5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The style-exemplar corpus (channel B — the writer's block-level retrieval of Dafna's real prose) was FROZEN at the one-time seed backfill: new finals were enrolled into style_corpus but never broken into exemplars, so the richest style channel never grew (8137/8126/8174 had 0 exemplars). The writer kept retrieving only March–April seed paragraphs no matter how many finals were signed. Extract the per-decision exemplar logic (section→paragraph→Voyage-embed→replace) into a shared service `legal_mcp.services.style_exemplars.extract_and_store` — the SINGLE implementation now used by BOTH the one-time backfill and the live enrollment path (G2; no parallel extractor). `_enroll_final_in_library` calls it on every final upload (source='internal_committee', the same source the writer's search_style_exemplars reads). Voyage embeds over REST → container-safe; best-effort, surfaced in the upload response, never fails the upload. Effect: every signed final now grows the exemplar corpus, so the writer's block-level style retrieval improves with each decision — the core "learn from every decision" fix for channel B. Path A (style_distance_history) will track whether the larger exemplar pool reduces style-distance over time. Invariants: G2 (one extraction path shared by backfill + enroll), INV-LRN5 (style/structure prose only — substance routes elsewhere), INV-LRN4 (the draft↔final loop now feeds the exemplar channel, not just the lesson channel). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../src/legal_mcp/services/style_exemplars.py | 83 +++++++++++++++++++ scripts/backfill_style_exemplars.py | 63 ++++---------- web/app.py | 16 ++++ 3 files changed, 113 insertions(+), 49 deletions(-) create mode 100644 mcp-server/src/legal_mcp/services/style_exemplars.py diff --git a/mcp-server/src/legal_mcp/services/style_exemplars.py b/mcp-server/src/legal_mcp/services/style_exemplars.py new file mode 100644 index 0000000..2428b59 --- /dev/null +++ b/mcp-server/src/legal_mcp/services/style_exemplars.py @@ -0,0 +1,83 @@ +"""Block-level style-exemplar extraction (channel B of Style Acquisition). + +Splits one of Dafna's decisions into section→paragraph units, embeds them +(Voyage), and stores them in `style_exemplars` so the writer can retrieve real +block-level prose by section/outcome/practice_area (07-learning §0.2 channel B). + +This is the SINGLE source of truth for exemplar extraction (G2): both the +one-time backfill (`scripts/backfill_style_exemplars.py`) and the live +final-enrollment path (`_enroll_final_in_library`) call `extract_and_store`. +Before this, exemplars were frozen at the seed backfill — new finals enrolled +into style_corpus but were never broken into exemplars, so the richest style +channel never grew. Embedding is Voyage-over-REST → container-safe. +""" +from __future__ import annotations + +import logging + +from legal_mcp.services import db, embeddings +from legal_mcp.services.chunker import _split_into_sections + +logger = logging.getLogger(__name__) + +# chunker section_type → style_exemplars.section +_SECTION_MAP = { + "facts": "background", + "appellant_claims": "claims", + "respondent_claims": "claims", + "legal_analysis": "discussion", + "conclusion": "summary", + "ruling": "summary", + "intro": "other", + "other": "other", +} + +MIN_WORDS = 25 # skip tiny fragments +MAX_WORDS = 450 # skip over-long blobs (likely un-split) +MAX_PER_SECTION = 15 + + +def _paragraphs(section_text: str) -> list[str]: + """Split a section into paragraph units (blank-line separated; fall back to lines).""" + raw = [p.strip() for p in section_text.split("\n\n")] + if len(raw) <= 1: + raw = [p.strip() for p in section_text.split("\n")] + out = [] + for p in raw: + wc = len(p.split()) + if MIN_WORDS <= wc <= MAX_WORDS: + out.append(p) + return out[:MAX_PER_SECTION] + + +def units_for(full_text: str) -> list[tuple[str, str]]: + """(section, paragraph) units for a decision — pure, no I/O.""" + units: list[tuple[str, str]] = [] + for section_type, section_text in _split_into_sections(full_text or ""): + section = _SECTION_MAP.get(section_type, "other") + for para in _paragraphs(section_text): + units.append((section, para)) + return units + + +async def extract_and_store( + decision_number: str, source: str, full_text: str, + practice_area: str = "", outcome: str = "", +) -> int: + """Idempotently (re)build a decision's block-level style exemplars: split → + embed → replace. Returns the number of exemplars stored. Raises on failure; + callers in best-effort paths (enroll) should wrap in try/except.""" + units = units_for(full_text) + if not units: + return 0 + texts = [u[1] for u in units] + vecs = await embeddings.embed_texts(texts, input_type="document") + await db.delete_style_exemplars(decision_number, source) + for (section, para), vec in zip(units, vecs): + await db.insert_style_exemplar( + decision_number=decision_number, source=source, + practice_area=practice_area, outcome=outcome, + section=section, paragraph_text=para, word_count=len(para.split()), + embedding=vec, + ) + return len(units) diff --git a/scripts/backfill_style_exemplars.py b/scripts/backfill_style_exemplars.py index 3d9f28f..8487ee4 100644 --- a/scripts/backfill_style_exemplars.py +++ b/scripts/backfill_style_exemplars.py @@ -19,40 +19,15 @@ import argparse import asyncio import logging -from legal_mcp.services import db, embeddings -from legal_mcp.services.chunker import _split_into_sections +from legal_mcp.services import db +from legal_mcp.services import style_exemplars as sx logging.basicConfig(level=logging.INFO, format="%(message)s") log = logging.getLogger("backfill_exemplars") -# chunker section_type → style_exemplars.section -_SECTION_MAP = { - "facts": "background", - "appellant_claims": "claims", - "respondent_claims": "claims", - "legal_analysis": "discussion", - "conclusion": "summary", - "ruling": "summary", - "intro": "other", - "other": "other", -} - -MIN_WORDS = 25 # skip tiny fragments -MAX_WORDS = 450 # skip over-long blobs (likely un-split) -MAX_PER_SECTION = 15 - - -def _paragraphs(section_text: str) -> list[str]: - """Split a section into paragraph units (blank-line separated; fall back to lines).""" - raw = [p.strip() for p in section_text.split("\n\n")] - if len(raw) <= 1: - raw = [p.strip() for p in section_text.split("\n")] - out = [] - for p in raw: - wc = len(p.split()) - if MIN_WORDS <= wc <= MAX_WORDS: - out.append(p) - return out[:MAX_PER_SECTION] +# Section mapping + paragraph splitting now live in the shared service +# (legal_mcp.services.style_exemplars) so the backfill and the live +# final-enrollment path use ONE extraction implementation (G2). async def _gather_sources() -> list[dict]: @@ -94,28 +69,18 @@ async def main(apply: bool) -> None: total_paras = 0 for src in sources: - units: list[tuple[str, str]] = [] # (section, paragraph) - for section_type, section_text in _split_into_sections(src["full_text"]): - section = _SECTION_MAP.get(section_type, "other") - for para in _paragraphs(section_text): - units.append((section, para)) - if not units: + n = len(sx.units_for(src["full_text"])) + if not n: continue - total_paras += len(units) - log.info(" %-14s %-16s → %d פסקאות", src["source"], src["decision_number"], len(units)) + total_paras += n + log.info(" %-14s %-16s → %d פסקאות", src["source"], src["decision_number"], n) if not apply: continue - - await db.delete_style_exemplars(src["decision_number"], src["source"]) - texts = [u[1] for u in units] - vecs = await embeddings.embed_texts(texts, input_type="document") - for (section, para), vec in zip(units, vecs): - await db.insert_style_exemplar( - decision_number=src["decision_number"], source=src["source"], - practice_area=src["practice_area"], outcome=src["outcome"], - section=section, paragraph_text=para, word_count=len(para.split()), - embedding=vec, - ) + await sx.extract_and_store( + decision_number=src["decision_number"], source=src["source"], + full_text=src["full_text"], practice_area=src["practice_area"], + outcome=src["outcome"], + ) if apply: cov = await db.count_style_exemplars() diff --git a/web/app.py b/web/app.py index 4bceca2..42d5c23 100644 --- a/web/app.py +++ b/web/app.py @@ -3769,6 +3769,22 @@ async def _enroll_final_in_library( logger.warning("inline metadata extraction failed for %s: %s", case_number, e) out["metadata_error"] = str(e) + # Grow the style-exemplar corpus (channel B): break this final into block-level + # paragraphs the writer retrieves (07-learning §0.2). Before this, exemplars were + # frozen at the one-time seed backfill — new finals never got exemplarized, so the + # richest style channel never grew. Same extraction the backfill uses (G2). Voyage + # embeds over REST → container-safe. Best-effort; surfaced, never fails the upload. + try: + from legal_mcp.services import style_exemplars as _sx + n_ex = await _sx.extract_and_store( + decision_number=case_number, source="internal_committee", + full_text=final_text, practice_area=case.get("practice_area", ""), + ) + out["exemplars"] = n_ex + except Exception as e: + logger.warning("style-exemplar extraction failed for %s: %s", case_number, e) + out["exemplars_error"] = str(e) + # The precedents this decision cites → link to the library; flag the ones not found. try: await cit_tools.extract_internal_citations(case_law_id=case_law_id, limit=0)