LLM+本地知识库？简单又没那么简单

最近在研究如何将大语言模型结合本地知识库进行问答，虽然网上已经有很多教程，但大部分都是基于LangChain进行文本分割，然后调用模型向量化的API。这种方式的确很简单，但有这么几个前提：

大模型不使用ChatGPT的话，其实效果很差
尽管有多重切分方式，但还是很容易把文档中的一些语义撕裂。

由于众所周知的原因，使用ChatGPT的embedding的API很麻烦，因此自己想玩LLM以及本地知识库其实很多时候用的是自己搭建的模型，那么对于文档预处理的步骤就更加重要了，看了网上很多资料，还在学习中，本文就先把最近学习到的一些东西做个总结，也不太确定是否正确，若有错误，可以评论或私信，本人还是个刚开始学习AI的小白，还请多多指教。

构建本地知识库的前提

在构建本地知识库问答系统的时候，第一步要对本地的知识文档进行处理，因为希望更傻瓜式的去使用它，因此不太希望有人力参与对文档进行处理，比如分段、摘要等等。但如果不做任何处理，直接使用文档喂给大模型肯定是会超出tokeni限制。因此第一步会将文档的知识转成向量存储到向量数据库中，在进行知识问答的时候，先将问题在向量数据库中进行匹配，将匹配到的结果提供给LLM让其针对结果进行整理和回答。常见的也是最简单的知识库问答实现就是使用Langchain来进行文档预处理。

将一个文档转成向量数据库中的数据往往可以分成两个大的步骤，tokenizer 和 embedding。

Tokenizer负责将文本拆分成词元(token)。它将一个字符序列转换成一个词元序列。常见的tokenizer有基于空格、标点符号的简单tokenizer,还有更复杂的基于字典的tokenizer等。
Embedding则是将词元转换成词向量的表示。它为每个词元映射到一个稠密的向量空间,使得语义相关的词元之间的向量更加相近。Embedding可以通过事先训练好的词向量表获得,也可以在神经网络中进行学习。

我们最终将词汇或者语句转成向量是通过embedding得到的，但一般来说，我们不太可能将一整篇文档转换成向量。因为文档的长度往往都是比较长，会超过绝大部分模型的token限制；此外我们进行知识搜索的时候也不是要搜到整篇文档，而是文档中相关联的知识。那么tokenizer的第一步就是将文档拆分成合适的片段。

使用Langchain的方式进行文档拆分是非常简单的，具体实现细节可以参考Langchain的官方文档，这里只给出一个简单的demo：

# 导入文本  
loader = UnstructuredFileLoader(file_path)  
# 将文本转成 Document 对象  
document = loader.load()  
  
# 初始化文本分割器  
text_splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=20)  
  
# 切分文本  
split_documents = text_splitter.split_documents(document)

前两行不用解释太多，就是将文档加载进来，主要是第三步，文本分割器，Langchain提供了多种文本分割器的实现，比如字符文本分割器、Markdown文本分割器、上面demo中的递归字符文本分割器等等。使用Langchain的分割器非常方便，几乎几行代码就可以将文本切分成规定的样子，切分后的文本再送入tokenizer中处理，得到的结果再进行embedding，一个简单的本地知识库就做好了。

那么这种方式有什么问题呢？

虽然Langchain提供了多种分割器，但每一种都有着极强的局限性。对于一整篇文档而言，是存在上下文关联的，如果分割的时候正好撕开了上下文的关系，会导致向量搜索的结果很差，自然LLM的问答效果也会很差。
分割器在垂直领域的文档中的表现容易变差，因为不理解文档内容，可能会导致拆分一些专业词汇或语句。

因此文本分割的方式对于构建知识库来说是非常重要的，LLM无法直接理解自然语言，有效的分词确保语言的细微差别得以准确捕捉，尤其是在面对多样化或特定领域文本时。

虽然 tokenizer 更偏重的是文本处理，embedding 才是处理语义，但如果文本被分割的稀碎那后面的步骤全部都是基于一堆碎片做处理自然不会有什么价值，因此文本分割以及分词是非常重要的。

最近感兴趣的几个方案

目前先只考虑纯文本类型的文档，但可以是word，PDF、txt或者Markdown。如果穿插图表在其中的话暂时还没放在考虑范围内。

层级摘要

绝大部分文本文档都是有着段落层级关系，就像是Markdown或者word等，有着明显的层级关系。就想Langchain的Markdown分割器就可以很好的进行Markdown文件的切割。一般来说，一个段落的核心内容不会割裂太多，那么最简单粗暴的就是分割成段落，然后tokenizer分词，在标准的Markdown的文件上表现一般不会太差。但也要考虑一个问题，段落的长度也是不确定的，对于长段落，很多tokenizer可能无法处理。那么对于段落还要进行拆分，一般默认是限制最大长度然后割裂开，这种自然也会存在语句的断裂导致语义破坏。

针对这种情况，我们可以使用现有的一些模型对文档的长段落进行摘要，生成一个包含主要信息的较短版本。然后将这些摘要作为输入，使用能够处理长序列的模型（如Longformer或BigBird）进行向量化。这些模型的设计允许处理较长的文档，并且它们的稀疏注意力机制有助于维持段落间的上下文联系。

下面是一段伪码的示例，简单的展示了如何利用层级摘要来进行长文档处理。

from transformers import LongformerTokenizer, LongformerModel
from transformers import BartForConditionalGeneration, BartTokenizer
import torch

# 假设我们有一篇很长的文档，已经分成了多个段落
document_paragraphs = [
    "First long paragraph ...",
    "Second long paragraph ...",
    "Third long paragraph ...",
    # Assume there are more paragraphs
]

# 初始化BART分词器和模型用于摘要
summary_model_name = 'facebook/bart-large-cnn'
summary_tokenizer = BartTokenizer.from_pretrained(summary_model_name)
summary_model = BartForConditionalGeneration.from_pretrained(summary_model_name)

# 初始化Longformer分词器和模型用于处理长文档
longformer_tokenizer = LongformerTokenizer.from_pretrained('allenai/longformer-base-4096')
longformer_model = LongformerModel.from_pretrained('allenai/longformer-base-4096')

# 函数来摘要每个段落
def summarize_paragraphs(paragraphs):
    summarized_paragraphs = []
    for paragraph in paragraphs:
        inputs = summary_tokenizer(paragraph, return_tensors="pt", max_length=1024, truncation=True)
        summary_ids = summary_model.generate(inputs['input_ids'], num_beams=4, max_length=200, early_stopping=True)
        summary = summary_tokenizer.decode(summary_ids[0], skip_special_tokens=True)
        summarized_paragraphs.append(summary)
    return summarized_paragraphs

# 函数来编码文档的各个段落
def encode_paragraphs_with_longformer(paragraphs):
    # 使用Longformer分词器将所有段落编码为一个大的输入序列
    inputs = longformer_tokenizer(paragraphs, return_tensors="pt", padding=True, truncation=True)
    # 获取模型的输出
    outputs = longformer_model(**inputs)
    return outputs.last_hidden_state

# 首先对每个段落进行摘要
summarized_paragraphs = summarize_paragraphs(document_paragraphs)

# 然后使用Longformer编码摘要后的段落
document_encoding = encode_paragraphs_with_longformer(summarized_paragraphs)

# 'document_encoding' 现在包含了整个文档的编码表示

这种方式的优缺点也很明显：优点：

高效处理长文档：通过摘要减少了处理的文本长度，使长文档更易于管理。
保持关键信息：摘要过程提取关键信息，有助于减少信息丢失。
语义连贯性：长文档模型如Longformer保持了文档的整体语义结构。

缺点：

摘要的准确性：自动摘要可能会丢失一些重要细节，特别是在复杂的文档中。
资源消耗：运行高级深度学习模型需要相对较多的计算资源。
依赖模型质量：这种方案的效果依赖于摘要和长文档模型的质量。

这个方法我试了一下，emmm。。这个就像缺点中说的，摘要很重要，如果摘要的内容出的不好，对整个向量化的效果影响非常巨大，很考验模型本身的能力，所以需要选择合适的模型来做摘要和向量化。

滑动窗口

这个方案应该是最简单的方式，也是目前应该算是比较好实现的一种算法。其实就是对于每个长段落，使用滑动窗口技术进行分块处理，确保窗口之间有重叠部分，这样每个窗口都会包含来自相邻窗口的上下文信息。在Langchain的分割器中，往往也会设置一个overlap的值，这个其实就是重叠部分的设置，而且滑动窗口的步长是可以自己控制，可以多尝试几次找到相对比较合适的步长和重叠的大小。

处理完所有窗口后，我们获得了每个窗口的向量表示。这些向量表示包含了各自窗口的语义信息以及通过全局注意力机制获得的相关上下文信息。

最后，我们可以将这些向量合成一个连贯的段落或文档向量表示。这可以通过多种方法实现，如简单的向量平均、最大池化（max pooling）或者更复杂的融合技术。合成的向量表示旨在捕捉整个文档的核心语义，同时保持段落之间的流畅过渡。

为什么将这些向量还要合成起来？有这么几个好处：

保持文档完整性：尽管文档被切分成多个窗口以便处理，但最终目标是理解和表示整个文档的内容和语义。合成一个向量表示有助于捕捉整个文档的综合信息，而不仅仅是片段信息。
提高检索效率：在知识问答系统中，检索过程往往需要与整个文档的向量表示进行匹配。单一的文档向量表示可以显著提高检索效率，避免了与文档中的每个小片段逐一比较的需要。
减少信息丢失：在处理长文档时，单个窗口可能无法完全捕捉文档的所有关键信息。通过合成一个全面的向量表示，可以最大程度地减少由于窗口化处理带来的信息丢失。
增强上下文理解：合成的向量可以更好地反映不同窗口之间的上下文关系，特别是对于跨越多个窗口的主题或概念。这对于理解文档的整体结构和流程至关重要。
适应系统架构：许多知识问答和信息检索系统的架构设计是基于处理单一的文档向量，而不是处理多个分散的向量。因此，将窗口向量合成单一向量有助于这些系统更有效地处理和检索信息。

常见的合成方法有

简单平均：计算所有窗口向量的平均值。这种方法假设每个窗口的重要性相同。
加权平均：在某些情况下，不同窗口的重要性可能不同。加权平均方法允许根据窗口的相关性或其他标准对其进行加权。
最大池化：在这种方法中，从每个窗口向量中选择最大值作为最终向量的相应元素。这种方法有助于捕捉文档中最显著的特征。
串联：将所有窗口的向量串联起来形成一个长向量。这种方法可以保留更多的原始信息，但会导致维度非常高。
注意力机制：更复杂的方法可能包括使用注意力机制，以动态确定各个窗口的重要性，并据此合成向量。在选择合适的合成方法时，需要考虑文档的特点、处理的目的以及计算资源等因素。例如，对于信息密度较高的文档，使用加权平均或注意力机制可能更为合适；而对于信息较为均匀分布的文档，简单平均或最大池化可能就足够有效。

import torch
from transformers import BertTokenizer, BertModel
import numpy as np
import os

def sliding_window_split(text, window_size, step):
    """
    滑动窗口分割
    """
    chunks = []
    start = 0
    text = text.strip()
    while start < len(text):
        end = start + window_size
        chunks.append(text[start:end])
        start += step

        # Ensure we don't miss the end part of the text
        if end >= len(text) and start < len(text):
            chunks.append(text[start:])
            break

    return chunks

tokenizer = BertTokenizer.from_pretrained('bert-base-uncased')
model = BertModel.from_pretrained('bert-base-uncased')

def vectorize(chunk):
    """
    对一个chunk进行向量化
    """
    inputs = tokenizer(chunk, return_tensors='pt', max_length=512, truncation=True, padding='max_length')
    outputs = model(**inputs)
    chunk_vector = outputs.last_hidden_state.mean(dim=1)
    return chunk_vector

def vectorize_chunks(chunks):
    """
    对一个list的chunk进行向量化
    """
    vectors = []
    for chunk in chunks:
        chunk_vector = vectorize(chunk)
        vectors.append(chunk_vector)
    return vectors

def aggregate_vectors(vectors):
    return torch.mean(torch.stack(vectors), dim=0)

