大量数据用于训练大型语言模型 (LLM)，该模型包含数百万和数十亿个模型参数，目的是生成文本，例如文本补全、文本摘要、语言翻译和回答问题。虽然 LLM 从训练数据源中开发知识库，但总有一个训练截止日期，在此日期之后 LLM 将不会知道任何新生成的数据。

例如，训练 OpenAI 的 GPT-3.5-turbo-instruct LLM 的截止日期是 2021 年 9 月，因此，GPT-3.5-turbo-instruct LLM 可能无法准确回答有关 2022、2023 或 2024 年事件的问题。这种不属于 LLM 原始训练数据的数据称为外部数据。

检索增强生成 (RAG) 是一种旨在帮助解决此类情况的技术，它从授权的外部来源检索与输入提示相关的适当信息，并增强输入，以便 LLM 可以生成准确且相关的响应。实际上，RAG 构成了 LLM 和外部数据之间的网关。这种增强消除了重新训练或进一步微调 LLM 模型的需要。

NSDT工具推荐： Three.js AI纹理开发包 - YOLO合成数据生成器 - GLTF/GLB在线编辑 - 3D模型格式在线转换 - 可编程3D场景编辑器 - REVIT导出3D模型插件 - 3D模型语义搜索引擎 - Three.js虚拟轴心开发包 - 3D模型在线减面 - STL模型在线切割

1、LLM 的典型 M.O.

LLM 是自回归的，根据输入提示生成新的标记并将其标记为一系列标记。下一个最佳标记的生成是基于概率的，可以表示如下：

P( Yn ∣ X0, X1, ... Xn-1, θ )

本质上，新生成的第 n 个 token 的概率 Yn 取决于 n-1 个前一个 token 序列 X 的出现概率以及学习到的模型参数 θ。这里需要注意的是，tokenized 输入序列 X 在生成下一个 token 时起着至关重要的作用。此外，自注意力机制补充了有效的自回归，其中序列中的每个输入 token 通过关注和权衡序列中其他 token 的重要性来计算其表示。序列中 token 之间的这种复杂关系和依赖关系还使 LLM 能够破译最有可能的下一个最佳 token，该 token 与输入序列中的 token “很好地融合”。LLM 将新 token 附加到先前的 token 以形成新的输入序列，并重复自回归过程，直到满足完成条件，例如达到最大 token 计数。

这种自注意力驱动的自回归意味着 LLM 主要依赖输入序列来生成下一个最佳 token。只要输入序列有助于通过自注意力确定下一个最佳标记，LLM 就会继续处于“良性”循环中，生成连贯、可理解且相关的输出。相反，如果提示输入无助于确定下一个最佳标记，LLM 将开始依赖模型参数。在这种情况下，如果模型经过训练，包含足够的与输入提示相关的“知识”，则模型可能成功生成下一个最佳标记。相反，如果提示输入涉及 LLM 从未训练过的“外部数据”，则模型可能会进入“恶性”循环，生成不连贯、难以理解且可能不相关的输出。

有多种技术都用于解决这个问题。提示工程就是其中之一，其目标是通过调整提示来增强上下文，以便 LLM 可以生成相关输出，从而解决“缺失上下文”问题。 RAG 是另一种技术，其目标是通过以自动方式从外部数据源检索与输入提示最相关的信息上下文并增强提示，专门解决“由于外部数据而缺少上下文”的问题。

2、RAG 的挑战

RAG 的主要职责是从外部数据源（例如信息数据库、API 和其他文档存储库，如 Wikipedia）搜索和检索与输入提示上下文相关的数据。简单的关键字搜索无法解决问题。相反，RAG 需要语义搜索。为了方便语义搜索，从外部源检索到的文本信息被转换为数值表示或向量，通常称为文本嵌入，并存储在向量数据库中。

有各种模型或算法可以从文本创建这些嵌入。首先将提示转换为其向量表示，以搜索和检索最接近匹配的外部数据向量。然后计算提示向量与先前存储的外部数据向量之间的向量相似性（或向量距离）。使用阈值对最相似或最接近的向量进行排序和过滤，并检索其相应的文本信息以增强提示的上下文。以下概念图捕获了启用 RAG 的不同组件之间的典型交互：

