Speech ML/AI Evaluation And Incident Exam

Gate on final WER and CER by slice, entity recall, punctuation if user-visible, first partial latency, finalization latency, partial churn, endpointing false cuts, correction rate, and downstream task success. Run offline fixtures, shadow traffic, and a small canary. Promote only if no high-risk approved aggregate, consented, or high-value slice crosses a defined regression threshold. Roll back on tail-latency SLO breach, correction-rate spike, or severe entity recall regression.

Evaluate ASR robustness on noisy and entity-heavy queries, retrieval recall with ASR variants, ACL correctness, freshness, grounded answer rate, refusal behavior, citation traceability, tool-use safety, TTS intelligibility, first audio byte, complete turn latency, escalation precision, and cost per successful answer. Include human review for high-risk intents and synthetic tests for stale documents, ambiguous policies, and prompt-injection attempts.

Split product surfaces into interactive and noninteractive use. For interactive voice agents, require first audio byte, chunk cadence, barge-in recovery, pronunciation, safety, and abandonment to stay within gates. For long-form generation, slower synthesis may be acceptable if quality improves and cost is controlled. Consider segmented first responses, voice warm pools, cache policies, or model-tier routing before rejecting the release.

Slice by model, decoder config, VAD, chunk size, endpointing, client version, language, noise, and network path. Compare partial churn, first partial latency, finalization delay, and downstream correction rate. Mitigate with decoder rollback, more conservative partial display, endpointing rollback, or affected-slice routing. Add canary gates for partial stability and UI-visible rewrite rate.

Check tenant index version, document ingestion watermark, ACLs, embedding version, chunking config, reranker version, cache TTL, and whether ASR entity substitutions changed retrieval queries. Pin the tenant to the last good index or force human handoff for affected intents. Add per-tenant freshness gates and ASR-noisy retrieval evals before index promotion.

Check prompt length, retrieved document count, conversation length, retry storms, speculative decoding fallback, model mix, cache hit rate, batch efficiency, and tenant-specific routes. Stabilize with token budgets, retrieval caps, priority queues, fallback model tiers, admission control for noninteractive work, and release rollback if a config change caused the spike. Add cost per successful turn to launch gates.

Invariant: every required slice and metric must pass. Missing candidate metrics are failures because a release gate cannot assume safety. Test missing slices, missing metrics, exact threshold, improvements, a lower-is-better regression, and a higher-is-better regression.

def can_promote(baseline, candidate, thresholds):
    failures = []

    for slice_name, base_metrics in baseline.items():
        cand_metrics = candidate.get(slice_name)
        if cand_metrics is None:
            failures.append((slice_name, "missing_slice", None))
            continue

        for metric, base_value in base_metrics.items():
            if metric not in thresholds:
                continue
            cand_value = cand_metrics.get(metric)
            if cand_value is None:
                failures.append((slice_name, metric, "missing_metric"))
                continue

            direction = thresholds[metric]["direction"]
            allowed = thresholds[metric]["allowed_regression"]

            if direction == "lower_is_better":
                regression = cand_value - base_value
            elif direction == "higher_is_better":
                regression = base_value - cand_value
            else:
                raise ValueError(f"unknown direction for {metric}: {direction}")

            if regression - allowed > 1e-12:
                failures.append((slice_name, metric, regression))

    return {
        "promote": not failures,
        "failures": failures,
    }

Invariant: streak is the number of consecutive days up to the current point that violate the threshold in the configured direction. Reset it on a healthy day. Common mistakes include flagging one-day noise, assuming every metric drifts upward, and forgetting to return the first sustained breach. In production, reject invalid windows and missing telemetry instead of treating unknown days as healthy.

def first_sustained_drift(points, metric, threshold, consecutive_days, direction):
    if consecutive_days <= 0:
        raise ValueError("consecutive_days must be positive")
    if direction not in {"higher_is_worse", "lower_is_worse"}:
        raise ValueError(f"unknown direction: {direction}")

    streak = 0
    for point in points:
        if metric not in point:
            raise ValueError(f"missing {metric} for day {point.get('day')}")

        value = point[metric]
        if direction == "higher_is_worse":
            breached = value > threshold
        else:
            breached = value < threshold

        if breached:
            streak += 1
            if streak >= consecutive_days:
                return point["day"]
        else:
            streak = 0
    return None

A rollback target should reduce user impact quickly without hiding safety or privacy issues. Only rank candidates that are explicitly known-good; an unknown version is not a safe rollback target during an incident. Prefer lower regression score, smaller blast radius, and short rollback time. If two options are close, choose the one with better observability and smaller config difference from the current serving path.

def rank_rollback_targets(candidates):
    required = {
        "version",
        "known_good",
        "regression_score",
        "blast_radius_percent",
        "rollback_minutes",
    }

    safe_candidates = []
    for candidate in candidates:
        missing = required - candidate.keys()
        if missing:
            raise ValueError(f"missing rollback fields: {sorted(missing)}")
        if candidate["known_good"]:
            safe_candidates.append(candidate)

    if not safe_candidates:
        raise ValueError("no known-good rollback target")

    def score(candidate):
        return (
            candidate["regression_score"],
            candidate["blast_radius_percent"],
            candidate["rollback_minutes"],
            -candidate.get("observability_score", 0),
        )

    return sorted(safe_candidates, key=score)