OpenAI Realtime 服务器:协议翻译、会话池与音频回传

30 秒导读: 前面几章把一条语音对话流水线(VAD → STT → LM → TTS)搭好了。本章讲最后一层包装:怎么把这条内部流水线,伪装成一个 OpenAI Realtime API 兼容的 WebSocket 服务——让任何按 OpenAI Realtime 协议写的客户端,连上来就能语音对话。核心是三块:RealtimeService(协议事件 ↔ 内部消息的翻译层)、四个 Handler(响应生命周期的分工)、websocket_router(每条连接一条 send loop + 会话池 + 音频批量回传)。

本章属于 hf-speech-to-speech 系列的核心机制三。前置阅读:

• 01-pipeline-runtime.md —— BaseHandler、线程模型、四条内部队列。
• 03-speculative-turns.md —— speculative turns / barge-in,本章反复引用它做"陈旧话轮过滤"。
• 04-llm-chat-tools.md —— Chat 有界历史、GenerateResponseRequest。

---

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

一句话定义: 一个把本地语音流水线"翻译"成 OpenAI Realtime 协议的 WebSocket 网关。

要解决的问题

OpenAI 的 Realtime API 定义了一套 WebSocket 上的 JSON 事件协议:客户端往上传 input_audio_buffer.append(base64 音频),服务端往下发 response.output_audio.delta(base64 音频),中间还有 session.update、response.create、conversation.item.create、input_audio_buffer.speech_started 等等几十种事件。

这套协议已经有了成熟的客户端生态。于是本项目的想法是:

别自己发明协议——直接说 OpenAI Realtime 的"方言"。 客户端不用改一行代码,把 endpoint 从 OpenAI 换成 ws://localhost:8765/v1/realtime,背后跑的却是你本机的 Whisper + 本地 LLM + 本地 TTS。

用起来什么样

服务端起来后就是一个标准 WebSocket:

客户端 ──▶ {"type":"input_audio_buffer.append","audio":"<base64 PCM16>"}
客户端 ──▶ {"type":"input_audio_buffer.append","audio":"..."}
         (VAD 检测到你说完了,内部流水线自动跑起来)
服务端 ◀── {"type":"input_audio_buffer.speech_started","item_id":"item_..."}
服务端 ◀── {"type":"conversation.item.input_audio_transcription.completed","transcript":"你好"}
服务端 ◀── {"type":"response.created","response":{...}}
服务端 ◀── {"type":"response.output_audio.delta","delta":"<base64 回复音频>"}
服务端 ◀── {"type":"response.output_audio.delta","delta":"..."}
服务端 ◀── {"type":"response.done","response":{"status":"completed",...}}

仓库里 scripts/listen_and_play_realtime.py、scripts/synthetic_conversation_realtime_client.py 就是这样的客户端。

一句话直觉

把它想成一个"同声传译"。 左边坐着一个只会说 OpenAI Realtime 协议的客户端,右边坐着一条只懂内部队列消息(AudioInItem、GenerateResponseRequest、AssistantTextEvent…)的流水线。RealtimeService 就是中间那位翻译:左边说一句,它翻成右边听得懂的;右边产出什么,它翻回左边的协议事件。

---

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

一张结构图

先看这张图怎么读:上半是"入站"(客户端 → 流水线),下半是"出站"(流水线 → 客户端);中间那根竖线 RealtimeService 是双向翻译层,两边都要经过它。

                 ┌──────────────────── 一个 WebSocket 连接 ────────────────────┐
   客户端 ─JSON─▶│  realtime_endpoint (websocket_router.py)                   │
                 │      │  parse_client_event → 分派                          │
                 │      ▼                                                     │
      入 站      │  RealtimeService  ── 翻译 ──▶  内部队列                    │
                 │   (service.py)                 ├ input_queue   (音频→VAD)  │
                 │   audio/session/               └ text_prompt_queue(转写→LM)│
                 │   response/conversation                                    │
                 └───────────────────────────┬────────────────────────────────┘
                                             │  VAD→STT→LM→TTS(见 01/03/04 章)
                 ┌───────────────────────────▼────────────────────────────────┐
      出 站      │  _send_loop_for (一个 unit 一条)                            │
                 │   轮询 text_output_queue(协议事件)+ output_queue(音频)    │
                 │      │  dispatch_pipeline_event → RealtimeService 翻回协议  │
   客户端 ◀─JSON─│      ▼  encode_audio_chunk / finish_response               │
                 └────────────────────────────────────────────────────────────┘

