diff --git a/AGENTS.md b/AGENTS.md index 399f93ba..a0d1c5b9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -73,8 +73,11 @@ input_df → evaluate-repair loop (up to max_repair_iterations): EvaluateWorkflow → leakage_mass, utility_score, _needs_repair RepairWorkflow → _rewritten_text (failing rows only) - → FinalJudgeWorkflow (non-critical) → _judge_evaluation, needs_human_review → output: {text_col}_rewritten, utility_score, leakage_mass, needs_human_review, … + +Later, `RewriteWorkflow.evaluate()` can run the non-critical judge steps: + → DetectionJudgeWorkflow → detection_valid, detection_invalid_entities + → FinalJudgeWorkflow → judge_evaluation ``` Records with no detected entities skip all LLM sub-workflows and pass through with default metrics (utility=1.0, leakage=0.0). diff --git a/src/anonymizer/engine/rewrite/rewrite_workflow.py b/src/anonymizer/engine/rewrite/rewrite_workflow.py index a948fd9e..2a118445 100644 --- a/src/anonymizer/engine/rewrite/rewrite_workflow.py +++ b/src/anonymizer/engine/rewrite/rewrite_workflow.py @@ -321,17 +321,6 @@ def run( ) all_failed.extend(eval_repair_failed) - # --- Step 6: final judge (non-critical) --- - entity_rows, judge_failed = self._run_final_judge( - entity_rows, - model_configs=model_configs, - selected_models=selected_models, - privacy_goal=privacy_goal, - evaluation=evaluation, - preview_num_records=preview_num_records, - ) - all_failed.extend(judge_failed) - # --- Merge and return --- _apply_passthrough_defaults(passthrough_rows) combined = merge_and_reorder(entity_rows, passthrough_rows) @@ -452,31 +441,29 @@ def _run_final_judge( df: pd.DataFrame, *, model_configs: list[ModelConfig], - selected_models: RewriteModelSelection, + selected_models: EvaluateModelSelection, privacy_goal: PrivacyGoal, - evaluation: EvaluationCriteria, preview_num_records: int | None, ) -> tuple[pd.DataFrame, list[FailedRecord]]: try: judge_columns = self._judge_wf.columns( selected_models=selected_models, privacy_goal=privacy_goal, - evaluation=evaluation, ) + effective_preview = min(preview_num_records, len(df)) if preview_num_records is not None else None judge_seed = select_seed_cols(df, derive_seed_columns(judge_columns, df)) judge_result = self._adapter.run_workflow( judge_seed, model_configs=model_configs, columns=judge_columns, workflow_name="rewrite-final-judge", - preview_num_records=preview_num_records, + preview_num_records=effective_preview, ) df = _join_judge_columns(df, judge_result.dataframe) return df, judge_result.failed_records except Exception: - logger.warning("Final judge step failed; populating defaults", exc_info=True) + logger.warning("Final judge step failed; defaulting to judge_evaluation=None", exc_info=True) df[COL_JUDGE_EVALUATION] = None - df[COL_NEEDS_HUMAN_REVIEW] = True return df, [] # --------------------------------------------------------------------------- @@ -540,25 +527,14 @@ def evaluate( entity_rows[COL_DETECTION_INVALID_ENTITIES] = None # --- Holistic judge (privacy / quality / style) --- - try: - judge_columns = self._judge_wf.columns( - selected_models=selected_models, - privacy_goal=privacy_goal, - ) - effective_preview = min(preview_num_records, len(entity_rows)) if preview_num_records is not None else None - judge_seed = select_seed_cols(entity_rows, derive_seed_columns(judge_columns, entity_rows)) - judge_result = self._adapter.run_workflow( - judge_seed, - model_configs=model_configs, - columns=judge_columns, - workflow_name="rewrite-final-judge", - preview_num_records=effective_preview, - ) - entity_rows = _join_judge_columns(entity_rows, judge_result.dataframe) - all_failed.extend(judge_result.failed_records) - except Exception: - logger.warning("Final judge step failed; defaulting to judge_evaluation=None", exc_info=True) - entity_rows[COL_JUDGE_EVALUATION] = None + entity_rows, judge_failed = self._run_final_judge( + entity_rows, + model_configs=model_configs, + selected_models=selected_models, + privacy_goal=privacy_goal, + preview_num_records=preview_num_records, + ) + all_failed.extend(judge_failed) combined = merge_and_reorder(entity_rows, passthrough_rows) return RewriteResult(dataframe=combined, failed_records=all_failed) diff --git a/src/anonymizer/measurement/metrics/llm_calls.py b/src/anonymizer/measurement/metrics/llm_calls.py index 545c61cf..74489d18 100644 --- a/src/anonymizer/measurement/metrics/llm_calls.py +++ b/src/anonymizer/measurement/metrics/llm_calls.py @@ -35,7 +35,7 @@ def estimate_llm_calls_by_stage( "rewrite_pipeline": 5 if rewrite_body_calls else 0, "rewrite_evaluate": 3 * (1 + repair_iterations) if rewrite_body_calls else 0, "rewrite_repair": repair_iterations if rewrite_body_calls else 0, - "rewrite_final_judge": 1 if rewrite_body_calls else 0, + "rewrite_final_judge": 0, } diff --git a/tests/engine/test_rewrite_workflow.py b/tests/engine/test_rewrite_workflow.py index e84ec96d..9d85f676 100644 --- a/tests/engine/test_rewrite_workflow.py +++ b/tests/engine/test_rewrite_workflow.py @@ -508,10 +508,6 @@ def test_repair_loop_runs_up_to_max_iterations( repaired_df = eval_needs_repair.copy() repaired_df[COL_REWRITTEN_TEXT_NEXT] = "Repaired text" - judge_df = repaired_df.copy() - judge_df[COL_JUDGE_EVALUATION] = None - judge_df[COL_NEEDS_HUMAN_REVIEW] = True - replace_df = stub_df_with_entities.copy() replace_df["_replacement_map"] = [{"replacements": []}] @@ -528,8 +524,6 @@ def test_repair_loop_runs_up_to_max_iterations( WorkflowRunResult(dataframe=repaired_df, failed_records=[]), # evaluate-final (after loop exhaustion) WorkflowRunResult(dataframe=eval_needs_repair, failed_records=[]), - # judge - WorkflowRunResult(dataframe=judge_df, failed_records=[]), ] with patch(_REPLACE_PATCH) as mock_replace_cls: @@ -583,14 +577,6 @@ def test_only_failing_rows_sent_to_repair( eval_after_repair[COL_LEAKAGE_MASS] = 0.1 eval_after_repair[COL_ANY_HIGH_LEAKED] = False - judge_df = rewrite_gen_df.copy() - judge_df[COL_NEEDS_REPAIR] = False - judge_df[COL_UTILITY_SCORE] = 0.9 - judge_df[COL_LEAKAGE_MASS] = 0.1 - judge_df[COL_ANY_HIGH_LEAKED] = False - judge_df[COL_JUDGE_EVALUATION] = None - judge_df[COL_NEEDS_HUMAN_REVIEW] = False - replace_df = df.copy() replace_df["_replacement_map"] = [{"replacements": []}, {"replacements": []}] @@ -599,7 +585,6 @@ def test_only_failing_rows_sent_to_repair( WorkflowRunResult(dataframe=eval_df, failed_records=[]), WorkflowRunResult(dataframe=repaired_row, failed_records=[]), WorkflowRunResult(dataframe=eval_after_repair, failed_records=[]), - WorkflowRunResult(dataframe=judge_df, failed_records=[]), ] with patch(_REPLACE_PATCH) as mock_replace_cls: @@ -654,10 +639,6 @@ def test_repair_iterations_tracked_per_row( eval_pass[COL_ANY_HIGH_LEAKED] = False eval_pass[COL_REPAIR_ITERATIONS] = 1 - judge_df = eval_pass.copy() - judge_df[COL_JUDGE_EVALUATION] = None - judge_df[COL_NEEDS_HUMAN_REVIEW] = False - replace_df = stub_df_with_entities.copy() replace_df["_replacement_map"] = [{"replacements": []}] @@ -666,7 +647,6 @@ def test_repair_iterations_tracked_per_row( WorkflowRunResult(dataframe=eval_needs_repair, failed_records=[]), WorkflowRunResult(dataframe=repaired_df, failed_records=[]), WorkflowRunResult(dataframe=eval_pass, failed_records=[]), - WorkflowRunResult(dataframe=judge_df, failed_records=[]), ] with patch(_REPLACE_PATCH) as mock_replace_cls: @@ -744,10 +724,6 @@ def test_evaluate_dropping_rows_degrades_gracefully( eval_df[COL_LEAKAGE_MASS] = 0.1 eval_df[COL_ANY_HIGH_LEAKED] = False - judge_df = eval_df.copy() - judge_df[COL_JUDGE_EVALUATION] = None - judge_df[COL_NEEDS_HUMAN_REVIEW] = False - replace_df = df.copy() replace_df["_replacement_map"] = [{"replacements": []}, {"replacements": []}] @@ -756,7 +732,6 @@ def test_evaluate_dropping_rows_degrades_gracefully( adapter.run_workflow.side_effect = [ WorkflowRunResult(dataframe=rewrite_gen_df, failed_records=[]), WorkflowRunResult(dataframe=eval_df, failed_records=[eval_failed]), - WorkflowRunResult(dataframe=judge_df, failed_records=[]), ] with patch(_REPLACE_PATCH) as mock_replace_cls: @@ -816,10 +791,6 @@ def test_repair_dropping_rows_degrades_gracefully( eval_after[COL_LEAKAGE_MASS] = 0.1 eval_after[COL_ANY_HIGH_LEAKED] = False - judge_df = eval_after.copy() - judge_df[COL_JUDGE_EVALUATION] = None - judge_df[COL_NEEDS_HUMAN_REVIEW] = False - replace_df = df.copy() replace_df["_replacement_map"] = [{"replacements": []}, {"replacements": []}] @@ -831,8 +802,6 @@ def test_repair_dropping_rows_degrades_gracefully( WorkflowRunResult(dataframe=repaired_df, failed_records=[repair_failed]), # re-evaluate WorkflowRunResult(dataframe=eval_after, failed_records=[]), - # judge - WorkflowRunResult(dataframe=judge_df, failed_records=[]), ] with patch(_REPLACE_PATCH) as mock_replace_cls: @@ -884,17 +853,12 @@ def test_passthrough_rows_get_defaults( eval_df[COL_LEAKAGE_MASS] = 0.3 eval_df[COL_ANY_HIGH_LEAKED] = False - judge_df = eval_df.copy() - judge_df[COL_JUDGE_EVALUATION] = None - judge_df[COL_NEEDS_HUMAN_REVIEW] = False - replace_df = entity_df.copy() replace_df["_replacement_map"] = [{"replacements": []}] adapter.run_workflow.side_effect = [ WorkflowRunResult(dataframe=rewrite_gen_df, failed_records=[]), WorkflowRunResult(dataframe=eval_df, failed_records=[]), - WorkflowRunResult(dataframe=judge_df, failed_records=[]), ] with patch(_REPLACE_PATCH) as mock_replace_cls: diff --git a/tests/test_measurement.py b/tests/test_measurement.py index a77b19f7..5c9403df 100644 --- a/tests/test_measurement.py +++ b/tests/test_measurement.py @@ -1770,7 +1770,7 @@ def test_rewrite_llm_call_estimate_splits_by_stage() -> None: "rewrite_pipeline": 5, "rewrite_evaluate": 9, "rewrite_repair": 2, - "rewrite_final_judge": 1, + "rewrite_final_judge": 0, }