dir_path = 'dir_path'
texts = {}
# 遍历文件夹
files = os.listdir(dir_path)
for file in files:
  file_path = os.path.join(dir_path, file)
  with open(file_path, 'r') as f:
    text = f.read()
    texts[file] = text

# 遍历每个文件，进行向量化
# 每个文件对应的片段向量化之后的数组
file_vectors = {}
file_vectors_aggr = {}
file_chunks = {}
for file, text in texts.items():
  print(file)
  # 分割文档
  chunks = sliding_window_split(text, 1024, 555)
  file_chunks[file] = chunks
  # 分割后的向量数组
  vectors = vectorize_chunks(chunks)
  file_vectors[file] = vectors
  aggregated_vector = aggregate_vectors(vectors)
  file_vectors_aggr[file] = aggregated_vector

question = '什么是MVCC'
# 向量化问题，并移除梯度
q_vec = vectorize(question).detach()

# 计算相似度
for file, vector in file_vectors_aggr.items():
    # 转换为NumPy数组
    vector_np = vector.detach().numpy().flatten()  # 将向量转换为一维数组
    q_vec_np = q_vec.numpy().flatten()    # 同样将问题向量转换为一维数组

    # 计算余弦相似度
    similarity = np.dot(q_vec_np, vector_np) / (np.linalg.norm(q_vec_np) * np.linalg.norm(vector_np))
    # print(file, similarity)
    if (similarity > 0.48):
        file_vecs = file_vectors[file]
        for index, vec in enumerate(file_vecs):
          # 转换为NumPy数组
          vec_np = vec.detach().numpy().flatten()  # 将向量转换为一维数组
          q_vec_np = q_vec.numpy().flatten()    # 同样将问题向量转换为一维数组

          # 计算余弦相似度
          similarity = np.dot(q_vec_np, vec_np) / (np.linalg.norm(q_vec_np) * np.linalg.norm(vec_np))
          if similarity > 0.8:
            print(file, similarity)
            print(file_chunks[file][index])

我自己试了一下这个方法，但暂时体验也不是很好，可能得原因有很多，比如窗口大小和步长设置不合理、向量聚合方式不合理、模型本身能力不合适等。我这里由于硬件的限制无法测验更多的场景，只能说这个方法虽然简单，但体验还是有很大提升空间的。

自定义稀疏注意力机制

最后再说一个听起来比较高大上的方式，但复杂是真的复杂，这个也是我跟ChatGPT聊天得到的一个建议，反正我是实现不了。通常需要深入了解Transformer模型的工作原理，特别是它的注意力机制。Transformer模型通过注意力机制处理输入序列，但是传统的全连接（full attention）模式在处理长序列时计算量非常大。因此，为了处理更长的文档，需要实现一种计算上更高效的稀疏注意力模式。

在稀疏注意力模式中，模型并不是计算序列中每个元素与其他所有元素之间的注意力，而是只计算与每个元素相关联的一小部分元素的注意力。这可以通过多种方式实现，例如滑动窗口、固定模式或自定义模式。以下是实现自定义稀疏注意力的大致步骤：

定义稀疏结构：你需要定义一种模式或者结构来确定哪些token之间应该计算注意力。例如，可以设计一个模式，使得每个token只与其前后的k个token以及特定的全局token（如句首token）计算注意力。如果基于现有模型，比如BERT去做的话，需要深入理解BERT或类似模型的内部结构，特别是其注意力层的工作原理。这包括对输入进行查询（query）、键（key）和值（value）的转换，以及如何计算和应用注意力权重。
修改注意力计算：接下来需要修改自注意力层的实现，使其根据定义的稀疏结构计算注意力。这可能涉及改变softmax函数前的注意力得分计算，或在计算这些得分之前应用某种形式的掩码或过滤机制。这样做可以让模型的注意力更加集中于重要的输入部分。在进行这些修改时，通常保留模型的其余部分不变。这意味着，除了注意力层之外，模型的其他层（如嵌入层、前馈网络等）和它们的预训练权重将保持不变。
优化内存和计算：由于稀疏结构可能会导致不规则的内存访问模式和计算，可能需要进一步优化，例如使用特殊的数据结构或并行计算策略来提高效率
训练模型：使用自定义的稀疏注意力模式，你需要重新训练你的模型，或者至少对现有模型进行fine-tuning，以适应新的注意力结构。