部件一句话职责

部件	干什么	文件
RealtimeServer	拥有 unit 池 + 单个 uvicorn 服务;由 ThreadManager 线程启动	api/openai_realtime/server.py
PipelineUnit	一条隔离的流水线:自己的队列、事件、service、handler 链	api/openai_realtime/pipeline_unit.py
SessionState	每个 WebSocket 的短命状态(ws 引用、session_id、drained 事件)	pipeline_unit.py:12
create_app / realtime_endpoint	FastAPI 应用 + WebSocket 路由:收事件、claim/release unit	api/openai_realtime/websocket_router.py:294
_send_loop_for	每 unit 一条出站循环:轮询队列、批量音频、翻回协议	websocket_router.py:448
RealtimeService	翻译核心:协议事件 ↔ 内部消息,持每连接 ConnState	api/openai_realtime/service.py:179
四个 Handler	Audio/Session/Response/Conversation,按领域拆分响应生命周期	api/openai_realtime/handlers/

主线走一遍(不进代码)

一次"你说一句话、它回一句语音"的完整往返:

1. claim unit:客户端连上 /v1/realtime,路由从池里挑一个空闲 PipelineUnit 占住,register() 出一个 session_id,回 session.created。
2. 入站音频:客户端发 input_audio_buffer.append,AudioHandler 解 base64、重采样到 16kHz、切成 512-sample 的 PCM16 块,塞进 input_queue 给 VAD。
3. 内部流水线:VAD 判断你说完了 → STT 转写 → 落进 Chat 并往 text_prompt_queue 投一个 GenerateResponseRequest → LM 生成 → TTS 合成音频,音频进 output_queue,文本事件进 text_output_queue。
4. 出站翻译:_send_loop_for 把 text_output_queue 里的 TranscriptionCompletedEvent、AssistantTextEvent 翻成协议事件,把 output_queue 里的音频攒够一批 base64 编码成 response.output_audio.delta,遇到结束哨兵就发 response.done。
5. release:客户端断开,路由投 SESSION_END 进 input_queue,等它排空整条链才真正释放 unit。

---

3. 核心原理

本章挑五个机制,由浅入深:入站翻译 → STT→LM 桥 → 响应生命周期 → 出站 send loop → 会话池与释放。

3.1 入站:parse_client_event 与 _pipeline_dispatch 两张路由表

RealtimeService 是一个实例,被所有连接共享(service.py:179-185);每条连接的状态藏在内部的 _conns: dict[str, ConnState] 里,用 session_id 索引。它有两个方向的路由。

入站方向:协议事件 → 内部动作。 客户端发来的原始 JSON 先过 parse_client_event:按 type 字段查 _EVENT_TYPE_TO_MODEL 表,用 pydantic 校验成强类型事件对象,校验失败或类型未知就返回 None(路由层据此回一个 error)。

# service.py:255  parse_client_event(简化)
model_cls = _EVENT_TYPE_TO_MODEL.get(event_type)  # 只认这 5 种客户端事件
if model_cls is None:
    return None                                    # 未知类型 → 上层报错
try:
    return model_cls.model_validate(raw)           # 校验成 pydantic 模型
except ValidationError:
    return None

真源码见 service.py:255-269 parse_client_event;_EVENT_TYPE_TO_MODEL 只列了 5 种入站事件(service.py:66-72):input_audio_buffer.append / session.update / conversation.item.create / response.create / response.cancel。解析后由 realtime_endpoint 用 isinstance 分派到对应 handler(websocket_router.py:342-379)。

出站方向:内部流水线事件 → 协议事件。 内部流水线产出的是 PipelineEvent 子类,由另一张表 _pipeline_dispatch 路由:

内部事件	翻成什么	handler
SpeechStartedEvent	input_audio_buffer.speech_started	audio.on_speech_started
SpeechStoppedEvent	input_audio_buffer.speech_stopped	audio.on_speech_stopped
PartialTranscriptionEvent	...transcription.delta	conversation.on_partial_transcription
TranscriptionCompletedEvent	...transcription.completed + 触发 LM	_on_transcription_completed
TokenUsageEvent	(无协议事件,只计量)	_on_token_usage
ResponseFailedEvent	error + response.done(failed)	_on_response_failed
AssistantTextEvent	文本 delta / 音频转写 / 工具调用	response.on_assistant_text

这张表在 service.py:206-213。注意 AssistantTextEvent 不在表里——它在 _dispatch_pipeline_event 里被单独提前分派(service.py:346-351),因为它要走 speculative-turn 的"提交/丢弃"逻辑(见 §3.3、03 章)。

3.2 ConnState:每连接的全部协议状态

每条连接一个 ConnState(service.py:148-176),它把协议层要维护的全部可变状态装在一个 pydantic 模型里。挑几个关键字段:

字段	作用
session_id / conversation_id	协议 ID,_generate_id 生成
runtime_config	这条连接的 Chat(有界历史)+ session 配置
in_response / response_pending	响应生命周期的两个门闩:是否正在出响应 / 是否已触发但还没出第一块
current_response_id / current_item_id / content_index	当前响应/输出项/内容索引,拼协议事件要用
audio_remainder	上次没凑够 512-sample 的 PCM 尾巴,下次拼上
pending_output_text_parts	纯文本响应逐块攒的文字,finish 时一次性拼成 output_text.done
response_usage	本响应的 UsageMetrics(token / 音频时长 / 工具调用数)
speculative_user_turn_id / speculative_user_item_id / …	speculative 话轮的用户侧游标(见 §3.3)

register()/unregister() 管这些状态的生死(service.py:217-240):register 建一个带全新 Chat 的 ConnState 存进 _conns;unregister 弹出它、关掉它的 Chat(st.runtime_config.chat.close(),防止后台压缩线程去改一个已关会话的历史,见 04 章),并把本连接累计用量 += 进全局 total_usage。

3.3 STT→LM 桥:_on_transcription_completed

这是入站方向最关键的一步:一句话被转写完了,要做三件事——发协议事件、落进 chat 历史、触发 LM 生成。核心在 service.py:389-437。

难点在于 speculative turns(边说边改)。 用户可能"说完 → 又续了半句",于是同一个 turn_id 会有多轮转写;后来的转写要替换而不是追加前一版落进历史的用户话轮。逻辑分三支(service.py:404-417):

# service.py:404  _on_transcription_completed(简化:用户话轮的落库策略)
if transcript:
    if same_speculative_turn and st.speculative_user_item_id:
        # 同一 speculative 话轮的修订:就地替换旧的用户消息文本
        replaced = cfg.chat.replace_user_message_text(st.speculative_user_item_id, transcript)
        if not replaced:                                  # 旧消息已不在历史里
            item = cfg.chat.add_item(make_user_message(transcript))
            st.speculative_user_item_id = item.id
    else:                                                 # 新话轮:直接追加
        item = cfg.chat.add_item(make_user_message(transcript))
        st.speculative_user_item_id = item.id
elif same_speculative_turn and st.speculative_user_item_id:
    cfg.chat.remove_user_message(st.speculative_user_item_id)  # 修订成空 → 撤掉
    st.speculative_user_item_id = None

