首页
/ 基于 ElevenLabs 与 Claude 构建低延迟语音助手:claude-cookbooks 实战指南

基于 ElevenLabs 与 Claude 构建低延迟语音助手:claude-cookbooks 实战指南

2026-09-07 17:16:29作者:蔡怀权

本篇指南以 third_party/ElevenLabs 目录为核心,讲解如何在 Anthropic 的 claude-cookbooks 示例库中,把 ElevenLabs 的语音合成(TTS)与语音识别(STT)能力接入 Claude,逐步搭建一套低延迟的对话式语音助手。你将掌握从环境配置、四种渐进式延迟优化思路,到生产级 WebSocket 流式实现(含无缝音频播放、连续对话)的完整方案,并学会排查麦克风、音频爆音与网络等常见问题。

本目录是 claude-cookbooks 项目 third_party/ 下由 ElevenLabs 提供的第三方集成 cookbook。它的目标是:用 ElevenLabs 处理"说话→文字"与"文字→语音"两个环节,用 Claude 负责"理解意图并生成回答",再通过流式(streaming)技术把整条链路的延迟压到最低。

目录包含哪些内容

third_party/ElevenLabs/ 下共有 5 个文件,形成一条完整的学习与落地链路:

文件 作用
low_latency_stt_claude_tts.ipynb 交互式教程,分步演示语音助手从零到 WebSocket 流式的构建过程,每一步都测量延迟收益
stream_voice_assistant_websocket.py 生产可用的完整脚本:连续麦克风输入、无间隙音频播放、基于 WebSocket 的最低延迟实现
requirements.txt 全部 Python 依赖及版本约束
.env.example API Key 环境变量模板
README.md 使用说明、排障与扩展思路(即本篇解析的对象)

官方建议的使用顺序是:先建环境 → 跑通 notebook 理解优化原理 → 再运行生产脚本体验完整效果。

环境准备四步走

1. 创建虚拟环境

进入 ElevenLabs 目录并创建隔离环境(避免污染全局 Python 环境):

cd /path/to/claude-cookbooks/third_party/ElevenLabs
python -m venv venv

# macOS / Linux
source venv/bin/activate
# Windows
venv\Scripts\activate

2. 获取两把 API Key

语音链路需要两个服务的密钥,缺一不可:

  • ElevenLabs API key:在 ElevenLabs 开发者后台的 API Keys 页面创建。创建时请确保授予以下最小权限集,否则后续调用会失败:
    • Text to speech(文本转语音)
    • Speech to text(语音转文本)
    • Read access on voices(读取音色库)
    • Read access on models(读取模型列表)
  • Anthropic API key:在 Anthropic 控制台的 API Keys 设置页创建,用于调用 Claude 模型生成回答。

从源码看,脚本 stream_voice_assistant_websocket.py 启动时会立即 assert 两个环境变量。缺少任意一个都会抛出形如 AssertionError: ELEVENLABS_API_KEY is not set 的错误并终止运行,因此这两把 Key 是硬性前提。

3. 配置 .env

将模板复制为 .env 并填入真实密钥:

cp .env.example .env

.env 文件内容对应 .env.example 的格式:

# ElevenLabs API Key
# Get your API key from: https://elevenlabs.io/app/developers/api-keys
ELEVENLABS_API_KEY=your_elevenlabs_api_key_here

# Anthropic API Key
# Get your API key from: https://console.anthropic.com/settings/keys
ANTHROPIC_API_KEY=your_anthropic_api_key_here

脚本与 notebook 都通过 dotenvload_dotenv() 自动加载该文件(见 stream_voice_assistant_websocket.py 第 49-53 行),因此无需手动 export。注意检查密钥前后不要混入空格或换行。

4. 安装依赖

在虚拟环境激活状态下执行:

pip install -r requirements.txt

依赖清单与用途(见 requirements.txt):

