教育行业AI应用Cerebrium创建实时RAG语音智能体

背景

     Cerebrium平台是一个集AI助手、LLM API访问、AI应用构建、低延迟语音AI机器人以及实时机器学习模型训练与部署于一体的综合性平台。它以其先进的技术和广泛的应用场景，为开发者和用户提供了高效、智能的AI解决方案。如今，教育资源非常容易获取。只要有网络连接，任何人在任何地方都可以在线观看世界上一些最知名的机构和个人提供的内容。然而，在这个循环中却缺少了一些东西。如果你对讲座内容不理解怎么办？如何向视频提问？如果你能做到这一点呢？Cerebrium将托管整个应用程序，因此，如果您还没有Cerebrium账户，可以在这里注册（我们将为您提供30美元的免费点数），按照这里的文档进行设置。

数据处理

为了开始工作，我们需要创建 Cerebrium 项目

cerebrium init voice-educator

main.py - 代码所在的入口点文件。

cerebrium.toml - 包含所有构建和环境设置的配置文件。

首先，我们需要开始数据处理任务--下载 Andrej 的 Youtube 视频，将其分块并嵌入。由于这将是一次性任务，而不是我们实时应用程序的一部分，因此请创建一个名为 data-processing.py 的新 Python 文件。按照良好的做法，让我们创建一个 Python 虚拟环境，以确保应用程序的依赖关系保持一致。我们还将使用环境变量，因此请安装 python-dotenv Pip 软件包以启用环境变量：

python -m venv educator
source educator/bin/activate
pip install python-dotenv

您应该已经进入了新环境！为了创建应用程序的 RAG 元素，我们需要下载 Andrej 的 Youtube 视频，将其转录并上传到我们的矢量数据库。这样，应用程序就能检索到所需的转录内容，从而为嵌入式 LLM 提供正确的上下文。要在本地下载视频，我们可以使用 pytube 库。

一. 在本教程中，我们只需下载视频音频。

运行 pip install pytube，并创建以下脚本：

from pytube import YouTube
import os

def download_videos(link: str, download_path: str):

    if not os.path.exists(download_path):
         os.makedirs(download_path)

    yt = YouTube(link)

    audio_download = yt.streams.get_audio_only()

    print("Downloading Audio...")
     audio_download.download(filename=f"{yt.title}.mp3", output_path = download_path)
     download_file_path = f"{download_path}/{yt.title}.mp3"

    return yt.title, download_file_path,

二.接下来，我们需要将音频文件转录为文本。

我们使用 Deegram API 完成这项工作。您可以在这里注册 Deepgram 账户（他们提供丰厚的免费级别）。然后，你可以在初始界面创建一个 API 密钥。运行 pip install deepgram httpx。在项目根目录下创建 .env 文件，并添加 API 密钥，我们将其命名为 DEEPGRAM_API_KEY

然后，我们可以创建以下代码来转录音频文件并返回文本：

from deepgram import (
     DeepgramClient,
     PrerecordedOptions,
     FileSource,
)
import httpx
from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

def transcribe_file(audio_file: str):
     print('transcribing')
     try:
         # STEP 1 Create a Deepgram client using the API key
         deepgram = DeepgramClient(os.getenv("DEEPGRAM_API_KEY"))

        with open(audio_file, "rb") as file:
             buffer_data = file.read()

        payload: FileSource = {
             "buffer": buffer_data,
         }

        #STEP 2: Configure Deepgram options for audio analysis
         options = PrerecordedOptions(
             model="nova-2",
             smart_format=True,
         )

        # STEP 3: Call the transcribe_file method with the text payload and options
         response = deepgram.listen.prerecorded.v("1").transcribe_file(payload, options, timeout=httpx.Timeout(300.0, connect=10.0))
         return response.results.channels[0].alternatives[0].transcript

    except Exception as e:
         print(f"Exception: {e}")