• 有文字 + 同一 speculative 话轮 → replace_user_message_text 就地替换(chat.py:257)。
• 有文字 + 新话轮 → add_item 追加(chat.py:174)。
• 修订成空文字 → remove_user_message 把先前落进去的撤掉(chat.py:274)。

音频计量的抵扣。 同一 speculative 话轮被重算时,先把上一版计入的音频时长减掉再重记,避免同一段音频被计两次(service.py:392-400)。

触发 LM。 落库后,只要有 text_prompt_queue 且转写非空,就把 response_pending 置真,并投一个 GenerateResponseRequest,带上 turn_id/turn_revision/speech_stopped_at_s(service.py:424-435)——这些字段让 LM 侧和后续的陈旧过滤能对齐话轮身份。

陈旧话轮过滤。 所有出站事件在 _dispatch_pipeline_event 里先过 _is_stale_turn_event(service.py:358-377):如果某个 AssistantTextEvent/TokenUsageEvent 属于一个已经被更新话轮取代的旧话轮,直接丢弃(返回 []),或在"重开宽限期"未决时返回 None 让上层稍后重试。这套判定全委托给 SpeculativeTurnTracker,细节见 03 章。

3.4 响应生命周期:四个 Handler 的分工

响应从"创建"到"结束"的状态机,被拆到四个 handler(service.py:201-204 构造,都继承 RealtimeBaseHandler):

Handler	负责	文件
SessionHandler	session.update 深合并、session.created	handlers/session.py
AudioHandler	入站音频切块、出站音频编码、VAD 的 speech_started/stopped	handlers/audio.py
ResponseHandler	响应的 create / cancel / finish + 所有 ID 管理	handlers/response.py
ConversationHandler	conversation item 注入、转写事件翻译	handlers/conversation.py

RealtimeService 上的同名方法基本是转发门面(service.py:273-303),真逻辑在 handler 里。

_ensure_response / _end_response:两个门闩

响应的核心是一对函数(response.py:38-73):

• _ensure_response:如果还没有 current_response_id,就现造一个(resp_... id + 新 item),置 in_response=True;清 response_pending。幂等——已经有响应就直接返回现有 id。这让"显式 response.create"和"隐式 VAD→…→TTS"两条路都能安全汇聚到同一个响应上。
• _end_response:结算用量(completed/cancelled 计数、total_usage +=、response_usage.reset())、清空所有响应级状态(current_response_id、content_index、in_response、pending_output_text_parts…)。

finish_response:audio-done vs text-done

响应关闭时发什么"done"事件,取决于它是音频响应还是纯文本响应(response.py:206-255):

finish_response(status)
   │
   ├─ response_wants_audio? ──是──▶ 发 response.output_audio.done   (任何终态都发)
   │                          否
   ├─ status=="completed" 且有攒的文字? ──是──▶ 发 response.output_text.done
   │        (把 pending_output_text_parts 拼成完整文本;cancelled/failed 不发)
   │
   └─ 总是最后发 response.done(带 _build_response 的完整状态) → _end_response

关键取舍:纯文本响应的 output_text.delta 是流式逐块发的(response.py:307-317,每块同时 append 进 pending_output_text_parts),但对应的 output_text.done 只在关闭时发一次,携带拼好的全文——且只有 completed 才发,被取消/失败的文本响应直接 response.done 收尾。

handle_response_create:in-band vs out-of-band

客户端可以显式发 response.create 触发一次生成。这里最巧的是 in-band / out-of-band 的 input 分流(response.py:137-195):

• in-band(默认,conversation != "none"):response.input 里的 item 当场 _append_item 落进默认会话历史,这样它们出现在后续上下文里(response.py:161-166)。
• out-of-band(conversation == "none",由 is_out_of_band 判定,utils.py:26):不碰默认会话——input 只随请求"搭车"进 LM,在那边 seed 一个用完即弃的临时 chat;它的 conversation_id 报成 null(response.py:118),assistant 输出也不回写历史。

