文本转语音应用的体验差异，往往不在于模型能否生成音频，而在于用户要等多久才能听到第一个音节。将 Qwen3-TTS 部署到 Amazon SageMaker AI，并使用 AWS vLLM-Omni Deep Learning Container 持续返回音频块，可以让前端在整段语音生成完成之前就开始播放。
这类架构的重点不只是“部署一个模型端点”，还包括持久双向连接、音频分块、前端增量播放，以及连接中断和并发压力的处理。本文聚焦第一阶段：部署 TTS 模型，并通过 Gradio 构建一个可交互的流式语音界面。
请求—响应不够：为什么要使用持久连接
传统推理接口通常采用一次 HTTP 请求对应一次完整响应的模式：客户端提交文本，服务端生成整段 WAV 文件，然后一次性返回。实现简单，但首包延迟会随文本长度和生成耗时上升。
流式 TTS 则把过程拆成多个连续事件：
- 客户端发送文本、音色和采样参数。
- 推理服务开始生成音频 token 或波形块。
- 服务端立即推送已经生成的音频块。
- 客户端一边接收，一边解码和播放。
- 服务端发送完成事件，连接可以保留并处理后续请求。
一个典型的数据路径如下：
Browser
│ Gradio 的持续连接
▼
Gradio application
│ 经过认证的双向流
▼
SageMaker endpoint
│
▼
AWS vLLM-Omni container + Qwen3-TTS
不要让浏览器直接持有 AWS 长期凭证。更稳妥的方式是由 Gradio 后端或独立 API 网关负责身份验证、请求签名和上游连接，浏览器只连接应用层服务。
用可配置脚本创建 SageMaker 端点
实际使用的镜像 URI、Qwen3-TTS 模型 ID和容器环境变量，应以目标区域中的 AWS vLLM-Omni DLC 及其启动约定为准。下面的脚本故意不猜测容器内部的环境变量名称，而是通过 MODEL_ENV_JSON 注入，这样可以直接按所用镜像的要求调整。
先安装依赖：
python -m venv .venv
source .venv/bin/activate
pip install "sagemaker>=2.220,<3" boto3
创建 deploy.py：
import json
import os
import sagemaker
from sagemaker.model import Model
region = os.environ.get("AWS_REGION", "us-east-1")
role_arn = os.environ["SAGEMAKER_ROLE_ARN"]
image_uri = os.environ["VLLM_OMNI_IMAGE_URI"]
endpoint_name = os.environ.get("ENDPOINT_NAME", "qwen3-tts-vllm-omni")
instance_type = os.environ.get("INSTANCE_TYPE", "ml.g5.2xlarge")
# 示例：'{"MODEL_ID":"<qwen3-tts-model-id>","DTYPE":"bfloat16"}'
# 键名必须与实际 DLC 的启动契约一致。
container_env = json.loads(os.environ.get("MODEL_ENV_JSON", "{}"))
if not container_env:
raise RuntimeError("请通过 MODEL_ENV_JSON 提供 DLC 所需的模型配置")
session = sagemaker.Session(boto_session=sagemaker.boto3.Session(region_name=region))
model = Model(
image_uri=image_uri,
role=role_arn,
env=container_env,
sagemaker_session=session,
)
model.deploy(
initial_instance_count=1,
instance_type=instance_type,
endpoint_name=endpoint_name,
container_startup_health_check_timeout=1800,
)
print(f"Endpoint ready: {endpoint_name}")
配置变量并部署：
export AWS_REGION=us-east-1
export SAGEMAKER_ROLE_ARN='arn:aws:iam::<account-id>:role/<sagemaker-execution-role>'
export VLLM_OMNI_IMAGE_URI='<regional-vllm-omni-dlc-image-uri>'
export ENDPOINT_NAME='qwen3-tts-vllm-omni'
export INSTANCE_TYPE='ml.g5.2xlarge'
export MODEL_ENV_JSON='{"MODEL_ID":"<qwen3-tts-model-id>","DTYPE":"bfloat16"}'
python deploy.py
运行前需要确认三件事：实例类型能够容纳模型；执行角色有拉取镜像和读取模型文件的权限；模型许可证允许目标使用场景。GPU 实例会持续计费，不测试时应删除端点，而不只是停止本地 Gradio 程序。
将音频块接入 Gradio
下面是一个可以改造的最小 Gradio 客户端。它假设上游暴露 WebSocket，并采用以下示例协议：客户端发送一条 JSON synthesize 消息，服务端持续返回 PCM16 单声道二进制块，最后返回 {"event":"done"}。
这是一个应用层协议示例，并不是对 SageMaker 或 DLC 固定报文格式的声明。接入实际端点时，需要按照所用连接器修改 URL、认证方式和消息字段；如果上游返回 WAV、Opus 或带元数据的帧，也要相应更换解码逻辑。
安装依赖：
pip install "gradio>=4.44,<6" "websockets>=12,<16" "numpy>=1.26,<3"
创建 app.py：
import asyncio
import json
import os
import gradio as gr
import numpy as np
import websockets
VOICE_WS_URL = os.environ.get("VOICE_WS_URL", "ws://127.0.0.1:8080/tts")
VOICE_AUTH_TOKEN = os.environ.get("VOICE_AUTH_TOKEN", "")
SAMPLE_RATE = int(os.environ.get("SAMPLE_RATE", "24000"))
async def synthesize(text: str, voice: str):
text = text.strip()
if not text:
raise gr.Error("请输入要合成的文本")
headers = {}
if VOICE_AUTH_TOKEN:
headers["Authorization"] = f"Bearer {VOICE_AUTH_TOKEN}"
# websockets 14+ 使用 additional_headers。
async with websockets.connect(
VOICE_WS_URL,
additional_headers=headers,
open_timeout=20,
ping_interval=20,
max_size=None,
) as ws:
await ws.send(json.dumps({
"event": "synthesize",
"text": text,
"voice": voice,
"format": "pcm_s16le",
"sample_rate": SAMPLE_RATE,
}))
while True:
message = await asyncio.wait_for(ws.recv(), timeout=60)
if isinstance(message, bytes):
pcm = np.frombuffer(message, dtype="<i2")
# Gradio 可以消费 (采样率, NumPy 数组) 形式的流式音频。
yield SAMPLE_RATE, pcm
continue
event = json.loads(message)
if event.get("event") == "done":
break
if event.get("event") == "error":
raise gr.Error(event.get("message", "上游语音生成失败"))
with gr.Blocks(title="Qwen3-TTS Streaming Demo") as demo:
gr.Markdown("## Qwen3-TTS 实时语音生成")
text = gr.Textbox(
label="文本",
lines=5,
value="你好，这是一段边生成边播放的语音。",
)
voice = gr.Textbox(label="音色", value="default")
generate = gr.Button("开始生成", variant="primary")
audio = gr.Audio(label="流式音频", streaming=True, autoplay=True)
generate.click(
fn=synthesize,
inputs=[text, voice],
outputs=audio,
concurrency_limit=4,
)
if __name__ == "__main__":
demo.queue(default_concurrency_limit=4).launch(
server_name="0.0.0.0",
server_port=7860,
)
启动界面：
export VOICE_WS_URL='wss://<your-streaming-gateway>/tts'
export VOICE_AUTH_TOKEN='<short-lived-token>'
export SAMPLE_RATE=24000
python app.py
生产环境中，上游连接通常还需要 AWS SigV4 签名、临时凭证或由受控代理完成鉴权。不要为了快速演示而把 IAM Access Key 写进 JavaScript、镜像或 Git 仓库。
实时体验取决于整条链路
模型生成速度只是其中一个指标。调试时至少应记录以下延迟：
- 连接建立时间：包括 DNS、TLS、认证和 WebSocket 握手。
- 首音频块延迟：从提交文本到收到第一块可播放 PCM 的时间。
- 实时系数：生成耗时与音频时长的比值；小于 1 才能持续赶上播放速度。
- 块间抖动：音频块到达不均匀会造成停顿，即使平均吞吐量足够。
- 端到端完成时间：用于容量规划和并发评估。
音频块不是越小越好。块太大会增加首包等待，块太小则会放大协议、Python 调度和网络开销。可以从 20～100 毫秒音频对应的数据量开始测试，再结合网络状况和播放器缓冲策略调整。这个区间是工程起点，不是 Qwen3-TTS 的固定要求。
还要为长文本设置上限。较稳妥的做法是按句子切分输入，在标点处排队生成，并保留少量播放缓冲。这样既能降低单次请求风险，也更容易取消尚未播放的后续内容。
上线前检查清单
将演示升级为服务时，建议逐项确认：
- 容器镜像、模型权重和实例类型已经在同一区域验证。
- Gradio 或网关使用短期凭证，浏览器无法接触 AWS 密钥。
- 连接具备超时、心跳、取消和断线清理机制。
- 客户端明确知道采样率、声道数、字节序和音频编码。
- 端点设置并发上限、输入长度限制和请求级追踪 ID。
- 监控首音频块延迟、错误率、GPU 利用率和活跃连接数。
- 对用户输入、生成内容、音色授权和日志保留实施合规控制。
- 测试结束后删除闲置 SageMaker 端点，避免 GPU 成本继续累积。
这一阶段解决的是“文本如何尽快变成可播放语音”。真正的双向语音助手还需要补上语音输入、端点检测、语音转写、打断控制和对话状态管理。先把 TTS 流稳定下来，再扩展完整语音回路，通常比一次性叠加所有组件更容易定位性能问题。