POST https://computevault.unodetech.xyz/v1/realtime/sessions
Content-Type: application/json
Authorization: Bearer $API_KEY

{
  "model": "gpt-4o-realtime-preview-2024-12-17",
  "voice": "verse"
}

sequenceDiagram
    participant Client
    participant Server
    participant OpenAI
    
    alt WebRTC 连接
        Client->>Server: 请求临时令牌
        Server->>OpenAI: 创建会话
        OpenAI-->>Server: 返回临时令牌
        Server-->>Client: 返回临时令牌
        
        Client->>OpenAI: 创建 WebRTC offer
        OpenAI-->>Client: 返回 answer
        
        Note over Client,OpenAI: 建立 WebRTC 连接
        
        Client->>OpenAI: 创建数据通道
        OpenAI-->>Client: 确认数据通道
    else WebSocket 连接
        Server->>OpenAI: 建立 WebSocket 连接
        OpenAI-->>Server: 确认连接
        
        Note over Server,OpenAI: 开始实时对话
    end

import WebSocket from "ws";

const url = "wss://computevault.unodetech.xyz/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17";
const ws = new WebSocket(url, {
  headers: {
    "Authorization": "Bearer " + process.env.API_KEY,
    "OpenAI-Beta": "realtime=v1",
  },
});

ws.on("open", function open() {
  console.log("Connected to server.");
});

ws.on("message", function incoming(message) {
  console.log(JSON.parse(message.toString()));
});

# 需要安装 websocket-client 库:
# pip install websocket-client

import os
import json
import websocket

API_KEY = os.environ.get("API_KEY")

url = "wss://computevault.unodetech.xyz/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17"
headers = [
    "Authorization: Bearer " + API_KEY,
    "OpenAI-Beta: realtime=v1"
]

def on_open(ws):
    print("Connected to server.")

def on_message(ws, message):
    data = json.loads(message)
    print("Received event:", json.dumps(data, indent=2))

ws = websocket.WebSocketApp(
    url,
    header=headers,
    on_open=on_open,
    on_message=on_message,
)

ws.run_forever()

/*
注意：在浏览器等客户端环境中，我们建议使用WebRTC。
但在Deno和Cloudflare Workers等类浏览器环境中，
也可以使用标准WebSocket接口。
*/

const ws = new WebSocket(
  "wss://computevault.unodetech.xyz/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17",
  [
    "realtime",
    // 认证
    "openai-insecure-api-key." + API_KEY, 
    // 可选
    "openai-organization." + OPENAI_ORG_ID,
    "openai-project." + OPENAI_PROJECT_ID,
    // Beta协议，必需
    "openai-beta.realtime-v1"
  ]
);

ws.on("open", function open() {
  console.log("Connected to server.");
});

ws.on("message", function incoming(message) {
  console.log(message.data);
});

// 接收服务器事件
ws.on("message", function incoming(message) {
  // 需要从JSON解析消息数据
  const serverEvent = JSON.parse(message.data)
  console.log(serverEvent);
});

// 发送事件，创建符合客户端事件格式的JSON数据结构
const event = {
  type: "response.create",
  response: {
    modalities: ["audio", "text"],
    instructions: "Give me a haiku about code.",
  }
};
ws.send(JSON.stringify(event));

# 发送客户端事件，将字典序列化为JSON
def on_open(ws):
    print("Connected to server.")
    
    event = {
        "type": "response.create",
        "response": {
            "modalities": ["text"],
            "instructions": "Please assist the user."
        }
    }
    ws.send(json.dumps(event))

# 接收消息需要从JSON解析消息负载
def on_message(ws, message):
    data = json.loads(message)
    print("Received event:", json.dumps(data, indent=2))

pip install -r requirements.txt

# Ubuntu/Debian
sudo apt-get install portaudio19-dev python3-pyaudio

# CentOS/RHEL
sudo yum install portaudio-devel

WEBSOCKET_URL = "wss://new-api.weroam.xyz/v1/realtime"
API_KEY = "sk-QMA3eCob2EmDI2graXo4onUupApVnrk8wPXemJ7lSfK4QPa0"
MODEL = "gpt-4o-realtime-preview-2024-12-17"

logging.basicConfig(level=logging.DEBUG)

├── openai_realtime_client.py  # 主程序文件
├── requirements.txt           # Python依赖
└── README.md                  # 说明文档

#!/usr/bin/env python3
"""
OpenAI Realtime WebSocket 语音示例
支持实时语音对话，包括音频录制、发送和播放
"""

import asyncio
import json
import base64
import websockets
import pyaudio
import wave
import threading
import time
from typing import Optional
import logging

# 配置日志
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('websocket_debug.log', encoding='utf-8')
    ]
)
logger = logging.getLogger(__name__)

