编排核心:一次对话回合怎么从音频走到音频

30 秒导读: 这是 Unmute 后端的主线。用户对着麦克风说话,几百毫秒后音箱里传出机器人的回答——中间这条流水线由谁在驱动?答案不是一个 while True 的大状态机,而是 FastRTC 给的两个回调 receive()(音频进)和 emit()(音频出),外加一个从聊天记录末条消息实时推断出来的会话状态。本章讲清这条从音频到音频的回路怎么转,以及"什么时候该让机器人开口"这个决定是在哪里、怎么做出的。

本章是全书主线。它只讲中央控制回路:双回调、会话状态、响应触发时机、唯一出口、以及对外的事件协议。至于"怎么判断用户说完了"(停顿检测)留给听:STT/VAD/轮次,"怎么逐词喂 TTS、按真实时间放音"留给说:TTS 实时队列,"Quest 这个异步 RAII 怎么管生命周期与打断"留给生命周期与打断。

1. 这是什么(零基础也能懂)

一句话定义: Unmute 后端是一个 WebSocket 服务,一端连着浏览器(收发 Opus 压缩音频),另一端连着三个远程模型服务(STT 转写、LLM 生成文本、TTS 合成语音),它负责把这几样编排成一场自然的语音对话。

它要解决的问题: 文本大模型只会读写字符串,没有耳朵也没有嘴。要让你能"打电话"跟它聊,得有人在中间盯着:实时听音频、判断你说完没有、把你的话喂给 LLM、把 LLM 吐的字逐个送去合成语音、再把语音流回你耳朵——而且你随时插嘴它得立刻闭嘴。这个"盯着"的角色,就是本章讲的编排回路。

一个直觉类比: 把后端想成一个同声传译的调度员。他不自己翻译(那是三个模型的事),他只做三件事:决定"现在轮到谁说"、把话从一个人的嘴传到下一个人的耳、以及在有人插话时喊停。Unmute 的巧妙在于:这个"轮到谁说"的判断,它不用一个专门的变量记着,而是每次现查——看看对话记录最后一句是谁说的。

用起来什么样(协议层面): 客户端连上 wss://.../v1/realtime,不断发 input_audio_buffer.append(base64 的 Opus 音频块),服务端不断回 response.audio.delta(base64 的 Opus 音频)以及一堆状态事件(response.created、转写增量、speech_started 等)。这套事件名字故意抄的 OpenAI Realtime API,所以 OpenAI 的客户端能几乎无改动地连过来。

本节不出现代码细节。记住一件事就够:输入是音频流,输出是音频流,中间隔着三个模型,后端是那个把它们串起来的人。

2. 顶层全景(它大概怎么转)

2.1 两层回路,别混淆

后端其实有两层回路,初学最容易把它们搅在一起。第一层是 WebSocket 的收发,第二层是音频处理逻辑。它们靠一个 asyncio.Queue(叫 output_queue)解耦。

外层(协议层): main_websocket.py 里的 receive_loop 和 emit_loop,只管 WebSocket 上的字节:解 JSON、解 Opus、编 Opus、发 JSON。
内层(逻辑层): UnmuteHandler 的 receive() 和 emit(),只管音频语义:喂给 STT、决定何时生成、从队列取要发的东西。

外层拿到一帧解码后的 PCM 就调内层的 receive();外层想发东西就调内层的 emit() 取一件。两层各跑各的 async 任务,互不阻塞。

2.2 一张图:从音频到音频

