重建ai语音助手：使用langgraph和whisper的5个关键步骤提升定制化能力

在探索了ElevenLabs的AI agent功能并亲身体验了我如何快速构建一个功能齐全的voicebot而无需编写大量代码后，我不禁想——如果我从头开始构建它会怎么样？

在我之前的文章中，我分享了如何结合ElevenLabs和Make.com创建一个能够轻松安排会议和处理客户互动的voicebot。no-code方法带来了令人印象深刻的结果，但作为一个代码爱好者，我希望获得更多的控制和定制。

这一次，我决定走不同的路线——从头开始重建AI voicebot，使用LangGraph、ElevenLabs的API和Whisper。我的目标是更好地理解其内部工作原理，微调其性能，并探索现成解决方案无法提供的新可能性。在本文中，我不仅会带你深入了解构建我自己版本的ElevenLabs AI agent的幕后，还会为你提供一个清晰的框架，以便你自己重建一个voicebot。在这个过程中，我会分享我面临的挑战、所做的决定，以及定制编码解决方案的灵活性如何解锁无尽的可能性。

为了实现这个项目，我依赖于三个构成AI voicebot基础的关键组件：

构建模块

语音转文本：Whisper

Whisper，由OpenAI开发，是一种先进的语音转文本模型，以其准确性和多语言能力而闻名。它高效地将口语转换为文本，使其成为处理高精度语音输入的理想解决方案。其深度学习能力使其能够理解各种口音和方言，确保在不同人群中提供流畅的用户体验。在这个项目中，我利用OpenAI的Python SDK处理音频输入：

openai_client.audio.transcriptions.create(
   model="whisper-1", 
   file=audio_bytes,
)

对话式AI代理：LangGraph

LangGraph，由LangChain开发，作为AI voicebot的核心，能够创建结构化的、互动的对话，具备工具调用和记忆功能。它允许管理复杂的对话流程和决策过程，确保机器人能够智能地响应各种用户输入。借助LangGraph，我能够设计出高度动态和适应性强的对话体验，能够记住之前的互动，并有效利用各种工具来增强功能。

文本转语音：ElevenLabs API

ElevenLabs API 驱动文本转语音组件，提供自然且逼真的语音响应。该组件通过提供类人语音输出增强用户互动，使对话更加引人入胜和直观。在这个项目中，我利用了 ElevenLabs 的 Python SDK 使用 eleven_turbo_v2_5 模型将文本转换为语音：

elevenlabs_client.text_to_speech.convert(
    voice_id="YUdpWWny7k5yb4QCeweX",
    output_format="mp3_22050_32",
    text=cleaned_text,
    model_id="eleven_turbo_v2_5", 
    voice_settings=VoiceSettings(
        stability=0.0,
        similarity_boost=1.0,
        style=0.0,
        use_speaker_boost=True,
    ),
)

通过这些组件，我能够从零开始创建一个高度响应和智能的 voicebot。

构建AI语音助手的逐步指南

现在我们已经覆盖了基本构建模块，让我们深入实现部分。我们将首先创建一个使用LangGraph的基本对话式聊天机器人，具有记忆功能和一个获取当前日期和时间的工具。一旦聊天机器人正常运行，我们将通过添加两个额外的节点来增强它——一个用于使用Whisper将输入音频处理为文本，另一个用于使用ElevenLabs API将聊天机器人的响应转换为音频。

这种逐步的方法将提供一个坚实的基础，同时逐步构建语音助手的功能。

步骤 1：构建基本的对话式聊天机器人

首先，我们将创建一个简单的聊天机器人，它可以记住之前的消息并获取当前的日期和时间。这将涉及定义一个状态管理系统，集成一个工具以从API获取日期和时间，并使用LangGraph构建对话流程。

使用LangGraph构建的基本对话聊天机器人

定义状态管理系统

我们首先定义聊天机器人的状态，以跟踪对话消息。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

State类使用注解列表来存储消息，并在新消息到达时将其附加。

实现记忆功能

接下来，我们引入记忆功能，以存储和检索之前的消息，确保对话的连续性。

from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

MemorySaver组件帮助在交互之间持久化聊天机器人的状态。

定义工具

接下来，我们定义一个工具，从API获取当前的日期和时间。

from langchain_core.tools import tool
import requests

@tool
def get_date_and_time() -> dict:
    """
    调用工具从API获取当前的日期和时间。
    """
    try:
        response = requests.get("https://timeapi.io/api/Time/current/zone?timeZone=Europe/Brussels")
        response.raise_for_status()
        data = response.json()
        return {"date_time": data["dateTime"], "timezone": data["timeZone"]}
    except requests.RequestException as e:
        return {"error": str(e)}

