实时会话与语音

实时（Realtime）语音会话走 WebSocket，是「连接保活、多轮事件流」的会话协议，与一问一答的 HTTP 完全不同。网关把 OpenAI 与 DashScope 两个实时上游统一成一个客户端入口：GET /v1/realtime（OpenAI Realtime 客户端协议）。客户端侧永远说 OpenAI Realtime 协议，上游是 openai_realtime 还是 dashscope_realtime 由模型配置决定，网关在中间桥接。

前置

• 网关已运行，控制台里已配一个 openai_realtime（或 dashscope_realtime）协议的模型。没有配实时模型时，请求 /v1/realtime 会 404 model_not_found——先确认模型存在再谈接线。
• 客户端用任意 WebSocket 库即可，不需要厂商 SDK。

连接与认证

与 HTTP 数据面不同，WebSocket 握手通过 Authorization 头认证。浏览器原生 WebSocket 无法自定义头，而网关认证中间件只读请求头（不支持 ?api_key= 查询参数认证）——因此在浏览器里要用支持自定义头的客户端/库：

• 头：Authorization: Bearer <访问密钥> + OpenAI-Beta: realtime=v1
• URL：ws://localhost:7890/v1/realtime?model=<网关里配的实时模型名>

Python

import asyncio, json, websockets

async def main():
    async with websockets.connect(
        "ws://localhost:7890/v1/realtime?model=gpt-realtime",
        additional_headers={
            "Authorization": "Bearer <你的访问密钥>",
            "OpenAI-Beta": "realtime=v1",
        },
    ) as ws:
        await ws.send(json.dumps({
            "type": "response.create",
            "response": {"modalities": ["text"], "instructions": "用中文说一句问候"},
        }))
        async for raw in ws:
            ev = json.loads(raw)
            if ev["type"] == "response.output_text.delta":
                print(ev["delta"], end="")
            elif ev["type"] == "response.done":
                break

asyncio.run(main())

TypeScript

import WebSocket from "ws";

const ws = new WebSocket(
  "ws://localhost:7890/v1/realtime?model=gpt-realtime",
  { headers: { Authorization: "Bearer <你的访问密钥>", "OpenAI-Beta": "realtime=v1" } },
);

ws.on("open", () => {
  ws.send(JSON.stringify({
    type: "response.create",
    response: { modalities: ["text"], instructions: "用中文说一句问候" },
  }));
});

ws.on("message", (raw) => {
  const ev = JSON.parse(raw.toString());
  if (ev.type === "response.output_text.delta") process.stdout.write(ev.delta);
  else if (ev.type === "response.done") ws.close();
});

事件序列

一次文本轮次的最小事件流：

1. session.update（可选）——设会话级指令、模态、输入/输出音频格式。
2. response.create——发起一轮回复（可带 input 为 conversation.item.create 的上下文）。
3. response.output_text.delta——文本增量（重复到达，客户端拼起来）。
4. response.audio.delta——音频增量（base64 PCM，若开了 audio 模态）。
5. response.done——本轮结束。

本文示例只演示文本模态（最小闭环）。音频模态在同一条连接上开 modalities: ["audio"]，并推送 base64 PCM 帧（input_audio_buffer.append），见下方采样率注意。

音频采样率差异

两个实时上游的采样率不对称，网关桥接按帧透传、不做重采样——用错率会 1.5× 慢放、音调偏低：

上游协议	输入采样率	输出采样率
openai_realtime（OpenAI）	24 kHz	24 kHz
dashscope_realtime（DashScope）	16 kHz	24 kHz

播放时用输出率解输出流、别用输入率。厂商侧约定，详见 Realtime 实时会话。

验证状态：连接与认证接线已在真实网关验证（带正确密钥请求未配置的实时模型 → 404 model_not_found，说明路由与认证已通；不带密钥 → 401）。事件流部分因示例网关未配置实时模型，未能端到端实跑，代码按 OpenAI Realtime 协议编写。你的网关上配好 gpt-realtime 或 DashScope qwen*-omni-realtime 模型后即可直接运行。

openai_audio：语音转写与合成

非会话式的语音能力走 HTTP（openai_audio 协议），经 OpenAI SDK 的 client.audio：

from openai import OpenAI

client = OpenAI(base_url="http://localhost:7890/v1", api_key="<你的访问密钥>")

# 语音转写（ASR）
with open("speech.mp3", "rb") as f:
    t = client.audio.transcriptions.create(model="whisper-1", file=f)
print(t.text)

# 文本转语音（TTS）
speech = client.audio.speech.create(model="tts-1", voice="alloy", input="你好，世界")
speech.write_to_file("out.mp3")

openai_audio 与 dashscope 多模态互译转写/翻译；但文字转语音（TTS）在 DashScope 上游无对应能力，会返回 unsupported_feature。能力矩阵见 协议互通矩阵。

下一步

• 实时会话的配置、计费（按轮）、会话作用域细节：见 Realtime 实时会话。
• DashScope 原生 SDK 与异步视频任务：见 DashScope 原生与异步任务。