OpenAI Realtime API 实战：低延迟语音对话应用

一句话结论：Realtime API 把"AI 语音对话"从科幻变成日常应用。延迟 < 300ms，已经接近真人对话节奏。

背景

2026 年 5 月，OpenAI Realtime API GA：

• 语音输入 → AI 推理 → 语音输出 < 300ms
• 支持打断（用户说话时 AI 立即停止）
• 情感识别（声调、语速）
• 多语言自动检测

5 个核心能力

1. 实时语音对话

import websockets
import asyncio
import base64
import json

async def realtime_chat():
    async with websockets.connect(
        "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime",
        extra_headers={"Authorization": f"Bearer {API_KEY}"},
    ) as ws:
        # 配置会话
        await ws.send(json.dumps({
            "type": "session.update",
            "session": {
                "modalities": ["text", "audio"],
                "voice": "alloy",
                "instructions": "你是友好的客服助手",
            },
        }))
        
        # 发送音频帧（PCM 16kHz）
        while True:
            audio_chunk = get_microphone_input()
            await ws.send(json.dumps({
                "type": "input_audio_buffer.append",
                "audio": base64.b64encode(audio_chunk).decode(),
            }))
            
            # 接收响应
            async for msg in ws:
                event = json.loads(msg)
                if event["type"] == "response.audio.delta":
                    audio = base64.b64decode(event["delta"])
                    play_audio(audio)

2. 函数调用

# Realtime API 支持 function calling
await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "tools": [
            {
                "type": "function",
                "name": "get_weather",
                "description": "获取天气信息",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "location": {"type": "string"},
                    },
                    "required": ["location"],
                },
            },
        ],
    },
}))

3. 视觉输入

# 摄像头 + 语音同时输入
await ws.send(json.dumps({
    "type": "conversation.item.create",
    "item": {
        "type": "message",
        "role": "user",
        "content": [
            {"type": "input_text", "text": "这是什么？"},
            {
                "type": "input_image",
                "image": base64.b64encode(camera_frame).decode(),
            },
        ],
    },
}))

4. 情感识别

# Realtime API 自动检测情感
# 在 response 中返回 sentiment_score
event = json.loads(msg)
if event["type"] == "response.audio.done":
    sentiment = event.get("sentiment", "neutral")
    # 应用：调整 AI 语气

5. 打断处理

# 用户说话时，AI 立即停止当前输出
# 自动处理，无需手动实现
async def handle_input(ws):
    while True:
        # 检测用户说话（VAD - Voice Activity Detection）
        if user_is_speaking():
            # 发送打断事件
            await ws.send(json.dumps({
                "type": "conversation.item.truncate",
                "item_id": current_response_id,
                "content_index": 0,
                "audio_end_ms": current_audio_position,
            }))

4 个真实应用案例

案例 1：AI 客服

# 24/7 智能客服
async def ai_customer_service():
    async with connect_realtime() as ws:
        # 配置：礼貌、专业、能调用订单系统
        configure_session(ws, instructions=CS_INSTRUCTIONS)
        
        while True:
            audio = await get_customer_audio()
            await send_audio(ws, audio)
            # AI 自动回应 + 必要时调用函数

案例 2：语言学习

# AI 陪练英语口语
configure_session(ws, instructions="""
你是英语口语陪练：
1. 用英语对话（用户级别 A2）
2. 每次纠正用户发音 / 语法错误
3. 提供改进建议
""")

案例 3：远程医疗

# AI 预问诊（医生远程诊断前）
configure_session(ws, instructions="""
你是医疗 AI 助理：
1. 询问患者症状
2. 记录关键信息
3. 不要给出诊断（要医生判断）
""")

案例 4：智能音箱

# 嵌入智能音箱
# 实时对话 + IoT 控制

性能优化

降低延迟

# 1. 用流式 chunk 而不是整段音频
chunk_size = 1024  # 字节
# 16kHz PCM 1 帧 = 1024 字节 = 64ms

# 2. WebSocket 压缩
ws = await websockets.connect(
    url,
    compression="deflate",  # 减少 50% 流量
)

# 3. CDN 加速（Cloudflare Workers）

控制成本

Realtime API 定价：
- 音频输入：$100 / 1M tokens (≈ $0.06/分钟)
- 音频输出：$200 / 1M tokens (≈ $0.12/分钟)
- 文字输入：$5 / 1M tokens

示例：1 小时对话 ≈ $10

隐私保护

# 1. 数据加密
# WebSocket 强制 WSS

# 2. 不存储音频
# 默认不持久化（用户主动开启才保存）

# 3. 用户授权
# 第一次使用时明确告知 + 同意

5 个常见坑

坑 1：音频采样率不匹配

# OpenAI 要求 24kHz PCM
# 麦克风默认可能是 16kHz / 44.1kHz
# ✅ 转换采样率
import scipy.signal as signal
audio_24k = signal.resample(audio, len(audio) * 24000 // sample_rate)

坑 2：网络抖动

# 实时对话对网络敏感
# ✅ 用 WebSocket（而不是 HTTP）
# ✅ 自动重连
async def with_retry():
    while True:
        try:
            await realtime_chat()
        except websockets.ConnectionClosed:
            await asyncio.sleep(1)
            continue

坑 3：会话上下文管理

# 对话长了 token 消耗会爆
# ✅ 定期总结历史
await ws.send(json.dumps({
    "type": "conversation.item.create",
    "item": {
        "type": "message",
        "role": "system",
        "content": [{"type": "input_text", "text": "前面对话的总结..."}],
    },
}))

坑 4：VAD 不准确

# 默认 VAD 可能误判（环境噪音）
# ✅ 自定义 VAD
await ws.send(json.dumps({
    "type": "session.update",
    "session": {
        "turn_detection": {
            "type": "server_vad",
            "threshold": 0.5,  # 灵敏度
            "silence_duration_ms": 200,  # 静默多久算结束
        },
    },
}))

坑 5：浏览器权限

// 必须 HTTPS 才能获取麦克风权限
// ✅ 本地开发用 localhost（视为安全）
// ✅ 生产必须 HTTPS

何时用 Realtime API

✅ 适合

• 实时语音客服
• 语音助手 / 智能音箱
• 语言学习陪练
• 远程医疗预问诊

❌ 不适合

• 离线场景（用本地 Whisper）
• 预算敏感（用 ASR + LLM + TTS 链式调用）
• 延迟容忍（用传统 pipeline）

我的看法

Realtime API 标志着 AI 语音应用进入"实时"时代：

1. 延迟 < 300ms → 真人类对话节奏
2. 多模态输入 → 语音 + 视频 + 文本
3. 情感理解 → 不止文字，还有语气

对独立开发者的机会：

• 垂直场景语音助手（医疗 / 法律 / 教育）
• AI 客服（成本 < 人工 1/10）
• 智能硬件（嵌入 AI 大脑）

参考

• Realtime API 文档
• 官方示例代码
• 定价详情

---

本文基于 Realtime API GA 版本，2026 年 5 月实测。