三.接下来，我们需要将文本嵌入一个矢量数据库

     这样我们的应用程序就可以轻松检索 LLM 所需的上下文，从而做出有效的响应。有很多文章介绍了这项任务需要选择的不同模型和策略。我们建议您查看这里的工具，看看哪些可能适合您的使用情况：

为简单起见，我们将使用 OpenAI 进行嵌入，并使用 Pinecone 作为我们的向量存储。Pinecone数据库是一个专为大规模向量集的高效索引和检索而设计的实时、高性能向量数据库。 您可以在此处注册 OpenAI 账户，在此处注册 PineCone 账户。我们将使用 Langchain 框架来创建我们的 RAG 应用程序，因此我们也将使用它来分块、上传、嵌入和存储我们的转录文本。您需要在 Pinecone 中创建一个索引，我们将把嵌入式数据上传到该索引中。由于我们使用的是 OpenAI 嵌入模型，因此必须将维度设置为 1536，而且我们将使用余弦指标来衡量相似性。

运行

pip install -qU langchain-text-splitters langchain_openai langchain_pinecone

然后，您需要从 OpenAI 和 PineCone 获取 API 密钥，并将其添加到 .env 文件中。我们将其分别称为 OPENAI_API_KEY 和 PINECONE_API_KEY

四.然后您就可以执行下面的代码

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings

def embed_text(text: str):

    print('Embedding')

    text_splitter = RecursiveCharacterTextSplitter(
         chunk_size=800,
         chunk_overlap=200,
         length_function=len,
         is_separator_regex=False,
     )
     split_text = text_splitter.split_text(text)
     return split_text
    

def save_embeddings_to_db(title: str, url: str, docs: [str]):
    
     index_name = "andrej-youtube"
     embeddings = OpenAIEmbeddings()
     # Connect to Pinecone index and insert the chunked docs as contents
     PineconeVectorStore.from_texts(docs, embeddings, index_name=index_name)

上述代码会获取我们转录的文本，根据我们设置的大小和重叠值将其分块，然后上传到 Pinecone 中的索引。

最后，让我们将这一切整合在一起：

if __name__ == "__main__":

    video_links = [
         "https://www.youtube.com/watch?v=l8pRSuU81PU",
         "https://www.youtube.com/watch?v=zduSFxRajkE",
         "https://www.youtube.com/watch?v=zjkBMFhNj_g",
         "https://www.youtube.com/watch?v=kCc8FmEb1nY"
     ]

    for link in video_links:
         title, download_path = download_videos(link, "./videos")
         texts = transcribe_file(download_path)
         docs = embed_text(texts)
         save_embeddings_to_db(title, link, docs=docs)

然后，您可以使用 python data-processing.py 运行脚本。你应该会看到一些日志，执行时间大约为 5 分钟。然后你就可以在 Pinecone 中导航到你的索引，并看到一些记录。

语音智能体

我们之前做过一篇教程，介绍如何使用 Deepgram、Daily 和 Pipecat 框架在 Cerebrium 上构建语音智能体。在这里我们只讨论我们所做的更改以及如何使用 Pipecat 实现 RAG。

Deepgram平台是一个提供先进的AI语音识别和自然语言处理技术的综合性平台。

语音到文本（Speech-to-Text）API：Deepgram的核心功能之一是将音频数据转换为文本。这一功能让开发者能够快速将语音转录功能集成到他们自己的应用程序和服务中，实现自动转录、内容索引和数据挖掘。
文本到语音（Text-to-Speech）API：Deepgram最新推出的文本到语音（TTS）服务，提供了自然、类似人类的声音，并且具有低延迟特性，非常适合用于对话式AI代理和应用程序。

支持场景：

客户服务和呼叫中心：Deepgram可以用于自动转录客户服务电话，帮助企业提高服务效率，通过语音分析改善客户体验，并从通话中提取有价值的数据和洞察。
媒体和内容制作：Deepgram可用于快速准确地转录视频、播客和其他媒体内容，节省编辑和后期制作的时间，同时提高内容的可访问性。医疗转录：在医疗领域，Deepgram可以帮助医生和医疗专业人员转录临床笔记、患者咨询和手术记录，提高记录的准确性和可检索性。
语音助手和聊天机器人：Deepgram的技术可以集成到语音助手和聊天机器人中，提供更自然、更准确的语音交互体验，提高用户满意度。

