利用LangChain下载Bilibili视频并构建RAG应用教程

概述

本文将展示如何利用LangChain下载Bilibili视频并将其转录内容用于构建一个RAG（Retrieval-Augmented Generation）应用。

教程包括从视频下载、转录到RAG应用的实现步骤。以下是系统的架构图：

环境准备

在开始之前，请确保您的环境已配置好，包括安装必要的Python库，并能调用OpenAI的大模型

pip install openai langchain langchain-openai langchain_pinecone langchain[docarray] docarray pydantic==1.10.8 pytube python-dotenv tiktoken pinecone-client scikit-learn ruff

这里提供完整的代码和流程，但是需要网络环境好以及能调用OpenAI的大模型才能完整运行代码

本教程重点展示实现过程，不涉及对RAG基础原理的解释和说明

第一步：提取视频

首先，我们需要将Bilibili视频下载到本地，并对视频内容进行转录。

由于这一步不是本教程的重点，这里只提供代码和相关API工具链接，感兴趣的读者可以自行研究。

1.1 将目标视频链接转化为Base64编码

为了使用API下载视频，首先需要将视频的URL转换为Base64编码。以下是具体代码：

import requests
import os
import base64
from dotenv import load_dotenv

load_dotenv()

ROLL_APP_ID = os.getenv("ROLL_APP_ID")
ROLL_APP_SECRET = os.getenv("ROLL_APP_SECRET")

def url_to_base64(url):
    # 将URL编码为字节
    url_bytes = url.encode('utf-8')
    # 对字节进行Base64编码
    base64_bytes = base64.urlsafe_b64encode(url_bytes)
    # 将Base64字节转换为字符串
    base64_url = base64_bytes.decode('utf-8')
    return base64_url

# 示例URL
url = "https://www.bilibili.com/video/BV1im4y117pn"
base64_url = url_to_base64(url)
print("Base64编码后的URL:", base64_url)

1.2 使用API获得视频下载链接

API参数以及返回数据解析

# 接口地址和请求参数
API_URL = "https://www.mxnzp.com/api/bilibili/video"

API_PARAMS = {
    "url": base64_url,
    "app_id": ROLL_APP_ID,  # 替换为你的 app_id
    "app_secret": ROLL_APP_SECRET  # 替换为你的 app_secret
}

# 获取视频下载链接
def get_video_download_url():
    response = requests.get(API_URL, params=API_PARAMS)
    data = response.json()
    
    if data['code'] == 1:
        video_url = data['data']['list'][0]['url']
        video_title= data['data']['title']
        return [video_title,video_url]
    else:
        print("Error:", data['msg'])
        return None
    
video = get_video_download_url()
print(video)

1.3 视频下载到本地

import requests

def download_video_by_url(video):
    print("正在下载视频...")

    # 添加请求头
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.85 Safari/537.36"
    }

    response = requests.get(video[1], headers=headers)
    
    if response.status_code == 200:
        # 获取视频数据
        video_data = response.content

        # 指定视频文件的保存路径
        save_path = video[0] + ".mp4"  # 请指定保存路径和文件名
        # 使用urllib保存视频文件
        with open(save_path, "wb") as video_file:
            video_file.write(video_data)
            print("视频下载完成")
            return save_path
    else:
        print("无法下载视频，HTTP状态码:", response.status_code)

# 定义视频文件名
video_file_name = video[0] +  ".mp4"

if not os.path.exists(video_file_name):
    video_file_name = download_video_by_url(video)

2. 用whisper转录音频内容

这里选择用whisper做音频转录，不是因为它效果最好，只是它免费，并且可以下载到本地使用

顺便说明一下，whisper对中文的支持度不高，如果要做一个可应用的产品，需要调用其他转录工具

import tempfile
import whisper
from moviepy.editor import VideoFileClip

# 定义视频文件名和转录文件名
transcription_file_name = video_file_name +  ".txt"

if not os.path.exists(transcription_file_name):
    # 加载 Whisper 模型（可选择“tiny”, "base", "small", "large")
    whisper_model = whisper.load_model("small")

    with tempfile.TemporaryDirectory() as tmpdir:
        print(f"Temporary directory created at: {tmpdir}")
        
        # 将视频文件转换为音频文件
        video_path = os.path.join(tmpdir, video_file_name)
        audio_path = os.path.join(tmpdir, "audio.mp3")
        
        video = VideoFileClip(video_file_name)
        video.audio.write_audiofile(audio_path)
        print(f"Extracted audio file to: {audio_path}")

        if not os.path.exists(audio_path):
            raise FileNotFoundError(f"Extracted audio file not found at: {audio_path}")

        # 使用 Whisper 模型进行转录
        transcription = whisper_model.transcribe(audio_path, fp16=False, language="zh")["text"].strip()

        # 将转录结果保存到文件
        with open(transcription_file_name, "w") as file:
            file.write(transcription)
        print(f"Transcription saved to: {transcription_file_name}")

3. 设置langchain部分

需要在环境变量里配置API key

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
OPENAI_BASE_URL = os.getenv("OPENAI_API_BASE")

3.1. 选择大语言模型

这里用gpt-3.5，因为便宜

首先加载大语言模型

from langchain_openai.chat_models import ChatOpenAI

model = ChatOpenAI(openai_api_key=OPENAI_API_KEY,openai_api_base=OPENAI_BASE_URL, model_name="gpt-3.5-turbo")

3.2. 拆分文档

有许多不同的方法可以拆分文档。在这个例子中，我们将使用一个简单的拆分器，将文档拆分成固定大小的块（chunk）

不同文档拆分方法的更多信息，请参阅文档拆分器

我们将转录内容拆分成每块1000个字符，并且每块之间有20个字符的重叠

from langchain_community.document_loaders import TextLoader

loader = TextLoader(transcription_file_name)
text_documents = loader.load()
text_documents

from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=20)
documents = text_splitter.split_documents(text_documents)

3.3. 加载并嵌入拆分后的转录文本

from langchain_openai.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(openai_api_key=OPENAI_API_KEY,openai_api_base=OPENAI_BASE_URL)

3.4. 创建检索和生成答案的链式结构

from langchain.prompts import ChatPromptTemplate

template = """
Answer the question based on the context below. If you can't 
answer the question, reply "I don't know".

Context: {context}

Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

from langchain_community.vectorstores import DocArrayInMemorySearch
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

vectorstore2 = DocArrayInMemorySearch.from_documents(documents, embeddings)

chain = (
    {"context": vectorstore2.as_retriever(), "question": RunnablePassthrough()}
    | prompt
    | model
    | StrOutputParser()
)

3.5. 测试RAG应用

chain.invoke("这篇文章讲述了关于人生的什么内容，并给出详细解释答案的原因")

总结

目前展示的RAG流程，是把知识库放在本地载入，这样并不优雅

更好的实现方式是通过一个向量数据库来储存知识库的内容