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

Updated configuration to utilize segment ids provided in the input transcript

Browse Source
This commit is contained in:
Eric Rakestraw
2026-04-21 15:01:31 -05:00
parent 7e1a3f721a
commit ca01e46d77
13 changed files with 186 additions and 97 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
src/audita/chunking.py
View File

@@ -41,7 +41,7 @@ class IndexedSegment:
return self.segment.model_dump(mode="json")
def prompt_payload(self) -> dict:
return {"segment_id": self.index, "original_text": self.segment.text}
return {"id": self.segment.id, "original_text": self.segment.text}
@dataclass(frozen=True)
@@ -116,7 +116,7 @@ def chunk_indexed_segments(
sections.append(_make_section(len(sections), current, current_tokens))
for section in sections:
parse_transcript_json(section.transcript_json())
parse_transcript_json(section.transcript_json(), require_sequential_ids=False)
return sections

2
src/audita/config.py
View File

@@ -9,7 +9,7 @@ from .errors import AuditaConfigError
DEFAULT_MODEL = "openrouter/mistralai/mistral-small-3.2-24b-instruct"
DEFAULT_BASE_URL = "https://openrouter.ai/api/v1"
DEFAULT_MAX_SECTION_TOKENS = 16000
DEFAULT_CONFIDENCE_THRESHOLD = 0.80
DEFAULT_CONFIDENCE_THRESHOLD = 0.60
DEFAULT_MAX_RETRIES = 3
DEFAULT_WORK_DIR = "/tmp/audita"
DEFAULT_GLOSSARY_MAX_LLM_PASSES = 3

45
src/audita/corrections.py
View File

@@ -1,5 +1,5 @@
from dataclasses import asdict, dataclass
from typing import Iterable, List, Optional, Tuple
from typing import Dict, Iterable, List, Optional, Tuple
from .errors import AuditaValidationError
from .schemas import CorrectionCandidate, TranscriptSegment
@@ -7,7 +7,7 @@ from .schemas import CorrectionCandidate, TranscriptSegment
@dataclass(frozen=True)
class SkippedCorrection:
segment_id: int
id: int
reason: str
original_text: str
corrected_text: str
@@ -22,8 +22,8 @@ class SkippedCorrection:
class CorrectionApplicationResult:
transcript: List[TranscriptSegment]
skipped: List[SkippedCorrection]
applied_segment_ids: List[int]
ignored_segment_ids: List[int]
applied_ids: List[int]
ignored_ids: List[int]
def apply_corrections(
@@ -35,40 +35,43 @@ def apply_corrections(
raise AuditaValidationError("Confidence threshold must be between 0.0 and 1.0.")
revised = list(transcript)
id_to_position = _id_to_position(revised)
skipped: List[SkippedCorrection] = []
applied_segment_ids: List[int] = []
ignored_segment_ids: List[int] = []
applied_ids: List[int] = []
ignored_ids: List[int] = []
for correction in corrections:
if correction.confidence < confidence_threshold:
ignored_segment_ids.append(correction.segment_id)
ignored_ids.append(correction.id)
continue
reason, actual_text = _target_error(revised, correction)
reason, actual_text = _target_error(revised, id_to_position, correction)
if reason is not None:
skipped.append(_skip(correction, reason, actual_text=actual_text))
continue
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})
applied_segment_ids.append(correction.segment_id)
position = id_to_position[correction.id]
segment = revised[position]
revised_text = segment.text.replace(correction.original_text, correction.corrected_text)
revised[position] = segment.model_copy(update={"text": revised_text})
applied_ids.append(correction.id)
return CorrectionApplicationResult(
transcript=revised,
skipped=skipped,
applied_segment_ids=applied_segment_ids,
ignored_segment_ids=ignored_segment_ids,
applied_ids=applied_ids,
ignored_ids=ignored_ids,
)
def _target_error(
transcript: List[TranscriptSegment],
id_to_position: Dict[int, int],
correction: CorrectionCandidate,
) -> Tuple[Optional[str], Optional[str]]:
if correction.segment_id >= len(transcript):
return "segment_id does not exist in transcript", None
if correction.id not in id_to_position:
return "id does not exist in transcript", None
segment = transcript[correction.segment_id]
segment = transcript[id_to_position[correction.id]]
if correction.original_text == "":
return "original_text is empty", segment.text
if correction.original_text == correction.corrected_text:
@@ -77,19 +80,21 @@ def _target_error(
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
def _id_to_position(transcript: List[TranscriptSegment]) -> Dict[int, int]:
return {segment.id: position for position, segment in enumerate(transcript)}
def _skip(
correction: CorrectionCandidate,
reason: str,
actual_text: Optional[str] = None,
) -> SkippedCorrection:
return SkippedCorrection(
segment_id=correction.segment_id,
id=correction.id,
reason=reason,
original_text=correction.original_text,
corrected_text=correction.corrected_text,

43
src/audita/pipeline.py
View File

@@ -41,12 +41,12 @@ def process_transcript(
for pass_number in range(1, config.glossary_max_llm_passes + 1):
if pass_number == 1:
indexed_segments = _indexed_segments_for_ids(working, list(range(len(working))))
indexed_segments = _indexed_segments_for_ids(working, [segment.id for segment in working])
else:
retry_segment_ids = sorted(unresolved_retry_skips)
if not retry_segment_ids:
retry_ids = sorted(unresolved_retry_skips)
if not retry_ids:
break
indexed_segments = _indexed_segments_for_ids(working, retry_segment_ids)
indexed_segments = _indexed_segments_for_ids(working, retry_ids)
if not indexed_segments:
break
@@ -80,13 +80,13 @@ def process_transcript(
)
working = application_result.transcript
for segment_id in application_result.applied_segment_ids:
unresolved_retry_skips.pop(segment_id, None)
for segment_id in application_result.ignored_segment_ids:
unresolved_retry_skips.pop(segment_id, None)
for correction_id in application_result.applied_ids:
unresolved_retry_skips.pop(correction_id, None)
for correction_id in application_result.ignored_ids:
unresolved_retry_skips.pop(correction_id, None)
for skipped in application_result.skipped:
if _is_retryable_skip(skipped, len(working)):
unresolved_retry_skips[skipped.segment_id] = skipped
if _is_retryable_skip(skipped, working):
unresolved_retry_skips[skipped.id] = skipped
else:
final_nonretry_skips.append(skipped)
@@ -97,8 +97,8 @@ def process_transcript(
"section_count": len(sections),
"segment_count": len(indexed_segments),
"corrections_returned": len(corrections),
"applied_count": len(application_result.applied_segment_ids),
"ignored_below_threshold_count": len(application_result.ignored_segment_ids),
"applied_count": len(application_result.applied_ids),
"ignored_below_threshold_count": len(application_result.ignored_ids),
"skipped_count": len(application_result.skipped),
"retry_segment_count": len(unresolved_retry_skips),
}
@@ -109,13 +109,13 @@ def process_transcript(
break
final_skipped = final_nonretry_skips + [
unresolved_retry_skips[segment_id] for segment_id in sorted(unresolved_retry_skips)
unresolved_retry_skips[correction_id] for correction_id in sorted(unresolved_retry_skips)
]
_write_skipped_corrections(run_dir, final_skipped)
for skipped in final_skipped:
_log(
progress,
f"Skipping correction for segment {skipped.segment_id}: {skipped.reason}",
f"Skipping correction for id {skipped.id}: {skipped.reason}",
)
revised = _sort_transcript_chronologically(working)
except Exception as exc:
@@ -164,7 +164,7 @@ def _write_and_validate_section(run_dir: Path, section: TranscriptSection) -> No
section_path = run_dir / f"section-{section.section_index:04d}.json"
section_json = section.transcript_json()
section_path.write_text(section_json, encoding="utf-8")
parse_transcript_json(section_json)
parse_transcript_json(section_json, require_sequential_ids=False)
def _write_skipped_corrections(run_dir: Path, skipped: List[SkippedCorrection]) -> None:
@@ -182,17 +182,18 @@ def _write_skipped_corrections(run_dir: Path, skipped: List[SkippedCorrection])
def _indexed_segments_for_ids(
transcript: List[TranscriptSegment],
segment_ids: List[int],
ids: List[int],
) -> List[IndexedSegment]:
id_to_position = {segment.id: position for position, segment in enumerate(transcript)}
return [
IndexedSegment(index=segment_id, segment=transcript[segment_id])
for segment_id in segment_ids
if 0 <= segment_id < len(transcript)
IndexedSegment(index=id_to_position[segment_id], segment=transcript[id_to_position[segment_id]])
for segment_id in ids
if segment_id in id_to_position
]
def _is_retryable_skip(skipped: SkippedCorrection, transcript_length: int) -> bool:
return 0 <= skipped.segment_id < transcript_length
def _is_retryable_skip(skipped: SkippedCorrection, transcript: List[TranscriptSegment]) -> bool:
return any(segment.id == skipped.id for segment in transcript)
def _sort_transcript_chronologically(

4
src/audita/prompts.py
View File

@@ -44,10 +44,10 @@ def build_glossary_correction_messages(
"- 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 from the input segment.\n"
"- Use the exact 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"
"- Each returned correction must contain only 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"

36
src/audita/schemas.py
View File

@@ -11,11 +11,19 @@ from .errors import AuditaValidationError
class TranscriptSegment(BaseModel):
model_config = ConfigDict(extra="forbid")
id: int = Field(ge=1)
speaker: StrictStr
start: float
end: float
text: StrictStr
@field_validator("id", mode="before")
@classmethod
def require_integer_id(cls, value: Any) -> int:
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError("must be an integer")
return value
@field_validator("speaker", "text")
@classmethod
def require_non_empty_text(cls, value: str) -> str:
@@ -82,12 +90,12 @@ class Glossary(BaseModel):
class CorrectionCandidate(BaseModel):
model_config = ConfigDict(extra="forbid")
segment_id: int = Field(ge=0)
id: int = Field(ge=1)
original_text: StrictStr
corrected_text: StrictStr
confidence: float = Field(ge=0.0, le=1.0)
@field_validator("segment_id", mode="before")
@field_validator("id", mode="before")
@classmethod
def require_integer_id(cls, value: Any) -> int:
if isinstance(value, bool) or not isinstance(value, int):
@@ -114,23 +122,39 @@ class CorrectionSet(BaseModel):
_TRANSCRIPT_ADAPTER = TypeAdapter(List[TranscriptSegment])
def validate_transcript_data(data: Any) -> List[TranscriptSegment]:
def validate_transcript_data(
data: Any,
require_sequential_ids: bool = True,
) -> List[TranscriptSegment]:
if not isinstance(data, list):
raise AuditaValidationError("Transcript must be a JSON array.")
if not data:
raise AuditaValidationError("Transcript must contain at least one segment.")
try:
return _TRANSCRIPT_ADAPTER.validate_python(data)
transcript = _TRANSCRIPT_ADAPTER.validate_python(data)
except ValidationError as exc:
raise AuditaValidationError(f"Transcript schema validation failed: {exc}") from exc
if require_sequential_ids:
_validate_sequential_ids(transcript)
return transcript
def parse_transcript_json(raw: str) -> List[TranscriptSegment]:
def _validate_sequential_ids(transcript: List[TranscriptSegment]) -> None:
ids = [segment.id for segment in transcript]
expected = list(range(1, len(transcript) + 1))
if ids != expected:
raise AuditaValidationError("Transcript segment ids must be sequential starting at 1.")
def parse_transcript_json(
raw: str,
require_sequential_ids: bool = True,
) -> List[TranscriptSegment]:
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
raise AuditaValidationError(f"Transcript is not valid JSON: {exc}") from exc
return validate_transcript_data(data)
return validate_transcript_data(data, require_sequential_ids=require_sequential_ids)
def parse_glossary_yaml(raw: str) -> Glossary: