实时语音助手

⚡ 实时语音助手

这是一个使用 OpenAI 实时 API 的基本实时语音助手示例。它展示了实现超低延迟语音对话的核心组件，只需最少的配置即可完成。

本演示所展示的内容

• 核心实时组件：RealtimeAgent、RealtimeRunner 和 RealtimeSession
• 基础语音对话：超低延迟语音交互
• 函数工具：在语音对话期间可调用的简单工具
• 代理交接：简单的专业代理交接
• 事件处理：实时会话的关键事件处理

核心概念：实时语音处理

实时代理通过 OpenAI 的实时 API 提供了超低延迟的语音对话。与传统语音管道不同，实时代理保持持久的 WebSocket 连接，以便即时处理音频。可以将实时代理视为实时对话伙伴，它们：

• 以极低延迟即时处理音频并作出响应
• 在对话过程中优雅地处理打断
• 保持持久连接，实现自然的对话流程
• 支持实时工具执行和代理交接
• 在实时生成过程中应用安全防护措施

根据官方文档，实时代理能够以尽可能低的延迟实现自然的语音对话。

┌─────────────────────────────────────────────────────────────┐
│                   REALTIME VOICE WORKFLOW                   │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   LIVE AUDIO INPUT                                        |
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐    1. WEBSOCKET CONNECTION                 │
│  │ PERSISTENT  │    ◦ Continuous audio streaming            │
│  │ CONNECTION  │    ◦ Ultra-low latency pipeline            │
│  └─────────────┘    ◦ Real-time processing                  │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐    2. INSTANT PROCESSING                   │
│  │ REALTIME    │    ◦ Immediate speech recognition          │
│  │ AGENTS      │    ◦ Live agent reasoning                  │
│  └─────────────┘    ◦ Real-time tool execution              │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐    3. IMMEDIATE RESPONSE                   │
│  │   LIVE      │    ◦ Real-time audio generation            │
│  │ RESPONSE    │    ◦ Streaming audio output                │
│  └─────────────┘    ◦ Interruption handling                 │
│       │                                                     │
│       ▼                                                     │
│   INSTANT AUDIO OUTPUT                                    |
│                                                             │
│  ↺ CONTINUOUS CONVERSATION LOOP                             │
└─────────────────────────────────────────────────────────────┘
​

快速入门

1. 安装 OpenAI Agents SDK：

   pip install openai-agents
   ​

2. 设置环境：

   cp env.example .env
   # Edit .env and add your OpenAI API key
   ​

3. 运行基本实时代理：

   python agent.py
   ​

4. 开始对话：代理将实时回应。试试以下指令：

   • “巴黎的天气怎么样？”
   • “明天下午2点预约”

本示例包含的内容

核心实时组件

根据官方指南：

• RealtimeAgent：带有指令、工具和交接功能的代理
• RealtimeRunner：管理配置并返回会话
• RealtimeSession：单个对话会话，支持事件流

基础函数工具

• get_weather(city): Simple weather information
• book_appointment(date, time, service)：基础预约工具

简单代理交接

• 主助理：通用对话代理
• 计费代理：专门的计费支持（演示交接模式）

关键事件处理

• 音频转录：用户和助理的语音转录
• 工具调用：函数执行通知
• 错误事件：基础错误处理

示例语音交互

基础对话

• “巴黎的天气怎么样？” → 工具调用并即时响应
• “明天下午2点预约” → 预约工具

代理交接

• “我需要计费帮助” → 交接至计费支持代理

关键实现模式

根据官方指南：

1. 创建 RealtimeAgent

from agents.realtime import RealtimeAgent

agent = RealtimeAgent(
    name="Assistant",
    instructions="You are a helpful voice assistant...",
    tools=[get_weather, book_appointment],
    handoffs=[realtime_handoff(billing_agent)]
)
​

2. 设置 RealtimeRunner

from agents.realtime import RealtimeRunner

runner = RealtimeRunner(
    starting_agent=agent,
    config={
        "model_settings": {
            "model_name": "gpt-4o-realtime-preview",
            "voice": "alloy",
            "modalities": ["text", "audio"]
        }
    }
)
​

3. 启动会话并处理事件

session = await runner.run()

async with session:
    async for event in session:
        if event.type == "response.audio_transcript.done":
            print(f"Assistant: {event.transcript}")
​

基础实时概念

来自官方指南：

1. 会话流程：创建代理 → 设置 runner → 启动会话 → 处理事件
2. 事件处理：监听音频转录、工具调用和错误
3. 语音配置：从6种声音中选择（alloy、echo、fable、onyx、nova、shimmer）
4. 断句检测：服务器端语音活动检测，实现自然对话

实时与传统语音对比

高级实时功能

语音活动检测 (VAD)

• 服务器VAD：OpenAI优化的语音检测
• 可配置阈值：针对不同环境调整灵敏度
• 静音检测：智能断句边界检测
• 前缀填充：准确捕捉语音起始

音频配置选项

• 声音选择：从6种不同声音中选择（alloy、echo、fable、onyx、nova、shimmer）
• 音频格式：支持PCM16、G.711 μ-law和G.711 A-law
• 转录模型：集成Whisper进行语音转文本
• 多模态支持：文本和音频模态

实时防护措施

