简介

通过 WebSocket 建立低延迟实时语音会话,兼容 OpenAI Realtime API。连接后按事件协议收发音频与文本,适用于语音助手、实时翻译等场景。 Base URL:wss://api.novawander.cn/v1/realtime
必须使用 WebSocket 协议,不是普通 HTTP POST。model 必须作为 查询参数 传入,否则网关返回 400。

认证

建立 WebSocket 连接时在请求头携带:
Authorization
string
必填
Bearer Token,如 Bearer sk-xxxxxxxxxx
OpenAI-Beta
string
必填
固定为 realtime=v1
浏览器环境无法自定义 WebSocket Header,可使用 OpenAI 子协议传 Key(与 Header 方式等价,由网关解析):
Sec-WebSocket-Protocol: realtime, openai-insecure-api-key.sk-xxxxxxxxxx, openai-beta.realtime-v1

连接

wss://api.novawander.cn/v1/realtime?model=gpt-realtime-1.5
model
string
必填
实时模型 ID,如 gpt-realtime-1.5、gpt-realtime-2。须与当前 API Key 可用模型列表一致。
连接成功时服务端返回 101 Switching Protocols。随后会收到 session.created,客户端应发送 session.update 完成会话配置。

会话配置

连接成功后发送 session.update 设置指令、输出模态与音频格式:
{
  "type": "session.update",
  "session": {
    "type": "realtime",
    "instructions": "You are a concise assistant. Reply in Chinese. 回答用中文回答",
    "output_modalities": ["audio"],
    "audio": {
      "input":  { "format": { "type": "audio/pcm", "rate": 24000 } },
      "output": { "format": { "type": "audio/pcm", "rate": 24000 }, "voice": "sage" }
    }
  }
}
字段说明
session.type固定为 realtime
instructions系统指令,控制助手行为与回复语言
output_modalities["audio"] 输出语音;["text"] 仅输出文字(不播放音频)
audio.input.format输入音频格式,PCM16、24 kHz、单声道
audio.output.format输出音频格式,同上
audio.output.voice音色,见下方可选值
可选音色(output_modalities 含 audio 时有效): alloy · ash · ballad · coral · echo · sage · shimmer · verse · marin 等待 session.updated 后即可开始对话。

发送消息

每轮对话分两步:
  1. conversation.item.create — 写入用户消息
  2. response.create — 触发模型生成回复

文字输入

{
  "type": "conversation.item.create",
  "item": {
    "type": "message",
    "role": "user",
    "content": [
      { "type": "input_text", "text": "你好,请用中文介绍一下自己" }
    ]
  }
}

音频输入

录音须为 PCM16 raw bytes(无 WAV 头),Base64 编码后发送:
{
  "type": "conversation.item.create",
  "item": {
    "type": "message",
    "role": "user",
    "content": [
      { "type": "input_audio", "audio": "<base64_pcm16>" }
    ]
  }
}
同一轮可组合多种输入,例如图片 + 文字、图片 + 录音:
{
  "type": "conversation.item.create",
  "item": {
    "type": "message",
    "role": "user",
    "content": [
      { "type": "ms_image", "image": "data:image/jpeg;base64,<base64>" },
      { "type": "input_text", "text": "这张图里有什么?" }
    ]
  }
}
图片输入限制:部分上游(如 Azure OpenAI Realtime)目前不支持图片,仅支持 input_text 与 input_audio,发送图片可能触发服务端 500 错误。Vision 能力仅在非 Realtime 的 Chat Completions / Responses 接口中可用。详见 Microsoft 官方说明。图片 Base64 长度建议不超过 1,048,576 字符(约 750 KB 原图)。
写入消息后发送触发回复:
{ "type": "response.create" }

接收回复

常见服务端事件:
事件类型说明
response.text.delta / response.output_text.delta文字回复增量(output_modalities: ["text"])
response.audio_transcript.delta / response.output_audio_transcript.delta音频回复的文字转写增量
response.audio.delta / response.output_audio.delta音频 PCM16 增量(Base64),需解码后播放
response.done本轮结束,含 usage 用量
error错误;部分上游错误后连接会被关闭,需重连
音频增量字段可能是 audio 或 delta,客户端应兼容两种写法。

用量说明

每次 response.done 返回的是本次 Response 的用量,不是会话累计。官方原文:“The tokens used for a Response can be read from the response.done event”。多轮对话时建议同时记录每轮用量与会话累计用量。 usage 结构示例:
{
  "total_tokens": 320,
  "input_tokens": 180,
  "output_tokens": 140,
  "input_token_details": {
    "text_tokens": 50,
    "audio_tokens": 130,
    "image_tokens": 0,
    "cached_tokens": 20,
    "cached_tokens_details": {
      "text_tokens": 8,
      "audio_tokens": 12
    }
  },
  "output_token_details": {
    "text_tokens": 40,
    "audio_tokens": 100
  }
}

