From 1189817bfeac8b7cb85ab0752ce589cb187748e1 Mon Sep 17 00:00:00 2001 From: memadi Date: Tue, 10 Nov 2026 17:26:50 -0800 Subject: [PATCH 1/4] fix(replace): retry missing substitution mappings once Compare schema-valid maps with the original requested entity-label pairs. Retry missing pairs once while retaining accepted substitutions, and report unresolved or failed retry records without discarding complete records. Refs #388 Signed-off-by: memadi --- docs/concepts/replace.md | 18 +- plans/388/substitution-completeness-retry.md | 76 ++++ src/anonymizer/engine/constants.py | 1 + .../engine/replace/llm_replace_workflow.py | 213 +++++++++- .../engine/replace/replace_runner.py | 2 +- .../ndd/test_substitution_completeness.py | 163 ++++++++ tests/engine/test_replace_runner.py | 10 +- .../engine/test_substitution_completeness.py | 378 ++++++++++++++++++ 8 files changed, 835 insertions(+), 26 deletions(-) create mode 100644 plans/388/substitution-completeness-retry.md create mode 100644 tests/engine/ndd/test_substitution_completeness.py create mode 100644 tests/engine/test_substitution_completeness.py diff --git a/docs/concepts/replace.md b/docs/concepts/replace.md index 29daf31f0..cd91b20e5 100644 --- a/docs/concepts/replace.md +++ b/docs/concepts/replace.md @@ -3,7 +3,7 @@ # Replace -Replace mode replaces each [detected entity](detection.md) with an alternative token. Anonymizer provides four strategies. +Replace mode replaces each [detected entity](detection.md) with an alternative token. Anonymizer provides four strategies. --- @@ -22,6 +22,22 @@ Replace mode replaces each [detected entity](detection.md) with an alternative t Replaces entities with LLM-generated synthetic values that are contextually plausible. This is the only replacement strategy that requires an LLM call. +After schema validation and filtering, Anonymizer checks that every unique +requested `(entity value, label)` pair has a mapping. If mappings are missing, +it retries only the missing pairs once, retaining the accepted substitutions. +A record whose map is complete is returned; a record still missing mappings +after the retry is excluded from successful output and reported in +`failed_records`. Other complete records continue normally. Repeated occurrences +of the same pair need only one mapping, and records without entities need no +replacement-generator call. + +This completeness retry applies to schema-valid responses. DataDesigner's +schema-validation retries still apply to each call. If the completeness retry +fails validation or its workflow fails, the affected records are reported as +failed while complete records are retained. Completeness checks exact pairs; +it does not guarantee the semantic quality of a generated substitute or +validate span offsets. + ```python from anonymizer import AnonymizerConfig, Substitute diff --git a/plans/388/substitution-completeness-retry.md b/plans/388/substitution-completeness-retry.md new file mode 100644 index 000000000..302c3fda8 --- /dev/null +++ b/plans/388/substitution-completeness-retry.md @@ -0,0 +1,76 @@ + + + +# Substitution validation for issue #388 + +For Substitute, compare schema-valid generated mappings with the exact unique +`(entity value, label)` pairs supplied before generation. Retry missing pairs once, +retain already accepted mappings, and fail the affected record if coverage is +still incomplete. Complete records continue and records without entities need no +model call. The user selected this behavior during the investigation of #388. + +## Scope and approach + +Add a validated-map entry point in `engine/replace/llm_replace_workflow.py`, wire +Substitute to it in `replace_runner.py`, and document the successful-output and +failed-record behavior. Keep `generate_map_only()` and rewrite policy unchanged. +Reuse schema validation, filtering, collision handling, and NddAdapter execution. +Snapshot expected pairs before generation; do not infer coverage from entry counts +or model-returned request columns. Track positions so duplicate IDs and reordered +model output retain their original record association. + +Schema validation remains first. This adds one retry for a returned, schema-valid +but incomplete map. DataDesigner's schema-validation retries still apply to each +call. Initial workflow errors retain existing behavior; a wholly failed +completeness retry becomes failures for the affected records while complete +records survive. This work does not introduce permissive transport schemas or +recover invalid entries from a schema-rejected response. + +Reject whitespace-only synthetic values in the shared replacement schema using +`pattern="\\S"` alongside `min_length=1`. This requires at least one non-whitespace +character in both Pydantic validation and the JSON Schema supplied to DataDesigner. +Keep valid one-character values and preserve surrounding whitespace on valid +strings. The shared rule also applies to rewrite replacement maps; their existing +schema-retry and failure policy stays unchanged. Test the exported schema through +DataDesigner's actual response parser as well as direct validation, mixed batches, +and recovery through configured schema retries. + +## Trade-offs and validation + +A targeted retry avoids regenerating accepted values or making another request +for already complete records. Supply the accepted map as retry context so the +missing replacements can remain consistent. Compare exact pairs after filtering; +an unrelated mapping or duplicate does not fill a gap. Repeated occurrences of +the same pair require a single mapping. + +Before discarding a schema-valid Substitute entry, prefer its exact requested +pair. If it has no exact match, reconcile only U+00A0 (no-break space), U+2009 +(thin space), and U+202F (narrow no-break space) with ASCII space in a comparison +key. The label must match exactly and the normalized key must identify exactly +one pair in the full pre-call inventory. Restore that pair's trusted original +without changing source text, offsets, or the synthetic value. An exact entry +takes precedence over an approximate entry regardless of response order. +Ambiguous matches and other text differences use the existing retry/failure +policy. Do not strip or collapse whitespace, normalize case, or apply broader +Unicode normalization. Keep rewrite's existing exact matching unchanged. + +Test recovery without another model call, exact-entry precedence, ambiguity +across the full inventory on both attempts, exact label matching, unsupported +characters, preservation of offsets/text, and the real preview/batch paths. + +Exercise recovery, terminal incomplete retries, retry workflow drops, mixed +batches, no entities, repeated occurrences, mismatched labels/originals, preview +limits, duplicate IDs and output reordering. Verify preview and batch paths +through the real adapter with a local scripted provider. Failure summaries and +new logs contain counts and labels rather than entity values. Run relevant unit +tests and repository checks. No public API or default configuration changes are +planned; the bundled public skill template is unaffected. + +## Rollout + +Create three local commits on `codex/bugfix/388-origin-investigation`, authored +and signed off by memadi: missing-map completeness and one retry, narrow Unicode +whitespace reconciliation, and rejection of whitespace-only synthetic values. +Do not push. Completeness and reconciliation apply only to Substitute; the +nonblank schema rule also applies to rewrite replacement maps. Tests and +documentation make each behavior explicit. diff --git a/src/anonymizer/engine/constants.py b/src/anonymizer/engine/constants.py index f12f3f5b4..e37c3ef20 100644 --- a/src/anonymizer/engine/constants.py +++ b/src/anonymizer/engine/constants.py @@ -48,6 +48,7 @@ COL_REPLACEMENT_MAP = "_replacement_map" COL_REPLACEMENT_MAP_RAW = COL_REPLACEMENT_MAP + "__raw" COL_REPLACEMENT_MAP_SOURCE = "_replacement_map_source" +COL_REPLACEMENT_RETRY_CONTEXT = "_replacement_retry_context" # LlmReplaceWorkflow internal prompt-construction columns. Created by # `LlmReplaceWorkflow.generate_map_only` for the replacement-generator prompt diff --git a/src/anonymizer/engine/replace/llm_replace_workflow.py b/src/anonymizer/engine/replace/llm_replace_workflow.py index 0a24c5825..4d097a263 100644 --- a/src/anonymizer/engine/replace/llm_replace_workflow.py +++ b/src/anonymizer/engine/replace/llm_replace_workflow.py @@ -21,13 +21,16 @@ COL_ENTITY_EXAMPLES, COL_REPLACEMENT_MAP, COL_REPLACEMENT_MAP_SOURCE, + COL_REPLACEMENT_RETRY_CONTEXT, ENTITY_LABEL_EXAMPLES, + _jinja, ) -from anonymizer.engine.ndd.adapter import FailedRecord, NddAdapter +from anonymizer.engine.ndd.adapter import RECORD_ID_COLUMN, FailedRecord, NddAdapter, WorkflowRunResult from anonymizer.engine.ndd.model_loader import resolve_model_alias from anonymizer.engine.prompt_utils import substitute_placeholders -from anonymizer.engine.row_partitioning import merge_and_reorder, split_rows +from anonymizer.engine.row_partitioning import ROW_ORDER_COL, merge_and_reorder, split_rows from anonymizer.engine.schemas import EntitiesByValueSchema, EntityReplacementMapSchema +from anonymizer.interface.errors import AnonymizerWorkflowError logger = logging.getLogger("anonymizer.replace.llm_workflow") REPLACEMENT_MAP_SOURCE_LLM = "llm" @@ -51,7 +54,7 @@ class LlmReplaceWorkflow: def __init__(self, adapter: NddAdapter) -> None: self._adapter = adapter - def generate_map_only( + def generate_validated_maps( self, dataframe: pd.DataFrame, *, @@ -61,8 +64,128 @@ def generate_map_only( entities_column: str = COL_ENTITIES_BY_VALUE, preview_num_records: int | None = None, ) -> LlmReplaceResult: - replace_alias = resolve_model_alias("replacement_generator", selected_models) + """Generate complete Substitute maps, retrying missing pairs once. + + Expected pairs are captured before generation. Schema-valid entries are + filtered before checking coverage; accepted entries survive the retry. + Rows still incomplete after that retry are reported as failed records. + DataDesigner's schema-validation retries apply to each call. A failed + completeness retry also fails only the affected records. + """ + source = dataframe.copy() + source[ROW_ORDER_COL] = range(len(source)) + if RECORD_ID_COLUMN not in source and not source.empty: + source[RECORD_ID_COLUMN] = source.apply(NddAdapter._compute_record_id, axis=1) + expected = [EntitiesByValueSchema.from_raw(raw) for raw in source[entities_column]] + missing = [_entity_pairs(entities) for entities in expected] + accepted: list[dict[str, list[dict[str, str]]]] = [{"replacements": []} for _ in expected] + pending = [position for position, pairs in enumerate(missing) if pairs] + if preview_num_records is not None: + pending = pending[:preview_num_records] + eligible = {position for position, pairs in enumerate(missing) if not pairs} | set(pending) + failed: list[FailedRecord] = [] + dropped: set[int] = set() + + for attempt in range(2): + if not pending: + break + batch = source.iloc[pending].copy() + requested = [_entities_for_pairs(missing[position]) for position in pending] + batch[COL_ENTITIES_FOR_REPLACE] = [_enrich_entities_for_template(entities) for entities in requested] + batch[COL_ENTITIES_FOR_REPLACE_JSON] = batch[COL_ENTITIES_FOR_REPLACE].apply(json.dumps) + batch[COL_ENTITY_EXAMPLES] = [_create_entity_examples(entities) for entities in requested] + batch[COL_REPLACEMENT_RETRY_CONTEXT] = [json.dumps(accepted[position]) for position in pending] + try: + result = self._run_generation( + batch, + model_configs=model_configs, + selected_models=selected_models, + instructions=instructions, + preview_num_records=preview_num_records, + retry=bool(attempt), + ) + except AnonymizerWorkflowError: + if not attempt: + raise + # A wholly failed retry may raise instead of returning an empty + # frame. Complete records from the first attempt still survive. + for position in pending: + failed.append( + FailedRecord( + record_id=str(source[RECORD_ID_COLUMN].iloc[position]), + step="replace-map-generation", + reason="Replacement map retry failed before a complete map was returned", + ) + ) + dropped.update(pending) + pending = [] + break + failed.extend(result.failed_records) + returned: set[int] = set() + for _, row in result.dataframe.iterrows(): + position = int(row[ROW_ORDER_COL]) + returned.add(position) + filtered = _filter_replacement_map_to_input_entities( + raw_map=row.get(COL_REPLACEMENT_MAP), + parsed_entities=expected[position], + record_id=str(source[RECORD_ID_COLUMN].iloc[position]), + ) + additions = [ + entry + for entry in filtered["replacements"] + if (entry["original"], entry["label"]) in missing[position] + ] + accepted[position]["replacements"].extend(additions) + missing[position] -= {(entry["original"], entry["label"]) for entry in additions} + + # Positional association survives duplicate record IDs and reordered + # workflow output. Do not retry rows already dropped by DataDesigner. + reported = Counter(failure.record_id for failure in result.failed_records) + for position in sorted(set(pending) - returned): + dropped.add(position) + record_id = str(source[RECORD_ID_COLUMN].iloc[position]) + if reported[record_id]: + reported[record_id] -= 1 + else: + failed.append( + FailedRecord(record_id, "replace-map-generation", "Record missing from workflow output") + ) + pending = [position for position in pending if position in returned and missing[position]] + if pending and attempt == 0: + logger.warning( + "Substitution maps incomplete in %d record(s); retrying %d missing pair(s) once " + "(missing_by_label=%s)", + len(pending), + sum(len(missing[position]) for position in pending), + dict(sorted(Counter(label for position in pending for _, label in missing[position]).items())), + ) + for position in pending: + labels = dict(sorted(Counter(label for _, label in missing[position]).items())) + failed.append( + FailedRecord( + record_id=str(source[RECORD_ID_COLUMN].iloc[position]), + step="replace-map-generation", + reason=f"Replacement map incomplete after one retry: {len(missing[position])} requested pair(s) " + f"have no mapping (missing_by_label={labels})", + ) + ) + kept = sorted(eligible - dropped - set(pending)) + output = source.iloc[kept].drop(columns=[ROW_ORDER_COL]).copy().reset_index(drop=True) + output[COL_REPLACEMENT_MAP] = pd.Series([accepted[position] for position in kept], dtype=object) + output[COL_REPLACEMENT_MAP_SOURCE] = REPLACEMENT_MAP_SOURCE_LLM + return LlmReplaceResult(dataframe=output, failed_records=failed) + + def generate_map_only( + self, + dataframe: pd.DataFrame, + *, + model_configs: list[ModelConfig], + selected_models: ReplaceModelSelection, + instructions: str | None = None, + entities_column: str = COL_ENTITIES_BY_VALUE, + preview_num_records: int | None = None, + ) -> LlmReplaceResult: working_df = dataframe.copy() # Parse the per-row entity payload once, then reuse it for prompt inputs. @@ -87,21 +210,11 @@ def generate_map_only( effective_preview_num_records = ( min(preview_num_records, len(entity_rows)) if preview_num_records is not None else None ) - run_result = self._adapter.run_workflow( + run_result = self._run_generation( entity_rows, model_configs=model_configs, - columns=[ - LLMStructuredColumnConfig( - name=COL_REPLACEMENT_MAP, - prompt=_get_replacement_mapping_prompt( - entities_column=COL_ENTITIES_FOR_REPLACE, - instructions=instructions, - ), - model_alias=replace_alias, - output_format=EntityReplacementMapSchema, - ) - ], - workflow_name="replace-map-generation", + selected_models=selected_models, + instructions=instructions, preview_num_records=effective_preview_num_records, ) output_df = run_result.dataframe.copy() @@ -122,6 +235,55 @@ def generate_map_only( failed_records=run_result.failed_records, ) + def _run_generation( + self, + dataframe: pd.DataFrame, + *, + model_configs: list[ModelConfig], + selected_models: ReplaceModelSelection, + instructions: str | None, + preview_num_records: int | None, + retry: bool = False, + ) -> WorkflowRunResult: + return self._adapter.run_workflow( + dataframe, + model_configs=model_configs, + columns=[ + LLMStructuredColumnConfig( + name=COL_REPLACEMENT_MAP, + prompt=_get_replacement_mapping_prompt( + entities_column=COL_ENTITIES_FOR_REPLACE, + instructions=instructions, + retry=retry, + ), + model_alias=resolve_model_alias("replacement_generator", selected_models), + output_format=EntityReplacementMapSchema, + ) + ], + workflow_name="replace-map-generation", + preview_num_records=min(preview_num_records, len(dataframe)) if preview_num_records is not None else None, + ) + + +def _entity_pairs(entities: EntitiesByValueSchema) -> set[tuple[str, str]]: + return { + (entity.value, label) + for entity in entities.entities_by_value + for label in entity.labels + if entity.value and label + } + + +def _entities_for_pairs(pairs: set[tuple[str, str]]) -> EntitiesByValueSchema: + grouped: dict[str, list[str]] = {} + for value, label in sorted(pairs): + grouped.setdefault(value, []).append(label) + return EntitiesByValueSchema.model_validate( + { + "entities_by_value": [{"value": value, "labels": labels} for value, labels in grouped.items()], + } + ) + def _enrich_entities_for_template(parsed: EntitiesByValueSchema) -> list[dict[str, str | list[str]]]: """Add ``labels_str`` for Jinja template rendering.""" @@ -237,10 +399,15 @@ def _collision_safe_synthetic(label: str, *, index: int, protected_original_valu index += 1 -def _get_replacement_mapping_prompt(*, entities_column: str, instructions: str | None = None) -> str: +def _get_replacement_mapping_prompt( + *, + entities_column: str, + instructions: str | None = None, + retry: bool = False, +) -> str: instruction_block = f"\nAdditional instructions: {instructions}\n" if instructions else "" prompt = """Generate synthetic replacements for sensitive entities. ONE value per entity, used consistently. -<> +<><> The replacements must: - prevent re-identification @@ -310,6 +477,14 @@ def _get_replacement_mapping_prompt(*, entities_column: str, instructions: str | prompt, { "<>": instruction_block, + "<>": ( + "\nThis is the one retry for missing mappings. Generate mappings only for the entities listed below. " + "Keep them consistent with these accepted mappings; do not change the accepted values:\n" + + _jinja(COL_REPLACEMENT_RETRY_CONTEXT) + + "\n" + if retry + else "" + ), "<>": entities_column, "<>": COL_ENTITY_EXAMPLES, }, diff --git a/src/anonymizer/engine/replace/replace_runner.py b/src/anonymizer/engine/replace/replace_runner.py index a7f42526a..a00bd0582 100644 --- a/src/anonymizer/engine/replace/replace_runner.py +++ b/src/anonymizer/engine/replace/replace_runner.py @@ -111,7 +111,7 @@ def run( elif isinstance(replace_method, Substitute): if self._llm_workflow is None: raise ValueError("Substitute requires an llm_workflow, but none was provided.") - map_result = self._llm_workflow.generate_map_only( + map_result = self._llm_workflow.generate_validated_maps( dataframe, model_configs=model_configs, selected_models=selected_models, diff --git a/tests/engine/ndd/test_substitution_completeness.py b/tests/engine/ndd/test_substitution_completeness.py new file mode 100644 index 000000000..0682f94af --- /dev/null +++ b/tests/engine/ndd/test_substitution_completeness.py @@ -0,0 +1,163 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json +from collections import Counter +from pathlib import Path + +import pandas as pd +import pytest +from data_designer.config.models import ModelConfig +from data_designer.config.run_config import RunConfig +from data_designer.interface.data_designer import DataDesigner + +from anonymizer.config.models import ReplaceModelSelection +from anonymizer.config.replace_strategies import Substitute +from anonymizer.engine.constants import ( + COL_ENTITIES_BY_VALUE, + COL_FINAL_ENTITIES, + COL_REPLACED_TEXT, + COL_REPLACEMENT_MAP, + COL_TAGGED_TEXT, + COL_TEXT, +) +from anonymizer.engine.ndd.adapter import RECORD_ID_COLUMN, NddAdapter +from anonymizer.engine.replace.llm_replace_workflow import LlmReplaceWorkflow +from anonymizer.engine.replace.replace_runner import ReplacementResult, ReplacementWorkflow +from tests.engine.ndd.stub_provider import Script, stub_provider + +_ALICE = {"original": "Alice", "label": "first_name", "synthetic": "Maya"} +_BOB = {"original": "Bob", "label": "first_name", "synthetic": "Noah"} +_CARA = {"original": "Cara", "label": "first_name", "synthetic": "Lena"} + + +def _inputs() -> pd.DataFrame: + return pd.DataFrame( + { + RECORD_ID_COLUMN: ["partial", "complete", "empty"], + COL_TEXT: ["Alice sees Bob.", "Cara.", "No entities."], + COL_TAGGED_TEXT: ["PARTIAL_RECORD Alice sees Bob.", "COMPLETE_RECORD Cara.", "No entities."], + COL_FINAL_ENTITIES: [ + { + "entities": [ + {"value": "Alice", "label": "first_name", "start_position": 0, "end_position": 5}, + {"value": "Bob", "label": "first_name", "start_position": 11, "end_position": 14}, + ] + }, + {"entities": [{"value": "Cara", "label": "first_name", "start_position": 0, "end_position": 4}]}, + {"entities": []}, + ], + COL_ENTITIES_BY_VALUE: [ + {"entities_by_value": [{"value": value, "labels": ["first_name"]} for value in values]} + for values in [["Alice", "Bob"], ["Cara"], []] + ], + } + ) + + +def _prompt_text(prompt: str) -> str: + messages = json.loads(prompt) + return "\n".join( + message["content"] + if isinstance(message["content"], str) + else "\n".join(part.get("text", "") for part in message["content"]) + for message in messages + ) + + +def _requested_inventory(prompt: str) -> str: + return _prompt_text(prompt).split("Entities to replace:", 1)[1].split("Examples:", 1)[0] + + +def _run(tmp_path: Path, script: Script, preview_num_records: int | None) -> ReplacementResult: + with stub_provider(script) as provider: + designer = DataDesigner( + artifact_path=tmp_path / "artifacts", model_providers=[provider], auto_configure_logging=False + ) + designer.set_run_config( + RunConfig(max_conversation_restarts=0, max_conversation_correction_steps=0, otel_metrics_port=None) + ) + return ReplacementWorkflow(llm_workflow=LlmReplaceWorkflow(adapter=NddAdapter(designer))).run( + _inputs(), + replace_method=Substitute(), + model_configs=[ModelConfig(alias="m", model="stub-model", provider="stub", skip_health_check=True)], + selected_models=ReplaceModelSelection(replacement_generator="m"), + preview_num_records=preview_num_records, + ) + + +@pytest.mark.parametrize("preview_num_records", [None, 3], ids=["run", "preview"]) +@pytest.mark.parametrize( + "extra_entry", + [ + pytest.param(None, id="omitted"), + pytest.param({**_BOB, "original": "bob"}, id="changed-original"), + pytest.param({**_BOB, "label": "name"}, id="changed-label"), + ], +) +def test_real_adapter_retries_only_missing_pairs_and_keeps_accepted_values( + tmp_path: Path, preview_num_records: int | None, extra_entry: dict[str, str] | None +) -> None: + partial_prompts: list[str] = [] + complete_requests = 0 + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + nonlocal complete_requests + text = _prompt_text(prompt) + if "COMPLETE_RECORD" in text: + complete_requests += 1 + entries = [_CARA] + else: + partial_prompts.append(prompt) + entries = ( + [_BOB, {**_ALICE, "synthetic": "Changed"}] + if len(partial_prompts) == 2 + else [_ALICE, *([extra_entry] if extra_entry is not None else [])] + ) + return 200, json.dumps({"replacements": entries}), "stop" + + result = _run(tmp_path, reply, preview_num_records) + + assert result.failed_records == [] + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["partial", "complete", "empty"] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Maya sees Noah.", "Lena.", "No entities."] + assert len(partial_prompts) == 2 + assert complete_requests == 1 + assert '"Alice"' in _requested_inventory(partial_prompts[0]) + assert '"Bob"' in _requested_inventory(partial_prompts[0]) + assert '"Alice"' not in _requested_inventory(partial_prompts[1]) + assert '"Bob"' in _requested_inventory(partial_prompts[1]) + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0] == {"replacements": [_ALICE, _BOB]} + + +@pytest.mark.parametrize("preview_num_records", [None, 3], ids=["run", "preview"]) +@pytest.mark.parametrize("invalid_retry", [False, True], ids=["still-missing", "schema-invalid"]) +def test_real_adapter_fails_only_the_record_that_cannot_complete_its_retry( + tmp_path: Path, preview_num_records: int | None, invalid_retry: bool +) -> None: + requests: Counter[str] = Counter() + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + record = "complete" if "COMPLETE_RECORD" in _prompt_text(prompt) else "partial" + requests[record] += 1 + if record == "complete": + entries = [_CARA] + elif requests[record] == 2 and invalid_retry: + entries = [{"original": "Bob", "label": "first_name", "synthetic": ""}] + else: + entries = [_ALICE] + return 200, json.dumps({"replacements": entries}), "stop" + + result = _run(tmp_path, reply, preview_num_records) + + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["complete", "empty"] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Lena.", "No entities."] + assert requests == {"partial": 2, "complete": 1} + assert len(result.failed_records) == 1 + failure = result.failed_records[0] + assert failure.record_id == "partial" + assert failure.step == "replace-map-generation" + for value in ["Alice", "Bob", "Maya", "Noah", "Cara", "Lena"]: + assert value not in failure.reason diff --git a/tests/engine/test_replace_runner.py b/tests/engine/test_replace_runner.py index abebad791..8f189682c 100644 --- a/tests/engine/test_replace_runner.py +++ b/tests/engine/test_replace_runner.py @@ -93,7 +93,7 @@ def test_substitute_runner_applies_generated_map( stub_entities: list[dict], ) -> None: llm_workflow = Mock() - llm_workflow.generate_map_only.return_value = LlmReplaceResult( + llm_workflow.generate_validated_maps.return_value = LlmReplaceResult( dataframe=pd.DataFrame( { COL_TEXT: ["Alice works at Acme"], @@ -117,7 +117,7 @@ def test_substitute_runner_applies_generated_map( model_configs=stub_model_configs, selected_models=stub_replace_model_selection, ) - assert llm_workflow.generate_map_only.call_count == 1 + assert llm_workflow.generate_validated_maps.call_count == 1 assert result.dataframe[COL_REPLACED_TEXT].iloc[0] == "Maya works at NovaCorp" assert len(result.failed_records) == 1 assert result.dataframe[COL_REPLACEMENT_APPLICATION].iloc[0] == { @@ -141,7 +141,7 @@ def test_malformed_entities_row_fails_while_others_are_replaced( stub_replace_model_selection: ReplaceModelSelection, stub_entities: list[dict], ) -> None: - def _generate_map_only(dataframe: pd.DataFrame, **_: object) -> LlmReplaceResult: + def _generate_validated_maps(dataframe: pd.DataFrame, **_: object) -> LlmReplaceResult: replacements = [{"original": "Alice", "label": "first_name", "synthetic": "Maya"}] return LlmReplaceResult( dataframe=dataframe.assign(**{COL_REPLACEMENT_MAP: [{"replacements": replacements}] * len(dataframe)}), @@ -149,7 +149,7 @@ def _generate_map_only(dataframe: pd.DataFrame, **_: object) -> LlmReplaceResult ) llm_workflow = Mock() - llm_workflow.generate_map_only.side_effect = _generate_map_only + llm_workflow.generate_validated_maps.side_effect = _generate_validated_maps dataframe = pd.DataFrame( { COL_TEXT: ["Alice works at Acme", "Bob works at Acme"], @@ -500,7 +500,7 @@ def test_runner_does_not_invoke_judges( The judges live behind a separate ``evaluate()`` call. """ llm_workflow = Mock() - llm_workflow.generate_map_only.return_value = LlmReplaceResult( + llm_workflow.generate_validated_maps.return_value = LlmReplaceResult( dataframe=pd.DataFrame( { COL_TEXT: ["Alice works at Acme"], diff --git a/tests/engine/test_substitution_completeness.py b/tests/engine/test_substitution_completeness.py new file mode 100644 index 000000000..fa8a0b211 --- /dev/null +++ b/tests/engine/test_substitution_completeness.py @@ -0,0 +1,378 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import logging +from unittest.mock import Mock + +import pandas as pd +import pytest +from data_designer.config.models import ModelConfig + +from anonymizer.config.models import ReplaceModelSelection +from anonymizer.config.replace_strategies import Substitute +from anonymizer.engine.constants import ( + COL_ENTITIES_BY_VALUE, + COL_ENTITIES_FOR_REPLACE, + COL_FINAL_ENTITIES, + COL_REPLACED_TEXT, + COL_REPLACEMENT_MAP, + COL_TAGGED_TEXT, + COL_TEXT, +) +from anonymizer.engine.ndd.adapter import RECORD_ID_COLUMN, FailedRecord, WorkflowRunResult +from anonymizer.engine.replace.llm_replace_workflow import LlmReplaceWorkflow +from anonymizer.engine.replace.replace_runner import ReplacementWorkflow +from anonymizer.engine.row_partitioning import ROW_ORDER_COL + +_ALICE = {"original": "Alice", "label": "first_name", "synthetic": "Maya"} +_BOB = {"original": "Bob", "label": "first_name", "synthetic": "Noah"} + + +def _input_frame() -> pd.DataFrame: + return pd.DataFrame( + { + RECORD_ID_COLUMN: ["partial", "complete", "empty"], + COL_TEXT: ["Alice sees Bob.", "Bob sees Bob.", "No entities."], + COL_TAGGED_TEXT: ["Alice sees Bob.", "Bob sees Bob.", "No entities."], + COL_FINAL_ENTITIES: [ + { + "entities": [ + {"value": "Alice", "label": "first_name", "start_position": 0, "end_position": 5}, + {"value": "Bob", "label": "first_name", "start_position": 11, "end_position": 14}, + ] + }, + { + "entities": [ + {"value": "Bob", "label": "first_name", "start_position": 0, "end_position": 3}, + {"value": "Bob", "label": "first_name", "start_position": 9, "end_position": 12}, + ] + }, + {"entities": []}, + ], + COL_ENTITIES_BY_VALUE: [ + { + "entities_by_value": [ + {"value": "Alice", "labels": ["first_name"]}, + {"value": "Bob", "labels": ["first_name"]}, + ] + }, + {"entities_by_value": [{"value": "Bob", "labels": ["first_name"]}]}, + {"entities_by_value": []}, + ], + } + ) + + +@pytest.mark.parametrize( + "extra_entry", + [ + pytest.param(None, id="omitted"), + pytest.param(dict(_ALICE), id="duplicate-entry"), + pytest.param({**_BOB, "original": "Unrequested"}, id="unrequested-original"), + pytest.param({**_BOB, "original": "bob"}, id="original-case"), + pytest.param({**_BOB, "label": "name"}, id="wrong-label"), + ], +) +def test_incomplete_maps_retry_missing_pairs_and_preserve_accepted_mappings( + extra_entry: dict[str, str] | None, + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, + caplog: pytest.LogCaptureFixture, +) -> None: + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy(deep=True)) + output = frame.copy() + if len(calls) == 1: + output[COL_REPLACEMENT_MAP] = [ + {"replacements": [_ALICE, *([extra_entry] if extra_entry is not None else [])]} + if record_id == "partial" + else {"replacements": [_BOB]} + for record_id in frame[RECORD_ID_COLUMN] + ] + else: + output[COL_REPLACEMENT_MAP] = [ + {"replacements": [_BOB, {**_ALICE, "synthetic": "Changed"}]} for _ in range(len(output)) + ] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + source = _input_frame() + original = source.copy(deep=True) + with caplog.at_level(logging.WARNING, logger="anonymizer"): + result = ReplacementWorkflow(llm_workflow=LlmReplaceWorkflow(adapter=adapter)).run( + source, + replace_method=Substitute(), + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + + pd.testing.assert_frame_equal(source, original) + assert result.failed_records == [] + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["partial", "complete", "empty"] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Maya sees Noah.", "Noah sees Noah.", "No entities."] + assert len(calls) == 2 + assert calls[0][RECORD_ID_COLUMN].tolist() == ["partial", "complete"] + assert calls[1][RECORD_ID_COLUMN].tolist() == ["partial"] + assert calls[1][COL_ENTITIES_FOR_REPLACE].tolist() == [ + [{"value": "Bob", "labels": ["first_name"], "labels_str": "first_name"}] + ] + for value in ["Alice", "Bob", "Maya", "Noah", "Changed", "Unrequested"]: + assert value not in caplog.text + + +@pytest.mark.parametrize("drop_retry", [False, True], ids=["incomplete-retry", "dropped-retry"]) +def test_second_unsuccessful_attempt_fails_only_the_incomplete_record( + drop_retry: bool, + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, + caplog: pytest.LogCaptureFixture, +) -> None: + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy()) + if len(calls) == 2 and drop_retry: + return WorkflowRunResult( + dataframe=frame.iloc[:0], + failed_records=[ + FailedRecord(record_id="partial", step="replace-map-generation", reason="Model failed") + ], + ) + output = frame.copy() + output[COL_REPLACEMENT_MAP] = [ + {"replacements": []} if record_id == "partial" else {"replacements": [_BOB]} + for record_id in frame[RECORD_ID_COLUMN] + ] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + with caplog.at_level(logging.WARNING, logger="anonymizer"): + result = ReplacementWorkflow(llm_workflow=LlmReplaceWorkflow(adapter=adapter)).run( + _input_frame(), + replace_method=Substitute(), + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + + assert len(calls) == 2 + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["complete", "empty"] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Noah sees Noah.", "No entities."] + assert [failure.record_id for failure in result.failed_records] == ["partial"] + for value in ["Alice", "Bob", "Maya", "Noah"]: + assert value not in result.failed_records[0].reason + assert value not in caplog.text + + +def test_unicode_space_mutation_does_not_count_as_the_requested_original( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + source = _input_frame().iloc[:1].copy() + source[COL_ENTITIES_BY_VALUE] = [{"entities_by_value": [{"value": "Bob\u202fJones", "labels": ["full_name"]}]}] + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy()) + output = frame.copy() + output[COL_REPLACEMENT_MAP] = [ + { + "replacements": [ + { + "original": "Bob\u2009Jones" if len(calls) == 1 else "Bob\u202fJones", + "label": "full_name", + "synthetic": "Noah Smith", + } + ] + } + ] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + result = LlmReplaceWorkflow(adapter=adapter).generate_validated_maps( + source, + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + assert len(calls) == 2 + assert result.failed_records == [] + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0]["replacements"][0]["original"] == "Bob\u202fJones" + + +def test_completeness_compares_against_the_original_pre_call_inventory( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy()) + output = frame.copy() + output[COL_ENTITIES_BY_VALUE] = [{"entities_by_value": [{"value": "Alice", "labels": ["first_name"]}]}] + output[COL_REPLACEMENT_MAP] = [{"replacements": [_ALICE] if len(calls) == 1 else [_BOB]}] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + source = _input_frame().iloc[:1] + result = LlmReplaceWorkflow(adapter=adapter).generate_validated_maps( + source, + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + + assert len(calls) == 2 + assert calls[1][COL_ENTITIES_FOR_REPLACE].iloc[0] == [ + {"value": "Bob", "labels": ["first_name"], "labels_str": "first_name"} + ] + assert result.failed_records == [] + assert result.dataframe[COL_ENTITIES_BY_VALUE].iloc[0] == source[COL_ENTITIES_BY_VALUE].iloc[0] + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0]["replacements"] == [_ALICE, _BOB] + + +def test_missing_label_for_a_shared_value_still_requires_its_own_mapping( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + source = _input_frame().iloc[:1].copy() + source[COL_ENTITIES_BY_VALUE] = [ + {"entities_by_value": [{"value": "Jordan", "labels": ["first_name", "last_name"]}]} + ] + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy()) + output = frame.copy() + output[COL_REPLACEMENT_MAP] = [ + { + "replacements": [ + { + "original": "Jordan", + "label": "first_name" if len(calls) == 1 else "last_name", + "synthetic": "Taylor", + } + ] + } + ] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + result = LlmReplaceWorkflow(adapter=adapter).generate_validated_maps( + source, + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + + assert len(calls) == 2 + assert calls[1][COL_ENTITIES_FOR_REPLACE].iloc[0] == [ + {"value": "Jordan", "labels": ["last_name"], "labels_str": "last_name"} + ] + assert result.failed_records == [] + assert {entry["label"] for entry in result.dataframe[COL_REPLACEMENT_MAP].iloc[0]["replacements"]} == { + "first_name", + "last_name", + } + + +def test_complete_maps_and_no_entity_records_do_not_retry( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + output = frame.copy() + output[COL_REPLACEMENT_MAP] = [{"replacements": [_BOB]} for _ in range(len(output))] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + workflow = LlmReplaceWorkflow(adapter=adapter) + result = workflow.generate_validated_maps( + _input_frame().iloc[1:], + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + assert result.failed_records == [] + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["complete", "empty"] + adapter.run_workflow.assert_called_once() + + adapter.reset_mock() + for source in [_input_frame().iloc[2:], _input_frame().iloc[:0]]: + result = workflow.generate_validated_maps( + source, + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + assert len(result.dataframe) == len(source) + assert result.failed_records == [] + adapter.run_workflow.assert_not_called() + + +def test_retry_keeps_record_positions_when_ids_repeat_and_output_is_reordered( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + source = _input_frame().iloc[[0, 0]].copy() + source[RECORD_ID_COLUMN] = ["duplicate", "duplicate"] + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy()) + output = frame.iloc[::-1].copy() + output[COL_REPLACEMENT_MAP] = [ + { + "replacements": [ + { + "original": "Alice" if len(calls) == 1 else "Bob", + "label": "first_name", + "synthetic": ["Maya", "Lena"][position] if len(calls) == 1 else ["Noah", "Liam"][position], + } + ] + } + for position in output[ROW_ORDER_COL] + ] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + result = ReplacementWorkflow(llm_workflow=LlmReplaceWorkflow(adapter=adapter)).run( + source, + replace_method=Substitute(), + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + assert len(calls) == 2 + assert result.failed_records == [] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Maya sees Noah.", "Lena sees Liam."] + + +def test_preview_retries_only_records_in_the_requested_entity_sample( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy()) + output = frame.iloc[:1].copy() + output[COL_REPLACEMENT_MAP] = [{"replacements": [_ALICE] if len(calls) == 1 else [_BOB]}] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + result = LlmReplaceWorkflow(adapter=adapter).generate_validated_maps( + _input_frame(), + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + preview_num_records=1, + ) + assert len(calls) == 2 + assert all(frame[RECORD_ID_COLUMN].tolist() == ["partial"] for frame in calls) + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["partial", "empty"] + assert result.failed_records == [] From 9671002da078ea085fd94a18191adad9d35b9707 Mon Sep 17 00:00:00 2001 From: memadi Date: Tue, 10 Nov 2026 17:28:25 -0800 Subject: [PATCH 2/4] fix(replace): reconcile Unicode whitespace in original values Recover a copied original only when known space variants identify one requested pair under the same label. Prefer exact entries and restore the trusted original without changing source text, offsets or synthetic values. Ambiguous matches retain the existing retry and failure behavior. Refs #388 Signed-off-by: memadi --- docs/concepts/replace.md | 9 + .../engine/replace/llm_replace_workflow.py | 40 ++- tests/engine/ndd/test_substitution_unicode.py | 157 ++++++++++ .../engine/test_substitution_completeness.py | 4 +- tests/engine/test_substitution_unicode.py | 290 ++++++++++++++++++ 5 files changed, 496 insertions(+), 4 deletions(-) create mode 100644 tests/engine/ndd/test_substitution_unicode.py create mode 100644 tests/engine/test_substitution_unicode.py diff --git a/docs/concepts/replace.md b/docs/concepts/replace.md index cd91b20e5..f354bae10 100644 --- a/docs/concepts/replace.md +++ b/docs/concepts/replace.md @@ -31,6 +31,15 @@ after the retry is excluded from successful output and reported in of the same pair need only one mapping, and records without entities need no replacement-generator call. +Before discarding a mapping, Substitute can reconcile a copied original that +differs only in no-break spaces (`U+00A0`), thin spaces (`U+2009`), narrow no-break +spaces (`U+202F`), or ASCII spaces. The label must match exactly and the comparison +must identify exactly one pair in the original requested inventory. Exact +mappings take precedence. The mapping's original is restored from that inventory; +the source text, span offsets, and synthetic value stay unchanged. This comparison +does not strip or collapse whitespace or change other characters. Ambiguous or +unmatched entries remain missing and follow the retry/failure flow above. + This completeness retry applies to schema-valid responses. DataDesigner's schema-validation retries still apply to each call. If the completeness retry fails validation or its workflow fails, the affected records are reported as diff --git a/src/anonymizer/engine/replace/llm_replace_workflow.py b/src/anonymizer/engine/replace/llm_replace_workflow.py index 4d097a263..c31b2ff5f 100644 --- a/src/anonymizer/engine/replace/llm_replace_workflow.py +++ b/src/anonymizer/engine/replace/llm_replace_workflow.py @@ -41,6 +41,10 @@ # dtypes that would break trace_dataframe.to_parquet round-trip. _INTERNAL_COLUMNS = (COL_ENTITY_EXAMPLES, COL_ENTITIES_FOR_REPLACE, COL_ENTITIES_FOR_REPLACE_JSON) +# Compare known space variants without folding other characters, changing +# whitespace counts, or admitting tabs and line breaks. +_UNICODE_SPACE_TRANSLATION = str.maketrans({"\u00a0": " ", "\u2009": " ", "\u202f": " "}) + @dataclass(frozen=True) class LlmReplaceResult: @@ -129,6 +133,7 @@ def generate_validated_maps( raw_map=row.get(COL_REPLACEMENT_MAP), parsed_entities=expected[position], record_id=str(source[RECORD_ID_COLUMN].iloc[position]), + reconcile_unicode_whitespace=True, ) additions = [ entry @@ -307,8 +312,14 @@ def _filter_replacement_map_to_input_entities( raw_map: object, parsed_entities: EntitiesByValueSchema, record_id: str = "", + reconcile_unicode_whitespace: bool = False, ) -> dict[str, list[dict[str, str]]]: - """Keep only replacement entries that correspond to actual requested entities.""" + """Keep requested entries, optionally reconciling unambiguous Unicode spaces. + + Exact entries take precedence. Reconciliation compares the original under + the same exact label against the full requested inventory, then restores + the trusted original. Ambiguous or otherwise changed originals are dropped. + """ if isinstance(raw_map, BaseModel): raw_map = raw_map.model_dump(mode="python") if not isinstance(raw_map, dict): @@ -328,13 +339,31 @@ def _filter_replacement_map_to_input_entities( if entity.value and label } protected_original_values = {value for value, _ in allowed_pairs} + exact_pairs = {(entry.original, entry.label) for entry in parsed_map.replacements} & allowed_pairs + equivalent_pairs: dict[tuple[str, str], set[tuple[str, str]]] = {} + if reconcile_unicode_whitespace: + for value, label in allowed_pairs: + key = (value.translate(_UNICODE_SPACE_TRANSLATION), label) + equivalent_pairs.setdefault(key, set()).add((value, label)) filtered: list[dict[str, str]] = [] seen: set[tuple[str, str]] = set() synthetic_collision_labels: Counter[str] = Counter() + reconciled_labels: Counter[str] = Counter() for replacement in parsed_map.replacements: key = (replacement.original, replacement.label) - if key not in allowed_pairs or key in seen: + if key not in allowed_pairs: + normalized_key = (replacement.original.translate(_UNICODE_SPACE_TRANSLATION), replacement.label) + candidates = equivalent_pairs.get(normalized_key, set()) + if len(candidates) != 1: + continue + key = next(iter(candidates)) + if key in exact_pairs: + continue + if key in seen: continue + if replacement.original != key[0]: + replacement = replacement.model_copy(update={"original": key[0]}) + reconciled_labels[replacement.label] += 1 if replacement.synthetic in protected_original_values: synthetic_collision_labels[replacement.label] += 1 seen.add(key) @@ -353,6 +382,13 @@ def _filter_replacement_map_to_input_entities( seen.add(key) filtered.append(replacement.model_dump()) + if reconciled_labels: + logger.debug( + "Replacement map reconciled Unicode whitespace for record %s; reconciled=%d (reconciled_by_label=%s)", + record_id or "", + sum(reconciled_labels.values()), + dict(sorted(reconciled_labels.items())), + ) if synthetic_collision_labels: logger.warning( "Replacement map repaired synthetic-original collision entries for record %s; repaired=%d " diff --git a/tests/engine/ndd/test_substitution_unicode.py b/tests/engine/ndd/test_substitution_unicode.py new file mode 100644 index 000000000..698fbe3e1 --- /dev/null +++ b/tests/engine/ndd/test_substitution_unicode.py @@ -0,0 +1,157 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json +from pathlib import Path + +import pandas as pd +import pytest +from data_designer.config.models import ModelConfig +from data_designer.config.run_config import RunConfig +from data_designer.interface.data_designer import DataDesigner + +from anonymizer.config.models import ReplaceModelSelection +from anonymizer.config.replace_strategies import Substitute +from anonymizer.engine.constants import ( + COL_ENTITIES_BY_VALUE, + COL_FINAL_ENTITIES, + COL_REPLACED_TEXT, + COL_REPLACEMENT_APPLICATION, + COL_REPLACEMENT_MAP, + COL_TAGGED_TEXT, + COL_TEXT, +) +from anonymizer.engine.ndd.adapter import RECORD_ID_COLUMN, NddAdapter +from anonymizer.engine.replace.llm_replace_workflow import LlmReplaceWorkflow +from anonymizer.engine.replace.replace_runner import ReplacementResult, ReplacementWorkflow +from tests.engine.ndd.stub_provider import Script, stub_provider + +_NARROW = "Acme\u202fCorp" +_THIN = "Acme\u2009Corp" +_PLAIN = "Acme Corp" + + +def _input(values: list[str]) -> pd.DataFrame: + text = " | ".join(values) + spans = [] + start = 0 + for value in values: + spans.append( + {"value": value, "label": "company_name", "start_position": start, "end_position": start + len(value)} + ) + start += len(value) + len(" | ") + return pd.DataFrame( + { + RECORD_ID_COLUMN: ["unicode-record"], + COL_TEXT: [text], + COL_TAGGED_TEXT: [text], + COL_FINAL_ENTITIES: [{"entities": spans}], + COL_ENTITIES_BY_VALUE: [ + {"entities_by_value": [{"value": value, "labels": ["company_name"]} for value in values]} + ], + } + ) + + +def _run(tmp_path: Path, dataframe: pd.DataFrame, script: Script, preview_num_records: int | None) -> ReplacementResult: + with stub_provider(script) as provider: + designer = DataDesigner( + artifact_path=tmp_path / "artifacts", model_providers=[provider], auto_configure_logging=False + ) + designer.set_run_config( + RunConfig(max_conversation_restarts=0, max_conversation_correction_steps=0, otel_metrics_port=None) + ) + return ReplacementWorkflow(llm_workflow=LlmReplaceWorkflow(adapter=NddAdapter(designer))).run( + dataframe, + replace_method=Substitute(), + model_configs=[ModelConfig(alias="m", model="stub-model", provider="stub", skip_health_check=True)], + selected_models=ReplaceModelSelection(replacement_generator="m"), + preview_num_records=preview_num_records, + ) + + +def _requested_inventory(prompt: str) -> str: + messages = json.loads(prompt) + text = "\n".join( + message["content"] + if isinstance(message["content"], str) + else "\n".join(part.get("text", "") for part in message["content"]) + for message in messages + ) + return text.split("Entities to replace:", 1)[1].split("Examples:", 1)[0] + + +@pytest.mark.parametrize("preview_num_records", [None, 1], ids=["run", "preview"]) +def test_real_adapter_recovers_unicode_original_without_retry_or_source_changes( + tmp_path: Path, preview_num_records: int | None +) -> None: + prompts: list[str] = [] + dataframe = _input([_NARROW]) + original_entities = json.dumps(dataframe[COL_FINAL_ENTITIES].iloc[0]) + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + prompts.append(prompt) + return ( + 200, + json.dumps({"replacements": [{"original": _THIN, "label": "company_name", "synthetic": "Nova Labs"}]}), + "stop", + ) + + result = _run(tmp_path, dataframe, reply, preview_num_records) + + assert result.failed_records == [] + assert len(prompts) == 1 + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Nova Labs"] + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0] == { + "replacements": [{"original": _NARROW, "label": "company_name", "synthetic": "Nova Labs"}] + } + assert result.dataframe[COL_REPLACEMENT_APPLICATION].iloc[0] == { + "targeted_span_count": 1, + "applied_span_count": 1, + "skipped_span_count": 0, + "skipped_span_label_counts": {}, + } + assert dataframe[COL_TEXT].iloc[0] == _NARROW + assert result.dataframe[COL_TEXT].iloc[0] == _NARROW + assert json.dumps(dataframe[COL_FINAL_ENTITIES].iloc[0]) == original_entities + assert result.dataframe[COL_FINAL_ENTITIES].iloc[0] == dataframe[COL_FINAL_ENTITIES].iloc[0] + + +@pytest.mark.parametrize("preview_num_records", [None, 1], ids=["run", "preview"]) +@pytest.mark.parametrize("recover_on_retry", [True, False], ids=["exact-retry", "still-ambiguous"]) +def test_real_adapter_does_not_guess_between_unicode_equivalent_requested_originals( + tmp_path: Path, preview_num_records: int | None, recover_on_retry: bool +) -> None: + prompts: list[str] = [] + dataframe = _input([_NARROW, _PLAIN]) + ambiguous = {"original": _THIN, "label": "company_name", "synthetic": "Unassigned Company"} + exact = [ + {"original": _NARROW, "label": "company_name", "synthetic": "Nova Labs"}, + {"original": _PLAIN, "label": "company_name", "synthetic": "Summit Press"}, + ] + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + prompts.append(prompt) + entries = exact if len(prompts) == 2 and recover_on_retry else [ambiguous] + return 200, json.dumps({"replacements": entries}), "stop" + + result = _run(tmp_path, dataframe, reply, preview_num_records) + + assert len(prompts) == 2 + assert _NARROW in _requested_inventory(prompts[1]) + assert _PLAIN in _requested_inventory(prompts[1]) + assert dataframe[COL_TEXT].iloc[0] == f"{_NARROW} | {_PLAIN}" + if recover_on_retry: + assert result.failed_records == [] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Nova Labs | Summit Press"] + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0] == {"replacements": exact} + else: + assert result.dataframe.empty + assert len(result.failed_records) == 1 + failure = result.failed_records[0] + assert failure.record_id == "unicode-record" + assert failure.step == "replace-map-generation" + for value in [_NARROW, _PLAIN, _THIN, "Unassigned Company"]: + assert value not in failure.reason diff --git a/tests/engine/test_substitution_completeness.py b/tests/engine/test_substitution_completeness.py index fa8a0b211..cfba1549b 100644 --- a/tests/engine/test_substitution_completeness.py +++ b/tests/engine/test_substitution_completeness.py @@ -169,7 +169,7 @@ def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: assert value not in caplog.text -def test_unicode_space_mutation_does_not_count_as_the_requested_original( +def test_unlisted_unicode_space_mutation_does_not_count_as_the_requested_original( stub_model_configs: list[ModelConfig], stub_replace_model_selection: ReplaceModelSelection, ) -> None: @@ -184,7 +184,7 @@ def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: { "replacements": [ { - "original": "Bob\u2009Jones" if len(calls) == 1 else "Bob\u202fJones", + "original": "Bob\u2003Jones" if len(calls) == 1 else "Bob\u202fJones", "label": "full_name", "synthetic": "Noah Smith", } diff --git a/tests/engine/test_substitution_unicode.py b/tests/engine/test_substitution_unicode.py new file mode 100644 index 000000000..4b04e6d4c --- /dev/null +++ b/tests/engine/test_substitution_unicode.py @@ -0,0 +1,290 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import copy +import logging +from unittest.mock import Mock + +import pandas as pd +import pytest +from data_designer.config.models import ModelConfig + +from anonymizer.config.models import ReplaceModelSelection +from anonymizer.config.replace_strategies import Substitute +from anonymizer.engine.constants import ( + COL_ENTITIES_BY_VALUE, + COL_ENTITIES_FOR_REPLACE, + COL_FINAL_ENTITIES, + COL_REPLACED_TEXT, + COL_REPLACEMENT_APPLICATION, + COL_REPLACEMENT_MAP, + COL_TAGGED_TEXT, + COL_TEXT, +) +from anonymizer.engine.ndd.adapter import RECORD_ID_COLUMN, WorkflowRunResult +from anonymizer.engine.replace.llm_replace_workflow import ( + LlmReplaceWorkflow, + _filter_replacement_map_to_input_entities, +) +from anonymizer.engine.replace.replace_runner import ReplacementResult, ReplacementWorkflow +from anonymizer.engine.schemas import EntitiesByValueSchema, EntityReplacementMapSchema + +_LABEL = "company_name" +_NARROW = "Acme\u202fCorp" +_THIN = "Acme\u2009Corp" + + +def _entities(pairs: list[tuple[str, str]]) -> EntitiesByValueSchema: + grouped: dict[str, list[str]] = {} + for value, label in pairs: + grouped.setdefault(value, []).append(label) + return EntitiesByValueSchema.model_validate( + {"entities_by_value": [{"value": value, "labels": labels} for value, labels in grouped.items()]} + ) + + +def _source(pairs: list[tuple[str, str]]) -> pd.DataFrame: + spans = [] + cursor = 0 + for value, label in pairs: + spans.append({"value": value, "label": label, "start_position": cursor, "end_position": cursor + len(value)}) + cursor += len(value) + 3 + text = " | ".join(value for value, _ in pairs) + "." + return pd.DataFrame( + { + RECORD_ID_COLUMN: ["unicode-row"], + COL_TEXT: [text], + COL_TAGGED_TEXT: [text], + COL_FINAL_ENTITIES: [{"entities": spans}], + COL_ENTITIES_BY_VALUE: [_entities(pairs).model_dump()], + } + ) + + +def _run( + source: pd.DataFrame, + responses: list[list[dict[str, str]]], + model_configs: list[ModelConfig], + selected_models: ReplaceModelSelection, +) -> tuple[ReplacementResult, list[pd.DataFrame]]: + calls: list[pd.DataFrame] = [] + + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + calls.append(frame.copy()) + output = frame.copy() + entries = responses[min(len(calls) - 1, len(responses) - 1)] + output[COL_REPLACEMENT_MAP] = [{"replacements": entries} for _ in range(len(output))] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + result = ReplacementWorkflow(llm_workflow=LlmReplaceWorkflow(adapter=adapter)).run( + source, + replace_method=Substitute(), + model_configs=model_configs, + selected_models=selected_models, + ) + return result, calls + + +@pytest.mark.parametrize( + ("requested", "echoed"), + [ + pytest.param(_NARROW, _THIN, id="narrow-to-thin"), + pytest.param("Acme Corp", "Acme\u00a0Corp", id="ascii-to-no-break"), + pytest.param("Acme\u00a0Corp", "Acme Corp", id="no-break-to-ascii"), + ], +) +def test_unicode_space_recovery_preserves_source_spans_and_synthetic_without_retry( + requested: str, + echoed: str, + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, + caplog: pytest.LogCaptureFixture, +) -> None: + source = _source([(requested, _LABEL)]) + snapshot = copy.deepcopy(source.to_dict(orient="list")) + synthetic = "Nova\u202fLabs" + with caplog.at_level(logging.DEBUG, logger="anonymizer"): + result, calls = _run( + source, + [[{"original": echoed, "label": _LABEL, "synthetic": synthetic}]], + stub_model_configs, + stub_replace_model_selection, + ) + + assert source.to_dict(orient="list") == snapshot + assert len(calls) == 1 + assert result.failed_records == [] + row = result.dataframe.iloc[0] + assert row[COL_TEXT] == source[COL_TEXT].iloc[0] + assert row[COL_FINAL_ENTITIES] == snapshot[COL_FINAL_ENTITIES][0] + assert row[COL_REPLACED_TEXT] == synthetic + "." + assert row[COL_REPLACEMENT_MAP] == { + "replacements": [{"original": requested, "label": _LABEL, "synthetic": synthetic}] + } + assert row[COL_REPLACEMENT_APPLICATION]["applied_span_count"] == 1 + assert row[COL_REPLACEMENT_APPLICATION]["skipped_span_count"] == 0 + for value in [requested, echoed, synthetic, "Acme", "Nova"]: + assert value not in caplog.text + + +def test_reconciliation_uses_label_when_other_values_share_the_comparison_key() -> None: + raw = {"replacements": [{"original": "Acme Corp", "label": _LABEL, "synthetic": "Nova Labs"}]} + result = _filter_replacement_map_to_input_entities( + raw_map=raw, + parsed_entities=_entities([(_NARROW, _LABEL), (_THIN, "organization")]), + reconcile_unicode_whitespace=True, + ) + assert result == {"replacements": [{"original": _NARROW, "label": _LABEL, "synthetic": "Nova Labs"}]} + + +@pytest.mark.parametrize("model_response", [False, True], ids=["dict", "pydantic-model"]) +def test_reconciliation_does_not_mutate_raw_response_or_requested_inventory(model_response: bool) -> None: + payload = {"replacements": [{"original": _THIN, "label": _LABEL, "synthetic": "Nova Labs"}]} + raw = EntityReplacementMapSchema.model_validate(payload) if model_response else payload + requested = _entities([(_NARROW, _LABEL)]) + before_payload = copy.deepcopy(payload) + before_requested = requested.model_dump() + result = _filter_replacement_map_to_input_entities( + raw_map=raw, parsed_entities=requested, reconcile_unicode_whitespace=True + ) + assert result["replacements"][0]["original"] == _NARROW + assert payload == before_payload + if isinstance(raw, EntityReplacementMapSchema): + assert raw.model_dump() == before_payload + assert requested.model_dump() == before_requested + + +@pytest.mark.parametrize("exact_first", [False, True], ids=["exact-later", "exact-first"]) +def test_exact_mapping_takes_precedence_over_reconciled_mapping(exact_first: bool) -> None: + exact = {"original": _NARROW, "label": _LABEL, "synthetic": "Exact Labs"} + approximate = {"original": _THIN, "label": _LABEL, "synthetic": "Other Labs"} + result = _filter_replacement_map_to_input_entities( + raw_map={"replacements": [exact, approximate] if exact_first else [approximate, exact]}, + parsed_entities=_entities([(_NARROW, _LABEL)]), + reconcile_unicode_whitespace=True, + ) + assert result == {"replacements": [exact]} + + +def test_exact_match_survives_ambiguous_whitespace_comparison_key() -> None: + entry = {"original": _NARROW, "label": _LABEL, "synthetic": "Nova Labs"} + result = _filter_replacement_map_to_input_entities( + raw_map={"replacements": [entry]}, + parsed_entities=_entities([(_NARROW, _LABEL), (_THIN, _LABEL)]), + reconcile_unicode_whitespace=True, + ) + assert result == {"replacements": [entry]} + + +@pytest.mark.parametrize( + ("echoed", "label"), + [ + pytest.param("acme\u2009Corp", _LABEL, id="case"), + pytest.param("Acme\u2009\u2009Corp", _LABEL, id="space-count"), + pytest.param(" Acme\u2009Corp", _LABEL, id="leading-space"), + pytest.param("Acme\u2009Corp ", _LABEL, id="trailing-space"), + pytest.param("Acme\tCorp", _LABEL, id="tab"), + pytest.param("Acme\nCorp", _LABEL, id="newline"), + pytest.param("Acme\u2003Corp", _LABEL, id="unlisted-em-space"), + pytest.param("Acme\u200bCorp", _LABEL, id="zero-width-space"), + pytest.param("\uff21cme\u2009Corp", _LABEL, id="compatibility-character"), + pytest.param("AcmeCorp", _LABEL, id="deleted-space"), + pytest.param(_THIN, "organization", id="different-label"), + pytest.param(_THIN, "Company_Name", id="label-case"), + ], +) +def test_reconciliation_does_not_accept_differences_outside_allowlisted_space_variants(echoed: str, label: str) -> None: + result = _filter_replacement_map_to_input_entities( + raw_map={"replacements": [{"original": echoed, "label": label, "synthetic": "Nova Labs"}]}, + parsed_entities=_entities([(_NARROW, _LABEL)]), + reconcile_unicode_whitespace=True, + ) + assert result == {"replacements": []} + + +def test_ambiguous_whitespace_mapping_retries_once_then_fails( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + result, calls = _run( + _source([(_NARROW, _LABEL), (_THIN, _LABEL)]), + [[{"original": "Acme Corp", "label": _LABEL, "synthetic": "Nova Labs"}]], + stub_model_configs, + stub_replace_model_selection, + ) + assert len(calls) == 2 + assert all(len(frame[COL_ENTITIES_FOR_REPLACE].iloc[0]) == 2 for frame in calls) + assert result.dataframe.empty + assert [failure.record_id for failure in result.failed_records] == ["unicode-row"] + assert "2 requested pair(s)" in result.failed_records[0].reason + assert "Acme" not in result.failed_records[0].reason + + +def test_retry_reconciliation_checks_full_original_inventory_not_only_missing_pairs( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + result, calls = _run( + _source([(_NARROW, _LABEL), (_THIN, _LABEL)]), + [ + [{"original": _NARROW, "label": _LABEL, "synthetic": "Nova Labs"}], + [{"original": "Acme Corp", "label": _LABEL, "synthetic": "Orion Media"}], + ], + stub_model_configs, + stub_replace_model_selection, + ) + assert len(calls) == 2 + assert calls[1][COL_ENTITIES_FOR_REPLACE].iloc[0] == [{"value": _THIN, "labels": [_LABEL], "labels_str": _LABEL}] + assert result.dataframe.empty + assert len(result.failed_records) == 1 + assert "1 requested pair(s)" in result.failed_records[0].reason + + +def test_whitespace_reconciliation_can_complete_the_retry( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + result, calls = _run( + _source([(_NARROW, _LABEL)]), + [[], [{"original": _THIN, "label": _LABEL, "synthetic": "Nova Labs"}]], + stub_model_configs, + stub_replace_model_selection, + ) + assert len(calls) == 2 + assert result.failed_records == [] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Nova Labs."] + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0]["replacements"][0]["original"] == _NARROW + + +def test_rewrite_map_generation_keeps_existing_exact_match_behavior( + stub_model_configs: list[ModelConfig], + stub_replace_model_selection: ReplaceModelSelection, +) -> None: + def generate(frame: pd.DataFrame, **_: object) -> WorkflowRunResult: + output = frame.copy() + output[COL_REPLACEMENT_MAP] = [ + {"replacements": [{"original": _THIN, "label": _LABEL, "synthetic": "Nova Labs"}]} + ] + return WorkflowRunResult(dataframe=output, failed_records=[]) + + adapter = Mock() + adapter.run_workflow.side_effect = generate + result = LlmReplaceWorkflow(adapter=adapter).generate_map_only( + _source([(_NARROW, _LABEL)]), + model_configs=stub_model_configs, + selected_models=stub_replace_model_selection, + ) + adapter.run_workflow.assert_called_once() + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0] == {"replacements": []} + + +def test_default_filter_does_not_enable_unicode_reconciliation() -> None: + result = _filter_replacement_map_to_input_entities( + raw_map={"replacements": [{"original": _THIN, "label": _LABEL, "synthetic": "Nova Labs"}]}, + parsed_entities=_entities([(_NARROW, _LABEL)]), + ) + assert result == {"replacements": []} From 7ffb9f7a1f1000cc034b02604fbd9f541e13a695 Mon Sep 17 00:00:00 2001 From: memadi Date: Tue, 10 Nov 2026 17:29:29 -0800 Subject: [PATCH 3/4] fix(schema): reject whitespace-only synthetic replacements Require a non-whitespace character in synthetic values in both Pydantic and exported JSON Schema validation. Preserve valid one-character and padded values, including replacement maps generated for rewrite mode. Refs #388 Signed-off-by: memadi --- docs/concepts/replace.md | 7 + src/anonymizer/engine/schemas/replace.py | 6 +- .../ndd/test_substitution_blank_values.py | 207 ++++++++++++++++++ tests/engine/test_substitution_schema.py | 81 +++++++ 4 files changed, 300 insertions(+), 1 deletion(-) create mode 100644 tests/engine/ndd/test_substitution_blank_values.py create mode 100644 tests/engine/test_substitution_schema.py diff --git a/docs/concepts/replace.md b/docs/concepts/replace.md index f354bae10..eb3c4ae2d 100644 --- a/docs/concepts/replace.md +++ b/docs/concepts/replace.md @@ -22,6 +22,13 @@ Replace mode replaces each [detected entity](detection.md) with an alternative t Replaces entities with LLM-generated synthetic values that are contextually plausible. This is the only replacement strategy that requires an LLM call. +Synthetic values must contain at least one non-whitespace character. Schema +validation rejects empty strings and values made entirely of spaces, tabs, +newlines, or Unicode whitespace. Valid one-character substitutes, such as an +initial or a single-digit age, remain allowed. Valid values are preserved as +generated, including any surrounding whitespace. This shared schema rule also +applies to the replacement maps generated for rewrite mode. + After schema validation and filtering, Anonymizer checks that every unique requested `(entity value, label)` pair has a mapping. If mappings are missing, it retries only the missing pairs once, retaining the accepted substitutions. diff --git a/src/anonymizer/engine/schemas/replace.py b/src/anonymizer/engine/schemas/replace.py index 86a890df1..05e69dbbf 100644 --- a/src/anonymizer/engine/schemas/replace.py +++ b/src/anonymizer/engine/schemas/replace.py @@ -9,7 +9,11 @@ class EntityReplacementSchema(BaseModel): original: str = Field(min_length=1, description="The original entity value") label: str = Field(min_length=1, description="The entity label/type") - synthetic: str = Field(min_length=1, description="The synthetic replacement value") + synthetic: str = Field( + min_length=1, + pattern=r"\S", + description="The synthetic replacement value; must contain a non-whitespace character", + ) class EntityReplacementMapSchema(BaseModel): diff --git a/tests/engine/ndd/test_substitution_blank_values.py b/tests/engine/ndd/test_substitution_blank_values.py new file mode 100644 index 000000000..c18499d75 --- /dev/null +++ b/tests/engine/ndd/test_substitution_blank_values.py @@ -0,0 +1,207 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json +from collections import Counter +from pathlib import Path + +import pandas as pd +import pytest +from data_designer.config.models import ModelConfig +from data_designer.config.run_config import RunConfig +from data_designer.interface.data_designer import DataDesigner + +from anonymizer.config.models import ReplaceModelSelection +from anonymizer.config.replace_strategies import Substitute +from anonymizer.engine.constants import ( + COL_ENTITIES_BY_VALUE, + COL_FINAL_ENTITIES, + COL_REPLACED_TEXT, + COL_REPLACEMENT_MAP, + COL_TAGGED_TEXT, + COL_TEXT, +) +from anonymizer.engine.ndd.adapter import RECORD_ID_COLUMN, NddAdapter +from anonymizer.engine.replace.llm_replace_workflow import LlmReplaceResult, LlmReplaceWorkflow +from anonymizer.engine.replace.replace_runner import ReplacementResult, ReplacementWorkflow +from tests.engine.ndd.stub_provider import Script, stub_provider + + +def _inputs() -> pd.DataFrame: + values = ["Alice", "Bob"] + return pd.DataFrame( + { + RECORD_ID_COLUMN: ["bad", "good"], + COL_TEXT: values, + COL_TAGGED_TEXT: ["BAD_RECORD Alice", "GOOD_RECORD Bob"], + COL_FINAL_ENTITIES: [ + { + "entities": [ + { + "value": value, + "label": "first_name", + "start_position": 0, + "end_position": len(value), + } + ] + } + for value in values + ], + COL_ENTITIES_BY_VALUE: [ + {"entities_by_value": [{"value": value, "labels": ["first_name"]}]} for value in values + ], + } + ) + + +def _record(prompt: str) -> str: + return "bad" if "BAD_RECORD" in prompt else "good" + + +def _response(record: str, synthetic: str) -> tuple[int, str, str]: + return ( + 200, + json.dumps( + { + "replacements": [ + { + "original": "Alice" if record == "bad" else "Bob", + "label": "first_name", + "synthetic": synthetic, + } + ] + } + ), + "stop", + ) + + +def _run( + tmp_path: Path, + script: Script, + preview_num_records: int | None, + *, + schema_restarts: int = 0, + map_only: bool = False, +) -> LlmReplaceResult | ReplacementResult: + with stub_provider(script) as provider: + designer = DataDesigner( + artifact_path=tmp_path / "artifacts", model_providers=[provider], auto_configure_logging=False + ) + designer.set_run_config( + RunConfig( + max_conversation_restarts=schema_restarts, + max_conversation_correction_steps=0, + otel_metrics_port=None, + ) + ) + workflow = LlmReplaceWorkflow(adapter=NddAdapter(designer)) + models = [ModelConfig(alias="m", model="stub-model", provider="stub", skip_health_check=True)] + selection = ReplaceModelSelection(replacement_generator="m") + if map_only: + return workflow.generate_map_only( + _inputs(), + model_configs=models, + selected_models=selection, + preview_num_records=preview_num_records, + ) + return ReplacementWorkflow(llm_workflow=workflow).run( + _inputs(), + replace_method=Substitute(), + model_configs=models, + selected_models=selection, + preview_num_records=preview_num_records, + ) + + +@pytest.mark.parametrize("preview_num_records", [None, 2], ids=["run", "preview"]) +@pytest.mark.parametrize("blank", [" ", "\t", "\n", " \t\n\u00a0\u2009\u202f"]) +def test_blank_substitute_is_schema_rejected_while_valid_record_survives( + tmp_path: Path, preview_num_records: int | None, blank: str +) -> None: + requests: Counter[str] = Counter() + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + record = _record(prompt) + requests[record] += 1 + return _response(record, blank if record == "bad" else "N") + + result = _run(tmp_path, reply, preview_num_records) + + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["good"] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["N"] + assert requests == {"bad": 1, "good": 1} + assert len(result.failed_records) == 1 + failure = result.failed_records[0] + assert (failure.record_id, failure.step) == ("bad", "replace-map-generation") + assert "could not be parsed" in failure.reason + for value in ["Alice", "Bob"]: + assert value not in failure.reason + + +@pytest.mark.parametrize("preview_num_records", [None, 2], ids=["run", "preview"]) +def test_blank_substitute_uses_existing_schema_restart_before_completeness_check( + tmp_path: Path, preview_num_records: int | None +) -> None: + requests: Counter[str] = Counter() + bad_prompts: list[str] = [] + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + record = _record(prompt) + requests[record] += 1 + if record == "bad": + bad_prompts.append(prompt) + return _response(record, "\t" if requests[record] == 1 else "Maya") + return _response(record, "N") + + result = _run(tmp_path, reply, preview_num_records, schema_restarts=1) + + assert result.failed_records == [] + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["bad", "good"] + assert result.dataframe[COL_REPLACED_TEXT].tolist() == ["Maya", "N"] + assert requests == {"bad": 2, "good": 1} + assert len(bad_prompts) == 2 and bad_prompts[0] == bad_prompts[1] + + +@pytest.mark.parametrize("preview_num_records", [None, 2], ids=["run", "preview"]) +@pytest.mark.parametrize("synthetic", ["N", " Maya "]) +def test_nonblank_substitute_is_preserved_without_another_call( + tmp_path: Path, preview_num_records: int | None, synthetic: str +) -> None: + requests: Counter[str] = Counter() + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + record = _record(prompt) + requests[record] += 1 + return _response(record, synthetic) + + result = _run(tmp_path, reply, preview_num_records) + + assert result.failed_records == [] + assert requests == {"bad": 1, "good": 1} + assert result.dataframe[COL_REPLACED_TEXT].tolist() == [synthetic, synthetic] + assert [entry["replacements"][0]["synthetic"] for entry in result.dataframe[COL_REPLACEMENT_MAP]] == [ + synthetic, + synthetic, + ] + + +@pytest.mark.parametrize("preview_num_records", [None, 2], ids=["run", "preview"]) +def test_shared_rewrite_map_generation_rejects_blank_without_completeness_retry( + tmp_path: Path, preview_num_records: int | None +) -> None: + requests: Counter[str] = Counter() + + def reply(_model: str, prompt: str, _earlier: int) -> tuple[int, str, str]: + record = _record(prompt) + requests[record] += 1 + return _response(record, "\u202f\u2009" if record == "bad" else "N") + + result = _run(tmp_path, reply, preview_num_records, map_only=True) + + assert result.dataframe[RECORD_ID_COLUMN].tolist() == ["good"] + assert requests == {"bad": 1, "good": 1} + assert [failure.record_id for failure in result.failed_records] == ["bad"] + assert result.dataframe[COL_REPLACEMENT_MAP].iloc[0]["replacements"][0]["synthetic"] == "N" diff --git a/tests/engine/test_substitution_schema.py b/tests/engine/test_substitution_schema.py new file mode 100644 index 000000000..30bc60e68 --- /dev/null +++ b/tests/engine/test_substitution_schema.py @@ -0,0 +1,81 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +import json +from typing import Literal + +import pytest +from data_designer.config.utils.storage_schema import compile_json_storage_schema +from data_designer.engine.models.parsers.errors import ParserException +from data_designer.engine.models.recipes.response_recipes import StructuredResponseRecipe +from pydantic import ValidationError + +from anonymizer.engine.schemas.replace import EntityReplacementMapSchema + +ValidationBackend = Literal["pydantic", "ndd"] + + +def _payload(fields: dict[str, str]) -> dict[str, list[dict[str, str]]]: + return { + "replacements": [ + {"original": "Alice", "label": "first_name", "synthetic": "Maya"}, + {"original": "Bob", "label": "first_name", **fields}, + ] + } + + +def _validate(payload: dict[str, list[dict[str, str]]], backend: ValidationBackend) -> object: + if backend == "pydantic": + return EntityReplacementMapSchema.model_validate(payload).model_dump() + schema, _ = compile_json_storage_schema(EntityReplacementMapSchema.model_json_schema()) + return StructuredResponseRecipe(json_schema=schema).parse(json.dumps(payload)) + + +@pytest.mark.parametrize("backend", ["pydantic", "ndd"]) +@pytest.mark.parametrize( + "synthetic", + [ + pytest.param(" ", id="spaces"), + pytest.param("\t", id="tab"), + pytest.param("\n", id="newline"), + pytest.param("\r\n", id="carriage-return-newline"), + pytest.param("\u00a0", id="no-break-space"), + pytest.param("\u2009", id="thin-space"), + pytest.param("\u202f", id="narrow-no-break-space"), + pytest.param(" \t\r\n\u00a0\u2009\u202f", id="mixed-whitespace"), + ], +) +def test_schema_rejects_whitespace_only_substitutes(synthetic: str, backend: ValidationBackend) -> None: + with pytest.raises((ValidationError, ParserException)): + _validate(_payload({"synthetic": synthetic}), backend) + + +@pytest.mark.parametrize("backend", ["pydantic", "ndd"]) +@pytest.mark.parametrize( + "fields", [pytest.param({}, id="missing-field"), pytest.param({"synthetic": ""}, id="empty-string")] +) +def test_schema_still_rejects_missing_and_empty_substitutes(fields: dict[str, str], backend: ValidationBackend) -> None: + with pytest.raises((ValidationError, ParserException)): + _validate(_payload(fields), backend) + + +@pytest.mark.parametrize("backend", ["pydantic", "ndd"]) +@pytest.mark.parametrize("synthetic", ["N", "7", "林", " Maya ", "\tN\n", "\u202fM\u2009"]) +def test_schema_preserves_nonblank_substitutes(synthetic: str, backend: ValidationBackend) -> None: + payload = _payload({"synthetic": synthetic}) + + assert _validate(payload, backend) == payload + + +def test_schema_identifies_the_bad_entry_in_a_mixed_map() -> None: + with pytest.raises(ValidationError) as exc_info: + EntityReplacementMapSchema.model_validate(_payload({"synthetic": " \t\u202f"})) + + assert [error["loc"] for error in exc_info.value.errors()] == [("replacements", 1, "synthetic")] + + +@pytest.mark.parametrize("backend", ["pydantic", "ndd"]) +def test_schema_does_not_change_mapping_completeness_rules(backend: ValidationBackend) -> None: + assert _validate({"replacements": []}, backend) == {"replacements": []} From 73293a54e2b6c8f131685551b92af228ae290e35 Mon Sep 17 00:00:00 2001 From: memadi Date: Tue, 10 Nov 2026 18:48:59 -0800 Subject: [PATCH 4/4] docs(replace): clarify schema and completeness retries Distinguish existing DataDesigner whole-response schema retries from the single missing-pair retry. Align failure diagnostics, Rewrite validation, per-row call counts and bundled agent guidance with the implemented flow. Update the issue plan to reflect the published branch and PR. Refs #388 Signed-off-by: memadi --- docs/concepts/choosing-a-strategy.md | 7 +++-- docs/concepts/replace.md | 22 ++++++++++---- docs/concepts/rewrite.md | 8 +++-- docs/troubleshooting.md | 32 +++++++++++++++++--- plans/388/substitution-completeness-retry.md | 19 +++++++----- skills/anonymizer/SKILL.md | 5 +-- skills/anonymizer/references/interactive.md | 2 +- 7 files changed, 69 insertions(+), 26 deletions(-) diff --git a/docs/concepts/choosing-a-strategy.md b/docs/concepts/choosing-a-strategy.md index 75b66fa99..3e6a781c7 100644 --- a/docs/concepts/choosing-a-strategy.md +++ b/docs/concepts/choosing-a-strategy.md @@ -140,12 +140,15 @@ Both modes start from the same [detection](detection.md) pipeline. The differenc | Is the goal "scrub the entities and keep everything else"? | ✅ | — | | Is the goal "produce a privacy-safe version of this text that downstream models can train on"? | — | ✅ | | Are there inferable / latent identifiers that aren't explicitly stated (e.g. "during her third round of chemo" → cancer treatment)? | ❌ leaves them | ✅ removes them | -| Additional LLM calls (beyond shared detection) | ~1 (Substitute) or 0 (Redact/Annotate/Hash) | Many (domain → disposition → QA → rewrite → evaluate → repair → judge) | +| Additional LLM calls (beyond shared detection) | Typically 1 per row with entities, plus retries (Substitute); 0 for Redact/Annotate/Hash | Many (domain → disposition → QA → rewrite → evaluate → repair → judge) | | Output text length | ≈ same as input | Often shorter / restructured | | Best for | Structured records, log scrubbing, known-list redaction | Free-text data with implicit identifiers (clinical notes, biographies, depositions, support transcripts) | **Picking between them.** If your data has inferable identifiers that survive entity-only scrubbing (clinical notes, biographies, depositions), Rewrite is the right fit. For structured records, logs, or single-cell PII, Replace is faster and preserves shape. If you're unsure, walk through a few sample rows before deciding. +Substitute's existing whole-response schema retries and its single missing-pair +retry can add calls. See [Substitute validation and retries](replace.md#substitute). + --- ## 4. (Replace) Which strategy @@ -175,7 +178,7 @@ Keep instructions short (one or two sentences). Long instructions compete with t !!! note "Substitute is per-row, not per-dataset" - Within a single row, repeated mentions of the same value get one consistent replacement (entities are grouped by value before the LLM call). **Across rows the LLM has no shared memory** — each row is an independent call, so "Alice" in row 1 and "Alice" in row 47 will likely get different replacements. If you need stable cross-row mappings (e.g. to re-join records by an identifier), use `Hash` instead, or post-process `result.trace_dataframe["_replacement_map"]`. + Within a single row, repeated mentions of the same value get one consistent replacement (entities are grouped by value before the LLM call). **Across rows the LLM has no shared memory** — each row has an independent request context, so "Alice" in row 1 and "Alice" in row 47 will likely get different replacements. If you need stable cross-row mappings (e.g. to re-join records by an identifier), use `Hash` instead, or post-process `result.trace_dataframe["_replacement_map"]`. --- diff --git a/docs/concepts/replace.md b/docs/concepts/replace.md index eb3c4ae2d..4d75e2d9c 100644 --- a/docs/concepts/replace.md +++ b/docs/concepts/replace.md @@ -47,12 +47,22 @@ the source text, span offsets, and synthetic value stay unchanged. This comparis does not strip or collapse whitespace or change other characters. Ambiguous or unmatched entries remain missing and follow the retry/failure flow above. -This completeness retry applies to schema-valid responses. DataDesigner's -schema-validation retries still apply to each call. If the completeness retry -fails validation or its workflow fails, the affected records are reported as -failed while complete records are retained. Completeness checks exact pairs; -it does not guarantee the semantic quality of a generated substitute or -validate span offsets. +This completeness retry applies to schema-valid responses. DataDesigner already +retries schema-invalid responses, validating the record's entire mapping response +on each attempt. Its default is up to five conversation restarts after the initial +attempt, controlled by `max_conversation_restarts` in the `RunConfig` passed as +`Anonymizer(data_designer_run_config=...)`. These schema retries also apply to the +missing-pair generation call. The one completeness retry therefore does not mean +a maximum of two LLM requests. Valid entries from a schema-rejected response are +not retained for a targeted per-entry retry. + +If the completeness retry fails validation after DataDesigner's recovery attempts, +or its workflow fails, the affected records are reported as failed while complete +records are retained. Initial generation retains the existing failure handling: +dropped records appear in `failed_records`, but a failure of the entire initial +workflow can raise `AnonymizerWorkflowError`. Completeness checks exact pairs; +it does not guarantee the semantic quality of a generated substitute or validate +span offsets. ```python from anonymizer import AnonymizerConfig, Substitute diff --git a/docs/concepts/rewrite.md b/docs/concepts/rewrite.md index c06322184..165ee63e0 100644 --- a/docs/concepts/rewrite.md +++ b/docs/concepts/rewrite.md @@ -9,10 +9,14 @@ Instead of replacing individual entities, rewrite mode transforms the entire tex ## How it works -[Detection](detection.md) runs first (same as [Replace mode](replace.md), plus latent entity detection for context-inferable information). This includes identifying signals that may not be explicitly tagged but can be deduced from combinations of details (e.g., location inferred from contextual cues). The text is then classified by domain, and each entity or attribute is assigned a sensitivity disposition based on contextual risk, recognizing that quasi-identifiers can emerge from any aspect of the text. +[Detection](detection.md) runs first (same as [Replace mode](replace.md), plus latent entity detection for context-inferable information). This includes identifying signals that may not be explicitly tagged but can be deduced from combinations of details (e.g., location inferred from contextual cues). The text is then classified by domain, and each entity or attribute is assigned a sensitivity disposition based on contextual risk, recognizing that quasi-identifiers can emerge from any aspect of the text. The text is then rewritten to reduce identifiability, applying targeted transformations that disrupt inference (e.g., weakening or removing linking details) rather than simply rewording content. The rewritten output is evaluated for both quality and privacy leakage using adversarial testing. If thresholds are exceeded, the system automatically refines the rewrite. Any records that failed to meet standards are tagged for human review. +!!! note "Replacement-map validation" + + Rewrite's generated replacement maps share the [Substitute schema rules](replace.md#substitute): synthetic values must contain at least one non-whitespace character, while valid one-character values remain allowed. DataDesigner's existing whole-response schema retries apply to map generation. The single missing-pair retry and Unicode-space reconciliation described for Substitute do not apply to Rewrite; its handling of omitted mappings remains unchanged. + --- ## Key concepts @@ -168,7 +172,7 @@ flagged[["utility_score", "leakage_mass", "any_high_leaked"]].head() - Lower `risk_tolerance` (e.g. `minimal`) to trigger more aggressive repair. - Enable `strict_entity_protection=True` to ensure no detected entity is left unchanged in the output regardless of risk. -**Last resort:** Manually edit or exclude records that resist automated repair — some text is inherently difficult to rewrite without losing utility or leaking identifiers, and requires your judgement as the expert. +**Last resort:** Manually edit or exclude records that resist automated repair — some text is inherently difficult to rewrite without losing utility or leaking identifiers, and requires your judgement as the expert. --- diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 9a6b1f970..ce1d7c96f 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -7,7 +7,7 @@ Symptom-first guide to common problems and how to fix them. Each entry says how For a controlled campaign that establishes quality thresholds before testing concurrency, see [Tune quality before throughput](tuning.md). -When something looks wrong, **first confirm the run completed cleanly** (no `failed_records` — these are rows that didn't make it through the pipeline at all, usually a rate-limit / infra issue). Once you know the pipeline ran, **run `preview` and inspect rows with quality issues** (`needs_human_review=True`) — the trace columns it produces are where to start. +When something looks wrong, **first inspect `failed_records`**. Records can be excluded because of model/provider failures, invalid generated responses, or unresolved Substitute mappings. Use each record's `step` and `reason` to distinguish these cases. Once you know the pipeline ran, **run `preview` and inspect rows with quality issues** (`needs_human_review=True`) — the trace columns it produces are where to start. !!! note "This guide is written against the Python API" @@ -21,7 +21,7 @@ Most fixes start with one of these. ### Did the run actually complete cleanly? -Before debugging quality, confirm the run wasn't degraded by infra problems. Anonymizer logs a WARNING with the number of failed records, and every dropped record appears on `result.failed_records` with its step and reason: +Before debugging quality, inspect workflow and substitution-validation failures. Anonymizer logs a WARNING with the number of failed records, and every reported failed record appears on `result.failed_records` with its step and reason: ```python for fr in result.failed_records: @@ -36,14 +36,16 @@ Common patterns and what they mean: | `timed out` | Requests took longer than the alias's timeout, which is 60 seconds when `inference_parameters.timeout` is not set; large `max_tokens` budgets on reasoning models often need longer | Set `inference_parameters.timeout` for that alias in `models.yaml` (the bundled configs use 300). If the reason says local request admission timed out, the request was never sent: lower `max_parallel_requests` instead. See [Models](concepts/models.md) | | `no error was recorded for this row` | The row never completed and left no error of its own, while other rows failed; `most common:` names their most frequent cause | Investigate the named most-common cause first as a run-level hint; it does not establish why this particular row is missing. Only if rows still go missing, check the shutdown error rate in `data_designer_run_config` | | `does not match the source span` | The GLiNER server returned entity offsets that are inconsistent with the text | Restart the self-hosted GLiNER server from a current Anonymizer release (see the [self-hosting guide](concepts/self-hosting-gliner.md)); servers started from older code can return offsets that do not match the text | +| `Replacement map incomplete after one retry` | Substitute still lacks one or more exact requested mappings after its missing-pair retry | See [Incomplete Substitute mappings](#incomplete-substitute-mappings) below; this is separate from schema validation retries | +| `Replacement map retry failed` | The missing-pair generation workflow failed before returning a complete map | Inspect the workflow logs for schema, provider, or request errors; already-complete records are retained | | `Record missing from workflow output` with no other cause, at `step="entity-detection"`, `step="replace-map-generation"` or `step="rewrite-*"` | Often rate limiting (429s) exhausting the retry budget of the model pool | See [Rate limits / dropped rows](#rate-limits-dropped-rows) below | | `Output is missing required tracking column` | Internal; file an issue | None | -If `failed_records` is non-empty, **fix that first.** Strategy/prompt knobs (`risk_tolerance`, `protect`, `gliner_threshold`) won't help: those rows didn't fail because of bad output, they failed because the workflow never produced a usable output for them. +If `failed_records` is non-empty, **investigate those records first**. Provider failures need model/provider configuration changes; incomplete substitution maps need the [Substitute validation and retry checks](concepts/replace.md#substitute). Treat the reported step and reason as the starting point rather than assuming every failure is an infrastructure problem. ### Rate limits / dropped rows -If you have rows failing with `Record missing from workflow output`, the chain is: +For rate-limited rows failing at `step="entity-detection"`, a common chain is: GLiNER candidate set → chunked into validator calls → each call dispatched to a model in the validator pool → `ThrottledModelClient` does AIMD on 429s → if every alias in the pool fails on at least one chunk, the row drops. Fix in this order: @@ -211,7 +213,27 @@ Symptom: every record uses all `max_repair_iterations` and still ends up flagged ### Output rows are missing (`FailedRecord`s) -See [Did the run actually complete cleanly?](#did-the-run-actually-complete-cleanly) and [Rate limits / dropped rows](#rate-limits-dropped-rows). The short version: this is almost always a model/provider issue, not a strategy issue, and the fix is in `models.yaml` / `providers.yaml`. +Start with each failed record's step and reason in [Did the run actually complete cleanly?](#did-the-run-actually-complete-cleanly). Provider errors may require changes to `models.yaml` / `providers.yaml`; incomplete Substitute maps follow the validation and retry behavior below. A workflow-wide error can instead raise `AnonymizerWorkflowError`. + +### Incomplete Substitute mappings + +Substitute compares the schema-valid, filtered map with the unique `(entity value, +label)` pairs supplied before generation. It retains accepted substitutions and +retries only missing pairs once. If the map is still incomplete, the record is +excluded from successful output and reported at `step="replace-map-generation"` +with `Replacement map incomplete after one retry` in its reason. + +Check the missing-label counts in the reason, the requested entities, and the +`replacement_generator` model and `Substitute.instructions`. A copied original +that differs only in a supported Unicode space is reconciled only when it has +one unambiguous match under the same label. Other differences or ambiguous matches +remain missing. See [Substitute](concepts/replace.md#substitute) for the exact rules. + +A response with an absent field, an empty synthetic value, or a whitespace-only +synthetic value fails schema validation instead. DataDesigner applies its existing +whole-response schema retries before the map reaches the completeness check. +Those retries are separate from the single missing-pair retry; this is not a +per-entry recovery mechanism for a schema-rejected response. ### Replacement looks repetitive or unrealistic (Substitute) diff --git a/plans/388/substitution-completeness-retry.md b/plans/388/substitution-completeness-retry.md index 302c3fda8..23bbdac50 100644 --- a/plans/388/substitution-completeness-retry.md +++ b/plans/388/substitution-completeness-retry.md @@ -27,7 +27,7 @@ records survive. This work does not introduce permissive transport schemas or recover invalid entries from a schema-rejected response. Reject whitespace-only synthetic values in the shared replacement schema using -`pattern="\\S"` alongside `min_length=1`. This requires at least one non-whitespace +`pattern=r"\S"` alongside `min_length=1`. This requires at least one non-whitespace character in both Pydantic validation and the JSON Schema supplied to DataDesigner. Keep valid one-character values and preserve surrounding whitespace on valid strings. The shared rule also applies to rewrite replacement maps; their existing @@ -64,13 +64,16 @@ limits, duplicate IDs and output reordering. Verify preview and batch paths through the real adapter with a local scripted provider. Failure summaries and new logs contain counts and labels rather than entity values. Run relevant unit tests and repository checks. No public API or default configuration changes are -planned; the bundled public skill template is unaffected. +introduced. The bundled skill's guidance is updated; its template imports and API +calls remain unchanged. ## Rollout -Create three local commits on `codex/bugfix/388-origin-investigation`, authored -and signed off by memadi: missing-map completeness and one retry, narrow Unicode -whitespace reconciliation, and rejection of whitespace-only synthetic values. -Do not push. Completeness and reconciliation apply only to Substitute; the -nonblank schema rule also applies to rewrite replacement maps. Tests and -documentation make each behavior explicit. +The three implementation commits, authored and signed off by memadi, are on +`memadi/bugfix/388-substitution-coverage` and published in +[PR #396](https://github.com/NVIDIA-NeMo/Anonymizer/pull/396). They cover missing-map +completeness and one retry, narrow Unicode-whitespace reconciliation, and rejection +of whitespace-only synthetic values. Completeness and reconciliation apply only +to Substitute; the nonblank schema rule also applies to rewrite replacement maps. +Tests and documentation distinguish the existing DataDesigner whole-response +schema retries from the new single missing-pair retry. diff --git a/skills/anonymizer/SKILL.md b/skills/anonymizer/SKILL.md index e8eb0df7e..9b8e610c6 100644 --- a/skills/anonymizer/SKILL.md +++ b/skills/anonymizer/SKILL.md @@ -31,10 +31,11 @@ regulatory and business context. # Rules - **Always preview before running the full pipeline.** Preview is cheap; a full run can be expensive and slow. -- **If `result.failed_records` is non-empty after preview, fix that *before* tweaking strategy.** Dropped rows are a model/provider/infra problem (rate limits, auth, etc.), not a config problem. Strategy knobs won't help. See `docs/troubleshooting.md` "Did the run actually complete cleanly?" or the [published troubleshooting guide](https://nvidia-nemo.github.io/Anonymizer/dev/troubleshooting/). +- **If `result.failed_records` is non-empty after preview, inspect each failure's `step` and `reason` before quality tuning.** Failed rows can indicate model/provider/infra issues (rate limits, auth, etc.) or invalid or incomplete substitution maps. Treat them as failed records, not successful anonymization. See `docs/troubleshooting.md` "Did the run actually complete cleanly?" or the [published troubleshooting guide](https://nvidia-nemo.github.io/Anonymizer/dev/troubleshooting/). - **Ask the user which mode.** Briefly describe both: Replace detects entities and replaces each in place (faster, cheaper, keeps shape); Rewrite transforms the full text to also remove inferable identifiers (more expensive, may restructure). Use the data shape as a hint — free-text with implicit identifiers (clinical notes, biographies, depositions) leans Rewrite; structured records / log lines lean Replace — but the user picks. - **For cross-record consistency** (same value → same replacement everywhere), use `Hash`, not `Substitute`. `Substitute` is consistent within a row only. - **In Replace mode, default to `Substitute`** if the user hasn't specified a strategy. It's the most general-purpose choice and matches the bulk of production usage. +- **`Substitute` checks mapping completeness before returning a record.** After schema validation and filtering, every unique requested `(entity value, label)` pair needs a mapping. Missing pairs are retried once while accepted mappings are retained; a record still incomplete or whose completeness retry fails is reported in `failed_records`. Unambiguous supported Unicode-space differences can be reconciled under the same exact label. DataDesigner's whole-response schema retries are separate from this single missing-pair retry. The shared replacement schema rejects whitespace-only synthetic values in both modes, but Rewrite does not use Substitute's completeness retry or Unicode-space reconciliation. - **`Annotate` is for inspection, not production.** Its output keeps the original entity text and is not privacy-safe. Use it during iteration to confirm detection is working, then switch. - **Evaluation is opt-in and runs as a separate step** (Replace and Rewrite modes). After `preview()` / `run()`, call `anonymizer.evaluate(result)` to score the output with LLM-as-judge. **Entity coverage always runs** in both modes — it reports detection recall over the judge's unique candidate values (`entity_coverage` + `missed_entities`). On top of that: Replace `Substitute` adds three quality judges (type fidelity, relational consistency, attribute fidelity); Rewrite adds the holistic privacy/quality/style judge. Detection validity is **opt-in** via `EvaluateConfig(compute_detection_validity=True)` (off by default). Evaluation is diagnostic — it scores quality, it does not change the anonymized output. - **Always set `data_summary` on the input**, even briefly. It is the single cheapest quality lever and it improves both detection and rewrite. @@ -190,7 +191,7 @@ def main() -> None: result.trace_dataframe.to_pickle("preview_trace.pkl") print("Saved: preview.parquet (pd.read_parquet), preview_trace.pkl (pd.read_pickle)") - # Failure-first protocol: dropped rows are infra issues, not strategy issues. + # Inspect failure reasons: infrastructure and substitution validation can both fail rows. if result.failed_records: print(f"\n⚠️ {len(result.failed_records)} record(s) failed:") for fr in result.failed_records[:3]: diff --git a/skills/anonymizer/references/interactive.md b/skills/anonymizer/references/interactive.md index 199835091..b588d7fec 100644 --- a/skills/anonymizer/references/interactive.md +++ b/skills/anonymizer/references/interactive.md @@ -31,7 +31,7 @@ Iterative design with the user. Do not disengage from the loop until the user sa 5. **Build** — Write the Python script to the current directory using the Output Template in `SKILL.md`. Name the file after the dataset (e.g. `anonymize_clinical_notes.py`). 6. **Preview** — Run `python