依赖 版本约束 用途
anthropic >=0.71.0 调用 Claude 模型与流式 API
elevenlabs >=2.15.0 官方 Python SDK,封装 TTS / STT / WebSocket
sounddevice >=0.5.1 底层基于 PortAudio 的麦克风采集与扬声器播放
numpy >=1.26.0 音频采样数组处理
scipy >=1.16.2 将录音写入 WAV 格式
websocket-client >=1.8.0 维护 ElevenLabs TTS WebSocket 连接
python-dotenv >=1.0.0 读取 .env 环境变量
pydub >=0.25.1 解码 MP3 音频流(依赖 FFmpeg)

通过 notebook 理解"逐步压低延迟"的优化路线

low_latency_stt_claude_tts.ipynb 的价值不在于一步到位,而在于把语音助手拆成可测量的环节,让你看清延迟从哪来、被谁吃掉。整条路线分五步推进。

第 1 步:用 TTS 生成"用户输入"音频

教程先调用 ElevenLabs 文本转语音接口生成一段 "Hello, Claude." 的示例语音,模拟用户说话。从 notebook 代码看,这里用的是 eleven_v3 模型,输出格式为 mp3_44100_128

audio = elevenlabs_client.text_to_speech.convert(
    voice_id=VOICE_ID,               # 动态选取的音色 ID
    output_format="mp3_44100_128",   # 44.1kHz、128kbps 的 MP3
    model_id="eleven_v3",
    text="Hello, Claude. ",
)
audio_data = io.BytesIO()
for chunk in audio:
    audio_data.write(chunk)

第 2 步:STT 语音识别并测量耗时

把上一步的音频交给 ElevenLabs 的 scribe_v1 语音识别模型转成文字,同时用 time.time() 记录转录耗时。这一步确立了全流程的第一个延迟来源——识别不是瞬时的

transcription = elevenlabs_client.speech_to_text.convert(
    file=audio_data, model_id="scribe_v1"
)

第 3 步:非流式调用 Claude 作为"延迟基准"

把转录文本发给 Claude(本 cookbook 统一使用 claude-haiku-4-5,兼顾速度与质量,temperature=0 保证输出稳定),用非流式 messages.create 等待完整回复,并记录总耗时 non_streaming_response_time。这一步是所有后续优化的对照基线。

第 4 步:Claude 流式输出,砍掉"首 token 等待"

同样的问题改用 messages.stream 上下文管理器,逐块读取 text_stream。记录从发起请求到收到第一个 token的时间:

with anthropic_client.messages.stream(...) as stream:
    for text in stream.text_stream:
        if first_token_time is None:
            first_token_time = time.time()

notebook 会打印首 token 时间,并对比非流式总响应时间,计算"感知延迟降低了百分之多少"。这里揭示了一个核心观点:用户等的是"第一句话",而不是"整段写完",因此首 token 延迟才是感知延迟的关键

第 5 步:TTS 也流式化,音频提前开播

文字流式化后,TTS 端同样存在"先生成完整音频再播放"的静默期。notebook 改用 elevenlabs_client.text_to_speech.stream(模型换为专为低延迟设计的 eleven_turbo_v2_5),边生成边接收音频分块,记录首个音频块到达时间 streaming_tts_time_to_first_chunk,用实际数据证明流式 TTS 显著减少了从"开始合成"到"听到声音"之间的空档。

进阶尝试:按句切分直送 TTS,及其代价

既然两端都能流式,自然想到把它们串起来:用正则 re.compile(r"[.!?]+") 检测 Claude 输出中的句子边界,每凑齐一个完整句子就立即送去 TTS 合成(余下不完整片段留在 sentence_buffer 中,结束后再处理)。notebook 代码展示了完整的句级拼接与补尾逻辑。

但这种做法有一个质量缺陷——音频"断片感":每个句子被独立合成,TTS 模型看不到后续内容,导致句与句之间的韵律(节奏、重音、语调)不连贯,听起来生硬、不自然。这就引出了最终答案:ElevenLabs 的 WebSocket API。

