跳到主要内容

数据截至 (上游 commit b21e54d6a845)

HuggingFace speech-to-speech — 架构与原理

30 秒导读: 这是一个开源的语音对话后端。你对着麦克风说话,它检测你说完了、转成文字、送给大模型、把回答合成成语音流回来——全程用 OpenAI Realtime 协议,所以任何写好的 Realtime 客户端把 URL 一改就能连过来。四个环节的模型全部可换,LLM 可以是 OpenAI,也可以是你本机的 llama.cpp。

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

一句话定义​

一个把「听 → 懂 → 说」串起来的低延迟语音代理服务器,对外长得跟 OpenAI 的 Realtime API 一模一样。

解决谁的什么问题​

假设你要做一个会说话的机器人玩具,或者一个电话客服 bot。你需要的东西是:

  • 判断用户什么时候说完了(不能他一停顿就抢话)
  • 把语音转成文字
  • 让大模型生成回答
  • 把回答变成声音,边生成边播(不能等整段生成完)
  • 用户中途插话时,立刻闭嘴

这五件事每一件都有现成的模型/库,但把它们拼起来并且低延迟、不串台是脏活。这个项目就是那份脏活的开源实现。README 称它已经作为 Reachy Mini 机器人的对话后端在生产环境运行(此为 README 陈述,代码内无从验证)。

它能做什么​

能力说明
Realtime 兼容服务WebSocket /v1/realtime + WebRTC POST /v1/realtime/calls
四段全部可换VAD / STT / LLM / TTS 各有多个后端,CLI 一个参数切换
打断(barge-in)用户开口即取消正在播的回答
实时字幕说话过程中就流式吐出转写增量
工具调用云端 API 原生 tools,或本地模型用 <code>...</code> 提示词模拟
纯本地栈Parakeet(STT)+ llama.cpp / mlx-lm(LLM)+ Qwen3-TTS(TTS)
LLM 反向代理把配置好的远端 LLM 再暴露成普通 OpenAI 端点给客户端做副业任务

用起来什么样​

最小的一次跑通(依据:README.md Quickstart 段、pyproject.toml:105-106 的 [project.scripts]):

pip install speech-to-speech
export OPENAI_API_KEY=...

# 终端 1:起服务器(默认 Parakeet STT + OpenAI Responses API + Qwen3-TTS)
speech-to-speech serve

# 终端 2:用自带的麦克风/扬声器客户端连上去
speech-to-speech talk --url ws://127.0.0.1:8765/v1/realtime

程序化连接就是标准 OpenAI SDK,只是把 base_url 指向本机:

# 示意,非源码(改编自 README 的 Realtime API 段)
from openai import OpenAI

client = OpenAI(
    base_url="http://localhost:8765/v1",
    websocket_base_url="ws://localhost:8765/v1",
    api_key="not-needed",          # 服务端本身不校验
)

with client.realtime.connect(model="local") as conn:
    conn.send({"type": "session.update", "session": {...}})
    for event in conn:             # speech_started / transcription / audio delta ...
        print(event.type)

一句话直觉​

把它想成一条工厂流水线: 每个工位(VAD、STT、LLM、TTS)是一个独立线程,工位之间用传送带(队列)连接,每件半成品上贴着一张标签(turn_id + turn_revision + cancel_generation)。用户改口时,不需要去停机器——只要把标签作废,下游工位看到过期标签就直接把半成品扔进垃圾桶。

这条「标签作废」的暗线,是理解整个项目的钥匙。

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

主线数据流​

怎么读这张图:从上到下是一次问答的时间顺序;注意 STT 不直接连 LLM,中间要绕一次协议层。

            客户端(浏览器 / 打包客户端 / 任意 Realtime SDK)
                    │ 上行 PCM                      ▲ 下行 JSON 事件 + 音频
                    ▼                               │
              ┌──────────┐                   ┌──────────────┐
              │   VAD    │                   │  发送循环    │
              │切出语音段│                   │ _send_loop   │
              └────┬─────┘                   └──────▲───────┘
                   │ 语音段                         │ PCM + 有序事件
                   ▼                                │
              ┌──────────┐                    ┌───────────┐
              │   STT    │                    │    TTS    │
              └────┬─────┘                    └─────▲─────┘
                   │ 转写完成事件                   │ 要念的句子
                   ▼                                │
         ┌──────────────────────┐           ┌───────────────┐
         │  RealtimeService     │  生成请求 │  输出处理器   │
         │ 协议状态机+会话历史  │ ────────► │ LMOutputProc  │
         └──────────────────────┘           └───────▲───────┘
                                                    │ 文本片段 / 工具调用
                                             ┌──────┴──────┐
                                             │     LLM     │
                                             └─────────────┘

