用 Python 实现 OpenAI 低延迟语音 AI：从原理到可运行代码
用 Python 实现 OpenAI 低延迟语音 AI：从原理到可运行代码
很多人以为语音 AI 慢是因为模型本身慢。其实不是——真正的瓶颈在于流式传输的架构设计，而不是推理速度。
OpenAI 最近在 HN 上分享了他们如何在全球范围内做低延迟语音 AI 的技术细节。核心思路其实可以直接借鉴到自己的项目里。这篇文章把关键原理拆出来，配上完整可运行的代码，你照着做就能搭一个响应飞快的语音对话 demo。
为什么语音 AI 感觉"卡"
普通的 API 调用流程是这样的：
用户说话 → 录音完整 → 发送 → 等待全部响应 → 播放
这个流程里有三个等待点，每个都在累积延迟。OpenAI 的做法是把每个等待点都改成流式：
用户说话 → 边录边发 → 边收边播
具体来说有几个关键技术点：
1. VAD（语音活动检测）：不等用户说完，实时判断"说话结束了没有"
2. 流式 STT：音频边录边转文字，不等完整录音
3. 流式 LLM：文字边生成边发出，不等完整回复
4. 流式 TTS：文字边收边合成语音，不等完整文本
四个环节全部流式，延迟从"四段串行等待"变成"几乎感知不到"。
动手实现：一个完整的流式语音对话 demo
下面这个 demo 用 Python 实现核心流程：麦克风录音 → 流式调用 LLM → 流式 TTS 播放。
依赖安装：
pip install openai pyaudio numpy webrtcvad
第一步：VAD 检测说话结束
import pyaudio
import webrtcvad
import numpy as np
import collections
def record_with_vad(sample_rate=16000, frame_duration_ms=30, silence_threshold=1.5):
"""
使用 VAD 检测说话结束，避免固定时长录音的尴尬停顿
silence_threshold: 静音多少秒后认为说话结束
"""
vad = webrtcvad.Vad(2) # 灵敏度 0-3，2 是平衡值
pa = pyaudio.PyAudio()
frame_size = int(sample_rate * frame_duration_ms / 1000)
stream = pa.open(
format=pyaudio.paInt16,
channels=1,
rate=sample_rate,
input=True,
frames_per_buffer=frame_size
)
print("🎤 开始录音，说话后自动检测结束...")
frames = []
silence_frames = 0
silence_limit = int(silence_threshold * 1000 / frame_duration_ms)
speaking = False
while True:
frame = stream.read(frame_size, exception_on_overflow=False)
is_speech = vad.is_speech(frame, sample_rate)
if is_speech:
speaking = True
silence_frames = 0
frames.append(frame)
elif speaking:
silence_frames += 1
frames.append(frame)
if silence_frames > silence_limit:
print("✅ 检测到说话结束")
break
stream.stop_stream()
stream.close()
pa.terminate()
audio_data = b''.join(frames)
return audio_data
第二步：流式调用 LLM 并同步 TTS
from openai import OpenAI
import threading
import queue
import io
# 用无量 API，国内直连，不需要代理
client = OpenAI(
api_key="your_api_key_here",
base_url="https://api2everything.xyz/v1"
)
def stream_llm_response(user_text: str, text_queue: queue.Queue):
"""
流式获取 LLM 回复，把文本块放入队列供 TTS 消费
"""
print(f"💬 用户: {user_text}")
print("🤖 AI: ", end="", flush=True)
stream = client.chat.completions.create(
model="gpt-4o-mini", # 也可以换 claude-3-5-haiku，更便宜
messages=[
{"role": "system", "content": "你是一个语音助手，回答简洁，适合朗读。"},
{"role": "user", "content": user_text}
],
stream=True,
max_tokens=300
)
buffer = ""
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
buffer += delta
# 遇到句子结束符就发给 TTS，不等全部生成完
if any(p in buffer for p in ["。", "！", "？", ".", "!", "?"]):
text_queue.put(buffer.strip())
buffer = ""
# 把剩余文本也发出去
if buffer.strip():
text_queue.put(buffer.strip())
text_queue.put(None) # 结束信号
print()
def stream_tts_playback(text_queue: queue.Queue):
"""
从队列消费文本，流式合成并播放语音
"""
pa = pyaudio.PyAudio()
while True:
text = text_queue.get()
if text is None:
break
# 调用 TTS 接口
response = client.audio.speech.create(
model="tts-1", # tts-1-hd 质量更好但稍慢
voice="alloy",
input=text,
response_format="pcm" # PCM 格式延迟最低
)
# 直接流式播放，不写文件
stream = pa.open(
format=pyaudio.paInt16,
channels=1,
rate=24000,
output=True
)
for audio_chunk in response.iter_bytes(chunk_size=1024):
stream.write(audio_chunk)
stream.stop_stream()
stream.close()
pa.terminate()
def voice_chat_loop():
"""
主循环：录音 → STT → LLM → TTS，全程流式
"""
while True:
# 1. VAD 录音
audio_data = record_with_vad()
# 2. STT：把录音转文字
audio_file = io.BytesIO(audio_data)
audio_file.name = "audio.wav"
transcript = client.audio.transcriptions.create(
model="whisper-1",
file=audio_file,
language="zh"
)
user_text = transcript.text
if not user_text.strip():
continue
# 3. LLM + TTS 并行：LLM 生成文字的同时 TTS 开始合成
text_queue = queue.Queue()
tts_thread = threading.Thread(
target=stream_tts_playback,
args=(text_queue,)
)
tts_thread.start()
stream_llm_response(user_text, text_queue)
tts_thread.join()
if __name__ == "__main__":
print("🚀 语音对话启动，Ctrl+C 退出")
voice_chat_loop()
几个优化点
模型选择影响延迟：gpt-4o-mini 比 gpt-4o 快很多，语音场景下够用。如果想更便宜，换 deepseek-v3 成本能再降一半，无量 API 上这些模型都有。
TTS 格式选 PCM：比 MP3 少了解码步骤，首字节延迟能低 100-200ms。
句子切割粒度：上面代码按标点切割，实际可以调整——切太细 TTS 合成频繁，切太粗等待时间长，根据场景调。
并发架构：LLM 流式输出和 TTS 合成用独立线程，两者真正并行，不是串行等待。
关于 API 费用
这套流程调用了三个接口：Whisper STT + LLM + TTS，如果直接用 OpenAI 官方，一次完整对话大概 $0.01-0.02。
用 无量 API 的话，同样的模型便宜 60% 以上，国内直连不需要代理，Base URL 改一行就切过去了。注册还送 ¥1 余额，拿来跑这个 demo 完全够。
代码跑起来有问题的，评论区说一下，我看到都会回。觉得有用的话点个赞，后续还会写 WebSocket 版本的实时双工语音流。