RAG 面临的挑战是，进行向量驱动的语义搜索并非易事，需要大量计算资源，因为它涉及计算数据库中可能大量的向量的向量相似度或距离。为每个输入提示从庞大的向量数据库计算每个存储向量的相似度或距离度量将变得不可行。此外，语义匹配质量越低，LLM 的生成输出质量就越低。因此，找到一种有效进行语义搜索的方法变得至关重要。

3、解决方案

采用多种算法解决方案进行有效的语义搜索。此类算法的典型方法是将外部数据向量分组或聚类为最近邻居，并通过映射到此类聚类来索引它们。大多数向量数据库都提供此类索引作为内置功能。在语义搜索期间，首先针对输入提示向量评估匹配的聚类。对于每个评估的聚类，选择索引向量。然后计算输入提示向量与所选向量之间的相似度。这里的期望是，找到“最近邻居”作为中间步骤可以显著减少相似性计算的数量。最后，检索与通过阈值过滤的最相似或最近的向量相对应的文本信息。k-Nearest Neighbors、Ball-of-Radius-R、
Locality-Sensitive-Hashing、DBSCAN-Clustering、树状层次结构和图形层次结构等算法通常由向量数据库实现，以方便语义搜索。

没有一刀切的解决方案，因为不同的算法系列在内存效率、计算效率、延迟、准确性、向量维数、数据集大小等方面有不同的权衡。例如，聚类方法通过缩小语义搜索的向量空间来提高速度，而树状或图形方法可以提高低维向量数据的准确性。

4、自组织映射

自组织映射 (SOM) 是一种基于神经网络的降维算法，由 Teuvo Kohonen 在 20 世纪 80 年代开发。它通常用于将高维特征向量降维为低维（通常是二维）特征向量。SOM 背后的核心思想是将高维数据向量表示为低维空间中的特定节点，同时保留向量在原始空间中的拓扑结构。低维空间中的节点数（SOM 节点）是固定的（超参数）。SOM 节点的确切位置通过多个训练周期进行评估。迭代训练的目标是调整 SOM 节点在低维空间中的位置，以便它们映射到高维特征空间中的最近邻向量。换句话说，目标是将高维空间中的最近邻向量映射到也是低维空间中最近邻的 SOM 节点。

5、SOM 用于 RAG

在本文中，我想分享我使用 SOM 进行实验的笔记和发现，SOM 是一种可能的算法，可以推动 RAG 的语义搜索。与其他算法相比，SOM 之所以理想，有三个关键原因：

• 向量的高维性可能成为大多数其他算法（如树和图）的瓶颈——即所谓的维数灾难。相反，SOM 是为降维而构建的，因此，它可以有效地应用于高维和低维场景。
• SOM 对可能渗入原始高维向量空间的随机变化不太敏感，从而产生噪声。其他算法可能对这种噪声很敏感，影响它们将高维向量聚类或分组为最近邻的方式。由于 SOM 在低维向量空间中使用中间 SOM 节点，这些节点被评估为来自高维空间的映射向量的局部平均值，因此它可以有效地降低噪声。
• 外部数据集的庞大规模可能会限制其他算法创建语义向量空间，从而影响语义匹配的延迟和准确性。另一方面，SOM 可以处理海量数据集，因为低维空间中的 SOM 节点数量可以通过与底层数据集大小成比例的超参数进行微调。虽然使用大型数据集训练 SOM 可能需要更长时间，但训练完成后查询时间映射仍然更快。

我演示了一个使用 SOM 进行 RAG 语义搜索的简单示例，以使用 OpenAI 的 GPT-3.5-turbo-instruct LLM 增强问答上下文。使用 OpenAI 的 GPT-3.5-turbo-instruct LLM 的主要原因是 OpenAI 的 GPT-3.5-turbo-instruct LLM 的训练截止日期是 2021 年 9 月，因此，GPT-3.5-turbo-instruct LLM 可能无法准确回答有关 2022、2023 或 2024 年事件的问题。因此，有关 2022、2023 或 2024 年事件的信息可以成为 OpenAI 的 GPT-3.5-turbo-instruct LLM 的“外部数据”。

我使用 Wikipedia API 作为此类“外部数据”的来源来获取事件信息。以下是我用于开发和训练示例的步骤以及示例代码。

5.1 基于 PyTorch 的 Kohonen 的 SOM 实现

我利用 PyTorch 张量来表示向量，并使用 PyTorch 实现了 Kohonen 的 SOM。该算法使用二维格子，其大小成为超参数。该算法的数学方面源自精心设计的观点，并在这篇文章中提到了清晰的解释。

以下代码片段展示了 Kohonen 的 SOM 的 Python 类。完整代码可在此 GitHub 找到。值得注意的是，此实现是独立的，因此可以在 RAG 示例之外使用。

class KohonenSOM():
    """
    The code is developed based on the following article:
    http://www.ai-junkie.com/ann/som/som1.html
    
    The vector and matrix operations are developed using PyTorch Tensors.
    """
    def __init__( ... )
    ...
    def find_topk_best_matching_units( self, data_points : torch.Tensor, topk : int = 1 ) -> List[ List[ int ] ] :
        if len( data_points.size() ) == 1:
            #batching 
            data_points = data_points.view( 1, data_points.shape[0] )

        topk = int( topk )
        
        distances = self.dist_evaluator( data_points, self.lattice_node_weights )
        
        topk_best_matching_unit_indexes = torch.topk( distances, topk, dim=1, largest=False ).indices
        topk_best_matching_units = []
        
        for i in range( data_points.shape[0] ):
            best_matching_unit_indexes = topk_best_matching_unit_indexes[i]
            best_matching_units = [ self.lattice_coordinates[ bmu_index.item() ].tolist() for bmu_index in best_matching_unit_indexes ]
            topk_best_matching_units.append( best_matching_units )
        
        return topk_best_matching_units

5.2 基于 SOM 的向量索引器实现

向量索引器是一种实用程序，它使用 Kohonen 的 SOM 用来自外部数据集的数据向量来训练 SOM 节点。其主要目的是将每个数据向量映射到最近的前 k 个 SOM 节点，从而实现数据向量的高效索引。以下代码片段显示了向量索引器 Python 类的训练和索引功能。其完整代码可在此 GitHub 位置获得。虽然其实现目前仅限于示例需求，但可以扩展以满足其他要求。

class SOMBasedVectorIndexer():
    ...

    def train_n_gen_indexes( 
                                self, input_vectors : torch.Tensor, 
                                train_epochs : int = 100 
                           ):
        if self.generated_indexes:
            print( "WARNING: Indexes were already generated. Ignoring the request..." )
            return
        
        self.som.train( input_vectors, train_epochs )
   
        topk_bmu_indexes = self.som.find_topk_best_matching_units( input_vectors, topk = self.topk_bmu_for_indexing )
        
        for idx in tqdm( range( len( topk_bmu_indexes ) ), desc="SOM-Based Indexed Vectors"  ):
            bmu_indexes = topk_bmu_indexes[ idx ]
            
            for bmu_index in bmu_indexes:
                bmu_index_key = tuple( bmu_index )
                idx_set = self.som_node_idx_map.get( bmu_index_key, set() )
                idx_set.add( idx )
                self.som_node_idx_map[ bmu_index_key ] = idx_set
        
        self.generated_indexes = True

5.3 基于 OpenAI 嵌入的文本到矢量编码器

编码器的主要功能是使用 OpenAI 的文本嵌入 API 将文本转换为矢量表示。值得注意的是，使用嵌入 API 需要 OpenAI 帐户和 API 密钥。首次开通帐户时，OpenAI 会提供补充信用补助，这些补助足以用于测试目的访问 API。以下是展示 OpenAI 编码器 Python 类的批量编码功能的代码片段。完整代码可在此 GitHub 位置获得。

import openai
from openai.embeddings_utils import get_embedding
...
from vector_encoder_parent import VectorEncoder
...

class OpenAIEmbeddingsVectorEncoder( VectorEncoder ):
    def __init__( ... )
    ...
    def encode_batch( self, list_of_text : List[ str ] ) -> torch.Tensor :
        if list_of_text == None or len( list_of_text ) == 0:
            raise ValueError( "ERROR: Required list_of_text is None or empty" )
        
        list_of_text = [ str( text ) for text in list_of_text ]
        
        openai.api_key = self.openai_key
        response = openai.Embedding.create(
                                            input = list_of_text,
                                            engine = self.vector_encoder_id
                                          )
        
        embeddings = [ data["embedding"] for data in response["data"] ] 
        vectors = torch.tensor( embeddings, dtype=torch.float )
        return vectors