Daily.co平台是一家专注于提供视频会议和实时通信解决方案的公司。支持场景：

视频会议API：Daily.co提供了一套简单易用的视频会议API，可以轻松地集成到网站和应用程序中，帮助用户快速构建自己的视频通信功能。
高质量视频通话：Daily.co支持高清视频通话，确保用户在会议中能够享受到流畅的视频和音频体验。
跨平台支持：Daily.co的视频会议功能可以在各种设备和平台上运行，包括Web、iOS和Android，用户可以随时随地进行视频通话。

这次实现的不同之处在于，我们将为 LLM 使用外部 API，而不是本地模型。我们这样做有两个原因：

展示如何在需要 OpenAI 或 Anthropic 等性能更强的模型时使用外部 LLM。这确实会带来延迟权衡（~800ms 与本地 ~100ms）。
对于性能类似的模型，你可以让你的实现在 CPU 上运行，而不是在 H100 上运行，这样你就不会受到容量/成本的限制。

我建议你现在就克隆版本库，我将解释代码的变化，并只展示变化的片段。首先，让我们把 .env 文件中的秘密上传到 Cerebrium 账户，以便在应用程序中使用它们。导航至 Cerebrium 控制面板中的 “秘密”，然后上传 .env 文件--你应该会看到你的值弹出。我们将在本教程稍后部分引用这些值。

在 cerebrium.toml 文件中，确保设置了以下内容：

[cerebrium.deployment]
name = "educator"
python_version = "3.11"
include = "[./*, main.py, cerebrium.toml]"
exclude = "[.*]"
shell_commands = []
docker_base_image_url="prod.registry/daily:latest"

[cerebrium.hardware]
cpu = 2
memory = 8.0
compute = "CPU"
provider = "aws"
region = "us-east-1"

[cerebrium.scaling]
min_replicas = 0
max_replicas = 5
cooldown = 60

[cerebrium.dependancies.pip]
deepgram-sdk = "latest"
"pipecat-ai[silero, daily, openai, deepgram, elevenlabs]" = "latest"
aiohttp = ">=3.9.4"
torchaudio = ">=2.3.0"
channels = ">=4.0.0"
requests = "==2.32.2"
openai = "latest"
langchain = "latest"
langchain_community = "latest"
langchain_openai = "latest"
langchain_pinecone = "latest"
pinecone = "latest"

就是这样：

将我们的基础 Docker 镜像设置为本地包含 Deepgram 模型的日常镜像。这将使 STT 转换极其快速，因为它是在本地进行的，而不是通过网络。
我们将计算类型设置为 CPU，因为我们调用的是 LLM 的 API，不需要 GPU。
我们列出了应用程序所需的 pip 包

克隆声音

ElevenLabs平台是一家专注于为用户提供创新的AI语音合成解决方案的在线平台。该平台将人工智能技术与个性化语音合成相结合，为用户带来了全新的语音克隆和语音生成体验。ElevenLabs主要服务

语音合成

• 文本转语音：用户可以将任何文本内容转换为专业的语音输出，支持多种语言和声音风格。
• 语音克隆：通过VoiceLab工具，用户可以创建即时语音克隆（IVCs）和专业语音克隆（PVCs），从样本或自己的声音中克隆声音，或者从零设计全新的合成声音。
• 多语言支持：ElevenLabs支持多种语言，包括英语、德语、波兰语、西班牙语、意大利语、法语、葡萄牙语和印地语等，满足跨语言交流的需求。

项目管理

• ElevenLabs提供了项目管理工具，允许用户为长篇内容（如文章和有声书）创建配音，提高创作效率。

AI模型

