使用ElevenLabs和Claude构建低延迟语音助手

本教程面向有编程基础的开发者，深入讲解如何结合ElevenLabs的语音转文字（STT）和文字转语音（TTS）服务与Anthropic的Claude大语言模型，构建低延迟的语音助手。文章从环境配置、API密钥管理开始，逐步演示了非流式与流式API的延迟对比，并创新性地提出逐句合成与WebSocket流式传输两种优化方案，最终给出一个生产级语音助手的完整实现。教程不仅涵盖代码实现，还深入分析了延迟来源、架构设计权衡及生产环境中的关键挑战，帮助开发者掌握实时语音AI系统的工程实践。

技术背景

随着大语言模型（LLM）能力的提升，语音交互成为AI应用的重要入口。然而，传统语音助手通常采用“录音-转写-生成-合成-播放”的串行流程，导致端到端延迟高，用户体验差。本教程旨在解决这一痛点，通过流式处理和并行化技术，将感知延迟降至最低。适用场景包括实时对话系统、语音客服、智能助手等。主要限制包括：依赖第三方API的稳定性和成本，以及网络延迟对实时性的影响。

核心原理

核心原理是流水线并行与流式处理。传统流程中，STT、LLM、TTS三个环节串行执行，每个环节需等待完整输出。通过流式API，LLM可以在生成第一个token时就开始传输，TTS也可以在收到部分文本时就开始合成，从而重叠各阶段的时间。此外，WebSocket协议支持双向实时通信，允许文本块持续流入TTS引擎，保持上下文连贯性，实现自然韵律。关键设计决策包括：选择低延迟模型（如claude-haiku-4-5和eleven_turbo_v2_5）、动态选择声音、以及使用句子边界检测来平衡延迟与质量。

架构分析

系统架构分为四层：输入层（麦克风/音频文件）、STT层（ElevenLabs Scribe）、LLM层（Claude）、TTS层（ElevenLabs）。数据流为：音频→STT→文本→LLM→文本流→TTS→音频流→播放。关键组件包括：客户端初始化（ElevenLabs和Anthropic）、模型与声音选择、流式响应处理、音频缓冲与播放。设计思想强调解耦与异步：各服务通过API独立运行，通过流式接口实现数据管道。优化点包括：使用流式API减少首token延迟，使用WebSocket保持TTS上下文，以及使用音频队列实现无缝播放。

完整实战教程

使用ElevenLabs和Claude构建低延迟语音助手

本笔记本演示如何使用ElevenLabs进行语音转文字和文字转语音，结合Claude生成智能响应，构建低延迟语音助手。我们将通过流式响应来衡量性能提升，以最小化延迟。

在本笔记本中，我们将演示如何：

使用ElevenLabs TTS将文本转换为语音
使用ElevenLabs语音转文字转录音频
使用Claude生成响应
使用Claude的流式API优化延迟

安装

首先，安装所需的依赖：

%pip install --upgrade pip

%pip install -r requirements.txt

导入

导入 ElevenLabs 集成、Claude API 访问和音频播放所需的库：

import io
import os
import time

import anthropic
import elevenlabs
from dotenv import load_dotenv
from IPython.display import Audio

API 密钥

为 ElevenLabs 和 Anthropic 设置你的 API 密钥。

设置说明：

将此目录中的 .env.example 复制为 .env
编辑 .env 并添加你的实际 API 密钥：

获取你的 ElevenLabs API 密钥：https://elevenlabs.io/app/developers/api-keys
获取你的 Anthropic API 密钥：https://console.anthropic.com/settings/keys

密钥将从 .env 文件中自动加载。

# Load environment variables from .env file

load_dotenv()

ELEVENLABS_API_KEY = os.getenv("ELEVENLABS_API_KEY")
ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")

初始化客户端

为 ElevenLabs 和 Anthropic 服务创建客户端实例：

assert ELEVENLABS_API_KEY is not None, (
    "ERROR: ELEVENLABS_API_KEY not found. Please copy .env.example to .env and add your API keys."
)
assert ANTHROPIC_API_KEY is not None, (
    "ERROR: ANTHROPIC_API_KEY not found. Please copy .env.example to .env and add your API keys."
)

elevenlabs_client = elevenlabs.ElevenLabs(
    api_key=ELEVENLABS_API_KEY, base_url="https://api.elevenlabs.io"
)

anthropic_client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)

列出可用的模型和声音

探索可用的 ElevenLabs 模型和声音。我们将自动选择第一个可用的声音作为助手的回复声音：

print("Available Models and Voices:\n")
for model in elevenlabs_client.models.list():
    print(f"{model.name}: {model.model_id}")