请注意，OpenAI 向量编码器类扩展了一个通用父类“VectorEncoder”，该类定义了要通过继承实现的抽象编码函数。可以通过从此父类继承来实现其他类型的向量编码器，以实现其他编码方案的可插入性。父向量编码器类的完整代码可在此 GitHub 位置找到。

5.4 Wikipedia API 驱动的数据源实现

此实用程序类旨在封装与 Wikipedia API 集成的数据检索逻辑。其主要功能是获取指定日历年数组的事件，格式化检索到的事件，并将其加载到 Pandas 数据框中。下面的代码片段捕获了实用程序类的主要功能，而完整代码可在此 GitHub 位置找到。

import requests
import pandas as pd
from dateutil.parser import parse
...
class WikiEventsDataSource():
    ...
    def fetch_n_prepare_data( self ):
        if self.fetched:
            print( "WARNING: Wiki events for the specified years already fetched. Ignoring the request..." )
            return
        
        main_df = pd.DataFrame()
        
        for year in self.event_years_to_fetch:
            wiki_api_params = {
                                "action": "query", 
                                "prop": "extracts",
                                "exlimit": 1,
                                "titles": year,
                                "explaintext": 1,
                                "formatversion": 2,
                                "format": "json"
                              }
        
            response = requests.get( "https://en.wikipedia.org/w/api.php", params=wiki_api_params )
            response_dict = response.json()
        
            df = pd.DataFrame()
            df[ "text" ] = response_dict["query"]["pages"][0]["extract"].split("\n")
            df = self.__clean_df__( df, year )
            
            main_df = pd.concat( [ main_df, df ] )
        
        self.df = main_df.reset_index(drop=True)
        self.fetched = True

5.5 基于 SOM 的 RAG 实用程序实现

基于 SOM 的 RAG 实用程序是示例实现的关键元素。它利用向量编码器、索引器和数据源来实现底层语义搜索的核心逻辑。基于 SOM 的 RAG 实用程序的完整代码可在此 GitHub 位置找到。

该实用程序实现了三个主要功能。第一个功能是从外部数据源加载数据并将其编码为向量，如以下代码片段所示。

...
from vector_encoder_parent import VectorEncoder
from vector_indexer import SOMBasedVectorIndexer

class SOM_Based_RAG_Util():
    ...
    def load_n_vectorize_data( self, data_source ):
        if self.data_loaded_n_vectorized:
            print( "WARNING: Data already loaded and vectorized. Ignoring the request..." )
            return
        
        data_source.fetch_n_prepare_data()
        self.df = data_source.get_data()
        
        vectors = None
        
        for i in tqdm( range(0, len(self.df), self.vectorize_batch_size ), desc="Vectorized Data Batch" ):
            list_of_text = self.df.iloc[ i:i+self.vectorize_batch_size ]["text"].tolist()
            batch_encoded_vectors = self.vector_encoder.encode_batch( list_of_text )
            
            if vectors == None:
                vectors = batch_encoded_vectors
            else:
                vectors = torch.cat( [ vectors, batch_encoded_vectors], dim=0 )
                
        self.vectors = vectors.to( self.device )
        self.data_loaded_n_vectorized = True

第二个功能是训练基于 SOM 的索引器以构建 Kohonen 的 SOM 节点，然后索引数据向量，如下面的代码片段所示。

def train_n_index_data_vectors( self, train_epochs : int = 100  ):
        if not self.data_loaded_n_vectorized:
            raise ValueError( "ERROR: Data not loaded and vectorized." )
        
        if self.data_vectors_indexed:
            print( "WARNING: Data vectors already indexed. Ignoring the request..." )
            return
        
        self.vector_indexer.train_n_gen_indexes( self.vectors, train_epochs )
        self.data_vectors_indexed = True

第三个函数是根据查询文本从先前存储的外部数据集中查找相似信息。此函数使用编码器将查询文本转换为向量，然后通过基于 SOM 的索引器搜索最可能的匹配项。然后，此函数使用余弦相似度或其他指定的相似度评估器计算查询向量与发现的数据向量之间的相似度。最后，此函数筛选相似度大于或等于指定相似度阈值的数据向量。以下代码片段捕获了函数实现。

