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

Refactor to request matching/replacement substrings from LLMs, rather than requesting a complete replacement for the full original text

Browse Source
This commit is contained in:
Eric Rakestraw
2026-04-21 12:33:15 -05:00
parent 23532cade1
commit 4296e3576e
5 changed files with 102 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
src/audita/corrections.py
View File

@@ -1,5 +1,5 @@
from dataclasses import asdict, dataclass
from typing import Dict, Iterable, List, Optional, Tuple
from typing import Iterable, List, Optional, Tuple
from .errors import AuditaValidationError
from .schemas import CorrectionCandidate, TranscriptSegment
@@ -32,33 +32,20 @@ def apply_corrections(
if not 0.0 <= confidence_threshold <= 1.0:
raise AuditaValidationError("Confidence threshold must be between 0.0 and 1.0.")
correction_by_id: Dict[int, CorrectionCandidate] = {}
revised = list(transcript)
skipped: List[SkippedCorrection] = []
for correction in corrections:
segment_id = correction.segment_id
if segment_id in correction_by_id:
skipped.append(
_skip(
correction,
"duplicate correction for segment already handled",
)
)
if correction.confidence < confidence_threshold:
continue
reason, actual_text = _target_error(transcript, correction)
reason, actual_text = _target_error(revised, correction)
if reason is not None:
skipped.append(_skip(correction, reason, actual_text=actual_text))
continue
correction_by_id[segment_id] = correction
revised: List[TranscriptSegment] = []
for original_id, segment in enumerate(transcript):
correction = correction_by_id.get(original_id)
if correction is not None and correction.confidence >= confidence_threshold:
revised.append(segment.model_copy(update={"text": correction.corrected_text}))
else:
revised.append(segment)
segment = revised[correction.segment_id]
revised_text = segment.text.replace(correction.original_text, correction.corrected_text, 1)
revised[correction.segment_id] = segment.model_copy(update={"text": revised_text})
indexed_revised = list(enumerate(revised))
indexed_revised.sort(key=lambda item: (item[1].start, item[1].end, item[0]))
@@ -76,8 +63,16 @@ def _target_error(
return "segment_id does not exist in transcript", None
segment = transcript[correction.segment_id]
if correction.original_text != segment.text:
return "original_text does not exactly match segment text", segment.text
if correction.original_text == "":
return "original_text is empty", segment.text
if correction.original_text == correction.corrected_text:
return "original_text and corrected_text are identical", segment.text
match_count = segment.text.count(correction.original_text)
if match_count == 0:
return "original_text does not match any substring in segment text", segment.text
if match_count > 1:
return "original_text appears multiple times in segment text", segment.text
return None, None

6
src/audita/prompts.py
View File

@@ -31,11 +31,13 @@ def build_glossary_correction_messages(section: TranscriptSection, glossary: Glo
"- Inappropriate example: correcting \"Lyra\" to \"Jesters\" should be omitted because those words are not similar in spoken English, even if \"Jesters\" appears in the glossary.\n"
"- Do not replace one clear glossary term, character name, location, or ordinary word with a different glossary term unless it is a plausible mishearing.\n"
"- Assign high confidence only when the correction is supported by glossary evidence, local context, and spoken-word similarity; otherwise omit the correction.\n"
"- Use the exact segment_id and original_text from the input segment.\n"
"- Use the exact segment_id from the input segment.\n"
"- For returned corrections, original_text must be only the exact text span that needs replacement, not the full segment text unless the whole segment is the replacement span.\n"
"- corrected_text must be only the replacement text for that span, not the full corrected segment text unless the whole segment is the replacement span.\n"
"- Each returned correction must contain only segment_id, original_text, corrected_text, and confidence.\n"
"- Do not return corrections where original_text and corrected_text are identical.\n"
"- Do not return speaker, start, or end fields.\n"
"- Return only changed segments; do not return entries for unchanged segments.\n"
"- corrected_text must contain the full corrected text for that segment.\n"
"- confidence must be between 0.0 and 1.0.\n"
"- If no corrections are needed, return an empty corrections list.\n\n"
f"Glossary:\n{glossary_json}\n\n"

96
tests/test_corrections.py
View File

@@ -9,7 +9,7 @@ def _transcript():
return parse_transcript_json(
"""
[
{"speaker": "Eric", "start": 10.0, "end": 11.0, "text": "I ask Chontia."},
{"speaker": "Eric", "start": 10.0, "end": 11.0, "text": "I ask Chontia for help."},
{"speaker": "Mike", "start": 0.0, "end": 1.0, "text": "Then Lyra."}
]
"""
@@ -21,8 +21,8 @@ def test_apply_corrections_uses_threshold_and_sorts_chronologically():
corrections = [
CorrectionCandidate(
segment_id=0,
original_text="I ask Chontia.",
corrected_text="I ask Chauntea.",
original_text="Chontia",
corrected_text="Chauntea",
confidence=0.8,
)
]
@@ -30,7 +30,7 @@ def test_apply_corrections_uses_threshold_and_sorts_chronologically():
result = apply_corrections(transcript, corrections, confidence_threshold=0.8)
assert [segment.speaker for segment in result.transcript] == ["Mike", "Eric"]
assert result.transcript[1].text == "I ask Chauntea."
assert result.transcript[1].text == "I ask Chauntea for help."
assert result.skipped == []
@@ -39,57 +39,55 @@ def test_apply_corrections_ignores_below_threshold():
corrections = [
CorrectionCandidate(
segment_id=0,
original_text="I ask Chontia.",
corrected_text="I ask Chauntea.",
original_text="Chontia",
corrected_text="Chauntea",
confidence=0.79,
)
]
result = apply_corrections(transcript, corrections, confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia."
assert result.transcript[1].text == "I ask Chontia for help."
assert result.skipped == []
def test_apply_corrections_skips_duplicate_targets():
def test_apply_corrections_allows_multiple_distinct_spans_in_one_segment():
transcript = _transcript()
first = CorrectionCandidate(
segment_id=0,
original_text="I ask Chontia.",
corrected_text="I ask Chauntea.",
original_text="Chontia",
corrected_text="Chauntea",
confidence=0.8,
)
second = CorrectionCandidate(
segment_id=0,
original_text="I ask Chontia.",
corrected_text="I ask Something Else.",
original_text="help",
corrected_text="guidance",
confidence=0.9,
)
result = apply_corrections(transcript, [first, second], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chauntea."
assert len(result.skipped) == 1
assert result.skipped[0].segment_id == 0
assert "duplicate" in result.skipped[0].reason
assert result.transcript[1].text == "I ask Chauntea for guidance."
assert result.skipped == []
def test_apply_corrections_skips_mismatched_original_text():
def test_apply_corrections_skips_missing_substring():
transcript = _transcript()
correction = CorrectionCandidate(
segment_id=0,
original_text="Different text.",
corrected_text="I ask Chauntea.",
corrected_text="Chauntea",
confidence=0.8,
)
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia."
assert result.transcript[1].text == "I ask Chontia for help."
assert len(result.skipped) == 1
assert result.skipped[0].segment_id == 0
assert result.skipped[0].actual_text == "I ask Chontia."
assert "original_text" in result.skipped[0].reason
assert result.skipped[0].actual_text == "I ask Chontia for help."
assert "does not match any substring" in result.skipped[0].reason
def test_apply_corrections_skips_missing_segment_id():
@@ -103,12 +101,66 @@ def test_apply_corrections_skips_missing_segment_id():
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert [segment.text for segment in result.transcript] == ["Then Lyra.", "I ask Chontia."]
assert [segment.text for segment in result.transcript] == ["Then Lyra.", "I ask Chontia for help."]
assert len(result.skipped) == 1
assert result.skipped[0].segment_id == 99
assert "does not exist" in result.skipped[0].reason
def test_apply_corrections_skips_no_op():
transcript = _transcript()
correction = CorrectionCandidate(
segment_id=0,
original_text="Chontia",
corrected_text="Chontia",
confidence=0.8,
)
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia for help."
assert len(result.skipped) == 1
assert "identical" in result.skipped[0].reason
def test_apply_corrections_skips_ambiguous_repeated_substring():
transcript = parse_transcript_json(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Bane met Bane."}
]
"""
)
correction = CorrectionCandidate(
segment_id=0,
original_text="Bane",
corrected_text="Bain",
confidence=0.8,
)
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert result.transcript[0].text == "Bane met Bane."
assert len(result.skipped) == 1
assert "multiple times" in result.skipped[0].reason
def test_apply_corrections_skips_empty_original_text():
transcript = _transcript()
correction = CorrectionCandidate(
segment_id=0,
original_text="",
corrected_text="Chauntea",
confidence=0.8,
)
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia for help."
assert len(result.skipped) == 1
assert "empty" in result.skipped[0].reason
def test_apply_corrections_rejects_invalid_threshold():
with pytest.raises(AuditaValidationError):
apply_corrections(_transcript(), [], confidence_threshold=1.1)

8
tests/test_pipeline.py
View File

@@ -49,8 +49,8 @@ def _transcript():
def test_pipeline_processes_with_fake_llm_and_cleans_work_dir(tmp_path):
correction = CorrectionCandidate(
segment_id=0,
original_text="I ask Chontia.",
corrected_text="I ask Chauntea.",
original_text="Chontia",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient([CorrectionSet(corrections=[correction])])
@@ -71,7 +71,7 @@ def test_pipeline_skips_bad_correction_and_preserves_diagnostics(tmp_path):
correction = CorrectionCandidate(
segment_id=0,
original_text="Different text.",
corrected_text="I ask Chauntea.",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient([CorrectionSet(corrections=[correction])])
@@ -93,4 +93,4 @@ def test_pipeline_skips_bad_correction_and_preserves_diagnostics(tmp_path):
assert skipped_path.exists()
diagnostics = json.loads(skipped_path.read_text(encoding="utf-8"))
assert diagnostics["skipped_corrections"][0]["segment_id"] == 0
assert "original_text" in diagnostics["skipped_corrections"][0]["reason"]
assert "does not match any substring" in diagnostics["skipped_corrections"][0]["reason"]

3
tests/test_prompts.py
View File

@@ -34,6 +34,9 @@ def test_prompt_requires_acoustically_plausible_transcription_errors():
assert '"gestures" to "Jesters"' in prompt_text
assert '"Lyra" to "Jesters"' in prompt_text
assert "should be omitted" in prompt_text
assert "exact text span that needs replacement" in prompt_text
assert "replacement text for that span" in prompt_text
assert "Do not return corrections where original_text and corrected_text are identical" in prompt_text
def test_prompt_uses_simplified_segment_payload():