print()

voices = elevenlabs_client.voices.search().voices
for voice in voices:
    print(f"{voice.name}: {voice.voice_id}")

# Select the first voice for assistant responses
selected_voice = voices[0]
VOICE_ID = selected_voice.voice_id

print(f"\nSelected voice: {selected_voice.name} with ID: {VOICE_ID}")

生成输入音频

使用 ElevenLabs 文本转语音功能创建一个示例音频文件。这将模拟我们语音助手的用户输入：

audio = elevenlabs_client.text_to_speech.convert(
    voice_id=VOICE_ID,  # Use the dynamically selected voice
    output_format="mp3_44100_128",
    model_id="eleven_v3",
    text="Hello, Claude. ",
)

audio_data = io.BytesIO()
for chunk in audio:
    audio_data.write(chunk)

Audio(audio_data.getvalue())

语音转写

使用 ElevenLabs 的语音转文本模型对音频输入进行转写。我们将测量转写延迟：

audio_data.seek(0)

start_time = time.time()

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

end_time = time.time()
transcription_time = end_time - start_time

print(f"Transcribed text: {transcription.text}")
print(f"Transcription time: {transcription_time:.2f} seconds")

从 Claude 获取响应

将转录的文本发送给 Claude，并测量响应时间。我们使用 claude-haiku-4-5 来获得快速、高质量的响应：

start_time = time.time()

message = anthropic_client.messages.create(
    model="claude-haiku-4-5",
    max_tokens=1000,
    temperature=0,
    messages=[{"role": "user", "content": transcription.text}],
)

end_time = time.time()
non_streaming_response_time = end_time - start_time

print(message.content[0].text)
print(f"\nResponse time: {non_streaming_response_time:.2f} seconds")

使用流式传输进行优化

通过使用 Claude 的流式 API 来改善响应延迟。这使我们能够更快地接收到第一个 token，从而显著降低感知延迟：

start_time = time.time()
first_token_time = None

claude_full_response = ""

with anthropic_client.messages.stream(
    model="claude-haiku-4-5",
    max_tokens=1000,
    temperature=0,
    messages=[{"role": "user", "content": transcription.text}],
) as stream:
    for text in stream.text_stream:
        claude_full_response += text
        print(text, end="", flush=True)
        if first_token_time is None:
            first_token_time = time.time()

streaming_time_to_first_token = first_token_time - start_time
print(
    f"\n\nStreaming time to first token: {streaming_time_to_first_token:.2f} seconds - reducing perceived latency by {(non_streaming_response_time - streaming_time_to_first_token) * 100 / non_streaming_response_time:.2f}%"
)

文本转语音。与上述类似，我们可以流式传输响应以减少静默时间。

start_time = time.time()
first_audio_chunk_time = None

audio_buffer = io.BytesIO()

audio_generator = elevenlabs_client.text_to_speech.stream(
    voice_id=VOICE_ID,
    output_format="mp3_44100_128",
    text=claude_full_response,
    model_id="eleven_turbo_v2_5",
)

for chunk in audio_generator:
    if first_audio_chunk_time is None:
        first_audio_chunk_time = time.time()
    audio_buffer.write(chunk)

streaming_tts_time_to_first_chunk = first_audio_chunk_time - start_time
print(f"Streaming TTS time to first audio chunk: {streaming_tts_time_to_first_chunk:.2f} seconds")

Audio(audio_buffer.getvalue())

将 Claude 流式输出直接转换为语音（逐句合成）

我们已经分别优化了 Claude 的流式输出和 TTS，但能否将它们结合起来呢？让我们在获得完整句子后，立即将 Claude 的响应流式传输并合成音频。

这种方法会检测句子边界（使用标点符号，如句号、感叹号、问号），并立即将每个句子发送到 TTS，从而进一步降低延迟。

import re

sentence_pattern = re.compile(r"[.!?]+")
sentence_buffer = ""
audio_chunks = []

start_time = time.time()
first_audio_time = None

with anthropic_client.messages.stream(
    model="claude-haiku-4-5",
    max_tokens=1000,
    temperature=0,
    messages=[{"role": "user", "content": transcription.text}],
) as stream:
    for text in stream.text_stream:
        print(text, end="", flush=True)
        sentence_buffer += text

        if sentence_pattern.search(sentence_buffer):
            sentences = sentence_pattern.split(sentence_buffer)

            # Process all complete sentences (all but the last element)
            for i in range(len(sentences) - 1):
                complete_sentence = sentences[i].strip()
                if complete_sentence:
                    audio_gen = elevenlabs_client.text_to_speech.stream(
                        voice_id=VOICE_ID,
                        output_format="mp3_44100_128",  # Free tier format
                        text=complete_sentence,
                        model_id="eleven_turbo_v2_5",
                    )

                    sentence_audio = io.BytesIO()
                    for chunk in audio_gen:
                        if first_audio_time is None:
                            first_audio_time = time.time()
                        sentence_audio.write(chunk)

                    audio_chunks.append(sentence_audio.getvalue())

            sentence_buffer = sentences[-1]