• ElevenLabs的AI模型经过大量音频数据的训练，能够处理从自然对话到戏剧性朗读等多种语音任务。平台提供多种模型选择，包括英语专用的v1模型以及多语言v1（实验性）和多语言v2模型等。

定制化服务

• 企业客户可以通过ElevenLabs平台微调语音模型，建立自己的专有语音模型，满足个性化需求。

为了让我们的演示更加逼真，我们想用 ElevenLabs 克隆安德烈的声音，这样视频中的声音听起来就不那么像机器人了。你可以在这里注册一个 ElevenLabs 账户。不过，如果你想进行语音克隆，就需要升级到他们的入门计划，费用为 5 美元。要在 ElevenLabs 上克隆语音，你需要上传一段小于 4MB 的录音。由于我们在数据处理步骤中已经下载了音频文件，因此只需使用 Clideo 等平台（免费）对其进行裁剪即可。Clideo平台是一个提供在线视频处理服务的综合性平台，其主要功能包括视频压缩、视频合并、视频裁剪以及视频分享等。剪切完文件后，就可以将其上传到 ElevenLabs，这样就会得到一个语音 ID，我们稍后将在应用程序中使用它。

您需要将 ElevenLabs API 密钥上传到 Secrets！我们称之为 ELEVENLABS_API_KEY--下一步我们将使用它

Pipecat

PipeCat的主要特点和功能包括：

1. 多模态支持：PipeCat支持语音、文本、图像等多种模态的输入和输出，允许开发者构建更加丰富和自然的交互体验。

2. 灵活的管道系统：PipeCat的核心是一个灵活的管道系统，允许开发者将不同功能模块（如文本处理、语音识别、自然语言理解和回复生成等）串联起来，形成完整的对话流程。

3. 集成多种AI服务：PipeCat支持集成多种AI服务提供商的API，如Anthropic、Azure、Google等，从而可以充分利用这些服务商提供的先进技术和算法。

4. 广泛的应用场景：PipeCat构建的对话系统可以应用于各种场景，如个人教练、会议助手、故事讲述玩具、客户支持机器人等，具有很高的灵活性和可扩展性。

以下是我们对 Pipecat 的基本实现进行了一些改动：
我们使用 ElevenLabs 作为 TTS 元素，并使用上一步中克隆的语音。我们将 voiceID 更新为 ElevenLabs 指定的值。
我们实现了 Pipecat Langchain 集成，创建了一个可以记忆对话历史的对话代理。我们将在下一步编辑这部分代码：

from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.chains import create_history_aware_retriever, create_retrieval_chain
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

message_store = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
     if session_id not in message_store:
         message_store[session_id] = ChatMessageHistory()
     return message_store[session_id]

async def main(room_url: str, token: str):
    
     async with aiohttp.ClientSession() as session:
         transport = DailyTransport(
             room_url,
             token,
             "Andrej Karpathy",
             DailyParams(
                 audio_out_enabled=True,
                 transcription_enabled=True,
                 vad_enabled=True,
                 vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
             )
         )

        stt = DeepgramSTTService(
              name="STT",
              api_key=None,
              url='ws://127.0.0.1:8082/v1/listen'
         )

        tts = ElevenLabsTTSService(
             aiohttp_session=session,
             api_key="49a9831645c1cf792f99eb6b73c77f1f",#get_secret("ELEVENLABS_API_KEY"),
             voice_id="uGLvhQYfq0IUmSfqitRE",#get_secret("ELEVENLABS_VOICE_ID"),
         )

                ##We are about to replace this langchain
         prompt = ChatPromptTemplate.from_messages(
             [
                 ("system",
                  "Be nice and helpful. Answer very briefly and without special characters like `#` or `*`. "
                  "Your response will be synthesized to voice and those characters will create unnatural sounds.",
                  ),
                 MessagesPlaceholder("chat_history"),
                 ("human", "{input}"),
             ])
         chain = prompt | ChatOpenAI(model="gpt-4o", temperature=0.7)
         history_chain = RunnableWithMessageHistory(
             chain,
             get_session_history,
             history_messages_key="chat_history",
             input_messages_key="input")
         lc = LangchainProcessor(history_chain)
                 ##end of Langchain segment

        avt = AudioVolumeTimer()
         tl = TranscriptionTimingLogger(avt)

        tma_in = LLMUserResponseAggregator()
         tma_out = LLMAssistantResponseAggregator()

        pipeline = Pipeline([
             transport.input(),   # Transport user input
             avt,  # Audio volume timer
             stt,  # Speech-to-text
             tl,  # Transcription timing logger
             tma_in,              # User responses
             lc,                 # LLM
             tts,                 # TTS
             transport.output(),  # Transport bot output
             tma_out,             # Assistant spoken responses
         ])

        task = PipelineTask(pipeline, PipelineParams(
             allow_interruptions=True,
             enable_metrics=True,
             report_only_initial_ttfb=True,
         ))