此函数获取当前的日期和时间，并以JSON格式返回。

设置语言模型

我们现在设置语言模型，并将其与定义的工具集成，使聊天机器人能够在其响应中调用外部工具。

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools([get_date_and_time])

构建对话流程

使用LangGraph，我们现在构建一个简单的对话流程，仅依赖一个可以访问get_date_and_time工具的“聊天机器人”节点。

from langgraph.graph import StateGraph
from langchain_core.runnables import RunnableConfig

def chatbot(state: State, config: RunnableConfig):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge("chatbot", "chatbot")
graph = graph_builder.compile(checkpointer=memory)

此设置允许聊天机器人处理消息，将其存储在记忆中，并相应地进行响应。

Step 2: 用语音功能增强聊天机器人

接下来，我们将向聊天机器人工作流中添加两个额外的节点——一个用于使用Whisper将传入音频处理为文本，另一个用于使用ElevenLabs API将文本响应转换为音频。这个增强将使聊天机器人变成一个功能齐全的语音助手。

音频输入和输出处理的额外节点。

输入：捕获音频

为了提供无缝和互动的体验，聊天机器人必须能够准确地听到和理解用户。这就是OpenAI的Whisper模型发挥作用的地方，它使聊天机器人能够有效地捕获和转录口语。

语音转文本过程包括几个关键步骤：

音频捕获: 使用sounddevice库，聊天机器人监听用户麦克风的语音输入。当检测到声音时，录音过程自动开始，并在静默一段时间后停止。

处理音频输入: 配置多个参数以优化语音检测：

• SAMPLE_RATE: 定义采样率以确保足够的语音捕获。
• THRESHOLD: 设置检测语音活动的最低音频水平。
• SILENCE_DURATION: 确定在停止录音之前应保持多长时间的静默。
• CHUNK_SIZE: 指定用于处理的音频块的大小。

语音识别: 一旦录音完成，使用OpenAI的Whisper API处理并转录为文本，该API支持高精度的多语言语音识别。

import io
import threading
import numpy as np
import sounddevice as sd
from scipy.io.wavfile import write

from openai import OpenAI

from langgraph.graph import  MessagesState, HumanMessage

openai_client = OpenAI()

SAMPLE_RATE = 16000  
THRESHOLD = 500  
SILENCE_DURATION = 1.5  
CHUNK_SIZE = 1024

def record_audio_until_silence(state: MessagesState):
    """等待用户开始说话，录制音频，并在检测到静默后停止。"""

    audio_data = []  
    silent_chunks = 0  
    started_recording = False

    def record_audio():
        """持续录制音频，等待用户开始说话。"""
        nonlocal silent_chunks, audio_data, started_recording

        with sd.InputStream(samplerate=SAMPLE_RATE, channels=1, dtype='int16') as stream:
            print("等待您开始说话...")

            while not started_recording:
                audio_chunk, _ = stream.read(CHUNK_SIZE)
                audio_array = np.frombuffer(audio_chunk, dtype=np.int16)

                if np.abs(audio_array).max() > THRESHOLD:
                    started_recording = True
                    print("检测到声音。开始录音。")
                    audio_data.append(audio_chunk)
                    break

            while True:
                audio_chunk, _ = stream.read(CHUNK_SIZE)
                audio_data.append(audio_chunk)
                audio_array = np.frombuffer(audio_chunk, dtype=np.int16)

                if np.abs(audio_array).max() < THRESHOLD:
                    silent_chunks += 1
                else:
                    silent_chunks = 0

                if silent_chunks > (SILENCE_DURATION * SAMPLE_RATE / CHUNK_SIZE):
                    print("检测到静默。停止录音。")
                    break

    recording_thread = threading.Thread(target=record_audio)
    recording_thread.start()
    recording_thread.join()

    audio_data = np.concatenate(audio_data, axis=0)

    audio_bytes = io.BytesIO()
    write(audio_bytes, SAMPLE_RATE, audio_data)  
    audio_bytes.seek(0)  
    audio_bytes.name = "audio.wav"

    transcription = openai_client.audio.transcriptions.create(
        model="whisper-1", 
        file=audio_bytes,
        language='nl'
    )

    print("这是转录结果:", transcription.text)

    return {"messages": [HumanMessage(content=transcription.text)]}

输出：通过语音使聊天机器人活起来

为了提供真正引人入胜的用户体验，聊天机器人需要以自然和类人方式进行交流。这就是ElevenLabs强大的文本转语音（TTS）功能发挥作用的地方，它使我们能够将聊天机器人的响应无缝转换为生动的音频。

该过程涉及几个关键步骤：

