Speech Model CI/CD And Release Engineering

The model may see shifted frequency content and degraded phoneme evidence, especially for sibilants, noise, and accents. The release should fail a contract test before canary. A strong answer asks for sample-rate metadata, feature checksums, slice WER, client version splits, and replay fixtures that verify the exact production feature path.

REQUIRED_FIELDS = {
    "model_id",
    "version",
    "owner",
    "task",
    "data_manifest",
    "training_code_sha",
    "feature_contract",
    "eval_report",
    "serving_image",
    "rollback_target",
}


def validate_registry_entry(entry):
    if not isinstance(entry, dict):
        return {
            "release_blocked": True,
            "missing_fields": sorted(REQUIRED_FIELDS),
        }
    missing = sorted(field for field in REQUIRED_FIELDS if not entry.get(field))
    return {
        "release_blocked": bool(missing),
        "missing_fields": missing,
    }

The invariant is that every promoted model must be traceable and reversible. The common mistake is validating only model weights and ignoring feature, eval, and rollback metadata.

import math


REQUIRED_SLICE_FIELDS = {
    "wer_relative_delta",
    "p95_latency_ms_delta",
    "requests",
}


def finite_number(value):
    return isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(value)


def canary_promotion(metrics, approved_slices, min_requests_per_slice=100):
    if not isinstance(metrics, dict):
        return {"decision": "hold", "blockers": ["invalid_metrics"]}
    if not isinstance(approved_slices, set) or not approved_slices:
        raise ValueError("approved_slices must be a non-empty set")
    if not finite_number(min_requests_per_slice) or min_requests_per_slice <= 0:
        raise ValueError("min_requests_per_slice must be positive")

    blockers = []

    slices = metrics.get("slices")
    if not isinstance(slices, dict) or not slices:
        blockers.append("missing_slices")
        slices = {}

    for name, values in slices.items():
        if name not in approved_slices:
            blockers.append(f"{name}:unapproved_slice")
            continue
        if not isinstance(values, dict):
            blockers.append(f"{name}:invalid_metrics")
            continue
        missing = REQUIRED_SLICE_FIELDS - values.keys()
        if missing:
            blockers.append(f"{name}:missing_{sorted(missing)[0]}")
            continue
        if not finite_number(values["requests"]) or values["requests"] < min_requests_per_slice:
            blockers.append(f"{name}:low_exposure")
        if not finite_number(values["wer_relative_delta"]) or values["wer_relative_delta"] > 0.02:
            blockers.append(f"{name}:wer")
        if not finite_number(values["p95_latency_ms_delta"]) or values["p95_latency_ms_delta"] > 75:
            blockers.append(f"{name}:latency")

    if metrics.get("privacy_errors") is None:
        blockers.append("missing_privacy_errors")
    elif not finite_number(metrics["privacy_errors"]) or metrics["privacy_errors"] > 0:
        blockers.append("privacy")

    if metrics.get("unit_cost_relative_delta") is None:
        blockers.append("missing_unit_cost")
    elif not finite_number(metrics["unit_cost_relative_delta"]) or metrics["unit_cost_relative_delta"] > 0.15:
        blockers.append("unit_cost")

    if blockers:
        return {"decision": "hold", "blockers": blockers}
    return {"decision": "promote", "blockers": []}

Average quality improvement is not enough. A canary should protect critical slices, privacy, latency, unit economics, and telemetry completeness before promotion. Missing top-level or slice telemetry should hold the release, not crash late, smuggle NaN through a comparison, expose raw cohort labels, or default to success.

REQUIRED_RELEASE_FIELDS = {
    "family",
    "state",
    "promoted_at",
    "model_id",
    "version",
    "serving_image",
}


def rollback_target(releases, family):
    candidates = []
    for release in releases:
        if not isinstance(release, dict):
            raise ValueError("release records must be dictionaries")
        missing = REQUIRED_RELEASE_FIELDS - release.keys()
        if missing:
            raise ValueError(f"release missing {sorted(missing)[0]}")
        if release["family"] == family and release["state"] == "stable":
            candidates.append(release)

    if not candidates:
        raise ValueError(f"no stable rollback target for {family}")

    candidates.sort(key=lambda release: release["promoted_at"], reverse=True)
    return {
        "model_id": candidates[0]["model_id"],
        "version": candidates[0]["version"],
        "serving_image": candidates[0]["serving_image"],
    }

The target must be a known stable serving bundle, not merely the previous commit. A release system validates its metadata before an incident so rollback does not fail late on a malformed record.

Cover unit tests for feature extraction, schema compatibility, sanitized replay fixtures, WER/CER slices, command intent recall, partial churn, endpointing, latency, memory, privacy logging, canary thresholds, model registry metadata, and a tested rollback target. Dictation and commands need separate quality gates.

Add load tests with realistic concurrency, queue age, warmup, voice mix, long utterances, memory fragmentation, retry behavior, and autoscaling. Require serving-image benchmarks in CI, capacity signoff, narrow canaries, and rollback verification before the next release.

Build a paved release path with registry templates, automated eval packs, synthetic replay, shadow traffic, slice dashboards, progressive canaries, fast rollback, and clear ownership. Speed comes from removing manual ambiguity, not skipping gates.