feat(#99 / T10): get_style_guide — יחסי-זהב נמדדים מהקורפוס לצד היעד

style_distance.measure_corpus_ratios(): מפצל כל החלטה ב-style_corpus לסעיפים
(chunker) ומחשב ממוצע %-סעיף — אגרגט "_all" + פר-תוצאה (כשיש). cached.
get_style_guide מציג שורת "נמדד בפועל" עם ⚠️ על פער מטווח-היעד.

מצב נוכחי: style_corpus.outcome לא מאוכלס → מוצג אגרגט כל-ההחלטות (n=48:
רקע 26.4% / טענות 9.7% / דיון 43.8% / סיכום 20.1%); פיצול לפי-תוצאה future-ready.
המדידה גם מאירה מגבלות זיהוי-סעיפים (כוונת T10 — לסמן פער לבדיקה). חופף-חלקית
ל-T7 שמודד adherence per-draft; זה מודד את הקורפוס. כשל מדידה מוצג, לא נבלע.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-06-06 21:01:42 +00:00
parent a571ad535b
commit e4651a9d06
2 changed files with 91 additions and 0 deletions

View File

@@ -27,6 +27,62 @@ _BLOCK_TO_SECTION = {
"block-yod-alef": "summary",
}
# chunker section_type → golden-ratio section (for corpus measurement, T10)
_CHUNK_SECTION_TO_GOLDEN = {
"facts": "background", "intro": "background",
"appellant_claims": "claims", "respondent_claims": "claims",
"legal_analysis": "discussion",
"conclusion": "summary", "ruling": "summary",
}
_CORPUS_RATIOS_CACHE: dict | None = None
async def measure_corpus_ratios() -> dict:
"""Measure ACTUAL section %-of-total from Dafna's style_corpus, averaged per
outcome — the empirical counterpart to lessons.GOLDEN_RATIOS (T10). Splits each
decision via chunker (accurate, not the filtered exemplars). Cached for the
process. Returns {outcome: {"n": int, "sections": {sec: pct}}}."""
global _CORPUS_RATIOS_CACHE
if _CORPUS_RATIOS_CACHE is not None:
return _CORPUS_RATIOS_CACHE
from legal_mcp.services.chunker import _split_into_sections
pool = await db.get_pool()
async with pool.acquire() as conn:
rows = await conn.fetch("SELECT full_text, outcome FROM style_corpus WHERE full_text <> ''")
# Per-outcome AND an "_all" aggregate. style_corpus.outcome is currently
# unpopulated for the imported corpus, so per-outcome may be empty — "_all"
# is the meaningful signal today, and per-outcome becomes live once outcomes
# are backfilled. No silent loss: callers see which buckets have data via n.
by_outcome: dict[str, list[dict]] = {}
for r in rows:
sect_words: dict[str, int] = {}
for stype, stext in _split_into_sections(r["full_text"]):
g = _CHUNK_SECTION_TO_GOLDEN.get(stype)
if g:
sect_words[g] = sect_words.get(g, 0) + len(stext.split())
total = sum(sect_words.values())
if total < 100: # sections didn't parse — skip
continue
pct = {s: w / total * 100 for s, w in sect_words.items()}
by_outcome.setdefault("_all", []).append(pct)
outcome = canonical_outcome(r["outcome"] or "")
if outcome:
by_outcome.setdefault(outcome, []).append(pct)
result: dict = {}
for outcome, decs in by_outcome.items():
avg = {}
for sec in ("background", "claims", "discussion", "summary"):
vals = [d.get(sec, 0.0) for d in decs]
if vals:
avg[sec] = round(sum(vals) / len(vals), 1)
result[outcome] = {"n": len(decs), "sections": avg}
_CORPUS_RATIOS_CACHE = result
return result
def count_anti_patterns(text: str) -> dict:
"""Count each anti-pattern occurrence in text. Lower = closer to Dafna."""