用逐句语音合成降低 Agent 播报延迟

语音 Agent 已经开始生成答案，用户却还在等声音，问题可能出在调用顺序：系统等整段文本生成完，才交给语音模型。AWS 这篇文章用 Kokoro-82M 和 SageMaker AI 双向流，把合成起点提前到第一句完成时，实测缩短了首音频等待时间。

对做前端和全栈的开发者，更有用的是它拆开的三个问题：怎样从 token 流里取出完整句子，怎样让连接持续接收后续句子，以及音频经过 SSE 时会付出什么代价。方案有实测，但「流式」能加速到哪一步，需要分清。

首句生成后就合成，究竟快在哪里

原文开头举了一个场景：一句话可能在 600 毫秒内生成，完整回答却要五秒。如果 TTS（文本转语音）等全文结束才启动，前面的等待就被白白串联起来。这里的数字是问题示例，不是后文的基准测试结果。

作者采用的模型是 Kokoro-82M，参数量为 8200 万，放在 ml.g5.xlarge GPU 实例上的 SageMaker AI 实时端点。自定义容器提供两个入口：POST /invocations 接收一次完整输入并流式返回结果；WS /invocations-bidirectional-stream 在同一连接里持续收发多条消息。

加速来自文本生成与语音合成的重叠，而不是多句话同时合成。

端点处理仍然严格串行：接收一句，合成并发送音频，发出 synthesis_complete，再接收下一句。第一句音频还在传输时，第二句不会在这个处理循环里并发合成。不过，语言模型可以继续生成后面的文字。

让这个重叠成立，需要把 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

要看清 _SENTENCE_RE：它识别的是英文 . ! ? 后跟空白的边界，不能直接当作中文分句器。flush() 也必须在输入结束时调用，否则没有句末标点的尾句会留在缓冲区里。

连接显示正常，为什么请求仍然全失败

作者遇到的一个坑很适合后端联调时参考：SageMaker AI 双向传输把客户端消息送到容器时，使用的是二进制 WebSocket 帧。

Starlette 常见的 receive_text() 会读取 message["text"]。二进制帧没有这个字段，于是每次请求都抛出 KeyError('text')。如果异常处理只把 str(e) 发回去，客户端看到的可能只是 '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

关键是 message.get("bytes", b"").decode("utf-8") 这一行：没有文本字段时，按 UTF-8 解码二进制负载。验证时必须真的发送二进制帧。

连接生命周期同样重要。发送 synthesis_complete 后，容器应回到 receive() 等待下一句，直到客户端明确发送 {"type": "close"} 才结束会话。每句都断开重连，会抵消提前合成带来的收益。作者用同一会话发送三句话，验证了三个音频块分别对应各自文本。

部署也有前提：需要 AWS CDK v2、Python 3.12 或更高版本、uv、Docker，以及支持 invoke_endpoint_with_bidirectional_stream 的 Boto3。Python 版本限制来自底层 HTTP/2 客户端的预编译 wheel。容器还要声明 com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true。

先执行 uv run speech-infra deploy kokoro-82m，再执行 uv run agent-infra deploy --bedrock-model-id <your-bedrock-model-id>。后者部署调用端点的 AgentCore runtime。权限至少包括 sagemaker:CreateEndpoint、sagemaker:CreateEndpointConfig、sagemaker:CreateModel、sagemaker:InvokeEndpointWithBidirectionalStream 和 bedrock:InvokeModelWithResponseStream。

示例容器固定了存在已知安全公告的旧 PyTorch，编号为 PYSEC-2026-2286，修复版本是 torch 2.10.0。原文明确要求在概念验证之外使用前检查并升级。

音频走 SSE，还是直接传二进制

已有聊天页面通常通过 SSE（服务器发送事件）接收文本。接入语音时，把音频转成 base64，再塞进 JSON 事件，是改动较小的方案。

作者测了 24 段音频，对比 MP3 和 AAC，并用 Amazon Transcribe 检查不同编码下的转写质量。base64 的负载膨胀接近预期的 33%。不过，转写结果相近不能替代完整的听感评测。

另一个问题是边界。SSE 用空行分隔事件，MP3 有自己的音频帧边界，网络读取又有传输块边界。作者一次测量中，四个逻辑 MP3 帧被分成六个交付块，其中两个在帧中间结束。客户端必须缓存和重组，不能把每次读取结果直接当成完整事件或可播放单元。

模型侧的替代方案是在双向会话中混传原始音频和 JSON 控制帧。原文的识别规则是：只有整个块能解码成 JSON 对象，才作为控制消息；否则作为音频。不能只看首字节是否为 {，因为原始 PCM 也可能以 0x7b 开头。

但完整案例并没有彻底消除 base64。

系统有三段通信：浏览器与 Agent、Agent 与 Bedrock、Agent 与 Kokoro。模型侧可以用二进制，Agent 返回调用方时却仍然使用 SSE 和 base64，因为 BedrockAgentCoreApp 的 entrypoint 必须产出可 JSON 序列化的字典，SDK 自动包装成 data: {json}\n\n。

因此，"bidi" 和 "batch" 两种模式都承担了浏览器方向的编码成本。run_bidi 用两个并发任务分别处理文本和音频，通过共享队列按到达顺序输出事件；run_batch 等全文生成完再调用合成。这个对照主要比较合成开始时间。

作者先用相同的 267 字符、四句文本，在真实端点上各跑五次：全文一次发送，与只发送首句相比，平均首音频字节时间从 329.3 ms 降到 156.7 ms，相对改善 52.4%。

再接入真实语言模型，原文报告本地容器的首音频速度为 batch 的 2.4 倍，部署到 AgentCore 后为 3.5 倍。复现命令是 uv run agent-infra demo --endpoint-url <your-endpoint-url> --prompt "<your prompt>"，比较 agent_start 到首个 audio_chunk 的时间。

我的判断

我认可它优先调整调用顺序的思路。只要首句和全文之间有明显生成间隔，就有机会隐藏一部分合成等待；回答很短、首句特别长，收益就会缩小。

但这些结果还不能直接变成用户体验承诺。端点测试每种方式只有五次，端到端部分没有给出绝对耗时、试验次数和尾部延迟。首个 audio_chunk 出现，也不等于浏览器已经完成解码并开始播放。部署后差距扩大，作者用网络等待被重叠过程吸收来解释，仍需在自己的链路上验证。

已有 SSE 的团队先用 base64，是合理取舍。33% 是音频负载增幅，不能直接推导成整体延迟或费用增加 33%。新建语音链路可以考虑二进制，但采用 AgentCore 的这个入口时，仍受 JSON 输出约束。

成本方面，原文认为小模型有利于降低推理成本，却没有给出价格、并发吞吐或单位音频费用对照，因此「便宜」在这里主要是架构判断。

能拿来做什么

• 给已有流式 Agent 加语音时，先实现句子缓冲和结束时的 flush()，再按实际语言调整分句规则，并用同一提示词比较 batch 与逐句合成。
• 做前端播放联调时，同时记录首个音频事件和实际开始播放的时间，并测试网络块被拆开的情况，避免把事件到达误当成用户听到。
• 接 SageMaker AI 双向流时，用二进制帧、连续多句和显式关闭消息验证完整会话，不只检查端点是否处于 InService。
• 复现实验后，依次执行 uv run agent-infra destroy --bedrock-model-id <your-bedrock-model-id> 和 uv run speech-infra destroy kokoro-82m；不再使用的 S3 权重缓存和 ECR 镜像仓库需要另行清理，这两个命令不会删除它们。