ElevenLabs：低延迟语音

实时语音转文字 + Claude + 文字转语音

🌐 查看英文原文 | 源码 Notebook

使用 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 Keys

为 ElevenLabs 和 Anthropic 设置你的 API keys。

设置说明：

将 .env.example 复制为 .env 到此目录
编辑 .env 并添加你的实际 API keys：
- 获取 ElevenLabs API key：https://elevenlabs.io/app/developers/api-keys
- 获取 Anthropic API key：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}")

Available Models and Voices:

Eleven v3 (alpha): eleven_v3
Eleven Multilingual v2: eleven_multilingual_v2
Eleven Flash v2.5: eleven_flash_v2_5
Eleven Turbo v2.5: eleven_turbo_v2_5
Eleven Turbo v2: eleven_turbo_v2
Eleven Flash v2: eleven_flash_v2
Eleven Multilingual v1: eleven_multilingual_v1
Eleven English v1: eleven_monolingual_v1
Eleven English v2: eleven_english_sts_v2
Eleven Multilingual v2: eleven_multilingual_sts_v2

Rachel: 21m00Tcm4TlvDq8ikWAM
Drew: 29vD33N1CtxCmqQRPOHJ
Clyde: 2EiwWnXFnvU5JabPnv8n
Paul: 5Q0t7uMcjvnagumLfvZi
Aria: 9BWtsMINqrJLrRacOk9x
Domi: AZnzlk1XvdvUeBnXmlld
Dave: CYw3kZ02Hs0563khs1Fj
Roger: CwhRBWXzGAHq8TQ4Fs17
Fin: D38z5RcWu1voky8WS1ja
Sarah: EXAVITQu4vr4xnSDxMaL

Selected voice: Rachel with ID: 21m00Tcm4TlvDq8ikWAM

生成输入音频

使用 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())

<IPython.lib.display.Audio object>

语音转写

使用 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")

Transcribed text: Hello, Claude.
Transcription time: 0.54 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")

Hello! It's nice to meet you. How can I help you today?

Response time: 1.03 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}%"
)

Hello! It's nice to meet you. How can I help you today?

Streaming time to first token: 0.71 seconds - reducing perceived latency by 30.71%

文字转语音。与上述类似，我们可以流式传输回复以减少静音时间。

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())

Streaming TTS time to first audio chunk: 0.39 seconds

<IPython.lib.display.Audio object>

将 Claude 流式输出直接传给 TTS（逐句传输）

我们已经分别优化了 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)

Hello! It's nice to meet you. How can I help you today?

Time to first audio: 1.48 seconds

<IPython.lib.display.Audio object>

问题：断连的音频

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

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

WebSocket 流式传输：两全其美

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

WebSocket API： - 接受流式文本输入（无需句子缓冲） - 在各块之间保持上下文以实现自然的韵律 - 在音频块准备好后立即返回 - 在保持最佳音频质量的同时实现最低延迟

让我们来实现这个终极优化：

构建生产级语音助手

本教程中演示的技术为构建实时语音助手奠定了基础。WebSocket 流式传输方法在保持自然音频质量的同时最小化了延迟。

关键实现挑战

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

连续音频播放：音频块必须无缝播放，不能有间隙或爆音。这需要仔细的缓冲区管理和预缓冲策略。
麦克风输入：从麦克风实时录音，需要正确处理音频格式和采样率。
对话状态：在多轮对话中维护对话历史，以便 Claude 可以引用之前的上下文。
音频质量：在不同音频格式（PCM、WAV）之间转换，并避免编码产生的伪影。

完整实现

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

stream_voice_assistant_websocket.py——一个生产级的对话式语音助手，具有以下功能： - 麦克风录音，支持回车键停止控制 - ElevenLabs 语音转文字转写 - Claude 流式传输，带有对话历史 - 基于 WebSocket 的 TTS，延迟极低 - 自定义音频队列实现无缝播放 - 持续对话循环

运行脚本以体验功能完整的语音助手：

python stream_voice_assistant_websocket.py

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

Tip

中文读者提示：本章节的代码和输出为英文原文。如需理解具体实现细节，可参考上方中文导读和代码注释。如有疑问，欢迎在 GitHub Issues 讨论。