怎么读这张图:上半是"进"的方向(左→右),下半是"出"的方向(右→左),正中间的 output_queue 是唯一汇合点。 所有要发给客户端的东西——不管是音频、文本、还是状态事件——都先进这个队列。

                          ┌─────────────────────────────────────────────┐
   浏览器                 │              UnmuteHandler(内层逻辑)          │
   (Opus 音频)           │                                             │
      │  append           │   receive(pcm)                              │
      ▼                   │      │  喂音频 ──────────► STT 转写(远程)   │
 ┌──────────┐  PCM 帧     │      │  推断会话状态                        │      ┌─────────┐
 │receive_  ├────────────►│      │  该说话了? ──► _generate_response    │─────►│ LLM(远程)│
 │loop      │  解 Opus    │             │                             │◄─────┤ 逐词流   │
 │(外层)    │            │             ▼  逐词喂                       │      └─────────┘
 └──────────┘             │        ┌─────────┐  文本delta            │      ┌─────────┐
                          │        │ TTS 循环 │◄──────────────────────│─────►│ TTS(远程)│
 ┌──────────┐  Opus 帧    │        └────┬────┘  音频/文本回来          │◄─────┤ 逐帧音频 │
 │emit_loop ├◄────────────│  emit()     │                             │      └─────────┘
 │(外层)    │  编 Opus    │      ▲      ▼                             │
 └────┬─────┘  取一件      │      └── output_queue(唯一出口)◄─────────┘
      │                   └─────────────────────────────────────────────┘
      ▼
   浏览器(听到回答)

2.3 部件一句话职责

部件	干什么	在哪(文件:符号)
`websocket_route`	接受 `/v1/realtime` 连接,协商 subprotocol,建 handler	`unmute/main_websocket.py:291` `websocket_route`
`_run_route`	用一个 TaskGroup 并发拉起收/发/生命周期四个任务	`unmute/main_websocket.py:380` `_run_route`
`receive_loop`	从 WS 收 JSON、解 Opus 成 PCM、调 `handler.receive()`	`unmute/main_websocket.py:406` `receive_loop`
`emit_loop`	调 `handler.emit()` 取一件、必要时编 Opus、发 JSON	`unmute/main_websocket.py:512` `emit_loop`
`UnmuteHandler.receive`	每来一帧音频:喂 STT、看状态、决定是否生成	`unmute/unmute_handler.py:280` `receive`
`UnmuteHandler.emit`	从 `output_queue` 取一件东西交给外层	`unmute/unmute_handler.py:393` `emit`
`_generate_response`	触发一次机器人回合(拉起 LLM+TTS)	`unmute/unmute_handler.py:177` `_generate_response`
`Chatbot.conversation_state`	从 `chat_history` 末条消息现算会话状态	`unmute/llm/chatbot.py:21` `conversation_state`
`output_queue`	所有对外输出的唯一汇合队列	`unmute/unmute_handler.py:89`(`self.output_queue`)

2.4 主线走一遍(高层,不进代码)

一次完整回合大致是这样一条链:

浏览器发 Opus → receive_loop 解成 PCM → handler.receive()。
receive() 把音频喂给 STT,同时不停问 conversation_state():现在是谁的回合?
STT 转写陆续回来,追加进 chat_history 的 user 消息 → 状态自然变成 user_speaking。
receive() 里检测到停顿 → flush STT → flush 完成后调 _generate_response()。
_generate_response() 先往 chat_history 塞一条空的 assistant 消息(状态立刻变 bot_speaking),再拉起 LLM 任务。
LLM 逐词吐字 → 逐词喂给 TTS → TTS 逐帧回音频 → 全都塞进 output_queue。
emit_loop 不停调 emit() 把队列里的东西编成 Opus / JSON 发回浏览器。
TTS 收尾时往 chat_history 补一条空 user 消息 → 状态回到 waiting_for_user,等下一轮。

注意第 2、3、5、8 步:会话状态从来没被谁"设置"过,它是聊天记录末条消息的副产物。 这是理解整个后端的钥匙,下一节展开。

3. 核心原理

3.1 会话状态:不是变量,是"末条消息谁说的"

它要解决的小问题: 语音对话必须知道"现在轮到谁"——用户在说、机器人在说、还是都没说在等。多数系统会用一个 state 枚举变量,手动在各处 state = BOT_SPEAKING 地改。改漏一处就状态错乱。

Unmute 的思路: 干脆不存这个变量,每次要用时从 chat_history 末条消息现推。规则极简:

末条消息	推出的状态
role 是 `assistant`	`bot_speaking`
role 是 `user` 且内容非空	`user_speaking`
role 是 `user` 但内容为空	`waiting_for_user`
role 是 `system`(刚开场)	`waiting_for_user`