if sentence_buffer.strip():
    audio_gen = elevenlabs_client.text_to_speech.stream(
        voice_id=VOICE_ID,
        output_format="mp3_44100_128",
        text=sentence_buffer.strip(),
        model_id="eleven_turbo_v2_5",
    )
    sentence_audio = io.BytesIO()
    for chunk in audio_gen:
        sentence_audio.write(chunk)
    audio_chunks.append(sentence_audio.getvalue())

sentence_streaming_time_to_first_audio = first_audio_time - start_time
print(f"\n\nTime to first audio: {sentence_streaming_time_to_first_audio:.2f} seconds")

combined_pcm = b"".join(audio_chunks)
Audio(combined_pcm)

问题：音频不连贯

虽然这种方法实现了极低的延迟，但存在一个质量问题。每个句子都是独立合成的，导致音频听起来不连贯且不自然。韵律（节奏、重音、语调）在句子之间无法流畅衔接。

这是因为TTS模型没有关于后续内容的上下文，因此每个句子都被当作独立的语音单元来处理。

WebSocket 流式传输：两全其美

ElevenLabs 提供了 WebSocket API，完美解决了这个问题。我们无需等待完整句子，而是可以将从 Claude 接收到的文本块直接流式传输到 TTS 引擎。

WebSocket API：

接受流式文本输入（无需句子缓冲）
跨块保持上下文，实现自然韵律
音频块一旦就绪即可返回
以最低延迟实现最佳音频质量

让我们实现这一终极优化：

构建生产级语音助手

本笔记本中演示的技术为构建实时语音助手奠定了基础。WebSocket 流式传输方法在保持自然音频质量的同时，最大限度地减少了延迟。

关键实现挑战

在构建生产系统时，除了基本的流式传输之外，您还需要解决几个额外的挑战：

连续音频播放：音频块必须无缝播放，不能有间隙或爆音。这需要仔细的缓冲区管理和预缓冲策略。

麦克风输入：从麦克风实时录音，并正确处理音频格式和采样率。

对话状态：跨轮次维护对话历史，以便 Claude 可以引用之前的上下文。

音频质量：在不同音频格式（PCM、WAV）之间转换，并避免编码产生的伪影。

完整实现

我们构建了一个完整的语音助手脚本，演示了所有这些技术：

stream_voice_assistant_websocket.py - 一个生产就绪的对话式语音助手，具有以下特点：

麦克风录音，支持按 Enter 停止控制
ElevenLabs 语音转文字转录
Claude 流式传输，带对话历史
基于 WebSocket 的 TTS，延迟极低
自定义音频队列，实现无缝播放
连续对话循环

运行该脚本，体验一个功能齐全的语音助手：

python stream_voice_assistant_websocket.py

这展示了本笔记本中的流式优化如何转化为具有生产级音频处理的真实应用。

常见问题

常见错误：1) API密钥未配置或无效，导致认证失败；2) 音频格式不匹配，导致播放异常；3) 网络不稳定导致流中断；4) 句子缓冲逻辑错误，导致音频不连贯。排查方法：检查.env文件和环境变量，确认API密钥有效；使用日志记录各阶段耗时；测试不同音频格式（如mp3_44100_128）；确保网络连接稳定。生产优化：使用连接池和重试机制；实现音频预缓冲和动态调整；采用多线程处理音频播放；监控API调用成本并优化模型选择。

企业应用扩展

本教程展示了基础实现，可扩展方向包括：1) 集成更多LLM（如GPT-4）或TTS服务；2) 添加语音活动检测（VAD）以自动判断说话结束；3) 实现多轮对话状态管理；4) 优化音频编码（如使用Opus）以减少带宽；5) 部署为微服务，使用消息队列（如Kafka）解耦组件；6) 添加用户个性化声音克隆；7) 实现流式STT（如ElevenLabs的流式转写）以进一步降低延迟。

总结

内容来源与验证

内容等级：PREMIUM_A
质量等级：A
难度：中级
验证等级：V1
原文作者：Adriaan-ANT
原文日期：2025-11-24
来源许可：MIT
来源版本：f0d859da3ddcecf8e65350f8ec4c8190b4057b1d
查看原文

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员