WebSocket 流式:延迟与音质兼得

WebSocket 方案解决了句级拼接的所有问题,notebook 与生产脚本一致认为它是"两全其美":

  • 接受流式文本输入,无需在客户端缓存成完整句子;
  • 跨文本块维持上下文,韵律自然连贯;
  • 音频块就绪即回传,无需等待整段生成完毕;
  • 最低延迟的前提下保持最高音质

交互流程(与 README.md 的描述一致,按下回车开始/结束录音):

  1. 按下 Enter 开始录音;
  2. 对着麦克风提问;
  3. 再次按 Enter 结束录音;
  4. 助手用自然语音回答;
  5. 可反复对话,按 Ctrl+C 退出。
python stream_voice_assistant_websocket.py

深入生产脚本的实现原理

stream_voice_assistant_websocket.py 把 notebook 的所有优化落地为可持续运行的对话循环。它的设计目标包括:实时麦克风录音、带上下文记忆的连续对话、WebSocket 最小延迟、自定义音频队列实现无缝隙播放。

常量与全局配置

脚本开头定义了若干关键常量(见脚本 62-79 行):

  • SAMPLE_RATE = 44100CHANNELS = 1:录音采用 44.1kHz 单声道;
  • TTS_MODEL_ID = "eleven_turbo_v2_5":专为低延迟设计的 TTS 模型;
  • TTS_OUTPUT_FORMAT = "mp3_44100_128":MP3 格式,兼容 ElevenLabs 免费套餐
  • 客户端初始化时显式指定 base_url="https://api.elevenlabs.io"

启动时会调用 elevenlabs_client.voices.search() 自动选取第一个可用音色并打印名称与 ID,方便你确认当前用的是哪个声音。

AudioQueue:回调驱动的无缝播放器

这是整个脚本最核心的自定义类,负责把异步到达的音频块连续播出来。它维护一个受线程锁保护的字节缓冲(bytearray),配合 sounddeviceOutputStream 回调式播放:

  • PRE_BUFFER_SIZE = 8192:缓冲至少攒够该字节数才开播,用预缓冲消除开播瞬间的爆音;
  • BUFFER_CLEANUP_THRESHOLD = 100000:读指针超过该阈值即压缩缓冲区,防止长时间对话导致内存无限增长;
  • REMAINING_BYTES_THRESHOLD = 1000:剩余未播字节低于该值即视为播放结束;
  • 播放线程以 blocksize=2048 采样块填充输出缓冲,数据不足时补静音(outdata[:] = 0)保持流不中断。

add() 方法用 pydub 把收到的 MP3 块解码为 PCM,再归一化为 float32 数组(除以 32768)写入缓冲;解码失败的分块会被静默跳过(见下节 Troubleshooting)。

record_audio:回车控制的实时录音

record_audio() 用两个 input() 分别充当"开始"与"停止"开关:第一次回车后打开 sd.InputStream,回调把每个音频帧追加进列表;第二次回车后停止并关闭流,把所有帧 np.concatenate 成整段音频,乘 32767 转回 int16 后用 scipy.io.wavfile 写入内存中的 WAV 缓冲返回。

stream_claude_and_synthesize_ws:Claude 直连 TTS 流

核心流程可拆成三段:

  1. 建立 WebSocket:向 wss://api.elevenlabs.io/v1/text-to-speech/{VOICE_ID}/stream-input 发起连接,查询参数带上 model_idoutput_formaton_open 时先发送一条控制消息初始化会话,其中携带 voice_settingsstability: 0.5similarity_boost: 0.8)与 xi_api_key
  2. 边收边发:在独立守护线程中运行 ws.run_forever(),同时用 anthropic_client.messages.stream 流式读取 Claude 回复。每收到一个文本块,除了打印与累计到 response_text,立即 ws.send(json.dumps({"text": text, "try_trigger_generation": True})) 不做任何句子缓冲直接推给 TTS;
  3. 收尾判定:文本流结束后发送空文本块({"text": ""})通知 TTS 处理剩余内容,随后轮询等待服务端返回带 isFinal: true 的消息,确认合成完毕再关闭连接。

