將RAG與Semantic Search融入LLM：提升準確性與效率

前言

前幾篇分享了 IBM Watsonx.ai 平台，以及在平台上使用 LLM 完成客戶體驗分析、與LLM串連處理較複雜的問題。在這一篇中，我們想來嘗試使用檢索增強生成（RAG）的技術，RAG 通過整合外部數據來增強基礎模型的回答能力，這不僅能解決模型訓練數據的局限性問題，還可以提供更精準和相關的信息，甚至是回答專業領域問題，以及處理大型文檔任務時，能夠更精準，且不容易發生憑空想像的問題。

RAG基本介紹

RAG 技術是一種結合了搜尋檢索和生成能力的自然語言處理架構，它可以從外部知識庫搜尋相關信息，然後使用這些信息來生成回應或完成特定的 NLP 任務。

RAG 模型主要由兩個部分構成：檢索器和生成器。

1. 檢索器負責從外部知識庫（例如，文本數據庫或預先訓練的知識嵌入）中檢索相關的知識訊息
2. 生成器會利用檢索到的知識來生成回應。
   

透過這個架構，模型可以從大量的文本數據中找到與問題相關的答案，並生成結構化的回應。RAG 技術的優點是利用大量的真實數據，得出更準確及更具有語境的回答。

為什麼需要 RAG

它主要解決大型語言模型處理即時訊息，以及更新資訊的限制:

1. 處理即時或最新信息： 傳統的大型語言模型在特定時間點的數據快照上進行訓練，因此它們僅能回答訓練數據中包含的信息。例如，問一個 LLM 今天的天氣如何，它可能無法回答，因為今天的天氣信息不包含在它的訓練數據中。RAG 透過探索即時天氣的 API 或相關信息來獲得資訊，並在閱讀這些 API 的結果後，生成需要的答案，以彌補處理即時問題的限制。
2. 提供更準確和相關的回答： 通過檢索和整合與用戶查詢相關的外部信息，RAG 能夠使模型提供更準確、更具相關性的答案。這對於需要專業知識或特定領域信息的查詢特別有用，這些專業知識可能來自某些長篇的文件中，那就很適合透過 RAG 去檢索與生成，且生成的內容也會根據此長文，導出精確的回答。
3. 擴展模型的應用範圍： RAG 通過整合不同的數據源（如實時API數據、專業文件等），使模型能夠解答原本超出其訓練範圍的問題，從而擴展了模型的應用範圍。
4. 優化成本和效率： 對於需要處理大量文檔或數據的場景，RAG 可以通過只檢索和處理與特定查詢最相關的信息，來提高處理效率和降低運算成本。

語義檢索（Semantic Search）基本介紹

與RAG緊密相連的另一項關鍵技術是語義檢索。語義檢索技術主要用於從大規模數據集中快速識別出與特定查詢語義上最相關的信息片段。這種技術通過將文本轉換為數值向量（嵌入），並計算這些向量之間的相似度，來確定哪些文本片段與用戶的查詢最為匹配。

如果想讓模型回答關於一本潛水艇手冊（長達492頁）的問題，直接將整本手冊作為上下文提供給模型是不切實際的。這是因為這樣的大量文本會超過模型的處理能力和最大標記限制。此外，由於基於使用的 token 數計費，過長的提示會導致成本增加。

這種情況下，語義檢索（Semantic Search）技術就顯得非常重要。其核心思想是將大型文檔分解成較小的文本片段。這樣不僅使得文本處理和檢索變得更高效，也使整個過程更經濟實用。

具體來說，我們將文本片段轉換成 vector，然後測量這些向量之間的距離，從而識別出與用戶提問在語義上最接近的文本片段。這樣，我們就只需將最相關的文本部分作為上下文提供給模型，而不是整個大型文檔。總而言之，透過將大型文檔分解並使用語義檢索技術，我們可以更高效地利用語言模型來回答複雜和專業的問題，同時控制成本和提升回答的準確性。

實作

RAG 回答 PDF 問題

導入 pdf 文件

def pdf_to_text(path: str,
                start_page: int = 1,
                end_page: Optional[int or None] = None) -> list[str]:
    """
    Converts PDF to plain text.

    Params:
        path (str): Path to the PDF file.
        start_page (int): Page to start getting text from.
        end_page (int): Last page to get text from.
    """
    loader = PyPDFLoader(path)
    pages = loader.load()
    total_pages = len(pages)

    if end_page is None:
        end_page = len(pages)

    text_list = []
    for i in range(start_page-1, end_page):
        text = pages[i].page_content
        text = text.replace('\n', ' ')
        text = re.sub(r'\s+', ' ', text)
        text_list.append(text)

    return text_list
    
