Token360
登录注册

连接 Realtime 会话

WS/v1/realtime?model={model}

Token360 提供兼容 OpenAI Realtime 协议的 WebSocket 接口，可用于低延迟文本及音频会话。

请仅从可信的服务端环境连接该接口，不要在浏览器或移动端代码中暴露 API Key。

建立连接

URLwss://api.token360.ai/v1/realtime?model=gpt-realtime-2.1
鉴权在 WebSocket 握手中发送 Authorization: Bearer <TOKEN360_API_KEY>
模型gpt-realtime-2.1
协议基于 WebSocket 的 OpenAI 兼容 GA Realtime 事件协议

不要发送旧版 OpenAI-Beta: realtime=v1 请求头。

文本会话流程

  1. 建立 WebSocket 连接并等待 session.created。
  2. 发送 session.update，将 output_modalities 设置为 ["text"]。
  3. 等待 session.updated 后再发送第一条内容。
  4. 发送包含 input_text 的 conversation.item.create。
  5. 发送 response.create。
  6. 持续读取 response.output_text.delta，直到收到 response.done。

必须等待 session.updated。默认会话配置可能使用音频输出，因此纯文本客户端应在创建第一个响应前明确切换为文本输出。

音频会话

语音到语音场景应通过 session.update 配置音频输入与输出，使用 input_audio_buffer.append 连续发送 Base64 音频分片，然后提交音频输入并创建响应。同一会话内应保持音频格式一致。

音频会话不要使用纯文本的 output_modalities: ["text"] 配置。

常用客户端事件

session.update配置输出模态、指令、工具、音频等会话选项
conversation.item.create添加文本或其他会话内容
input_audio_buffer.append追加 Base64 编码的音频分片
input_audio_buffer.commit将已缓存音频提交为用户输入
response.create请求模型生成响应

常用服务端事件

session.created连接已就绪，可以配置会话
session.updated会话配置已被接受
response.output_text.delta增量文本输出
response.output_audio.delta增量 Base64 音频输出
response.done包含最终状态和 usage 的终止事件
error请求或会话错误

错误处理与重连

  • WebSocket 握手未升级成功时，检查返回的状态和消息，以判断鉴权、权限、模型可用性或请求校验问题。
  • 收到 error 事件时，检查其中的 error.code、error.type、error.param 和 error.message。
  • 对临时断线使用带抖动且有上限的指数退避重连。
  • 新建 WebSocket 会生成新会话，需要重新发送必要的会话配置，并由应用恢复需要保留的状态。
  • Realtime 会话不是永久连接。请在 60 分钟前主动重连，不要依赖无限期保持连接。

查询用量

终止事件 response.done 会返回本次响应的 usage。也可以使用响应 ID 查询已完成请求的用量：

Shell
1curl "https://api.token360.ai/v1/billing/requests/RESPONSE_ID" \
2  -H "Authorization: Bearer $TOKEN360_API_KEY"

请求

安装 WebSocket 客户端：

Shell
1pip install "websockets>=14"

在环境变量中设置 TOKEN360_API_KEY，然后运行：

Python
1import asyncio
2import json
3import os
4
5import websockets
6
7
8URL = (
9    "wss://api.token360.ai/v1/realtime"
10    "?model=gpt-realtime-2.1"
11)
12
13
14async def receive_event(ws):
15    event = json.loads(await ws.recv())
16    if event.get("type") == "error":
17        raise RuntimeError(event.get("error", {}).get("message", "Realtime error"))
18    return event
19
20
21async def wait_for(ws, event_type):
22    while True:
23        event = await receive_event(ws)
24        if event.get("type") == event_type:
25            return event
26
27
28async def main():
29    headers = {
30        "Authorization": f"Bearer {os.environ['TOKEN360_API_KEY']}"
31    }
32
33    async with websockets.connect(URL, additional_headers=headers) as ws:
34        await wait_for(ws, "session.created")
35
36        await ws.send(json.dumps({
37            "type": "session.update",
38            "session": {
39                "type": "realtime",
40                "output_modalities": ["text"]
41            }
42        }))
43        await wait_for(ws, "session.updated")
44
45        await ws.send(json.dumps({
46            "type": "conversation.item.create",
47            "item": {
48                "type": "message",
49                "role": "user",
50                "content": [{
51                    "type": "input_text",
52                    "text": "请只回复：Realtime 已就绪"
53                }]
54            }
55        }))
56        await ws.send(json.dumps({"type": "response.create"}))
57
58        output = []
59        while True:
60            event = await receive_event(ws)
61            if event.get("type") == "response.output_text.delta":
62                output.append(event.get("delta", ""))
63            if event.get("type") == "response.done":
64                response = event.get("response", {})
65                print("".join(output))
66                print(json.dumps(response.get("usage", {}), indent=2))
67                break
68
69
70asyncio.run(main())

其他服务端 WebSocket 库也使用相同的事件顺序。

此页面对您有帮助吗？
连接 Realtime 会话 · Token360