值得注意的两处细节:

  • System Prompt 约束:脚本明确告诉 Claude"回答将被 ElevenLabs 转为语音,不要写 Markdown,因为无法被朗读"。这是语音场景极易被忽略的工程点;
  • 首音频计时AudioQueue 首次播放时记录 first_audio_time,主循环用它减去录音结束时刻打印"Time to first audio",让你在每个轮次都能量化端到端延迟。

main 主循环:带记忆的连续对话

main() 维护 conversation_history 列表:每轮把转录文本追加为 user 消息,把完整回答追加为 assistant 消息后再进入下一轮。这样 Claude 能引用前几轮上下文,实现真正"连续"的对话体验。整套流程被 try/except KeyboardInterrupt 包裹,随时可干净退出。

Troubleshooting 常见问题排查

播放出现爆音 / 咔哒声

现象:播放中偶尔出现短暂爆音、咔哒声或音频丢失。

原因:脚本使用 MP3 格式(免费套餐的硬性要求)。实时分块流式传输 MP3 时,FFmpeg 偶尔会收到无法解码的不完整帧,典型出现在:流开始阶段(首块过小)、短暂网络延迟期间、音频生成收尾阶段(末块不完整)。脚本解码逻辑中的 try-except 会静默跳过失败块以阻止报错,但会造成短暂音频空洞,表现为爆音或咔哒声。

影响:播放整体正常;爆音通常轻微或几乎不可察觉;WebSocket 连接稳定,功能不受损。

解决:这是免费套餐下 MP3 格式的预期行为。若想彻底消除爆音:升级到 ElevenLabs 付费套餐,或将输出格式由 mp3_44100_128 改为 pcm_44100——PCM 无解码问题,流式播放更干净。

API Key 相关报错

现象:抛出 AssertionError: ELEVENLABS_API_KEY is not setANTHROPIC_API_KEY is not set

处理:先确认已执行 cp .env.example .env;再编辑 .env 确保两个 Key 都填对;检查密钥是否有拼写错误或多余空格;最后确认 ElevenLabs Key 具备上文列出的最小权限集。

依赖缺失 / 无法出声

现象ImportError: PortAudio library not found 或播放失败。

处理sounddevice 依赖系统级 PortAudio 库,pydub 解码 MP3 依赖 FFmpeg,需按平台安装:

  • macOSbrew install portaudio ffmpeg
  • Ubuntu/Debiansudo apt-get install portaudio19-dev ffmpeg
  • Windows:从 ffmpeg.org 下载 FFmpeg 并加入系统 PATH(PortAudio 通常随 sounddevice 自动安装)

随后重新执行 pip install -r requirements.txt

麦克风无权限

现象OSError: [Errno -9999] Unanticipated host error 或麦克风不可用。

处理

  • macOS:系统设置 → 隐私与安全性 → 麦克风,为终端(或所用 Python IDE)开启麦克风权限;
  • Windows:设置 → 隐私 → 麦克风,为 Python/终端开启麦克风访问;
  • Linux:确认当前用户属于 audio 组:sudo usermod -a -G audio $USER(随后注销重登)。

可用下面的命令验证麦克风是否被系统正确识别:

python -c "import sounddevice as sd; print(sd.query_devices())"

WebSocket 连接失败

现象:连接错误、超时或流中断。

处理:检查网络连接是否稳定;确认防火墙没有拦截 WebSocket 的 443 端口;临时关闭 VPN 或代理再试;确认没有超出 API 速率限制(可在 ElevenLabs 控制台用量页查看)。若持续异常,可关注 ElevenLabs 服务状态页是否有故障通告。