text_list = pdf_to_text("pdfs/DQ2.pdf")
print(text_list[:2])

文字預處理

以頁為單位分割，並記錄每個片段的頁碼。

def text_to_chunks(texts: list[str],
                   start_page: int = 1) -> list[list[str]]:
    """
    Splits the text into equally distributed chunks.

    Args:
        texts (str): List of texts to be converted into chunks.
        word_length (int): Maximum number of words in each chunk.
        start_page (int): Starting page number for the chunks.
    """
    text_toks = [t.split(' ') for t in texts]
    chunks = []

    for idx, words in enumerate(text_toks):

        chunk = ' '.join(words).strip()
        chunk = f'[Page no. {idx+start_page}]' + ' ' + '"' + chunk + '"'
        chunks.append(chunk)

    return chunks
    
chunks = text_to_chunks(text_list)

轉換成 vector 使用 text embeddings

應用 hugging dace 的 universal-sentence-encoder_4 模型將文字轉換成 vector

def get_text_embedding(texts: list[list[str]],
                       batch: int = 1000) -> list[Any]:
        """
        Get the embeddings from the text.

        Args:
            texts (list(str)): List of chucks of text.
            batch (int): Batch size.
        """
        embeddings = []
        for i in range(0, len(texts), batch):
            text_batch = texts[i:(i+batch)]
            # Embeddings model
            emb_batch = emb_function(text_batch)
            embeddings.append(emb_batch)
        embeddings = np.vstack(embeddings)
        return embeddings
        
embeddings = get_text_embedding(chunks)

print(embeddings.shape)
print(f"Our text was embedded into {embeddings.shape[1]} dimensions")

可視化 Semantic Search

剛剛轉換成 384 維度的向量，我們使用 t-SNE 算法，降到二維，讓我們可以就圖形做觀察

# Create a t-SNE model
tsne = TSNE(n_components=2, random_state=42)
embeddings_with_question = np.vstack([embeddings, emb_question])
embeddings_2d = tsne.fit_transform(embeddings_with_question)

def visualize_embeddings(embeddings_2d: np.ndarray,
                         question: Optional[bool] = False,
                         neighbors: Optional[np.ndarray] = None) -> None:
    """
    Visualize 384-dimensional embeddings in 2D using t-SNE, label each data point with its index,
    and optionally plot a question data point as a red dot with the label 'q'.

    Args:
        embeddings (numpy.array): An array of shape (num_samples, 384) containing the embeddings.
        question (numpy.array, optional): An additional 384-dimensional embedding for the question.
                                          Default is None.
    """

    # Scatter plot the 2D embeddings and label each data point with its index
    plt.figure(figsize=(10, 8))
    num_samples = embeddings.shape[0]
    if neighbors is not None:
        for i, (x, y) in enumerate(embeddings_2d[:num_samples]):
            if i in neighbors:
                plt.scatter(x, y, color='purple', alpha=0.7)
                plt.annotate(str(i), xy=(x, y), xytext=(5, 2), textcoords='offset points', color='black')
            else:
                plt.scatter(x, y, color='blue', alpha=0.7)
                plt.annotate(str(i), xy=(x, y), xytext=(5, 2), textcoords='offset points', color='black')
    else:
        for i, (x, y) in enumerate(embeddings_2d[:num_samples]):
            plt.scatter(x, y, color='blue', alpha=0.7)
            plt.annotate(str(i), xy=(x, y), xytext=(5, 2), textcoords='offset points', color='black')

    # Plot the question data point if provided
    if question:
        x, y = embeddings_2d[-1]  # Last point corresponds to the question
        plt.scatter(x, y, color='red', label='q')
        plt.annotate('q', xy=(x, y), xytext=(5, 2), textcoords='offset points', color='black')

    plt.title('t-SNE Visualization of 384-dimensional Embeddings')
    plt.xlabel('Dimension 1')
    plt.ylabel('Dimension 2')
    plt.show()

visualize_embeddings(embeddings_2d[:-1])

我們要透過文字片段的距離，來找到可能包含問題答案的片段，這裡的距離選用歐幾里德距離計算相似度與接近性，採用最近鄰居(NearestNeighbors)算法，選擇最接近的 k 個數據，k 定為 5。

