在 SageMaker AI 上部署 WhisperX：生成词级时间戳与说话人标注

2026-09-25 29 预计阅读时间: 1 分钟
来源: aws.amazon.com AI 摘要 Original link

Disclaimer: This article is an AI-assisted summary. Read it together with the original source when precision matters. The summary may omit context, version differences, or edge cases and is not official documentation.

预计阅读时间：13 分钟

普通语音转写只能回答“说了什么”，会议纪要、客服质检和访谈分析还需要知道“谁在什么时候说了什么”。AWS WhisperX Deep Learning Container 将 Whisper、基于 wav2vec2 的强制对齐，以及说话人分离封装进可使用 GPU 的镜像，可以部署为 Amazon SageMaker AI 实时端点或异步端点，输出词级时间戳和说话人标签。

真正进入生产环境后，难点不只在模型本身。GPU AMI 版本是否固定、请求应该走实时还是异步、端点能否缩容到零，以及音频和转写结果如何存放，都会直接影响稳定性与账单。

一条音频如何变成带说话人的词级结果

WhisperX 的处理链可以拆成三个阶段：

  1. Whisper 转写：识别音频内容并生成初步分段。
  2. wav2vec2 强制对齐：把文本重新对齐到音频，得到更精确的词级起止时间。
  3. 说话人分离：识别不同说话人的时间区间，再把说话人标签映射到单词或句段。

典型结果可以整理成下面这种结构：

{
  "segments": [
    {
      "speaker": "SPEAKER_00",
      "start": 0.82,
      "end": 3.41,
      "text": "今天我们讨论发布计划。",
      "words": [
        {"word": "今天", "start": 0.82, "end": 1.21, "speaker": "SPEAKER_00"},
        {"word": "计划", "start": 2.93, "end": 3.41, "speaker": "SPEAKER_00"}
      ]
    }
  ]
}

词级时间戳适合生成字幕、点击文字跳转音频、敏感词定位和通话质检。说话人标签通常只是会话内的匿名编号，并不等于身份识别；如果业务需要把 SPEAKER_00 映射到具体姓名，还需要额外的身份确认或声纹系统，并处理相应的隐私与合规问题。

实时端点还是异步端点

两种托管方式使用相同的模型镜像，但服务目标不同。

维度 实时端点 异步端点
适合场景 短音频、交互式应用、低延迟 API 长录音、批量任务、后台处理
请求方式 客户端等待 HTTP 响应 输入放入 S3，任务排队执行，结果写回 S3
扩缩容 通常至少保留一个实例 可以配置自动缩容到零
成本特征 空闲时仍可能产生实例费用 突发任务和低频工作负载更经济
客户端设计 超时、重试和并发控制 任务状态、结果通知和幂等处理

如果输入是数十分钟的会议或访谈，不要为了复用同步 API 而把客户端超时拉得很长。异步推理更适合承接长音频：请求先进入队列，计算结束后再从 S3 获取结果。实时端点更适合长度受控、用户正在等待结果的语音片段。

还要注意，SageMaker 的“实时”表示同步请求模式，并不意味着 WhisperX 自动具备流式增量解码能力。若产品需要边说边出字，应单独评估切片、上下文拼接和流式 ASR 方案。

用 boto3 创建可切换模式的端点

下面的脚本可以直接改造成部署工具。运行前需要准备：

  • 已推送或由 AWS 提供的 WhisperX DLC 镜像 URI；
  • SageMaker 执行角色 ARN；
  • 用于异步输入、输出的 S3 桶；
  • 与镜像兼容、经过验证的 GPU 实例类型；
  • 镜像要求的 GPU 推理 AMI 版本。

不要猜测 AMI 版本。将来源部署说明中经过验证的值显式写入 INFERENCE_AMI_VERSION，避免平台默认 AMI 更新后与镜像中的 CUDA、驱动或框架组合不兼容。

安装依赖：

python -m pip install --upgrade boto3

保存为 deploy_whisperx.py：

import os
import time
import boto3

