深入探讨如何在 AWS 上构建成本优化的自托管 AI 转录流水线——使用 Spot 实例、基于 SQS 队列的自动扩缩容,以及 Whisper Large-v3 Turbo——在价格上击败 OpenAI Whisper API。

大多数基于语音转文本构建的团队都默默接受 API 账单。我没有。当一个副业项目每月在转录 API 费用上花费 180 美元 时,我在 AWS 上重建了整个流水线,将处理成本降至 每小时 0.10 美元以下——比 OpenAI Whisper API 本身更便宜。

本文将介绍架构、权衡以及使其成功的具体决策。这里的一切都驱动着我作为独立创始人构建的 AI 转录平台 StrikeScribe,因此这些数据来自真实的生产工作负载,而非基准测试。

如果你正在构建基于 Whisper、Deepgram、AssemblyAI 或 OpenAI 音频 API 的任何产品,这就是大多数人错过的套利机会。

问题:托管转录 API 在成本上无法扩展

托管 API 非常适合上手。你发送音频,就能得到文本。但定价模式会惩罚用量:

  • OpenAI Whisper API: 每小时约 0.36 美元(按 0.006 美元/分钟计算)
  • Otter、Fireflies 等: 订阅阶梯会限制你的时长并收取高昂的超额费用
  • AssemblyAI / Deepgram: 具有竞争力,但仍按分钟计费,且随用量线性增长

当你每月处理数百小时——会议、播客、调研通话、IoT 音频流——时,那条线性成本曲线就会成为主要开支。你是在为本可以直接租用的 GPU 算力支付溢价。

洞见是:底层模型(Whisper)是开源的。 你支付的是他人的便利层费用。如果你能高效地自行运行推理,经济效益就会逆转。

架构

以下是高层流水线:

Upload → S3 → SQS Queue → Autoscaling Worker Group (GPU Spot Instances)
   → Whisper Large-v3 Turbo → Pyannote (diarization) → Postgres → Client

Enter fullscreen mode Exit fullscreen mode

让我拆解真正影响成本的部分。

1. Whisper Large-v3 Turbo:质量与速度的甜蜜点

Whisper Large-v3 Turbo 能以接近 Large-v3 的准确率实现更短的推理时间。对于成本敏感的流水线,推理速度就是成本,因为你在为 GPU 秒数买单。Turbo 大幅降低了每音频小时所需的秒数。

为实现更快吞吐量,我通过 faster-whisper(CTranslate2 后端)运行它,其内存和计算效率显著高于参考实现:

from faster_whisper import WhisperModel

# int8_float16 keeps VRAM low while preserving accuracy
model = WhisperModel(
    "large-v3",
    device="cuda",
    compute_type="int8_float16",
)

segments, info = model.transcribe(
    "audio.wav",
    beam_size=5,
    vad_filter=True,           # skip silence, save compute
    vad_parameters={"min_silence_duration_ms": 500},
)

for segment in segments:
    print(f"[{segment.start:.2f} -> {segment.end:.2f}] {segment.text}")

Enter fullscreen mode Exit fullscreen mode

两个低成本优化:

  • vad_filter=True——语音活动检测会完全跳过静音区域。真实录音(会议、通话)充满死寂。这一项本身就显著降低了算力消耗。
  • int8_float16 量化——让你能在更廉价、显存更少的 GPU 上运行 Large-v3,且对大多数音频的准确率影响微乎其微。

2. SQS + 队列深度自动扩缩容

天真的做法是保持一个始终在线的 GPU 实例。那是昂贵的做法——你为闲置时间付费。

相反,我通过 SQS 队列 将上传与处理解耦。一个轻量级的 Node.js 控制器轮询队列深度,并根据积压情况扩缩容工作节点集群:

import { SQSClient, GetQueueAttributesCommand } from "@aws-sdk/client-sqs";
import { AutoScalingClient, SetDesiredCapacityCommand } from "@aws-sdk/client-auto-scaling";

const sqs = new SQSClient({ region: "us-east-1" });
const asg = new AutoScalingClient({ region: "us-east-1" });

async function scaleWorkers() {
  const { Attributes } = await sqs.send(
    new GetQueueAttributesCommand({
      QueueUrl: process.env.QUEUE_URL,
      AttributeNames: ["ApproximateNumberOfMessages"],
    })
  );

  const backlog = parseInt(Attributes.ApproximateNumberOfMessages, 10);

  // one worker per N queued jobs, capped to control spend const desired = Math.min(Math.ceil(backlog / 5), MAX_WORKERS);

  await asg.send(
    new SetDesiredCapacityCommand({
      AutoScalingGroupName: process.env.ASG_NAME,
      DesiredCapacity: desired,
    })
  );
}

