用 Faiss 向量数据库定制属于自己的 AI-Vtuber

前言

使用 ChatGPT 可以很方便地构造一个聊天机器人，已有的成熟开源项目有很多，例如：

• chatgpt mirai qq bot
• QChatGPT
• ......

以上这些项目都基于文本与用户交互的，体验上很枯燥。但如果我们给这个机器人“加上一个皮套”，让她动起来，这样我们聊天的时候也就不会感到无聊了 😄

比如 Navi-Studio 大大基于 Unity 打造的开源聊天机器人项目：

• 
• 仓库地址

不过碍于自己不会 C# Unity 那一套，上面的项目对于我来说实在是很难上手。可能对于一些小白来说，也不知道应该如何使用 😭

‍

那么是否存在一个简单好用，方便部署的开源虚拟人项目，让小白快速地上手使用呢？

‍

‍

AI-Vtuber Live2D 虚拟主播

项目介绍

AI-Vtuber 这个项目是 IKaros 大大开发的，支持 openai、claude、chatterBot、chatGLM 等模型的虚拟主播，同时支持 B 站、快手、抖音直播。支持 edge-tts 等语言输出。

项目后端使用 python 开发，Live2D 则使用 Web 加载。技术上并不难，很方便部署与使用。我自己也是很快地将其运行起来。

‍

阅读源码的时候发现，该项目对虚拟人设定的配置上，是简单地使用“前后缀”进行限制：

‍

不过好在原作者也提供了 langchain_pdf 功能，让我们可以自己上传 pdf 等“设定资料”，让虚拟主播按照我们的设定的内容去回答：

• 
• config.json 设置

‍

问题与改进

美中不足的是，IKaros 大佬使用的是 OpenAI API 实现知识库问答的功能。总所周知，OpenAI API 是很烧钱的，虚拟主播每天都要回答数量巨多的弹幕，每一次回答都需要从向量数据库中检索信息，再发送给 OpenAI 得到答复。

这一过程就需要耗费很多 OpenAI 的 token，每天开销很大 😭😭

‍

这时候该我出场了 😆，这个项目如此切合我的技术栈，又怎么能不贡献一下自己代码呢？

我主要贡献的部分是 langchain_pdf_local 模块，在本地构建向量数据库（Faiss），然后用免费的 Claude 大模型服务，免费地实现原本需要 OpenAI API 才能做到的内容。

‍

使用 Faiss 向量数据库实现本地查询

Faiss 向量数据库 是 Facebook 开源的向量数据库库，支持在庞大数据集下快速搜索。

仅仅有 Faiss 向量数据库还不行，还得有将文本数据转换为“向量数据”的工具，这里我使用 HuggingFace 的 GanymedeNil/text2vec-large-chinese 模型，该模型在中文数据的向量化上效果很好

‍

以上是向量数据库构建所需要的技术，我们还得需要 Langchain 这个框架将 GPT 模型与向量数据库关联起来，这样才能让 GPT 能够从向量数据库中读取所需要的数据信息

• langchain 是一个基于大语言模型（LLMs）用于构建端到端语言模型应用的框架，它可以让开发者使用语言模型来实现各种复杂的任务，例如文本到图像的生成、文档问答、聊天机器人等

‍

实现效果

我们随便以一个直播间作为测试：

1. 输入“伊卡洛斯。”，程序会自动获取弹幕信息
2. 然后根据这个信息，从本地向量数据库中检索出最符合的几条信息
3. 之后我们根据组装这些信息，发给 Claude，让 Claude 结合这些信息回答

‍

构造向量数据库的文本 PDF 是原作者 Ikaros 最喜欢的“伊卡洛斯”的人物设定：

‍

‍

Langchain + Claude + Faiss 技术实现细节

接下来是具体的实现过程，将大致地讲一下代码层面上实现的原理

‍

AI-Vtuber 支持 B 站、快手、抖音直播，每个模式对应不同的启动入口：

其中 B 站 的效果最好、最稳定。接下来以 B 站 作为演示：

‍

程序核心 My_handle()

在 bilibili.py 中，我们读取 config.json 后创建 My_handle() 实例，这个实例整个服务的核心，负责处理 GPT 等各种模块的交互：

def start_server():
    common = Common()
    # 日志文件路径
    log_path = "./log/log-" + common.get_bj_time(1) + ".txt"
    Configure_logger(log_path)

    my_handle = My_handle("config.json")
    if my_handle is None:
        logging.info("程序初始化失败！")
        exit(0)

    # 初始化 Bilibili 直播间
    room = live.LiveDanmaku(my_handle.get_room_id())

......

‍

