实时语音

通过双向实时会话交换音频与文本。

GET/v1/realtime

WebSocket

Headers请求头

左右滑动查看完整参数表

请求头
参数名	类型	必填	说明	默认值 / 示例
`Authorization`	string	必填	平台 API Key，使用 Bearer 认证。	示例 `Bearer YOUR_API_KEY`

Query Parameters查询参数

左右滑动查看完整参数表

查询参数
参数名	类型	必填	说明	默认值 / 示例
`model`	string	必填	具有实时语音能力的模型 ID；需平台开放。	示例 `MODEL_ID`

使用说明

此路径使用 WebSocket，连接成功时升级为 101 Switching Protocols，不能按普通 JSON REST 请求调用。
WebSocket 接口使用会话事件交换音频与文本，认证方式、音频编码和事件字段按模型协议配置。
将 MODEL_ID 替换为平台可用的语音模型 ID；需要音色的接口请填写该模型支持的音色。

WebSocket 连接与协议版本

连接地址为 wss://ai.kedang.net/v1/realtime?model=MODEL_ID。握手使用 Authorization: Bearer YOUR_API_KEY，成功后通过 WebSocket JSON 事件交换数据。本节是 OpenAI Realtime GA 协议参考；模型、通道与音色需在账户中开放并采用同一协议版本。不同版本的会话字段和音频事件名可能不同，不能据此推断所有语音模型均可连接。

在服务端保存 API Key。浏览器直接接入应使用服务端签发的短期凭据，并确认账户开放对应会话接口。示例使用服务端 Python WebSocket 连接，MODEL_ID 与 VOICE_ID 均需替换。

步骤	事件	处理内容
初始化	session.created → session.update → session.updated	配置 GA 会话的输入/输出音频格式与音色
上传音频	input_audio_buffer.append	audio 字段为裸音频字节的 Base64
结束本轮输入	input_audio_buffer.commit → input_audio_buffer.committed	关闭自动轮次检测时由客户端手动提交
请求回复	response.create	发起音频输出
接收音频	response.output_audio.delta	解码 delta 并按顺序追加字节
接收文本	response.output_audio_transcript.delta	可选的语音回复转写增量
终止与错误	response.done / error	检查 response.status；失败或连接中断不作为成功

音频编码与完整交互示例

本节固定使用 PCM16：24000 Hz、单声道、16 位有符号小端裸数据；Base64 内不包含 WAV 文件头。audio/pcm 的 rate=24000 与输入、输出文件的真实编码保持一致。示例关闭自动 VAD，上传一段完整音频，再提交并请求回复。

GA 的格式配置位于 session.audio.input.format 与 session.audio.output.format，音色位于 session.audio.output.voice。120 秒超时和 4800 字节分块是客户端示例选择；允许的模型、音色及其他格式以账户对应协议为准。

展开 Python 3.11+：连接、上传、接收并保存音频

# Python 3.11+；安装依赖：python -m pip install "websockets>=14"
# 本示例使用 Realtime GA 协议字段与事件名。
import asyncio
import base64
import json
import os
import tempfile
from pathlib import Path
from urllib.parse import urlencode
from websockets.asyncio.client import connect

MODEL_ID = "MODEL_ID"  # 账户开放且采用本节 GA 协议的实时语音模型
VOICE_ID = "VOICE_ID"  # 填写该模型可用的音色


async def read_event(socket):
    event = json.loads(await socket.recv())
    if event.get("type") == "error":
        raise RuntimeError(event.get("error", event))
    return event


async def wait_event(socket, expected):
    while True:
        event = await read_event(socket)
        if event.get("type") == expected:
            return event


async def exchange_audio(socket, pcm, output):
    await wait_event(socket, "session.created")
    await socket.send(json.dumps({
        "type": "session.update",
        "session": {
            "type": "realtime",
            "output_modalities": ["audio"],
            "audio": {
                "input": {"format": {"type": "audio/pcm", "rate": 24000}, "turn_detection": None},
                "output": {"format": {"type": "audio/pcm", "rate": 24000}, "voice": VOICE_ID},
            },
        },
    }))
    await wait_event(socket, "session.updated")
    # 裸 PCM16：小端、有符号、单声道、24000 Hz。4800 字节约为 100 ms。
    for offset in range(0, len(pcm), 4800):
        audio = base64.b64encode(pcm[offset:offset + 4800]).decode("ascii")
        await socket.send(json.dumps({"type": "input_audio_buffer.append", "audio": audio}))
    # 此例关闭自动轮次检测，先提交输入缓冲，再主动请求一个回复。
    await socket.send(json.dumps({"type": "input_audio_buffer.commit"}))
    await wait_event(socket, "input_audio_buffer.committed")
    await socket.send(json.dumps({"type": "response.create", "response": {"output_modalities": ["audio"]}}))
    received_audio = False
    while True:
        event = await read_event(socket)
        kind = event.get("type")
        if kind == "response.output_audio.delta":
            chunk = base64.b64decode(event["delta"], validate=True)
            if chunk:
                output.write(chunk)
                received_audio = True
        elif kind == "response.output_audio_transcript.delta":
            print(event.get("delta", ""), end="", flush=True)
        elif kind == "response.done":
            if event.get("response", {}).get("status") != "completed":
                raise RuntimeError(event.get("response", event))
            if not received_audio:
                raise RuntimeError("响应完成但没有音频数据，请检查输出模态与模型配置。")
            return
        # response.output_audio.done 仅表示一个音频部分结束；仍等 response.done。


async def main():
    pcm = Path("input.pcm").read_bytes()
    if not pcm or len(pcm) % 2 or pcm.startswith(b"RIFF"):
        raise ValueError("input.pcm 必须为非空裸 PCM16，不包含 WAV 文件头")
    url = "wss://ai.kedang.net/v1/realtime?" + urlencode({"model": MODEL_ID})
    headers = {"Authorization": "Bearer " + os.environ["KEDANG_API_KEY"]}
    temporary = None
    try:
        with tempfile.NamedTemporaryFile(dir=".", suffix=".pcm.part", delete=False) as output:
            temporary = Path(output.name)
            # 120 秒为本客户端示例的总期限，超时会关闭连接并保留已有目标文件。
            async with asyncio.timeout(120):
                async with connect(url, additional_headers=headers, open_timeout=10) as socket:
                    await exchange_audio(socket, pcm, output)
        temporary.replace("reply.pcm")
        print("\n音频已保存为 reply.pcm")
    finally:
        if temporary is not None:
            temporary.unlink(missing_ok=True)


if __name__ == "__main__":
    asyncio.run(main())

展开 FFmpeg：转换输入音频与包装输出 WAV

# 将已有音频转换为本节示例使用的裸 PCM16。
ffmpeg -i input.wav -ac 1 -ar 24000 -c:a pcm_s16le -f s16le input.pcm

# 将收到的同规格裸 PCM 包装为 WAV，以便播放器播放。
ffmpeg -f s16le -ar 24000 -ac 1 -i reply.pcm reply.wav