• 初始化ElevenLabs客户端: 配置ElevenLabs API客户端并设置API密钥，以便与文本转语音服务进行通信。
• 处理聊天机器人响应: 在将文本转换为语音之前，清理聊天机器人的响应，以去除可能干扰音频输出的格式化伪影。
• 将文本转换为语音: 将清理后的文本发送到ElevenLabs API，利用高级语音设置控制稳定性、相似度提升和说话者风格等方面。
• 播放生成的音频: 一旦TTS转换完成，音频响应将回放给用户，确保对话流畅自然。

以下是文本转语音转换的实现方式：

import os
from elevenlabs import play, VoiceSettings
from elevenlabs.client import ElevenLabs
from langgraph.graph import  MessagesState

elevenlabs_client = ElevenLabs(api_key=os.getenv("ELEVEN_API_KEY"))

def play_audio(state: MessagesState):
    """播放来自ElevenLabs远程图的音频响应。"""

    response = state['messages'][-1]

    cleaned_text = response.content.replace("**", "")

    response = elevenlabs_client.text_to_speech.convert(
        voice_id="YUdpWWny7k5yb4QCeweX",  
        output_format="mp3_22050_32",
        text=cleaned_text,
        model_id="eleven_turbo_v2_5", 
        language_code="nl",
        voice_settings=VoiceSettings(
            stability=0.0,
            similarity_boost=1.0,
            style=0.0,
            use_speaker_boost=True,
        ),
    )

    play(response)

图形构建：集成音频处理

通过定义处理文本和语音交互的功能，下一步是将这些能力无缝集成到一个统一的工作流程中。通过构建一个智能对话图，我们可以确保语音识别、基于文本的对话和语音响应之间的顺畅流动。

在这个阶段，我们将定义一个新的基于LangGraph的工作流程，将音频输入和输出功能与我们的聊天机器人连接，从而实现以下操作顺序：

1. 捕获音频输入：

   • 聊天机器人监听用户输入，并将其从语音转换为文本。
   • 一旦检测到语音并进行处理，生成的文本将传递给聊天机器人进行分析。

2. 处理基于文本的对话：

   • 聊天机器人根据转录的输入和其内部逻辑生成响应。
   • 它可以利用诸如日期和时间检索功能等工具来丰富交互。

3. 生成和播放音频输出：

   • 聊天机器人的文本响应通过ElevenLabs API转化为语音。
   • 响应被回放给用户，完成交互循环，并允许用户再次响应，从而无缝继续对话。

通过这种方式构建对话工作流程，我们创建了一个能够高效处理语音和文本交互的动态语音助手。

以下是我们构建完整图形以集成所有组件的方法：

from langgraph.graph import StateGraph, MessagesState, END, START

builder = StateGraph(MessagesState)

builder.add_node("audio_input", record_audio_until_silence)
builder.add_node("agent", graph)
builder.add_node("audio_output", play_audio)
builder.add_edge(START, "audio_input")
builder.add_edge("audio_input", "agent")
builder.add_edge("agent", "audio_output")
builder.add_edge("audio_output", "audio_input")
audio_graph = builder.compile(checkpointer=memory)

第3步：测试语音助手

一旦聊天机器人和音频功能完全整合，测试系统以确保其顺畅运行是至关重要的。我们将通过流式传输输入命令并观察聊天机器人的响应来启动测试对话。

from langchain_core.messages import convert_to_messages
from langchain_core.messages import HumanMessage

config = {"configurable": {"thread_id": "1"}}

for chunk in audio_graph.stream({"messages": HumanMessage(content="Follow the user's instructions:")}, stream_mode="values", config=config):
    chunk["messages"][-1].pretty_print()

结论

从零开始构建这个AI voicebot是一次令人兴奋和有益的经历。使用预构建的工具是一回事，但从头开发一个解决方案则提供了对整体如何协同工作的更深刻理解。从将speech-to-text与Whisper集成到使用ElevenLabs生成逼真的响应，每一步都带来了宝贵的见解和进一步定制的可能性。

如果你是一个喜欢与AI捣鼓的人，或者想将你的项目超越no-code平台的限制，深入代码将打开一个机会的世界。当然，这需要更多的努力，但将你的解决方案精确调整到你的需求使一切都值得。无论你是为了乐趣而构建，还是为了解决现实世界的挑战，创建AI驱动的语音助手的旅程才刚刚开始。

感谢您成为社区的一部分

在您离开之前：

• 确保点赞并关注作者
• 关注我们：X | LinkedIn | YouTube | Newsletter | Podcast
• 查看 CoFeed，这是获取最新科技动态的智能方式
• 在 Differ 上启动您自己的免费 AI 驱动博客
• 欲获取更多内容，请访问 plainenglish.io + stackademic.com