还有个细节:out-of-band 响应投 GenerateResponseRequest 时 turn_id=None(response.py:181-188)。null 话轮身份让所有 speculative 陈旧门闩都把它当"永远最新",于是生成中途来了新用户话轮也不会误丢它的输出。

失败前置校验:非字符串 tool_choice 直接报 tool_choice_not_supported;已有响应在跑就报 conversation_already_has_active_response(response.py:144-154)。

3.5 出站:_send_loop_for —— 每 unit 一条 send loop

出站方向由 create_app 在 lifespan 里为每个 unit 各起一条 _send_loop_for 协程(websocket_router.py:262-263)。一条 send loop 一轮做两件事(websocket_router.py:448-597):

① 先处理文本事件队列 text_output_queue。 取一个事件翻成协议事件发出。特别地,SpeechStartedEvent(用户开口)如果撞上正在出的响应,且 interrupt_response 允许,就取消当前响应 + 冲刷队列(barge-in,websocket_router.py:487-503,细节见 03 章)。

② 再处理音频队列 output_queue。 这里是音频回传的核心,分几种情况:

从 output_queue 取一个 chunk:
  ├─ PIPELINE_END 哨兵    → 排空待发响应事件 + finish_response + break(收摊)
  ├─ AUDIO_RESPONSE_DONE  → 若 generation 已陈旧:静默丢弃、重开监听
  │                         否则:finish_response、response_playing 清除、重开监听
  ├─ SESSION_END 控制消息 → session.drained.set()(通知释放路径,见 §3.6)
  ├─ 其它控制消息 / 该丢的陈旧音频 → 跳过
  └─ 真音频 → 攒批(见下)→ encode_audio_chunk → response.output_audio.delta

音频批量 MAX_AUDIO_BATCH_BYTES。 逐块发 base64 太碎,于是取到一块真音频后,继续 get_nowait 把后续块拼进一个 bytearray,直到快超过 MAX_AUDIO_BATCH_BYTES(6400 字节,websocket_router.py:38)为止,再一次编码发出(websocket_router.py:561-593)。凑批时若撞上结束哨兵或另一块会超限,就把那块暂存到 session.pending_output_item,下一轮优先处理——保证音频不丢、顺序不乱。

encode_audio_chunk(audio.py:153-199)负责把 PCM 重采样到客户端要的采样率、base64 编码成 response.output_audio.delta;并在第一块音频时,若还没 response.created(隐式响应路径:VAD→…→TTS 没有显式 response.create),补发一个 response.created。

3.6 会话池:claim、drain、release

RealtimeServer 持一个 pool: list[PipelineUnit](大小 num_pipelines,s2s_pipeline.py:642)。一个 unit 同时只服务一条连接。

claim(占用)。 连接进来,_claim_unit 线性扫池子,找第一个 session is None 的 unit,当场塞一个占位 SessionState(websocket_router.py:282-292)。因为 asyncio 单线程、这段中间没有 await,所以"找空位 + 占位"是原子的,不会两个连接抢到同一个 unit。池满就拒连:回一个 session_limit_reached 错误并以 code 1008 关闭(websocket_router.py:298-310)。

release(释放)必须等排空。 断开时不能马上把 unit.session = None——因为流水线里可能还有上一会话的在途消息,若立刻放开,新客户端 claim 到同一 unit,旧输出会被错分派到新会话。所以释放走一个"排空后再放"的协议(websocket_router.py:385-399 + _release_unit_after_drain):

客户端断开(finally 块):
  1. old_session.released_at = now            (/v1/pool 用它显示 draining 时长)
  2. _clean_unit(unit)                        (取消在途 + 冲刷四条队列)
  3. input_queue.put(SESSION_END)             (投哨兵,让它穿过整条 handler 链)
  4. create_task(_release_unit_after_drain)   (另起任务,finally 立即返回)
        │
        └─ 循环 sleep(0.05) 直到 session.drained.set()
              ↑ 由 send loop 在 output_queue 尾端看到 SESSION_END 时置位(§3.5)
           然后:service.unregister(session_id); unit.session = None  ← 真正释放