真实实现:

# unmute/llm/chatbot.py:21  conversation_state（节选）
last_message = self.chat_history[-1]
if last_message["role"] == "assistant":
    return "bot_speaking"
elif last_message["role"] == "user":
    if last_message["content"].strip() != "":
        return "user_speaking"
    else:
        return "waiting_for_user"

这段就是整个"状态机"的全部。ConversationState 这个类型别名(chatbot.py:7)只有三个取值。

为什么这么设计妙: 状态和数据成了同一个东西。想让机器人"开始说话",不需要改状态再拉起 LLM——只要往历史里塞一条 assistant 消息,状态自动就是 bot_speaking 了。这正是 _generate_response() 干的第一件事(见 3.3)。想标记"用户回合结束、等下一轮",就塞一条空 user 消息(_tts_loop 收尾时,unmute_handler.py:577)。状态永远不可能和历史对不上,因为它就是从历史算的。

关键细节: 空消息是"占位/信号"的通用手法。空 assistant 消息 = "我开始应答了,别人别抢"(在锁里塞,防竞态);空 user 消息 = "该你了"。add_chat_message_delta 里对空消息的判定(chatbot.py:75 返回 last_message == "")专门用来识别"这是不是一条全新消息的开头"。

3.2 双回调:receive() 进、emit() 出

它要解决的小问题: WebSocket 的收和发是两个独立节奏——用户音频随时来,机器人音频按 TTS 出帧的速度走,两者不能互相卡住。

思路: Unmute 让 UnmuteHandler 继承 FastRTC 的 AsyncStreamHandler(unmute_handler.py:80),只需实现两个方法,FastRTC(这里其实是 main_websocket.py 自己接管的等价循环)会分别在两个 task 里反复调它们:

receive(frame) —— 每来一帧 PCM 调一次,是"进"的入口。
emit() —— 反复调,每次返回一件要发的东西(或 None),是"出"的出口。

receive() 在干嘛(unmute_handler.py:280): 每帧音频进来,它按顺序做:累计样本数当时钟(n_samples_received,receive:289)→ 把音频喂给 STT(stt.send_audio,receive:336)→ 根据当前状态决定要不要触发生成或打断。它是所有"何时说话"决策的发生地。

emit() 在干嘛(unmute_handler.py:393): 极简——就是从 output_queue 取一件:

# unmute/unmute_handler.py:393  emit（节选）
output_queue_item = await wait_for_item(self.output_queue)
if output_queue_item is not None:
    return output_queue_item
else:
    # 没东西发时,顺手更新一下调试快照(GradioUpdate)
    ...

emit() 不含任何业务逻辑,它只是队列的"出料口"。所有"要发什么"的决定,都在别处(STT 循环、LLM 任务、TTS 循环)通过往 output_queue 塞东西来表达。

一个隐藏的巧思——用音频当时钟。 receive() 里的 audio_received_sec()(unmute_handler.py:273)返回的是"至今收到多少秒音频",而不是 time.time()。整个后端的计时(静默超时、"多久没插话"、VAD 免打断窗口)都基于它。好处:逻辑与真实墙钟解耦,离线跑录音、加速回放都不会乱。

3.3 `_generate_response`:一次机器人回合怎么被点燃

它要解决的小问题: 到底在哪几个时刻该让机器人开口?点燃后要保证不重入、不和别的回合打架。

两个触发时机(都在 receive() 里):

首个响应(开场白): 当 chat_history 只有系统提示这一条、且客户端已下发过 instructions 时,主动生成一句欢迎(unmute_handler.py:322-328)。这就是为什么你一连上,机器人会先开口。
用户停顿、flush 完成后: 检测到停顿会给 STT 灌一段静音去 flush 缓冲;等 flush 走完(stt.current_time > self.stt_end_of_flush_time),再调 _generate_response()(unmute_handler.py:363-370)。停顿检测本身的细节见第 02 章。

_generate_response() 本体只有 4 行,但顺序讲究(unmute_handler.py:177):