实例化 My_handle() 时，会根据 config.json 中不同的 chat_type 选择实例化不同的模式：

‍

而我所贡献的 langchain_pdf_local 的入口也在此处添加：

‍

Langchain_pdf_local 模块

进入 langchain_pdf_local.py 后，初始化函数首先会构造“向量数据库”和具体的 GPT 客户端：

• load_zip_as_db 函数会读取存在的 zip 压缩文件，将里面的一个或多个文本文件解析为向量数据

‍

load_zip_as_db

    def load_zip_as_db(self, zip_file,
                       pdf_loader,
                       model_name,
                       chunk_size=300,
                       chunk_overlap=20):
        if chunk_size <= chunk_overlap:
            logging.error("chunk_size小于chunk_overlap. 创建失败.")
            return
        if zip_file is None:
            logging.error("文件为空. 创建失败.")
            return

        self.local_db = create_faiss_index_from_zip(
            path_to_zip_file=zip_file,
            embedding_model_name=self.langchain_pdf_embedding_model,
            pdf_loader=pdf_loader,
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap
        )

        logging.info("成功创建向量知识库!")

‍

chunk_size 和 chunk_overlap

在向量数据库构建中，chunk_size 和 chunk_overlap 是两个重要参数，它们主要用于在处理大规模数据时分割数据集。下面对这两个参数进行解释：

1. chunk_size：这个参数定义了每个数据块（chunk）的大小。当处理大规模数据时，将数据集分割成较小的块可以减少内存消耗，加快处理速度。chunk_size 的值决定了每个数据块中向量的数量。
2. chunk_overlap：这个参数定义了相邻数据块之间的重叠部分。在某些情况下，数据块之间可能存在一定程度的相互依赖。为了确保这些依赖得到正确处理，可以引入部分重叠。chunk_overlap 的值决定了两个相邻数据块中共享的向量数量。

这两个参数的作用主要体现在以下几个方面：

• 减少内存消耗：将大规模数据集分割成较小的块可以降低内存使用，从而避免内存不足导致的程序崩溃。
• 加快处理速度：在处理较小数据块时，计算速度通常会更快，从而提高整体处理效率。
• 确保数据完整性：在处理有相互依赖的数据时，通过设置 chunk_overlap 参数，可以确保数据的完整性得到维护。

在设置这两个参数时，需要注意以下数学关系限制：

• chunk_size 必须大于 0，表示每个数据块至少包含一个向量。
• chunk_overlap 必须大于等于 0，表示相邻数据块之间可以有重叠部分。通常情况下，chunk_overlap 应该小于 chunk_size，以避免过多的冗余计算。

在实际应用中，需要根据数据集的特点和硬件资源情况来合理设置这两个参数，以获得最佳性能。

‍

create_faiss_index_from_zip

load_zip_as_db 函数主要调用这个方法生成向量数据库，这个方法在 utils/faiss_handler.py 中：

代码逻辑如下：

1. 获取 embedding 模型信息

2. 检测向量数据库存储信息是否存在

   1. 如果存在，则直接加载旧的数据库。使用 load_exist_faiss_file 方法
   2. 如果不存在，则解析压缩包的数据

3. 将解析好的向量数据库加载进 FAISS 数据库中，方便后续检索

‍

核心代码如下：

# utils/faiss_handler.py

......
    # 创建存储向量数据库的目录
    # 存储的文件格式
    # structure: ./data/vector_base
    #               - source data
    #               - embeddings
    #               - faiss_index
    store_path = os.getcwd() + "/data/vector_base/"
    if not os.path.exists(store_path):
        os.makedirs(store_path)
        project_path = store_path
        source_data = os.path.join(project_path, "source_data")
        embeddings_data = os.path.join(project_path, "embeddings")
        index_data = os.path.join(project_path, "faiss_index")
        os.makedirs(source_data)  # ./vector_base/source_data
        os.makedirs(embeddings_data)  # ./vector_base/embeddings
        os.makedirs(index_data)  # ./vector_base/faiss_index
    else:
        logging.warning(
            "向量数据库已存在，默认加载旧的向量数据库。如果需要加载新的数据，请删除data目录下的vector_base，再重新启动")
        logging.info("正在加载已存在的向量数据库文件")
        db = load_exist_faiss_file(store_path)
        if db is None:
            logging.error("加载旧数据库为空，数据库文件可能存在异常。请彻底删除vector_base文件夹后，再重新导入数据")
            exit(-1)
        return db