class OpenAIRealtimeClient:
    def __init__(self, 
                 websocket_url: str,
                 api_key: str,
                 model: str = "gpt-4o-realtime-preview-2024-12-17"):
        self.websocket_url = websocket_url
        self.api_key = api_key
        self.model = model
        self.websocket = None
        self.is_recording = False
        self.is_connected = False
        
        # 音频配置
        self.audio_format = pyaudio.paInt16
        self.channels = 1
        self.rate = 24000  # OpenAI Realtime API 要求的采样率
        self.chunk = 1024
        self.audio = pyaudio.PyAudio()
        
        # 音频流
        self.input_stream = None
        self.output_stream = None
        
    async def connect(self):
        """连接到WebSocket服务器"""
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "OpenAI-Beta": "realtime=v1"
        }
        
        logger.info("=" * 80)
        logger.info("🚀 开始WebSocket连接")
        logger.info("=" * 80)
        logger.info(f"连接URL: {self.websocket_url}")
        logger.info(f"API密钥: {self.api_key[:10]}...")
        logger.info(f"请求头: {json.dumps(headers, ensure_ascii=False, indent=2)}")
        
        try:
            self.websocket = await websockets.connect(
                self.websocket_url,
                additional_headers=headers
            )
            self.is_connected = True
            logger.info("✅ WebSocket连接成功")
            
            # 发送会话配置
            await self.send_session_config()
            
        except Exception as e:
            logger.error(f"❌ WebSocket连接失败: {e}")
            logger.error(f"错误类型: {type(e).__name__}")
            raise
    
    async def send_session_config(self):
        """发送会话配置"""
        config = {
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "instructions": "你是一个有用的AI助手，可以进行实时语音对话。",
                "voice": "alloy",
                "input_audio_format": "pcm16",
                "output_audio_format": "pcm16",
                "input_audio_transcription": {
                    "model": "whisper-1"
                },
                "turn_detection": {
                    "type": "server_vad",
                    "threshold": 0.5,
                    "prefix_padding_ms": 300,
                    "silence_duration_ms": 500
                },
                "tools": [],
                "tool_choice": "auto",
                "temperature": 0.8,
                "max_response_output_tokens": 4096
            }
        }
        
        config_json = json.dumps(config, ensure_ascii=False, indent=2)
        logger.info("=" * 60)
        logger.info("📤 发送会话配置:")
        logger.info(f"消息类型: {config['type']}")
        logger.info(f"配置内容:\n{config_json}")
        logger.info("=" * 60)
        
        await self.websocket.send(json.dumps(config))
        logger.info("✅ 会话配置已发送")
    
    def start_audio_streams(self):
        """启动音频输入和输出流"""
        try:
            # 输入流（麦克风）
            self.input_stream = self.audio.open(
                format=self.audio_format,
                channels=self.channels,
                rate=self.rate,
                input=True,
                frames_per_buffer=self.chunk
            )
            
            # 输出流（扬声器）
            self.output_stream = self.audio.open(
                format=self.audio_format,
                channels=self.channels,
                rate=self.rate,
                output=True,
                frames_per_buffer=self.chunk
            )
            
            logger.info("音频流已启动")
            
        except Exception as e:
            logger.error(f"启动音频流失败: {e}")
            raise
    
    def stop_audio_streams(self):
        """停止音频流"""
        if self.input_stream:
            self.input_stream.stop_stream()
            self.input_stream.close()
            self.input_stream = None
            
        if self.output_stream:
            self.output_stream.stop_stream()
            self.output_stream.close()
            self.output_stream = None
            
        logger.info("音频流已停止")
    
    async def start_recording(self):
        """开始录音并发送音频数据"""
        self.is_recording = True
        logger.info("开始录音...")
        
        try:
            while self.is_recording and self.is_connected:
                # 读取音频数据
                audio_data = self.input_stream.read(self.chunk, exception_on_overflow=False)
                
                # 将音频数据编码为base64
                audio_base64 = base64.b64encode(audio_data).decode('utf-8')
                
                # 发送音频数据
                message = {
                    "type": "input_audio_buffer.append",
                    "audio": audio_base64
                }
                
                # 记录音频数据发送（每10次记录一次，避免日志过多）
                if hasattr(self, '_audio_count'):
                    self._audio_count += 1
                else:
                    self._audio_count = 1
                
                if self._audio_count % 10 == 0:  # 每10次记录一次
                    logger.debug(f"🎤 发送音频数据 #{self._audio_count}: 长度={len(audio_base64)} 字符")
                
                await self.websocket.send(json.dumps(message))
                
                # 短暂延迟以避免过度发送
                await asyncio.sleep(0.01)
                
        except Exception as e:
            logger.error(f"录音过程中出错: {e}")
        finally:
            logger.info("录音已停止")
    
    async def stop_recording(self):
        """停止录音"""
        self.is_recording = False
        
        # 发送录音结束信号
        if self.websocket and self.is_connected:
            message = {
                "type": "input_audio_buffer.commit"
            }
            
            logger.info("=" * 60)
            logger.info("📤 发送录音结束信号:")
            logger.info(f"消息类型: {message['type']}")
            logger.info("=" * 60)
            
            await self.websocket.send(json.dumps(message))
            logger.info("✅ 录音结束信号已发送")
    
    async def handle_response(self):
        """处理WebSocket响应"""
        try:
            async for message in self.websocket:
                data = json.loads(message)
                message_type = data.get("type", "unknown")
                
                # 详细记录所有接收到的消息
                logger.info("=" * 60)
                logger.info("📥 收到WebSocket消息:")
                logger.info(f"消息类型: {message_type}")
                
                # 根据消息类型进行不同的处理
                if message_type == "response.audio.delta":
                    # 处理音频响应
                    audio_data = base64.b64decode(data.get("delta", ""))
                    logger.info(f"🎵 音频数据: 长度={len(audio_data)} 字节")
                    if audio_data and self.output_stream:
                        self.output_stream.write(audio_data)
                        logger.info("✅ 音频数据已播放")
                
                elif message_type == "response.text.delta":
                    # 处理文本响应
                    text = data.get("delta", "")
                    logger.info(f"💬 文本增量: '{text}'")
                    if text:
                        print(f"AI: {text}", end="", flush=True)
                
                elif message_type == "response.text.done":
                    # 文本响应完成
                    logger.info("✅ 文本响应完成")
                    print("\n")
                
                elif message_type == "response.audio.done":
                    # 音频响应完成
                    logger.info("✅ 音频响应完成")
                
                elif message_type == "error":
                    # 处理错误
                    error_info = data.get('error', {})
                    logger.error("❌ 服务器错误:")
                    logger.error(f"错误详情: {json.dumps(error_info, ensure_ascii=False, indent=2)}")
                
                elif message_type == "session.created":
                    # 会话创建成功
                    logger.info("✅ 会话已创建")
                
                elif message_type == "session.updated":
                    # 会话更新成功
                    logger.info("✅ 会话已更新")
                
                elif message_type == "conversation.item.created":
                    # 对话项目创建
                    logger.info("📝 对话项目已创建")
                
                elif message_type == "conversation.item.input_audio_buffer.speech_started":
                    # 语音开始
                    logger.info("🎤 语音开始检测")
                
                elif message_type == "conversation.item.input_audio_buffer.speech_stopped":
                    # 语音结束
                    logger.info("🔇 语音结束检测")
                
                elif message_type == "conversation.item.input_audio_buffer.committed":
                    # 音频缓冲区已提交
                    logger.info("📤 音频缓冲区已提交")
                
                else:
                    # 其他未知消息类型
                    logger.info(f"❓ 未知消息类型: {message_type}")
                
                # 记录完整的消息内容（除了音频数据，因为太长）
                if message_type != "response.audio.delta":
                    logger.info(f"完整消息内容:\n{json.dumps(data, ensure_ascii=False, indent=2)}")
                
                logger.info("=" * 60)
                    
        except websockets.exceptions.ConnectionClosed:
            logger.info("WebSocket连接已关闭")
            self.is_connected = False
        except Exception as e:
            logger.error(f"处理响应时出错: {e}")
            self.is_connected = False
    
    async def start_conversation(self):
        """开始对话"""
        try:
            # 启动音频流
            self.start_audio_streams()
            
            # 创建任务
            response_task = asyncio.create_task(self.handle_response())
            recording_task = asyncio.create_task(self.start_recording())
            
            logger.info("对话已开始，按 Ctrl+C 停止")
            
            # 等待任务完成
            await asyncio.gather(response_task, recording_task)
            
        except KeyboardInterrupt:
            logger.info("收到停止信号")
        except Exception as e:
            logger.error(f"对话过程中出错: {e}")
        finally:
            await self.cleanup()
    
    async def cleanup(self):
        """清理资源"""
        self.is_recording = False
        self.is_connected = False
        
        # 停止音频流
        self.stop_audio_streams()
        
        # 关闭WebSocket连接
        if self.websocket:
            await self.websocket.close()
            logger.info("WebSocket连接已关闭")
        
        # 终止PyAudio
        self.audio.terminate()
        logger.info("资源清理完成")
    
    async def run(self):
        """运行客户端"""
        try:
            await self.connect()
            await self.start_conversation()
        except Exception as e:
            logger.error(f"运行客户端时出错: {e}")
        finally:
            await self.cleanup()


async def main():
    """主函数"""
    # 配置参数
    WEBSOCKET_URL = "wss://new-api.weroam.xyz/v1/realtime?model=gpt-4o-realtime-preview-2024-12-17"
    API_KEY = "sk-EpnduEXFxjAt0AF55W08WBmzqZHlv9f4tmCDWd9TcJqBwVjV"
    MODEL = "gpt-4o-realtime-preview-2024-12-17"
    
    # 创建客户端
    client = OpenAIRealtimeClient(
        websocket_url=WEBSOCKET_URL,
        api_key=API_KEY,
        model=MODEL
    )
    
    # 运行客户端
    await client.run()


if __name__ == "__main__":
    print("OpenAI Realtime WebSocket 语音示例")
    print("=" * 50)
    print("功能说明:")
    print("- 实时语音对话")
    print("- 自动语音识别")
    print("- 文本和音频响应")
    print("- 按 Ctrl+C 停止")
    print("=" * 50)
    
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n程序已停止")
    except Exception as e:
        print(f"程序出错: {e}")