# unmute/unmute_handler.py:177  _generate_response
async def _generate_response(self):
    # 先塞一条空 assistant 消息:既宣告"我要应答了",又把状态切成 bot_speaking
    await self.add_chat_message_delta("", "assistant")
    quest = Quest.from_run_step("llm", self._generate_response_task)
    await self.quest_manager.add(quest)

为什么先塞空消息、且这句话很关键: 塞完这一刻,conversation_state() 立刻变成 bot_speaking。这既防止同一帧内又触发一次生成(状态已经不是"等用户"了),又让后续的 STT 转写知道"现在机器人在说,来的字算插话"。注释也点明这是"在锁里做,避免竞态"。

真正干活的是 _generate_response_task(unmute_handler.py:184),它编排 LLM→TTS 这条子流水线:

发一个 ResponseCreated 事件进队列(:187),告诉客户端"新回合开始了"。
拉起 TTS(start_up_tts,:199)——TTS 是个独立 Quest,先备好等着收词。
建 VLLMStream 调 LLM,首句用高温 0.7、后续用 0.3(:204-207),让开场白更活泼、后续更稳。
async for delta in rechunk_to_words(...):LLM 的流被重切成一个个词(:225),每个词:发一个文本 delta 事件、喂给 TTS(tts.send(delta),:250)。
每喂一个词前都检查 len(chat_history) > generating_message_i——若历史变长了说明被打断了,立刻 break(:246)。这是打断能生效的关键闸口之一。
词发完,给 TTS 发 EOS(:259)收尾。

一个易错点: LLM 的 token 流不是按词对齐的(可能半个词一个 chunk),所以有 rechunk_to_words(llm/llm_utils.py:65)先把它攒成完整的词再喂 TTS——因为 TTS 是逐词发音的。这条计时/切词的细节属于第 03 章。

3.4 output_queue:唯一出口,也是打断的开关

它要解决的小问题: 音频、文本、状态事件来自好几个并发任务(STT 循环、LLM 任务、TTS 循环),它们怎么有序地发给客户端?打断时又怎么保证"还没发出去的机器人音频"被干净地丢掉?

思路:所有对外输出只走一条队列 output_queue。 谁想发东西都 output_queue.put(...),emit() 是唯一的消费者。这样天然串行化,不会几路输出交错。

队列里流的东西是个联合类型 HandlerOutput(unmute_handler.py:69):可能是 (sample_rate, np.ndarray) 的音频、ora.ServerEvent 的协议事件、AdditionalOutputs 的调试快照、或 CloseStream 的关闭信号。emit_loop 按类型分别处理(见 4.2)。

打断为什么靠"换队列": 当用户插话,interrupt_bot()(unmute_handler.py:583)直接把 self.output_queue 整个换成一个新的空队列(:596),旧队列连同里面排着的机器人音频一起被丢弃。更妙的是 _tts_loop 在开头就把当时的队列存进了局部变量 output_queue(:511),所以即使旧 TTS 任务还没停,它往里塞的东西也只会进旧队列——绝不会污染新回合的新队列。这段打断/生命周期机制详见第 04 章。

4. 深入实现:协议层怎么把回路接起来

4.1 连接与并发骨架

/v1/realtime 的 WebSocket 一被接受,协商 subprotocol 是必须的一步:

# unmute/main_websocket.py:314  websocket_route（节选）
await websocket.accept(subprotocol="realtime")
handler = UnmuteHandler()
async with handler:
    await handler.start_up()
    await _run_route(websocket, handler)

注释解释了为什么(:311-313):OpenAI 的 Realtime 客户端在握手时声明它支持 "realtime" 协议,服务端不回同一个值,客户端就认为"这不是对的端点"而拒绝连接。这一行是 Unmute 能冒充 OpenAI 端点的前提。外面还套了个 SEMAPHORE(:308,MAX_CLIENTS = 4),因为它选择用"多开进程"而非"单进程扛更多"来扩容,以绕开 GIL(:69-71)。

_run_route 用一个 TaskGroup 同时拉起四个协程,任一失败则整组取消(main_websocket.py:392):