.........

    # 数据分片
    chunks = get_chunks(all_docs, chunk_size, chunk_overlap)

    # 向量数据
    text_embeddings = embeddings.embed_documents(chunks)
    text_embedding_pairs = list(zip(chunks, text_embeddings))

    # 向量数据保存位置
    embeddings_save_to = os.path.join(embeddings_data, 'text_embedding_pairs.pickle')

    # 保存数据
    with open(embeddings_save_to, 'wb') as handle:
        pickle.dump(text_embedding_pairs, handle, protocol=pickle.HIGHEST_PROTOCOL)

    # 将向量数据保存进FAISS中
    db = FAISS.from_embeddings(text_embedding_pairs, embeddings)
    db.save_local(index_data)

.......

1. 新数据则构建向量数据库
2. 存在已经创建的向量数据库则直接加载即可

‍

‍

load_exist_faiss_file

加载本地存在的数据库

在上面的函数中，需要写入 db_meta.json，这个 json 元信息就是存储当前模型的基础

只有解析出这个 json 信息，才能选择正确的模型进行加载数据库

‍

核心代码如下：

    db_meta_json = find_file("db_meta.json", path)

    # 获取模型数据
    embedding = EMBEDDINGS_MAPPING[db_meta_dict["embedding_model"]]

......

    # 加载index.faiss
    faiss_path = find_file_dir("index.faiss", path)
    if faiss_path is not None:
        db = FAISS.load_local(faiss_path, embedding)
        return db
    else:
        logging.error("加载index.faiss失败，模型已损坏。请彻底删除vector_base文件夹后，再重新导入一次数据")
        exit(-1)

‍

‍

云端构建向量数据库

本地电脑受限于性能，可能在构建向量数据库的时候异常地缓慢。

所以我自己单独写了一个 colab 脚本，通过白嫖 colab 的算力，快速地将大量 pdf 等文本文件转换为向量数据库：

• text2vec_colab —— HildaM

运行完成后，解压 vector_base.zip 放到 data 目录下即可：

‍

‍

本地模型设置

配置模型的位置在 utils/embeddings.py 文件中。配置好模型后，通过 EMBEDDINGS_MAPPING 全局列表向外提供：

"""
    模型1："sebastian-hofstaetter/distilbert-dot-tas_b-b256-msmarco"
    https://huggingface.co/sebastian-hofstaetter/distilbert-dot-tas_b-b256-msmarco
"""
DEFAULT_MODEL_NAME = "sebastian-hofstaetter/distilbert-dot-tas_b-b256-msmarco"
DEFAULT_MODEL_KWARGS = {'device': 'cpu'}
DEFAULT_ENCODE_KWARGS = {'normalize_embeddings': False}
default_vec_model = HuggingFaceEmbeddings(
    model_name=DEFAULT_MODEL_NAME,
    model_kwargs=DEFAULT_MODEL_KWARGS,
    encode_kwargs=DEFAULT_ENCODE_KWARGS
)

"""
    模型2："GanymedeNil/text2vec-large-chinese"
    https://huggingface.co/GanymedeNil/text2vec-large-chinese
"""
TEXT2VEC_LARGE_CHINESE = "GanymedeNil/text2vec-large-chinese"
text2vec_large_chinese = HuggingFaceEmbeddings(
    model_name=TEXT2VEC_LARGE_CHINESE,
    model_kwargs=DEFAULT_MODEL_KWARGS,
    encode_kwargs=DEFAULT_ENCODE_KWARGS
)


"""
    模型列表
"""
EMBEDDINGS_MAPPING = {
    DEFAULT_MODEL_NAME: default_vec_model,
    TEXT2VEC_LARGE_CHINESE: text2vec_large_chinese
}

‍

‍

美中不足的地方

我自认为我自己贡献的模块并不完美，目前只支持 Claude 和 Openai 两个 GPT 模型

后续我打算将模型层进行抽象，全局只需要配置一次 GPT 模型 key 信息，这样切换其他模式的时候就不需要再重新调整

‍

同时，Live2D 的配置我也要好好学一下。虽然 Ikaros 大佬已经配置好 Live2D 了，但是 Live2D 的动作与后端回应的话不相符，口型对不上。

打算参考一下 极客湾派蒙 项目中“情绪识别”模块的实现，将该功能融入到 AI-Vtuber 中。

‍

‍

总结

这篇博客主要记录了我参与开发 AI-Vtuber 项目的过程，也简单介绍了 langchain_pdf_local 模块的开发过程。

IKaros 大佬很积极维护项目，每天几乎都有 commit 😭，深感追不上大佬的节奏。。。。

我也会继续跟进下去，也希望在这个项目中践行自己所掌握的 AI 知识，尽力完善 AI-Vtuber 项目！

‍