在 Amazon SageMaker AI 上用 Kokoro TTS 为 AI 智能体流式输出低延迟语音

客户向 AI 语音代理提出一个问题，然后等待。代理其实已经有了答案，正在生成一段很长的多段落回复，但客户在五秒钟内什么也没听到。等到答案到达时，客户已经感到沮丧、困惑，也不再关注了。

文本转语音集成恰恰造成了这一差距。一句回复可能在 600 毫秒内生成，但完整回复需要五秒钟，而集成会等待全部内容生成完毕后才开始合成。

本文展示如何弥合这一差距。只要代理生成出一句话，我们就把它流式传输到 Amazon SageMaker AI 上托管的 Kokoro-82M 文本转语音（TTS）模型，因此合成从第一句话就开始，而不是等到回复完成。把音频传回浏览器也需要自己的解决方案，因此我们还会介绍实现这一点的两种传输方式，以及它们在实践中各自的成本。

解决方案概述

该解决方案通过让合成与生成并行流式进行，而不是等待生成完成，来弥合这一差距。当代理生成每一句话时，该句子会立即流式传输到托管的 Kokoro-82M 文本转语音模型，因此音频播放可以在代理说完完整回复之前就开始。Kokoro-82M 仅有 8200 万个参数，小到足以在单台中等 GPU 实例上低成本运行。这使得处理真实流量的语音代理能够同时保持低延迟和低推理成本。

Kokoro-82M 在 SageMaker AI 实时端点的自定义容器中运行，部署在 ml.g5.xlarge GPU 实例上。该容器暴露两条路由：POST /invocations，为单个完整输入流式返回响应；以及 WS /invocations-bidirectional-stream，在双向通信中通过一条连接承载多条消息。

下图展示了从客户端到模型的请求路径：

图 1：通过双向流从客户端到 SageMaker AI 实时端点的请求路径

先决条件

开始之前，你需要：

• 一个 AWS 账户，具有创建 SageMaker AI 端点、AgentCore 运行时（Amazon Bedrock AgentCore 的一项能力）、AWS Identity and Access Management（IAM）角色以及 Amazon Elastic Container Registry（Amazon ECR）存储库的权限，至少包括 sagemaker:CreateEndpoint、sagemaker:CreateEndpointConfig、sagemaker:CreateModel、sagemaker:InvokeEndpointWithBidirectionalStream 和 bedrock:InvokeModelWithResponseStream 操作。
• 示例容器固定使用了一个带有已知安全公告的旧版 PyTorch（PYSEC-2026-2286，已在 torch 2.10.0 中修复）。在任何非概念验证用途之前，请审查并升级该固定版本。
• AWS Cloud Development Kit (AWS CDK) v2，已在目标账户和 AWS 区域中安装并完成引导。
• 从声明了 com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true Docker 标签的容器部署的 SageMaker AI 实时端点。
• Python 3.12 或更高版本、uv 以及 Docker，用于构建容器镜像（底层 HTTP/2 客户端仅为 Python 3.12 及更新版本提供预构建 wheel）。
• 支持 invoke_endpoint_with_bidirectional_stream 的 AWS SDK for Python（Boto3）。
• 熟悉 WebSockets 和 Starlette 或类似的 ASGI 框架。
• 本文引用的 GitHub 存储库 中的部署代码。

演练

演练首先部署端点和代理，然后逐步介绍客户端侧的更改，使增量流式传输能够端到端工作。

部署端点和代理

Kokoro-82M 端点和代理运行时通过两个 CLI 工具部署，每个工具都封装了自己的 AWS CDK 堆栈：

uv run speech-infra deploy kokoro-82m
uv run agent-infra deploy --bedrock-model-id <your-bedrock-model-id>

speech-infra deploy 配置 Kokoro-82M 端点。这包括基础栈和模型缓存栈，它们创建底层 IAM 角色并将模型权重同步到 Amazon Simple Storage Service（Amazon S3）。随后，端点栈构建容器镜像，并创建 SageMaker AI 模型、端点配置和端点。agent-infra deploy 配置托管 Strands 代理的 Amazon Bedrock AgentCore 运行时，该代理会调用你刚刚部署的端点，因此请先部署端点。

步骤 1：处理 SageMaker AI 实际发送的 WebSocket 帧类型