为什么 STT 不直连 LLM? 因为「转写完了」和「该生成回答了」是两件事:协议层要先把转写落进会话历史、发出 input_audio_transcription.completed、判断这个回合是否已经作废,才决定要不要下单。这一跳发生在 RealtimeService._on_transcription_completed(src/speech_to_speech/api/openai_realtime/service.py:615-668),它构造 GenerateResponseRequest 丢进 text_prompt_queue。TranscriptionNotifier.process 本身从不产出任何东西给下游队列(结尾是 yield from (),src/speech_to_speech/STT/transcription_notifier.py:105)。

部件一句话职责​

部件干什么在哪个文件
VADHandler用 Silero 判定说话起止,切出语音段,发 speech_started/stoppedsrc/speech_to_speech/VAD/vad_handler.py
SmartTurnAnalyzer语音段结束后再判一次「这句话说完了吗」,决定等多久src/speech_to_speech/VAD/smart_turn.py
STT handler语音段 → 文本(渐进 + 最终两种)src/speech_to_speech/STT/parakeet_tdt_handler.py 等
TranscriptionNotifier把转写结果翻译成协议中立事件src/speech_to_speech/STT/transcription_notifier.py
LLM handler调 provider,把流式响应归一化成文本/工具事件src/speech_to_speech/LLM/base_openai_compatible_language_model.py
LMOutputProcessor保序:把文本、工具、TTS 输入放同一条队列src/speech_to_speech/LLM/lm_output_processor.py
TTS handler文本 → PCM 流src/speech_to_speech/TTS/qwen3_tts_handler.py 等
RealtimeService协议状态机:事件解析、会话状态、历史、用量src/speech_to_speech/api/openai_realtime/service.py
_send_loop_for出口总闸:决定哪些输出真正发给客户端src/speech_to_speech/api/openai_realtime/websocket_router.py:806-1091
PipelineUnit一条完整独立流水线(队列 + 状态 + handler 实例)src/speech_to_speech/api/openai_realtime/pipeline_unit.py

主线走一遍(高层)​

  1. 客户端发 input_audio_buffer.append(base64 PCM)。AudioHandler.append_pcm 重采样到 16 kHz,切成 512 采样一块喂给 VAD(src/speech_to_speech/api/openai_realtime/handlers/audio.py:122-149)。
  2. VAD 累积到「有效语音」时发 speech_started;检测到静音收尾时把整段音频作为 VADAudio(mode="final") 推给 STT。
  3. STT 出 Transcription;TranscriptionNotifier 转成 TranscriptionCompletedEvent 放到旁路队列 text_output_queue。
  4. 发送循环把该事件交给 RealtimeService,后者写历史 + 下 GenerateResponseRequest。
  5. LLM 边流边按句子成批吐 LLMResponseChunk;LMOutputProcessor 把每一段拆成「事件 + TTS 输入」放同一队列(保序的关键)。
  6. TTS 合成 PCM,连同事件一起走 send_audio_chunks_queue;发送循环编码成 response.output_audio.delta 发出去。
  7. 用户中途开口 → 发送循环调 cancel_scope.cancel(),世代号 +1,所有在途产物瞬间过期。

3. 阅读地图​

建议顺序(由浅入深):

章节讲什么什么时候读
01 骨架handler/队列/线程模型、后端注册表、进程启动必读,后面所有章的基础
02 断句与投机回合VAD、Smart Turn、turn_id/revision、打断必读,这是本项目最核心的原创机制
03 语音转文字渐进转写、过期过滤、MLX 全局锁想理解「实时字幕怎么做」时读
04 语言模型与历史provider 事件归一化、按句批量、Chat 事务想接自己的 LLM 时读
05 工具与预取两种工具调用路径、投机预取事务想做 function calling 时读
06 协议层状态机、发送循环闸门、WebRTC、LLM 代理想改协议行为或做客户端时读
07 语音合成文本合并、流式切块、延迟指标想优化「开口延迟」时读
08 精华与边界可借鉴技术、局限、对比、总代码地图收尾必读

4. 代码地图(总入口)​

主题文件路径符号名
CLI 命令分发src/speech_to_speech/cli.pyparse_command, main
参数解析与流水线组装src/speech_to_speech/s2s_pipeline.pyparse_arguments, _build_handlers, build_pipeline
handler 基类(线程主循环)src/speech_to_speech/baseHandler.pyBaseHandler.run
后端注册表src/speech_to_speech/backend_registry.pySTT_BACKENDS, LLM_BACKENDS, TTS_BACKENDS
投机回合跟踪器src/speech_to_speech/pipeline/speculative_turns.pySpeculativeTurnTracker
取消世代src/speech_to_speech/pipeline/cancel_scope.pyCancelScope
队列消息类型src/speech_to_speech/pipeline/messages.pyVADAudio, LLMResponseChunk, TTSInput, EndOfResponse
协议中立事件src/speech_to_speech/pipeline/events.pySpeechStartedEvent, AssistantOutputEvent
协议状态机src/speech_to_speech/api/openai_realtime/service.pyRealtimeService, ConnState
出口发送循环src/speech_to_speech/api/openai_realtime/websocket_router.py_send_loop_for, create_app
官方设计说明(英文,项目自带)src/speech_to_speech/api/openai_realtime/README.md—