SESSION_END 从 input_queue 一路穿过 VAD/STT/LM/TTS 每个 handler,最后回到 output_queue——send loop 在那儿看到它就 session.drained.set()(websocket_router.py:547-550)。只有那时释放路径才清 unit.session,新连接才能 claim。

无超时兜底的取舍。 _release_unit_after_drain 故意不设超时释放(websocket_router.py:227-255):如果某个 handler(比如一个还在读的 LM HTTP 调用)迟迟不结束,unit 就一直不释放。作者宁可减少池容量,也不愿冒"跨会话泄漏"的风险(新客户端 claim 到还有旧输出在飞的 unit)。超过 SESSION_END_DRAIN_TIMEOUT_S(10s)只打一条 warning,运维可在 /v1/pool 看到卡住的 unit(draining_for_s 很大)。

为什么另起 task 而不是直接 await? 因为 WebSocketDisconnect 传播后,同一 task 里后续的 await 可能被 Starlette 的 runner 跳过/取消而永不恢复(websocket_router.py:396-399 注释)。所以把 drain 丢进独立 task,finally 立即返回。

冲刷保序:_flush_queue。 _clean_unit 冲刷四条队列时用 _flush_queue(websocket_router.py:94-113):可保留匹配 preserve 的项(如音频结束哨兵、用户文本事件),并把保留项在队列锁下原子地插回队首,确保它们排在 drain 期间 pipeline 线程可能新塞的东西之前——顺序不乱。

3.7 计量:UsageMetrics 与 /v1/usage、/v1/pool

计量是两级累加(service.py:106-146):

• UsageMetrics(每响应):input_tokens/output_tokens/audio_duration_s/responses_completed/tool_calls/turns 等。它自定义了 __iadd__,支持 total += response_usage 把每响应用量滚进总量,reset() 清零。
• GlobalUsageMetrics(服务级):在每响应字段之上加 connections、errors_by_type(按错误类型计数,record_error)。

累加时机:_end_response 每关一个响应就 total_usage += response_usage 再 reset;unregister 断连时再把残余 response_usage 滚进去。

两个只读 HTTP endpoint:

• GET /v1/usage(websocket_router.py:401-420):跨整个池聚合。用递归 _merge 把数值字段相加、dict 字段(如 errors_by_type)按 key 合并——保证每个 unit 的错误计数都不被丢。
• GET /v1/pool(websocket_router.py:422-446):报每个 unit 的状态:idle(空闲)/ active(有活跃会话)/ draining(客户端已断但 SESSION_END 还没排空,附 draining_for_s)——就是 §3.6 那个"无超时兜底"下运维观察卡住 unit 的窗口。

---

4. 深入实现:多管线的深拷贝隔离

池里每个 PipelineUnit 必须是完全隔离的——否则一条连接的取消信号、文本队列会串到另一条。隔离在 _build_realtime_pipeline_unit(s2s_pipeline.py:448-578)里靠深拷贝 + 各建各的运行时对象实现。

① 所有 handler kwargs 深拷贝。 每个 STT/LM/TTS 的参数对象都 deepcopy 一份(s2s_pipeline.py:476-488),注释点明原因:"Per-unit deep copies so mutations (cancel_scope, text_output_queue) don't bleed across pipelines."

② 每 unit 各建一套运行时对象。 队列、事件、CancelScope、SpeculativeTurnTracker 全是这个 unit 独有的(s2s_pipeline.py:490-501):

# s2s_pipeline.py:490  每 unit 独有的运行时对象(节选)
should_listen = Event()
cancel_scope = CancelScope()                 # 本 unit 的取消信号
speculative_turns = SpeculativeTurnTracker() # 本 unit 的话轮跟踪
recv_audio_chunks_queue = Queue()            # = PipelineUnit.input_queue
text_prompt_queue = Queue()                  # 转写 → LM
send_audio_chunks_queue = Queue()            # = output_queue
text_output_queue = Queue()                  # 协议事件侧信道