根据指南文档，实时防护措施包括：

• 去抖动：定期运行（而非每字触发），提升性能
• 可配置：可调节去抖动长度（默认100字符）
• 非阻塞：不抛出异常，而是生成事件
• 实时性：触发时可立即中断响应

会话事件类型

• 音频事件：response.audio.delta, response.audio.done
• Transcription Events: response.audio_transcript.done, input_audio_transcription.completed
• Tool Events: response.function_call_arguments.done
• Lifecycle Events: session.created, session.updated, response.done
• Error Events: error, guardrail_tripped

Requirements & Dependencies

Core Dependencies

• openai-agents>=1.0.0: OpenAI Agents SDK with realtime support
• python-dotenv>=1.0.0: Environment variable management
• Python 3.9 or higher (required for realtime features)

API Requirements

• OpenAI API Key: Required for Realtime API access
• Model Access: Access to gpt-4o-realtime-preview模型
• WebSocket支持：稳定互联网连接，实现持久连接

系统要求

• 实时能力：低延迟网络连接
• 音频硬件：麦克风和扬声器用于语音交互
• 处理能力：足够CPU用于实时音频处理

配置选项

模型设置

"model_settings": {
    "model_name": "gpt-4o-realtime-preview",  # Realtime model
    "voice": "alloy",                         # Voice selection
    "modalities": ["text", "audio"],          # Supported modalities
    "input_audio_format": "pcm16",           # Audio input format
    "output_audio_format": "pcm16"           # Audio output format
}
​

断句检测设置

"turn_detection": {
    "type": "server_vad",           # Voice activity detection
    "threshold": 0.5,               # Detection sensitivity (0.0-1.0)
    "prefix_padding_ms": 300,       # Audio padding before speech
    "silence_duration_ms": 200      # Silence to detect turn end
}
​

转录配置

"input_audio_transcription": {
    "model": "whisper-1",           # Transcription model
    "language": "en",               # Language preference
    "prompt": "Custom prompt..."    # Domain-specific terms
}
​

️ 安全与防护措施

实时安全功能

• 去抖动处理：防护措施定期运行，提升性能
• 即时干预：可实时中断不安全的响应
• 事件驱动报警：生成guardrail_tripped事件，而非抛出异常
• 可配置灵敏度：根据需求调整去抖动长度

安全实现

@output_guardrail
def sensitive_data_guardrail(ctx, agent, output: str) -> GuardrailFunctionOutput:
    if contains_sensitive_data(output):
        return GuardrailFunctionOutput(
            tripwire_triggered=True,
            output_info="Blocked sensitive data"
        )
    return GuardrailFunctionOutput(tripwire_triggered=False)
​

生产注意事项

性能优化

• 连接管理：保持持久WebSocket连接
• 错误恢复：实现自动重连逻辑
• 资源监控：跟踪会话期间的内存和CPU使用情况
• 事件处理：优化事件处理，适应高吞吐场景

扩展性模式

• 会话隔离：每个用户独立实时会话
• 负载均衡：将会话分布到多个实例
• 连接池：高效管理WebSocket连接
• 优雅关闭：妥善处理会话清理

监控与分析

• 事件追踪：监控所有实时事件，获取洞察
• 性能指标：跟踪延迟、吞吐量和错误率
• 用户分析：分析对话模式和成功率
• 安全指标：监控防护措施激活及效果

Beta 注意事项

正如官方文档所述，实时代理目前处于Beta阶段。请注意：

• API稳定性：随着API演进，可能出现破坏性变更
• 功能开发：新功能可能定期添加
• 测试要求：建议在生产部署前进行全面测试
• 反馈渠道：提供反馈，帮助改进实时体验

专业提示

• 从简单入手：先从基础实时对话开始，再逐步加入复杂功能
• 监控事件：使用全面的事件日志了解行为
• 优化防护措施：平衡安全性和实时性能需求
• 测试打断：确保自然处理对话打断
• 规划扩展：设计会话管理，适应生产工作负载

相关文档

• 实时快速入门：官方入门指南
• 实时指南：全面实时文档
• 语音代理：所有语音代理功能概览
• 代理基础：基础代理概念

故障排除

常见问题

• 高延迟：检查网络连接和WebSocket稳定性
• 音频质量：验证麦克风设置和音频格式
• 事件处理：监控事件处理性能和错误
• 防护措施性能：优化去抖动设置，满足实时需求
• 模型访问：确保能访问gpt-4o-realtime-preview模型

调试策略

• 事件日志：启用全面事件调试
• 连接监控：跟踪WebSocket连接健康状况
• 性能剖析：监控会话期间的CPU和内存使用
• 音频管道：验证音频输入输出处理

下一步

掌握实时语音代理后：

• 生产部署：将实时代理规模化应用于生产环境
• 自定义集成：将实时语音融入现有应用
• 高级功能：探索前沿实时能力
• 多模态体验：结合实时语音与其他模态

免责声明：
本文档采用基于机器的 AI 翻译服务进行翻译。尽管我们力求准确，但请注意，自动翻译可能存在错误或不准确之处。应以原文语言版本的文档作为权威依据。如需获取关键信息，建议使用专业的人工翻译。对于因使用本翻译而产生的任何误解或误读，我们概不负责。