优点：

优化的性能：能够有效处理长文档，同时减少计算资源的消耗。
定制的处理：可以根据具体需求调整注意力机制，更准确地捕捉文档的关键部分。
维护上下文：即使在长文档中也能保持良好的上下文连贯性。

缺点：

技术复杂度高：需要深入的技术知识来定制和调整模型。
实施难度：修改现有模型的注意力机制可能在实现上较为复杂。
调优成本：可能需要大量时间进行实验和调优来达到最佳效果。
性能测试：任何对模型的修改都需要经过彻底的测试，以确保改进实际上提高了模型的性能，并没有引入意外的副作用。

对于这种方式，需要深入研究模型内部结构，感觉可能更适合研究人员去玩，对于我这种可能刚开始学习的小白而言，走微调的路子可能更适合，比如使用LoRA也可以拿来试试：假设基于BERT进行微调：

import torch
import torch.nn as nn
from transformers import BertModel, BertConfig

class LoRA_BERT(nn.Module):
    def __init__(self, bert_model_name, rank):
        super(LoRA_BERT, self).__init__()
        self.bert = BertModel.from_pretrained(bert_model_name)
        self.rank = rank
        self.lora_adjustments = nn.ModuleDict()

        # 为BERT的每个注意力层添加LoRA调整
        for i, layer in enumerate(self.bert.encoder.layer):
            self.lora_adjustments[f"layer_{i}"] = LoRAAdjustment(layer.attention.self.query, rank)

    def forward(self, input_ids, attention_mask):
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        sequence_output = outputs.last_hidden_state

        return sequence_output

class LoRAAdjustment(nn.Module):
    def __init__(self, query_layer, rank):
        super(LoRAAdjustment, self).__init__()
        self.query_layer = query_layer
        self.rank = rank
        hidden_size = query_layer.out_features

        # 初始化LoRA参数
        self.delta_W = nn.Parameter(torch.randn(hidden_size, rank))
        self.delta_b = nn.Parameter(torch.randn(hidden_size))

    def forward(self, x):
        # 动态计算调整后的权重和偏置
        adjusted_weight = self.query_layer.weight + self.delta_W @ self.delta_W.t()
        adjusted_bias = self.query_layer.bias + self.delta_b

        return torch.nn.functional.linear(x, adjusted_weight, adjusted_bias)


bert_model_name = 'bert-base-uncased'
model = LoRA_BERT(bert_model_name, rank=4)

这个demo里面，

LoRA_BERT 类集成了预训练的BERT模型，并为其每个注意力层添加了LoRA调整。
LoRAAdjustment 类是一个自定义的模块，它为给定的层添加低秩权重和偏差调整。
在这个简化的例子中，只对BERT的query层进行了调整。在实际应用中，可能还需要对key和value层也进行类似的调整。上面代码是用GPT生产的一个简单DEMO，最佳的rank值，以及如何最有效地应用LoRA调整都是需要在实际实验过程中进行调整的。

总结

其实通过几种方法的尝试，除非对大模型本身能力进行优化，不然每一种方式都会存在很大的限制。不考虑硬件成本和技术开发成本的话，训练或调优出一个特定的模型才是最优解，当然如果基座模型的确足够强大，直接LangChain进行文档分割就足够了，相信使用ChatGPT进行embedding过的同学们应该能感受到，这种方法可能会引入一定程度的误差，但仍可以在多数情况下提供有效的结果，尝试不同的分割器和不同的参数是能找到相对较好的方案的。如果有条件的话，比如有一定的数据量且有硬件资源，可以对开源模型进行调优，让他能更好的适配本地文库的知识，比如构建的是公司内部文档的知识库，那新增一些新的公司文档，对于调优后的模型而言，也会有较好的表现。

构建本地知识库的前提

最近感兴趣的几个方案

层级摘要

滑动窗口

自定义稀疏注意力机制

总结

发送评论 编辑评论

发送评论编辑评论