③ 把共享对象注入各 handler kwargs。 用 vars(kw)[...] = 把这个 unit 的 cancel_scope、speculative_turns、text_output_queue 塞进 VAD / LM / TTS 的参数里(s2s_pipeline.py:503-515),让同一 unit 内所有 handler 共享同一套取消/话轮/输出信道,而不同 unit 之间彼此不可见。

④ 建 service、建 handler 链、返回 PipelineUnit。 RealtimeService 拿到这个 unit 的 text_prompt_queue/should_listen/speculative_turns(s2s_pipeline.py:522-527);_build_pipeline_handlers 造出 [vad, stt, transcription_notifier, lm, lm_processor, tts] 链(s2s_pipeline.py:445),每个 handler 打上 pipeline_index;最后包成 PipelineUnit(s2s_pipeline.py:567-578)。

池子在 build_pipeline 的 mode == "realtime" 分支里造出来(s2s_pipeline.py:639-675):for i in range(pool_size) 建 pool_size 个 unit,交给 RealtimeServer,再把所有 unit 的 handler 摊平交给一个 ThreadManager 统一起线程。

---

5. 巧妙之处(可借鉴的技术)

• 状态与 ws 同生命周期(SessionState)。 把 ws 引用、session_id、send-loop 的临时暂存(pending_output_item)全放在 SessionState 上(pipeline_unit.py:12-35)。unit 释放时整个对象一起消失,不可能有陈旧哨兵漏进下一次 claim。send loop 每轮只快照一次 unit.session,即便中途被释放,也用旧快照跑完这轮,一致。

• 排空再释放,拒绝超时兜底(§3.6)。 用 SESSION_END 哨兵走完整条 handler 链来"证明"流水线已归零,而不是拍脑袋等个固定时间。且明确拒绝超时强制释放,把"跨会话数据泄漏"排在"池容量"之前——一个清醒的可用性 vs 正确性取舍,并用 /v1/pool 把代价可观测化。

• 幂等的响应汇聚(_ensure_response)。 显式 response.create 和隐式 VAD→…→TTS 两条完全不同的触发路径,靠 _ensure_response 的幂等性(response.py:38-46)干净地汇到同一个响应上,不产生重复 response.created。

• out-of-band 的 null 话轮 = 永不被抢占(response.py:181-188)。 用"null turn_id 让陈旧门闩恒判最新"这个小技巧,让带外响应天然免疫 barge-in 的误丢——不用给它加特例分支。

• 音频批量的暂存续传(session.pending_output_item)。 攒批撞到边界(哨兵/超限)时不丢那一块,而是暂存下轮优先发(websocket_router.py:573-585),兼顾吞吐(少发几个大包)与保序、不丢。

• 计量的 dict 深合并(_merge)。 /v1/usage 聚合时对 errors_by_type 这类 dict 递归合并、数值叶子相加(websocket_router.py:406-415),避免"以第一个 unit 的值覆盖其余"的常见聚合 bug。

---

6. 边界与局限

诚实地说清它刻意不做什么、会在哪崩:

• cascaded 的固有延迟。 这是 STT → LM → TTS 的级联架构,不是端到端语音大模型。每一级都要等上一级出结果,首字延迟天然高于原生 speech-to-speech 模型;本层只能优化调度(批量、barge-in),消不掉级联本身的延迟。

• httpx 阻塞读不可即时取消。 cancel_scope.is_stale(gen) 只在流迭代器推进时被检查;一个卡在 httpx 里的阻塞读,cancel_scope.cancel() 无法把它中途叫停(base_openai_compatible_language_model.py:540-542)。缓解手段只有 request_timeout / ReadTimeout。这也是 §3.6 那个 unit 可能长时间 draining 的根因之一。