MODE = os.getenv("MODE", "async")  # async 或 realtime
REGION = os.environ["AWS_REGION"]
ROLE_ARN = os.environ["SAGEMAKER_ROLE_ARN"]
IMAGE_URI = os.environ["WHISPERX_IMAGE_URI"]
INSTANCE_TYPE = os.getenv("INSTANCE_TYPE", "ml.g5.xlarge")
AMI_VERSION = os.environ["INFERENCE_AMI_VERSION"]
S3_OUTPUT_PATH = os.getenv("S3_OUTPUT_PATH")

if MODE not in {"async", "realtime"}:
    raise ValueError("MODE must be async or realtime")
if MODE == "async" and not S3_OUTPUT_PATH:
    raise ValueError("S3_OUTPUT_PATH is required in async mode")

suffix = str(int(time.time()))
model_name = f"whisperx-{suffix}"
config_name = f"whisperx-{MODE}-{suffix}"
endpoint_name = os.getenv("ENDPOINT_NAME", f"whisperx-{MODE}")

sm = boto3.client("sagemaker", region_name=REGION)

sm.create_model(
    ModelName=model_name,
    ExecutionRoleArn=ROLE_ARN,
    PrimaryContainer={
        "Image": IMAGE_URI,
        "Environment": {
            # 变量名必须以所用 WhisperX 镜像的文档为准。
            "HF_HOME": "/tmp/huggingface"
        },
    },
)

variant = {
    "VariantName": "AllTraffic",
    "ModelName": model_name,
    "InitialInstanceCount": 1,
    "InstanceType": INSTANCE_TYPE,
    "InitialVariantWeight": 1.0,
    "InferenceAmiVersion": AMI_VERSION,
    "ContainerStartupHealthCheckTimeoutInSeconds": 900,
    "ModelDataDownloadTimeoutInSeconds": 900,
}

request = {
    "EndpointConfigName": config_name,
    "ProductionVariants": [variant],
}

if MODE == "async":
    request["AsyncInferenceConfig"] = {
        "OutputConfig": {
            "S3OutputPath": S3_OUTPUT_PATH
        },
        "ClientConfig": {
            "MaxConcurrentInvocationsPerInstance": 1
        },
    }

sm.create_endpoint_config(**request)

try:
    sm.describe_endpoint(EndpointName=endpoint_name)
except sm.exceptions.ClientError as exc:
    if exc.response["Error"]["Code"] != "ValidationException":
        raise
    sm.create_endpoint(
        EndpointName=endpoint_name,
        EndpointConfigName=config_name,
    )
else:
    sm.update_endpoint(
        EndpointName=endpoint_name,
        EndpointConfigName=config_name,
    )

print(f"Waiting for endpoint: {endpoint_name}")
sm.get_waiter("endpoint_in_service").wait(EndpointName=endpoint_name)
print(f"Endpoint is ready: {endpoint_name}")

异步部署示例：

export AWS_REGION=us-east-1
export SAGEMAKER_ROLE_ARN='arn:aws:iam::123456789012:role/SageMakerExecutionRole'
export WHISPERX_IMAGE_URI='123456789012.dkr.ecr.us-east-1.amazonaws.com/whisperx:validated-tag'
export INFERENCE_AMI_VERSION='替换为镜像文档要求的精确 AMI 版本'
export INSTANCE_TYPE='ml.g5.xlarge'
export S3_OUTPUT_PATH='s3://my-transcription-bucket/outputs/'
export ENDPOINT_NAME='whisperx-async'
export MODE='async'

python deploy_whisperx.py

脚本中的镜像 URI、实例类型和 AMI 值是部署输入，不应直接照抄占位内容。执行角色还必须拥有拉取 ECR 镜像、读取输入 S3 对象、写入输出路径以及写 CloudWatch Logs 的权限。

提交异步转写任务

异步推理要求请求体已经位于 S3。下面假设所用容器接受 JSON，并使用 audio_uri 指向音频文件；实际字段名和可选参数必须按照镜像的请求协议调整。

先创建请求文件并上传：

cat > request.json <<'JSON'
{
  "audio_uri": "s3://my-transcription-bucket/audio/meeting.wav",
  "language": "zh",
  "diarize": true,
  "word_timestamps": true,
  "min_speakers": 2,
  "max_speakers": 6
}
JSON

aws s3 cp request.json s3://my-transcription-bucket/requests/meeting.json

