流式语音代理

文档摘要

流式语音代理使用 OpenAI Agents SDK 实现实时语音交互的示例，支持连续音频流。此示例展示了先进的语音处理管道功能，包括实时语音检测、实时处理以及基于轮次的对话管理。本示例演示的内容实时音频处理：连续音频输入与输出流活动检测：自动检测语音开始与结束轮次管理：智能对话轮次处理实时代理处理：对话过程中实时生成代理响应打断处理：对话流程管理的生命周期事件流式回调：实时监控与调试核心概念：流式语音管道流式语音管道以实时方式连续处理音频。

流式语音代理

使用 OpenAI Agents SDK 实现实时语音交互的示例，支持连续音频流。此示例展示了先进的语音处理管道功能，包括实时语音检测、实时处理以及基于轮次的对话管理。

本示例演示的内容

实时音频处理：连续音频输入与输出流
活动检测：自动检测语音开始与结束
轮次管理：智能对话轮次处理
实时代理处理：对话过程中实时生成代理响应
打断处理：对话流程管理的生命周期事件
流式回调：实时监控与调试

核心概念：流式语音管道

流式语音管道以实时方式连续处理音频。可以将其想象成一个实时对话助手，它：

持续监听音频输入
自动检测你何时开始和停止说话
使用 AI 代理实时处理语音
随时生成并回传响应
自动管理对话轮次


┌─────────────────────────────────────────────────────────────┐
│                   STREAMING VOICE WORKFLOW                  │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│   CONTINUOUS AUDIO INPUT                                  │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐    1. REAL-TIME CAPTURE                    │
│  │  STREAMING  │    ◦ Continuous microphone input           │
│  │   AUDIO     │    ◦ Chunk-based processing                │
│  │  RECORDER   │    ◦ Activity detection                    │
│  └─────────────┘                                            │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐    2. LIVE TRANSCRIPTION                   │
│  │ STREAMING   │    ◦ Real-time speech-to-text              │
│  │TRANSCRIPTION│    ◦ Turn boundary detection               │
│  └─────────────┘                                            │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐    3. CONCURRENT PROCESSING                │
│  │  PARALLEL   │    ◦ Agent workflow execution              │
│  │ AGENT EXEC  │    ◦ Tool calls & handoffs                 │
│  └─────────────┘    ◦ Multiple turns in session             │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────────┐    4. STREAMING RESPONSE                   │
│  │  LIVE TTS   │    ◦ Real-time text-to-speech              │
│  │  PLAYBACK   │    ◦ Chunked audio output                  │
│  └─────────────┘    ◦ Immediate response playback           │
│       │                                                     │
│       ▼                                                     │
│   CONTINUOUS AUDIO OUTPUT                                 │
│                                                             │
│  ↺ LOOP FOR MULTIPLE TURNS                                  │
└─────────────────────────────────────────────────────────────┘

快速入门

安装语音依赖项：


pip install 'openai-agents[voice]'
pip install sounddevice numpy soundfile librosa

设置环境：


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

运行流式语音代理：
```
python agent.py
```
开始讲话：代理会自动检测到你的讲话，并实时作出回应！

本示例包含的内容

实时音频管理

StreamedAudioRecorder：带线程的连续麦克风输入
AudioPlayer：带流管理的实时音频播放
活动检测：自动检测语音开始与结束
轮次处理：智能对话轮次管理

高级代理功能

多语言支持：英语、西班牙语和法语代理
增强工具：天气、时间、提醒和新闻
实时切换：流式过程中的语言检测
会话管理：多轮对话跟踪

流式工具

get_weather(city): Real-time weather information
get_time(): Current time with live updates
set_reminder(message, minutes): Demo reminder functionality
get_news_summary()：模拟新闻更新

高级监控

StreamingWorkflowCallbacks：实时事件监控
VoiceSessionManager：会话生命周期管理
轮次跟踪：对话分析与统计
生命周期事件：轮次开始与结束事件处理

示例交互

自然对话流程

开始讲话 → 代理自动检测语音
暂停 → 代理立即处理并作出回应
继续讲话 → 新一轮对话自动开始
单次会话中多轮对话

实时工具使用

“纽约的天气怎么样？” → 立即获取天气信息
“现在几点了？” → 获取实时时间
“设置一个15分钟后给莎拉打电话的提醒” → 提醒确认
“给我一份新闻摘要” → 当前新闻更新

实时语言切换

用英语讲话 → 英语代理作出回应
切换到“¿Qué tiempo hace en Madrid？” → 西班牙语代理接手
切换到“Quelle heure est-il？” → 法语代理作出回应
无缝的语言检测与切换

关键实现模式

1. 流式管道设置