Enter fullscreen mode Exit fullscreen mode

当队列为空时,集群会缩容至零 GPU 工作节点。你仅在有音频需要处理时才为 GPU 时间付费。这是最重要的结构性成本杠杆。

3. Spot 实例(以及应对中断)

GPU Spot 实例比按需实例便宜 60–90%。代价是:AWS 可能在 2 分钟警告后回收实例。对于像转录这样的无状态批处理工作负载,这非常合适——前提是你能优雅地处理中断。

工作节点会监听 Spot 中断通知,并将正在处理的任务重新入队,确保不会丢失任何内容:

// Poll the instance metadata endpoint for the interruption signal
async function checkForInterruption() {
  try {
    const res = await fetch(
      "http://169.254.169.254/latest/meta-data/spot/instance-action",
      { signal: AbortSignal.timeout(1000) }
    );
    if (res.status === 200) {
      // Reclaim imminent — put the current job back on the queue
      await requeueCurrentJob();
      await gracefulShutdown();
    }
  } catch {
    // 404 = not interrupted, carry on
  }
}

setInterval(checkForInterruption, 5000);

Enter fullscreen mode Exit fullscreen mode

因为任务是幂等的且由队列支撑,被中断的任务只需由其他工作节点接手。不会有工作丢失,也无需人工恢复。这种弹性正是 Spot 实例能用于生产而非仅用于业余批处理任务的关键。

4. 使用 Pyannote 进行说话人分离

原始转录文本只是价值的一半——知道说了什么才能将一堵文字墙变成可用的会议纪要。Pyannote 负责说话人分离:

from pyannote.audio import Pipeline

pipeline = Pipeline.from_pretrained(
    "pyannote/speaker-diarization-3.1",
    use_auth_token=HF_TOKEN,
)

diarization = pipeline("audio.wav")

for turn, _, speaker in diarization.itertracks(yield_label=True):
    print(f"{speaker}: {turn.start:.1f}s - {turn.end:.1f}s")

Enter fullscreen mode Exit fullscreen mode

随后将分离后的片段与 Whisper 的时间戳对齐,即可为每一行标注说话人。这会增加算力,但它正是“一份转录文本”与“会议智能”之间的区别。

成果

该流水线在生产环境中的表现:

  • 处理成本:低于每音频小时 0.10 美元——击败 OpenAI Whisper API
  • 一次压力测试:来自 9 台树莓派 24/7 录制的 216 小时音频 总处理费用 不足 3 美元
  • 与“始终在线 / 托管 API”基线相比,整体基础设施成本降低约 60%

四个杠杆——Turbo 模型、VAD 静音跳过、缩容至零的自动扩缩容,以及 Spot 定价——的叠加效应是乘数级的,而非简单的相加。每一项单独都有帮助;叠加起来,它们改变了整个单位经济。

何时不应该自托管

公平地说,这并非免费午餐。自托管对我而言合理的原因是:

  • 用量足够高,以至于 API 费用远超工程时间成本
  • 工作负载是批处理而非实时(Spot + 队列延迟可接受)
  • 我愿意承担 AWS 基础设施、GPU 驱动和模型更新的运维责任

如果你用量低、需要亚秒级的实时字幕,或不想维护基础设施,那么托管 API 确实是正确的选择。盈亏平衡点通常在每月数百小时的低位区间。

要点

套利模型——找到竞争对手定价过高的环节,然后用你可控的原语更便宜地构建——在当前 AI 基础设施领域非常真实。大量“AI 产品”的定价只是 GPU 算力和开源模型上的溢价。

如果你正在处理有意义的音频量,我验证有效的技术栈是:

  1. faster-whisper + Large-v3 Turbo,搭配 int8_float16 和 VAD
  2. 可缩容至零的 SQS 队列深度自动扩缩容
  3. 具备中断安全、幂等任务的 GPU Spot 实例
  4. 使用 Pyannote 进行说话人分离以增加真实的产品价值

我已将所有这些集成到 StrikeScribe 中——上传音频或视频,即可获得可搜索的转录文本加上结构化的 AI 洞见,无需注册,也无需会议机器人。如果你想看到这个流水线的输出端实际运行,这就是最简单的方式。

欢迎在评论区提出架构问题——尤其是 Spot 中断处理和说话人分离对齐,这通常是大家容易卡住的两点。