diff --git a/docs/spec/07-learning.md b/docs/spec/07-learning.md index 0865259..5c37194 100644 --- a/docs/spec/07-learning.md +++ b/docs/spec/07-learning.md @@ -271,6 +271,26 @@ LegalBench (gemini-2.5-flash) · Trust-or-Escalate (ICLR 2025) | סטטוס: ver החלטת-יו"ר 2026-06-19; מקור-אמת: [`../legal-principles-redesign.md`](../legal-principles-redesign.md). **הפרה ידועה:** — (חדש) +### INV-LRN8: סינתזת-לקחים מעוגנת + מגודרת-שער-מדורג (#158 → G2/G10/INV-AH) +**כלל:** לקחי-סגנון (`decision_lessons`) חופפים מאוחדים ל**לקח-על אחד** עשיר ומוכלל, כך +שהסט שזורם לכותב **קטֵן ומשתבח** (פותר את החיתוך-השקט limit=15, #157) — Authorial Style +Profiling (§0.1: פרופיל-מופשט מנצח ערימת-דוגמאות). המנגנון מחקה את סינתזת-הקנוני (INV-LRN6) +**על אותה טבלה** — אין מאגר-מקביל (G2): (א) אשכול greedy לפי cosine (`LESSON_SYNTH_CLUSTER_THRESHOLD`) +בתוך shard של `practice_area`+`category`; (ב) מיזוג ע"י `claude_session` **מעוגן-מקור** — נובע +מלקחי-המקור בלבד, סגנון-בלבד (לא מהות, INV-LRN5), abstain אם לא-מתמזג (INV-AH); (ג) **שער-drift** — +הלקח-הממוזג מוטמע-מחדש ומושווה (cosine) ל-centroid האשכול; מתחת ל-`LESSON_SYNTH_DRIFT_FLOOR` +נדחה. הלקח-על נכתב `source='synthesis'` עם `synthesized_from`, והמקורות → `review_status='superseded'` +(provenance, לא נצרכים-כותב). **שער מדורג-הפיך (הכרעת-יו"ר 2026-06-28):** מאחר שכל המקורות `approved`, +הלקח-על זורם `approved` עם veto-יו"ר ב-/training — דחייתו משחזרת את המקורות ל-`approved` +(`db.revert_lesson_synthesis`). idempotency: lookup-cosine מול synthesis קיים לפני INSERT. +**מסלול-יחיד (G2):** הכלי `lesson_synthesize_pending` והסקריפט `backfill_lesson_synthesis.py` עוברים +שניהם דרך `services/lesson_synthesis.py`. audit CSV ב-`data/audit/lesson-synthesis-*.csv`. +**מקורות:** Authorial Style Profiling · grounding-vs-hallucination (Stanford RegLab) · CoVe (arXiv:2309.11495) | סטטוס: verified +**אכיפה:** `services/lesson_synthesis.py` (אשכול/מיזוג/drift), `db.{fetch_synthesis_candidates, +apply_lesson_synthesis,revert_lesson_synthesis,nearest_synthesis_lesson,synthesis_shards}`, SCHEMA_V46 +(`embedding`+`synthesized_from`). config `LESSON_SYNTH_*`. +**הפרה ידועה:** — (חדש) + --- ## 4. הג'ובים המתוזמנים (תמיכת-תשתית ללולאה) diff --git a/mcp-server/src/legal_mcp/config.py b/mcp-server/src/legal_mcp/config.py index 16105c2..daf4158 100644 --- a/mcp-server/src/legal_mcp/config.py +++ b/mcp-server/src/legal_mcp/config.py @@ -275,6 +275,19 @@ HALACHA_CANONICAL_SYNTH_MODEL = os.environ.get("HALACHA_CANONICAL_SYNTH_MODEL", HALACHA_CANONICAL_SYNTH_EFFORT = os.environ.get("HALACHA_CANONICAL_SYNTH_EFFORT", "high") HALACHA_CANONICAL_SYNTH_DRIFT_FLOOR = float(os.environ.get("HALACHA_CANONICAL_SYNTH_DRIFT_FLOOR", "0.80")) +# Lesson synthesis (#158 / INV-LRN8) — mirrors the canonical-halacha synthesis above +# for decision_lessons: cluster overlapping style lessons (cosine ≥ CLUSTER_THRESHOLD, +# within a practice_area+category shard) and merge each cluster into one richer +# "super-lesson" via a local claude_session pass, grounded in the source lessons +# (INV-AH) with a re-embedding DRIFT_FLOOR guard. Opus by default — chair-facing +# quality. The synthesised set is smaller, so the writer's limit=15 stops truncating +# (the real fix for the silent cap, #157). +LESSON_SYNTH_MODEL = os.environ.get("LESSON_SYNTH_MODEL", HALACHA_EXTRACT_MODEL) +LESSON_SYNTH_EFFORT = os.environ.get("LESSON_SYNTH_EFFORT", "high") +LESSON_SYNTH_DRIFT_FLOOR = float(os.environ.get("LESSON_SYNTH_DRIFT_FLOOR", "0.80")) +# Cosine floor for two lessons to land in the same cluster (greedy, within shard). +LESSON_SYNTH_CLUSTER_THRESHOLD = float(os.environ.get("LESSON_SYNTH_CLUSTER_THRESHOLD", "0.82")) + # Mistral OCR (fallback for scanned PDFs — replaces Google Cloud Vision) MISTRAL_API_KEY = os.environ.get("MISTRAL_API_KEY", "") diff --git a/mcp-server/src/legal_mcp/server.py b/mcp-server/src/legal_mcp/server.py index 99fa0f8..00f360f 100644 --- a/mcp-server/src/legal_mcp/server.py +++ b/mcp-server/src/legal_mcp/server.py @@ -1185,6 +1185,16 @@ async def record_curator_findings(case_number: str, findings: list[dict]) -> str return await workflow.record_curator_findings(case_number, findings) +@mcp.tool() +async def lesson_synthesize_pending( + practice_area: str = "", category: str = "", apply: bool = True, +) -> str: + """סינתזת-לקחים (#158 / INV-LRN8): ממזגת לקחי-סגנון חופפים ל"לקח-על" אחד עשיר (source='synthesis'), + המקורות→superseded; הסט שזורם לכותב קטֵן ומשתבח (limit=15 מפסיק לחתוך). מעוגן-מקור (INV-AH) + + שער-drift; שער מדורג-הפיך (G10). apply=False = dry-run. practice_area/category ריקים = כל ה-shards.""" + return await workflow.lesson_synthesize_pending(practice_area, category, apply) + + @mcp.tool() async def halacha_corroboration(halacha_id: str) -> dict: """החזר את ה-corroboration של הלכה: הציטוטים שמתקפים אותה, הטיפול, וסיכום (X11, read-only).""" diff --git a/mcp-server/src/legal_mcp/services/db.py b/mcp-server/src/legal_mcp/services/db.py index 0aae633..3fe8afa 100644 --- a/mcp-server/src/legal_mcp/services/db.py +++ b/mcp-server/src/legal_mcp/services/db.py @@ -1770,6 +1770,18 @@ CREATE TABLE IF NOT EXISTS style_distance_history ( CREATE INDEX IF NOT EXISTS idx_style_distance_history_measured ON style_distance_history(measured_at); """ +SCHEMA_V46_SQL = """ +-- decision_lessons synthesis (#158 / INV-LRN8): consolidate overlapping lessons into +-- one richer "super-lesson" (source='synthesis', synthesized_from=[merged ids]); the +-- merged sources flip review_status='superseded' (kept as provenance, no longer fed to +-- the writer). embedding powers cosine clustering + the drift guard. Mirrors the +-- canonical-halacha synthesis (V41) ON THE SAME TABLE — no parallel store (G2). +ALTER TABLE decision_lessons ADD COLUMN IF NOT EXISTS embedding vector(1024); +ALTER TABLE decision_lessons ADD COLUMN IF NOT EXISTS synthesized_from UUID[] NOT NULL DEFAULT '{}'; +CREATE INDEX IF NOT EXISTS idx_decision_lessons_vec + ON decision_lessons USING ivfflat (embedding vector_cosine_ops) WITH (lists = 30); +""" + # Stable, arbitrary key for the session-level advisory lock that serialises # schema DDL across processes. Every short-lived process (cron drains, services) @@ -1838,6 +1850,7 @@ async def _apply_schema_ddl(conn: asyncpg.Connection) -> None: await conn.execute(SCHEMA_V43_SQL) await conn.execute(SCHEMA_V44_SQL) await conn.execute(SCHEMA_V45_SQL) + await conn.execute(SCHEMA_V46_SQL) async def init_schema() -> None: @@ -2691,6 +2704,139 @@ async def get_style_corpus_id_by_decision(decision_number: str) -> UUID | None: return row["id"] if row else None +# ── decision_lessons synthesis (#158 / INV-LRN8) ─────────────────── +# Consolidate overlapping lessons into one richer 'synthesis' row; sources are +# marked 'superseded' (provenance, not writer-fed). Mirrors V41 on the same table. + +async def fetch_synthesis_candidates(practice_area: str, category: str) -> list[dict]: + """Lessons eligible for synthesis in one (practice_area, category) shard. + + Only ``review_status='approved'`` lessons that are themselves NOT a synthesis + output and NOT already superseded — i.e. live, writer-fed lessons. Returns id, + lesson_text, review_status, source, style_corpus_id and the stored embedding + (python list or None; the caller lazily backfills NULLs via set_lesson_embedding). + """ + pool = await get_pool() + async with pool.acquire() as conn: + rows = await conn.fetch( + "SELECT dl.id, dl.lesson_text, dl.review_status, dl.source, " + " dl.style_corpus_id, dl.embedding " + "FROM decision_lessons dl JOIN style_corpus sc ON sc.id = dl.style_corpus_id " + "WHERE dl.review_status = 'approved' " + " AND dl.source <> 'synthesis' " + " AND dl.category = $1 " + " AND ($2 = '' OR sc.practice_area = $2) " + "ORDER BY dl.created_at", + category, practice_area, + ) + return [dict(r) for r in rows] + + +async def synthesis_shards(min_size: int = 2) -> list[dict]: + """(practice_area, category) shards that have ≥min_size live approved lessons — + the candidate shards a synthesis pass should consider. Largest first.""" + pool = await get_pool() + rows = await pool.fetch( + "SELECT sc.practice_area AS practice_area, dl.category AS category, count(*) AS n " + "FROM decision_lessons dl JOIN style_corpus sc ON sc.id = dl.style_corpus_id " + "WHERE dl.review_status = 'approved' AND dl.source <> 'synthesis' " + "GROUP BY 1, 2 HAVING count(*) >= $1 ORDER BY count(*) DESC", + min_size, + ) + return [dict(r) for r in rows] + + +async def set_lesson_embedding(lesson_id: UUID, embedding: list[float]) -> None: + """Lazy backfill: store a lesson's embedding so clustering/drift don't re-embed.""" + pool = await get_pool() + async with pool.acquire() as conn: + await conn.execute( + "UPDATE decision_lessons SET embedding = $2 WHERE id = $1", + lesson_id, embedding, + ) + + +async def nearest_synthesis_lesson( + vec: list[float], category: str, threshold: float, +) -> "tuple[str, float] | None": + """Nearest existing synthesis lesson (same category) by cosine, for idempotency — + so a re-run doesn't create a near-duplicate super-lesson. None if below threshold.""" + pool = await get_pool() + row = await pool.fetchrow( + "SELECT id::text AS id, 1 - (embedding <=> $1) AS sim " + "FROM decision_lessons " + "WHERE source = 'synthesis' AND category = $2 AND embedding IS NOT NULL " + " AND review_status <> 'rejected' " + "ORDER BY embedding <=> $1 LIMIT 1", + vec, category, + ) + if not row: + return None + sim = float(row["sim"]) + return (row["id"], sim) if sim >= threshold else None + + +async def apply_lesson_synthesis( + *, + corpus_id: UUID, + lesson_text: str, + category: str, + embedding: list[float], + source_ids: list[UUID], + review_status: str, +) -> dict: + """Atomically commit a synthesis: insert the super-lesson (source='synthesis') + and flip its source lessons to 'superseded' (provenance, no longer writer-fed). + Returns the new row. INV-LRN8 / G2 — single write path.""" + pool = await get_pool() + async with pool.acquire() as conn: + async with conn.transaction(): + row = await conn.fetchrow( + "INSERT INTO decision_lessons " + "(style_corpus_id, lesson_text, category, source, created_by, " + " review_status, embedding, synthesized_from) " + "VALUES ($1, $2, $3, 'synthesis', 'synthesis', $4, $5, $6) " + "RETURNING id, style_corpus_id, lesson_text, category, source, " + " review_status, created_by, created_at, updated_at", + corpus_id, lesson_text, category, review_status, embedding, source_ids, + ) + await conn.execute( + "UPDATE decision_lessons SET review_status = 'superseded', updated_at = now() " + "WHERE id = ANY($1::uuid[])", + source_ids, + ) + return dict(row) if row else {} + + +async def revert_lesson_synthesis(synthesis_id: UUID) -> dict: + """Chair veto of a super-lesson: mark it 'rejected' and restore its source lessons + to 'approved' (so they flow to the writer again). Idempotent; returns counts.""" + pool = await get_pool() + async with pool.acquire() as conn: + async with conn.transaction(): + row = await conn.fetchrow( + "SELECT synthesized_from, source FROM decision_lessons WHERE id = $1", + synthesis_id, + ) + if not row or row["source"] != "synthesis": + return {"reverted": False, "reason": "not a synthesis lesson"} + source_ids = list(row["synthesized_from"] or []) + await conn.execute( + "UPDATE decision_lessons SET review_status = 'rejected', updated_at = now() " + "WHERE id = $1", + synthesis_id, + ) + restored = 0 + if source_ids: + res = await conn.execute( + "UPDATE decision_lessons SET review_status = 'approved', updated_at = now() " + "WHERE id = ANY($1::uuid[]) AND review_status = 'superseded'", + source_ids, + ) + restored = int(res.split()[-1]) if res.split()[-1].isdigit() else 0 + return {"reverted": True, "restored_sources": restored} + + async def update_decision_lesson( lesson_id: UUID, *, diff --git a/mcp-server/src/legal_mcp/services/lesson_synthesis.py b/mcp-server/src/legal_mcp/services/lesson_synthesis.py new file mode 100644 index 0000000..8c9b087 --- /dev/null +++ b/mcp-server/src/legal_mcp/services/lesson_synthesis.py @@ -0,0 +1,256 @@ +"""Decision-lesson synthesis (#158 / INV-LRN8). + +The learning channels (panel, curator, chair) accumulate overlapping ``decision_lessons`` +on the same style dimension. The writer only consumes the 15 most-recent APPROVED ones +per practice_area (a silent cap, #157), so beyond that lessons pile up unused. This pass +**clusters** near-duplicate lessons within a (practice_area, category) shard and **merges** +each cluster into ONE richer, generalised "super-lesson" — so the writer-fed set shrinks +to a small, high-quality profile (Authorial Style Profiling, ספ §0.1) and the cap stops +biting. + +Mirrors the canonical-halacha synthesis (V41 / INV-LRN6) on the SAME table (no parallel +store, G2). Invariants: + • INV-AH — the super-lesson is GROUNDED in the source lessons only; the model abstains + rather than invent, and a re-embedding DRIFT guard rejects a rewrite that + drifts from the cluster centroid. + • INV-LRN1/G10 — graduated gate (chair decision 2026-06-28): since every source is + already ``approved``, the super-lesson flows as ``approved`` (reversible — + chair veto in /training restores the sources). A non-approved source ⇒ proposed. + • G2 — single synthesis path; the MCP tool and the backfill script both call + :func:`run_shard` / :func:`synthesize_cluster` here. + • G9 — every outcome (accepted / abstained / drift_rejected / merged-duplicate) returned. + +LLM calls go through ``claude_session`` (local ``claude -p`` CLI) only — never from the +FastAPI container (see claude_session docstring). +""" + +from __future__ import annotations + +import logging +import math +from uuid import UUID + +from legal_mcp import config +from legal_mcp.services import claude_session, db, embeddings + +logger = logging.getLogger(__name__) + +_SYSTEM = ( + "אתה עורך-דין בכיר המזקק כללי-סגנון-וכתיבה לבסיס-ידע של ועדת ערר לתכנון ובנייה. " + "תפקידך למזג כמה לקחי-סגנון חופפים לכלל אחד, כללי ומדויק, על *איך* כותבים — לא להמציא " + "כלל חדש ולא להוסיף מהות משפטית." +) + + +def _cosine(a, b) -> float: + dot = sum(x * y for x, y in zip(a, b)) + na = math.sqrt(sum(x * x for x in a)) + nb = math.sqrt(sum(y * y for y in b)) + if na == 0 or nb == 0: + return 0.0 + return dot / (na * nb) + + +def _centroid(vecs: list[list[float]]) -> list[float]: + n = len(vecs) + dim = len(vecs[0]) + return [sum(v[i] for v in vecs) / n for i in range(dim)] + + +def _build_prompt(members: list[dict]) -> str: + blocks = [] + for i, m in enumerate(members, 1): + blocks.append(f"### לקח {i}\n{m['lesson_text']}") + evidence = "\n\n".join(blocks) + return f"""{_SYSTEM} + +לקחי-המקור (כולם מאותו תחום וקטגוריה, חופפים בנושא): +{evidence} + +## המשימה +מזג את לקחי-המקור לכלל-סגנון **אחד** עשיר ומוכלל המשותף לכולם. שמר כל ניואנס מובחן שמופיע +באחד הלקחים, אך נסח אותו פעם אחת, נקי וכללי. + +## כללים מחייבים (INV-AH — עיגון, ללא הזיה) +1. **עיגון-מקור בלבד.** הכלל חייב לנבוע מלקחי-המקור שלמעלה. אסור להוסיף כלל, חריג או דוגמה שאינם עולים מהם. +2. **סגנון/שיטה בלבד, לא מהות.** אל תכניס הלכה, עובדה, מספר-תיק או תקדים ספציפי — רק *איך* דפנה כותבת. +3. **כללי ובלתי-תלוי-תיק.** הסר פרטים קונקרטיים; נסח כלל רב-תחולה. +4. **רגיסטר נקי** בעברית, משפט אחד עד שלושה, בלי מילות-מסגרת ("יש לזכור ש...") — רק הכלל עצמו. +5. **הימנעות עדיפה על המצאה.** אם הלקחים אינם באמת מתמזגים לכלל אחד מעוגן — החזר grounded=false. + +## פלט — JSON בלבד, ללא markdown וללא הסבר: +{{ + "lesson_text": "<כלל-הסגנון הממוזג>", + "grounded": true, + "reason": "<משפט קצר: מה אוחד>" +}}""" + + +def _greedy_clusters(candidates: list[dict], threshold: float) -> list[list[dict]]: + """Greedy single-link clustering by cosine over candidate embeddings. Each candidate + has an 'embedding' (python list). Returns clusters of size ≥2 only (singletons are + nothing to merge).""" + remaining = [c for c in candidates if c.get("embedding") is not None] + clusters: list[list[dict]] = [] + used: set = set() + for i, seed in enumerate(remaining): + if seed["id"] in used: + continue + cluster = [seed] + used.add(seed["id"]) + for other in remaining[i + 1:]: + if other["id"] in used: + continue + if _cosine(seed["embedding"], other["embedding"]) >= threshold: + cluster.append(other) + used.add(other["id"]) + if len(cluster) >= 2: + clusters.append(cluster) + return clusters + + +async def _ensure_embeddings(candidates: list[dict]) -> list[dict]: + """Lazy-backfill: embed any candidate whose stored embedding is NULL, persist it, + and return the candidates with embeddings populated (skips ones that still fail).""" + missing = [c for c in candidates if c.get("embedding") is None] + if missing: + vecs = await embeddings.embed_texts([c["lesson_text"] for c in missing]) + for c, v in zip(missing, vecs): + c["embedding"] = list(v) + await db.set_lesson_embedding(c["id"], c["embedding"]) + return [c for c in candidates if c.get("embedding") is not None] + + +async def synthesize_cluster( + members: list[dict], + *, + model: str | None = None, + effort: str | None = None, + drift_floor: float | None = None, +) -> dict: + """Merge one cluster of lessons. PURE — no DB writes. Returns: + {status, proposed, embedding, members:[ids], drift_cosine, reason} + status ∈ {accepted, abstained, drift_rejected, llm_error, too_small}. + """ + model = model or config.LESSON_SYNTH_MODEL + effort = effort or config.LESSON_SYNTH_EFFORT + drift_floor = config.LESSON_SYNTH_DRIFT_FLOOR if drift_floor is None else drift_floor + ids = [str(m["id"]) for m in members] + base = {"members": ids, "proposed": "", "embedding": None, + "drift_cosine": None, "reason": ""} + if len(members) < 2: + return {**base, "status": "too_small", "reason": "cluster < 2"} + + try: + result = await claude_session.query_json( + _build_prompt(members), model=model, effort=effort, tools="", + ) + except Exception as e: + logger.warning("synthesize_cluster %s: LLM error: %s", ids, e) + return {**base, "status": "llm_error", "reason": str(e)} + + if not isinstance(result, dict) or not result.get("lesson_text"): + return {**base, "status": "llm_error", "reason": "malformed LLM output"} + if not result.get("grounded", True): + return {**base, "status": "abstained", + "reason": result.get("reason") or "model abstained (not grounded)"} + + proposed = str(result["lesson_text"]).strip() + if not proposed: + return {**base, "status": "abstained", "reason": "empty proposal"} + + # Drift guard: the merged lesson must stay near the cluster centroid. + new_emb = list((await embeddings.embed_texts([proposed]))[0]) + centroid = _centroid([m["embedding"] for m in members]) + drift = _cosine(new_emb, centroid) + if drift < drift_floor: + return {**base, "status": "drift_rejected", "proposed": proposed, + "drift_cosine": round(drift, 4), + "reason": f"drift {drift:.3f} < floor {drift_floor}"} + + return {**base, "status": "accepted", "proposed": proposed, "embedding": new_emb, + "drift_cosine": round(drift, 4), "reason": result.get("reason") or "merged"} + + +async def run_shard( + practice_area: str, + category: str, + *, + apply: bool, + model: str | None = None, + effort: str | None = None, + drift_floor: float | None = None, + cluster_threshold: float | None = None, +) -> dict: + """Synthesize all clusters in one (practice_area, category) shard. + + Returns {practice_area, category, candidates, clusters:[result...]}. Each result is + a synthesize_cluster outcome augmented with ``applied`` and (when applied) ``new_id``. + With apply=False this is a pure dry-run (no writes beyond lazy embedding backfill). + """ + cluster_threshold = (config.LESSON_SYNTH_CLUSTER_THRESHOLD + if cluster_threshold is None else cluster_threshold) + candidates = await db.fetch_synthesis_candidates(practice_area, category) + candidates = await _ensure_embeddings(candidates) + clusters = _greedy_clusters(candidates, cluster_threshold) + + results = [] + for members in clusters: + res = await synthesize_cluster( + members, model=model, effort=effort, drift_floor=drift_floor, + ) + res["applied"] = False + if apply and res["status"] == "accepted": + # Idempotency: skip if a near-identical synthesis already exists (re-run safe). + dup = await db.nearest_synthesis_lesson( + res["embedding"], category, config.HALACHA_CANONICAL_THRESHOLD, + ) + if dup: + res["status"] = "duplicate_skipped" + res["reason"] = f"near existing synthesis {dup[0]} (sim {dup[1]:.3f})" + else: + # graduated gate: all sources are approved (fetch filter) → approved. + row = await db.apply_lesson_synthesis( + corpus_id=members[0]["style_corpus_id"], + lesson_text=res["proposed"], + category=category, + embedding=res["embedding"], + source_ids=[m["id"] for m in members], + review_status="approved", + ) + res["applied"] = True + res["new_id"] = str(row.get("id", "")) + results.append(res) + + return {"practice_area": practice_area or "*", "category": category, + "candidates": len(candidates), "clusters": results} + + +async def run_pending( + practice_area: str = "", + category: str = "", + *, + apply: bool, + model: str | None = None, + effort: str | None = None, + drift_floor: float | None = None, + cluster_threshold: float | None = None, +) -> list[dict]: + """Run synthesis across shards. If practice_area+category are given, one shard; + otherwise iterate every shard with ≥2 live approved lessons. Single entry point (G2).""" + if practice_area and category: + shards = [{"practice_area": practice_area, "category": category}] + else: + shards = await db.synthesis_shards(min_size=2) + if practice_area: + shards = [s for s in shards if s["practice_area"] == practice_area] + if category: + shards = [s for s in shards if s["category"] == category] + out = [] + for s in shards: + out.append(await run_shard( + s["practice_area"], s["category"], apply=apply, + model=model, effort=effort, drift_floor=drift_floor, + cluster_threshold=cluster_threshold, + )) + return out diff --git a/mcp-server/src/legal_mcp/tools/workflow.py b/mcp-server/src/legal_mcp/tools/workflow.py index 96d1cfb..a272a8e 100644 --- a/mcp-server/src/legal_mcp/tools/workflow.py +++ b/mcp-server/src/legal_mcp/tools/workflow.py @@ -497,6 +497,33 @@ def _norm(s: str) -> str: return " ".join((s or "").split()) +async def lesson_synthesize_pending( + practice_area: str = "", category: str = "", apply: bool = True, +) -> str: + """סינתזת-לקחים (#158 / INV-LRN8): ממזגת לקחי-סגנון חופפים ל"לקח-על" אחד עשיר ומוכלל, + כך שהסט שזורם לכותב קטֵן ומשתבח (התקרה limit=15 מפסיקה לחתוך). מאשכלת לפי דמיון (cosine) + בתוך shard של practice_area+category, ומסנתזת מעוגן-מקור (INV-AH) עם שער-drift. + + Args: + practice_area: לצמצם ל-shard אחד (ריק = כל התחומים). + category: style/structure/lexicon/tabular (ריק = כל הקטגוריות). + apply: True = כותב (לקח-על approved + מקורות→superseded); False = dry-run. + """ + from legal_mcp.services import lesson_synthesis + shards = await lesson_synthesis.run_pending(practice_area, category, apply=apply) + applied = sum(1 for s in shards for c in s["clusters"] if c.get("applied")) + clusters = sum(len(s["clusters"]) for s in shards) + return ok({ + "apply": apply, + "shards": shards, + "clusters_found": clusters, + "synthesized": applied, + }, message=( + f"סינתזת-לקחים: {clusters} אשכולות ב-{len(shards)} shards · " + f"{applied} לקחי-על {'נכתבו (approved)' if apply else 'דמו (dry-run)'}." + )) + + async def list_chair_feedback( case_number: str = "", category: str = "", diff --git a/scripts/SCRIPTS.md b/scripts/SCRIPTS.md index c07f9fb..582617d 100644 --- a/scripts/SCRIPTS.md +++ b/scripts/SCRIPTS.md @@ -71,6 +71,7 @@ | `compute_principle_gold.py` | python | **#153 (נטוש)** — גישת זיהוי-זהב ברמת-עיקרון דרך התאמת-`match_context`→הלכה. **הוחלף** ע"י "מאומת=אזכור" (`build_verified_layer.py`) אחרי שההתאמה נכשלה (match_context=רשימת-הפניות). נשמר לעיון. | deprecated | | `cull_principles.py` | python | **#152 Phase C — סינון רטרואקטיבי של קורפוס-העקרונות דרך פאנל-3 (הפיך).** מריץ על כל עיקרון 'original' קיים את אותו משטר שה-extractor משתמש בו להבא (`services/panel_extraction.panel_keep_score`, G2): 3 שופטים (Claude מקומי + DeepSeek + Gemini) מצביעים keep+score → כלל-האישור (3 קולות→שורד · 2 וציון≥0.85→שורד · 2 ו<0.85→יו"ר · ≤1→נדחה) → תקרת `HALACHA_PANEL_MAX_NEW`=5 לכל החלטה לפי ציון (`apply_cap`). נדחה → `halachot.review_status='rejected'` + ה-canonical שלו `rejected` (הפיך, גיבוי-CSV ב-`data/audit/` לפני כל כתיבה). מרוסן ב-`usage_limits` (עוצר-רך בתקרת-שימוש, resumable). `--dry-run` (ברירת-מחדל) / `--apply` / `--sample N` (החלטות אקראיות) / `--limit N` / `--no-throttle` / `--verbose`. **חובה מקומי** (3 שופטים). הרץ: `cd mcp-server && HOME=/home/chaim .venv/bin/python ../scripts/cull_principles.py --apply`. | **חד-פעמי** (סינון ראשוני) + ניתן-לחזרה | | `backfill_canonical_synthesis.py` | python | **V41 Phase 4 — סינתזת-LLM ל-`canonical_statement` (idempotent + resumable).** עובר על canonicals ב-`review_status='pending_synthesis'` (רב-instance ראשונים) ומזקק לכל אחד ניסוח אחד כללי ומעוגן בציטוטי-המופעים (INV-AH) דרך `services/canonical_synthesis.py` (מסלול-יחיד, G2). שערים: עיגון/הימנעות, **drift-floor** (cosine מול המקור, ברירת-מחדל 0.80 — סטייה גדולה→נשמר המקור), ואיסור ציטוטי-תיק חדשים. בכל מקרה הסטטוס מתקדם ל-`pending_review` לשער-היו"ר (G10/INV-LRN6). מודל Opus (`HALACHA_CANONICAL_SYNTH_MODEL`). מרוסן ע"י `usage_limits` (עוצר-רך בתקרת-שימוש, resumable). `--dry-run` (ברירת-מחדל) / `--apply` / `--sample N` (מדגם אקראי לבדיקה) / `--limit N` / `--no-throttle` / `--verbose`. CSV-audit ל-`data/audit/canonical-synthesis-*.csv`. **חובה מקומי** (claude_session). הרץ: `cd mcp-server && HOME=/home/chaim .venv/bin/python ../scripts/backfill_canonical_synthesis.py --apply`. שוטף: כלי-MCP `canonical_synthesize_pending`. | **חד-פעמי** (המסה הראשונית) + idempotent לחדשים | +| `backfill_lesson_synthesis.py` | python | **#158 / INV-LRN8 — סינתזת-LLM של `decision_lessons` (idempotent + resumable).** עובר על shards של `practice_area`+`category` עם ≥2 לקחי-סגנון `approved`, מאשכל near-duplicates (cosine ≥ `LESSON_SYNTH_CLUSTER_THRESHOLD`), וממזג כל אשכול ל"לקח-על" אחד מעוגן-מקור (INV-AH) עם **drift-floor** (`LESSON_SYNTH_DRIFT_FLOOR` מול centroid) דרך `services/lesson_synthesis.py` (מסלול-יחיד, G2). הלקח-על נכתב `source='synthesis'` (`review_status='approved'`, שער-מדורג-הפיך), המקורות→`superseded`. כך הסט שזורם לכותב קטֵן ומשתבח (פותר את חיתוך limit=15, #157). מודל Opus (`LESSON_SYNTH_MODEL`), מרוסן `usage_limits`. `--dry-run` (ברירת-מחדל) / `--apply` / `--practice-area` / `--category` / `--no-throttle` / `--verbose`. CSV-audit ל-`data/audit/lesson-synthesis-*.csv`. **חובה מקומי** (claude_session). הרץ: `cd mcp-server && HOME=/home/chaim .venv/bin/python ../scripts/backfill_lesson_synthesis.py --apply`. שוטף: כלי-MCP `lesson_synthesize_pending`. | idempotent — הרצה לפי צורך/כשמצטברים לקחים | | `halacha_batch_reconcile.py` | python | **#82.7** — dedup חוצה-פסקים offline (שמרני, **dry-run בלבד**). dedup-on-insert משווה רק תוך-פסק; כאן סף מחמיר (cosine ≥0.95, `--cosine`) ולא-הרסני: מאתר זוגות הלכות near-duplicate בין פסקים שונים (pgvector `<=>` exact) עם איתות לקסיקלי (Jaccard/Levenshtein) ומדווח ל-CSV ב-`data/audit/` לסקירת היו"ר. לא מדלג/ממזג/מוחק. `--include-pending`. **`--link`** רושם את הזוגות שנמצאו כ-`equivalent_halachot` (parallel authority, #84.2 — **deprecated post-V41** — השתמש ב-`backfill_canonical_halachot.py --apply` במקום). רץ עם venv של mcp-server. | **deprecated** — הוחלף ב-`backfill_canonical_halachot.py` (V41). נשמר לצורכי audit | | `calibrate_halacha_dedup.py` | python | **#82.1** — כיול ספי ה-dedup הלקסיקלי (#82.3) מול gold-set הניקוי. קורא `halacha-cleanup-manifest-*.csv` (זוגות duplicate↔survivor מתויגי-אדם), טוען טקסט-survivor מה-DB, ו-sweep של (jaccard_min × levenshtein_min) עם P/R/F1, מסמן את נקודת-העבודה המוגדרת. אימת ש-(0.55, 0.70) → **precision 1.0** (אפס false-merge), recall 0.30 — מתאים לאיתות-משני שחוסם auto-approve. `--manifest `. רץ עם venv של mcp-server | חד-פעמי — כיול (בוצע 2026-06-06) | | `ab_halacha_opus48.py` | python | **A/B לא-הרסני לחילוץ הלכות (Claude)** — מריץ מחדש חילוץ הלכות על פסק-דין בודד דרך מודל/effort נבחרים (`AB_MODEL`/`AB_EFFORT`, ברירת-מחדל `claude-opus-4-8`/`xhigh`) ומשווה לסטטיסטיקות ההלכות הקיימות ב-DB **בלי למחוק/לכתוב כלום**. משכפל את `halacha_extractor.extract()` (אותם פרומפטים, בחירת-צ'אנקים, אימות-ציטוט) ומחליף רק את קריאת ה-LLM ב-`claude -p --model --effort`. מפיק `data/ab_halacha__.json`. הרצה: `DOTENV_PATH=/home/chaim/.env DATA_DIR=.../data .venv/bin/python scripts/ab_halacha_opus48.py `. **ממצא 2026-05-31 (שטיין 1128-08-20):** Opus 4.8@xhigh חילץ 51 מול 124 בייצור (100% quote-verified מול 96%) אך ביטחון מכויל-נמוך יותר (חציון 0.75 מול 0.82) — ולכן **לא** מקטין את תור-האישור-הידני תחת sweep אוטו-אישור conf≥0.78 (26 מול 24). שיפור איכות, לא צמצום-תור. | ידני (החלטת מודל-חילוץ) | diff --git a/scripts/backfill_lesson_synthesis.py b/scripts/backfill_lesson_synthesis.py new file mode 100644 index 0000000..90141db --- /dev/null +++ b/scripts/backfill_lesson_synthesis.py @@ -0,0 +1,147 @@ +#!/usr/bin/env python3 +"""Backfill — LLM synthesis of decision_lessons (#158 / INV-LRN8). + +WHAT THIS DOES +-------------- +Walks (practice_area, category) shards that hold ≥2 live APPROVED style lessons, +clusters near-duplicates (cosine), and asks a local ``claude_session`` model (Opus +by default) to merge each cluster into ONE richer, generalised "super-lesson" — +grounded in the source lessons (INV-AH) with a drift guard. Accepted merges are +written as ``source='synthesis'`` rows (review_status='approved', graduated gate) +and their sources flip to ``superseded`` (provenance, no longer writer-fed). The +writer-fed set shrinks, so its limit=15 stops truncating (#157). + +All logic lives in services/lesson_synthesis.py (G2) — this is the batch driver: +shard ordering, throttling, dry-run reporting and a CSV audit trail. + +IDEMPOTENCY / RESUME +-------------------- +Re-running is safe: superseded sources are excluded from candidates, and an accepted +merge that matches an existing synthesis (cosine) is skipped (duplicate_skipped). + +USAGE +----- +cd ~/legal-ai/mcp-server +.venv/bin/python ../scripts/backfill_lesson_synthesis.py --dry-run # all shards, no writes +.venv/bin/python ../scripts/backfill_lesson_synthesis.py --dry-run --practice-area rishuy_uvniya --category style +.venv/bin/python ../scripts/backfill_lesson_synthesis.py --apply # full throttled run +""" +from __future__ import annotations + +import argparse +import asyncio +import csv +import os +import sys +from collections import Counter +from datetime import datetime, timezone + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "mcp-server", "src")) + +from legal_mcp.services import db, lesson_synthesis # noqa: E402 + +try: # stdlib-only module, importable from system python too + from legal_mcp.services import usage_limits +except Exception: # pragma: no cover + usage_limits = None + +AUDIT_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "audit") + + +def _throttled() -> tuple[bool, str]: + if usage_limits is None: + return False, "usage_limits unavailable" + usage = usage_limits.subscription_usage() + if usage is None: + return False, "usage read failed (proceeding)" + over, _reset, detail = usage_limits.ceiling_status(usage) + return over, detail + + +def _short(s: str, n: int = 100) -> str: + s = (s or "").replace("\n", " ") + return s if len(s) <= n else s[: n - 1] + "…" + + +async def _run(apply: bool, practice_area: str, category: str, + throttle: bool, verbose: bool) -> int: + if practice_area and category: + shards = [{"practice_area": practice_area, "category": category}] + else: + shards = await db.synthesis_shards(min_size=2) + if practice_area: + shards = [s for s in shards if s["practice_area"] == practice_area] + if category: + shards = [s for s in shards if s["category"] == category] + + mode = "APPLY" if apply else "DRY-RUN" + print(f"[{mode}] {len(shards)} shards with ≥2 approved lessons " + f"(throttle={'on' if throttle else 'off'})\n") + if not shards: + print("nothing to do.") + return 0 + + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + os.makedirs(AUDIT_DIR, exist_ok=True) + audit_path = os.path.join( + AUDIT_DIR, f"lesson-synthesis-{'apply' if apply else 'dryrun'}-{stamp}.csv") + counts: Counter[str] = Counter() + stopped = False + + with open(audit_path, "w", newline="", encoding="utf-8") as fh: + w = csv.writer(fh) + w.writerow(["practice_area", "category", "cluster_size", "status", + "drift_cosine", "applied", "reason", "after", "member_ids"]) + for n, s in enumerate(shards, 1): + if throttle: + over, detail = _throttled() + if over: + print(f"\n⏸ usage ceiling reached ({detail}) — stopping at " + f"shard {n - 1}/{len(shards)}. Re-run to resume.") + stopped = True + break + pa, cat = s["practice_area"], s["category"] + res = await lesson_synthesis.run_shard(pa, cat, apply=apply) + print(f"[{n}/{len(shards)}] {pa}/{cat}: {res['candidates']} candidates → " + f"{len(res['clusters'])} clusters") + for c in res["clusters"]: + counts[c["status"]] += 1 + w.writerow([pa, cat, len(c["members"]), c["status"], + c.get("drift_cosine"), c.get("applied"), + c.get("reason", ""), c.get("proposed", ""), + "|".join(c["members"])]) + mark = {"accepted": "✓", "duplicate_skipped": "=", "abstained": "·", + "drift_rejected": "✗", "llm_error": "!", "too_small": "·"}.get(c["status"], "?") + print(f" {mark} {c['status']:<18} size={len(c['members'])} " + f"drift={c.get('drift_cosine')}{' [written]' if c.get('applied') else ''}") + if verbose and c.get("proposed"): + print(f" → {_short(c['proposed'])}") + + processed = sum(counts.values()) + print(f"\n── summary ({mode}) — {processed} clusters" + f"{' (stopped early)' if stopped else ''} ──") + for status, c in counts.most_common(): + print(f" {status:<18} {c}") + print(f"\naudit CSV: {audit_path}") + if not apply: + print("dry-run — nothing written. Re-run with --apply to commit.") + return 0 + + +def main() -> int: + p = argparse.ArgumentParser(description="LLM synthesis of decision_lessons (#158 / INV-LRN8)") + p.add_argument("--apply", action="store_true", help="commit to the DB (default: dry-run)") + p.add_argument("--dry-run", action="store_true", help="explicit dry-run (default)") + p.add_argument("--practice-area", default="", help="limit to one practice_area") + p.add_argument("--category", default="", help="limit to one category (style/structure/lexicon/tabular)") + p.add_argument("--no-throttle", action="store_true", help="skip usage-ceiling checks") + p.add_argument("--verbose", action="store_true", help="print merged text per cluster") + args = p.parse_args() + return asyncio.run(_run( + apply=args.apply, practice_area=args.practice_area, category=args.category, + throttle=not args.no_throttle, verbose=args.verbose, + )) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/web/app.py b/web/app.py index 728be50..c8cfedd 100644 --- a/web/app.py +++ b/web/app.py @@ -1591,6 +1591,13 @@ async def patch_corpus_lesson(lesson_id: str, body: LessonPatch): raise HTTPException(400, f"invalid category; allowed: {sorted(_LESSON_CATEGORIES)}") if body.review_status is not None and body.review_status not in _LESSON_REVIEW_STATUSES: raise HTTPException(400, f"invalid review_status; allowed: {sorted(_LESSON_REVIEW_STATUSES)}") + # Veto of a synthesized super-lesson (#158/INV-LRN8): rejecting it must also + # restore its merged source lessons (review_status superseded → approved) so they + # flow to the writer again. revert_lesson_synthesis no-ops on non-synthesis rows. + if body.review_status == "rejected": + rev = await db.revert_lesson_synthesis(lid) + if rev.get("reverted"): + return {"updated": True, "review_status": "rejected", **rev} result = await db.update_decision_lesson( lid, lesson_text=body.lesson_text,