语音助手正在改变人们与技术交互的方式，通过人工智能实现无缝实时通信。借助 Twilio 的电话基础设施和 OpenAI 的生成式模型，您可以轻松创建一个处理动态对话的语音助手。以下是实现过程的完整指南，并包括详细的代码示例和功能说明。
项目概述
使用 Twilio 和 OpenAI 构建的实时语音助手可以实现以下功能：
- 处理实时语音输入，并生成动态语音响应。
- 使用 OpenAI 的 检索增强生成（RAG） 提供上下文感知的答案。
- 通过模拟或集成真实服务扩展应用场景，例如订单查询、库存检查等。
您可以在 GitHub 仓库 中找到完整代码。
实时交互流程
- 用户发起电话呼叫，Twilio 处理呼叫并将音频路由到服务器。
- 服务器通过 WebSocket 将语音输入流式传输到 OpenAI 实时 API。
- OpenAI 分析输入并生成语音响应。
- Twilio 将生成的语音返回给用户，完成互动。
先决条件
- Twilio 帐户：用于接收和处理电话。
- OpenAI API 密钥：用于调用实时模型。
- Python 3.x：项目开发环境。
- ngrok：用于本地开发时暴露 Webhook。
项目设置
1. 克隆项目并安装依赖
克隆项目并安装所需库：
git clone https://github.com/ericrisco/twilio-realtime-openai-rag
cd twilio-realtime-openai-rag
pip install -r requirements.txt
2. 配置环境变量
创建 .env 文件，存储您的 API 密钥和配置：
VOICE='echo'
OPENAI_API_KEY='your_openai_api_key'
WEBSOCKET_URL='https://your-ngrok-url'
PORT=5000
3. 配置 Twilio 电话号码
- 登录 Twilio 控制台。
- 购买一个支持语音的电话号码。
- 配置该号码的 Webhook，将其指向您的服务器
/stream/incoming-call路由。
WEBSOCKET_URL='https://your-ngrok-url/stream/incoming-call'
4. 启动服务器
运行应用程序：
python app.py
在开发环境中，使用 ngrok 将本地服务器暴露到公网：
ngrok http 5000
代码结构和功能
1. app.py
主应用入口，使用 FastAPI 配置路由和 WebSocket 服务。
from fastapi import FastAPI, WebSocket
import uvicorn
app = FastAPI()
# WebSocket 路由，用于接收 Twilio 和 OpenAI 的实时音频流
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
while True:
data = await websocket.receive_text()
# 处理实时输入
response = await process_audio_input(data)
await websocket.send_text(response)
# 启动服务器
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=5000)
2. stream.py
管理 Twilio 与 OpenAI API 的 WebSocket 实时流。
import websockets
import asyncio
async def handle_stream(audio_stream):
async with websockets.connect("wss://api.openai.com/audio-stream") as websocket:
await websocket.send(audio_stream)
response = await websocket.recv()
return response
3. 辅助工具
模拟数据（JSON 数据库）
模拟产品库存和订单状态，用于演示或开发：
data.json 示例：
{
"products": [
{"id": 1, "name": "Product A", "stock": 100},
{"id": 2, "name": "Product B", "stock": 50}
],
"orders": [
{"order_id": "12345", "status": "Shipped"}
]
}
读取模拟数据的工具
import json
def read_data(file_path):
with open(file_path, 'r') as f:
return json.load(f)
4. 调用 OpenAI API
通过 WebSocket 与 OpenAI 实时 API 通信：
import websockets
import json
import os
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
async def openai_real_time(audio_input):
async with websockets.connect("wss://api.openai.com/v1/real-time-audio") as ws:
headers = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"Content-Type": "application/json"
}
await ws.send(json.dumps({"audio": audio_input, "headers": headers}))
response = await ws.recv()
return json.loads(response)
运行和测试语音助手
1. 拨打 Twilio 电话
配置好的 Twilio 号码接收呼叫时，将音频流发送到服务器的 WebSocket 路由。
2. 实时语音交互
Twilio 将用户语音输入流式传输到服务器，OpenAI 实时生成语音响应，Twilio 返回给用户。
3. 运行语音助手
完成配置后，拨打 Twilio 号码即可开始测试语音助手。例如，您可以询问：
- “请告诉我我的订单状态。”
- “某某产品还有库存吗？”
助手会实时处理请求，调用虚拟数据库或实际服务生成响应。
扩展功能
1. 集成真实服务
将模拟的 JSON 数据库替换为真实服务，例如 CRM 或库存管理系统。
def check_real_stock(product_id):
# 调用实际库存系统 API
return {"stock": 20}
2. 增强用户体验
- 多语言支持：结合 OpenAI 的翻译模型实现实时翻译。
- 自然语音增强：使用 OpenAI 的更高版本模型，生成更自然的语音响应。
3. 生产部署建议
- 容器化应用：使用 Docker 部署到云服务（如 AWS 或 GCP）。
- 增强安全性：在生产环境中启用 HTTPS，并实现身份验证机制。
结论
通过 Twilio 和 OpenAI，您可以轻松构建一个功能强大的实时语音助手。本项目不仅提供了实时语音交互的基础，还支持进一步扩展，实现更多高级功能。无论是用于客户支持还是复杂查询处理，这种语音助手都能显著提升用户体验。
更多细节请访问项目 GitHub 仓库。