Langchain RAG 管道

要使用 Langchain 创建 RAG 管道，我们只需创建一个检索链即可。这需要
一个 LLM，本例中将使用 OpenAI 的新 GPT-4o-mini 模型。
我们将使用 OpenAI 进行嵌入，使用 Pinecone 进行向量存储，只需在数据处理步骤中进行链接即可。
然后，我们将使用 Langchain 的 RunnableWithMessageHistory，以便在检索 LLM 上下文时使用我们的消息历史记录。

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
vectorstore = PineconeVectorStore.from_existing_index(
     "andrej-youtube", OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever()
question_answer_chain = create_stuff_documents_chain(llm, answer_prompt)
rag_chain =  create_retrieval_chain(retriever, question_answer_chain)

history_chain = RunnableWithMessageHistory(
     rag_chain,
     get_session_history,
     history_messages_key="chat_history",
     input_messages_key="input",
     output_messages_key="answer")

您可以通过 Langchain 实现 history_aware_retriever，根据聊天记录、向量存储和原始问题生成新的提示。我们发现这增加了太多延迟，对结果的影响也不够大。Langchain 的检索链会创建一个 dict，以上述 output_messages_key 参数显示的 “答案 ”键提供响应。因此，我们需要扩展 Pipecat Langchain 处理器，以满足这一需求。在 helpers.py 中添加以下代码：

from pipecat.processors.frameworks.langchain import LangchainProcessor

from langchain_core.messages import AIMessageChunk
from langchain_core.runnables import Runnable

from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.frames.frames import (
     Frame,
     AudioRawFrame,
     InterimTranscriptionFrame,
     TranscriptionFrame,
     TextFrame,
     StartInterruptionFrame,
     LLMFullResponseStartFrame,
     LLMFullResponseEndFrame,
     LLMResponseEndFrame,
     LLMResponseStartFrame,
     TTSStoppedFrame,
     MetricsFrame
)

class LangchainRAGProcessor(LangchainProcessor):
     def __init__(self, chain: Runnable, transcript_key: str = "input"):
         super().__init__(chain, transcript_key) 
         self._chain = chain
         self._transcript_key = transcript_key

    @staticmethod
     def __get_token_value(text: Union[str, AIMessageChunk]) -> str:
         match text:
             case str():
                 return text
             case AIMessageChunk():
                 return text.content
             case dict() as d if 'answer' in d:
                 return d['answer']
             case _:
                 return ""
            
     async def _ainvoke(self, text: str):
         logger.debug(f"Invoking chain with {text}")
         targetPhrases = [
           "you can continue with the lecture",
           "continue with the lecture",
           "you can continue with lecture",
           "continue with lecture",
           "play the video",
           "continue with the video"
         ]

        ##Simple fuzzy matching by checking if the target phrase is included in the transcript text
         matchFound = any(phrase in text for phrase in targetPhrases)
         if matchFound:
             print("Fuzzy match found for the phrase: 'You can continue with the lecture'")
             return
        
         await self.push_frame(LLMFullResponseStartFrame())
         try:
             async for token in self._chain.astream(
                 {self._transcript_key: text},
                 config={"configurable": {"session_id": self._participant_id}},
             ):
                 await self.push_frame(LLMResponseStartFrame())
                 await self.push_frame(TextFrame(self.__get_token_value(token)))
                 await self.push_frame(LLMResponseEndFrame())
         except GeneratorExit:
             logger.warning(f"{self} generator was closed prematurely")
         except Exception as e:
             logger.exception(f"{self} an unknown error occurred: {e}")
         finally:
             await self.push_frame(LLMFullResponseEndFrame())

这里有三点需要注意：

我们扩展了来自 Pipecat 的 LangchainProcessor，因为它已经包含了我们需要的很多功能--我只是编辑了其中的一些函数。在 __get_token_value 中，我们查找 AIMessageChuck 中是否包含 dict 对象 “answer”，因为它来自我们的检索链--在这种情况下，我们返回这个值。在我们的 _ainvoke 方法（本质上是调用 Langchain invoke）中，我们对用户所说的话进行模糊匹配，以便在用户说我们可以继续播放视频时获取信息。我们这样做是为了阻止信息进入 LLM 并得到响应。你可以通过函数调用来实现这一点，但为了简化演示，我使用了模糊匹配。

现在，您可以在 main.py 中的 history_chain 变量下添加以下内容：

lc = LangchainRAGProcessor(chain=history_chain)

部署到 Cerebrium

要将此应用程序部署到 Cerebrium，只需在终端运行以下命令： cerebrium deploy。

如果部署成功，您应该会看到类似下面的内容：

我们将把这些端点添加到前台界面。

连接前端

我们创建了 PipeCat 前端的公共分叉，以便向您展示此应用程序的演示。您可以在此处克隆该 repo。按照 README.md 中的说明操作，然后在您的 .env.development.local 中填入以下变量VITE_SERVER_URL=https://api.cortex.cerebrium.ai/v4/p-xxxxx/<APP_NAME> #这是基本 URL。请勿包含函数名称 VITE_SERVER_AUTH= #这是您可以从 Cerebrium 控制面板的 API 密钥部分获取的 JWT 令牌。现在，您可以运行 yarn dev 并访问网址： **http://localhost:5173/** 测试应用程序！

最终Demo, 在https://educationbot.cerebrium.ai/

总结

      本教程展示了如何使用 Cerebrium 和各种辅助服务（Daily、Deepgram、ElevenLabs、OpenAI 等）构建一个可扩展的个性化导师。通过下载并处理Karpathy的YouTube视频，利用Deepgram进行音频转文本，然后利用OpenAI模型对文本进行嵌入和检索来实现语境相关的自然语言生成（RAG）。此外，还借助了ElevenLabs进行语音合成，以及Pinecone作为向量存储库，以提高查询效率。整个应用通过Cerebrium部署，实现实时交互和个性化的学习体验。数据处理、语音代理构建、以及与前端界面集成的步骤，并强调了这种技术在教育领域的潜力及其对未来学习方式的影响。 将 RAG 与语音相结合，可以开发出无数种应用，而且因为它是完全可定制的，所以你可以在延迟、成本和准确性方面做出自己的权衡。在一个像人工智能一样快速发展的领域，我们 Cerebrium 的工作就是不断创新，思考未来的行业可能是什么样子，从而确保我们处于最佳的支持位置。我们的整个团队都来自南非，教育一直是一个重要的话题，因此我们思考我们能对如此重要的行业产生什么影响。正如纳尔逊-曼德拉曾经说过的一句名言： “教育是我们可以用来改变世界的最有力工具。“

如有想了解更多软件设计与架构, 系统IT,企业信息化, 团队管理 资讯，请关注我的微信订阅号：

作者：Petter Liu
出处：http://www.cnblogs.com/wintersun/
本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。 该文章也同时发布在我的独立博客中-Petter Liu Blog。