再通过 Python 提交任务：

import boto3

runtime = boto3.client("sagemaker-runtime", region_name="us-east-1")

response = runtime.invoke_endpoint_async(
    EndpointName="whisperx-async",
    InputLocation="s3://my-transcription-bucket/requests/meeting.json",
    ContentType="application/json",
    InferenceId="meeting-2025-001"
)

print("Output:", response["OutputLocation"])
print("Failure:", response.get("FailureLocation"))

InferenceId 应使用业务任务 ID 或幂等键，便于把 SageMaker 调用与数据库记录、日志和最终输出关联起来。生产系统不要依赖客户端不断高频轮询，可以结合 S3 事件、SNS 通知或任务表更新处理完成状态。

扩缩容与成本控制

异步端点最有价值的成本能力之一，是在没有积压请求时缩容到零。以下命令为现有异步端点注册可扩缩资源，并设置 0 到 4 个实例的范围：

aws application-autoscaling register-scalable-target \
  --service-namespace sagemaker \
  --resource-id endpoint/whisperx-async/variant/AllTraffic \
  --scalable-dimension sagemaker:variant:DesiredInstanceCount \
  --min-capacity 0 \
  --max-capacity 4

aws application-autoscaling put-scaling-policy \
  --policy-name whisperx-async-backlog \
  --service-namespace sagemaker \
  --resource-id endpoint/whisperx-async/variant/AllTraffic \
  --scalable-dimension sagemaker:variant:DesiredInstanceCount \
  --policy-type TargetTrackingScaling \
  --target-tracking-scaling-policy-configuration '{
    "TargetValue": 1.0,
    "CustomizedMetricSpecification": {
      "MetricName": "ApproximateBacklogSizePerInstance",
      "Namespace": "AWS/SageMaker",
      "Dimensions": [
        {"Name": "EndpointName", "Value": "whisperx-async"}
      ],
      "Statistic": "Average"
    },
    "ScaleInCooldown": 600,
    "ScaleOutCooldown": 60
  }'

阈值不能只看请求数量。十秒音频和两小时录音都算一个请求，但 GPU 占用时间完全不同。可以用真实流量做压测，记录音频时长、排队时间、处理时长、显存峰值和失败率，再调整：

  • MaxConcurrentInvocationsPerInstance：并发过高容易显存不足；过低则 GPU 利用率不佳。
  • 最大实例数：同时受预算、区域 GPU 配额和可用容量约束。
  • 缩容冷却时间：过短会增加冷启动，过长会让空闲 GPU 持续计费。
  • 音频长度上限：限制异常大文件，必要时预先切片。
  • Whisper 模型尺寸与计算精度：更大的模型通常更准确，但显存、延迟和成本也更高。

即使允许缩容到零，第一个任务仍可能经历实例启动、容器加载模型和缓存初始化。对延迟敏感的工作负载应保留最小容量 1，或者将实时与异步端点分开：实时端点服务短请求，异步端点消化长音频和离线任务。

上线前的检查清单

  • 固定镜像与 AMI：使用不可变镜像标签或 digest，并显式固定经过验证的 GPU AMI 版本。
  • 验证输出质量：分别测试重叠发言、噪声、口音、多语言和静音片段；说话人数量提示并非总能得到正确结果。
  • 保护数据：为 S3、ECR 和端点启用合适的加密与访问控制，定义音频和转写文本的保留周期。
  • 控制失败重试：区分临时基础设施错误与坏音频，防止无法处理的任务无限重试。
  • 监控四类指标：队列积压、端到端延迟、GPU/实例利用率、失败与超时数量。
  • 准备回滚路径：更新端点配置前保留上一版模型、镜像 digest、AMI 和参数组合。
  • 及时清理资源：实验结束后删除端点、端点配置和无用模型；仅删除模型对象不会停止端点计费。

WhisperX DLC 缩短了从模型组件到 GPU 推理服务的距离，但生产可用性来自端点模式、版本固定、扩缩容和数据治理的组合。短音频优先考虑实时端点，长音频和波峰明显的任务优先考虑异步端点；无论选择哪种方式，都应先用代表性录音验证准确率、延迟和单位音频成本，再决定实例规格与并发配置。

相关推荐