From d7788da1f46ea3878ee90c519773bd94ed06b650 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 22 Apr 2026 15:58:05 -0500 Subject: [PATCH] Implemented a validation step to confirm that grammatical changes do not change substantive meaning --- src/audita/cli.py | 13 ++ src/audita/config.py | 42 ++++++ src/audita/corrections.py | 2 + src/audita/llm.py | 18 ++- src/audita/passes.py | 5 +- src/audita/pipeline.py | 83 ++++++++++- src/audita/prompts.py | 23 +++ src/audita/protection.py | 3 + src/audita/schemas.py | 46 ++++++ src/audita/semantic_validation.py | 159 ++++++++++++++++++++ tests/test_cli.py | 2 + tests/test_config.py | 27 ++++ tests/test_pipeline.py | 239 +++++++++++++++++++++++++++++- tests/test_prompts.py | 30 +++- tests/test_semantic_validation.py | 165 +++++++++++++++++++++ 15 files changed, 849 insertions(+), 8 deletions(-) create mode 100644 src/audita/semantic_validation.py create mode 100644 tests/test_semantic_validation.py diff --git a/src/audita/cli.py b/src/audita/cli.py index 1f2c608..d0d4e8a 100644 --- a/src/audita/cli.py +++ b/src/audita/cli.py @@ -45,6 +45,17 @@ def _build_parser() -> argparse.ArgumentParser: process.add_argument("--max-retries", type=int, help="maximum Instructor retries for structured response validation") process.add_argument("--glossary-max-llm-passes", type=int, help="maximum total LLM passes for glossary corrections") process.add_argument("--grammar-max-llm-passes", type=int, help="maximum total LLM passes for grammar corrections") + process.add_argument( + "--grammar-validation-enabled", + action=argparse.BooleanOptionalAction, + default=None, + help="enable semantic validation for grammar corrections", + ) + process.add_argument( + "--grammar-validation-confidence-threshold", + type=float, + help="minimum validator confidence required to apply a validated grammar correction", + ) process.add_argument( "--normalize-max-segment-gap", type=float, @@ -81,6 +92,8 @@ def _process(args: argparse.Namespace) -> int: max_retries=args.max_retries, glossary_max_llm_passes=args.glossary_max_llm_passes, grammar_max_llm_passes=args.grammar_max_llm_passes, + grammar_validation_enabled=args.grammar_validation_enabled, + grammar_validation_confidence_threshold=args.grammar_validation_confidence_threshold, normalize_max_segment_gap=args.normalize_max_segment_gap, normalize_ellipsis_gap=args.normalize_ellipsis_gap, normalize_max_segment_duration=args.normalize_max_segment_duration, diff --git a/src/audita/config.py b/src/audita/config.py index ddadedb..7b81dd7 100644 --- a/src/audita/config.py +++ b/src/audita/config.py @@ -16,6 +16,8 @@ DEFAULT_MAX_RETRIES = 3 DEFAULT_WORK_DIR = "/tmp/audita" DEFAULT_GLOSSARY_MAX_LLM_PASSES = 3 DEFAULT_GRAMMAR_MAX_LLM_PASSES = 3 +DEFAULT_GRAMMAR_VALIDATION_ENABLED = True +DEFAULT_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD = 0.80 DEFAULT_NORMALIZE_MAX_SEGMENT_GAP = 5.0 DEFAULT_NORMALIZE_ELLIPSIS_GAP = 3.5 DEFAULT_NORMALIZE_MAX_SEGMENT_DURATION = 60.0 @@ -32,6 +34,8 @@ class ConfigOverrides: max_retries: Optional[int] = None glossary_max_llm_passes: Optional[int] = None grammar_max_llm_passes: Optional[int] = None + grammar_validation_enabled: Optional[bool] = None + grammar_validation_confidence_threshold: Optional[float] = None normalize_max_segment_gap: Optional[float] = None normalize_ellipsis_gap: Optional[float] = None normalize_max_segment_duration: Optional[float] = None @@ -50,6 +54,8 @@ class AuditaConfig: max_retries: int = DEFAULT_MAX_RETRIES glossary_max_llm_passes: int = DEFAULT_GLOSSARY_MAX_LLM_PASSES grammar_max_llm_passes: int = DEFAULT_GRAMMAR_MAX_LLM_PASSES + grammar_validation_enabled: bool = DEFAULT_GRAMMAR_VALIDATION_ENABLED + grammar_validation_confidence_threshold: float = DEFAULT_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD normalize_max_segment_gap: float = DEFAULT_NORMALIZE_MAX_SEGMENT_GAP normalize_ellipsis_gap: float = DEFAULT_NORMALIZE_ELLIPSIS_GAP normalize_max_segment_duration: float = DEFAULT_NORMALIZE_MAX_SEGMENT_DURATION @@ -104,6 +110,18 @@ class AuditaConfig: DEFAULT_GRAMMAR_MAX_LLM_PASSES, "AUDITA_GRAMMAR_MAX_LLM_PASSES", ) + grammar_validation_enabled = _select_bool( + selected.grammar_validation_enabled, + source.get("AUDITA_GRAMMAR_VALIDATION_ENABLED"), + DEFAULT_GRAMMAR_VALIDATION_ENABLED, + "AUDITA_GRAMMAR_VALIDATION_ENABLED", + ) + grammar_validation_confidence_threshold = _select_float( + selected.grammar_validation_confidence_threshold, + source.get("AUDITA_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD"), + DEFAULT_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD, + "AUDITA_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD", + ) normalize_max_segment_gap = _select_float( selected.normalize_max_segment_gap, source.get("AUDITA_NORMALIZE_MAX_SEGMENT_GAP"), @@ -140,6 +158,8 @@ class AuditaConfig: max_retries=max_retries, glossary_max_llm_passes=glossary_max_llm_passes, grammar_max_llm_passes=grammar_max_llm_passes, + grammar_validation_enabled=grammar_validation_enabled, + grammar_validation_confidence_threshold=grammar_validation_confidence_threshold, normalize_max_segment_gap=normalize_max_segment_gap, normalize_ellipsis_gap=normalize_ellipsis_gap, normalize_max_segment_duration=normalize_max_segment_duration, @@ -168,6 +188,10 @@ class AuditaConfig: raise AuditaConfigError("AUDITA_GLOSSARY_MAX_LLM_PASSES must be greater than or equal to one.") if self.grammar_max_llm_passes < 1: raise AuditaConfigError("AUDITA_GRAMMAR_MAX_LLM_PASSES must be greater than or equal to one.") + if not 0.0 <= self.grammar_validation_confidence_threshold <= 1.0: + raise AuditaConfigError( + "AUDITA_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD must be between 0.0 and 1.0." + ) if not math.isfinite(self.normalize_max_segment_gap): raise AuditaConfigError("AUDITA_NORMALIZE_MAX_SEGMENT_GAP must be finite.") if not math.isfinite(self.normalize_ellipsis_gap): @@ -206,6 +230,24 @@ def _select_int(cli_value: Optional[int], env_value: Optional[str], default: int raise AuditaConfigError(f"{name} must be an integer.") from exc +def _select_bool( + cli_value: Optional[bool], + env_value: Optional[str], + default: bool, + name: str, +) -> bool: + if cli_value is not None: + return cli_value + if env_value is None: + return default + normalized = env_value.strip().casefold() + if normalized in ("1", "true", "yes", "on"): + return True + if normalized in ("0", "false", "no", "off"): + return False + raise AuditaConfigError(f"{name} must be a boolean.") + + def _select_float( cli_value: Optional[float], env_value: Optional[str], diff --git a/src/audita/corrections.py b/src/audita/corrections.py index b6d7a0e..364a1f8 100644 --- a/src/audita/corrections.py +++ b/src/audita/corrections.py @@ -16,6 +16,8 @@ class SkippedCorrection: corrected_text: str confidence: float actual_text: Optional[str] = None + validation_confidence: Optional[float] = None + validation_reason: Optional[str] = None def to_dict(self) -> dict: return asdict(self) diff --git a/src/audita/llm.py b/src/audita/llm.py index f08a639..55f6b21 100644 --- a/src/audita/llm.py +++ b/src/audita/llm.py @@ -3,7 +3,7 @@ from typing import List from .config import AuditaConfig from .errors import AuditaLLMError from .prompts import Message -from .schemas import CorrectionSet +from .schemas import CorrectionSet, GrammarValidationSet class InstructorLLMClient: @@ -36,6 +36,22 @@ class InstructorLLMClient: "supports tool calling or structured outputs." ) from exc + def create_grammar_validations(self, messages: List[Message], config: AuditaConfig) -> GrammarValidationSet: + model = _normalize_openrouter_model(config.model) + try: + return self._client.chat.completions.create( + model=model, + messages=messages, + response_model=GrammarValidationSet, + max_retries=config.max_retries, + extra_body={"provider": {"require_parameters": True}}, + ) + except Exception as exc: + raise AuditaLLMError( + "LLM grammar validation request failed. Confirm the configured OpenRouter model " + "supports tool calling or structured outputs." + ) from exc + def _normalize_openrouter_model(model: str) -> str: prefix = "openrouter/" diff --git a/src/audita/passes.py b/src/audita/passes.py index 780916f..3bd830f 100644 --- a/src/audita/passes.py +++ b/src/audita/passes.py @@ -5,13 +5,16 @@ from typing import List, Protocol from .chunking import TranscriptSection from .config import AuditaConfig from .prompts import build_glossary_correction_messages, build_grammar_correction_messages -from .schemas import CorrectionCandidate, CorrectionSet, Glossary +from .schemas import CorrectionCandidate, CorrectionSet, Glossary, GrammarValidationSet class LLMClient(Protocol): def create_corrections(self, messages: List[dict], config: AuditaConfig) -> CorrectionSet: ... + def create_grammar_validations(self, messages: List[dict], config: AuditaConfig) -> GrammarValidationSet: + ... + class CorrectionPass(Protocol): def run( diff --git a/src/audita/pipeline.py b/src/audita/pipeline.py index 5235f2e..4254508 100644 --- a/src/audita/pipeline.py +++ b/src/audita/pipeline.py @@ -12,8 +12,13 @@ from .corrections import CorrectionGuard, ReplacementMode, SkippedCorrection, ap from .errors import AuditaError from .normalization import NormalizationResult, normalize_transcript from .passes import CorrectionPass, GlossaryCorrectionPass, GrammarCorrectionPass, LLMClient +from .prompts import build_grammar_validation_messages from .protection import ProtectedVocabulary -from .schemas import Glossary, SourceTranscriptSegment, TranscriptSegment, parse_transcript_json +from .semantic_validation import ( + filter_with_grammar_validations, + select_grammar_validation_candidates, +) +from .schemas import CorrectionCandidate, Glossary, SourceTranscriptSegment, TranscriptSegment, parse_transcript_json ProgressCallback = Callable[[str], None] @@ -27,6 +32,7 @@ class StageSpec: confidence_threshold: float replacement_mode: ReplacementMode correction_guard: Optional[CorrectionGuard] = None + protected_vocabulary: Optional[ProtectedVocabulary] = None def process_transcript( @@ -75,6 +81,7 @@ def process_transcript( confidence_threshold=config.glossary_confidence_threshold, replacement_mode="replace_all", correction_guard=protected_vocabulary.violation_reason, + protected_vocabulary=protected_vocabulary, ), StageSpec( name="grammar", @@ -83,6 +90,7 @@ def process_transcript( confidence_threshold=config.grammar_confidence_threshold, replacement_mode="require_unique", correction_guard=protected_vocabulary.violation_reason, + protected_vocabulary=protected_vocabulary, ), ] final_skipped: List[Tuple[str, SkippedCorrection]] = [] @@ -110,6 +118,7 @@ def process_transcript( stage_dir, stage_summaries, stage_summary["passes"], + llm_client, progress, ) final_skipped.extend((stage.name, skipped) for skipped in stage_skipped) @@ -146,6 +155,7 @@ def _run_correction_stage( stage_dir: Path, stage_summaries: List[dict], pass_summaries: List[dict], + llm_client: LLMClient, progress: Optional[ProgressCallback], ) -> Tuple[List[TranscriptSegment], List[SkippedCorrection]]: working = list(transcript) @@ -186,9 +196,19 @@ def _run_correction_stage( ) ) - application_result = apply_corrections( + corrections_for_application, validation_skips, validation_summary = _validate_grammar_corrections( working, corrections, + config, + stage, + pass_dir, + llm_client, + ) + final_nonretry_skips.extend(validation_skips) + + application_result = apply_corrections( + working, + corrections_for_application, stage.confidence_threshold, replacement_mode=stage.replacement_mode, correction_guard=stage.correction_guard, @@ -214,6 +234,7 @@ def _run_correction_stage( "ignored_below_threshold_count": len(application_result.ignored_ids), "skipped_count": len(application_result.skipped), "retry_segment_count": len(unresolved_retry_skips), + **validation_summary, } ) _write_run_metadata(run_dir, config, normalization_summary, stage_summaries) @@ -227,6 +248,62 @@ def _run_correction_stage( return working, final_skipped +def _validate_grammar_corrections( + transcript: List[TranscriptSegment], + corrections: List[CorrectionCandidate], + config: AuditaConfig, + stage: StageSpec, + pass_dir: Path, + llm_client: LLMClient, +) -> Tuple[List[CorrectionCandidate], List[SkippedCorrection], dict]: + validation_summary = { + "validation_candidate_count": 0, + "validation_approved_count": 0, + "validation_rejected_count": 0, + "validation_bypassed_count": 0, + } + if stage.name != "grammar": + return corrections, [], validation_summary + if not config.grammar_validation_enabled: + validation_summary["validation_bypassed_count"] = len(corrections) + return corrections, [], validation_summary + if stage.protected_vocabulary is None: + validation_summary["validation_bypassed_count"] = len(corrections) + return corrections, [], validation_summary + + candidates, bypassed_count = select_grammar_validation_candidates( + transcript, + corrections, + stage.confidence_threshold, + stage.replacement_mode, + stage.protected_vocabulary, + ) + validation_summary["validation_candidate_count"] = len(candidates) + validation_summary["validation_bypassed_count"] = bypassed_count + if not candidates: + return corrections, [], validation_summary + + payload = [candidate.to_prompt_payload() for candidate in candidates] + messages = build_grammar_validation_messages(payload) + prompt_path = pass_dir / "validation-prompt-0000.json" + prompt_path.write_text(json.dumps(messages, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + response = llm_client.create_grammar_validations(messages, config) + response_path = pass_dir / "validation-response-0000.json" + response_path.write_text(response.model_dump_json(indent=2) + "\n", encoding="utf-8") + + result = filter_with_grammar_validations( + corrections, + candidates, + response, + config.grammar_validation_confidence_threshold, + ) + validation_summary["validation_approved_count"] = result.approved_count + validation_summary["validation_rejected_count"] = result.rejected_count + validation_summary["validation_bypassed_count"] = result.bypassed_count + return result.corrections, result.skipped, validation_summary + + def _create_run_dir(work_dir: Path) -> Path: work_dir.mkdir(parents=True, exist_ok=True) timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ") @@ -247,6 +324,8 @@ def _write_run_metadata( "max_section_tokens": config.max_section_tokens, "glossary_confidence_threshold": config.glossary_confidence_threshold, "grammar_confidence_threshold": config.grammar_confidence_threshold, + "grammar_validation_enabled": config.grammar_validation_enabled, + "grammar_validation_confidence_threshold": config.grammar_validation_confidence_threshold, "max_retries": config.max_retries, "glossary_max_llm_passes": config.glossary_max_llm_passes, "grammar_max_llm_passes": config.grammar_max_llm_passes, diff --git a/src/audita/prompts.py b/src/audita/prompts.py index 7e949c4..4dfd8cd 100644 --- a/src/audita/prompts.py +++ b/src/audita/prompts.py @@ -109,3 +109,26 @@ def build_grammar_correction_messages( f"Transcript section:\n{section_json}" ) return [{"role": "system", "content": system}, {"role": "user", "content": user}] + + +def build_grammar_validation_messages(validation_payload: List[dict]) -> List[Message]: + payload_json = json.dumps(validation_payload, ensure_ascii=False, indent=2) + system = ( + "You are Audita, a conservative semantic validation assistant. " + "Evaluate whether each proposed grammar correction preserves the same spoken content and meaning. " + "Do not judge whether the correction is more polished; judge only whether it changes meaning." + ) + user = ( + "Review these proposed grammar corrections and decide whether each correction preserves meaning.\n\n" + "Rules:\n" + "- Return one validation decision for every correction_index in the input.\n" + "- Reject corrections that add or remove negation, reverse meaning, introduce antonyms, change names, change quantities, change actions, change who did what, or otherwise substantively alter the speaker's meaning.\n" + "- Reject corrections like changing \"became visible\" to \"became invisible\" because that reverses the meaning.\n" + "- Allow capitalization and punctuation changes when they preserve meaning.\n" + "- Allow spelling and homophone fixes only when the corrected segment preserves the same spoken content.\n" + "- Do not use domain knowledge to second-guess protected glossary terms; glossary-protected corrections are excluded from this validation step.\n" + "- Each returned validation must contain only correction_index, is_meaning_preserving, confidence, and reason.\n" + "- confidence must be between 0.0 and 1.0.\n\n" + f"Corrections to validate:\n{payload_json}" + ) + return [{"role": "system", "content": system}, {"role": "user", "content": user}] diff --git a/src/audita/protection.py b/src/audita/protection.py index 78a6a77..8a28629 100644 --- a/src/audita/protection.py +++ b/src/audita/protection.py @@ -61,6 +61,9 @@ class ProtectedVocabulary: ) return terms + def contains_term(self, text: str) -> bool: + return bool(self._terms(text)) + @dataclass(frozen=True) class _ProtectedTerm: diff --git a/src/audita/schemas.py b/src/audita/schemas.py index 9998be7..e94ba50 100644 --- a/src/audita/schemas.py +++ b/src/audita/schemas.py @@ -171,6 +171,52 @@ class CorrectionSet(BaseModel): corrections: List[CorrectionCandidate] = Field(default_factory=list) +class GrammarValidationDecision(BaseModel): + model_config = ConfigDict(extra="forbid") + + correction_index: int = Field(ge=0) + is_meaning_preserving: bool + confidence: float = Field(ge=0.0, le=1.0) + reason: StrictStr + + @field_validator("correction_index", mode="before") + @classmethod + def require_integer_index(cls, value: Any) -> int: + if isinstance(value, bool) or not isinstance(value, int): + raise ValueError("must be an integer") + return value + + @field_validator("is_meaning_preserving", mode="before") + @classmethod + def require_boolean(cls, value: Any) -> bool: + if not isinstance(value, bool): + raise ValueError("must be a boolean") + return value + + @field_validator("confidence", mode="before") + @classmethod + def require_number(cls, value: Any) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError("must be a JSON number") + number = float(value) + if not math.isfinite(number): + raise ValueError("must be finite") + return number + + @field_validator("reason") + @classmethod + def require_non_empty_reason(cls, value: str) -> str: + if not value.strip(): + raise ValueError("must not be empty") + return value + + +class GrammarValidationSet(BaseModel): + model_config = ConfigDict(extra="forbid") + + validations: List[GrammarValidationDecision] = Field(default_factory=list) + + _TRANSCRIPT_ADAPTER = TypeAdapter(List[TranscriptSegment]) _SOURCE_TRANSCRIPT_ADAPTER = TypeAdapter(List[SourceTranscriptSegment]) diff --git a/src/audita/semantic_validation.py b/src/audita/semantic_validation.py new file mode 100644 index 0000000..9d5273a --- /dev/null +++ b/src/audita/semantic_validation.py @@ -0,0 +1,159 @@ +import string +from dataclasses import dataclass +from typing import Dict, List, Optional, Tuple + +from .corrections import ReplacementMode, SkippedCorrection +from .errors import AuditaLLMError +from .protection import ProtectedVocabulary +from .schemas import CorrectionCandidate, GrammarValidationSet, TranscriptSegment + + +@dataclass(frozen=True) +class GrammarValidationCandidate: + correction_index: int + correction: CorrectionCandidate + original_segment_text: str + corrected_segment_text: str + + def to_prompt_payload(self) -> dict: + return { + "correction_index": self.correction_index, + "id": self.correction.id, + "original_segment_text": self.original_segment_text, + "corrected_segment_text": self.corrected_segment_text, + "original_text": self.correction.original_text, + "corrected_text": self.correction.corrected_text, + } + + +@dataclass(frozen=True) +class GrammarValidationFilterResult: + corrections: List[CorrectionCandidate] + skipped: List[SkippedCorrection] + candidate_count: int + approved_count: int + rejected_count: int + bypassed_count: int + + +def select_grammar_validation_candidates( + transcript: List[TranscriptSegment], + corrections: List[CorrectionCandidate], + confidence_threshold: float, + replacement_mode: ReplacementMode, + protected_vocabulary: ProtectedVocabulary, +) -> Tuple[List[GrammarValidationCandidate], int]: + id_to_segment = {segment.id: segment for segment in transcript} + candidates: List[GrammarValidationCandidate] = [] + bypassed_count = 0 + for index, correction in enumerate(corrections): + if correction.confidence < confidence_threshold: + bypassed_count += 1 + continue + if _contains_protected_term(protected_vocabulary, correction): + bypassed_count += 1 + continue + if is_capitalization_or_punctuation_only(correction.original_text, correction.corrected_text): + bypassed_count += 1 + continue + + segment = id_to_segment.get(correction.id) + corrected_segment_text = _preview_corrected_segment_text(segment, correction, replacement_mode) + if corrected_segment_text is None: + bypassed_count += 1 + continue + + candidates.append( + GrammarValidationCandidate( + correction_index=index, + correction=correction, + original_segment_text=segment.text, + corrected_segment_text=corrected_segment_text, + ) + ) + return candidates, bypassed_count + + +def filter_with_grammar_validations( + corrections: List[CorrectionCandidate], + candidates: List[GrammarValidationCandidate], + validation_set: GrammarValidationSet, + confidence_threshold: float, +) -> GrammarValidationFilterResult: + candidate_indexes = {candidate.correction_index for candidate in candidates} + decisions_by_index = {} + for decision in validation_set.validations: + if decision.correction_index in decisions_by_index: + raise AuditaLLMError("LLM grammar validation response included duplicate correction_index values.") + if decision.correction_index not in candidate_indexes: + raise AuditaLLMError("LLM grammar validation response included an unknown correction_index.") + decisions_by_index[decision.correction_index] = decision + + missing_indexes = sorted(candidate_indexes - set(decisions_by_index)) + if missing_indexes: + raise AuditaLLMError("LLM grammar validation response omitted correction_index values.") + + rejected_by_index: Dict[int, SkippedCorrection] = {} + for candidate in candidates: + decision = decisions_by_index[candidate.correction_index] + if not decision.is_meaning_preserving or decision.confidence < confidence_threshold: + rejected_by_index[candidate.correction_index] = SkippedCorrection( + id=candidate.correction.id, + reason="grammar validation rejected semantic change", + original_text=candidate.correction.original_text, + corrected_text=candidate.correction.corrected_text, + confidence=candidate.correction.confidence, + actual_text=candidate.original_segment_text, + validation_confidence=decision.confidence, + validation_reason=decision.reason, + ) + + filtered_corrections = [ + correction for index, correction in enumerate(corrections) if index not in rejected_by_index + ] + approved_count = len(candidates) - len(rejected_by_index) + return GrammarValidationFilterResult( + corrections=filtered_corrections, + skipped=[rejected_by_index[index] for index in sorted(rejected_by_index)], + candidate_count=len(candidates), + approved_count=approved_count, + rejected_count=len(rejected_by_index), + bypassed_count=len(corrections) - len(candidates), + ) + + +def is_capitalization_or_punctuation_only(original_text: str, corrected_text: str) -> bool: + return _semantic_key(original_text) == _semantic_key(corrected_text) + + +def _contains_protected_term( + protected_vocabulary: ProtectedVocabulary, + correction: CorrectionCandidate, +) -> bool: + return protected_vocabulary.contains_term(correction.original_text) or protected_vocabulary.contains_term( + correction.corrected_text + ) + + +def _preview_corrected_segment_text( + segment: Optional[TranscriptSegment], + correction: CorrectionCandidate, + replacement_mode: ReplacementMode, +) -> Optional[str]: + if segment is None: + return None + if correction.original_text == "" or correction.original_text == correction.corrected_text: + return None + match_count = segment.text.count(correction.original_text) + if match_count == 0: + return None + if replacement_mode == "require_unique" and match_count > 1: + return None + return segment.text.replace(correction.original_text, correction.corrected_text) + + +_PUNCTUATION = set(string.punctuation) | {"—", "–", "…", "“", "”", "‘", "’"} + + +def _semantic_key(text: str) -> str: + return "".join(character.casefold() for character in text if not character.isspace() and character not in _PUNCTUATION) diff --git a/tests/test_cli.py b/tests/test_cli.py index 9fb12d2..8b40fec 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -21,6 +21,8 @@ def test_process_help_includes_glossary_pass_flag(capsys): assert "--grammar-max-llm-passes" in output assert "--glossary-confidence-threshold" in output assert "--grammar-confidence-threshold" in output + assert "--grammar-validation-enabled" in output + assert "--grammar-validation-confidence-threshold" in output assert "--normalize-max-segment-gap" in output assert "--normalize-ellipsis-gap" in output assert "--normalize-max-segment-duration" in output diff --git a/tests/test_config.py b/tests/test_config.py index 511424d..ca24852 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -8,6 +8,8 @@ from audita.config import ( DEFAULT_GLOSSARY_MAX_LLM_PASSES, DEFAULT_GRAMMAR_CONFIDENCE_THRESHOLD, DEFAULT_GRAMMAR_MAX_LLM_PASSES, + DEFAULT_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD, + DEFAULT_GRAMMAR_VALIDATION_ENABLED, DEFAULT_MAX_RETRIES, DEFAULT_MAX_SECTION_TOKENS, DEFAULT_NORMALIZE_ELLIPSIS_GAP, @@ -30,6 +32,10 @@ def test_config_uses_defaults_with_api_key(): assert config.max_retries == DEFAULT_MAX_RETRIES assert config.glossary_max_llm_passes == DEFAULT_GLOSSARY_MAX_LLM_PASSES assert config.grammar_max_llm_passes == DEFAULT_GRAMMAR_MAX_LLM_PASSES + assert config.grammar_validation_enabled == DEFAULT_GRAMMAR_VALIDATION_ENABLED + assert config.grammar_validation_enabled is True + assert config.grammar_validation_confidence_threshold == DEFAULT_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD + assert config.grammar_validation_confidence_threshold == 0.8 assert config.normalize_max_segment_gap == DEFAULT_NORMALIZE_MAX_SEGMENT_GAP assert config.normalize_max_segment_gap == 5.0 assert config.normalize_ellipsis_gap == DEFAULT_NORMALIZE_ELLIPSIS_GAP @@ -50,6 +56,8 @@ def test_config_env_overrides_defaults(): "AUDITA_MAX_RETRIES": "5", "AUDITA_GLOSSARY_MAX_LLM_PASSES": "7", "AUDITA_GRAMMAR_MAX_LLM_PASSES": "4", + "AUDITA_GRAMMAR_VALIDATION_ENABLED": "false", + "AUDITA_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD": "0.91", "AUDITA_NORMALIZE_MAX_SEGMENT_GAP": "4.5", "AUDITA_NORMALIZE_ELLIPSIS_GAP": "1.5", "AUDITA_NORMALIZE_MAX_SEGMENT_DURATION": "45.0", @@ -64,6 +72,8 @@ def test_config_env_overrides_defaults(): assert config.max_retries == 5 assert config.glossary_max_llm_passes == 7 assert config.grammar_max_llm_passes == 4 + assert config.grammar_validation_enabled is False + assert config.grammar_validation_confidence_threshold == 0.91 assert config.normalize_max_segment_gap == 4.5 assert config.normalize_ellipsis_gap == 1.5 assert config.normalize_max_segment_duration == 45.0 @@ -79,6 +89,8 @@ def test_config_cli_overrides_env(): "AUDITA_MAX_RETRIES": "5", "AUDITA_GLOSSARY_MAX_LLM_PASSES": "7", "AUDITA_GRAMMAR_MAX_LLM_PASSES": "6", + "AUDITA_GRAMMAR_VALIDATION_ENABLED": "false", + "AUDITA_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD": "0.91", "AUDITA_NORMALIZE_MAX_SEGMENT_GAP": "4.5", "AUDITA_NORMALIZE_ELLIPSIS_GAP": "1.5", "AUDITA_NORMALIZE_MAX_SEGMENT_DURATION": "45.0", @@ -92,6 +104,8 @@ def test_config_cli_overrides_env(): max_retries=3, glossary_max_llm_passes=2, grammar_max_llm_passes=3, + grammar_validation_enabled=True, + grammar_validation_confidence_threshold=0.75, normalize_max_segment_gap=3.0, normalize_ellipsis_gap=1.0, normalize_max_segment_duration=30.0, @@ -106,6 +120,8 @@ def test_config_cli_overrides_env(): assert config.max_retries == 3 assert config.glossary_max_llm_passes == 2 assert config.grammar_max_llm_passes == 3 + assert config.grammar_validation_enabled is True + assert config.grammar_validation_confidence_threshold == 0.75 assert config.normalize_max_segment_gap == 3.0 assert config.normalize_ellipsis_gap == 1.0 assert config.normalize_max_segment_duration == 30.0 @@ -148,6 +164,17 @@ def test_config_rejects_invalid_stage_thresholds(): AuditaConfig.from_sources( env={"OPENROUTER_API_KEY": "key", "AUDITA_GRAMMAR_CONFIDENCE_THRESHOLD": "-0.1"} ) + with pytest.raises(AuditaConfigError): + AuditaConfig.from_sources( + env={"OPENROUTER_API_KEY": "key", "AUDITA_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD": "1.1"} + ) + + +def test_config_rejects_invalid_grammar_validation_enabled(): + with pytest.raises(AuditaConfigError): + AuditaConfig.from_sources( + env={"OPENROUTER_API_KEY": "key", "AUDITA_GRAMMAR_VALIDATION_ENABLED": "maybe"} + ) def test_legacy_confidence_threshold_env_is_ignored(): diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 6157512..6385fb5 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -1,23 +1,49 @@ import json +import pytest + from audita.config import AuditaConfig +from audita.errors import AuditaError from audita.pipeline import process_transcript -from audita.schemas import CorrectionCandidate, CorrectionSet, parse_glossary_yaml, parse_source_transcript_json +from audita.schemas import ( + CorrectionCandidate, + CorrectionSet, + GrammarValidationDecision, + GrammarValidationSet, + parse_glossary_yaml, + parse_source_transcript_json, +) class FakeLLMClient: - def __init__(self, responses): + def __init__(self, responses, validation_responses=None): self.responses = list(responses) + self.validation_responses = list(validation_responses or []) self.calls = 0 + self.validation_calls = 0 self.messages = [] + self.validation_messages = [] def create_corrections(self, messages, config): self.calls += 1 self.messages.append(messages) return self.responses.pop(0) + def create_grammar_validations(self, messages, config): + self.validation_calls += 1 + self.validation_messages.append(messages) + if not self.validation_responses: + raise AssertionError("Unexpected grammar validation request.") + return self.validation_responses.pop(0) -def _config(tmp_path, glossary_max_llm_passes=3, grammar_max_llm_passes=3): + +def _config( + tmp_path, + glossary_max_llm_passes=3, + grammar_max_llm_passes=3, + grammar_validation_enabled=False, + grammar_validation_confidence_threshold=0.8, +): return AuditaConfig( api_key="key", max_section_tokens=16000, @@ -26,6 +52,8 @@ def _config(tmp_path, glossary_max_llm_passes=3, grammar_max_llm_passes=3): max_retries=3, glossary_max_llm_passes=glossary_max_llm_passes, grammar_max_llm_passes=grammar_max_llm_passes, + grammar_validation_enabled=grammar_validation_enabled, + grammar_validation_confidence_threshold=grammar_validation_confidence_threshold, work_dir=tmp_path / "work", ) @@ -541,6 +569,8 @@ def test_pipeline_writes_stage_metadata_for_unresolved_retries(tmp_path): assert metadata["grammar_max_llm_passes"] == 3 assert metadata["glossary_confidence_threshold"] == 0.8 assert metadata["grammar_confidence_threshold"] == 0.8 + assert metadata["grammar_validation_enabled"] is False + assert metadata["grammar_validation_confidence_threshold"] == 0.8 assert [item["stage"] for item in metadata["stages"]] == ["glossary", "grammar"] assert [item["pass_number"] for item in metadata["stages"][0]["passes"]] == [1, 2] assert metadata["stages"][0]["passes"][0]["retry_segment_count"] == 1 @@ -589,6 +619,209 @@ def test_grammar_stage_runs_after_glossary_and_sees_corrected_text(tmp_path): assert revised[0].text == "I ask Chauntea." +def test_grammar_validation_rejects_semantic_change(tmp_path): + transcript = parse_source_transcript_json( + """ + [ + {"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "He became visible."} + ] + """ + ) + grammar_correction = CorrectionCandidate( + id=1, + original_text="visible", + corrected_text="invisible", + confidence=0.95, + ) + validation = GrammarValidationDecision( + correction_index=0, + is_meaning_preserving=False, + confidence=0.99, + reason="This reverses visible to invisible.", + ) + fake_client = FakeLLMClient( + [ + CorrectionSet(corrections=[]), + CorrectionSet(corrections=[grammar_correction]), + ], + validation_responses=[GrammarValidationSet(validations=[validation])], + ) + + revised = process_transcript( + transcript, + _glossary(), + _config(tmp_path, grammar_validation_enabled=True), + llm_client=fake_client, + ) + + assert revised[0].text == "He became visible." + assert fake_client.validation_calls == 1 + run_dirs = list((tmp_path / "work").iterdir()) + assert len(run_dirs) == 1 + diagnostics = json.loads((run_dirs[0] / "skipped-corrections.json").read_text(encoding="utf-8")) + skipped = diagnostics["skipped_corrections"][0] + assert skipped["stage"] == "grammar" + assert skipped["reason"] == "grammar validation rejected semantic change" + assert skipped["validation_confidence"] == 0.99 + assert skipped["validation_reason"] == "This reverses visible to invisible." + metadata = json.loads((run_dirs[0] / "metadata.json").read_text(encoding="utf-8")) + grammar_pass = metadata["stages"][1]["passes"][0] + assert grammar_pass["validation_candidate_count"] == 1 + assert grammar_pass["validation_approved_count"] == 0 + assert grammar_pass["validation_rejected_count"] == 1 + assert grammar_pass["validation_bypassed_count"] == 0 + + +def test_grammar_validation_accepts_meaning_preserving_fix(tmp_path): + transcript = parse_source_transcript_json( + """ + [ + {"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Keep in bind."} + ] + """ + ) + grammar_correction = CorrectionCandidate( + id=1, + original_text="bind", + corrected_text="mind", + confidence=0.95, + ) + validation = GrammarValidationDecision( + correction_index=0, + is_meaning_preserving=True, + confidence=0.95, + reason="This fixes the phrase keep in mind.", + ) + fake_client = FakeLLMClient( + [ + CorrectionSet(corrections=[]), + CorrectionSet(corrections=[grammar_correction]), + ], + validation_responses=[GrammarValidationSet(validations=[validation])], + ) + + revised = process_transcript( + transcript, + _glossary(), + _config(tmp_path, grammar_validation_enabled=True), + llm_client=fake_client, + ) + + assert revised[0].text == "Keep in mind." + assert fake_client.validation_calls == 1 + assert list((tmp_path / "work").iterdir()) == [] + + +def test_grammar_validation_bypasses_protected_vocabulary_correction(tmp_path): + transcript = parse_source_transcript_json( + """ + [ + {"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "The gestures arrived."} + ] + """ + ) + glossary = parse_glossary_yaml( + """ + glossary: + - name: "Jesters" + category: faction + summary: "The Jesters are a faction." + """ + ) + grammar_correction = CorrectionCandidate( + id=1, + original_text="gestures", + corrected_text="Jesters", + confidence=0.95, + ) + fake_client = FakeLLMClient( + [ + CorrectionSet(corrections=[]), + CorrectionSet(corrections=[grammar_correction]), + ] + ) + + revised = process_transcript( + transcript, + glossary, + _config(tmp_path, grammar_validation_enabled=True), + llm_client=fake_client, + ) + + assert revised[0].text == "The Jesters arrived." + assert fake_client.validation_calls == 0 + assert list((tmp_path / "work").iterdir()) == [] + + +def test_disabled_grammar_validation_preserves_current_behavior(tmp_path): + transcript = parse_source_transcript_json( + """ + [ + {"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "He became visible."} + ] + """ + ) + grammar_correction = CorrectionCandidate( + id=1, + original_text="visible", + corrected_text="invisible", + confidence=0.95, + ) + fake_client = FakeLLMClient( + [ + CorrectionSet(corrections=[]), + CorrectionSet(corrections=[grammar_correction]), + ] + ) + + revised = process_transcript( + transcript, + _glossary(), + _config(tmp_path, grammar_validation_enabled=False), + llm_client=fake_client, + ) + + assert revised[0].text == "He became invisible." + assert fake_client.validation_calls == 0 + assert list((tmp_path / "work").iterdir()) == [] + + +def test_missing_grammar_validation_decision_fails_and_preserves_diagnostics(tmp_path): + transcript = parse_source_transcript_json( + """ + [ + {"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "He became visible."} + ] + """ + ) + grammar_correction = CorrectionCandidate( + id=1, + original_text="visible", + corrected_text="invisible", + confidence=0.95, + ) + fake_client = FakeLLMClient( + [ + CorrectionSet(corrections=[]), + CorrectionSet(corrections=[grammar_correction]), + ], + validation_responses=[GrammarValidationSet(validations=[])], + ) + + with pytest.raises(AuditaError): + process_transcript( + transcript, + _glossary(), + _config(tmp_path, grammar_validation_enabled=True), + llm_client=fake_client, + ) + + run_dirs = list((tmp_path / "work").iterdir()) + assert len(run_dirs) == 1 + assert (run_dirs[0] / "grammar" / "pass-0001" / "validation-prompt-0000.json").exists() + assert (run_dirs[0] / "grammar" / "pass-0001" / "validation-response-0000.json").exists() + + def test_grammar_stage_cannot_reverse_glossary_protected_term(tmp_path): transcript = parse_source_transcript_json( """ diff --git a/tests/test_prompts.py b/tests/test_prompts.py index 96a8646..33aeb13 100644 --- a/tests/test_prompts.py +++ b/tests/test_prompts.py @@ -1,7 +1,11 @@ import json from audita.chunking import chunk_transcript -from audita.prompts import build_glossary_correction_messages, build_grammar_correction_messages +from audita.prompts import ( + build_glossary_correction_messages, + build_grammar_correction_messages, + build_grammar_validation_messages, +) from audita.schemas import parse_glossary_yaml, parse_transcript_json @@ -169,3 +173,27 @@ def test_grammar_prompt_uses_simplified_segment_payload(): assert "speaker" not in prompt_segments[0] assert "start" not in prompt_segments[0] assert "end" not in prompt_segments[0] + + +def test_grammar_validation_prompt_rejects_semantic_changes(): + messages = build_grammar_validation_messages( + [ + { + "correction_index": 0, + "id": 1, + "original_segment_text": "He became visible.", + "corrected_segment_text": "He became invisible.", + "original_text": "visible", + "corrected_text": "invisible", + } + ] + ) + prompt_text = "\n".join(message["content"] for message in messages) + + assert "preserves meaning" in prompt_text + assert "became visible" in prompt_text + assert "became invisible" in prompt_text + assert "reverses the meaning" in prompt_text + assert "spelling and homophone fixes only when" in prompt_text + assert "correction_index" in prompt_text + assert "is_meaning_preserving" in prompt_text diff --git a/tests/test_semantic_validation.py b/tests/test_semantic_validation.py new file mode 100644 index 0000000..0a21f64 --- /dev/null +++ b/tests/test_semantic_validation.py @@ -0,0 +1,165 @@ +import pytest + +from audita.errors import AuditaLLMError +from audita.semantic_validation import select_grammar_validation_candidates +from audita.semantic_validation import filter_with_grammar_validations +from audita.protection import ProtectedVocabulary +from audita.schemas import ( + CorrectionCandidate, + GrammarValidationDecision, + GrammarValidationSet, + parse_glossary_yaml, + parse_transcript_json, +) + + +def _transcript(): + return parse_transcript_json( + """ + [ + {"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "he became visible and then gestures arrived"} + ] + """ + ) + + +def _vocabulary(): + glossary = parse_glossary_yaml( + """ + glossary: + - name: "Jesters" + category: faction + summary: "The Jesters are a faction." + """ + ) + return ProtectedVocabulary.from_glossary(glossary) + + +def test_protected_vocabulary_correction_bypasses_validation(): + correction = CorrectionCandidate( + id=1, + original_text="gestures", + corrected_text="Jesters", + confidence=0.95, + ) + + candidates, bypassed_count = select_grammar_validation_candidates( + _transcript(), + [correction], + confidence_threshold=0.8, + replacement_mode="require_unique", + protected_vocabulary=_vocabulary(), + ) + + assert candidates == [] + assert bypassed_count == 1 + + +def test_capitalization_only_correction_bypasses_validation(): + correction = CorrectionCandidate( + id=1, + original_text="he", + corrected_text="He", + confidence=0.95, + ) + + candidates, bypassed_count = select_grammar_validation_candidates( + _transcript(), + [correction], + confidence_threshold=0.8, + replacement_mode="require_unique", + protected_vocabulary=_vocabulary(), + ) + + assert candidates == [] + assert bypassed_count == 1 + + +def test_punctuation_only_correction_bypasses_validation(): + correction = CorrectionCandidate( + id=1, + original_text="visible", + corrected_text="visible.", + confidence=0.95, + ) + + candidates, bypassed_count = select_grammar_validation_candidates( + _transcript(), + [correction], + confidence_threshold=0.8, + replacement_mode="require_unique", + protected_vocabulary=_vocabulary(), + ) + + assert candidates == [] + assert bypassed_count == 1 + + +def test_meaning_sensitive_substitution_requires_validation(): + correction = CorrectionCandidate( + id=1, + original_text="visible", + corrected_text="invisible", + confidence=0.95, + ) + + candidates, bypassed_count = select_grammar_validation_candidates( + _transcript(), + [correction], + confidence_threshold=0.8, + replacement_mode="require_unique", + protected_vocabulary=_vocabulary(), + ) + + assert len(candidates) == 1 + assert candidates[0].correction_index == 0 + assert candidates[0].original_segment_text == "he became visible and then gestures arrived" + assert candidates[0].corrected_segment_text == "he became invisible and then gestures arrived" + assert bypassed_count == 0 + + +def test_validation_rejects_duplicate_and_unknown_decisions(): + correction = CorrectionCandidate( + id=1, + original_text="visible", + corrected_text="invisible", + confidence=0.95, + ) + candidates, _ = select_grammar_validation_candidates( + _transcript(), + [correction], + confidence_threshold=0.8, + replacement_mode="require_unique", + protected_vocabulary=_vocabulary(), + ) + decision = GrammarValidationDecision( + correction_index=0, + is_meaning_preserving=True, + confidence=0.95, + reason="Preserves meaning.", + ) + + with pytest.raises(AuditaLLMError): + filter_with_grammar_validations( + [correction], + candidates, + GrammarValidationSet(validations=[decision, decision]), + confidence_threshold=0.8, + ) + + with pytest.raises(AuditaLLMError): + filter_with_grammar_validations( + [correction], + candidates, + GrammarValidationSet( + validations=[ + GrammarValidationDecision( + correction_index=99, + is_meaning_preserving=True, + confidence=0.95, + reason="Unknown.", + ) + ] + ), + confidence_threshold=0.8, + )