让 AI 构建实时语音应用与通话助手
解决传统文本交互的延迟与情感缺失：通过集成 OpenAI Realtime API、Vapi 及 Deepgram，让 AI 直接理解语音、实时调用工具并流利地合成回答，打造具有真实对话感的语音助手与电话机器人。
为什么需要这个技能
在现代客服、物联网设备及智能助理场景中，用户更倾向于用“说话”而非“打字”。然而，构建低延迟（低毫秒级响应）且能打断对话的语音应用极其复杂，涉及音频流处理、WebRTC 连接管理及 LLM 流式输出的精细编排。
此技能提供了从基础 API 对接到复杂中断处理（Interruption Handling）的完整方案，帮助开发者将文本型 AI 模型转化为真正的“听得懂、能说话”的实时语音系统。
适用场景
- 需要构建支持打断、低延迟的实时语音客服机器人。
- 开发可连接电话线路（PSTN/SIP）的智能外呼系统。
- 创建支持多轮对话的会议记录或实时翻译助手。
- 集成 WebRTC 实现设备间的直接语音交互。
核心工作流
构建语音 AI 应用通常遵循“接收音频 -> 实时转译 -> 思考 -> 合成音频 -> 播放”的管道。
1. 建立实时会话与配置 (OpenAI Realtime API)
首先需建立 WebSocket 连接，配置语音活动检测（VAD）以支持打断功能，并定义工具以赋予 AI 操作能力。
import asyncio
import websockets
import json
import base64
OPENAI_API_KEY = "sk-..."
async def voice_session():
url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview"
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"OpenAI-Beta": "realtime=v1"
}
async with websockets.connect(url, extra_headers=headers) as ws:
# Configure session
await ws.send(json.dumps({
"type": "session.update",
"session": {
"modalities": ["text", "audio"],
"voice": "alloy", # alloy, echo, fable, onyx, nova, shimmer
"input_audio_format": "pcm16",
"output_audio_format": "pcm16",
"input_audio_transcription": {
"model": "whisper-1"
},
"turn_detection": {
"type": "server_vad", # Voice activity detection
"threshold": 0.5,
"prefix_padding_ms": 300,
"silence_duration_ms": 500
},
"tools": [
{
"type": "function",
"name": "get_weather",
"description": "Get weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"}
}
}
}
]
}
}))2. 音频流处理与转译 (Deepgram STT)
为了降低延迟，应使用 Deepgram 进行流式实时转写（Transcription），并在用户说话时持续接收数据，支持 VAD 截断以准备打断当前回答。
import asyncio
from deepgram import DeepgramClient, LiveTranscriptionEvents
# Deepgram real-time transcription
deepgram = DeepgramClient(api_key="...")
async def transcribe_stream(audio_stream):
connection = deepgram.listen.live.v("1")
async def on_transcript(result):
transcript = result.channel.alternatives[0].transcript
if transcript:
print(f"Heard: {transcript}")
if result.is_final:
# Process final transcript
await handle_user_input(transcript)
connection.on(LiveTranscriptionEvents.Transcript, on_transcript)
await connection.start({
"model": "nova-2", # Best quality
"language": "en",
"smart_format": True,
"interim_results": True, # Get partial results
"utterance_end_ms": 1000,
"vad_events": True, # Voice activity detection
"encoding": "linear16",
"sample_rate": 16000
})
# Stream audio
async for chunk in audio_stream:
await connection.send(chunk)
await connection.finish()3. 低延迟文本转语音 (ElevenLabs TTS)
回复不应等待 LLM 完全生成完毕。策略是接收 LLM 的第一个词即送入 TTS 缓冲区，待积累约 50 个字符或遇到标点符号时即刻开始播放。
from elevenlabs import ElevenLabs
# ElevenLabs streaming synthesis
eleven = ElevenLabs(api_key="...")
def text_to_speech_stream(text: str):
"""Stream TTS audio chunks."""
audio_stream = eleven.text_to_speech.convert_as_stream(
voice_id="21m00Tcm4TlvDq8ikWAM", # Rachel
model_id="eleven_turbo_v2_5", # Fastest
text=text,
output_format="pcm_24000" # Raw PCM for low latency
)
for chunk in audio_stream:
yield chunk4. 基础设施集成 (LiveKit)
若需自建房间或复杂通信，可使用 LiveKit。利用其 WebRTC 房间管理模块，处理音频流的订阅与发布，并整合前端的音频源捕获逻辑。
from livekit import api, rtc
# Server-side: Create room and tokens
lk_api = api.LiveKitAPI(
url="wss://your-livekit.livekit.cloud",
api_key="...",
api_secret="..."
)
async def create_room(room_name: str):
room = await lk_api.room.create_room(
api.CreateRoomRequest(name=room_name)
)
return room下载和安装
下载 voice-ai-development 中文版 Skill ZIP
解压后将目录放入你的 AI 工具 skills 文件夹，重启工具后即可使用。具体路径参考内附的说明文档。
你可能还需要
暂无推荐