nn_2d = NearestNeighbors(n_neighbors=5, metric='cosine')
nn_2d.fit(embeddings_2d[:-1])

neighbors = nn_2d.kneighbors(embeddings_2d[-1].reshape(1, -1), return_distance=False)
neighbors

再將它可視化

visualize_embeddings(embeddings_2d, True, neighbors)

紫色的點是比較靠近的。

使用 Semantic Search找最近的片段

question = '如何查詢成交明細?'
emb_question = emb_function([question])

neighbors = nn.kneighbors(emb_question, return_distance=False)
neighbors

準備 prompt

def build_prompt(question):
    prompt = ""


    prompt += "[INST] <<SYS>>You are an AI assistant tasked with providing answers by summarizing related documents. You should follow these rules:\n"\
              "1. Summarize the content from the provided documents, using the following format:\n"\
              "Topic of the Document: Describe the topic of the document.\n"\
              "Step by Step Instruction: Provide user question-specific instructions or information from the document.\n"\
              "2. If no relevant information is found in the chat history, respond with \"I can't answer the question\".\n"\
              "By adhering to these rules, you will help users find accurate and valuable information.\n"\
              "<</SYS>>\n"

    prompt += 'Search results:\n'

    for c in topn_chunks:
        prompt += c + '\n\n'
    prompt += f"\n\n\nQuery: {question}\n\nAnswer: [/INST]"

    return prompt

prompt = build_prompt(question)
print(prompt)

最後了！ inference

def send_to_watsonxai(prompts,
                    model_name="meta-llama/llama-2-70b-chat",
                    decoding_method="greedy",
                    max_new_tokens=512,
                    min_new_tokens=0,
                    repetition_penalty=1.0
                    ):
    '''
   helper function for sending prompts and params to Watsonx.ai

    Args:
        prompts:list list of text prompts
        decoding:str Watsonx.ai parameter "sample" or "greedy"
        max_new_tok:int Watsonx.ai parameter for max new tokens/response returned
        temperature:float Watsonx.ai parameter for temperature (range 0>2)
        repetition_penalty:float Watsonx.ai parameter for repetition penalty (range 1.0 to 2.0)

    Returns: None
        prints response
    '''


    # Instantiate parameters for text generation
    model_params = {
        GenParams.DECODING_METHOD: decoding_method,
        GenParams.MIN_NEW_TOKENS: min_new_tokens,
        GenParams.MAX_NEW_TOKENS: max_new_tokens,
        GenParams.RANDOM_SEED: 42,
        GenParams.TEMPERATURE: 0,
        GenParams.REPETITION_PENALTY: repetition_penalty,
    }


    # Instantiate a model proxy object to send your requests
    model = Model(
        model_id=model_name,
        params=model_params,
        credentials=creds,
        project_id=project_id)


    for prompt in prompts:
        print(model.generate_text(prompt))

# Example questions for the DQ2:
#   可以怎麼查詢成交狀況?
#   有辦法查詢原物料行情嗎?
#   我想查詢指數的委買跟委賣分析
#   DQ2 可以做到什麼功能?


question = "有辦法查詢原物料行情嗎?"
prompt = build_prompt(question)

send_to_watsonxai(prompts=[prompt], min_new_tokens=1)

小心得

在了解 RAG（檢索增強生成）與語義檢索（Semantic Search）技術後，我發現雖然它們在技術層面上比之前我分享的內容來得複雜，但實際上其核心概念非常明確且直觀。這些技術的出現，主要是為了解決當前大型語言模型（LLM）在處理即時更新資訊或特定領域深度知識方面的局限性。RAG 和語義檢索這些“巧妙的方法”實際上代表了在現有技術條件下的創新與突破。我認為，隨著技術的不斷發展，這些限制很可能會很快被克服。與此同時，這些方法在節約計算資源和提高回答精確度方面的潛力也令我覺得它們非常值得嘗試和深入探究，RAG 和語義檢索不僅是當前問題的有效解決方案，也可能為未來語言模型的發展提供新的思路和靈感，感謝看到這裡的你，我們下次見！

參考

• RAG 檢索增強生成— 讓大型語言模型更聰明的秘密武器. 看完文章後歡迎按鼓勵，訂閱，並分享給所有想知道此類知識的所有人！ | by SimonLiu | Nov, 2023 | InfuseAI
• Universal Sentence Encoder
• Lab watsonx_pdf_rag
• 全端 LLM 應用開發-Day29-Langchain 的 chain