流式 TTS、韵律与打断

更新 2026-09-13

正在读取学习记录…

TTS 是文字转语音。流式TTS会边生成边返回音频，播放器拿到第一块就能开始播，不必等整句话完成。韵律是语句的停顿、重音和语调：同样“订单已发货”，前后停顿不同，听感会变化。文本按逗号和句号组织，能给合成器提供较自然的短语边界。

语音 Agent 正在说“您好，订单已发货。预计明天到达。”，用户说“等等”。这时打断要同时处理生成、队列和播放状态。像关闭一台正在排队打印的打印机：停止产生新任务之外，已经排队的任务也要取消，迟到的旧任务不能重新进队。

从输入到可核对结果

下面使用带请求ID、序号和文本标签的人工音频块记录，实际运行播放器队列与取消机制。CHUNKS 的第三个字段标注对应语句，播放器没有合成人声。

deque 是两端队列，append从尾部加入，popleft从头部取出。tick表示消费一块，cancel清空尚未消费的队列并把当前请求加入cancelled。enqueue再次收到这个请求的尾包时返回False。同一个请求内，序号必须接着当前请求已消费与待消费总数，避免颠倒语句。begin("r2") 启动新请求，清空旧队列、取消旧请求，并把当前请求的played重新置空；新的块0因此能接收。history独立保留所有请求实际播放过的块，旧历史不会干扰新请求编号。真实服务可能先返回压缩音频或PCM字节，队列应按实际音频时长而非字符数控制。TTS官方文档说明流式音频与声音输出选项。

打断时的材料	本例动作	用户体验
已消费1块	保留播放记录	已说“您好，”
队列剩余2块	清空队列	后两句不继续
旧请求尾包	按r1取消标记丢弃	不会突然接上旧话
下一轮回答	begin(r2)重置本轮计数	新块0能入队

r1取消清队列，history保留已播，begin r2从块0开始，拒收r1尾包 — 历史记录和当前请求计数分开：新回答从块0开始，旧ID尾包继续拒绝。

保存并运行这两个文件

在 PowerShell 创建练习目录：

New-Item -ItemType Directory -Force l9-tts
cd l9-tts

macOS 终端：

mkdir -p l9-tts
cd l9-tts

把下面第一段完整代码保存为 playback_lab.py，第二段保存为 test_playback_lab.py，两份文件放在这个目录。

from collections import deque

CANCEL_AFTER = 1
CHUNKS = [("r1", 0, "您好，"), ("r1", 1, "订单已发货。"), ("r1", 2, "预计明天到达。")]

class Player:
    def __init__(self):
        self.current = "r1"
        self.queue = deque()
        self.played = []
        self.history = []
        self.cancelled = set()

    def begin(self, request_id):
        # 一个请求的编号从0开始；旧请求已播放内容保留在独立历史中。
        if not request_id or request_id == self.current or request_id in self.cancelled:
            raise ValueError("新请求ID必须唯一且非空")
        self.cancel()
        self.current = request_id
        self.played = []

    def enqueue(self, chunk):
        request, index, text = chunk
        # 取消后仍可能收到网络尾包；用请求ID过滤这些迟到块。
        if request in self.cancelled or request != self.current:
            return False
        if index != len(self.played) + len(self.queue):
            raise ValueError("音频块序号不连续")
        self.queue.append(chunk)
        return True

    def tick(self):
        if self.queue:
            chunk = self.queue.popleft()
            self.played.append(chunk)
            self.history.append(chunk)
        return [chunk[2] for chunk in self.played]

    def cancel(self):
        self.cancelled.add(self.current)
        removed = len(self.queue)
        self.queue.clear()
        return removed

if __name__ == "__main__":
    player = Player()
    for chunk in CHUNKS:
        player.enqueue(chunk)
    for _ in range(CANCEL_AFTER):
        player.tick()
    print("已播放文本：" + "".join(chunk[2] for chunk in player.played))
    print("打断清空块数：" + str(player.cancel()))
    print("迟到块入队：" + str(player.enqueue(("r1", 3, "谢谢。"))))
    player.begin("r2")
    player.enqueue(("r2", 0, "请告诉我您要修改什么。"))
    print("新请求播放：" + "".join(player.tick()))
    print("历史已播块数：" + str(len(player.history)))

import unittest
import playback_lab as lab


