| """Interpretable baselines for endpoint classification and pause control. |
| |
| These baselines are deliberately small enough to audit. The acoustic model |
| uses waveform-derived statistics only; it never consumes source, language, or |
| synthetic flags that could become dataset shortcuts. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import io |
| import json |
| import math |
| from collections.abc import Iterable, Mapping, Sequence |
| from dataclasses import asdict, dataclass |
| from pathlib import Path |
| from typing import Any |
|
|
| from turn_detection.runtime.replay import PauseCheckpoint, ReplayRecord, summarize_replay |
| from turn_detection.runtime.types import TurnDecision, TurnState |
|
|
| AUDIO_FEATURE_NAMES = ( |
| "log_duration_seconds", |
| "rms_db", |
| "tail_rms_db", |
| "tail_energy_delta_db", |
| "tail_silence_fraction", |
| "tail_zero_crossing_rate", |
| "spectral_centroid_nyquist", |
| "spectral_rolloff85_nyquist", |
| "periodicity", |
| "energy_slope", |
| ) |
|
|
|
|
| def _numpy() -> Any: |
| try: |
| import numpy as np |
| except ImportError as exc: |
| raise RuntimeError("audio baselines require numpy") from exc |
| return np |
|
|
|
|
| def _sigmoid(values: Any) -> Any: |
| np = _numpy() |
| clipped = np.clip(values, -30.0, 30.0) |
| return 1.0 / (1.0 + np.exp(-clipped)) |
|
|
|
|
| def decode_audio_value(value: Any) -> tuple[Any, int]: |
| """Decode a Hugging Face audio value without materializing a dataset.""" |
|
|
| np = _numpy() |
| try: |
| import soundfile as sf |
| except ImportError as exc: |
| raise RuntimeError("audio baselines require soundfile") from exc |
|
|
| declared_rate: int | None = None |
| source: Any = value |
| if isinstance(value, Mapping): |
| declared_rate = value.get("sampling_rate", value.get("sample_rate")) |
| if value.get("array") is not None: |
| array = np.asarray(value["array"]) |
| source_rate = int(declared_rate) if declared_rate else 16_000 |
| return _mono_float(array), source_rate |
| if value.get("samples") is not None: |
| array = np.asarray(value["samples"]) |
| source_rate = int(declared_rate) if declared_rate else 16_000 |
| return _mono_float(array), source_rate |
| if value.get("bytes") is not None: |
| source = io.BytesIO(value["bytes"]) |
| elif value.get("path"): |
| source = value["path"] |
| else: |
| raise ValueError("audio mapping contains no bytes, path, array, or samples") |
| elif isinstance(value, (bytes, bytearray)): |
| source = io.BytesIO(bytes(value)) |
| elif isinstance(value, (str, Path)): |
| source = str(value) |
| else: |
| return _mono_float(np.asarray(value)), int(declared_rate or 16_000) |
|
|
| array, decoded_rate = sf.read(source, dtype="float32", always_2d=True) |
| return _mono_float(array), int(declared_rate or decoded_rate) |
|
|
|
|
| def _mono_float(audio: Any) -> Any: |
| np = _numpy() |
| samples = np.asarray(audio) |
| original_dtype = samples.dtype |
| if samples.size == 0: |
| raise ValueError("audio cannot be empty") |
| if samples.ndim == 2: |
| channel_axis = 1 if samples.shape[1] <= 8 else 0 |
| samples = samples.astype(np.float32).mean(axis=channel_axis) |
| elif samples.ndim != 1: |
| raise ValueError(f"expected one- or two-dimensional audio, got {samples.shape}") |
| if np.issubdtype(original_dtype, np.integer): |
| info = np.iinfo(original_dtype) |
| samples = samples.astype(np.float32) / float(max(abs(info.min), info.max)) |
| else: |
| samples = samples.astype(np.float32, copy=False) |
| samples = np.nan_to_num(samples, nan=0.0, posinf=1.0, neginf=-1.0) |
| peak = float(np.max(np.abs(samples))) |
| if peak > 1.0: |
| samples = samples / peak |
| return np.clip(samples, -1.0, 1.0) |
|
|
|
|
| def _linear_resample(samples: Any, source_rate: int, target_rate: int) -> Any: |
| np = _numpy() |
| if source_rate == target_rate: |
| return samples.astype(np.float32, copy=False) |
| output_length = max(1, round(len(samples) * target_rate / source_rate)) |
| old_positions = np.arange(len(samples), dtype=np.float64) |
| new_positions = np.linspace(0.0, max(0, len(samples) - 1), output_length) |
| return np.interp(new_positions, old_positions, samples).astype(np.float32) |
|
|
|
|
| def extract_audio_statistics( |
| audio: Any, |
| sample_rate: int, |
| *, |
| target_rate: int = 16_000, |
| max_seconds: float = 4.0, |
| ) -> Any: |
| """Return ten finite, interpretable suffix-acoustic features.""" |
|
|
| np = _numpy() |
| if sample_rate <= 0 or target_rate <= 0 or max_seconds <= 0: |
| raise ValueError("sample rates and max_seconds must be positive") |
| samples = _linear_resample(_mono_float(audio), sample_rate, target_rate) |
| duration_seconds = len(samples) / target_rate |
| samples = samples[-round(max_seconds * target_rate) :] |
| tail = samples[-min(len(samples), round(0.40 * target_rate)) :] |
| body = samples[: -len(tail)] if len(samples) > len(tail) else samples |
| eps = 1e-10 |
|
|
| def rms_db(values: Any) -> float: |
| return 10.0 * math.log10(float(np.mean(values * values)) + eps) |
|
|
| total_db = rms_db(samples) |
| tail_db = rms_db(tail) |
| body_db = rms_db(body) |
| tail_silence_fraction = float(np.mean(np.abs(tail) < 0.01)) |
| tail_zcr = float(np.mean((tail[1:] >= 0) != (tail[:-1] >= 0))) if len(tail) > 1 else 0.0 |
|
|
| analysis = tail[-min(len(tail), target_rate) :] |
| windowed = analysis * np.hanning(len(analysis)).astype(np.float32) |
| spectrum = np.abs(np.fft.rfft(windowed)) ** 2 |
| frequencies = np.fft.rfftfreq(len(windowed), d=1.0 / target_rate) |
| spectral_sum = float(spectrum.sum()) + eps |
| centroid = float((spectrum * frequencies).sum() / spectral_sum) / (target_rate / 2.0) |
| cumulative = np.cumsum(spectrum) |
| rolloff_index = int(np.searchsorted(cumulative, 0.85 * cumulative[-1])) |
| rolloff_index = min(rolloff_index, len(frequencies) - 1) |
| rolloff = float(frequencies[rolloff_index]) / (target_rate / 2.0) |
|
|
| centered = analysis - float(analysis.mean()) |
| energy = float(np.dot(centered, centered)) + eps |
| minimum_lag = max(1, target_rate // 400) |
| maximum_lag = min(len(centered) - 1, target_rate // 60) |
| periodicity = 0.0 |
| if maximum_lag >= minimum_lag and energy > eps: |
| correlations = [ |
| float(np.dot(centered[:-lag], centered[lag:])) / energy |
| for lag in range(minimum_lag, maximum_lag + 1) |
| ] |
| periodicity = max(0.0, max(correlations, default=0.0)) |
|
|
| frame_length = max(1, round(0.10 * target_rate)) |
| frame_hop = max(1, round(0.05 * target_rate)) |
| frame_rms: list[float] = [] |
| for start in range(0, max(1, len(samples) - frame_length + 1), frame_hop): |
| frame = samples[start : start + frame_length] |
| if len(frame): |
| frame_rms.append(rms_db(frame)) |
| if len(frame_rms) >= 2: |
| x = np.linspace(-1.0, 1.0, len(frame_rms)) |
| energy_slope = float(np.dot(x, np.asarray(frame_rms) - np.mean(frame_rms))) / float( |
| np.dot(x, x) |
| ) |
| else: |
| energy_slope = 0.0 |
|
|
| features = np.asarray( |
| [ |
| math.log1p(duration_seconds), |
| total_db, |
| tail_db, |
| tail_db - body_db, |
| tail_silence_fraction, |
| tail_zcr, |
| centroid, |
| rolloff, |
| periodicity, |
| energy_slope, |
| ], |
| dtype=np.float64, |
| ) |
| if not bool(np.isfinite(features).all()): |
| raise ValueError("audio statistics contained non-finite values") |
| return features |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class LogisticBaseline: |
| feature_names: tuple[str, ...] |
| mean: tuple[float, ...] |
| scale: tuple[float, ...] |
| weights: tuple[float, ...] |
| bias: float |
| threshold: float = 0.5 |
| sample_rate: int = 16_000 |
| max_seconds: float = 4.0 |
|
|
| def __post_init__(self) -> None: |
| lengths = {len(self.feature_names), len(self.mean), len(self.scale), len(self.weights)} |
| if len(lengths) != 1: |
| raise ValueError("model vectors must have the same length") |
| if any(value <= 0 for value in self.scale): |
| raise ValueError("standardization scales must be positive") |
| if not 0.0 <= self.threshold <= 1.0: |
| raise ValueError("threshold must be in [0, 1]") |
|
|
| def predict_proba(self, features: Any) -> Any: |
| np = _numpy() |
| matrix = np.asarray(features, dtype=np.float64) |
| standardized = (matrix - np.asarray(self.mean)) / np.asarray(self.scale) |
| return _sigmoid(standardized @ np.asarray(self.weights) + self.bias) |
|
|
| def to_dict(self) -> dict[str, Any]: |
| payload = asdict(self) |
| payload["model_type"] = "audio_statistics_logistic_regression" |
| payload["feature_names"] = list(self.feature_names) |
| payload["mean"] = list(self.mean) |
| payload["scale"] = list(self.scale) |
| payload["weights"] = list(self.weights) |
| return payload |
|
|
| @classmethod |
| def from_dict(cls, payload: Mapping[str, Any]) -> LogisticBaseline: |
| values = dict(payload) |
| values.pop("model_type", None) |
| for name in ("feature_names", "mean", "scale", "weights"): |
| values[name] = tuple(values[name]) |
| return cls(**values) |
|
|
|
|
| def fit_logistic_baseline( |
| features: Any, |
| labels: Any, |
| *, |
| epochs: int = 800, |
| learning_rate: float = 0.05, |
| l2: float = 1e-3, |
| feature_names: Sequence[str] | None = None, |
| ) -> LogisticBaseline: |
| """Fit deterministic class-balanced logistic regression with full-batch GD.""" |
|
|
| np = _numpy() |
| matrix = np.asarray(features, dtype=np.float64) |
| targets = np.asarray(labels, dtype=np.float64).reshape(-1) |
| if matrix.ndim != 2 or matrix.shape[0] != len(targets) or matrix.shape[1] == 0: |
| raise ValueError("features must be [examples, dimensions] and align with labels") |
| if len(targets) == 0 or not bool(np.isin(targets, [0.0, 1.0]).all()): |
| raise ValueError("labels must be a non-empty binary vector") |
| resolved_names = tuple(feature_names or ()) |
| if not resolved_names: |
| resolved_names = ( |
| AUDIO_FEATURE_NAMES |
| if matrix.shape[1] == len(AUDIO_FEATURE_NAMES) |
| else tuple(f"feature_{index}" for index in range(matrix.shape[1])) |
| ) |
| if len(resolved_names) != matrix.shape[1] or len(set(resolved_names)) != len(resolved_names): |
| raise ValueError("feature_names must be unique and match the feature dimension") |
| positives = float(targets.sum()) |
| negatives = float(len(targets) - positives) |
| if positives == 0 or negatives == 0: |
| raise ValueError("both endpoint classes are required") |
| if epochs < 1 or learning_rate <= 0 or l2 < 0: |
| raise ValueError("epochs/learning_rate must be positive and l2 non-negative") |
|
|
| mean = matrix.mean(axis=0) |
| scale = matrix.std(axis=0) |
| scale = np.where(scale < 1e-8, 1.0, scale) |
| standardized = (matrix - mean) / scale |
| weights = np.zeros(matrix.shape[1], dtype=np.float64) |
| bias = 0.0 |
| example_weights = np.where( |
| targets == 1.0, len(targets) / (2 * positives), len(targets) / (2 * negatives) |
| ) |
| denominator = float(example_weights.sum()) |
| for _ in range(epochs): |
| probabilities = _sigmoid(standardized @ weights + bias) |
| residual = (probabilities - targets) * example_weights |
| weights -= learning_rate * ((standardized.T @ residual) / denominator + l2 * weights) |
| bias -= learning_rate * float(residual.sum() / denominator) |
|
|
| return LogisticBaseline( |
| feature_names=resolved_names, |
| mean=tuple(float(value) for value in mean), |
| scale=tuple(float(value) for value in scale), |
| weights=tuple(float(value) for value in weights), |
| bias=float(bias), |
| ) |
|
|
|
|
| def extract_manifest_features( |
| rows: Iterable[Mapping[str, Any]], |
| *, |
| source_root: str | Path, |
| max_examples: int | None = None, |
| sample_rate: int = 16_000, |
| max_seconds: float = 4.0, |
| ) -> tuple[Any, Any, list[dict[str, Any]]]: |
| """Resolve manifest audio lazily and produce a numeric matrix plus provenance.""" |
|
|
| from turn_detection.data import ManifestRecordResolver |
|
|
| np = _numpy() |
| if max_examples is not None and max_examples < 1: |
| raise ValueError("max_examples must be positive when provided") |
| resolver = ManifestRecordResolver(source_root=source_root, max_cached_row_groups=1) |
| vectors: list[Any] = [] |
| labels: list[int] = [] |
| provenance: list[dict[str, Any]] = [] |
| try: |
| for row in rows: |
| if max_examples is not None and len(vectors) >= max_examples: |
| break |
| endpoint = row.get("endpoint", row.get("endpoint_bool")) |
| if endpoint not in (False, True, 0, 1): |
| raise ValueError(f"manifest row {row.get('record_id')!r} has no binary endpoint") |
| source_record = resolver.resolve(row, columns=("audio",)) |
| audio, decoded_rate = decode_audio_value(source_record["audio"]) |
| vectors.append( |
| extract_audio_statistics( |
| audio, |
| decoded_rate, |
| target_rate=sample_rate, |
| max_seconds=max_seconds, |
| ) |
| ) |
| labels.append(int(bool(endpoint))) |
| provenance.append( |
| { |
| "record_id": str(row.get("record_id", "")), |
| "group_id": row.get("group_id"), |
| "split": row.get("split"), |
| "label": int(bool(endpoint)), |
| } |
| ) |
| finally: |
| resolver.clear() |
| if not vectors: |
| raise ValueError("manifest selection produced no examples") |
| return np.stack(vectors), np.asarray(labels, dtype=np.int64), provenance |
|
|
|
|
| def fixed_timeout_sweep( |
| checkpoints: Iterable[PauseCheckpoint], |
| timeouts_ms: Sequence[float], |
| ) -> list[dict[str, Any]]: |
| """Evaluate model-free policies that fire after a fixed VAD silence.""" |
|
|
| materialized = list(checkpoints) |
| results: list[dict[str, Any]] = [] |
| for timeout in timeouts_ms: |
| if not math.isfinite(timeout) or timeout < 0: |
| raise ValueError("timeouts must be finite and non-negative") |
| replay_records: list[ReplayRecord] = [] |
| ended_turns: set[str] = set() |
| for checkpoint in materialized: |
| should_end = checkpoint.silence_ms >= timeout |
| emit_response = should_end and checkpoint.turn_id not in ended_turns |
| if emit_response: |
| ended_turns.add(checkpoint.turn_id) |
| state = TurnState.END if checkpoint.turn_id in ended_turns else TurnState.HOLD |
| replay_records.append( |
| ReplayRecord( |
| checkpoint=checkpoint, |
| decision=TurnDecision( |
| state=state, |
| endpoint_probability=None, |
| threshold=None, |
| silence_ms=checkpoint.silence_ms, |
| reason=f"fixed_timeout_{timeout:g}ms", |
| timestamp_ms=checkpoint.timestamp_ms, |
| emit_response=emit_response, |
| ), |
| ) |
| ) |
| results.append({"timeout_ms": float(timeout), **summarize_replay(replay_records)}) |
| return results |
|
|
|
|
| def load_checkpoints_jsonl(path: str | Path) -> list[PauseCheckpoint]: |
| checkpoints: list[PauseCheckpoint] = [] |
| with Path(path).open(encoding="utf-8") as handle: |
| for line_number, line in enumerate(handle, start=1): |
| if not line.strip(): |
| continue |
| try: |
| value = json.loads(line) |
| checkpoints.append(PauseCheckpoint(**value)) |
| except (json.JSONDecodeError, TypeError, ValueError) as exc: |
| raise ValueError(f"invalid checkpoint at {path}:{line_number}") from exc |
| return checkpoints |
|
|