字段含义

字段说明
total_tokensinput_tokens + output_tokens
input_tokens发给模型的全部输入(含对话历史)
input_token_details.text_tokens输入中的文字 token
input_token_details.audio_tokens输入中的音频 token
input_token_details.image_tokens输入中的图片 token
input_token_details.cached_tokensinput_tokens 的子集(不是额外叠加),命中 prompt cache 的部分;计费时按缓存价打折
cached_tokens_details.text_tokens缓存中属于文字的部分(OpenAI 原生返回)
cached_tokens_details.audio_tokens缓存中属于音频的部分(OpenAI 原生返回)
output_tokens本轮生成的 token
output_token_details.text_tokens输出中的文字 token,即音频回复的 transcript
output_token_details.audio_tokens输出中的音频 token
  1. output.text_tokens:即使选择音频输出,也会因自动 transcript 产生文字 token,属正常计费。详见 OpenAI Realtime 计费说明。
  2. input.cached_tokens:是 input_tokens 的子集,表示命中缓存的部分,不是额外叠加。OpenAI 原生会返回 cached_tokens_details(拆分文字/音频缓存);部分上游(如 Azure GA)不返回时,网关会按文字/音频输入比例估算。
  3. 多轮会话成本:每次 Response 都会把整个对话历史发给模型,越到后面 input_tokens 越大;但前几轮内容大概率命中 prompt cache,折扣很高(如 gpt-realtime-2 音频缓存 0.40/1M,相比音频输入0.40/1M,相比音频输入 32/1M 折扣 98.75%),实际增量成本远小于原始输入价格。官方原文:“turns later in the session will be more expensive”。

Python 完整示例

依赖:pip install websockets pyaudio
import asyncio
import base64
import json
import websockets

BASE_URL = "wss://api.novawander.cn/v1/realtime"
API_KEY  = "sk-xxxxxxxxxx"
MODEL    = "gpt-realtime-1.5"

async def main():
    url = f"{BASE_URL}?model={MODEL}"

    async with websockets.connect(url, additional_headers={
        "Authorization": f"Bearer {API_KEY}",
        "OpenAI-Beta": "realtime=v1",
    }) as ws:
        # 1. 配置会话
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "type": "realtime",
                "instructions": "You are a concise assistant. Reply in Chinese.",
                "output_modalities": ["audio"],
                "audio": {
                    "input":  {"format": {"type": "audio/pcm", "rate": 24000}},
                    "output": {"format": {"type": "audio/pcm", "rate": 24000}, "voice": "sage"},
                },
            },
        }))

        # 2. 等待会话就绪
        async for raw in ws:
            evt = json.loads(raw)
            if evt.get("type") in ("session.created", "session.updated"):
                break

        # 3. 发送文字并触发回复
        await ws.send(json.dumps({
            "type": "conversation.item.create",
            "item": {
                "type": "message",
                "role": "user",
                "content": [{"type": "input_text", "text": "你好"}],
            },
        }))
        await ws.send(json.dumps({"type": "response.create"}))

        # 4. 接收回复
        async for raw in ws:
            evt = json.loads(raw)
            t = evt.get("type", "")
            if t in ("response.text.delta", "response.output_text.delta",
                     "response.audio_transcript.delta", "response.output_audio_transcript.delta"):
                print(evt.get("delta", ""), end="", flush=True)
            elif t in ("response.audio.delta", "response.output_audio.delta"):
                audio_b64 = evt.get("audio", "") or evt.get("delta", "")
                if audio_b64:
                    # 解码 Base64 后播放 PCM16 音频
                    pass
            elif t == "response.done":
                print()
                resp = evt.get("response", evt)  # GA 可能结构不同
                usage = resp.get("usage")
                if usage:
                    print("本轮用量:", usage)
                break
            elif t == "error":
                print("错误:", evt)
                break

asyncio.run(main())
完整交互脚本(含麦克风录音、扬声器播放、图片输入、每轮用量与累计用量统计)可参考 test_realtime.py。

常见问题

  • 连接 400:检查 URL 是否包含 ?model=... 查询参数。
  • 连接 401:API Key 无效或未授权;可先调用 GET /v1/models 验证 Key 与模型是否可用。
  • 发送后无响应:确认已收到 session.updated 且已发送 response.create。
  • 服务端 500 后断连:上游错误会导致 WebSocket 关闭,需重新建立连接。
  • 音频无声:确认采样率为 24 kHz、PCM16 单声道,且正确解码 Base64 增量后播放。
更多事件类型与字段说明请参考 OpenAI Realtime API 文档。