介绍
你是否曾经想过与视频聊天有多么好？作为一个博客作者，我经常觉得看一个长达一小时的视频来获取相关信息很无聊。有时候，看一个视频以获取任何有用的信息感觉像是一份工作。所以，我构建了一个聊天机器人，让你可以与 YouTube 视频或任何视频进行聊天。这得益于 GPT-3.5-turbo、Langchain、ChromaDB、Whisper 和 Gradio。因此，在本文中，我将介绍如何使用 Langchain 构建一个功能强大的聊天机器人，用于与 YouTube 视频交互。
学习目标
- 使用 Gradio 构建 Web 界面
- 使用 Whisper 处理 YouTube 视频并提取文本数据
- 适当处理和格式化文本
- 创建文本数据的嵌入
- 配置 Chroma DB 来存储数据
- 使用 OpenAI chatGPT、ChromaDB 和嵌入函数初始化 Langchain 对话链
- 最后，查询并流式传输答案到 Gradio 聊天机器人
在开始编码之前，让我们先了解一下我们将使用的工具和技术。
本文是数据科学博客马拉松的一部分。
Langchain
Langchain 是一个用 Python 编写的开源工具，可以使大型语言模型具备数据感知能力和代理能力。那这到底是什么意思呢？大多数商业可用的 LLM，例如 GPT-3.5 和 GPT-4，对它们训练的数据有一个限制。例如，ChatGPT 只能回答它已经见过的问题。2021 年 9 月之后的任何内容对它来说都是未知的。这是 Langchain 解决的核心问题。无论是 Word 文档还是个人 PDF，我们都可以将数据提供给 LLM，并获得类似人类的响应。它具有针对矢量数据库、聊天模型和嵌入函数等工具的包装器，可以仅使用 Langchain 构建 AI 应用程序。
Langchain 还允许我们构建代理 – LLM 机器人。这些自主代理可以配置为多个任务，包括数据分析、SQL 查询，甚至编写基本代码。我们可以使用这些代理自动化许多事情。这很有帮助，因为我们可以将低级别的知识工作外包给 LLM，从而节省时间和精力。
在这个项目中，我们将使用 Langchain 工具构建一个视频聊天应用程序。有关 Langchain 的更多信息，请访问其官方网站。
Whisper
Whisper 是 OpenAI 的另一个产物。它是一个通用的语音转文本模型，可以将音频或视频转换为文本。它经过训练，可以执行多语言翻译、语音识别和分类等任务。
该模型有五种不同的大小：tiny、base、小猪AI、small 和 large，具有速度和准确性的权衡。模型的性能还取决于语言。下图显示了使用 large-v2 模型对 Fleur 数据集进行 WER（单词错误率）分解的语言情况。
矢量数据库
大多数机器学习算法无法处理原始的非结构化数据，如图像、音频、视频和文本。它们必须转换为向量嵌入的矩阵。这些向量嵌入在多维平面中表示所述数据。为了获得嵌入，我们需要高效捕获数据的语义含义的深度学习模型。这对于制作任何 AI 应用程序都非常重要。为了存储和查询这些数据，我们需要能够有效处理它们的数据库。这导致了创建称为矢量数据库的专业数据库。有多个开源数据库可用。Chroma、Milvus、Weaviate 和 FAISS 是最流行的一些。
向量存储的另一个独特之处在于，我们可以对非结构化数据进行高速搜索操作。一旦我们得到嵌入，就可以将它们用于聚类、搜索、排序和分类。由于数据点在一个向量空间中，我们可以计算它们之间的距离，以了解它们之间的关系。多种算法，如余弦相似度、欧氏距离、KNN 和 ANN（近似最近邻）被用来查找相似的数据点。
我们将使用 Chroma 向量存储 – 一个开源向量数据库。Chroma 还具有 Langchain 集成，这将非常有用。
Gradio
我们应用程序的第四个骑士 Gradio 是一个开源库，可以轻松分享机器学习模型。它还可以使用 Python 的组件和事件来构建演示网页应用程序。
如果您不熟悉 Gradio 和 Langchain，请在继续之前阅读以下文章。
- 让我们用 Gradio 建立 ChatGPT
- 为 PDF 构建 ChatGPT
现在让我们开始构建它。
设置开发环境
要设置开发环境，请创建一个 Python 虚拟环境或使用 Docker 创建一个本地开发环境。
现在安装所有这些依赖项
pytube==15.0.0
gradio == 3.27.0
openai == 0.27.4
langchain == 0.0.148
chromadb == 0.3.21
tiktoken == 0.3.3
openai-whisper==20230314导入库
import os
import tempfile
import whisper
import datetime as dt
import gradio as gr
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.chains import ConversationalRetrievalChain
from pytube import YouTube
from typing import TYPE_CHECKING, Any, Generator, List创建 Web 接口
我们将使用 Gradio 块和组件来构建应用程序的前端。下面是如何制作接口。随意根据您的需求进行自定义。
with gr.Blocks() as demo:
with gr.Row():
# with gr.Group():
with gr.Column(scale=0.70):
api_key = gr.Textbox(placeholder='输入 OpenAI API 密钥',
show_label=False, interactive=True).style(container=False)
with gr.Column(scale=0.15):
change_api_key = gr.Button('更改密钥')
with gr.Column(scale=0.15):
remove_key = gr.Button('删除密钥')
with gr.Row():
with gr.Column():
chatbot = gr.Chatbot(value=[]).style(height=650)
query = gr.Textbox(placeholder='在此输入查询',
show_label=False).style(container=False)
with gr.Column():
video = gr.Video(interactive=True,)
start_video = gr.Button('启动转录')
gr.HTML('OR')
yt_link = gr.Textbox(placeholder='在此粘贴 YouTube 链接',
show_label=False).style(container=False)
yt_video = gr.HTML(label=True)
start_ytvideo = gr.Button('启动转录')
gr.HTML('请在完成应用程序后重置应用程序以删除资源')
reset = gr.Button('重置应用程序')
if __name__ == "__main__":
demo.launch() 界面将显示如下
在这里，我们有一个文本框，可以输入 OpenAI 密钥。还有两个用于更改 API 密钥和删除密钥的键。我们还在左侧拥有聊天 UI 和一个用于呈现本地视频的框。在视频框的下方，我们有一个要求输入 YouTube 链接的框和写着“启动转录”的按钮。
Gradio 事件
现在我们将定义事件，使应用程序变得有交互性。将下面的代码添加到 gr.Blocks() 的末尾。
start_video.click(fn=lambda :(pause, update_yt),
outputs=[start2, yt_video]).then(
fn=embed_video, inputs=
,
outputs=
).success(
fn=lambda:resume,
outputs=[start2])
start_ytvideo.click(fn=lambda :(pause, update_video),
outputs=[start1,video]).then(
fn=embed_yt, inputs=[yt_link],
outputs = [yt_video, chatbot]).success(
fn=lambda:resume, outputs=[start1])
query.submit(fn=add_text, inputs=[chatbot, query],
outputs=[chatbot]).success(
fn=QuestionAnswer,
inputs=[chatbot,query,yt_link,video],
outputs=[chatbot,query])
api_key.submit(fn=set_apikey, inputs=api_key, outputs=api_key)
change_api_key.click(fn=enable_api_box, outputs=api_key)
remove_key.click(fn = remove_key_box, outputs=api_key)
reset.click(fn = reset_vars, outputs=[chatbot,query, video, yt_video, ])- start_video：被点击时将触发从视频中获取文本并创建对话链的过程。
- start_ytvideo：被点击时将执行相同的操作，但现在是从YouTube视频中获取文本，并在完成后将其呈现在下面。
- query：负责将来自LLM的响应流式传输到聊天UI中。
其余事件用于处理API密钥和重置应用程序。
我们已经定义了事件，但还没有定义负责触发事件的函数。
后端
为了不使它变得复杂和混乱，我们将概述我们将在后端处理的过程。
- 处理API密钥。
- 处理上传的视频。
- 转录视频以获取文本。
- 从视频文本中创建块。
- 从文本创建嵌入。
- 将向量嵌入存储在ChromaDB向量存储中。
- 使用Langchain创建对话检索链。
- 发送相关文档到OpenAI聊天模型（gpt-3.5-turbo）。
- 获取答案并在聊天UI上流式传输。
我们将在处理所有这些事情以及几个异常处理。
定义一些环境变量。
chat_history = []
result = None
chain = None
run_once_flag = False
call_to_load_video = 0
enable_box = gr.Textbox.update(value=None,placeholder= '上传您的OpenAI API密钥',
interactive=True)
disable_box = gr.Textbox.update(value = 'OpenAI API密钥已设置',
interactive=False)
remove_box = gr.Textbox.update(value = '已成功删除您的API密钥',
interactive=False)
pause = gr.Button.update(interactive=False)
resume = gr.Button.update(interactive=True)
update_video = gr.Video.update(value = None)
update_yt = gr.HTML.update(value=None) 处理API密钥
当用户提交密钥时，它将被设置为环境变量，我们还将禁用文本框以进行进一步输入。按下更改键将使其可变。单击移除密钥将删除密钥。
enable_box = gr.Textbox.update(value=None,placeholder= '上传您的OpenAI API密钥',
interactive=True)
disable_box = gr.Textbox.update(value = 'OpenAI API密钥已设置',interactive=False)
remove_box = gr.Textbox.update(value = '已成功删除您的API密钥',
interactive=False)
def set_apikey(api_key):
os.environ['OPENAI_API_KEY'] = api_key
return disable_box
def enable_api_box():
return enable_box
def remove_key_box():
os.environ['OPENAI_API_KEY'] = ''
return remove_box处理视频
接下来，我们将处理上传的视频和YouTube链接。我们将有两个不同的函数处理每种情况。对于YouTube链接，我们将创建一个iframe嵌入链接。对于每种情况，我们都会调用另一个负责创建链的函数make_chain()。
这些函数在有人上传视频或提供YouTube链接并按下转录按钮时触发。
def embed_yt(yt_link: str):
# This function embeds a YouTube video into the page.
# Check if the YouTube link is valid.
if not yt_link:
raise gr.Error('粘贴YouTube链接')
# Set the global variable `run_once_flag` to False.
# This is used to prevent the function from being called more than once.
run_once_flag = False
# Set the global variable `call_to_load_video` to 0.
# This is used to keep track of how many times the function has been called.
call_to_load_video = 0
# Create a chain using the YouTube link.
make_chain(url=yt_link)
# Get the URL of the YouTube video.
url = yt_link.replace('watch?v=', '/embed/')
# Create the HTML code for the embedded YouTube video.
embed_html = f"""<iframe width="750" height="315" src="{url}"
title="YouTube video player" frameborder="0"
allow="accelerometer; autoplay; clipboard-write;
encrypted-media; gyroscope; picture-in-picture"
allowfullscreen></iframe>"""
# Return the HTML code and an empty list.
return embed_html, []
def embed_video(video=str | None):
# This function embeds a video into the page.
# Check if the video is valid.
if not video:
raise gr.Error('上传视频')
# Set the global variable `run_once_flag` to False.
# This is used to prevent the function from being called more than once.
run_once_flag = False
# Create a chain using the video.
make_chain(video=video)
# Return the video and an empty list.
return video, []创建链
这是最重要的步骤之一。这涉及创建 Chroma 向量存储库和 Langchain 链。我们将为我们的用例使用一个对话检索链。我们将使用 OpenAI 嵌入，但对于实际部署，请使用任何免费的嵌入模型，如 Huggingface 句子编码器等。
def make_chain(url=None, video=None) -> (ConversationalRetrievalChain | Any | None):
global chain, run_once_flag
# 检查是否提供了 YouTube 链接或视频
if not url and not video:
raise gr.Error('请提供 YouTube 链接或上传视频')
if not run_once_flag:
run_once_flag = True
# 从 YouTube 链接或视频中获取标题
title = get_title(url, video).replace(' ','-')
# 处理视频文本
grouped_texts, time_list = process_text(url=url) if url else process_text(video=video)
# 将 time_list 转换为元数据格式
time_list = [{'source': str(t.time())} for t in time_list]
# 从具有元数据的处理文本中创建向量存储
vector_stores = Chroma.from_texts(texts=grouped_texts, collection_name='test',
embedding=OpenAIEmbeddings(),
metadatas=time_list)
# 从向量存储创建 ConversationalRetrievalChain
chain = ConversationalRetrievalChain.from_llm(ChatOpenAI(temperature=0.0),
retriever=
vector_stores.as_retriever(
search_kwargs={"k": 5}),
return_source_documents=True)
return chain- 从 YouTube 链接或视频文件中获取文本和元数据。
- 从文本和元数据创建 Chroma 向量存储。
- 使用 OpenAI gpt-3.5-turbo 和 Chroma 向量存储构建链。
- 返回链。
处理文本
在此步骤中，我们将对视频中的文本进行适当的分片，并创建我们在上述链构建过程中使用的元数据对象。
def process_text(video=None, url=None) -> tuple[list, list[dt.datetime]]:
global call_to_load_video
if call_to_load_video == 0:
print('yes')
# 基于给定的视频或 URL 调用 process_video 函数
result = process_video(url=url) if url else process_video(video=video)
call_to_load_video += 1
texts, start_time_list = [], []
# 从结果中的每个片段提取文本和开始时间
for res in result['segments']:
start = res['start']
text = res['text']
start_time = dt.datetime.fromtimestamp(start)
start_time_formatted = start_time.strftime("%H:%M:%S")
texts.append(''.join(text))
start_time_list.append(start_time_formatted)
texts_with_timestamps = dict(zip(texts, start_time_list))
# 将时间戳字符串转换为 datetime 对象
formatted_texts = {
text: dt.datetime.strptime(str(timestamp), '%H:%M:%S')
for text, timestamp in texts_with_timestamps.items()
}
grouped_texts = []
current_group = ''
time_list = [list(formatted_texts.values())[0]]
previous_time = None
time_difference = dt.timedelta(seconds=30)
# 根据时间差分组文本
for text, timestamp in formatted_texts.items():
if previous_time is None or timestamp - previous_time <= time_difference:
current_group += text
else:
grouped_texts.append(current_group)
time_list.append(timestamp)
current_group = text
previous_time = time_list[-1]
# 添加最后一组文本
if current_group:
grouped_texts.append(current_group)
return grouped_texts, time_list- process_text 函数接受 URL 或视频路径。然后在 process_video 函数中进行转录，并获得最终文本。
- 然后我们获取每个句子的开始时间（从 Whisper 中），并将它们分组为 30 秒。
- 最后，我们返回分组文本和每个组的开始时间。
处理视频
在此步骤中，我们将转录视频或音频文件并获取文本。我们将使用 Whisper 基本模型进行转录。
def process_video(video=None, url=None) -> dict[str, str | list]:
if url:
file_dir = load_video(url)
else:
file_dir = video
print('使用 Whisper 基本模型转录视频')
model = whisper.load_model("base")
result = model.transcribe(file_dir)
return result对于YouTube视频，由于我们无法直接处理它们，我们必须将它们单独处理。我们将使用一个名为 Pytube 的库来下载 YouTube 视频的音频或视频。因此，以下是您可以执行的操作。
def load_video(url: str) -> str:
# 此函数下载 YouTube 视频并返回下载文件的路径。
# 为给定的 URL 创建一个 YouTube 对象。
yt = YouTube(url)
# 获取目标目录。
target_dir = os.path.join('/tmp', 'Youtube')
# 如果目标目录不存在，则创建它。
if not os.path.exists(target_dir):
os.mkdir(target_dir)
# 获取视频的音频流。
stream = yt.streams.get_audio_only()
# 将音频流下载到目标目录。
print('----DOWNLOADING AUDIO FILE----')
stream.download(output_path=target_dir)
# 获取已下载文件的路径。
path = target_dir + '/' + yt.title + '.mp4'
# 返回已下载文件的路径。
return path- 为给定的 URL 创建一个 YouTube 对象。
- 创建一个临时目标目录路径
- 检查路径是否存在，否则创建目录
- 下载文件的音频。
- 获取视频的路径目录
这是从视频中获取文本到创建链的自下而上的过程。现在，唯一剩下的就是配置聊天机器人。
配置聊天机器人
现在，我们只需要向它发送一个查询和一个聊天历史记录即可获取答案。因此，我们将定义一个仅在提交查询时触发的函数。
def add_text(history, text):
if not text:
raise gr.Error('enter text')
history = history + [(text,'')]
return history
def QuestionAnswer(history, query=None, url=None, video=None) -> Generator[Any | None, Any, None]:
# 此函数使用模型链回答问题。
# 检查是否提供了 YouTube 链接或本地视频文件。
if video and url:
# 如果提供了 YouTube 链接和本地视频文件，则引发错误。
raise gr.Error('Upload a video or a YouTube link, not both')
elif not url and not video:
# 如果未提供输入，则引发错误。
raise gr.Error('Provide a YouTube link or Upload a video')
# 获取处理视频的结果。
result = chain({"question": query, 'chat_history': chat_history}, return_only_outputs=True)
# 将问题和答案添加到聊天历史记录中。
chat_history += [(query, result["answer"])]
# 对于答案中的每个字符，将其附加到历史记录的最后一个元素中。
for char in result['answer']:
history[-1][-1] += char
yield history, ''我们将聊天历史记录与查询一起提供，以保持对话的上下文。最后，我们将答案流回聊天机器人。不要忘记定义重置功能以重置所有值。
因此，这就是全部内容。现在，启动您的应用程序并开始与视频聊天。
这是最终产品的外观
视频演示：
实际应用案例
让最终用户与任何视频或音频进行聊天的应用程序可以具有广泛的用途。以下是此聊天机器人的一些实际应用案例。
- 教育：学生经常经历长时间的视频讲座。这个聊天机器人可以帮助学生从讲座视频中学习并快速提取有用信息，节省时间和精力。这将显着提高学习体验。
- 法律：法律专业人员经常进行漫长的法律程序和证言以分析案件、准备文件、研究或合规监控。像这样的聊天机器人可以在整理这些任务方面起到很大的作用。
- 内容摘要：此应用程序可以分析视频内容并生成摘要文本版本。这使用户可以在不完全观看视频的情况下掌握视频的重点。
- 客户互动：品牌可以为其产品或服务增加视频聊天机器人功能。这对于销售高票产品或需要大量解释的服务的企业非常有帮助。
- 视频翻译：我们可以将文本语料库翻译成其他语言。这可以促进跨语言交流、语言学习或为非母语者提供辅助功能。
以下是我能想到的一些潜在使用案例。聊天机器人为视频可以有更多有用的应用。
结论
因此，本文讲述了如何构建一个聊天机器人视频的功能演示Web应用程序。我们在整个文章中涵盖了许多概念。以下是本文的主要收获：
- 我们了解了Langchain – 一种流行的工具，可以轻松创建AI应用程序。
- Whisper是OpenAI的一种有效的语音转文字模型。这是一个开源模型，可以将音频和视频转换为文本。
- 我们了解了向量数据库如何促进向量嵌入的有效存储和查询。
- 我们使用Langchain、Chroma和OpenAI模型从头开始构建了一个完全功能的Web应用程序。
- 我们还讨论了我们聊天机器人的潜在实际应用案例。
这就是全部内容，希望你喜欢，也请考虑关注我的Twitter，了解更多与开发相关的内容。
GitHub代码库：sunilkumardash9/chatgpt-for-videos。如果您发现这有帮助，请为代码库打星标。
常见问题
本文中显示的媒体不归Analytics Vidhya所有，仅代表作者的观点。