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

Added support for multiple LLM passes to correct the transcript based upon the provided glossary

Browse Source
This commit is contained in:
Eric Rakestraw
2026-04-21 13:12:08 -05:00
parent 4296e3576e
commit 7e1a3f721a
12 changed files with 306 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
tests/test_cli.py
View File

@@ -10,3 +10,10 @@ def test_cli_help_uses_audita_program_name(capsys):
assert exc.value.code == 0
assert capsys.readouterr().out.startswith("usage: audita ")
def test_process_help_includes_glossary_pass_flag(capsys):
with pytest.raises(SystemExit) as exc:
main(["process", "--help"])
assert exc.value.code == 0
assert "--glossary-max-llm-passes" in capsys.readouterr().out

14
tests/test_config.py
View File

@@ -3,7 +3,7 @@ from pathlib import Path
import pytest
from audita.config import AuditaConfig, ConfigOverrides
from audita.config import DEFAULT_MAX_RETRIES, DEFAULT_MAX_SECTION_TOKENS, DEFAULT_WORK_DIR
from audita.config import DEFAULT_GLOSSARY_MAX_LLM_PASSES, DEFAULT_MAX_RETRIES, DEFAULT_MAX_SECTION_TOKENS, DEFAULT_WORK_DIR
from audita.errors import AuditaConfigError
@@ -12,6 +12,7 @@ def test_config_uses_defaults_with_api_key():
assert config.max_section_tokens == DEFAULT_MAX_SECTION_TOKENS
assert config.max_retries == DEFAULT_MAX_RETRIES
assert config.glossary_max_llm_passes == DEFAULT_GLOSSARY_MAX_LLM_PASSES
assert config.work_dir == Path(DEFAULT_WORK_DIR)
@@ -22,6 +23,7 @@ def test_config_env_overrides_defaults():
"AUDITA_MAX_SECTION_TOKENS": "42",
"AUDITA_CONFIDENCE_THRESHOLD": "0.9",
"AUDITA_MAX_RETRIES": "5",
"AUDITA_GLOSSARY_MAX_LLM_PASSES": "7",
"AUDITA_WORK_DIR": "/tmp/custom-audita",
}
)
@@ -29,6 +31,7 @@ def test_config_env_overrides_defaults():
assert config.max_section_tokens == 42
assert config.confidence_threshold == 0.9
assert config.max_retries == 5
assert config.glossary_max_llm_passes == 7
assert config.work_dir == Path("/tmp/custom-audita")
@@ -38,17 +41,20 @@ def test_config_cli_overrides_env():
"OPENROUTER_API_KEY": "key",
"AUDITA_MAX_SECTION_TOKENS": "42",
"AUDITA_MAX_RETRIES": "5",
"AUDITA_GLOSSARY_MAX_LLM_PASSES": "7",
"AUDITA_WORK_DIR": "/tmp/env-audita",
},
overrides=ConfigOverrides(
max_section_tokens=100,
max_retries=3,
glossary_max_llm_passes=2,
work_dir=Path("/tmp/cli-audita"),
),
)
assert config.max_section_tokens == 100
assert config.max_retries == 3
assert config.glossary_max_llm_passes == 2
assert config.work_dir == Path("/tmp/cli-audita")
@@ -63,3 +69,9 @@ def test_config_rejects_bad_env_int():
env={"OPENROUTER_API_KEY": "key", "AUDITA_MAX_SECTION_TOKENS": "many"}
)
def test_config_rejects_invalid_glossary_pass_count():
with pytest.raises(AuditaConfigError):
AuditaConfig.from_sources(
env={"OPENROUTER_API_KEY": "key", "AUDITA_GLOSSARY_MAX_LLM_PASSES": "0"}
)

18
tests/test_corrections.py
View File

