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

Moved the current implementation to src/audita_prototype, moved current test suite to tests/audita_prototype, and started a new, more modular application skeleton in src/audita

Browse Source
This commit is contained in:
Eric Rakestraw
2026-04-24 10:17:38 -05:00
parent 889b620f2e
commit f39de37974
60 changed files with 3034 additions and 664 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
README.md
View File

@@ -1,6 +1,13 @@
# Audita
Audita takes raw audio transcripts, deterministically merges short same-speaker segments into speaking turns, and uses an LLM to identify and fix misheard words, jargon, domain-specific terms, and conservative readability issues.
Audita is now a framework-first transcript correction application. The public `audita` package provides:
- deterministic transcript normalization
- token-batched module orchestration
- reusable module / filter / review-stage contracts
- structured run reporting and work-dir diagnostics
The previous working implementation has been preserved as `audita_prototype` inside this repository. Its full regression suite lives under `tests/audita_prototype`.
## Development
@@ -13,14 +20,23 @@ uv run pytest
## Usage
Set an OpenRouter API key, then process a transcript with a glossary:
Process a transcript with the new framework skeleton:
```sh
export OPENROUTER_API_KEY=...
uv run audita process transcript.json --glossary glossary.yaml --output corrected.json
```
To also write a structured JSON report describing what Audita applied or skipped:
The framework currently runs this default module sequence:
1. `glossary_primary`
2. `homophones`
3. `glossary_secondary`
4. `spoken_word`
5. `grammar`
At this stage the module implementations are stubs. The framework is executable end to end, performs deterministic normalization, runs the full module lifecycle, and emits structured reports, but does not yet apply substantive LLM-driven corrections.
To also write a structured JSON report:
```sh
uv run audita process transcript.json --glossary glossary.yaml --output corrected.json --report-json report.json
@@ -44,21 +60,14 @@ audita process transcript.json --glossary glossary.yaml --output corrected.json
Without `--output`, Audita writes the corrected transcript JSON to stdout and progress logs to stderr.
`--report-json` writes a separate machine-readable run report and never mixes report data into stdout.
Useful configuration can be supplied by CLI flag or environment variable. CLI flags take precedence over environment variables.
Useful configuration can be supplied by CLI flag or environment variable. CLI flags take precedence over environment variables. The framework does not currently require `OPENROUTER_API_KEY`, but the config fields remain available for future module implementations.
| Environment variable | CLI flag | Default | Purpose |
| --- | --- | --- | --- |
| `AUDITA_MODEL` | `--model` | `openrouter/google/gemma-4-31b-it` | OpenRouter model to use |
| `AUDITA_BASE_URL` | `--base-url` | `https://openrouter.ai/api/v1` | OpenAI-compatible API base URL |
| `AUDITA_MAX_SECTION_TOKENS` | `--max-section-tokens` | `6144` | Maximum estimated tokens per LLM transcript section |
| `AUDITA_GLOSSARY_CONFIDENCE_THRESHOLD` | `--glossary-confidence-threshold` | `0.80` | Minimum confidence required to apply a glossary correction |
| `AUDITA_GRAMMAR_CONFIDENCE_THRESHOLD` | `--grammar-confidence-threshold` | `0.80` | Minimum confidence required to apply a grammar correction |
| `AUDITA_GRAMMAR_VALIDATION_ENABLED` | `--grammar-validation-enabled` / `--no-grammar-validation-enabled` | `true` | Whether grammar corrections are checked by the semantic validator |
| `AUDITA_GRAMMAR_VALIDATION_CONFIDENCE_THRESHOLD` | `--grammar-validation-confidence-threshold` | `0.80` | Minimum validator confidence required for validated grammar corrections |
| `AUDITA_GRAMMAR_SPOKEN_FORM_VALIDATION_CONFIDENCE_THRESHOLD` | `--grammar-spoken-form-validation-confidence-threshold` | `0.80` | Minimum validator confidence required for spoken-form rescue corrections |
| `AUDITA_MODEL` | `--model` | `openrouter/google/gemma-4-31b-it` | Reserved LLM model setting for future module implementations |
| `AUDITA_BASE_URL` | `--base-url` | `https://openrouter.ai/api/v1` | Reserved OpenAI-compatible API base URL |
| `AUDITA_MAX_RETRIES` | `--max-retries` | `3` | Maximum Instructor retries for structured responses |
| `AUDITA_GLOSSARY_MAX_LLM_PASSES` | `--glossary-max-llm-passes` | `3` | Total glossary correction passes |
| `AUDITA_GRAMMAR_MAX_LLM_PASSES` | `--grammar-max-llm-passes` | `3` | Total grammar/readability correction passes |
| `AUDITA_MAX_SECTION_TOKENS` | `--max-section-tokens` | `6144` | Maximum estimated tokens per transcript batch |
| `AUDITA_NORMALIZE_MAX_SEGMENT_GAP` | `--normalize-max-segment-gap` | `4.0` | Same-speaker gaps eligible for deterministic merging |
| `AUDITA_NORMALIZE_ELLIPSIS_GAP` | `--normalize-ellipsis-gap` | `3.5` | Same-speaker gaps above this value are joined with ` ... ` |
| `AUDITA_NORMALIZE_MAX_SEGMENT_DURATION` | `--normalize-max-segment-duration` | `60.0` | Maximum merged segment duration |
@@ -66,6 +75,8 @@ Useful configuration can be supplied by CLI flag or environment variable. CLI fl
| `AUDITA_WORK_DIR` | `--work-dir` | `/tmp/audita` | Per-run scratch diagnostics directory |
| `AUDITA_WORK_DIR_RETENTION` | `--work-dir-retention` | `auto` | Whether to retain the per-run work directory: `auto`, `always`, or `never` |
`OPENROUTER_API_KEY` is required and is read from the environment.
`AUDITA_WORK_DIR` stores per-run diagnostics while processing. Under the default `AUDITA_WORK_DIR_RETENTION=auto`, clean successful runs are removed, while failed runs and successful runs with final skipped corrections are preserved. Use `always` to keep every run directory and `never` to remove successful run directories even when skips remain.
## Prototype Archive
The archived prototype remains importable as `audita_prototype` and is still covered by its original regression suite. This is intentional: the new `audita` package is a fresh framework skeleton, not a thin wrapper around the old code.

7
pyproject.toml
View File

@@ -1,7 +1,7 @@
[project]
name = "audita"
version = "0.1.0"
description = "Correct audio transcripts with staged LLM review passes."
version = "0.2.0"
description = "Framework-first audio transcript correction pipeline with an archived prototype."
readme = "README.md"
requires-python = ">=3.9"
license = { text = "BSD-3-Clause" }
@@ -26,6 +26,9 @@ audita = "audita.cli:main"
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["src/audita", "src/audita_prototype"]
[tool.pytest.ini_options]
pythonpath = ["src"]
testpaths = ["tests"]

5
src/audita/__init__.py
View File

@@ -1,6 +1,5 @@
"""Audita transcript correction package."""
"""Audita modular transcript correction framework."""
__all__ = ["__version__"]
__version__ = "0.1.0"
__version__ = "0.2.0"

1
src/audita/__main__.py
View File

@@ -3,4 +3,3 @@ from .cli import main
if __name__ == "__main__":
raise SystemExit(main())

61
src/audita/cli.py
View File