# unmute/main_websocket.py:392  _run_route（节选）
async with asyncio.TaskGroup() as tg:
    tg.create_task(receive_loop(websocket, handler, emit_queue), name="receive_loop()")
    tg.create_task(emit_loop(websocket, handler, emit_queue), name="emit_loop()")
    tg.create_task(handler.quest_manager.wait(), name="quest_manager.wait()")
    tg.create_task(debug_running_tasks(), name="debug_running_tasks()")

这里出现了第二个队列 emit_queue(:390):它专给协议层自己要发的事件用(比如 JSON 解析报错、session.updated 回执)。emit_loop 优先发 emit_queue,空了才去调 handler.emit()(见 4.2)。别和 handler 的 output_queue 搞混:output_queue 是音频逻辑的出口,emit_queue 是协议控制的出口。

4.2 Opus 解码入、Opus 编码出

入(receive_loop,main_websocket.py:406): 收到 input_audio_buffer.append,base64 解码得 Opus 字节,交给 sphn.OpusStreamReader 解成 PCM(在线程里做以免阻塞事件循环),非空就调 handler.receive():

# unmute/main_websocket.py:460  receive_loop（节选）
if isinstance(message, ora.InputAudioBufferAppend):
    opus_bytes = base64.b64decode(message.audio)
    ...
    pcm = await asyncio.to_thread(opus_reader.append_bytes, opus_bytes)
    if pcm.size:
        await handler.receive((SAMPLE_RATE, pcm[np.newaxis, :]))

有个防坑逻辑 wait_for_first_opus(:462-469):UI 重连时可能把上一条连接的旧 OGG 包发过来,它靠检查字节里"首包标志位"(opus_bytes[5] & 2)丢弃这些残包,直到看到真正的流首包。

出(emit_loop,main_websocket.py:512): 每轮先看 WS 是否已断(断了就抛 WebSocketClosedError 让 TaskGroup 收摊),再从 emit_queue 取;取不到才去调 handler.emit(),按返回类型分派:

`emit()` 返回的类型	`emit_loop` 怎么处理	位置
`None`	没东西发,continue	`main_websocket.py:535`
`AdditionalOutputs`	包成 `unmute.additional_outputs` 调试事件	`:537`
`CloseStream`	显式 `websocket.close()` 并 break(让收循环也停)	`:542`
`ora.ServerEvent`	直接当事件发	`:546`
`(sr, audio)` 音频	用 `OpusStreamWriter` 编成 Opus,非空则包成 `response.audio.delta`	`:549`

音频编码同样丢线程里做(asyncio.to_thread(opus_writer.append_pcm, ...),:551),并注意 Opus 有缓冲/分块,不是每次喂 PCM 都吐字节,所以有 if opus_bytes: 的判空(:553),没吐就 continue。最后统一 websocket.send_text(to_emit.model_dump_json()) 发出(:566)。

4.3 事件协议:抄 OpenAI Realtime API 的形状

对外协议定义在 openai_realtime_api_events.py,直接对标 OpenAI 的 Realtime API(文件头注释就贴了官方文档链接)。核心设计:

每个事件是个 pydantic BaseEvent,type 字段由 Literal 泛型参数自动填(openai_realtime_api_events.py:36 set_type_from_generic),例如 ResponseTextDelta 的 type 恒为 "response.text.delta"。
事件分两组:ServerEvent(发给客户端,:177)和 ClientEvent(客户端发来,:195)。收消息时用带 discriminator="type" 的 TypeAdapter 按 type 字段自动路由到正确的类(main_websocket.py:79 ClientEventAdapter)。
凡以 Unmute/unmute. 开头的是 Unmute 自己的扩展(如 UnmuteInterruptedByVAD、unmute.response.text.delta.ready),其余尽量沿用 OpenAI 的名字以保证客户端兼容。

为什么这样值得学: 借用一套成熟且被广泛实现的协议,等于免费拿到一整个客户端生态——OpenAI Realtime 的 SDK/示例几乎能直连。代价只是自己那几个扩展事件需要客户端额外认识。

5. 巧妙之处(可带走的精华)