@@ -16,7 +16,7 @@ def _transcript():
)
def test_apply_corrections_uses_threshold_and_sorts_chronologically():
def test_apply_corrections_uses_threshold_and_preserves_segment_id_order():
transcript = _transcript()
corrections = [
CorrectionCandidate(
@@ -29,8 +29,8 @@ 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 for help."
assert [segment.speaker for segment in result.transcript] == ["Eric", "Mike"]
assert result.transcript[0].text == "I ask Chauntea for help."
assert result.skipped == []
@@ -47,7 +47,7 @@ def test_apply_corrections_ignores_below_threshold():
result = apply_corrections(transcript, corrections, confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia for help."
assert result.transcript[0].text == "I ask Chontia for help."
assert result.skipped == []
@@ -68,7 +68,7 @@ def test_apply_corrections_allows_multiple_distinct_spans_in_one_segment():
result = apply_corrections(transcript, [first, second], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chauntea for guidance."
assert result.transcript[0].text == "I ask Chauntea for guidance."
assert result.skipped == []
@@ -83,7 +83,7 @@ def test_apply_corrections_skips_missing_substring():
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia for help."
assert result.transcript[0].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 for help."
@@ -101,7 +101,7 @@ 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 for help."]
assert [segment.text for segment in result.transcript] == ["I ask Chontia for help.", "Then Lyra."]
assert len(result.skipped) == 1
assert result.skipped[0].segment_id == 99
assert "does not exist" in result.skipped[0].reason
@@ -118,7 +118,7 @@ def test_apply_corrections_skips_no_op():
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia for help."
assert result.transcript[0].text == "I ask Chontia for help."
assert len(result.skipped) == 1
assert "identical" in result.skipped[0].reason
@@ -156,7 +156,7 @@ def test_apply_corrections_skips_empty_original_text():
result = apply_corrections(transcript, [correction], confidence_threshold=0.8)
assert result.transcript[1].text == "I ask Chontia for help."
assert result.transcript[0].text == "I ask Chontia for help."
assert len(result.skipped) == 1
assert "empty" in result.skipped[0].reason

118
tests/test_pipeline.py
View File

@@ -9,18 +9,21 @@ class FakeLLMClient:
def __init__(self, responses):
self.responses = list(responses)
self.calls = 0
self.messages = []
def create_corrections(self, messages, config):
self.calls += 1
self.messages.append(messages)
return self.responses.pop(0)
def _config(tmp_path):
def _config(tmp_path, glossary_max_llm_passes=3):
return AuditaConfig(
api_key="key",
max_section_tokens=16000,
confidence_threshold=0.8,
max_retries=3,
glossary_max_llm_passes=glossary_max_llm_passes,
work_dir=tmp_path / "work",
)
@@ -40,7 +43,8 @@ def _transcript():
return parse_transcript_json(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "I ask Chontia."}
{"speaker": "Eric", "start": 10.0, "end": 11.0, "text": "I ask Chontia."},
{"speaker": "Mike", "start": 0.0, "end": 1.0, "text": "Then Lyra."}
]
"""
)
@@ -62,7 +66,8 @@ def test_pipeline_processes_with_fake_llm_and_cleans_work_dir(tmp_path):
llm_client=fake_client,
)
assert revised[0].text == "I ask Chauntea."
assert [segment.speaker for segment in revised] == ["Mike", "Eric"]
assert revised[1].text == "I ask Chauntea."
assert fake_client.calls == 1
assert list((tmp_path / "work").iterdir()) == []
@@ -80,12 +85,12 @@ def test_pipeline_skips_bad_correction_and_preserves_diagnostics(tmp_path):
revised = process_transcript(
_transcript(),
_glossary(),
_config(tmp_path),
_config(tmp_path, glossary_max_llm_passes=1),
llm_client=fake_client,
progress=progress.append,
)
assert revised[0].text == "I ask Chontia."
assert revised[1].text == "I ask Chontia."
assert any("Skipping correction for segment 0" in message for message in progress)
preserved = list((tmp_path / "work").iterdir())
assert len(preserved) == 1
@@ -94,3 +99,106 @@ def test_pipeline_skips_bad_correction_and_preserves_diagnostics(tmp_path):
diagnostics = json.loads(skipped_path.read_text(encoding="utf-8"))
assert diagnostics["skipped_corrections"][0]["segment_id"] == 0
assert "does not match any substring" in diagnostics["skipped_corrections"][0]["reason"]
def test_pipeline_retries_skipped_segment_and_cleans_work_dir_when_fixed(tmp_path):
first_pass = CorrectionCandidate(
segment_id=0,
original_text="Contia",
corrected_text="Chauntea",
confidence=0.95,
)
second_pass = CorrectionCandidate(
segment_id=0,
original_text="Chontia",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[first_pass]),
CorrectionSet(corrections=[second_pass]),
]
)
revised = process_transcript(
_transcript(),
_glossary(),
_config(tmp_path, glossary_max_llm_passes=3),
llm_client=fake_client,
)
assert fake_client.calls == 2
assert [segment.speaker for segment in revised] == ["Mike", "Eric"]
assert revised[1].text == "I ask Chauntea."
assert list((tmp_path / "work").iterdir()) == []
def test_pipeline_retry_prompt_contains_only_valid_deduped_segment_ids(tmp_path):
first_bad = CorrectionCandidate(
segment_id=0,
original_text="Contia",
corrected_text="Chauntea",
confidence=0.95,
)
second_bad_same_segment = CorrectionCandidate(
segment_id=0,
original_text="Still wrong",
corrected_text="Chauntea",
confidence=0.95,
)
invalid_segment = CorrectionCandidate(
segment_id=99,
original_text="Missing",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[first_bad, second_bad_same_segment, invalid_segment]),
CorrectionSet(corrections=[]),
]
)
process_transcript(
_transcript(),
_glossary(),
_config(tmp_path, glossary_max_llm_passes=2),
llm_client=fake_client,
)
assert fake_client.calls == 2
retry_prompt = fake_client.messages[1][1]["content"]
retry_payload = json.loads(retry_prompt.split("Transcript section:\n", maxsplit=1)[1])
assert retry_payload == [{"segment_id": 0, "original_text": "I ask Chontia."}]
assert "Retry guidance" in retry_prompt
def test_pipeline_writes_pass_metadata_for_unresolved_retries(tmp_path):
first_pass = CorrectionCandidate(
segment_id=0,
original_text="Contia",
corrected_text="Chauntea",
confidence=0.95,
)
fake_client = FakeLLMClient(
[
CorrectionSet(corrections=[first_pass]),
CorrectionSet(corrections=[]),
]
)
process_transcript(
_transcript(),
_glossary(),
_config(tmp_path, glossary_max_llm_passes=2),
llm_client=fake_client,
)
run_dirs = list((tmp_path / "work").iterdir())
assert len(run_dirs) == 1
metadata = json.loads((run_dirs[0] / "metadata.json").read_text(encoding="utf-8"))
assert metadata["glossary_max_llm_passes"] == 2
assert [item["pass_number"] for item in metadata["passes"]] == [1, 2]
assert metadata["passes"][0]["retry_segment_count"] == 1
assert metadata["passes"][1]["retry_pass"] is True