• MLX 全局锁 → 多管线关掉 live transcription。 在 Apple Silicon 上所有 MLX 推理串行通过一把全局锁(utils/mlx_lock.py)。多管线(num_pipelines > 1)时,渐进式 STT(live transcription)会在这把锁上激烈争用、刷一堆超时 warning。于是启动时若 num_pipelines > 1 且 platform == darwin 且 enable_live_transcription,主动关掉 live transcription(s2s_pipeline.py:1022-1028);最终 STT 路径不受影响。

• 池满即拒连。 unit 数固定为 num_pipelines,满了直接拒(session_limit_reached,websocket_router.py:298-309),没有排队/等待。并发上限 = 池大小。

• 只支持 realtime 会话,不支持 transcription 会话。 session.update 收到 RealtimeTranscriptionSessionCreateRequest 直接报 invalid_session_type(session.py:35-39)。

• tool_choice 只支持字符串形式。 auto/required/none 之外的结构化 tool_choice 报错(response.py:145-149)。工具/TTS 的具体实现本章只点到,不展开——见 04 章。

---

7. 横向对比

同 shelf(ai-agent-reference)里,本项目这层解决的是"把一套内部流水线包装成别人已有协议的服务器"这类问题。它的取舍很典型:

• 协议兼容优先于自造协议——直接复用 OpenAI Realtime 的客户端生态,代价是要把内部消息一一翻译成协议事件。
• 正确性优先于容量——排空再释放、拒绝超时兜底,宁可空着 unit 也不跨会话泄漏。
• 可观测性补短板——用 /v1/pool 把"无超时"带来的卡住风险变成可运维的信号。

对照本仓库其它章:barge-in / speculative turns 的判定细节在 03 章;LM 的流式分句、工具调用、有界历史在 04 章;线程模型与 BaseHandler 在 01 章。本章只讲"最外面那层协议壳"。

---

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

主题	文件	关键符号
协议翻译核心 / 共享 service	src/speech_to_speech/api/openai_realtime/service.py	RealtimeService、ConnState、parse_client_event、_pipeline_dispatch、_dispatch_pipeline_event
STT→LM 桥	service.py	_on_transcription_completed、register、unregister
计量	service.py	UsageMetrics、GlobalUsageMetrics、_on_token_usage、get_usage、build_error_event
响应生命周期	handlers/response.py	ResponseHandler、_ensure_response、_end_response、finish_response、handle_response_create、on_assistant_text
入站音频 / 出站编码	handlers/audio.py	AudioHandler、handle_audio_append、on_speech_started、encode_audio_chunk
会话配置	handlers/session.py	SessionHandler、handle_session_update、build_session_created
conversation item / 转写翻译	handlers/conversation.py	ConversationHandler、handle_conversation_item_create、on_transcription_completed
handler 共享基类	handlers/base.py	RealtimeBaseHandler、_input_item_id
WebSocket 路由 / send loop / 会话池	api/openai_realtime/websocket_router.py	create_app、realtime_endpoint、_claim_unit、_send_loop_for、_release_unit_after_drain、_flush_queue、_clean_unit、MAX_AUDIO_BATCH_BYTES
HTTP 计量 / 池状态 endpoint	websocket_router.py	usage_endpoint(/v1/usage)、pool_endpoint(/v1/pool)
服务器 + unit 数据结构	api/openai_realtime/server.py、pipeline_unit.py	RealtimeServer、PipelineUnit、SessionState
多管线深拷贝隔离	src/speech_to_speech/s2s_pipeline.py	_build_realtime_pipeline_unit(:448-578)、build_pipeline(realtime 分支 :639-675)
MLX 锁 / live transcription 取舍	s2s_pipeline.py、utils/mlx_lock.py	main(:1017-1028 的关闭逻辑)、_mlx_lock
out-of-band / 音频模态判定	utils/utils.py	is_out_of_band、response_wants_audio、_generate_id
httpx 阻塞读不可取消	LLM/base_openai_compatible_language_model.py	生成路径 :540-543 注释、cancel_scope/is_stale