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)