状态即数据,不设独立状态变量。 会话状态从 chat_history 末条消息现算(chatbot.py:21),从根上杜绝"状态和数据不一致"。要切状态就改历史,不会漏改。
空消息当信号量。 空 assistant 消息 = "我开始应答、别抢"(在锁里塞,unmute_handler.py:180);空 user 消息 = "轮到你了"(:577)。一个 role+空串就编码了轮次交接。
唯一出口 + 换队列式打断。 所有输出走一条 output_queue,打断时整条换新队列(:596),旧音频连锅端掉;TTS 任务持有旧队列引用,天然不会串味(:511)。干净利落,不用挨个 flush。
用"收到的音频秒数"当时钟。 audio_received_sec()(:273)让所有计时脱离墙钟,离线/加速回放行为一致。
协议兼容 OpenAI Realtime。 事件形状对齐官方(openai_realtime_api_events.py)+ subprotocol="realtime" 握手(main_websocket.py:314),白捡客户端生态。

6. 边界与局限

单回合、单说话人。 会话状态只有三态、末条消息决定一切,没有多方并行发言的建模。
状态推断依赖历史被正确维护。 一旦哪条路径忘了塞该塞的空消息,状态就会卡住(代码里靠"在锁里塞空消息 + generating_message_i 校验,chatbot.py:52"来防这类竞态)。
"再见"用字符串后缀判定。 机器人道别靠 last_assistant_message.lower().endswith("bye!")(unmute_handler.py:621)触发 CloseStream;注释自己承认用 function calling 更稳,但那样换 LLM 更麻烦——一个刻意的取舍。
本章不覆盖: 停顿/VAD 如何判定该说话(见 02)、TTS 逐词喂与实时放音队列(见 03)、Quest 生命周期与打断机制(见 04)、人格与声音(见 05)。

7. 代码地图(导航索引)

主题	文件	符号
WS 入口 / subprotocol 握手	`unmute/main_websocket.py`	`websocket_route`
并发拉起收/发/生命周期任务	`unmute/main_websocket.py`	`_run_route`
收:解 Opus → `handler.receive()`	`unmute/main_websocket.py`	`receive_loop`
发:取一件 → 编 Opus/发 JSON	`unmute/main_websocket.py`	`emit_loop`
事件类型路由(按 `type` 判别)	`unmute/main_websocket.py`	`ClientEventAdapter`
音频进入口(每帧决策点)	`unmute/unmute_handler.py`	`UnmuteHandler.receive`
输出出口(取 `output_queue`)	`unmute/unmute_handler.py`	`UnmuteHandler.emit`
点燃一次机器人回合	`unmute/unmute_handler.py`	`_generate_response`
LLM→TTS 子流水线编排	`unmute/unmute_handler.py`	`_generate_response_task`
打断:换队列 + 拆 TTS/LLM	`unmute/unmute_handler.py`	`interrupt_bot`
唯一出口队列	`unmute/unmute_handler.py`	`UnmuteHandler.output_queue`
会话状态推断	`unmute/llm/chatbot.py`	`Chatbot.conversation_state`
增量追加消息 / 新消息判定	`unmute/llm/chatbot.py`	`Chatbot.add_chat_message_delta`
事件协议(对标 OpenAI Realtime)	`unmute/openai_realtime_api_events.py`	`ServerEvent` / `ClientEvent` / `BaseEvent`

1. 这是什么(零基础也能懂)​

2. 顶层全景(它大概怎么转)​

2.1 两层回路,别混淆​

2.2 一张图:从音频到音频​

2.3 部件一句话职责​

2.4 主线走一遍(高层,不进代码)​

3. 核心原理​

3.1 会话状态:不是变量,是"末条消息谁说的"​

3.2 双回调:receive() 进、emit() 出​

3.3 _generate_response:一次机器人回合怎么被点燃​

3.4 output_queue:唯一出口,也是打断的开关​

4. 深入实现:协议层怎么把回路接起来​

4.1 连接与并发骨架​

4.2 Opus 解码入、Opus 编码出​

4.3 事件协议:抄 OpenAI Realtime API 的形状​

5. 巧妙之处(可带走的精华)​

6. 边界与局限​

7. 代码地图(导航索引)​