def find_semantically_similar_data( self, query: str, sim_evaluator = None, sim_threshold : float = 0.8  ):
        if not self.data_vectors_indexed:
            raise ValueError( "ERROR: Data vectors not indexed." )
            
        if query == None or len( query.strip() ) == 0:
            raise ValueError( "ERROR: Required query text is not specified." )
            
        sim_threshold = float( sim_threshold )

        if sim_evaluator == None:
            sim_evaluator = nn.CosineSimilarity(dim=0, eps=1e-6)
        
        query_vector = self.vector_encoder.encode( query )
        query_vector = query_vector.view( self.vector_encoder.get_encoded_vector_dimensions() )
        query_vector = query_vector.to( self.device )
        
        nearest_indexes = self.vector_indexer.find_nearest_indexes( query_vector )
        nearest_indexes = nearest_indexes[0]
        
        sim_scores = []
        
        for idx in nearest_indexes:
            data_vector = self.vectors[ idx ]
            data_vector = data_vector.view( self.vector_encoder.get_encoded_vector_dimensions() )
            
            sim_score = sim_evaluator( query_vector, data_vector )
      
            if sim_score >= sim_threshold:
                sim_score_tuple = (idx, sim_score.item() )
                sim_scores.append( sim_score_tuple )
        
        sim_scores.sort( key = lambda x: x[1], reverse=True )
        
        semantically_similar_data = [ 
                                        { 
                                            'text': self.df[ 'text' ][ idx ],
                                            'sim_score' : sim_score
                                        } for idx, sim_score in sim_scores
                                    ]
        
        return semantically_similar_data

下面显示了基于 SOM 的 RAG 效用函数进行语义搜索的示例输出：

5.6 抽象问答聊天机器人及其基于 OpenAI 的实现

开发了一个抽象的“QuestionAnswerChatBot”Python 类，以促进类似聊天机器人的实现。它使用标准指令模板并用从 RAG 实用程序检索到的上下文相似信息填充它来增强提示的问题。

指定的最大新标记数限制了上下文增强的文本大小，而标记计数则推迟到底层实现。在 LLM 经济学中，标记就像货币。模型处理的每个标记都需要计算资源——内存、处理能力和时间。因此，LLM 需要处理的标记越多，计算成本就越高。

最后，一旦填充了 QA 指令，此类就会将 LLM 模型的提示委托给底层实现。以下代码片段捕获了主要功能；完整代码可在此 GitHub 位置获得。

from abc import ABC, abstractmethod
import torch
import math

class QuestionAnswerChatBot( ABC ):
    ...
    def find_answer_to_question( self, question : str, sim_threshold = 0.68, max_new_tokens : int = 5 ):
        if question == None or len( question.strip() ) == 0:
            raise ValueError( "ERROR: Required question is not specified" )
        
        sim_threshold = float( sim_threshold )
        max_new_tokens = int( max_new_tokens )
        
        qa_instruction = self.get_qa_instruction( question, sim_threshold = sim_threshold )
        
        answer_text = self.__get_answer_text__( qa_instruction, max_new_tokens = max_new_tokens )
        answer_text = self.__clean_answer_text__( qa_instruction, answer_text )
        
        return answer_text
    ...
    def __qa_template__( self ):
        qa_template = """Context: 
    
    {}
    
    ---
    
    Question: {}
    Answer:"""
        return qa_template

Python 类“
OpenAIQuestionAnswerChatBot”扩展了抽象“QuestionAnswerChatBot”，并使用 OpenAI LLM API 实现了聊天机器人功能。以下代码片段显示了该类的主要功能。完整代码可在此 GitHub 位置找到。

import openai
import tiktoken
from qa_chatbot import QuestionAnswerChatBot

class OpenAIQuestionAnswerChatBot( QuestionAnswerChatBot ):
    ...
    def __get_answer_text__( self, qa_instruction : str, max_new_tokens : int = 5 ) -> str :
        openai.api_key = self.openai_key
        
        basic_answer = openai.Completion.create(
                                                    model = self.openai_model_name,
                                                    prompt = qa_instruction, 
                                                    
                                               )
    
        answer_text = basic_answer[ "choices" ][0][ "text" ]
        return answer_text

    def __token_count__( self, text : str ):    
        return len( self.tokenizer.encode( text ) )