SageMaker AI 使用的双向传输会将客户端消息的每一部分作为二进制 WebSocket 帧发送到你的容器。编写 Starlette WebSocket 处理程序的常见方式是调用 receive_text()，它会读取 message["text"]，并在该键不存在时抛出 KeyError。由于此客户端发送的每条消息都以二进制形式到达，该处理程序会在每个请求上失败。

这种故障常被误诊。一个将 str(e) 转发给客户端的通用异常处理程序会把 KeyError('text') 变成一条错误帧，内容为 'text'，看起来像是缺少字段，而不是帧格式问题。端点本身全程报告 InService，因为它接受连接并响应每条消息，但每次都会出错。

显式接受任一帧类型，而不是假设只会出现一种：

from starlette.websockets import WebSocket, WebSocketDisconnect

async def _receive_message(websocket: WebSocket) -> str:
    message = await websocket.receive()
    if message["type"] == "websocket.disconnect":
        raise WebSocketDisconnect(message.get("code", 1000))
    payload = message.get("text")
    if payload is None:
        payload = message.get("bytes", b"").decode("utf-8")
    return payload

当没有文本字段时，它会将字节按 UTF-8 解码，而不是假设只有一种帧类型。通过发送二进制帧来测试，因为这种故障模式会产生格式正确的错误帧和健康的端点状态，而不是明显的崩溃。

步骤 2：在消息之间保持会话打开

当你的容器完成一条消息的音频合成后，本能反应是关闭会话，就像 HTTP 周期结束那样。但在这里这是错误的。在发送一条消息的音频和一个 synthesis_complete 控制帧后，容器循环应回到等待 receive()，以在同一仍然打开的会话上接收下一条消息。仅当客户端发送显式的 {"type": "close"} 消息时才关闭连接。

下图展示了一个会话按顺序承载两个句子，仅在最后关闭：

图 2：一个双向会话按顺序承载两个句子，仅在最后关闭

这样，你可以在一个持久会话上连续发送多个句子，而无需在它们之间重新连接。在一次响应后急于关闭会迫使每个句子建立新连接，从而抵消大部分延迟收益。

步骤 3：在代理生成句子时流式传输它们

会话行为就位后，针对延迟问题的客户端修复方案是一个类，它规范化两种形式的文本：预先已知的完整字符串，或实时令牌流。无论哪种方式，它都产生相同的输出：完整的句子，一旦每个句子完成即可发送。

import re

_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+")


class IncrementalSentenceChunker:
    def __init__(self) -> None:
        self._buffer = ""

    def feed(self, fragment: str) -> list[str]:
        self._buffer += fragment
        parts = _SENTENCE_RE.split(self._buffer)
        if len(parts) == 1:
            return []
        self._buffer = parts[-1]
        return [s.strip() for s in parts[:-1] if s.strip()]

    def flush(self) -> str | None:
        remainder = self._buffer.strip()
        self._buffer = ""
        return remainder or None

每当文本片段到达时，调用一次 feed()。它返回任何新完成的句子，并保留尾部文本。在输入结束时调用一次 flush()，以释放剩余的任何部分文本，即使没有尾部标点。这可以防止在句号到达之前就结束的流丢弃最后一个句子。

这不是并发合成。每个双向处理器都是一个严格串行的循环：接收一条消息，合成其音频，发送该音频，发送 synthesis_complete，然后接收下一条消息。第二句的音频不会在第一句仍在传输时生成。增量发送带来的好处更有限：第一句的合成在它一存在时就开始，而不是等待完整响应。对于第一句和最后一句之间具有真实生成时间的多句响应，这就是全部的延迟收益。

我们在一个实时端点上验证了这一点，在单个会话中发送了一个三句输入。三个数据块到达时，其音频与各自发送的文本相匹配：

第 4 步：选择一种将音频传回客户端的传输方式

将文本增量流式传入模型解决了问题的一半。另一半是将生成的音频输出到任何渲染它的地方，通常是浏览器。大多数智能体和聊天前端已经通过 Server-Sent Events（SSE）流式传输文本：单向，且大多数浏览器原生支持。SSE 是一种文本协议，并非为承载原始二进制载荷而构建。为现有 SSE 管道添加语音的团队自然会采用它已有的模式：对音频进行 base64 编码，并将其包装在 JSON 事件中。