@@ -3,20 +3,18 @@ import sys
from pathlib import Path
from typing import Optional, Sequence
from .config import AuditaConfig, ConfigOverrides
from .errors import AuditaError
from .io import load_glossary, load_transcript, write_report, write_transcript
from .core.config import AuditaConfig, ConfigOverrides
from .core.errors import AuditaError
from .core.io import load_glossary, load_transcript, write_report, write_transcript
from .core.schemas import transcript_to_json
from .pipeline import process_transcript_result
from .schemas import transcript_to_json
def main(argv: Optional[Sequence[str]] = None) -> int:
parser = _build_parser()
args = parser.parse_args(argv)
if args.command == "process":
return _process(args)
parser.print_help(sys.stderr)
return 2
@@ -24,44 +22,15 @@ def main(argv: Optional[Sequence[str]] = None) -> int:
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="audita")
subparsers = parser.add_subparsers(dest="command", required=True)
process = subparsers.add_parser("process", help="correct a transcript using a glossary")
process = subparsers.add_parser("process", help="run the modular transcript correction framework")
process.add_argument("transcript", type=Path, help="path to the input transcript JSON")
process.add_argument("--glossary", type=Path, required=True, help="path to the glossary YAML")
process.add_argument("--output", type=Path, help="write corrected transcript JSON to this path")
process.add_argument("--report-json", type=Path, help="write structured run report JSON to this path")
process.add_argument("--model", help="OpenRouter model to use")
process.add_argument("--base-url", help="OpenAI-compatible API base URL")
process.add_argument("--max-section-tokens", type=int, help="maximum estimated tokens per transcript section")
process.add_argument(
"--glossary-confidence-threshold",
type=float,
help="minimum confidence required to apply a glossary correction",
)
process.add_argument(
"--grammar-confidence-threshold",
type=float,
help="minimum confidence required to apply a grammar correction",
)
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(
"--grammar-spoken-form-validation-confidence-threshold",
type=float,
help="minimum validator confidence required to apply a spoken-form rescue correction",
)
process.add_argument("--model", help="OpenRouter model to use when module LLM stages are implemented")
process.add_argument("--base-url", help="OpenAI-compatible API base URL for future LLM stages")
process.add_argument("--max-retries", type=int, help="maximum structured-output retries for future LLM stages")
process.add_argument("--max-section-tokens", type=int, help="maximum estimated tokens per transcript batch")
process.add_argument(
"--normalize-max-segment-gap",
type=float,
@@ -97,17 +66,8 @@ def _process(args: argparse.Namespace) -> int:
overrides=ConfigOverrides(
model=args.model,
base_url=args.base_url,
max_section_tokens=args.max_section_tokens,
glossary_confidence_threshold=args.glossary_confidence_threshold,
grammar_confidence_threshold=args.grammar_confidence_threshold,
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,
grammar_spoken_form_validation_confidence_threshold=(
args.grammar_spoken_form_validation_confidence_threshold
),
max_section_tokens=args.max_section_tokens,
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,
@@ -124,7 +84,6 @@ def _process(args: argparse.Namespace) -> int:
config,
progress=lambda message: print(message, file=sys.stderr),
)
if args.output is not None:
write_transcript(args.output, result.transcript)
else:

1
src/audita/core/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Core data models and utilities for the Audita framework."""

146
src/audita/core/chunking.py Normal file
View File

@@ -0,0 +1,146 @@
import json
from dataclasses import dataclass
from math import ceil
from typing import Any, Callable, Generic, List, Optional, Protocol, Sequence, TypeVar
from .errors import AuditaValidationError
from .schemas import TranscriptSegment, parse_transcript_json
class TokenEstimatorProtocol(Protocol):
def estimate_json(self, value: Any) -> int:
...
class TokenEstimator:
def __init__(self, fallback_chars_per_token: int = 4) -> None:
self._fallback_chars_per_token = fallback_chars_per_token
self._encoding = None
try:
import tiktoken
self._encoding = tiktoken.get_encoding("cl100k_base")
except Exception:
self._encoding = None
def estimate_text(self, text: str) -> int:
if self._encoding is not None:
return len(self._encoding.encode(text))
return max(1, ceil(len(text) / self._fallback_chars_per_token))
def estimate_json(self, value: Any) -> int:
return self.estimate_text(json.dumps(value, ensure_ascii=False, separators=(",", ":")))
T = TypeVar("T")
@dataclass(frozen=True)
class TokenBatch(Generic[T]):
batch_index: int
items: List[T]
token_count: int
def chunk_payload_items(
items: Sequence[T],
max_tokens: int,
payload_fn: Callable[[T], Any],
estimator: Optional[TokenEstimatorProtocol] = None,
empty_error_message: str = "Batch input must contain at least one item.",
) -> List[TokenBatch[T]]:
if max_tokens <= 0:
raise AuditaValidationError("Maximum section token count must be greater than zero.")
if not items:
raise AuditaValidationError(empty_error_message)
token_estimator = TokenEstimator() if estimator is None else estimator
batches: List[TokenBatch[T]] = []
current: List[T] = []
current_tokens = 0
for item in items:
single_payload = [payload_fn(item)]
single_tokens = token_estimator.estimate_json(single_payload)
if single_tokens > max_tokens:
raise AuditaValidationError(
"A single transcript segment exceeds the maximum section token limit. "
"Raise the limit or pre-split the transcript."
)
candidate = current + [item]
candidate_tokens = token_estimator.estimate_json([payload_fn(candidate_item) for candidate_item in candidate])
if current and candidate_tokens > max_tokens:
batches.append(TokenBatch(batch_index=len(batches), items=list(current), token_count=current_tokens))
current = [item]
current_tokens = single_tokens
else:
current = candidate
current_tokens = candidate_tokens
if current:
batches.append(TokenBatch(batch_index=len(batches), items=list(current), token_count=current_tokens))
return batches
@dataclass(frozen=True)
class IndexedSegment:
index: int
segment: TranscriptSegment
def transcript_payload(self) -> dict:
return self.segment.model_dump(mode="json")
def prompt_payload(self) -> dict:
return {"id": self.segment.id, "original_text": self.segment.text}
@dataclass(frozen=True)
class TranscriptSection:
section_index: int
start_index: int
segments: List[IndexedSegment]
token_count: int
def transcript_payload(self) -> List[dict]:
return [item.transcript_payload() for item in self.segments]
def prompt_payload(self) -> List[dict]:
return [item.prompt_payload() for item in self.segments]
def transcript_json(self) -> str:
return json.dumps(self.transcript_payload(), ensure_ascii=False, indent=2) + "\n"
def chunk_transcript(
segments: List[TranscriptSegment],
max_section_tokens: int,
estimator: Optional[TokenEstimatorProtocol] = None,
) -> List[TranscriptSection]:
indexed = [IndexedSegment(index=index, segment=segment) for index, segment in enumerate(segments)]
return chunk_indexed_segments(indexed, max_section_tokens, estimator=estimator)
def chunk_indexed_segments(
indexed_segments: List[IndexedSegment],
max_section_tokens: int,
estimator: Optional[TokenEstimatorProtocol] = None,
) -> List[TranscriptSection]:
batches = chunk_payload_items(
indexed_segments,
max_section_tokens,
payload_fn=lambda item: item.prompt_payload(),
estimator=estimator,
empty_error_message="Transcript must contain at least one segment.",
)
sections = [
TranscriptSection(
section_index=batch.batch_index,
start_index=batch.items[0].index,
segments=list(batch.items),
token_count=batch.token_count,
)
for batch in batches
]
for section in sections:
parse_transcript_json(section.transcript_json(), require_sequential_ids=False)
return sections

194
src/audita/core/config.py Normal file
View File

@@ -0,0 +1,194 @@
import math
import os
from dataclasses import dataclass
from pathlib import Path
from typing import Mapping, Optional
from .errors import AuditaConfigError
DEFAULT_MODEL = "openrouter/google/gemma-4-31b-it"
DEFAULT_BASE_URL = "https://openrouter.ai/api/v1"
DEFAULT_MAX_RETRIES = 3
DEFAULT_MAX_SECTION_TOKENS = 6144
DEFAULT_WORK_DIR = "/tmp/audita"
DEFAULT_WORK_DIR_RETENTION = "auto"
DEFAULT_NORMALIZE_MAX_SEGMENT_GAP = 4.0
DEFAULT_NORMALIZE_ELLIPSIS_GAP = 3.5
DEFAULT_NORMALIZE_MAX_SEGMENT_DURATION = 60.0
DEFAULT_NORMALIZE_MAX_SEGMENT_TOKENS = 2048
@dataclass(frozen=True)
class ConfigOverrides:
model: Optional[str] = None
base_url: Optional[str] = None
max_retries: Optional[int] = None
max_section_tokens: Optional[int] = None
normalize_max_segment_gap: Optional[float] = None
normalize_ellipsis_gap: Optional[float] = None
normalize_max_segment_duration: Optional[float] = None
normalize_max_segment_tokens: Optional[int] = None
work_dir: Optional[Path] = None
work_dir_retention: Optional[str] = None
@dataclass(frozen=True)
class AuditaConfig:
api_key: Optional[str] = None
model: str = DEFAULT_MODEL
base_url: str = DEFAULT_BASE_URL
max_retries: int = DEFAULT_MAX_RETRIES
max_section_tokens: int = DEFAULT_MAX_SECTION_TOKENS
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
normalize_max_segment_tokens: int = DEFAULT_NORMALIZE_MAX_SEGMENT_TOKENS
work_dir: Path = Path(DEFAULT_WORK_DIR)
work_dir_retention: str = DEFAULT_WORK_DIR_RETENTION
@classmethod
def from_sources(
cls,
env: Optional[Mapping[str, str]] = None,
overrides: Optional[ConfigOverrides] = None,
) -> "AuditaConfig":
source = os.environ if env is None else env
selected = ConfigOverrides() if overrides is None else overrides
config = cls(
api_key=_select_optional_string(source.get("OPENROUTER_API_KEY")),
model=selected.model or source.get("AUDITA_MODEL") or DEFAULT_MODEL,
base_url=selected.base_url or source.get("AUDITA_BASE_URL") or DEFAULT_BASE_URL,
max_retries=_select_int(
selected.max_retries,
source.get("AUDITA_MAX_RETRIES"),
DEFAULT_MAX_RETRIES,
"AUDITA_MAX_RETRIES",
),
max_section_tokens=_select_int(
selected.max_section_tokens,
source.get("AUDITA_MAX_SECTION_TOKENS"),
DEFAULT_MAX_SECTION_TOKENS,
"AUDITA_MAX_SECTION_TOKENS",
),
normalize_max_segment_gap=_select_float(
selected.normalize_max_segment_gap,
source.get("AUDITA_NORMALIZE_MAX_SEGMENT_GAP"),
DEFAULT_NORMALIZE_MAX_SEGMENT_GAP,
"AUDITA_NORMALIZE_MAX_SEGMENT_GAP",
),
normalize_ellipsis_gap=_select_float(
selected.normalize_ellipsis_gap,
source.get("AUDITA_NORMALIZE_ELLIPSIS_GAP"),
DEFAULT_NORMALIZE_ELLIPSIS_GAP,
"AUDITA_NORMALIZE_ELLIPSIS_GAP",
),
normalize_max_segment_duration=_select_float(
selected.normalize_max_segment_duration,
source.get("AUDITA_NORMALIZE_MAX_SEGMENT_DURATION"),
DEFAULT_NORMALIZE_MAX_SEGMENT_DURATION,
"AUDITA_NORMALIZE_MAX_SEGMENT_DURATION",
),
normalize_max_segment_tokens=_select_int(
selected.normalize_max_segment_tokens,
source.get("AUDITA_NORMALIZE_MAX_SEGMENT_TOKENS"),
DEFAULT_NORMALIZE_MAX_SEGMENT_TOKENS,
"AUDITA_NORMALIZE_MAX_SEGMENT_TOKENS",
),
work_dir=selected.work_dir or Path(source.get("AUDITA_WORK_DIR") or DEFAULT_WORK_DIR),
work_dir_retention=_select_choice(
selected.work_dir_retention,
source.get("AUDITA_WORK_DIR_RETENTION"),
DEFAULT_WORK_DIR_RETENTION,
"AUDITA_WORK_DIR_RETENTION",
("auto", "always", "never"),
),
)
config.validate()
return config
def validate(self) -> None:
if not self.model.strip():
raise AuditaConfigError("AUDITA_MODEL must not be empty.")
if not self.base_url.strip():
raise AuditaConfigError("AUDITA_BASE_URL must not be empty.")
if self.max_retries < 0:
raise AuditaConfigError("AUDITA_MAX_RETRIES must be greater than or equal to zero.")
if self.max_section_tokens <= 0:
raise AuditaConfigError("AUDITA_MAX_SECTION_TOKENS must be greater than zero.")
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):
raise AuditaConfigError("AUDITA_NORMALIZE_ELLIPSIS_GAP must be finite.")
if not math.isfinite(self.normalize_max_segment_duration):
raise AuditaConfigError("AUDITA_NORMALIZE_MAX_SEGMENT_DURATION must be finite.")
if self.normalize_max_segment_gap < 0:
raise AuditaConfigError("AUDITA_NORMALIZE_MAX_SEGMENT_GAP must be greater than or equal to zero.")
if self.normalize_ellipsis_gap < 0:
raise AuditaConfigError("AUDITA_NORMALIZE_ELLIPSIS_GAP must be greater than or equal to zero.")
if self.normalize_ellipsis_gap > self.normalize_max_segment_gap:
raise AuditaConfigError(
"AUDITA_NORMALIZE_ELLIPSIS_GAP must be less than or equal to AUDITA_NORMALIZE_MAX_SEGMENT_GAP."
)
if self.normalize_max_segment_duration <= 0:
raise AuditaConfigError("AUDITA_NORMALIZE_MAX_SEGMENT_DURATION must be greater than zero.")
if self.normalize_max_segment_tokens <= 0:
raise AuditaConfigError("AUDITA_NORMALIZE_MAX_SEGMENT_TOKENS must be greater than zero.")
if self.work_dir_retention not in ("auto", "always", "never"):
raise AuditaConfigError("AUDITA_WORK_DIR_RETENTION must be one of auto, always, or never.")
def to_report_dict(self) -> dict:
return {
"api_key_configured": bool(self.api_key),
"model": self.model,
"base_url": self.base_url,
"max_retries": self.max_retries,
"max_section_tokens": self.max_section_tokens,
"normalize_max_segment_gap": self.normalize_max_segment_gap,
"normalize_ellipsis_gap": self.normalize_ellipsis_gap,
"normalize_max_segment_duration": self.normalize_max_segment_duration,
"normalize_max_segment_tokens": self.normalize_max_segment_tokens,
"work_dir_retention": self.work_dir_retention,
}
def _select_optional_string(value: Optional[str]) -> Optional[str]:
if value is None:
return None
stripped = value.strip()
return stripped or None
def _select_int(cli_value: Optional[int], env_value: Optional[str], default: int, name: str) -> int:
if cli_value is not None:
return cli_value
if env_value is None:
return default
try:
return int(env_value)
except ValueError as exc:
raise AuditaConfigError(f"{name} must be an integer.") from exc
def _select_float(cli_value: Optional[float], env_value: Optional[str], default: float, name: str) -> float:
if cli_value is not None:
return cli_value
if env_value is None:
return default
try:
return float(env_value)
except ValueError as exc:
raise AuditaConfigError(f"{name} must be a number.") from exc
def _select_choice(
cli_value: Optional[str],
env_value: Optional[str],
default: str,
name: str,
valid_choices: tuple[str, ...],
) -> str:
value = cli_value or env_value or default
if value not in valid_choices:
raise AuditaConfigError(f"{name} must be one of {', '.join(valid_choices)}.")
return value

0
src/audita/errors.py → src/audita/core/errors.py
View File

0
src/audita/io.py → src/audita/core/io.py
View File

172
src/audita/core/normalization.py Normal file
View File

@@ -0,0 +1,172 @@
from dataclasses import asdict, dataclass
from typing import List, Optional
from .chunking import TokenEstimator, TokenEstimatorProtocol
from .schemas import SourceTranscriptSegment, TranscriptSegment
@dataclass(frozen=True)
class NormalizationSummary:
source_segment_count: int
normalized_segment_count: int
merge_count: int
max_segment_gap: float
ellipsis_gap: float
max_segment_duration: float
max_segment_tokens: int
def to_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class NormalizationResult:
transcript: List[TranscriptSegment]
summary: NormalizationSummary
@dataclass(frozen=True)
class _WorkingSegment:
speaker: str
start: float
end: float
text: str
order: int
def normalize_transcript(
segments: List[SourceTranscriptSegment],
max_segment_gap: float,
ellipsis_gap: float,
max_segment_duration: float,
max_segment_tokens: int,
estimator: Optional[TokenEstimatorProtocol] = None,
) -> NormalizationResult:
token_estimator = TokenEstimator() if estimator is None else estimator
working = [
_WorkingSegment(
speaker=segment.speaker,
start=segment.start,
end=segment.end,
text=segment.text,
order=index,
)
for index, segment in enumerate(segments)
]
working.sort(key=lambda segment: (segment.start, segment.end, segment.order))
merge_count = 0
while True:
candidate_index = _shortest_mergeable_gap_index(
working,
max_segment_gap,
ellipsis_gap,
max_segment_duration,
max_segment_tokens,
token_estimator,
)
if candidate_index is None:
break
left = working[candidate_index]
right = working[candidate_index + 1]
working[candidate_index : candidate_index + 2] = [_merge_segments(left, right, ellipsis_gap)]
merge_count += 1
normalized = _assign_ids(working)
return NormalizationResult(
transcript=normalized,
summary=NormalizationSummary(
source_segment_count=len(segments),
normalized_segment_count=len(normalized),
merge_count=merge_count,
max_segment_gap=max_segment_gap,
ellipsis_gap=ellipsis_gap,
max_segment_duration=max_segment_duration,
max_segment_tokens=max_segment_tokens,
),
)
def _shortest_mergeable_gap_index(
segments: List[_WorkingSegment],
max_segment_gap: float,
ellipsis_gap: float,
max_segment_duration: float,
max_segment_tokens: int,
estimator: TokenEstimatorProtocol,
) -> Optional[int]:
best_index = None
best_gap = None
for index in range(len(segments) - 1):
left = segments[index]
right = segments[index + 1]
gap = right.start - left.end
if not _can_merge(
left,
right,
gap,
max_segment_gap,
ellipsis_gap,
max_segment_duration,
max_segment_tokens,
estimator,
):
continue
if best_gap is None or gap < best_gap:
best_index = index
best_gap = gap
return best_index
def _can_merge(
left: _WorkingSegment,
right: _WorkingSegment,
gap: float,
max_segment_gap: float,
ellipsis_gap: float,
max_segment_duration: float,
max_segment_tokens: int,
estimator: TokenEstimatorProtocol,
) -> bool:
if left.speaker != right.speaker:
return False
if gap < 0 or gap > max_segment_gap:
return False
if right.end - left.start > max_segment_duration:
return False
merged_text = _joined_text(left.text, right.text, gap, ellipsis_gap)
return _estimate_prompt_tokens(merged_text, estimator) <= max_segment_tokens
def _merge_segments(left: _WorkingSegment, right: _WorkingSegment, ellipsis_gap: float) -> _WorkingSegment:
gap = right.start - left.end
return _WorkingSegment(
speaker=left.speaker,
start=left.start,
end=right.end,
text=_joined_text(left.text, right.text, gap, ellipsis_gap),
order=left.order,
)
def _joined_text(left_text: str, right_text: str, gap: float, ellipsis_gap: float) -> str:
joiner = " " if gap <= ellipsis_gap else " ... "
return f"{left_text.rstrip()}{joiner}{right_text.lstrip()}"
def _estimate_prompt_tokens(text: str, estimator: TokenEstimatorProtocol) -> int:
return estimator.estimate_json([{"id": 1, "original_text": text}])
def _assign_ids(segments: List[_WorkingSegment]) -> List[TranscriptSegment]:
ordered = sorted(segments, key=lambda segment: (segment.start, segment.end, segment.order))
return [
TranscriptSegment(
id=index + 1,
speaker=segment.speaker,
start=segment.start,
end=segment.end,
text=segment.text,
)
for index, segment in enumerate(ordered)
]

131
src/audita/core/reporting.py Normal file
View File

@@ -0,0 +1,131 @@
import json
from dataclasses import asdict, dataclass
from pathlib import Path
from typing import List, Optional
from .schemas import TranscriptSegment
@dataclass(frozen=True)
class AppliedChange:
module_instance: str
module_key: str
proposal_index: int
id: int
original_text: str
corrected_text: str
confidence: float
segment_text_before: str
segment_text_after: str
def to_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class ReportedSkip:
module_instance: str
module_key: str
proposal_index: int
id: int
reason: str
original_text: str
corrected_text: str
confidence: float
actual_text: Optional[str] = None
source: Optional[str] = None
def to_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class ReviewStageReport:
name: str
candidate_count: int
approved_count: int
rejected_count: int
def to_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class DeterministicFilterReport:
name: str
passed_count: int
rejected_count: int
def to_dict(self) -> dict:
return asdict(self)
@dataclass(frozen=True)
class ModuleRunReport:
instance_name: str
module_key: str
replacement_policy: str
section_count: int
proposal_count: int
deterministic_filters: List[DeterministicFilterReport]
review_stages: List[ReviewStageReport]
approved_count: int
applied_count: int
skipped_count: int
def to_dict(self) -> dict:
return {
"instance_name": self.instance_name,
"module_key": self.module_key,
"replacement_policy": self.replacement_policy,
"section_count": self.section_count,
"proposal_count": self.proposal_count,
"deterministic_filters": [item.to_dict() for item in self.deterministic_filters],
"review_stages": [item.to_dict() for item in self.review_stages],
"approved_count": self.approved_count,
"applied_count": self.applied_count,
"skipped_count": self.skipped_count,
}
@dataclass(frozen=True)
class RunReport:
status: str
config: dict
normalization: Optional[dict]
pipeline: List[str]
modules: List[ModuleRunReport]
applied_changes: List[AppliedChange]
skipped_corrections: List[ReportedSkip]
totals: dict
work_dir_retention: str
work_dir_retained: bool
work_dir: Optional[str]
error: Optional[str] = None
def to_dict(self) -> dict:
return {
"status": self.status,
"config": self.config,
"normalization": self.normalization,
"pipeline": self.pipeline,
"modules": [item.to_dict() for item in self.modules],
"applied_changes": [item.to_dict() for item in self.applied_changes],
"skipped_corrections": [item.to_dict() for item in self.skipped_corrections],
"totals": self.totals,
"work_dir_retention": self.work_dir_retention,
"work_dir_retained": self.work_dir_retained,
"work_dir": self.work_dir,
"error": self.error,
}
def to_json(self) -> str:
return json.dumps(self.to_dict(), ensure_ascii=False, indent=2) + "\n"
@dataclass(frozen=True)
class ProcessResult:
transcript: List[TranscriptSegment]
report: RunReport
run_dir: Path
work_dir_retained: bool

222
src/audita/core/schemas.py Normal file
View File

@@ -0,0 +1,222 @@
import json
import math
from typing import Any, List, Optional
from pydantic import BaseModel, ConfigDict, Field, StrictStr, TypeAdapter, ValidationError, field_validator, model_validator
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:
if not value.strip():
raise ValueError("must not be empty")
return value
@field_validator("start", "end", 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")
if number < 0:
raise ValueError("must be non-negative")
return number
@model_validator(mode="after")
def require_ordered_times(self) -> "TranscriptSegment":
if self.end < self.start:
raise ValueError("end must be greater than or equal to start")
return self
class SourceTranscriptSegment(BaseModel):
model_config = ConfigDict(extra="forbid")
id: Optional[int] = Field(default=None, ge=1)
speaker: StrictStr
start: float
end: float
text: StrictStr
@field_validator("id", mode="before")
@classmethod
def require_optional_integer_id(cls, value: Any) -> Optional[int]:
if value is None:
return None
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:
if not value.strip():
raise ValueError("must not be empty")
return value
@field_validator("start", "end", 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")
if number < 0:
raise ValueError("must be non-negative")
return number
@model_validator(mode="after")
def require_ordered_times(self) -> "SourceTranscriptSegment":
if self.end < self.start:
raise ValueError("end must be greater than or equal to start")
return self
class GlossaryEntry(BaseModel):
model_config = ConfigDict(extra="forbid")
name: StrictStr
aliases: List[StrictStr] = Field(default_factory=list)
plural: Optional[StrictStr] = None
category: StrictStr
summary: StrictStr
@field_validator("name", "category", "summary")
@classmethod
def require_non_empty_text(cls, value: str) -> str:
if not value.strip():
raise ValueError("must not be empty")
return value
@field_validator("aliases")
@classmethod
def require_non_empty_aliases(cls, aliases: List[str]) -> List[str]:
for alias in aliases:
if not alias.strip():
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")
glossary: List[GlossaryEntry]
@field_validator("glossary")
@classmethod
def require_entries(cls, entries: List[GlossaryEntry]) -> List[GlossaryEntry]:
if not entries:
raise ValueError("glossary must contain at least one entry")
return entries
_TRANSCRIPT_ADAPTER = TypeAdapter(List[TranscriptSegment])
_SOURCE_TRANSCRIPT_ADAPTER = TypeAdapter(List[SourceTranscriptSegment])
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:
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 validate_source_transcript_data(data: Any) -> List[SourceTranscriptSegment]:
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 _SOURCE_TRANSCRIPT_ADAPTER.validate_python(data)
except ValidationError as exc:
raise AuditaValidationError(f"Transcript schema validation failed: {exc}") from exc
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, require_sequential_ids=require_sequential_ids)
def parse_source_transcript_json(raw: str) -> List[SourceTranscriptSegment]:
try:
data = json.loads(raw)
except json.JSONDecodeError as exc:
raise AuditaValidationError(f"Transcript is not valid JSON: {exc}") from exc
return validate_source_transcript_data(data)
def parse_glossary_yaml(raw: str) -> Glossary:
try:
import yaml
except ImportError as exc:
raise AuditaValidationError("PyYAML is required to read glossary files.") from exc
try:
data = yaml.safe_load(raw)
except yaml.YAMLError as exc:
raise AuditaValidationError(f"Glossary is not valid YAML: {exc}") from exc
if data is None:
data = {}
try:
return Glossary.model_validate(data)
except ValidationError as exc:
raise AuditaValidationError(f"Glossary schema validation failed: {exc}") from exc
def transcript_to_json(segments: List[TranscriptSegment]) -> str:
payload = [segment.model_dump(mode="json") for segment in segments]
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
def source_transcript_to_json(segments: List[SourceTranscriptSegment]) -> str:
payload = [segment.model_dump(mode="json", exclude_none=True) for segment in segments]
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
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.")

1
src/audita/framework/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Reusable module, filter, review, and runner contracts."""

20
src/audita/framework/llm.py Normal file
View File

@@ -0,0 +1,20 @@
from typing import Any, Sequence
from audita.core.config import AuditaConfig
from audita.core.errors import AuditaLLMError
class OpenRouterStructuredLLMClient:
"""Placeholder for future structured LLM integration."""
def run_structured(
self,
*,
stage_name: str,
messages: Sequence[dict],
response_model: Any,
config: AuditaConfig,
) -> Any:
raise AuditaLLMError(
f"Structured LLM stage '{stage_name}' is not implemented in the framework skeleton."
)

116
src/audita/framework/models.py Normal file
View File

@@ -0,0 +1,116 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Any, List, Optional, Protocol, Sequence
from audita.core.config import AuditaConfig
from audita.core.schemas import Glossary, TranscriptSegment
ReplacementPolicy = str
@dataclass(frozen=True)
class ModuleRunSpec:
instance_name: str
module_key: str
module: "TranscriptModule"
@dataclass(frozen=True)
class CorrectionProposal:
proposal_index: int
module_instance: str
module_key: str
id: int
original_text: str
corrected_text: str
confidence: float
def to_prompt_payload(self) -> dict:
return {
"proposal_index": self.proposal_index,
"id": self.id,
"original_text": self.original_text,
"corrected_text": self.corrected_text,
"confidence": self.confidence,
}
@dataclass(frozen=True)
class FilterDecision:
approved: bool
reason: Optional[str] = None
@dataclass(frozen=True)
class ReviewDecision:
proposal_index: int
approved: bool
confidence: Optional[float] = None
reason: Optional[str] = None
@dataclass(frozen=True)
class ModuleContext:
run_spec: ModuleRunSpec
glossary: Glossary
config: AuditaConfig
run_dir: Path
class StructuredLLMClient(Protocol):
def run_structured(
self,
*,
stage_name: str,
messages: Sequence[dict],
response_model: Any,
config: AuditaConfig,
) -> Any:
...
class DeterministicFilter(Protocol):
name: str
def evaluate(
self,
proposal: CorrectionProposal,
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
) -> FilterDecision:
...
class ReviewStage(Protocol):
name: str
def review(
self,
proposals: Sequence[CorrectionProposal],
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client: Optional[StructuredLLMClient],
run_dir: Path,
) -> Sequence[ReviewDecision]:
...
class TranscriptModule(Protocol):
module_key: str
replacement_policy: ReplacementPolicy
def deterministic_filters(self) -> Sequence[DeterministicFilter]:
...
def review_stages(self) -> Sequence[ReviewStage]:
...
def propose(
self,
transcript_section: Sequence[TranscriptSegment],
context: ModuleContext,
) -> Sequence[CorrectionProposal]:
...

330
src/audita/framework/runner.py Normal file
View File

@@ -0,0 +1,330 @@
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Dict, List, Optional, Sequence, Tuple, Union
from audita.core.chunking import TranscriptSection, chunk_transcript
from audita.core.config import AuditaConfig
from audita.core.reporting import (
AppliedChange,
DeterministicFilterReport,
ModuleRunReport,
ReportedSkip,
ReviewStageReport,
)
from audita.core.schemas import Glossary, TranscriptSegment
from .models import (
CorrectionProposal,
FilterDecision,
ModuleContext,
ModuleRunSpec,
ReviewDecision,
StructuredLLMClient,
)
ProgressCallback = Callable[[str], None]
@dataclass(frozen=True)
class PipelineRunResult:
transcript: List[TranscriptSegment]
module_reports: List[ModuleRunReport]
applied_changes: List[AppliedChange]
skipped_corrections: List[ReportedSkip]
class PipelineRunner:
def run(
self,
*,
transcript: List[TranscriptSegment],
glossary: Glossary,
module_specs: Sequence[ModuleRunSpec],
config: AuditaConfig,
run_dir: Path,
llm_client: Optional[StructuredLLMClient] = None,
progress: Optional[ProgressCallback] = None,
) -> PipelineRunResult:
working = list(transcript)
module_reports: List[ModuleRunReport] = []
applied_changes: List[AppliedChange] = []
skipped_corrections: List[ReportedSkip] = []
for run_spec in module_specs:
module_dir = run_dir / run_spec.instance_name
module_dir.mkdir(parents=True, exist_ok=True)
context = ModuleContext(run_spec=run_spec, glossary=glossary, config=config, run_dir=module_dir)
sections = chunk_transcript(working, config.max_section_tokens)
if progress is not None:
progress(
f"Running module {run_spec.instance_name} "
f"({len(sections)} sections, {len(working)} segments)"
)
result = _run_module(
working=working,
sections=sections,
context=context,
llm_client=llm_client,
)
working = result.transcript
module_reports.append(result.module_report)
applied_changes.extend(result.applied_changes)
skipped_corrections.extend(result.skipped_corrections)
return PipelineRunResult(
transcript=working,
module_reports=module_reports,
applied_changes=applied_changes,
skipped_corrections=skipped_corrections,
)
@dataclass(frozen=True)
class _ModuleExecutionResult:
transcript: List[TranscriptSegment]
module_report: ModuleRunReport
applied_changes: List[AppliedChange]
skipped_corrections: List[ReportedSkip]
def _run_module(
*,
working: List[TranscriptSegment],
sections: Sequence[TranscriptSection],
context: ModuleContext,
llm_client: Optional[StructuredLLMClient],
) -> _ModuleExecutionResult:
module = context.run_spec.module
raw_proposals: List[CorrectionProposal] = []
for section in sections:
proposed = list(module.propose([item.segment for item in section.segments], context))
raw_proposals.extend(proposed)
proposals = [
CorrectionProposal(
proposal_index=index,
module_instance=context.run_spec.instance_name,
module_key=context.run_spec.module_key,
id=proposal.id,
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
)
for index, proposal in enumerate(raw_proposals)
]
surviving = proposals
filter_reports: List[DeterministicFilterReport] = []
skipped: List[ReportedSkip] = []
for deterministic_filter in module.deterministic_filters():
next_survivors: List[CorrectionProposal] = []
rejected_count = 0
for proposal in surviving:
decision = deterministic_filter.evaluate(proposal, working, context.glossary, context.config)
if decision.approved:
next_survivors.append(proposal)
continue
rejected_count += 1
skipped.append(
ReportedSkip(
module_instance=proposal.module_instance,
module_key=proposal.module_key,
proposal_index=proposal.proposal_index,
id=proposal.id,
reason=decision.reason or f"{deterministic_filter.name} rejected proposal",
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
actual_text=_segment_text_by_id(working).get(proposal.id),
source=f"deterministic_filter:{deterministic_filter.name}",
)
)
filter_reports.append(
DeterministicFilterReport(
name=deterministic_filter.name,
passed_count=len(next_survivors),
rejected_count=rejected_count,
)
)
surviving = next_survivors
review_reports: List[ReviewStageReport] = []
for review_stage in module.review_stages():
candidate_count = len(surviving)
if not surviving:
review_reports.append(
ReviewStageReport(
name=review_stage.name,
candidate_count=0,
approved_count=0,
rejected_count=0,
)
)
continue
decisions = list(review_stage.review(surviving, working, context.glossary, context.config, llm_client, context.run_dir))
decisions_by_index = _index_review_decisions(decisions, surviving, review_stage.name)
approved: List[CorrectionProposal] = []
rejected_count = 0
for proposal in surviving:
decision = decisions_by_index[proposal.proposal_index]
if decision.approved:
approved.append(proposal)
continue
rejected_count += 1
skipped.append(
ReportedSkip(
module_instance=proposal.module_instance,
module_key=proposal.module_key,
proposal_index=proposal.proposal_index,
id=proposal.id,
reason=decision.reason or f"{review_stage.name} rejected proposal",
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
actual_text=_segment_text_by_id(working).get(proposal.id),
source=f"review_stage:{review_stage.name}",
)
)
review_reports.append(
ReviewStageReport(
name=review_stage.name,
candidate_count=candidate_count,
approved_count=len(approved),
rejected_count=rejected_count,
)
)
surviving = approved
updated_transcript = list(working)
applied_changes: List[AppliedChange] = []
for proposal in surviving:
apply_result = _apply_proposal(updated_transcript, proposal, module.replacement_policy)
if isinstance(apply_result, ReportedSkip):
skipped.append(apply_result)
continue
updated_transcript, applied_change = apply_result
applied_changes.append(applied_change)
module_report = ModuleRunReport(
instance_name=context.run_spec.instance_name,
module_key=context.run_spec.module_key,
replacement_policy=module.replacement_policy,
section_count=len(sections),
proposal_count=len(proposals),
deterministic_filters=filter_reports,
review_stages=review_reports,
approved_count=len(surviving),
applied_count=len(applied_changes),
skipped_count=len(skipped),
)
return _ModuleExecutionResult(
transcript=updated_transcript,
module_report=module_report,
applied_changes=applied_changes,
skipped_corrections=skipped,
)
def _index_review_decisions(
decisions: Sequence[ReviewDecision],
proposals: Sequence[CorrectionProposal],
stage_name: str,
) -> Dict[int, ReviewDecision]:
expected_indexes = {proposal.proposal_index for proposal in proposals}
indexed: Dict[int, ReviewDecision] = {}
for decision in decisions:
if decision.proposal_index in indexed:
raise ValueError(f"Review stage '{stage_name}' returned duplicate proposal indexes.")
if decision.proposal_index not in expected_indexes:
raise ValueError(f"Review stage '{stage_name}' returned an unknown proposal index.")
indexed[decision.proposal_index] = decision
missing = expected_indexes - set(indexed)
if missing:
raise ValueError(f"Review stage '{stage_name}' omitted proposal indexes: {sorted(missing)}")
return indexed
def _apply_proposal(
transcript: List[TranscriptSegment],
proposal: CorrectionProposal,
replacement_policy: str,
) -> Union[Tuple[List[TranscriptSegment], AppliedChange], ReportedSkip]:
index_by_id = {segment.id: index for index, segment in enumerate(transcript)}
segment_index = index_by_id.get(proposal.id)
if segment_index is None:
return ReportedSkip(
module_instance=proposal.module_instance,
module_key=proposal.module_key,
proposal_index=proposal.proposal_index,
id=proposal.id,
reason="proposal references unknown segment id",
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
source="application",
)
segment = transcript[segment_index]
match_count = segment.text.count(proposal.original_text)
if proposal.original_text == "":
return ReportedSkip(
module_instance=proposal.module_instance,
module_key=proposal.module_key,
proposal_index=proposal.proposal_index,
id=proposal.id,
reason="proposal original_text must not be empty",
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
actual_text=segment.text,
source="application",
)
if match_count == 0:
return ReportedSkip(
module_instance=proposal.module_instance,
module_key=proposal.module_key,
proposal_index=proposal.proposal_index,
id=proposal.id,
reason="proposal original_text does not match segment text",
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
actual_text=segment.text,
source="application",
)
if replacement_policy == "require_unique" and match_count != 1:
return ReportedSkip(
module_instance=proposal.module_instance,
module_key=proposal.module_key,
proposal_index=proposal.proposal_index,
id=proposal.id,
reason="proposal original_text must match exactly once",
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
actual_text=segment.text,
source="application",
)
updated = list(transcript)
updated_text = segment.text.replace(proposal.original_text, proposal.corrected_text)
updated[segment_index] = segment.model_copy(update={"text": updated_text})
return (
updated,
AppliedChange(
module_instance=proposal.module_instance,
module_key=proposal.module_key,
proposal_index=proposal.proposal_index,
id=proposal.id,
original_text=proposal.original_text,
corrected_text=proposal.corrected_text,
confidence=proposal.confidence,
segment_text_before=segment.text,
segment_text_after=updated_text,
),
)
def _segment_text_by_id(transcript: Sequence[TranscriptSegment]) -> Dict[int, str]:
return {segment.id: segment.text for segment in transcript}

16
src/audita/modules/__init__.py Normal file
View File

@@ -0,0 +1,16 @@
from audita.framework.models import ModuleRunSpec
from .glossary import GlossaryModule
from .grammar import GrammarModule
from .homophones import HomophonesModule
from .spoken_word import SpokenWordModule
def default_module_specs() -> list[ModuleRunSpec]:
return [
ModuleRunSpec(instance_name="glossary_primary", module_key="glossary", module=GlossaryModule()),
ModuleRunSpec(instance_name="homophones", module_key="homophones", module=HomophonesModule()),
ModuleRunSpec(instance_name="glossary_secondary", module_key="glossary", module=GlossaryModule()),
ModuleRunSpec(instance_name="spoken_word", module_key="spoken_word", module=SpokenWordModule()),
ModuleRunSpec(instance_name="grammar", module_key="grammar", module=GrammarModule()),
]

66
src/audita/modules/glossary.py Normal file
View File

@@ -0,0 +1,66 @@
from typing import Sequence
from audita.core.config import AuditaConfig
from audita.core.schemas import Glossary, TranscriptSegment
from audita.framework.models import (
CorrectionProposal,
DeterministicFilter,
FilterDecision,
ModuleContext,
ReviewDecision,
ReviewStage,
)
class _StubFilter:
def __init__(self, name: str) -> None:
self.name = name
def evaluate(
self,
proposal: CorrectionProposal,
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
) -> FilterDecision:
return FilterDecision(approved=True)
class _StubReviewStage:
def __init__(self, name: str) -> None:
self.name = name
def review(
self,
proposals: Sequence[CorrectionProposal],
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client,
run_dir,
) -> Sequence[ReviewDecision]:
return [ReviewDecision(proposal_index=proposal.proposal_index, approved=True) for proposal in proposals]
class GlossaryModule:
module_key = "glossary"
replacement_policy = "replace_all"
def deterministic_filters(self) -> Sequence[DeterministicFilter]:
return [
_StubFilter("glossary_direction_guard"),
_StubFilter("protected_glossary_guard"),
]
def review_stages(self) -> Sequence[ReviewStage]:
return [
_StubReviewStage("toward_glossary_term_review"),
_StubReviewStage("context_support_review"),
]
def propose(
self,
transcript_section: Sequence[TranscriptSegment],
context: ModuleContext,
) -> Sequence[CorrectionProposal]:
return []

65
src/audita/modules/grammar.py Normal file
View File

@@ -0,0 +1,65 @@
from typing import Sequence
from audita.core.config import AuditaConfig
from audita.core.schemas import Glossary, TranscriptSegment
from audita.framework.models import (
CorrectionProposal,
DeterministicFilter,
FilterDecision,
ModuleContext,
ReviewDecision,
ReviewStage,
)
class _StubFilter:
def __init__(self, name: str) -> None:
self.name = name
def evaluate(
self,
proposal: CorrectionProposal,
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
) -> FilterDecision:
return FilterDecision(approved=True)
class _StubReviewStage:
def __init__(self, name: str) -> None:
self.name = name
def review(
self,
proposals: Sequence[CorrectionProposal],
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client,
run_dir,
) -> Sequence[ReviewDecision]:
return [ReviewDecision(proposal_index=proposal.proposal_index, approved=True) for proposal in proposals]
class GrammarModule:
module_key = "grammar"
replacement_policy = "require_unique"
def deterministic_filters(self) -> Sequence[DeterministicFilter]:
return [
_StubFilter("protected_glossary_guard"),
_StubFilter("punctuation_capitalization_spacing_guard"),
]
def review_stages(self) -> Sequence[ReviewStage]:
return [
_StubReviewStage("edited_text_readability_review"),
]
def propose(
self,
transcript_section: Sequence[TranscriptSegment],
context: ModuleContext,
) -> Sequence[CorrectionProposal]:
return []

67
src/audita/modules/homophones.py Normal file
View File

@@ -0,0 +1,67 @@
from typing import Sequence
from audita.core.config import AuditaConfig
from audita.core.schemas import Glossary, TranscriptSegment
from audita.framework.models import (
CorrectionProposal,
DeterministicFilter,
FilterDecision,
ModuleContext,
ReviewDecision,
ReviewStage,
)
class _StubFilter:
def __init__(self, name: str) -> None:
self.name = name
def evaluate(
self,
proposal: CorrectionProposal,
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
) -> FilterDecision:
return FilterDecision(approved=True)
class _StubReviewStage:
def __init__(self, name: str) -> None:
self.name = name
def review(
self,
proposals: Sequence[CorrectionProposal],
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client,
run_dir,
) -> Sequence[ReviewDecision]:
return [ReviewDecision(proposal_index=proposal.proposal_index, approved=True) for proposal in proposals]
class HomophonesModule:
module_key = "homophones"
replacement_policy = "require_unique"
def deterministic_filters(self) -> Sequence[DeterministicFilter]:
return [
_StubFilter("protected_glossary_guard"),
_StubFilter("short_span_guard"),
]
def review_stages(self) -> Sequence[ReviewStage]:
return [
_StubReviewStage("acoustic_similarity_review"),
_StubReviewStage("contextual_plausibility_review"),
_StubReviewStage("antonym_reversal_review"),
]
def propose(
self,
transcript_section: Sequence[TranscriptSegment],
context: ModuleContext,
) -> Sequence[CorrectionProposal]:
return []

66
src/audita/modules/spoken_word.py Normal file
View File

@@ -0,0 +1,66 @@
from typing import Sequence
from audita.core.config import AuditaConfig
from audita.core.schemas import Glossary, TranscriptSegment
from audita.framework.models import (
CorrectionProposal,
DeterministicFilter,
FilterDecision,
ModuleContext,
ReviewDecision,
ReviewStage,
)
class _StubFilter:
def __init__(self, name: str) -> None:
self.name = name
def evaluate(
self,
proposal: CorrectionProposal,
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
) -> FilterDecision:
return FilterDecision(approved=True)
class _StubReviewStage:
def __init__(self, name: str) -> None:
self.name = name
def review(
self,
proposals: Sequence[CorrectionProposal],
transcript: Sequence[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client,
run_dir,
) -> Sequence[ReviewDecision]:
return [ReviewDecision(proposal_index=proposal.proposal_index, approved=True) for proposal in proposals]
class SpokenWordModule:
module_key = "spoken_word"
replacement_policy = "replace_all"
def deterministic_filters(self) -> Sequence[DeterministicFilter]:
return [
_StubFilter("protected_glossary_guard"),
_StubFilter("spoken_disfluency_guard"),
]
def review_stages(self) -> Sequence[ReviewStage]:
return [
_StubReviewStage("spoken_marker_cleanup_review"),
_StubReviewStage("meaning_preservation_review"),
]
def propose(
self,
transcript_section: Sequence[TranscriptSegment],
context: ModuleContext,
) -> Sequence[CorrectionProposal]:
return []

650
src/audita/pipeline.py
View File

@@ -1,83 +1,42 @@
import json
import shutil
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Callable, Dict, List, Optional, Tuple
from typing import Callable, List, Optional
from uuid import uuid4
from .chunking import IndexedSegment, TranscriptSection, chunk_indexed_segments
from .config import AuditaConfig
from .corrections import CorrectionGuard, ReplacementMode, SkippedCorrection, apply_corrections
from .errors import AuditaError
from .normalization import NormalizationResult, normalize_transcript
from .passes import CorrectionPass, GlossaryCorrectionPass, GrammarCorrectionPass, LLMClient
from .prompts import build_grammar_spoken_form_validation_messages, build_grammar_validation_messages
from .protection import ProtectedVocabulary
from .reporting import AppliedChange, ProcessResult, ReportedSkippedCorrection, RunReport
from .semantic_validation import (
filter_with_meaning_preserving_validations,
filter_with_spoken_form_validations,
keep_corrections_with_indexes,
select_grammar_validation_candidates,
)
from .schemas import CorrectionCandidate, Glossary, SourceTranscriptSegment, TranscriptSegment, parse_transcript_json
from .core.config import AuditaConfig
from .core.errors import AuditaError
from .core.normalization import normalize_transcript
from .core.reporting import ProcessResult, RunReport
from .core.schemas import Glossary, SourceTranscriptSegment, TranscriptSegment, source_transcript_to_json, transcript_to_json
from .framework.runner import PipelineRunner
from .modules import default_module_specs
ProgressCallback = Callable[[str], None]
@dataclass(frozen=True)
class StageSpec:
name: str
correction_pass: CorrectionPass
max_llm_passes: int
confidence_threshold: float
replacement_mode: ReplacementMode
correction_guard: Optional[CorrectionGuard] = None
protected_vocabulary: Optional[ProtectedVocabulary] = None
@dataclass(frozen=True)
class StageRunResult:
transcript: List[TranscriptSegment]
applied_changes: List[AppliedChange]
skipped_corrections: List[ReportedSkippedCorrection]
def process_transcript(
transcript: List[SourceTranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client: Optional[LLMClient] = None,
progress: Optional[ProgressCallback] = None,
) -> List[TranscriptSegment]:
return process_transcript_result(
transcript,
glossary,
config,
llm_client=llm_client,
progress=progress,
).transcript
return process_transcript_result(transcript, glossary, config, progress=progress).transcript
def process_transcript_result(
transcript: List[SourceTranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client: Optional[LLMClient] = None,
progress: Optional[ProgressCallback] = None,
) -> ProcessResult:
run_dir = _create_run_dir(config.work_dir)
stage_summaries: List[dict] = []
normalization_summary: Optional[dict] = None
applied_changes: List[AppliedChange] = []
final_skipped: List[ReportedSkippedCorrection] = []
working: List[TranscriptSegment] = []
normalized_transcript: List[TranscriptSegment] = []
report: Optional[RunReport] = None
try:
_log(progress, f"Created work directory {run_dir}")
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
normalization_result = normalize_transcript(
transcript,
max_segment_gap=config.normalize_max_segment_gap,
@@ -85,7 +44,7 @@ def process_transcript_result(
max_segment_duration=config.normalize_max_segment_duration,
max_segment_tokens=config.normalize_max_segment_tokens,
)
normalization_summary = normalization_result.summary.to_dict()
normalized_transcript = list(normalization_result.transcript)
_write_normalization_diagnostics(run_dir, transcript, normalization_result)
_log(
progress,
@@ -93,532 +52,113 @@ def process_transcript_result(
f"{normalization_result.summary.source_segment_count} to "
f"{normalization_result.summary.normalized_segment_count} segments",
)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
if llm_client is None:
from .llm import InstructorLLMClient
llm_client = InstructorLLMClient(config)
working = list(normalization_result.transcript)
protected_vocabulary = ProtectedVocabulary.from_glossary(glossary)
stages = [
StageSpec(
name="glossary",
correction_pass=GlossaryCorrectionPass(llm_client),
max_llm_passes=config.glossary_max_llm_passes,
confidence_threshold=config.glossary_confidence_threshold,
replacement_mode="replace_all",
correction_guard=protected_vocabulary.violation_reason,
protected_vocabulary=protected_vocabulary,
),
StageSpec(
name="grammar",
correction_pass=GrammarCorrectionPass(llm_client),
max_llm_passes=config.grammar_max_llm_passes,
confidence_threshold=config.grammar_confidence_threshold,
replacement_mode="require_unique",
correction_guard=protected_vocabulary.violation_reason,
protected_vocabulary=protected_vocabulary,
),
]
for stage in stages:
stage_dir = run_dir / stage.name
stage_dir.mkdir()
stage_summary = {
"stage": stage.name,
"max_llm_passes": stage.max_llm_passes,
"confidence_threshold": stage.confidence_threshold,
"replacement_mode": stage.replacement_mode,
"passes": [],
}
stage_summaries.append(stage_summary)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
stage_result = _run_correction_stage(
working,
glossary,
config,
stage,
run_dir,
normalization_summary,
stage_dir,
stage_summaries,
stage_summary["passes"],
llm_client,
progress,
)
working = stage_result.transcript
applied_changes.extend(stage_result.applied_changes)
final_skipped.extend(stage_result.skipped_corrections)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
_write_skipped_corrections(run_dir, final_skipped)
for skipped in final_skipped:
_log(
progress,
f"Skipping {skipped.stage} correction for id {skipped.id}: {skipped.reason}",
)
revised = _sort_transcript_chronologically(working)
except Exception as exc:
_write_skipped_corrections(run_dir, final_skipped)
report = _build_run_report(
module_specs = default_module_specs()
pipeline_runner = PipelineRunner()
pipeline_result = pipeline_runner.run(
transcript=normalized_transcript,
glossary=glossary,
module_specs=module_specs,
config=config,
normalization_summary=normalization_summary,
stage_summaries=stage_summaries,
applied_changes=applied_changes,
skipped_corrections=final_skipped,
run_dir=run_dir,
llm_client=None,
progress=progress,
)
revised = _sort_transcript_chronologically(pipeline_result.transcript)
work_dir_retained = _should_retain_run_dir(config.work_dir_retention, bool(pipeline_result.skipped_corrections))
report = RunReport(
status="success",
config=config.to_report_dict(),
normalization=normalization_result.summary.to_dict(),
pipeline=[spec.instance_name for spec in module_specs],
modules=pipeline_result.module_reports,
applied_changes=pipeline_result.applied_changes,
skipped_corrections=pipeline_result.skipped_corrections,
totals={
"output_segment_count": len(revised),
"applied_change_count": len(pipeline_result.applied_changes),
"skipped_correction_count": len(pipeline_result.skipped_corrections),
},
work_dir_retention=config.work_dir_retention,
work_dir_retained=work_dir_retained,
work_dir=str(run_dir) if work_dir_retained else None,
error=None,
)
_write_run_report(run_dir / "report.json", report)
if work_dir_retained:
_log(progress, f"Work directory preserved at {run_dir}")
else:
shutil.rmtree(run_dir)
_log(progress, "Removed work directory after successful run")
return ProcessResult(
transcript=revised,
report=report,
run_dir=run_dir,
work_dir_retained=work_dir_retained,
)
except Exception as exc:
report = RunReport(
status="failed",
config=config.to_report_dict(),
normalization=None,
pipeline=[spec.instance_name for spec in default_module_specs()],
modules=[],
applied_changes=[],
skipped_corrections=[],
totals={
"output_segment_count": len(normalized_transcript),
"applied_change_count": 0,
"skipped_correction_count": 0,
},
work_dir_retention=config.work_dir_retention,
work_dir_retained=True,
run_dir=run_dir,
transcript=working,
status="failed",
work_dir=str(run_dir),
error=str(exc),
)
_write_run_report(run_dir / "report.json", report)
message = f"{exc} Diagnostics preserved at {run_dir}"
if isinstance(exc, AuditaError):
raise type(exc)(message) from exc
raise AuditaError(message) from exc
work_dir_retained = _should_retain_run_dir(config.work_dir_retention, bool(final_skipped))
report = _build_run_report(
config=config,
normalization_summary=normalization_summary,
stage_summaries=stage_summaries,
applied_changes=applied_changes,
skipped_corrections=final_skipped,
work_dir_retention=config.work_dir_retention,
work_dir_retained=work_dir_retained,
run_dir=run_dir,
transcript=revised,
status="success",
error=None,
)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=work_dir_retained)
if work_dir_retained:
_write_run_report(run_dir / "report.json", report)
if final_skipped:
_log(progress, f"Skipped correction diagnostics preserved at {run_dir}")
else:
_log(progress, f"Work directory preserved at {run_dir}")
else:
shutil.rmtree(run_dir)
_log(progress, "Removed work directory after successful run")
return ProcessResult(
transcript=revised,
report=report,
run_dir=run_dir,
work_dir_retained=work_dir_retained,
)
raise type(exc)(f"{exc} Diagnostics preserved at {run_dir}") from exc
raise AuditaError(f"{exc} Diagnostics preserved at {run_dir}") from exc
def _run_correction_stage(
transcript: List[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
stage: StageSpec,
run_dir: Path,
normalization_summary: Optional[dict],
stage_dir: Path,
stage_summaries: List[dict],
pass_summaries: List[dict],
llm_client: LLMClient,
progress: Optional[ProgressCallback],
) -> StageRunResult:
working = list(transcript)
stage_applied_changes: List[AppliedChange] = []
unresolved_retry_skips: Dict[int, ReportedSkippedCorrection] = {}
final_nonretry_skips: List[ReportedSkippedCorrection] = []
for pass_number in range(1, stage.max_llm_passes + 1):
if pass_number == 1:
indexed_segments = _indexed_segments_for_ids(working, [segment.id for segment in working])
else:
retry_ids = sorted(unresolved_retry_skips)
if not retry_ids:
break
indexed_segments = _indexed_segments_for_ids(working, retry_ids)
if not indexed_segments:
break
pass_dir = stage_dir / f"pass-{pass_number:04d}"
pass_dir.mkdir()
sections = chunk_indexed_segments(indexed_segments, config.max_section_tokens)
corrections = []
for section in sections:
_write_and_validate_section(pass_dir, section)
_log(
progress,
f"Processing {stage.name} pass {pass_number}/{stage.max_llm_passes} "
f"section {section.section_index + 1}/{len(sections)} "
f"({len(section.segments)} segments, estimated {section.token_count} tokens)",
)
corrections.extend(
stage.correction_pass.run(
section,
glossary,
config,
pass_dir,
retry_pass=pass_number > 1,
)
)
corrections_for_application, validation_skips, validation_summary = _validate_grammar_corrections(
working,
corrections,
config,
stage,
pass_dir,
llm_client,
)
final_nonretry_skips.extend(
_reported_skip(stage.name, pass_number, skipped) for skipped in validation_skips
)
application_result = apply_corrections(
working,
corrections_for_application,
stage.confidence_threshold,
replacement_mode=stage.replacement_mode,
correction_guard=stage.correction_guard,
)
working = application_result.transcript
stage_applied_changes.extend(
AppliedChange(
stage=stage.name,
pass_number=pass_number,
id=applied.id,
original_text=applied.original_text,
corrected_text=applied.corrected_text,
confidence=applied.confidence,
segment_text_before=applied.segment_text_before,
segment_text_after=applied.segment_text_after,
)
for applied in application_result.applied_corrections
)
next_retry_skips: Dict[int, ReportedSkippedCorrection] = {}
for ignored in application_result.ignored:
reported_ignored = _reported_skip(stage.name, pass_number, ignored)
if _is_retryable_skip(reported_ignored, working):
next_retry_skips[reported_ignored.id] = reported_ignored
else:
final_nonretry_skips.append(reported_ignored)
for skipped in application_result.skipped:
reported_skip = _reported_skip(stage.name, pass_number, skipped)
if _is_retryable_skip(reported_skip, working):
next_retry_skips[reported_skip.id] = reported_skip
else:
final_nonretry_skips.append(reported_skip)
unresolved_retry_skips = next_retry_skips
pass_summaries.append(
{
"pass_number": pass_number,
"retry_pass": pass_number > 1,
"section_count": len(sections),
"segment_count": len(indexed_segments),
"corrections_returned": len(corrections),
"applied_count": len(application_result.applied_corrections),
"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, work_dir_retained=True)
if not unresolved_retry_skips:
break
final_skipped = final_nonretry_skips + [
unresolved_retry_skips[correction_id] for correction_id in sorted(unresolved_retry_skips)
]
return StageRunResult(
transcript=working,
applied_changes=stage_applied_changes,
skipped_corrections=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,
"spoken_form_validation_candidate_count": 0,
"spoken_form_validation_approved_count": 0,
"spoken_form_validation_rejected_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")
meaning_result = filter_with_meaning_preserving_validations(
candidates,
response,
config.grammar_validation_confidence_threshold,
)
validation_summary["validation_approved_count"] = meaning_result.approved_count
validation_summary["validation_rejected_count"] = meaning_result.rejected_count
validation_summary["spoken_form_validation_candidate_count"] = len(meaning_result.rescue_candidates)
candidate_indexes = {candidate.correction_index for candidate in candidates}
allowed_indexes = set(range(len(corrections))) - candidate_indexes
allowed_indexes.update(meaning_result.approved_correction_indexes)
if not meaning_result.rescue_candidates:
return keep_corrections_with_indexes(corrections, allowed_indexes), [], validation_summary
spoken_form_payload = [candidate.to_prompt_payload() for candidate in meaning_result.rescue_candidates]
spoken_form_messages = build_grammar_spoken_form_validation_messages(spoken_form_payload)
spoken_form_prompt_path = pass_dir / "spoken-form-validation-prompt-0000.json"
spoken_form_prompt_path.write_text(
json.dumps(spoken_form_messages, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
spoken_form_response = llm_client.create_grammar_spoken_form_validations(spoken_form_messages, config)
spoken_form_response_path = pass_dir / "spoken-form-validation-response-0000.json"
spoken_form_response_path.write_text(spoken_form_response.model_dump_json(indent=2) + "\n", encoding="utf-8")
spoken_form_result = filter_with_spoken_form_validations(
meaning_result.rescue_candidates,
spoken_form_response,
config.grammar_spoken_form_validation_confidence_threshold,
)
validation_summary["spoken_form_validation_approved_count"] = spoken_form_result.approved_count
validation_summary["spoken_form_validation_rejected_count"] = spoken_form_result.rejected_count
allowed_indexes.update(spoken_form_result.approved_correction_indexes)
return (
keep_corrections_with_indexes(corrections, allowed_indexes),
spoken_form_result.skipped,
validation_summary,
)
def _create_run_dir(work_dir: Path) -> Path:
work_dir.mkdir(parents=True, exist_ok=True)
def _create_run_dir(root: Path) -> Path:
root.mkdir(parents=True, exist_ok=True)
timestamp = datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
run_dir = work_dir / f"run-{timestamp}-{uuid4().hex[:8]}"
run_dir.mkdir()
run_dir = root / f"run-{timestamp}-{uuid4().hex[:8]}"
run_dir.mkdir(parents=False, exist_ok=False)
return run_dir
def _write_run_metadata(
run_dir: Path,
config: AuditaConfig,
normalization_summary: Optional[dict],
stage_summaries: List[dict],
work_dir_retained: bool,
) -> None:
metadata = {
**_config_summary(config),
"normalization": normalization_summary,
"stages": stage_summaries,
"work_dir_retained": work_dir_retained,
}
(run_dir / "metadata.json").write_text(
json.dumps(metadata, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def _should_retain_run_dir(retention: str, has_skips: bool) -> bool:
if retention == "always":
return True
if retention == "never":
return False
return has_skips
def _write_normalization_diagnostics(
run_dir: Path,
source: List[SourceTranscriptSegment],
result: NormalizationResult,
) -> None:
def _sort_transcript_chronologically(transcript: List[TranscriptSegment]) -> List[TranscriptSegment]:
return list(sorted(transcript, key=lambda segment: (segment.start, segment.end, segment.id)))
def _write_normalization_diagnostics(run_dir: Path, transcript, normalization_result) -> None:
normalization_dir = run_dir / "normalization"
normalization_dir.mkdir()
(normalization_dir / "source-transcript.json").write_text(
_segments_to_json(source),
encoding="utf-8",
)
normalization_dir.mkdir(parents=True, exist_ok=True)
(normalization_dir / "source-transcript.json").write_text(source_transcript_to_json(transcript), encoding="utf-8")
(normalization_dir / "normalized-transcript.json").write_text(
_segments_to_json(result.transcript),
transcript_to_json(normalization_result.transcript),
encoding="utf-8",
)
(normalization_dir / "summary.json").write_text(
json.dumps(result.summary.to_dict(), ensure_ascii=False, indent=2) + "\n",
json.dumps(normalization_result.summary.to_dict(), indent=2) + "\n",
encoding="utf-8",
)
def _segments_to_json(segments: List[SourceTranscriptSegment]) -> str:
payload = [segment.model_dump(mode="json", exclude_none=True) for segment in segments]
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
def _write_and_validate_section(run_dir: Path, section: TranscriptSection) -> None:
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, require_sequential_ids=False)
def _write_skipped_corrections(run_dir: Path, skipped: List[ReportedSkippedCorrection]) -> None:
skipped_path = run_dir / "skipped-corrections.json"
skipped_path.write_text(
json.dumps(
{
"skipped_corrections": [item.to_dict() for item in skipped]
},
ensure_ascii=False,
indent=2,
)
+ "\n",
encoding="utf-8",
)
def _indexed_segments_for_ids(
transcript: List[TranscriptSegment],
ids: List[int],
) -> List[IndexedSegment]:
id_to_position = {segment.id: position for position, segment in enumerate(transcript)}
return [
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: ReportedSkippedCorrection, transcript: List[TranscriptSegment]) -> bool:
return any(segment.id == skipped.id for segment in transcript)
def _sort_transcript_chronologically(
transcript: List[TranscriptSegment],
) -> List[TranscriptSegment]:
indexed = list(enumerate(transcript))
indexed.sort(key=lambda item: (item[1].start, item[1].end, item[0]))
return [segment for _, segment in indexed]
def _log(progress: Optional[ProgressCallback], message: str) -> None:
if progress is not None:
progress(message)
def _reported_skip(stage: str, pass_number: int, skipped: SkippedCorrection) -> ReportedSkippedCorrection:
return ReportedSkippedCorrection(
stage=stage,
pass_number=pass_number,
id=skipped.id,
reason=skipped.reason,
original_text=skipped.original_text,
corrected_text=skipped.corrected_text,
confidence=skipped.confidence,
actual_text=skipped.actual_text,
validation_confidence=skipped.validation_confidence,
validation_reason=skipped.validation_reason,
)
def _should_retain_run_dir(work_dir_retention: str, has_final_skipped: bool) -> bool:
if work_dir_retention == "always":
return True
if work_dir_retention == "never":
return False
return has_final_skipped
def _config_summary(config: AuditaConfig) -> dict:
return {
"model": config.model,
"base_url": config.base_url,
"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,
"grammar_spoken_form_validation_confidence_threshold": (
config.grammar_spoken_form_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,
"normalize_max_segment_gap": config.normalize_max_segment_gap,
"normalize_ellipsis_gap": config.normalize_ellipsis_gap,
"normalize_max_segment_duration": config.normalize_max_segment_duration,
"normalize_max_segment_tokens": config.normalize_max_segment_tokens,
"work_dir_retention": config.work_dir_retention,
}
def _build_run_report(
config: AuditaConfig,
normalization_summary: Optional[dict],
stage_summaries: List[dict],
applied_changes: List[AppliedChange],
skipped_corrections: List[ReportedSkippedCorrection],
work_dir_retention: str,
work_dir_retained: bool,
run_dir: Path,
transcript: List[TranscriptSegment],
status: str,
error: Optional[str],
) -> RunReport:
totals = {
"output_segment_count": len(transcript),
"applied_change_count": len(applied_changes),
"skipped_correction_count": len(skipped_corrections),
}
if normalization_summary is not None:
totals["source_segment_count"] = normalization_summary["source_segment_count"]
totals["normalized_segment_count"] = normalization_summary["normalized_segment_count"]
return RunReport(
status=status,
config=_config_summary(config),
normalization=normalization_summary,
stages=stage_summaries,
applied_changes=applied_changes,
skipped_corrections=skipped_corrections,
totals=totals,
work_dir_retention=work_dir_retention,
work_dir_retained=work_dir_retained,
work_dir=str(run_dir) if work_dir_retained else None,
error=error,
)
def _write_run_report(path: Path, report: RunReport) -> None:
path.write_text(report.to_json(), encoding="utf-8")
def _log(progress: Optional[ProgressCallback], message: str) -> None:
if progress is not None:
progress(message)

6
src/audita_prototype/__init__.py Normal file
View File

@@ -0,0 +1,6 @@
"""Audita transcript correction package."""
__all__ = ["__version__"]
__version__ = "0.1.0"

6
src/audita_prototype/__main__.py Normal file
View File

@@ -0,0 +1,6 @@
from .cli import main
if __name__ == "__main__":
raise SystemExit(main())

0
src/audita/chunking.py → src/audita_prototype/chunking.py
View File

137
src/audita_prototype/cli.py Normal file
View File

@@ -0,0 +1,137 @@
import argparse
import sys
from pathlib import Path
from typing import Optional, Sequence
from .config import AuditaConfig, ConfigOverrides
from .errors import AuditaError
from .io import load_glossary, load_transcript, write_report, write_transcript
from .pipeline import process_transcript_result
from .schemas import transcript_to_json
def main(argv: Optional[Sequence[str]] = None) -> int:
parser = _build_parser()
args = parser.parse_args(argv)
if args.command == "process":
return _process(args)
parser.print_help(sys.stderr)
return 2
def _build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(prog="audita")
subparsers = parser.add_subparsers(dest="command", required=True)
process = subparsers.add_parser("process", help="correct a transcript using a glossary")
process.add_argument("transcript", type=Path, help="path to the input transcript JSON")
process.add_argument("--glossary", type=Path, required=True, help="path to the glossary YAML")
process.add_argument("--output", type=Path, help="write corrected transcript JSON to this path")
process.add_argument("--report-json", type=Path, help="write structured run report JSON to this path")
process.add_argument("--model", help="OpenRouter model to use")
process.add_argument("--base-url", help="OpenAI-compatible API base URL")
process.add_argument("--max-section-tokens", type=int, help="maximum estimated tokens per transcript section")
process.add_argument(
"--glossary-confidence-threshold",
type=float,
help="minimum confidence required to apply a glossary correction",
)
process.add_argument(
"--grammar-confidence-threshold",
type=float,
help="minimum confidence required to apply a grammar correction",
)
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(
"--grammar-spoken-form-validation-confidence-threshold",
type=float,
help="minimum validator confidence required to apply a spoken-form rescue correction",
)
process.add_argument(
"--normalize-max-segment-gap",
type=float,
help="maximum same-speaker gap in seconds eligible for deterministic merging",
)
process.add_argument(
"--normalize-ellipsis-gap",
type=float,
help="minimum same-speaker gap in seconds that uses an ellipsis joiner",
)
process.add_argument(
"--normalize-max-segment-duration",
type=float,
help="maximum merged segment duration in seconds",
)
process.add_argument(
"--normalize-max-segment-tokens",
type=int,
help="maximum estimated tokens for a merged segment prompt payload",
)
process.add_argument("--work-dir", type=Path, help="directory for per-run scratch diagnostics")
process.add_argument(
"--work-dir-retention",
choices=("auto", "always", "never"),
help="whether to retain the per-run work directory",
)
return parser
def _process(args: argparse.Namespace) -> int:
try:
config = AuditaConfig.from_sources(
overrides=ConfigOverrides(
model=args.model,
base_url=args.base_url,
max_section_tokens=args.max_section_tokens,
glossary_confidence_threshold=args.glossary_confidence_threshold,
grammar_confidence_threshold=args.grammar_confidence_threshold,
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,
grammar_spoken_form_validation_confidence_threshold=(
args.grammar_spoken_form_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,
normalize_max_segment_tokens=args.normalize_max_segment_tokens,
work_dir=args.work_dir,
work_dir_retention=args.work_dir_retention,
)
)
transcript = load_transcript(args.transcript)
glossary = load_glossary(args.glossary)
result = process_transcript_result(
transcript,
glossary,
config,
progress=lambda message: print(message, file=sys.stderr),
)
if args.output is not None:
write_transcript(args.output, result.transcript)
else:
sys.stdout.write(transcript_to_json(result.transcript))
if args.report_json is not None:
write_report(args.report_json, result.report)
return 0
except AuditaError as exc:
print(f"audita: error: {exc}", file=sys.stderr)
return 1

0
src/audita/config.py → src/audita_prototype/config.py
View File

0
src/audita/corrections.py → src/audita_prototype/corrections.py
View File

15
src/audita_prototype/errors.py Normal file
View File

@@ -0,0 +1,15 @@
class AuditaError(Exception):
"""Base exception for user-facing Audita failures."""
class AuditaValidationError(AuditaError):
"""Raised when input data does not match Audita's expected schema."""
class AuditaConfigError(AuditaError):
"""Raised when runtime configuration is invalid or incomplete."""
class AuditaLLMError(AuditaError):
"""Raised when an LLM request or structured response fails."""

22
src/audita_prototype/io.py Normal file
View File

@@ -0,0 +1,22 @@
from pathlib import Path
from typing import List
from .reporting import RunReport
from .schemas import Glossary, SourceTranscriptSegment, TranscriptSegment
from .schemas import parse_glossary_yaml, parse_source_transcript_json, transcript_to_json
def load_transcript(path: Path) -> List[SourceTranscriptSegment]:
return parse_source_transcript_json(path.read_text(encoding="utf-8"))
def load_glossary(path: Path) -> Glossary:
return parse_glossary_yaml(path.read_text(encoding="utf-8"))
def write_transcript(path: Path, segments: List[TranscriptSegment]) -> None:
path.write_text(transcript_to_json(segments), encoding="utf-8")
def write_report(path: Path, report: RunReport) -> None:
path.write_text(report.to_json(), encoding="utf-8")

0
src/audita/llm.py → src/audita_prototype/llm.py
View File

0
src/audita/normalization.py → src/audita_prototype/normalization.py
View File

0
src/audita/passes.py → src/audita_prototype/passes.py
View File

624
src/audita_prototype/pipeline.py Normal file
View File

@@ -0,0 +1,624 @@
import json
import shutil
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import Callable, Dict, List, Optional, Tuple
from uuid import uuid4
from .chunking import IndexedSegment, TranscriptSection, chunk_indexed_segments
from .config import AuditaConfig
from .corrections import CorrectionGuard, ReplacementMode, SkippedCorrection, apply_corrections
from .errors import AuditaError
from .normalization import NormalizationResult, normalize_transcript
from .passes import CorrectionPass, GlossaryCorrectionPass, GrammarCorrectionPass, LLMClient
from .prompts import build_grammar_spoken_form_validation_messages, build_grammar_validation_messages
from .protection import ProtectedVocabulary
from .reporting import AppliedChange, ProcessResult, ReportedSkippedCorrection, RunReport
from .semantic_validation import (
filter_with_meaning_preserving_validations,
filter_with_spoken_form_validations,
keep_corrections_with_indexes,
select_grammar_validation_candidates,
)
from .schemas import CorrectionCandidate, Glossary, SourceTranscriptSegment, TranscriptSegment, parse_transcript_json
ProgressCallback = Callable[[str], None]
@dataclass(frozen=True)
class StageSpec:
name: str
correction_pass: CorrectionPass
max_llm_passes: int
confidence_threshold: float
replacement_mode: ReplacementMode
correction_guard: Optional[CorrectionGuard] = None
protected_vocabulary: Optional[ProtectedVocabulary] = None
@dataclass(frozen=True)
class StageRunResult:
transcript: List[TranscriptSegment]
applied_changes: List[AppliedChange]
skipped_corrections: List[ReportedSkippedCorrection]
def process_transcript(
transcript: List[SourceTranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client: Optional[LLMClient] = None,
progress: Optional[ProgressCallback] = None,
) -> List[TranscriptSegment]:
return process_transcript_result(
transcript,
glossary,
config,
llm_client=llm_client,
progress=progress,
).transcript
def process_transcript_result(
transcript: List[SourceTranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
llm_client: Optional[LLMClient] = None,
progress: Optional[ProgressCallback] = None,
) -> ProcessResult:
run_dir = _create_run_dir(config.work_dir)
stage_summaries: List[dict] = []
normalization_summary: Optional[dict] = None
applied_changes: List[AppliedChange] = []
final_skipped: List[ReportedSkippedCorrection] = []
working: List[TranscriptSegment] = []
try:
_log(progress, f"Created work directory {run_dir}")
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
normalization_result = normalize_transcript(
transcript,
max_segment_gap=config.normalize_max_segment_gap,
ellipsis_gap=config.normalize_ellipsis_gap,
max_segment_duration=config.normalize_max_segment_duration,
max_segment_tokens=config.normalize_max_segment_tokens,
)
normalization_summary = normalization_result.summary.to_dict()
_write_normalization_diagnostics(run_dir, transcript, normalization_result)
_log(
progress,
"Normalized transcript from "
f"{normalization_result.summary.source_segment_count} to "
f"{normalization_result.summary.normalized_segment_count} segments",
)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
if llm_client is None:
from .llm import InstructorLLMClient
llm_client = InstructorLLMClient(config)
working = list(normalization_result.transcript)
protected_vocabulary = ProtectedVocabulary.from_glossary(glossary)
stages = [
StageSpec(
name="glossary",
correction_pass=GlossaryCorrectionPass(llm_client),
max_llm_passes=config.glossary_max_llm_passes,
confidence_threshold=config.glossary_confidence_threshold,
replacement_mode="replace_all",
correction_guard=protected_vocabulary.violation_reason,
protected_vocabulary=protected_vocabulary,
),
StageSpec(
name="grammar",
correction_pass=GrammarCorrectionPass(llm_client),
max_llm_passes=config.grammar_max_llm_passes,
confidence_threshold=config.grammar_confidence_threshold,
replacement_mode="require_unique",
correction_guard=protected_vocabulary.violation_reason,
protected_vocabulary=protected_vocabulary,
),
]
for stage in stages:
stage_dir = run_dir / stage.name
stage_dir.mkdir()
stage_summary = {
"stage": stage.name,
"max_llm_passes": stage.max_llm_passes,
"confidence_threshold": stage.confidence_threshold,
"replacement_mode": stage.replacement_mode,
"passes": [],
}
stage_summaries.append(stage_summary)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
stage_result = _run_correction_stage(
working,
glossary,
config,
stage,
run_dir,
normalization_summary,
stage_dir,
stage_summaries,
stage_summary["passes"],
llm_client,
progress,
)
working = stage_result.transcript
applied_changes.extend(stage_result.applied_changes)
final_skipped.extend(stage_result.skipped_corrections)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=True)
_write_skipped_corrections(run_dir, final_skipped)
for skipped in final_skipped:
_log(
progress,
f"Skipping {skipped.stage} correction for id {skipped.id}: {skipped.reason}",
)
revised = _sort_transcript_chronologically(working)
except Exception as exc:
_write_skipped_corrections(run_dir, final_skipped)
report = _build_run_report(
config=config,
normalization_summary=normalization_summary,
stage_summaries=stage_summaries,
applied_changes=applied_changes,
skipped_corrections=final_skipped,
work_dir_retention=config.work_dir_retention,
work_dir_retained=True,
run_dir=run_dir,
transcript=working,
status="failed",
error=str(exc),
)
_write_run_report(run_dir / "report.json", report)
message = f"{exc} Diagnostics preserved at {run_dir}"
if isinstance(exc, AuditaError):
raise type(exc)(message) from exc
raise AuditaError(message) from exc
work_dir_retained = _should_retain_run_dir(config.work_dir_retention, bool(final_skipped))
report = _build_run_report(
config=config,
normalization_summary=normalization_summary,
stage_summaries=stage_summaries,
applied_changes=applied_changes,
skipped_corrections=final_skipped,
work_dir_retention=config.work_dir_retention,
work_dir_retained=work_dir_retained,
run_dir=run_dir,
transcript=revised,
status="success",
error=None,
)
_write_run_metadata(run_dir, config, normalization_summary, stage_summaries, work_dir_retained=work_dir_retained)
if work_dir_retained:
_write_run_report(run_dir / "report.json", report)
if final_skipped:
_log(progress, f"Skipped correction diagnostics preserved at {run_dir}")
else:
_log(progress, f"Work directory preserved at {run_dir}")
else:
shutil.rmtree(run_dir)
_log(progress, "Removed work directory after successful run")
return ProcessResult(
transcript=revised,
report=report,
run_dir=run_dir,
work_dir_retained=work_dir_retained,
)
def _run_correction_stage(
transcript: List[TranscriptSegment],
glossary: Glossary,
config: AuditaConfig,
stage: StageSpec,
run_dir: Path,
normalization_summary: Optional[dict],
stage_dir: Path,
stage_summaries: List[dict],
pass_summaries: List[dict],
llm_client: LLMClient,
progress: Optional[ProgressCallback],
) -> StageRunResult:
working = list(transcript)
stage_applied_changes: List[AppliedChange] = []
unresolved_retry_skips: Dict[int, ReportedSkippedCorrection] = {}
final_nonretry_skips: List[ReportedSkippedCorrection] = []
for pass_number in range(1, stage.max_llm_passes + 1):
if pass_number == 1:
indexed_segments = _indexed_segments_for_ids(working, [segment.id for segment in working])
else:
retry_ids = sorted(unresolved_retry_skips)
if not retry_ids:
break
indexed_segments = _indexed_segments_for_ids(working, retry_ids)
if not indexed_segments:
break
pass_dir = stage_dir / f"pass-{pass_number:04d}"
pass_dir.mkdir()
sections = chunk_indexed_segments(indexed_segments, config.max_section_tokens)
corrections = []
for section in sections:
_write_and_validate_section(pass_dir, section)
_log(
progress,
f"Processing {stage.name} pass {pass_number}/{stage.max_llm_passes} "
f"section {section.section_index + 1}/{len(sections)} "
f"({len(section.segments)} segments, estimated {section.token_count} tokens)",
)
corrections.extend(
stage.correction_pass.run(
section,
glossary,
config,
pass_dir,
retry_pass=pass_number > 1,
)
)
corrections_for_application, validation_skips, validation_summary = _validate_grammar_corrections(
working,
corrections,
config,
stage,
pass_dir,
llm_client,
)
final_nonretry_skips.extend(
_reported_skip(stage.name, pass_number, skipped) for skipped in validation_skips
)
application_result = apply_corrections(
working,
corrections_for_application,
stage.confidence_threshold,
replacement_mode=stage.replacement_mode,
correction_guard=stage.correction_guard,
)
working = application_result.transcript
stage_applied_changes.extend(
AppliedChange(
stage=stage.name,
pass_number=pass_number,
id=applied.id,
original_text=applied.original_text,
corrected_text=applied.corrected_text,
confidence=applied.confidence,
segment_text_before=applied.segment_text_before,
segment_text_after=applied.segment_text_after,
)
for applied in application_result.applied_corrections
)
next_retry_skips: Dict[int, ReportedSkippedCorrection] = {}
for ignored in application_result.ignored:
reported_ignored = _reported_skip(stage.name, pass_number, ignored)
if _is_retryable_skip(reported_ignored, working):
next_retry_skips[reported_ignored.id] = reported_ignored
else:
final_nonretry_skips.append(reported_ignored)
for skipped in application_result.skipped:
reported_skip = _reported_skip(stage.name, pass_number, skipped)
if _is_retryable_skip(reported_skip, working):
next_retry_skips[reported_skip.id] = reported_skip
else:
final_nonretry_skips.append(reported_skip)
unresolved_retry_skips = next_retry_skips
pass_summaries.append(
{
"pass_number": pass_number,
"retry_pass": pass_number > 1,
"section_count": len(sections),
"segment_count": len(indexed_segments),
"corrections_returned": len(corrections),
"applied_count": len(application_result.applied_corrections),
"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, work_dir_retained=True)
if not unresolved_retry_skips:
break
final_skipped = final_nonretry_skips + [
unresolved_retry_skips[correction_id] for correction_id in sorted(unresolved_retry_skips)
]
return StageRunResult(
transcript=working,
applied_changes=stage_applied_changes,
skipped_corrections=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,
"spoken_form_validation_candidate_count": 0,
"spoken_form_validation_approved_count": 0,
"spoken_form_validation_rejected_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")
meaning_result = filter_with_meaning_preserving_validations(
candidates,
response,
config.grammar_validation_confidence_threshold,
)
validation_summary["validation_approved_count"] = meaning_result.approved_count
validation_summary["validation_rejected_count"] = meaning_result.rejected_count
validation_summary["spoken_form_validation_candidate_count"] = len(meaning_result.rescue_candidates)
candidate_indexes = {candidate.correction_index for candidate in candidates}
allowed_indexes = set(range(len(corrections))) - candidate_indexes
allowed_indexes.update(meaning_result.approved_correction_indexes)
if not meaning_result.rescue_candidates:
return keep_corrections_with_indexes(corrections, allowed_indexes), [], validation_summary
spoken_form_payload = [candidate.to_prompt_payload() for candidate in meaning_result.rescue_candidates]
spoken_form_messages = build_grammar_spoken_form_validation_messages(spoken_form_payload)
spoken_form_prompt_path = pass_dir / "spoken-form-validation-prompt-0000.json"
spoken_form_prompt_path.write_text(
json.dumps(spoken_form_messages, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
spoken_form_response = llm_client.create_grammar_spoken_form_validations(spoken_form_messages, config)
spoken_form_response_path = pass_dir / "spoken-form-validation-response-0000.json"
spoken_form_response_path.write_text(spoken_form_response.model_dump_json(indent=2) + "\n", encoding="utf-8")
spoken_form_result = filter_with_spoken_form_validations(
meaning_result.rescue_candidates,
spoken_form_response,
config.grammar_spoken_form_validation_confidence_threshold,
)
validation_summary["spoken_form_validation_approved_count"] = spoken_form_result.approved_count
validation_summary["spoken_form_validation_rejected_count"] = spoken_form_result.rejected_count
allowed_indexes.update(spoken_form_result.approved_correction_indexes)
return (
keep_corrections_with_indexes(corrections, allowed_indexes),
spoken_form_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")
run_dir = work_dir / f"run-{timestamp}-{uuid4().hex[:8]}"
run_dir.mkdir()
return run_dir
def _write_run_metadata(
run_dir: Path,
config: AuditaConfig,
normalization_summary: Optional[dict],
stage_summaries: List[dict],
work_dir_retained: bool,
) -> None:
metadata = {
**_config_summary(config),
"normalization": normalization_summary,
"stages": stage_summaries,
"work_dir_retained": work_dir_retained,
}
(run_dir / "metadata.json").write_text(
json.dumps(metadata, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def _write_normalization_diagnostics(
run_dir: Path,
source: List[SourceTranscriptSegment],
result: NormalizationResult,
) -> None:
normalization_dir = run_dir / "normalization"
normalization_dir.mkdir()
(normalization_dir / "source-transcript.json").write_text(
_segments_to_json(source),
encoding="utf-8",
)
(normalization_dir / "normalized-transcript.json").write_text(
_segments_to_json(result.transcript),
encoding="utf-8",
)
(normalization_dir / "summary.json").write_text(
json.dumps(result.summary.to_dict(), ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
def _segments_to_json(segments: List[SourceTranscriptSegment]) -> str:
payload = [segment.model_dump(mode="json", exclude_none=True) for segment in segments]
return json.dumps(payload, ensure_ascii=False, indent=2) + "\n"
def _write_and_validate_section(run_dir: Path, section: TranscriptSection) -> None:
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, require_sequential_ids=False)
def _write_skipped_corrections(run_dir: Path, skipped: List[ReportedSkippedCorrection]) -> None:
skipped_path = run_dir / "skipped-corrections.json"
skipped_path.write_text(
json.dumps(
{
"skipped_corrections": [item.to_dict() for item in skipped]
},
ensure_ascii=False,
indent=2,
)
+ "\n",
encoding="utf-8",
)
def _indexed_segments_for_ids(
transcript: List[TranscriptSegment],
ids: List[int],
) -> List[IndexedSegment]:
id_to_position = {segment.id: position for position, segment in enumerate(transcript)}
return [
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: ReportedSkippedCorrection, transcript: List[TranscriptSegment]) -> bool:
return any(segment.id == skipped.id for segment in transcript)
def _sort_transcript_chronologically(
transcript: List[TranscriptSegment],
) -> List[TranscriptSegment]:
indexed = list(enumerate(transcript))
indexed.sort(key=lambda item: (item[1].start, item[1].end, item[0]))
return [segment for _, segment in indexed]
def _log(progress: Optional[ProgressCallback], message: str) -> None:
if progress is not None:
progress(message)
def _reported_skip(stage: str, pass_number: int, skipped: SkippedCorrection) -> ReportedSkippedCorrection:
return ReportedSkippedCorrection(
stage=stage,
pass_number=pass_number,
id=skipped.id,
reason=skipped.reason,
original_text=skipped.original_text,
corrected_text=skipped.corrected_text,
confidence=skipped.confidence,
actual_text=skipped.actual_text,
validation_confidence=skipped.validation_confidence,
validation_reason=skipped.validation_reason,
)
def _should_retain_run_dir(work_dir_retention: str, has_final_skipped: bool) -> bool:
if work_dir_retention == "always":
return True
if work_dir_retention == "never":
return False
return has_final_skipped
def _config_summary(config: AuditaConfig) -> dict:
return {
"model": config.model,
"base_url": config.base_url,
"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,
"grammar_spoken_form_validation_confidence_threshold": (
config.grammar_spoken_form_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,
"normalize_max_segment_gap": config.normalize_max_segment_gap,
"normalize_ellipsis_gap": config.normalize_ellipsis_gap,
"normalize_max_segment_duration": config.normalize_max_segment_duration,
"normalize_max_segment_tokens": config.normalize_max_segment_tokens,
"work_dir_retention": config.work_dir_retention,
}
def _build_run_report(
config: AuditaConfig,
normalization_summary: Optional[dict],
stage_summaries: List[dict],
applied_changes: List[AppliedChange],
skipped_corrections: List[ReportedSkippedCorrection],
work_dir_retention: str,
work_dir_retained: bool,
run_dir: Path,
transcript: List[TranscriptSegment],
status: str,
error: Optional[str],
) -> RunReport:
totals = {
"output_segment_count": len(transcript),
"applied_change_count": len(applied_changes),
"skipped_correction_count": len(skipped_corrections),
}
if normalization_summary is not None:
totals["source_segment_count"] = normalization_summary["source_segment_count"]
totals["normalized_segment_count"] = normalization_summary["normalized_segment_count"]
return RunReport(
status=status,
config=_config_summary(config),
normalization=normalization_summary,
stages=stage_summaries,
applied_changes=applied_changes,
skipped_corrections=skipped_corrections,
totals=totals,
work_dir_retention=work_dir_retention,
work_dir_retained=work_dir_retained,
work_dir=str(run_dir) if work_dir_retained else None,
error=error,
)
def _write_run_report(path: Path, report: RunReport) -> None:
path.write_text(report.to_json(), encoding="utf-8")

0
src/audita/prompts.py → src/audita_prototype/prompts.py
View File

0
src/audita/protection.py → src/audita_prototype/protection.py
View File

1
src/audita_prototype/py.typed Normal file
View File

@@ -0,0 +1 @@

0
src/audita/reporting.py → src/audita_prototype/reporting.py
View File

0
src/audita/schemas.py → src/audita_prototype/schemas.py
View File

0
src/audita/semantic_validation.py → src/audita_prototype/semantic_validation.py
View File

1
tests/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Test package root."""

1
tests/audita_prototype/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""Archived prototype regression suite."""

6
tests/test_chunking.py → tests/audita_prototype/test_chunking.py
View File

@@ -1,8 +1,8 @@
import pytest
from audita.chunking import chunk_transcript
from audita.errors import AuditaValidationError
from audita.schemas import parse_transcript_json
from audita_prototype.chunking import chunk_transcript
from audita_prototype.errors import AuditaValidationError
from audita_prototype.schemas import parse_transcript_json
class CountEstimator:

14
tests/test_cli.py → tests/audita_prototype/test_cli.py
View File

@@ -1,8 +1,8 @@
import pytest
from audita.cli import main
from audita.reporting import ProcessResult, RunReport
from audita.schemas import parse_transcript_json
from audita_prototype.cli import main
from audita_prototype.reporting import ProcessResult, RunReport
from audita_prototype.schemas import parse_transcript_json
def test_cli_help_uses_audita_program_name(capsys):
@@ -63,10 +63,10 @@ def test_cli_process_writes_report_json(monkeypatch, tmp_path):
work_dir_retained=False,
)
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", lambda overrides=None: object())
monkeypatch.setattr("audita.cli.load_transcript", lambda path: [])
monkeypatch.setattr("audita.cli.load_glossary", lambda path: object())
monkeypatch.setattr("audita.cli.process_transcript_result", lambda *args, **kwargs: result)
monkeypatch.setattr("audita_prototype.cli.AuditaConfig.from_sources", lambda overrides=None: object())
monkeypatch.setattr("audita_prototype.cli.load_transcript", lambda path: [])
monkeypatch.setattr("audita_prototype.cli.load_glossary", lambda path: object())
monkeypatch.setattr("audita_prototype.cli.process_transcript_result", lambda *args, **kwargs: result)
output_path = tmp_path / "out.json"
report_path = tmp_path / "report.json"

6
tests/test_config.py → tests/audita_prototype/test_config.py
View File

@@ -2,8 +2,8 @@ from pathlib import Path
import pytest
from audita.config import AuditaConfig, ConfigOverrides
from audita.config import (
from audita_prototype.config import AuditaConfig, ConfigOverrides
from audita_prototype.config import (
DEFAULT_GLOSSARY_CONFIDENCE_THRESHOLD,
DEFAULT_GLOSSARY_MAX_LLM_PASSES,
DEFAULT_GRAMMAR_CONFIDENCE_THRESHOLD,
@@ -20,7 +20,7 @@ from audita.config import (
DEFAULT_WORK_DIR,
DEFAULT_WORK_DIR_RETENTION,
)
from audita.errors import AuditaConfigError
from audita_prototype.errors import AuditaConfigError
def test_config_uses_defaults_with_api_key():

8
tests/test_corrections.py → tests/audita_prototype/test_corrections.py
View File

@@ -1,9 +1,9 @@
import pytest
from audita.corrections import apply_corrections
from audita.errors import AuditaValidationError
from audita.protection import ProtectedVocabulary
from audita.schemas import CorrectionCandidate, parse_glossary_yaml, parse_transcript_json
from audita_prototype.corrections import apply_corrections
from audita_prototype.errors import AuditaValidationError
from audita_prototype.protection import ProtectedVocabulary
from audita_prototype.schemas import CorrectionCandidate, parse_glossary_yaml, parse_transcript_json
def _transcript():

33
tests/audita_prototype/test_launcher.py Normal file
View File

@@ -0,0 +1,33 @@
import os
import shutil
import subprocess
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
def test_prototype_package_is_importable():
package_root = ROOT / "src" / "audita_prototype"
assert package_root.is_dir()
assert (package_root / "__main__.py").is_file()
def test_prototype_module_help_smoke():
env = os.environ.copy()
env["PYTHONPATH"] = str(ROOT / "src")
result = subprocess.run(
[sys.executable, "-m", "audita_prototype", "--help"],
cwd=ROOT,
text=True,
capture_output=True,
env=env,
check=False,
)
assert result.returncode == 0
assert result.stdout.startswith("usage: audita ")

4
tests/test_normalization.py → tests/audita_prototype/test_normalization.py
View File

@@ -1,5 +1,5 @@
from audita.normalization import normalize_transcript
from audita.schemas import parse_source_transcript_json
from audita_prototype.normalization import normalize_transcript
from audita_prototype.schemas import parse_source_transcript_json
class WordEstimator:

10
tests/test_pipeline.py → tests/audita_prototype/test_pipeline.py
View File

@@ -2,11 +2,11 @@ import json
import pytest
from audita.config import AuditaConfig
from audita.errors import AuditaError
from audita.io import write_report
from audita.pipeline import process_transcript, process_transcript_result
from audita.schemas import (
from audita_prototype.config import AuditaConfig
from audita_prototype.errors import AuditaError
from audita_prototype.io import write_report
from audita_prototype.pipeline import process_transcript, process_transcript_result
from audita_prototype.schemas import (
CorrectionCandidate,
GrammarSpokenFormValidationDecision,
GrammarSpokenFormValidationSet,

6
tests/test_prompts.py → tests/audita_prototype/test_prompts.py
View File

@@ -1,13 +1,13 @@
import json
from audita.chunking import chunk_transcript
from audita.prompts import (
from audita_prototype.chunking import chunk_transcript
from audita_prototype.prompts import (
build_glossary_correction_messages,
build_grammar_correction_messages,
build_grammar_spoken_form_validation_messages,
build_grammar_validation_messages,
)
from audita.schemas import parse_glossary_yaml, parse_transcript_json
from audita_prototype.schemas import parse_glossary_yaml, parse_transcript_json
def test_prompt_requires_acoustically_plausible_transcription_errors():

4
tests/test_protection.py → tests/audita_prototype/test_protection.py
View File

@@ -1,5 +1,5 @@
from audita.protection import ProtectedVocabulary
from audita.schemas import parse_glossary_yaml
from audita_prototype.protection import ProtectedVocabulary
from audita_prototype.schemas import parse_glossary_yaml
def _vocabulary():

8
tests/test_semantic_validation.py → tests/audita_prototype/test_semantic_validation.py
View File

@@ -1,8 +1,8 @@
import pytest
from audita.errors import AuditaLLMError
from audita.protection import ProtectedVocabulary
from audita.schemas import (
from audita_prototype.errors import AuditaLLMError
from audita_prototype.protection import ProtectedVocabulary
from audita_prototype.schemas import (
CorrectionCandidate,
GrammarSpokenFormValidationDecision,
GrammarSpokenFormValidationSet,
@@ -11,7 +11,7 @@ from audita.schemas import (
parse_glossary_yaml,
parse_transcript_json,
)
from audita.semantic_validation import (
from audita_prototype.semantic_validation import (
filter_with_meaning_preserving_validations,
filter_with_spoken_form_validations,
keep_corrections_with_indexes,

4
tests/test_validation.py → tests/audita_prototype/test_validation.py
View File

@@ -1,7 +1,7 @@
import pytest
from audita.errors import AuditaValidationError
from audita.schemas import parse_glossary_yaml, parse_source_transcript_json, parse_transcript_json
from audita_prototype.errors import AuditaValidationError
from audita_prototype.schemas import parse_glossary_yaml, parse_source_transcript_json, parse_transcript_json
def test_valid_transcript_parses():

27
tests/test_framework_chunking.py Normal file
View File

@@ -0,0 +1,27 @@
from audita.core.chunking import TokenEstimatorProtocol, chunk_transcript
from audita.core.schemas import parse_transcript_json
class FakeEstimator(TokenEstimatorProtocol):
def estimate_json(self, value):
if len(value) == 1:
return 4
return len(value) * 4
def test_chunk_transcript_batches_sections_by_token_limit():
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "one"},
{"id": 2, "speaker": "A", "start": 1.0, "end": 2.0, "text": "two"},
{"id": 3, "speaker": "A", "start": 2.0, "end": 3.0, "text": "three"}
]
"""
)
sections = chunk_transcript(transcript, max_section_tokens=8, estimator=FakeEstimator())
assert len(sections) == 2
assert [segment.segment.id for segment in sections[0].segments] == [1, 2]
assert [segment.segment.id for segment in sections[1].segments] == [3]

173
tests/test_framework_runner.py Normal file
View File

@@ -0,0 +1,173 @@
from pathlib import Path
from audita.core.config import AuditaConfig
from audita.core.schemas import parse_glossary_yaml, parse_transcript_json
from audita.framework.models import (
CorrectionProposal,
FilterDecision,
ModuleContext,
ModuleRunSpec,
ReviewDecision,
)
from audita.framework.runner import PipelineRunner
class AllowAllFilter:
name = "allow_all"
def evaluate(self, proposal, transcript, glossary, config):
return FilterDecision(approved=True)
class RejectAllFilter:
name = "reject_all"
def evaluate(self, proposal, transcript, glossary, config):
return FilterDecision(approved=False, reason="filter rejected proposal")
class AllowAllReviewStage:
name = "allow_all_review"
def review(self, proposals, transcript, glossary, config, llm_client, run_dir):
return [ReviewDecision(proposal_index=proposal.proposal_index, approved=True) for proposal in proposals]
class RecordingModule:
replacement_policy = "require_unique"
def __init__(self, module_key, proposals, recorder):
self.module_key = module_key
self._proposals = proposals
self._recorder = recorder
def deterministic_filters(self):
return [AllowAllFilter()]
def review_stages(self):
return [AllowAllReviewStage()]
def propose(self, transcript_section, context: ModuleContext):
self._recorder.append([segment.text for segment in transcript_section])
return list(self._proposals)
class RejectedModule:
module_key = "rejected"
replacement_policy = "require_unique"
def deterministic_filters(self):
return [RejectAllFilter()]
def review_stages(self):
return []
def propose(self, transcript_section, context):
return [
CorrectionProposal(
proposal_index=0,
module_instance=context.run_spec.instance_name,
module_key=context.run_spec.module_key,
id=1,
original_text="Hello",
corrected_text="Goodbye",
confidence=0.9,
)
]
def test_pipeline_runner_applies_modules_sequentially(tmp_path):
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "Alpha."}
]
"""
)
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Alpha"
category: noun
summary: "Alpha."
"""
)
seen = []
first = RecordingModule(
"first",
[
CorrectionProposal(
proposal_index=0,
module_instance="first",
module_key="first",
id=1,
original_text="Alpha",
corrected_text="Beta",
confidence=0.9,
)
],
seen,
)
second = RecordingModule(
"second",
[
CorrectionProposal(
proposal_index=0,
module_instance="second",
module_key="second",
id=1,
original_text="Beta",
corrected_text="Gamma",
confidence=0.9,
)
],
seen,
)
runner = PipelineRunner()
result = runner.run(
transcript=transcript,
glossary=glossary,
module_specs=[
ModuleRunSpec(instance_name="first", module_key="first", module=first),
ModuleRunSpec(instance_name="second", module_key="second", module=second),
],
config=AuditaConfig.from_sources(env={}),
run_dir=tmp_path / "run",
)
assert seen[0] == ["Alpha."]
assert seen[1] == ["Beta."]
assert result.transcript[0].text == "Gamma."
assert len(result.applied_changes) == 2
def test_pipeline_runner_reports_filter_rejections(tmp_path):
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "A", "start": 0.0, "end": 1.0, "text": "Hello."}
]
"""
)
glossary = parse_glossary_yaml(
"""
glossary:
- name: "Hello"
category: noun
summary: "Hello."
"""
)
runner = PipelineRunner()
result = runner.run(
transcript=transcript,
glossary=glossary,
module_specs=[ModuleRunSpec(instance_name="rejected", module_key="rejected", module=RejectedModule())],
config=AuditaConfig.from_sources(env={}),
run_dir=tmp_path / "run",
)
assert result.transcript[0].text == "Hello."
assert result.module_reports[0].skipped_count == 1
assert result.skipped_corrections[0].reason == "filter rejected proposal"

83
tests/test_new_cli.py Normal file
View File

@@ -0,0 +1,83 @@
import pytest
from audita.cli import main
from audita.core.reporting import ProcessResult, RunReport
from audita.core.schemas import parse_transcript_json
def test_cli_help_uses_audita_program_name(capsys):
with pytest.raises(SystemExit) as exc:
main(["--help"])
assert exc.value.code == 0
assert capsys.readouterr().out.startswith("usage: audita ")
def test_process_help_exposes_framework_flags(capsys):
with pytest.raises(SystemExit) as exc:
main(["process", "--help"])
assert exc.value.code == 0
output = capsys.readouterr().out
assert "--report-json" in output
assert "--model" in output
assert "--base-url" in output
assert "--max-retries" in output
assert "--max-section-tokens" in output
assert "--work-dir-retention" in output
assert "--normalize-max-segment-gap" in output
assert "--glossary-confidence-threshold" not in output
assert "--grammar-validation-enabled" not in output
def test_cli_process_writes_report_json(monkeypatch, tmp_path):
transcript = parse_transcript_json(
"""
[
{"id": 1, "speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Fixed."}
]
"""
)
report = RunReport(
status="success",
config={"model": "m", "base_url": "b"},
normalization={"source_segment_count": 1, "normalized_segment_count": 1, "merge_count": 0},
pipeline=["glossary_primary", "homophones", "glossary_secondary", "spoken_word", "grammar"],
modules=[],
applied_changes=[],
skipped_corrections=[],
totals={"output_segment_count": 1, "applied_change_count": 0, "skipped_correction_count": 0},
work_dir_retention="auto",
work_dir_retained=False,
work_dir=None,
error=None,
)
result = ProcessResult(
transcript=transcript,
report=report,
run_dir=tmp_path / "run",
work_dir_retained=False,
)
monkeypatch.setattr("audita.cli.AuditaConfig.from_sources", lambda overrides=None: object())
monkeypatch.setattr("audita.cli.load_transcript", lambda path: [])
monkeypatch.setattr("audita.cli.load_glossary", lambda path: object())
monkeypatch.setattr("audita.cli.process_transcript_result", lambda *args, **kwargs: result)
output_path = tmp_path / "out.json"
report_path = tmp_path / "report.json"
exit_code = main(
[
"process",
"transcript.json",
"--glossary",
"glossary.yaml",
"--output",
str(output_path),
"--report-json",
str(report_path),
]
)
assert exit_code == 0
assert report_path.exists()

31
tests/test_new_config.py Normal file
View File

@@ -0,0 +1,31 @@
import pytest
from audita.core.config import (
AuditaConfig,
ConfigOverrides,
DEFAULT_NORMALIZE_MAX_SEGMENT_GAP,
DEFAULT_WORK_DIR_RETENTION,
)
from audita.core.errors import AuditaConfigError
def test_default_config_allows_missing_api_key():
config = AuditaConfig.from_sources(env={})
assert config.api_key is None
assert config.normalize_max_segment_gap == DEFAULT_NORMALIZE_MAX_SEGMENT_GAP
assert config.work_dir_retention == DEFAULT_WORK_DIR_RETENTION
def test_cli_overrides_take_precedence():
config = AuditaConfig.from_sources(
env={"AUDITA_MAX_SECTION_TOKENS": "1000"},
overrides=ConfigOverrides(max_section_tokens=2000),
)
assert config.max_section_tokens == 2000
def test_invalid_work_dir_retention_is_rejected():
with pytest.raises(AuditaConfigError):
AuditaConfig.from_sources(env={"AUDITA_WORK_DIR_RETENTION": "sometimes"})

0
tests/test_launcher.py → tests/test_new_launcher.py
View File

86
tests/test_new_pipeline.py Normal file
View File

@@ -0,0 +1,86 @@
import json
from audita.core.config import AuditaConfig
from audita.core.io import write_report
from audita.core.schemas import parse_glossary_yaml, parse_source_transcript_json
from audita.pipeline import process_transcript, process_transcript_result
def _glossary():
return parse_glossary_yaml(
"""
glossary:
- name: "Jesters"
category: faction
summary: "A faction."
"""
)
def _transcript():
return parse_source_transcript_json(
"""
[
{"speaker": "Eric", "start": 0.0, "end": 1.0, "text": "Hello."},
{"speaker": "Eric", "start": 1.5, "end": 2.0, "text": "Again."},
{"speaker": "Mike", "start": 10.0, "end": 11.0, "text": "Done."}
]
"""
)
def test_process_transcript_runs_noop_framework(tmp_path):
revised = process_transcript(
_transcript(),
_glossary(),
AuditaConfig.from_sources(env={}, overrides=None),
)
assert [segment.id for segment in revised] == [1, 2]
assert revised[0].text == "Hello. Again."
assert revised[1].text == "Done."
def test_process_transcript_result_writes_report_and_preserves_skips_per_policy(tmp_path):
config = AuditaConfig.from_sources(
env={},
overrides=None,
)
config = AuditaConfig(
api_key=config.api_key,
model=config.model,
base_url=config.base_url,
max_retries=config.max_retries,
max_section_tokens=config.max_section_tokens,
normalize_max_segment_gap=config.normalize_max_segment_gap,
normalize_ellipsis_gap=config.normalize_ellipsis_gap,
normalize_max_segment_duration=config.normalize_max_segment_duration,
normalize_max_segment_tokens=config.normalize_max_segment_tokens,
work_dir=tmp_path / "work",
work_dir_retention="always",
)
result = process_transcript_result(_transcript(), _glossary(), config)
assert result.work_dir_retained is True
assert result.report.pipeline == [
"glossary_primary",
"homophones",
"glossary_secondary",
"spoken_word",
"grammar",
]
assert result.report.totals["applied_change_count"] == 0
assert (result.run_dir / "report.json").exists()
assert (result.run_dir / "normalization" / "summary.json").exists()
def test_external_report_can_be_written(tmp_path):
config = AuditaConfig.from_sources(env={})
result = process_transcript_result(_transcript(), _glossary(), config)
report_path = tmp_path / "report.json"
write_report(report_path, result.report)
payload = json.loads(report_path.read_text(encoding="utf-8"))
assert payload["pipeline"][0] == "glossary_primary"
assert payload["totals"]["applied_change_count"] == 0