以下是如何使用通过语义搜索检索到的类似信息来增强提示性问题的上下文的示例：

5.7 测试示例问题

以下是使用 OpenAI 的 GPT-3.5-turbo-instruct LLM 测试 RAG 的示例问题。它们旨在确保答案与 2022 年、2023 年和 2024 年发生的事件相关。

sample_questions = [
                        "Who won the 2022 soccer world cup?",
                        "When did Sweden join NATO?",
                        "Who joined NATO in 2023?",
                        "Who joined NATO in 2024?",
                        "Which is the 31st member of NATO?",
                        "Which is the 32nd member of NATO?",
                        "Who won the Cricket World Cup in 2023?",
                        "Who defeated India in Cricket World Cup final in 2023?",
                        "Name the former prime minister of Japan that was assassinated in 2022?",
                        "When did Chandrayaan-3 land near the south pole of the Moon?",
                        "Where did Chandrayaan-3 land on the Moon?",
                        "Who acquired Twitter in 2022?",
                        "Who owns Twitter?",
                        "Who acquired Activision Blizzard in 2023?"
                   ]

5.8 将所有内容整合在一起

可以在 GitHub 上找到将所有组件整合在一起的完整 Jupyter 笔记本。以下代码片段展示了基于 OpenAI 的主要 QA 聊天机器人的启动。请注意，OpenAI 的文本嵌入算法“text-embedding-ada-002”用于矢量编码。同样，聊天机器人使用 OpenAI 的标记器“cl100k_base”来计算标记以限制上下文文本，从而利用 TikToken Python 库的内置函数来增强问题提示。

openai_vector_encoder_id = "text-embedding-ada-002"
openai_encoded_vector_dimensions = 1536
openai_tokenizer_name = "cl100k_base" 
openai_model_name = "gpt-3.5-turbo-instruct"

vector_encoder = OpenAIEmbeddingsVectorEncoder( openai_encoded_vector_dimensions, openai_vector_encoder_id, openai_key )

event_years_to_fetch = [ 2022, 2023, 2024 ]
data_source = WikiEventsDataSource( event_years_to_fetch  )
...
som_driven_rag_util = SOM_Based_RAG_Util( 
                                            vector_encoder = vector_encoder,
                                            som_lattice_height = 20,
                                            som_lattice_width = 30,
                                            learning_rate = 0.3,
                                            topk_bmu_for_indexing = 10,
                                            device = device
                                        )
...
openai_chatbot = OpenAIQuestionAnswerChatBot( 
                                                vector_db_util = som_driven_rag_util,
                                                openai_tokenizer_name = openai_tokenizer_name,
                                                openai_model_name = openai_model_name,
                                                openai_key = openai_key,
                                                question_input_max_token_count = 100,
                                                context_trim_percent = 0.1,
                                                device = device
                                            )

下列序列图有助于可视化初始化和实际问答阶段的所有组件交互。

6、发现

下图捕获了 OpenAI 的 GPT-3.5-turbo-instruct LLM 中带有和不带有上下文增强的问题/答案。

可以理解的是，LLM 很难回答有关 2021 年 9 月截止日期之后发生的事件的问题。在大多数情况下，它清楚地回答说这些问题来自相对于其培训截止日期的未来时间。相反，当提示问题的上下文与从维基百科检索到的 2022 年、2023 年和 2024 年的相关信息相结合时，同一个 LLM 可以准确完美地回答所有问题。真正的功劳归功于 SOM，它构成了 RAG 语义搜索的基础，用于检索和增强提示问题的上下文与相关信息。

7、建议的后续步骤

虽然上述示例是概念验证，用于评估自组织映射是否适合通过 LLM 实现检索增强文本生成，但建议使用更全面的基准测试来评估其性能，并与使用更大外部数据集的其他算法进行比较，其中性能以 LLM 输出的质量来衡量（例如困惑度 + 准确度）。此外，由于当前示例启用了可插入框架，因此建议使用其他开源和免费 QA LLM 进行此类基准测试，以最大限度地减少 LLM 使用费用。

为了帮助在本地环境中运行示例，我包含了“requirements.txt”文件，其中包含我在环境中用于运行和测试上述示例的各种版本的 Python 库。此文件可在此 GitHub 位置获得。

原文链接