class MechanismTests(unittest.TestCase):
    # 正常、边界和失败输入都由这里提供，直接运行即可。

    def test_normal_cancel(self):
        player = lab.Player()
        for chunk in lab.CHUNKS:
            player.enqueue(chunk)
        self.assertEqual(player.tick(), ["您好，"])
        self.assertEqual(player.cancel(), 2)
        self.assertFalse(player.enqueue(("r1", 3, "尾包")))
        player.begin("r2")
        self.assertTrue(player.enqueue(("r2", 0, "新的回答")))
        self.assertEqual(player.tick(), ["新的回答"])
        self.assertEqual([chunk[0] for chunk in player.history], ["r1", "r2"])
        self.assertFalse(player.enqueue(("r1", 3, "旧请求尾包")))

    def test_boundary_empty(self):
        player = lab.Player()
        self.assertEqual(player.tick(), [])
        self.assertEqual(player.cancel(), 0)

    def test_failure_order(self):
        with self.assertRaises(ValueError):
            lab.Player().enqueue(("r1", 1, "提前的块"))
        player = lab.Player()
        for request in ("", "r1"):
            with self.assertRaises(ValueError):
                player.begin(request)


if __name__ == "__main__":
    unittest.main()

打开 PowerShell，逐行输入下列命令。cd 是进入文件夹，dir 是列出文件；确认列表里有刚保存的两个 .py 文件，再运行 Python。

# 在刚才创建的练习目录检查文件，再运行程序。
ls
python -X utf8 playback_lab.py
python -X utf8 -m unittest -v test_playback_lab.py

程序输出：

已播放文本：您好，
打断清空块数：2
迟到块入队：False
新请求播放：请告诉我您要修改什么。
历史已播块数：2

测试末尾应显示 Ran 3 tests 与 OK。若找不到文件，检查当前目录与扩展名；若导入失败，确认测试和主程序在同一文件夹。测试描述的三组材料就在上方代码，不需要另外准备输入。

改一处，再看结果

输出中的False说明打断后的尾包被拒绝。最后两行显示新请求r2从块0成功播放，历史含r1与r2的已播块共2块。一次取消如果只清空队列却没有请求标记，网络刚到的尾包还会进入队列，造成停止后又说话。

把 CANCEL_AFTER = 1 改成 CANCEL_AFTER = 2，保存并重跑。已播放文本增加“订单已发货。”，清空块数从2降为1。恢复为1后重跑。这改变的是打断发生的位置，没有重新生成语句。

真实播放设备通常还有自己的小缓冲，清空Python队列后需通知音频设备停止；远端生成也要单独取消。会话历史应只保留用户实际听到的部分，否则下一轮会以为用户已经听见被打断的内容。向用户播放模型合成声音时，应说明声音来自AI；需要调用远端服务的声音合成会产生服务费用，费用控制跟合成请求一起设置。

改动后仍在刚才的 PowerShell 目录运行：

python -X utf8 playback_lab.py

确认上述变化后，把参数恢复成原值，保存并再次执行这条命令。

把机制接回 Agent

把Agent文字答案交给TTS后，每个音频块需要带请求ID、连续序号、格式和时长。播放器边接边播；用户发言事件到来时，取消远端生成、停止设备播放、清空队列并更新会话已听到的位置。这四步应共同处理，才能避免下一轮接上旧回答。

课后练习展开习题收起习题

正在读取本地记录…

不调用模型，不执行代码。简答、代码和场景题仅保存答案并展示自查标准，不生成评分、徽章或认证。输入内容不会自动保存，请点击提交；不要填写个人信息或密钥。

练习 1 / 4 · v1.0.1

选择题 · 先判断边界

关于“流式 TTS、韵律与打断”，哪种做法更稳妥？

对应知识点：L9-tts

选择一个答案A. 先明确输入、证据和停止条件B. 默认输出永远正确C. 用一次成功推断全部场景

练习 2 / 4 · v1.0.0

简答题 · 解释失败

写出一个失败输入并说明记录字段。

对应知识点：L9-tts

你的回答

练习 3 / 4 · v1.0.0

代码题 · 边界测试

补充正常、边界、失败三组输入和预期结果。

对应知识点：L9-tts

你的代码或伪代码

练习 4 / 4 · v1.0.0

场景题 · 结果未知

证据不足时如何设计状态和人工接管？

对应知识点：L9-tts

你的回答