eric/audita
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

More bugfixes in the glossary protection guard module

Browse Source
This commit is contained in:
Eric Rakestraw
2026-04-22 12:17:52 -05:00
parent 15cdf01f58
commit 87c47517a0
9 changed files with 269 additions and 28 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/audita/corrections.py
View File

@@ -60,7 +60,7 @@ def apply_corrections(
segment = revised[position]
revised_text = segment.text.replace(correction.original_text, correction.corrected_text)
if correction_guard is not None:
reason = correction_guard(segment.text, revised_text)
reason = correction_guard(correction.original_text, correction.corrected_text)
if reason is not None:
skipped.append(_skip(correction, reason, actual_text=segment.text))
continue

4
src/audita/prompts.py
View File

@@ -13,7 +13,7 @@ def build_glossary_correction_messages(
glossary: Glossary,
retry_pass: bool = False,
) -> List[Message]:
glossary_json = json.dumps(glossary.model_dump(mode="json"), ensure_ascii=False, indent=2)
glossary_json = json.dumps(glossary.model_dump(mode="json", exclude_none=True), ensure_ascii=False, indent=2)
section_json = json.dumps(section.prompt_payload(), ensure_ascii=False, indent=2)
system = (
@@ -67,7 +67,7 @@ def build_grammar_correction_messages(
glossary: Glossary,
retry_pass: bool = False,
) -> List[Message]:
glossary_json = json.dumps(glossary.model_dump(mode="json"), ensure_ascii=False, indent=2)
glossary_json = json.dumps(glossary.model_dump(mode="json", exclude_none=True), ensure_ascii=False, indent=2)
section_json = json.dumps(section.prompt_payload(), ensure_ascii=False, indent=2)
system = (

71
src/audita/protection.py
View File

@@ -7,39 +7,40 @@ from .schemas import Glossary
@dataclass(frozen=True)
class ProtectedVocabulary:
canonical_by_folded: Dict[str, str]
terms_by_folded: Dict[str, "_ProtectedTermDefinition"]
pattern: Optional[Pattern[str]]
@classmethod
def from_glossary(cls, glossary: Glossary) -> "ProtectedVocabulary":
canonical_by_folded: Dict[str, str] = {}
for entry in glossary.glossary:
_add_term(canonical_by_folded, entry.name)
for alias in entry.aliases:
_add_term(canonical_by_folded, alias)
terms_by_folded: Dict[str, _ProtectedTermDefinition] = {}
for identity, entry in enumerate(glossary.glossary):
entry_terms = [entry.name, *entry.aliases]
for term in entry_terms:
_add_term(terms_by_folded, term, identity)
_add_term(terms_by_folded, f"{term}s", identity)
if entry.plural is not None:
_add_term(terms_by_folded, entry.plural, identity)
terms = list(canonical_by_folded.values())
terms = [definition.canonical for definition in terms_by_folded.values()]
if not terms:
return cls(canonical_by_folded=canonical_by_folded, pattern=None)
return cls(terms_by_folded=terms_by_folded, pattern=None)
alternatives = sorted((re.escape(term) for term in terms), key=len, reverse=True)
pattern = re.compile(r"(?<!\w)(" + "|".join(alternatives) + r")(?!\w)", flags=re.IGNORECASE)
return cls(canonical_by_folded=canonical_by_folded, pattern=pattern)
return cls(terms_by_folded=terms_by_folded, pattern=pattern)
def violation_reason(self, before: str, after: str) -> Optional[str]:
before_terms = self._terms(before)
after_terms = self._terms(after)
if not before_terms:
return None
after_folded = [term.folded for term in after_terms]
for before_term in before_terms:
if before_term.folded not in after_folded:
return "correction changes protected glossary term usage"
if before_terms:
after_identities = [term.identity for term in after_terms]
for before_term in before_terms:
if before_term.identity not in after_identities:
return "correction changes protected glossary term usage"
for after_term in after_terms:
canonical = self.canonical_by_folded[after_term.folded]
if after_term.text != canonical:
if after_term.text != after_term.canonical:
return "correction changes protected glossary term capitalization"
return None
@@ -47,20 +48,42 @@ class ProtectedVocabulary:
def _terms(self, text: str) -> List["_ProtectedTerm"]:
if self.pattern is None:
return []
return [
_ProtectedTerm(text=match.group(0), folded=match.group(0).casefold())
for match in self.pattern.finditer(text)
]
terms = []
for match in self.pattern.finditer(text):
matched_text = match.group(0)
definition = self.terms_by_folded[matched_text.casefold()]
terms.append(
_ProtectedTerm(
text=matched_text,
identity=definition.identity,
canonical=definition.canonical,
)
)
return terms
@dataclass(frozen=True)
class _ProtectedTerm:
text: str
folded: str
identity: int
canonical: str
def _add_term(canonical_by_folded: Dict[str, str], term: str) -> None:
@dataclass(frozen=True)
class _ProtectedTermDefinition:
identity: int
canonical: str
def _add_term(
terms_by_folded: Dict[str, _ProtectedTermDefinition],
term: str,
identity: int,
) -> None:
stripped = term.strip()
if not stripped:
return
canonical_by_folded.setdefault(stripped.casefold(), stripped)
terms_by_folded.setdefault(
stripped.casefold(),
_ProtectedTermDefinition(identity=identity, canonical=stripped),
)

8
src/audita/schemas.py
View File

@@ -99,6 +99,7 @@ class GlossaryEntry(BaseModel):
name: StrictStr
aliases: List[StrictStr] = Field(default_factory=list)
plural: Optional[StrictStr] = None
category: StrictStr
summary: StrictStr
@@ -117,6 +118,13 @@ class GlossaryEntry(BaseModel):
raise ValueError("aliases must not contain empty strings")
return aliases
@field_validator("plural")
@classmethod
def require_non_empty_plural(cls, plural: Optional[str]) -> Optional[str]:
if plural is not None and not plural.strip():
raise ValueError("plural must not be empty")
return plural
class Glossary(BaseModel):
model_config = ConfigDict(extra="forbid")

41
tests/test_corrections.py
View File

@@ -2,7 +2,8 @@ import pytest
from audita.corrections import apply_corrections
from audita.errors import AuditaValidationError
from audita.schemas import CorrectionCandidate, parse_transcript_json
from audita.protection import ProtectedVocabulary
from audita.schemas import CorrectionCandidate, parse_glossary_yaml, parse_transcript_json
def _transcript():
@@ -200,6 +201,44 @@ def test_apply_corrections_skips_when_guard_rejects_replacement():
assert result.skipped[0].actual_text == "Hrank moves."
def test_apply_corrections_guards_only_replacement_span():
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "You have to keep it bind. Svend sees the jesters."}
]
"""
)
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Svend"
category: pc
summary: "Svend is a player character."
- name: "Jesters"
category: faction
summary: "The Jesters are a faction."
"""
)
correction = CorrectionCandidate(
id=1,
original_text="keep it bind",
corrected_text="keep in mind",
confidence=0.8,
)
result = apply_corrections(
transcript,
[correction],
confidence_threshold=0.8,
replacement_mode="require_unique",
correction_guard=ProtectedVocabulary.from_glossary(glossary).violation_reason,
)
assert result.transcript[0].text == "You have to keep in mind. Svend sees the jesters."
assert result.skipped == []
def test_apply_corrections_without_guard_remains_permissive():
transcript = parse_transcript_json(
"""

43
tests/test_pipeline.py
View File

@@ -119,6 +119,49 @@ def test_glossary_stage_can_correct_toward_protected_term(tmp_path):
assert list((tmp_path / "work").iterdir()) == []
def test_glossary_guard_ignores_unrelated_protected_terms_elsewhere_in_segment(tmp_path):
transcript = parse_source_transcript_json(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "The gestures are near lyra."}
]
"""
)
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Jesters"
category: faction
summary: "The Jesters are a faction."
- name: "Lyra"
category: npc
summary: "Lyra is an NPC."
"""
)
glossary_correction = CorrectionCandidate(
id=1,
original_text="gestures",
corrected_text="Jesters",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[glossary_correction]),
CorrectionSet(corrections=[]),
]
)
revised = process_transcript(
transcript,
glossary,
_config(tmp_path),
llm_client=fake_client,
)
assert revised[0].text == "The Jesters are near lyra."
assert list((tmp_path / "work").iterdir()) == []
def test_glossary_stage_cannot_change_away_from_protected_term(tmp_path):
transcript = parse_source_transcript_json(
"""

