在 Amazon SageMaker AI 上用 Kokoro-82M 实现语音智能体低延迟流式合成

客户向 AI 语音智能体提出一个问题并等待。智能体其实已经有了答案，正在生成一段多段落的长回复，但客户在五秒内什么都听不到。等到答案终于传来时，客户已经感到沮丧、困惑，并且不再专注聆听。

文本转语音（TTS）集成正是会造成这种间隙。一句话的响应可能在 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 实时端点的请求路径

先决条件

开始之前，您需要：

• 一个具有创建 SageMaker AI 端点权限的 AWS 账户；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 步：在智能体生成句子的同时进行流式传输

在会话行为就位之后，针对延迟问题的客户端解决方案是一个类，它能规范化以两种形式到达的文本：一种是预先已知的完整字符串，另一种是实时的 token 流。无论哪种形式，它都产生相同的输出：完整的句子，每个句子一完成就可以立即发送。

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 都要付出相同的代价，因此二者之间测得的差异纯粹是时序上的：即第一个音频块何时发出。

测试与验证

为了分离 TTS 传输本身对首音频延迟的贡献，我们以两种方式发送同一段 267 字符、四个句子的文本：作为一条消息（batch），以及只发送其第一句（incremental）。我们针对实时端点各运行了五次试验。

如前文所测，只发送第一句使首音频字节的平均时间从 329.3 毫秒降至 156.7 毫秒，相对提升 52.4%，且这一提升仅归因于 TTS 传输本身。

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

使用相同的方法，双向模式的首音频延迟在本地容器上比批量模式快 2.4 倍，在已部署的 AgentCore 运行时上快 3.5 倍。

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

图 4：batch 和 bidi 模式在本地和已部署运行时上的相对首音频时间

在已部署场景中，一旦加入真实的网络跳转，差距会拉大而非缩小。run_bidi 将这一网络跳转吸收在它已经在生成与合成之间利用的重叠之中。run_batch 没有等价的重叠来掩盖这一代价，因为它一次只做一件事，所以每一毫秒新增的网络开销都直接落到它的串行等待上。

要在您自己的部署中复现这一对比，请运行智能体容器提供的相同的 bidi-vs-batch 测试：

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

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

清理资源

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

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 仓库，因为已存储的模型检查点和容器镜像在移除之前会持续产生少量存储费用。

结论

我们展示了如何从语音智能体中消除一种可避免的延迟来源——在开始语音合成之前等待智能体的完整响应——以及如何量化另一种更隐蔽的成本：SSE 中的 base64 编码音频会使载荷体积膨胀约三分之一，而且如果不做缓冲，还可能与 MP3 帧边界错位。在智能体每生成一句话时就将其流式传输到 SageMaker AI 双向端点，仅此一项就把首音频时间缩短了一半以上，而在已部署的智能体中缩短了 2.4 到 3.5 倍。在该会话上多路复用二进制音频和 JSON 控制帧（而不是把音频编码为文本），消除了简单 SSE 中继所需的体积膨胀和帧重新对齐工作。

如果您正在为已经通过 SSE 流式传输文本的智能体添加语音功能，可以从第 4 步的 base64 方案入手，并在延迟成为优先事项时再重新审视统一传输方案。要了解更多信息，请参阅 Amazon SageMaker AI 实时推理文档 和 Amazon Bedrock AgentCore 文档。如果您希望在您自己的架构中应用此模式，请联系您的 AWS 客户团队。

关于作者