@@ -166,17 +166,33 @@ def aggregate_cell(per_run: list[dict]) -> dict:
""" Mean each metric across repeated generations of the same (case, block, effort). """
if not per_run :
return { " distance " : 1.0 , " anti_pattern_total " : 0.0 , " change_percent " : 100.0 ,
" golden_ratio_deviation_pp " : None , " n " : 0 }
" golden_ratio_deviation_pp " : None , " anti_by_pattern " : { } , " " : 0 }
ratios = [ r [ " golden_ratio_deviation_pp " ] for r in per_run if r . get ( " golden_ratio_deviation_pp " ) is not None ]
return {
" distance " : round ( mean ( r [ " distance " ] for r in per_run ) , 4 ) ,
" anti_pattern_total " : round ( mean ( r [ " anti_pattern_total " ] for r in per_run ) , 2 ) ,
" change_percent " : round ( mean ( r [ " change_percent " ] for r in per_run ) , 2 ) ,
" golden_ratio_deviation_pp " : round ( mean ( ratios ) , 2 ) if ratios else None ,
" anti_by_pattern " : _mean_by_pattern ( per_run ) ,
" n " : len ( per_run ) ,
}
def _mean_by_pattern ( per_run : list [ dict ] ) - > dict :
""" Mean hits PER anti-pattern name across runs — the ' which rule broke ' view.
A pattern absent from a run counts as 0 (count_anti_patterns omits zero-hit
keys), so the mean is over ALL runs, not only the ones that tripped it.
"""
names : set [ str ] = set ( )
for r in per_run :
names | = set ( ( r . get ( " anti_by_pattern " ) or { } ) . keys ( ) )
return {
name : round ( mean ( ( r . get ( " anti_by_pattern " ) or { } ) . get ( name , 0 ) for r in per_run ) , 2 )
for name in sorted ( names )
}
def _current_default ( block_id : str ) - > str | None :
from legal_mcp . services . block_writer import BLOCK_CONFIG , DEFAULT_EFFORT
cfg = BLOCK_CONFIG . get ( block_id , { } )
@@ -420,13 +436,30 @@ async def _finals_for_calibration(case_filter: str | None) -> list[dict]:
async def _score_cell ( case_id , block_id : str , effort : str , final_section : str ,
final_total_words : int , outcome : str , repeats : int ) - > dict :
""" Generate `block_id` at `effort` `repeats` times; score each vs the final section. """
final_total_words : int , outcome : str , repeats : int ,
model : str | None = None , instructions : str = " " ) - > dict :
""" Generate `block_id` at `effort` `repeats` times; score each vs the final section.
`model` (optional) A/Bs the generation model via write_block(model_override=…).
None ⇒ the pinned GENERATION_MODEL, i.e. the production path unchanged.
`instructions` (optional) is appended to the block prompt for EVERY cell in
the run — a prompt-variant A/B (e.g. an explicit formatting rule). It is
applied to all models so the comparison stays a model comparison rather
than silently becoming a prompt comparison.
"""
from legal_mcp . services import block_writer
from legal_mcp . services . style_distance import block_distance_to_final
runs : list [ dict ] = [ ]
models_used : list [ str ] = [ ]
for _ in range ( repeats ) :
res = await block_writer . write_block ( case_id , block_id , effort_override = effort )
res = await block_writer . write_block (
case_id , block_id , instructions = instructions ,
effort_override = effort , model_override = model ,
)
# Record what the CLI was actually asked to run, so a silent fallback to
# a different build is visible in the report rather than mis-attributed.
models_used . append ( res . get ( " model_used " ) or " ? " )
scored = block_distance_to_final (
block_id , res . get ( " content " , " " ) , final_section , outcome ,
section_target_total_words = final_total_words ,
@@ -434,6 +467,8 @@ async def _score_cell(case_id, block_id: str, effort: str, final_section: str,
runs . append ( scored )
agg = aggregate_cell ( runs )
agg [ " effort " ] = effort
agg [ " model " ] = model
agg [ " models_used " ] = sorted ( set ( models_used ) )
agg [ " runs " ] = runs
return agg
@@ -446,6 +481,7 @@ async def _run(args, ts: str) -> dict:
efforts = args . efforts
blocks = args . blocks
models = args . models
finals = await _finals_for_calibration ( args . case )
cases_meta = [ ]
@@ -468,11 +504,15 @@ async def _run(args, ts: str) -> dict:
section = _BLOCK_TO_SECTION . get ( block_id )
plan [ block_id ] = [ c for c in cases_meta if section and c [ " sections " ] . get ( section ) ]
total_cells = sum ( len ( plan [ b ] ) for b in blocks ) * len ( efforts ) * args . repeats
total_cells = sum ( len ( plan [ b ] ) for b in blocks ) * len ( efforts ) * args . repeats * len ( models )
grid_summary = {
" n_finals " : len ( cases_meta ) ,
" finals " : [ c [ " case_number " ] for c in cases_meta ] ,
" blocks " : blocks , " efforts " : efforts , " repeats " : args . repeats ,
" models " : models ,
# Provenance: a prompt-variant run is NOT comparable to a baseline run,
# so the instruction text is recorded in the report, not just the shell.
" instructions " : getattr ( args , " instructions " , " " ) or " " ,
" total_generations " : total_cells ,
" per_block_n " : { b : len ( plan [ b ] ) for b in blocks } ,
}
@@ -480,6 +520,27 @@ async def _run(args, ts: str) -> dict:
if args . dry_run :
return { " dry_run " : True , " grid " : grid_summary , " by_block " : { } }
by_model : dict [ str , dict ] = { }
for model in models :
by_block = await _run_blocks_for_model (
model , blocks , efforts , plan , args , ts , grid_summary , by_model , _BLOCK_TO_SECTION ,
)
by_model [ model ] = by_block
# `by_block` stays the single-model shape (first model) so --rerank and the
# existing per-block report path keep working unchanged (G2 — no second
# result schema); multi-model runs additionally carry by_model.
out = { " dry_run " : False , " grid " : grid_summary , " by_block " : by_model [ models [ 0 ] ] }
if len ( models ) > 1 :
out [ " by_model " ] = by_model
return out
async def _run_blocks_for_model ( model , blocks , efforts , plan , args , ts , grid_summary ,
by_model_so_far , _BLOCK_TO_SECTION ) - > dict :
""" The per-block × """
from uuid import UUID
by_block : dict [ str , dict ] = { }
for block_id in blocks :
section = _BLOCK_TO_SECTION . get ( block_id )
@@ -497,11 +558,12 @@ async def _run(args, ts: str) -> dict:
cell = await _score_cell (
UUID ( c [ " case_id " ] ) , block_id , effort , final_section ,
c [ " final_total_words " ] , c [ " outcome " ] , args . repeats ,
model = model , instructions = getattr ( args , " instructions " , " " ) or " " ,
)
except Exception as exc : # noqa: BLE001 — harness must survive any cell failure
logger . warning (
" calibration cell skipped: case= %s block= %s effort= %s — %s " ,
c [ " case_number " ] , block_id , effort , exc ,
" calibration cell skipped: case= %s block= %s effort= %s model= %s %s " ,
c [ " case_number " ] , block_id , effort , model , exc ,
)
continue
per_effort_runs [ effort ] . append ( cell )
@@ -523,6 +585,7 @@ async def _run(args, ts: str) -> dict:
" anti_pattern_total " : round ( mean ( r [ " anti_pattern_total " ] for r in rows ) , 2 ) ,
" change_percent " : round ( mean ( r [ " change_percent " ] for r in rows ) , 2 ) ,
" golden_ratio_deviation_pp " : round ( mean ( ratios ) , 2 ) if ratios else None ,
" anti_by_pattern " : _mean_by_pattern ( rows ) ,
" n " : len ( rows ) ,
} )
rec = recommend_effort ( effort_rows )
@@ -532,6 +595,11 @@ async def _run(args, ts: str) -> dict:
" recommended " : rec [ " effort " ] if rec else None ,
" confidence " : rec [ " confidence " ] if rec else None ,
" confidence_margin " : rec . get ( " confidence_margin " ) if rec else None ,
" model " : model ,
# Model builds the CLI actually reported across this block's cells —
# a mismatch vs `model` means a silent fallback, not a real A/B.
" models_used " : sorted ( { m for e in per_effort_runs . values ( )
for cell in e for m in cell . get ( " models_used " , [ ] ) } ) ,
" efforts " : effort_rows ,
" per_case " : per_case ,
}
@@ -541,11 +609,15 @@ async def _run(args, ts: str) -> dict:
# Blocks not yet done are simply absent from by_block; _write_report tolerates
# partial results. main() does the final flush once the loop finishes.
try :
_write_report ( { " dry_run " : False , " grid " : grid_summary , " by_block " : by_block } , ts )
snap = { " dry_run " : False , " grid " : grid_summary , " by_block " : by_block }
if by_model_so_far or len ( grid_summary . get ( " models " , [ ] ) ) > 1 :
snap [ " by_model " ] = { * * by_model_so_far , model : by_block }
_write_report ( snap , ts )
except Exception as exc : # noqa: BLE001 — a write hiccup must not abort the run
logger . warning ( " incremental report write failed after block= %s — %s " ,block_id , exc )
logger . warning ( " incremental report write failed after block= %s model= %s — %s " ,
block_id , model , exc )
return { " dry_run " : False , " grid " : grid_summary , " by_block " : by_block }
return by_block
IL_TZ = ZoneInfo ( " Asia/Jerusalem " )
@@ -578,7 +650,10 @@ def _write_report(result: dict, ts: str) -> tuple[Path, Path]:
" ההמלצה אדוויזורית; ההכרעה בידי היו \" ר/המפעיל. \n " ,
f " - בלוקים: { ' , ' . join ( g [ ' blocks ' ] ) } " ,
f " - efforts: { ' , ' . join ( g [ ' efforts ' ] ) } · repeats/cell: { g [ ' repeats ' ] } " ,
f " - models: { ' , ' . join ( m or ' pinned-default ' for m in g . get ( ' models ' , [ None ] ) ) } " ,
f " - סך ייצורי-מודל: { g [ ' total_generations ' ] } " ,
( f " - ⚠️ **וריאנט-פרומפט** (לא בר-השוואה לריצת-בסיס): ` { g [ ' instructions ' ] } ` "
if g . get ( " instructions " ) else " - וריאנט-פרומפט: — (פרומפט ייצור כפי-שהוא) " ) ,
" " ,
]
if result . get ( " dry_run " ) :
@@ -613,6 +688,54 @@ def _write_report(result: dict, ts: str) -> tuple[Path, Path]:
f " | { r [ ' effort ' ] } { star } | { r [ ' distance ' ] : .4f } | { r [ ' anti_pattern_total ' ] } | "
f " { r [ ' change_percent ' ] } | { ratio if ratio is not None else ' — ' } | { r [ ' n ' ] } | " )
lines . append ( " " )
by_model = result . get ( " by_model " ) or { }
if len ( by_model ) > 1 :
lines + = [ " ## השוואת-מודלים (אותו block, אותו effort, אותם סופיים) \n " ,
" | block | effort | model | anti_total | change % | ratioΔpp | distance | n | " ,
" |---|---|---|---|---|---|---|---| " ]
for b in g [ " blocks " ] :
for eff in g [ " efforts " ] :
rows = [ ]
for m , bb in by_model . items ( ) :
for r in ( bb . get ( b ) or { } ) . get ( " efforts " , [ ] ) :
if r [ " effort " ] == eff :
rows . append ( ( m , r ) )
if len ( rows ) < 2 :
continue # nothing to compare for this cell — don't fake a row
best = min ( rows , key = lambda mr : ( mr [ 1 ] [ " anti_pattern_total " ] ,
mr [ 1 ] [ " golden_ratio_deviation_pp " ] or 0 ,
mr [ 1 ] [ " distance " ] ) ) [ 0 ]
for m , r in rows :
ratio = r [ " golden_ratio_deviation_pp " ]
star = " ⭐ " if m == best else " "
lines . append (
f " | { b } | { eff } | { m } { star } | { r [ ' anti_pattern_total ' ] } | "
f " { r [ ' change_percent ' ] } | { ratio if ratio is not None else ' — ' } | "
f " { r [ ' distance ' ] : .4f } | { r [ ' n ' ] } | " )
lines . append ( " " )
# WHICH rule broke — a total alone can't tell you what to fix.
bd_rows = [ ( b , eff , m , r ) for b in g [ " blocks " ] for eff in g [ " efforts " ]
for m , bb in by_model . items ( )
for r in ( bb . get ( b ) or { } ) . get ( " efforts " , [ ] ) if r [ " effort " ] == eff ]
if any ( r . get ( " anti_by_pattern " ) for * _ , r in bd_rows ) :
names = sorted ( { n for * _ , r in bd_rows for n in ( r . get ( " anti_by_pattern " ) or { } ) } )
lines + = [ " ### פילוח אנטי-דפוסים (איזה כלל הופר) \n " ,
" | block | effort | model | " + " | " . join ( names ) + " | " ,
" |---|---|---| " + " ---| " * len ( names ) ]
for b , eff , m , r in bd_rows :
cells = " | " . join ( str ( ( r . get ( " anti_by_pattern " ) or { } ) . get ( n , 0 ) ) for n in names )
lines . append ( f " | { b } | { eff } | { m } | { cells } | " )
lines . append ( " " )
# A silent CLI fallback would make the whole comparison meaningless — surface it.
for m , bb in by_model . items ( ) :
for b , bd in bb . items ( ) :
used = bd . get ( " models_used " ) or [ ]
if used and any ( not u . startswith ( str ( m ) ) for u in used ) :
lines . append ( f " > ⚠️ ** { b } / { m } **: ה-CLI דיווח ` { ' , ' . join ( used ) } ` — "
" ייתכן fallback שקט; ההשוואה לתא זה אינה תקפה. \n " )
lines . append ( " " )
lines . append ( " > דירוג-ההמלצה **style-clean** (#213): anti_total ראשי → ratioΔ → distance (tiebreak). "
" **change % מדווח-לא-מדורג** — מערבב סגנון עם שלמות-תוכן (07-learning §0.7), "
" anti_total הוא הסיגנל הנקי-לסגנון. confidence=⚠️ "
@@ -633,6 +756,12 @@ async def main() -> int:
help = " comma block ids to calibrate " )
ap . add_argument ( " --case " , default = None , help = " restrict to a single case_number " )
ap . add_argument ( " --repeats " , type = int , default = 1 , help = " generations per cell (avg out gen noise) " )
ap . add_argument ( " --models " , default = " " ,
help = " comma generation-model ids to A/B (e.g. claude-opus-4-8,claude-opus-5). "
" Empty (default) = the pinned GENERATION_MODEL, i.e. production unchanged. " )
ap . add_argument ( " --instructions " , default = " " ,
help = " extra prompt instruction appended to EVERY cell (prompt-variant A/B). "
" Applied to all models — the run stays a model comparison. Recorded in the report. " )
args = ap . parse_args ( )
logging . basicConfig ( level = logging . INFO , format = " %(asctime)s %(levelname)s %(message)s " )
@@ -653,6 +782,9 @@ async def main() -> int:
if bad_b :
print ( f " non-calibratable block(s): { bad_b } . valid: { VALID_BLOCKS } " , file = sys . stderr )
return 2
# [None] = "use the pinned GENERATION_MODEL" — keeps the default run byte-identical
# to the pre-#models behaviour instead of hard-coding the id in a second place (G2).
args . models = [ m . strip ( ) for m in args . models . split ( " , " ) if m . strip ( ) ] or [ None ]
ts = _ts ( )
result = await _run ( args , ts )