38
tests/test_prompts.py
View File

@@ -70,6 +70,44 @@ def test_prompt_uses_simplified_segment_payload():
assert "end" not in prompt_segments[0]
def test_prompts_do_not_include_inferred_plurals():
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "The gestures are nearby."}
]
"""
)
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Godfrey"
aliases:
- "Jester"
category: npc
summary: "Godfrey is an NPC."
"""
)
section = chunk_transcript(transcript, max_section_tokens=16000)[0]
glossary_messages = build_glossary_correction_messages(section, glossary)
glossary_json = glossary_messages[1]["content"].split("Glossary:\n", maxsplit=1)[1].split(
"\n\nTranscript section:",
maxsplit=1,
)[0]
grammar_messages = build_grammar_correction_messages(section, glossary)
grammar_json = grammar_messages[1]["content"].split("Protected glossary/context:\n", maxsplit=1)[1].split(
"\n\nTranscript section:",
maxsplit=1,
)[0]
for prompt_glossary in (json.loads(glossary_json), json.loads(grammar_json)):
entry = prompt_glossary["glossary"][0]
assert "plural" not in entry
assert "Godfreys" not in json.dumps(prompt_glossary)
assert "Jesters" not in json.dumps(prompt_glossary)
def test_grammar_prompt_limits_readability_corrections_and_protects_glossary():
transcript = parse_transcript_json(
"""

62
tests/test_protection.py
View File

@@ -14,6 +14,17 @@ def _vocabulary():
- name: "Popov"
category: npc
summary: "Popov is an allied NPC."
- name: "Jesters"
aliases:
- "Jester"
category: faction
summary: "The Jesters are a faction."
- name: "Svend"
category: pc
summary: "Svend is a player character."
- name: "Godfrey"
category: npc
summary: "Godfrey is an NPC."
"""
)
return ProtectedVocabulary.from_glossary(glossary)
@@ -59,6 +70,9 @@ def test_protection_allows_correction_toward_protected_term():
vocabulary = _vocabulary()
assert vocabulary.violation_reason("Pawpaw moves.", "Popov moves.") is None
assert vocabulary.violation_reason("gestures", "Jesters") is None
assert vocabulary.violation_reason("rank", "Hrank") is None
assert vocabulary.violation_reason("spend", "Svend") is None
def test_protection_allows_possessive_correction_toward_protected_term():
@@ -67,6 +81,54 @@ def test_protection_allows_possessive_correction_toward_protected_term():
assert vocabulary.violation_reason("Pawpaw's exhausted.", "Popov's exhausted.") is None
def test_protection_blocks_noncanonical_introduced_protected_term():
vocabulary = _vocabulary()
assert (
vocabulary.violation_reason("gestures", "jesters")
== "correction changes protected glossary term capitalization"
)
assert (
vocabulary.violation_reason("rank", "hrank")
== "correction changes protected glossary term capitalization"
)
assert (
vocabulary.violation_reason("spend", "svend")
== "correction changes protected glossary term capitalization"
)
def test_protection_allows_inferred_name_plural():
vocabulary = _vocabulary()
assert vocabulary.violation_reason("Godfrey's", "Godfreys") is None
def test_protection_allows_inferred_alias_plural():
vocabulary = _vocabulary()
assert vocabulary.violation_reason("gesture", "Jesters") is None
def test_protection_allows_explicit_plural():
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Mox"
plural: "Moxen"
category: faction
summary: "The Mox are a faction."
"""
)
vocabulary = ProtectedVocabulary.from_glossary(glossary)
assert vocabulary.violation_reason("Mox's", "Moxen") is None
assert (
vocabulary.violation_reason("Mox's", "moxen")
== "correction changes protected glossary term capitalization"
)
def test_protection_allows_punctuation_around_protected_term():
vocabulary = _vocabulary()

28
tests/test_validation.py
View File

@@ -177,6 +177,34 @@ def test_valid_glossary_parses():
)
assert glossary.glossary[0].name == "Lyra"
assert glossary.glossary[0].plural is None
def test_glossary_accepts_optional_plural():
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Godfrey"
plural: "Godfreys"
category: npc
summary: "Godfrey is an NPC."
"""
)
assert glossary.glossary[0].plural == "Godfreys"
def test_glossary_rejects_empty_plural():
with pytest.raises(AuditaValidationError):
parse_glossary_yaml(
"""
glossary:
- name: "Godfrey"
plural: ""
category: npc
summary: "Godfrey is an NPC."
"""
)
def test_glossary_rejects_empty_entries():