基于语音助手的扩展项目灵感

在跑通基础语音助手后,官方文档给出了六个方向,帮助你评估这套架构的复用边界:

  • 会议纪要助手(Meeting Note-Taker):实时录音转写会议,由 Claude 提炼摘要、行动项与关键结论;
  • 语言学习陪练(Language Learning Tutor):多语言口语对话与实时反馈,Claude 可纠正发音、优化措辞并按水平调节难度;
  • 互动故事机(Interactive Storyteller):Claude 旁白的"选择你自己的冒险"游戏,每个角色使用不同音色;
  • 免手编码助手(Hands-Free Coding Assistant):语音描述代码改动、Bug 或新功能,双手不离键盘,适合结对调试与自言自语式排错;
  • 语音智能家居(Voice-Activated Smart Home):用自然对话控制家居设备,支持"现在开暖气会不会太冷"这类复合语义问题;
  • 语音日记(Personal Voice Journal):口述日记由 Claude 按主题归档、跟踪情绪,并在需要时调出相关过往记录。

这些项目的共同点在于:输入侧复用"STT + 录音",推理侧替换 Claude 的 system prompt 与调用逻辑,输出侧自由切换音色与模型——恰好印证了本 cookbook 分层解耦的架构价值。

小结

ElevenLabs × Claude 的这套低延迟语音助手方案,本质上是一条"感知延迟驱动"的优化路径:先用非流式调用建立基线,再依次对 Claude 输出、TTS 输出做流式化,最终用 WebSocket 把两条流对接,让文本块到达即转语音、音频块到达即播放,同时借助 ElevenLabs 的跨块上下文保住自然韵律。配合 notebook 的逐级实验与 生产脚本 的完整工程细节,这套模板稍作改造即可支撑上述多种语音交互产品。

登录后查看全文
热门项目推荐
相关项目推荐

项目优选

收起
kernelkernel
deepin linux kernel
C
33
18
ops-transformerops-transformer
本项目是CANN提供的transformer类大模型算子库,实现网络在NPU上加速计算。
C++
1.13 K
2.75 K
pytorchpytorch
作为 Ascend for PyTorch 社区的核心组件,TorchNPU 是昇腾专为 PyTorch 打造的深度学习适配插件,使 PyTorch 框架能够直接调用昇腾 NPU,为开发者提供昇腾 AI 处理器的超强算力。
Python
857
1.35 K
docsdocs
暂无描述
Markdown
897
5.8 K
kernelkernel
openEuler内核是openEuler操作系统的核心,既是系统性能与稳定性的基石,也是连接处理器、设备与服务的桥梁。
C
529
593
ops-nnops-nn
本项目是CANN提供的神经网络类计算算子库,实现网络在NPU上加速计算。
C++
915
1.83 K
jiuwenswarmjiuwenswarm
JiuwenSwarm 是一款基于openJiuwen开发的智能AI Agent,它能够将大语言模型的强大能力,通过你日常使用的各类通讯应用,直接延伸至你的指尖。
Python
3.58 K
1.01 K
ops-mathops-math
本项目是CANN提供的数学类基础计算算子库,实现网络在NPU上加速计算。
C++
1.35 K
1.46 K
cann-learning-hubcann-learning-hub
CANN 学习中心仓,支持在线互动运行、边学边练,提供教程、示例与优化方案,一站式助力昇腾开发者快速上手。
Jupyter Notebook
1.01 K
515
AscendNPU-IRAscendNPU-IR
AscendNPU-IR是基于MLIR(Multi-Level Intermediate Representation)构建的,面向昇腾亲和算子编译时使用的中间表示,提供昇腾完备表达能力,通过编译优化提升昇腾AI处理器计算效率,支持通过生态框架使能昇腾AI处理器与深度调优
C++
547
388