pipeline = VoicePipeline(
    workflow=SingleAgentVoiceWorkflow(agent, callbacks=StreamingWorkflowCallbacks())
)

2. 连续音频输入


with StreamedAudioRecorder() as recorder:
    streamed_input = StreamedAudioInput()
    
    while session_active:
        if recorder.has_audio():
            audio_chunk = recorder.get_audio_chunk()
            streamed_input.push_audio(audio_chunk)

3. 实时音频输出


with AudioPlayer() as player:
    async for event in result.stream():
        if event.type == "voice_stream_event_audio":
            player.add_audio(event.data)
        elif event.type == "voice_stream_event_lifecycle":
            handle_turn_events(event)

4. 会话管理


class VoiceSessionManager:
    async def start_session(self):
        # Concurrent input/output processing
        input_task = asyncio.create_task(self._process_audio_input())
        output_task = asyncio.create_task(self._process_audio_output())
        await asyncio.gather(input_task, output_task)

流式语音最佳实践

活动检测：让管道自动处理语音检测
轮次管理：使用生命周期事件管理对话流程
并发处理：同时处理输入与输出流
缓冲区管理：在轮次之间加入静音缓冲区，使对话更自然
错误恢复：为流式失败实现健壮的错误处理
资源管理：正确清理音频流与资源

性能特征

流式管道优势

实时交互：对用户语音即时响应
自然对话：连续流畅的对话
活动检测：自动检测轮次边界
并发处理：输入与输出并行处理
可扩展性：高效处理多轮对话

技术优势

低延迟：语音与响应之间延迟极小
自适应：应对不同语音模式
健壮：自动错误恢复与继续
高效：基于分块处理，性能最优

流式功能

自动轮次检测

语音活动检测：自动检测用户何时开始讲话
静音检测：识别用户何时结束讲话
轮次边界：智能对话轮次管理
持续监听：随时准备接收下一次输入

实时处理

实时转录：边讲边转文字
流式代理响应：语音期间进行 AI 处理
即时音频输出：响应生成时立即转为语音
并行操作：多个进程同时运行

生命周期管理

轮次事件：turn_started and turn_ended notifications
Session Tracking: Multi-turn conversation analytics
State Management: Proper resource allocation and cleanup
Interruption Handling: Graceful handling of user interruptions

Requirements & Dependencies

Core Dependencies

openai-agents[voice]: OpenAI Agents SDK with voice support
sounddevice: Real-time audio I/O
numpy: Audio data processing
threading: Concurrent audio processing
asyncio: Asynchronous pipeline management

System Requirements

Real-Time Audio: Low-latency audio hardware
Microphone: Good quality microphone for speech detection
Processing Power: Sufficient CPU for real-time processing
Network: Stable connection for streaming API calls

Static Voice: Turn-based voice interaction
Voice Pipeline Documentation: Official pipeline docs
Streaming Events: Voice event handling

️ Advanced Customization

Custom Activity Detection

Implement custom speech detection algorithms
Add voice activity thresholds
Configure silence detection parameters

Enhanced Session Management

Add conversation memory across sessions
Implement user authentication
Add conversation logging and analytics

Real-Time Features

Add live transcription display
Implement real-time sentiment analysis
Add voice emotion detection

Streaming Considerations

Interruption Handling

The SDK currently doesn't support built-in interruptions. Use lifecycle events to:

Mute microphone during AI responses (turn_started)
Unmute microphone after responses (turn_ended
平稳处理用户打断

性能优化

缓冲区大小：优化音频分块大小，兼顾延迟与质量
并发限制：平衡处理线程，提升性能
内存管理：定期清理音频缓冲区
网络优化：优雅处理 API 调用失败

专业提示

从简单入手：先从基础流式开始，逐步添加功能
监控生命周期事件：利用回调了解轮次流程
测试音频硬件：确保低延迟音频设置
处理边缘情况：规划好网络问题与音频故障
优化对话体验：设计代理，让对话更自然流畅

下一步

掌握流式语音代理后：

生产部署：将流式语音扩展到实际应用
自定义语音模型：集成专用语音模型
多模态代理：结合语音、视觉与文本
企业级语音解决方案：构建稳健的语音应用

故障排除

常见问题

音频延迟：检查音频硬件与缓冲区设置
语音检测：调整麦克风音量与灵敏度
轮次管理：监控生命周期事件以便调试
资源使用：监控流式过程中的 CPU 与内存使用
网络问题：为 API 失败实现重试逻辑

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

作者与出处

原作者： Shubhamsaboo

来源：Shubhamsaboo

原文： https://github.com/Shubhamsaboo/awesome-llm-apps

许可证：Apache-2.0

整理：灏天文库整理

由灏天文库结构化整理，提供目录导航、全文检索与在线阅读，便于系统化学习

流式语音代理

文档摘要