我们在 24 个片段上测量了该方法的成本，通过 Amazon Transcribe 确认不同编码之间的质量没有差异，并比较了 MP3 与 AAC：

Base64 会使每个载荷膨胀接近预期的 33%。更微妙的成本是帧对齐。SSE 事件由空行分隔，但其中的音频被分隔为 MP3 帧。这两种方案不一定与传输层对数据分块的方式对齐。在一次测量的响应中，四个逻辑 MP3 帧以六个传输块到达，其中两个在帧中间结束。将每个传输块视为一个完整单元的消费者会在普通、健康的流量上失败。修复方法是缓冲传入字节，仅在找到分隔符后才采取行动：

buffer = b""
for event in resp["Body"]:
    if "PayloadPart" not in event:
        continue
    buffer += event["PayloadPart"]["Bytes"]
    while b"\n\n" in buffer:
        frame, buffer = buffer.split(b"\n\n", 1)
        handle(frame)

如今用于面向模型这一段的替代方案完全跳过了 base64 和 SSE 帧封装。它在已经承载增量文本输入的同一双向会话上多路复用原始二进制音频和 JSON 控制帧，并用一条规则区分它们。

import json


def _decode_control_frame(chunk: bytes) -> dict | None:
    try:
        decoded = json.loads(chunk.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    return decoded if isinstance(decoded, dict) else None

无法解码为 JSON 对象的数据块就是音频，按接收原样转发，路径中没有任何编码步骤。有一个边缘情况很重要：原始 PCM 音频可以合法地以字节 0x7b 开头，即 ASCII 字符 {，与 JSON 对象的开头字符相同。这条规则避免了对该音频的误分类，因为它要求整个数据块解析为有效 JSON，而不仅仅是第一个字节，而 PCM 帧基本上永远不会做到这一点。

注意：如果你的前端已经有用于文本的可用 SSE 管道，那么在 SSE 事件中使用 base64 编码的音频是务实的选择。它带来大约 33% 的额外载荷和一次缓冲重组步骤的成本，但复用了你已经在运营的基础设施。如果你是从头构建语音路径，或者延迟是优先事项，那么统一的二进制与 JSON 方法避免了这两项成本，并复用了已经承载文本的连接。

第 5 步：在真实智能体中整合起来

更完整的端到端案例是一个实时运行的生成式智能体，它逐 token 生成响应，在句子一存在之前无法知道句子四是什么。此示例作为 Strands 智能体运行，由 Amazon Bedrock 提供支持，部署到 Amazon Bedrock AgentCore 运行时，并使用第 4 步中的模式调用 Kokoro 端点。此设计连接了三条通信链路：面向浏览器、将 Server-Sent Events 回传给调用方的那条链路；智能体为生成文本而向 Amazon Bedrock 发起的流式调用；以及智能体与 Kokoro 端点之间用于合成的双向会话。

下图展示了这三条通信链路，以及生成与合成之间的重叠发生在哪里：

图 3：连接调用方、Amazon Bedrock 和 Kokoro 端点的三条通信链路

请求负载携带一个 mode 字段，取值为 "bidi" 或 "batch"，会被分发到同一个入口函数内的两条代码路径。两者之间的基准测试只改变一个变量：合成何时开始。在 run_bidi 中，两个任务并发运行。一个任务消费 Bedrock 文本流，并将已完成的句子送入分块器。另一个任务消费 Kokoro 会话，并在音频块到达时将其发出。两者都推入同一个共享队列，因此消费者按真实到达顺序产出事件：较早句子的音频可以在 Bedrock 仍在生成较晚句子时到达调用方。run_batch 则刻意相反：它等待完整响应后才发起一次合成调用，作为公平的基线保留。

反方向上，从智能体容器回到调用方的链路确实使用了 base64，原因与第 4 步所述不同。BedrockAgentCoreApp 的入口点必须产出可 JSON 序列化的普通字典，其 SDK 会自动将其转换为 data: {json}\n\n 帧。无法向它传入原始二进制帧。这是 JSON 原生托管运行时的结果，而不是第 4 步所测量并放弃的选择。run_bidi 和 run_batch 在这条链路上都付出相同的代价，因此它们之间测得的差异纯粹是时序上的：第一个音频块何时发出。

测试与验证

我们通过以两种方式发送同一段 267 个字符、四句话的文本，隔离出 TTS 传输本身对首个音频时间的贡献：作为一条消息发送（批处理），以及仅发送其第一句话（增量式）。我们针对实时端点各运行了五次试验。

如前所述，仅发送第一句话将首个音频字节的平均时间从 329.3 毫秒缩短到 156.7 毫秒，相对提升 52.4%，且仅归因于 TTS 传输本身。

随后，我们使用真实语言模型生成文本，测量了更完整的端到端案例：将完全相同的提示词分别通过 run_bidi 和 run_batch，针对同一个实时 Bedrock 模型和端点运行，因此唯一变量是合成何时开始。

采用相同方法，在本地容器上测得 bidi 模式的首个音频时间比批处理模式快 2.4 倍，在部署的 AgentCore 运行时上快 3.5 倍。

下图展示了两种环境下的这一差距：

图 4：批处理模式与 bidi 模式在本地和部署运行时上的相对首个音频时间

一旦加入真实网络跳转，部署场景下的差距会扩大而非缩小。run_bidi 在它已经利用的生成与合成重叠中吸收了该跳转。run_batch 没有等效的重叠来隐藏该成本，因为它一次只做一件事，所以每一毫秒新增的网络开销都直接落在其串行等待上。

要针对你自己的部署复现这一对比，请运行 agent 容器所暴露的同一个 bidi-vs-batch 测试：

uv run agent-infra demo --endpoint-url <your-endpoint-url> --prompt "<your prompt>"

这会将提示词分别通过两种模式发送，并报告每种模式下首个 audio_chunk 事件相对于 agent_start 的时间戳。如果你看到的是错误，请检查第 1 步中的 KeyError: 'text' 症状。如果端点报告 InService 而每个请求都失败，则表明存在该 bug，而不是部署问题。

清理

为避免持续产生费用，请按与部署相反的顺序拆除本演练创建的资源，先拆除 agent，再拆除端点：

uv run agent-infra destroy --bedrock-model-id <your-bedrock-model-id>
uv run speech-infra destroy kokoro-82m

agent-infra destroy 会移除 Amazon Bedrock AgentCore 运行时资源。speech-infra destroy 会移除 SageMaker AI 模型、端点配置和端点。两者都不会清空用于缓存模型权重的 Amazon S3 存储桶，也不会删除存放容器镜像的 Amazon ECR 仓库，因为二者都可在多个模型间共享。如果你不再需要它们，请手动清空并删除 Amazon S3 存储桶，并移除 Amazon ECR 仓库，因为存储的模型检查点和容器镜像在移除前会持续产生少量存储费用。

结论

我们展示了如何从语音 agent 中消除一个可避免的延迟来源，即在开始语音合成前等待 agent 的完整响应；以及如何量化第二个更隐蔽的成本：SSE 中的 base64 编码音频会使载荷膨胀约三分之一，并且如果不为此进行缓冲，还可能与 MP3 帧边界错位。在隔离测试中，只要 agent 生成每个句子就将其流式传输到 SageMaker AI 双向端点，可将首个音频的时间缩短一半以上；在已部署的 agent 中则可缩短 2.4 到 3.5 倍。在该会话上多路复用二进制音频和 JSON 控制帧，而不是将音频编码为文本，可消除朴素 SSE 中继所需的大小膨胀和帧重新对齐工作。

如果你正在为已经通过 SSE 流式传输文本的 agent 添加语音，请先从第 4 步中的 base64 方法开始，并在延迟成为优先事项后重新考虑统一传输。要了解更多信息，请参阅 Amazon SageMaker AI 实时推理文档 和 Amazon Bedrock AgentCore 文档。如果你希望获得帮助，将此模式应用到自己的架构中，请联系你的 AWS 客户团队。

关于作者

Joseph Cottingham

Joseph 是 AWS 生成式 AI 创新中心的一名机器学习工程师，他在那里设计和交付 AI 系统，帮助组织从想法走向生产。他的技术背景涵盖固件、软件和机器学习，并在高吞吐、事件驱动的架构方面拥有深厚专业知识，这些架构可让 AI 工作负载在大规模下保持高性能。Joseph 专注于最先进研究与运营现实的交叉领域，交付能够实质性改变组织及其运营系统的解决方案。