Try   HackMD

RAG實戰 - 以COSCUP問答bot為例

tags: RAG

本文章會分享筆者在參與COSCUP的問答bot的開發經驗。

目標

COSCUP是由台灣開放原始碼社群推動的年度研討會，每年都有眾多志工投入籌備工作。由於COSCUP擁有豐富的文件，但志工們經常需要翻閱過往資料，這為工作增加了負擔。本專案旨在減輕志工的工作量，利用RAG（Retrieval Augmented Generation）技術打造一個AI bot，以便將部分文件查閱工作轉化為與bot互動的形式。

image
bot使用範例

本文大致分為三個部分

  1. 背景介紹
  2. 我們採用的方法
  3. 未來想用但還沒實現的方法

RAG背景介紹

RAG (Retrieval-Augmented Generation)旨在提供LLM生成回答的方法，使其能夠在回答問題時，不僅依賴於模型本身的知識，還可以檢索相關的外部資料，提供更準確且相關性更高的答案。相比單純使用LLM生成答案，RAG能夠有效降低模型的幻覺（hallucination）問題，並提升回答的可靠性。

以下為一個經典的RAG流程

image

https://github.com/chatchat-space/Langchain-Chatchat

首先需要把文件存入資料庫中，通常會使用向量資料庫(Vector Store)，當使用者提出一個問題時，RAG 首先會從大型數據庫中找到與問題相關的文本片段，然後將這些片段作為上下文輸入到生成模型中，生成最終的回答。

其中會有幾個關鍵

  1. chunking方法: 如何把文件切塊
  2. model: 選擇embedding model
  3. 更新的策略

由於我們的文件大多是繁體中文，因此LLM選擇breeze-7b。

langchain

langchain為一個框架，使用langchain有優點但也有缺點。

優點
能夠非常快速的將基本RAG流程搭建起來，example，並且可以快速的與langsmith結合，langsmith能提供清楚明瞭的log。

缺點
由於高度抽象化導致後續維護變複雜，許多文章也在討論是否需要使用langchain，如這篇文章。

Document Source

我們採用了三種來源

  1. 本地文件(local document)
  2. github repo
  3. mattermost 聊天紀錄

mattermost為類似slack的通訊軟體

並且支援4種格式: md, txt, docx, pdf

本地文件

我們在langchain提供的不同loader建立了一個class，能夠指定資料夾並載入支援的檔案格式。

github文件

我們只抓github倉庫內的md文件，直接使用GithubFileLoader。

必須設置token。
https://github.com/settings/tokens?type=beta

from langchain_community.document_loaders.github import GithubFileLoader
loader = GithubFileLoader(
    repo=<Repo name>,
    access_token=<ACCESS_TOKEN>,
    github_api_url="https://api.github.com",
    file_filter=lambda file_path: file_path.endswith(
        ".md"
    ),
)
documents = loader.load()

本地文件與github repo的性質類似，都是將一個長文檔抓下來。然而，在處理聊天訊息時，我們需要將分散的訊息組合成段落，且聊天訊息的資訊密度通常較低。為了確保內容的準確性，我們選擇抓取特定人物在特定頻道的發言，並向前後各延伸抓取三則相關訊息，以此組合成完整的段落。

Mattermost 訊息

訊息會組成如下的對話段落，並存入資料庫內。

A: blabla
B: blabla
C: blabla
TARGET: XXXX
A: blabla
B: blabla
C: blabla

抓取mattermost訊息需要設置token，並且只能抓取已加入的頻道。
https://api.mattermost.com/

完整code: https://github.com/COSCUP/polyhistor/blob/main/vectorDB/read_mattermost.py

因為username是額外一個api，所以使用了cache機制讓同樣的id不需要抓第二次username。

Chunking

Chunking 是指將文件分成多個文字段落並存入資料庫中。由於這些切分後的文字段落會被納入後續的 prompt，因此其長度至關重要。過長的段落可能導致超出 LLM的輸入限制，而過短的段落則可能無法提供足夠的有效信息。

目前chunking方法能分為5個levels:

  1. 字符切分（Character Splitting）：將文檔按字符數量切分。
  2. 遞歸字符文本切分（Recursive Character Text Splitting）：基於分隔符列表遞歸式切分。
  3. 文檔特定切分（Document Specific Splitting）：針對不同類型的文檔（如 PDF、Python、Markdown）使用多種切分方法。
  4. 語義切分（Semantic Splitting）：基於embedding切分。
  5. 代理式切分（Agentic Splitting）：使用LLM進行文本切分。

chunking方法能夠同時採用，意即同一文件可經過不同的chunking方法存在資料庫內

本專案只使用到Document Specific Splitting

Character Splitting

單純的使用字符長度切分，並且可設置重疊長度。

from langchain_text_splitters import CharacterTextSplitter
text_splitter = CharacterTextSplitter(chunk_size = 35, chunk_overlap=5, separator='', strip_whitespace=False)

Recursive Character Text Splitting

在 Character Splitting 上加入分隔符號的判斷，預設會判斷

  • "\n\n"
  • "\n"
  • " "
from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(chunk_size = 65, chunk_overlap=0)

Document Specific Splitting

對不同文檔採用不同的方式，例如: md文件使用header

from langchain_text_splitters import MarkdownHeaderTextSplitter
headers = [("#", "Header 1"), ("##", "Header 2"), ("###", "Header 3")]
text_splitter = MarkdownTextSplitter(headers, strip_headers=False)

以下只介紹概念

Semantic Splitting

由於embedding能代表語意資訊，若能以句子為單位，利用前後句的相關性為chunking段落依據，當前後關聯差距過大時，就為分為不同的chunk。

image
Source

langchain實作: https://python.langchain.com/v0.2/docs/how_to/semantic-chunker/

Agentic Splitting

持續地問LLM現在需不需要分段，甚至是將原文摘要等等。
這會耗費非常多的token。

由於這兩個方法會使用到model，會花費相對多的時間。

Embedding model

Embedding model會將切分好的段落轉變成embedding，接著存入資料庫內。而embedding的好壞會直接影響Retriever的結果，因此選擇一個好的embedding model是非常重要的。

由於我們的文件大多為繁體中文，因此我們從C_MTEB的leaderboard上挑選測試的model：

  • dztech/bge-large-zh:v1.5
  • herald/dmeta-embedding-zh
  • milkey/m3e
  • chevalblanc/acge_text_embedding

Evaluation method

  1. 利用手上的文檔人工產出30道問題，產生 問題-出處的對應
  2. 測試各個model檢索結果

我們採用的指標為Recall，測試找到多少正確出處，並且為了貼近實際的使用情境，一道問題可能會對應多個出處。

image
https://en.wikipedia.org/wiki/Precision_and_recall

Result

Model Recall
dztech/bge-large-zh:v1.5 0.702
herald/dmeta-embedding-zh 0.729
milkey/m3e 0.810
chevalblanc/acge_text_embedding 0.837

根據測試出來的結果，我們選擇chevalblanc/acge_text_embedding為embedding model。

我們的測試方法只有驗證文件出處，而非確切的段落，因此有可能會出現檢索到正確的文件，但是是錯誤的段落。

MultiQuery retriever

我們採用MultiQuery retriever方法來檢索文檔，在傳統retriever領域中，query expansion與qeury rewriting是非常常用的技術，這兩種技術都是為了提高query的檢索正確性，而現在我們能使用 LLM 來做這個步驟，如下圖，我們能利用 LLM 將一個 query 擴增到數個與原 query 相似的query，保證檢索結果的多樣性。

image
https://medium.com/@kbdhunga/advanced-rag-multi-query-retriever-approach-ad8cd0ea0f5b

此方法在langchain已經有實作。
docs

我們選擇自己實作，並分為 multiqueryChain 與 answerchain

multiqueryChain

from langchain.load import dumps, loads
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

from src.utils.model import llm_model


def reciprocal_rank_fusion(results: list[list], k=60):
    fused_scores = {}
    for docs in results:
        for rank, doc in enumerate(docs):
            doc_str = dumps(doc)
            if doc_str not in fused_scores:
                fused_scores[doc_str] = 0
            fused_scores[doc_str]
            fused_scores[doc_str] += 1 / (rank + k)

    reranked_results = [(loads(doc), score) for doc, score in sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)]
    return reranked_results


def multiqueryChain(retriever, model):

    template = """
        You are a helpful assistant that generates multiple search queries based on a single input query.
        Generate multiple search queries related to: {original_query}
        The generated 4 queries should contain the original query and be related to the original query.
        OUTPUT (4 queries):
    """
    prompt = ChatPromptTemplate.from_template(template)
    model = llm_model(model)

    generate_queries = prompt | model | StrOutputParser() | (lambda x: x.split("\n"))
    chain = generate_queries | retriever.map() | reciprocal_rank_fusion
    return chain

我們定義了一個處理multiqueryChain結果的function

def parse_fusion_results(results, max_results=5) -> dict:
    content = []
    source_dict = set()
    metadata_rank = []

    for res in results[:max_results]:
        content.append(res[0].page_content)
        post_id = res[0].metadata.get("post_id")
        source = f"https://chat.coscup.org/coscup/pl/{post_id}" if post_id else res[0].metadata["source"]

        if source not in source_dict:
            source_dict.add(source)

            if source.startswith("http"):
                metadata_rank.append(source)
            else:
                metadata_rank.append(source.split("/")[-1])

    return {"metadata": metadata_rank, "context": "\n\n".join(content)}

answerChain

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

from src.utils.model import llm_model


def answerChain(model):
    template = """你是一位COSCUP的工作人員，請利用下面的資訊回答問題，回答時需遵守以下幾點：
    1. 回答需確保資訊正確、完整
    2. 使用繁體中文回答
    3. 回答需清晰、易懂
    4. 回答需盡量簡潔
    5. 回答需盡量符合問題
    6. 回答需使用台灣用語

    $$$$$$
    問題: {original_query}
    $$$$$$
    {context}

    $$$$$$
    回答:
    """
    prompt = ChatPromptTemplate.from_template(template)
    model = llm_model(model)
    chain = prompt | model | StrOutputParser()

    return chain

接著將兩個chain合併

我們使用的vectorstore為Qdrant

from langchain.chains import TransformChain
from langchain_core.runnables import RunnableParallel, RunnablePassthrough

retriever = vectorstore.as_retriever(search_kwargs={"k": 3, "score_threshold": 0.5}, search_type="mmr")
answerchain = answerChain(model=config.model.llm)

def retrieval_transform(inputs: dict) -> dict:
    multiquerychain = multiqueryChain(retriever=retriever, model=config.model.llm)
    fused_results: dict[str, list] = parse_fusion_results(multiquerychain.invoke({"original_query": inputs["original_query"]}))

    return fused_results

retrieval_chain = TransformChain(input_variables=["original_query"], output_variables=["metadata", "context"], transform=retrieval_transform)

rag_chain = (
    RunnableParallel(original_query=RunnablePassthrough())
    .assign(multi_query=retrieval_chain)
    .assign(context=lambda x: x["multi_query"]["context"], metadata=lambda x: x["multi_query"]["metadata"])
    .assign(answer=answerchain)
    .pick(["context", "metadata", "answer"])
)

out = rag_chain.invoke(data.query)

Update strategy

當文檔數量過多時，如果每次更新文件時都需要重新轉成embedding再存入資料庫內，會非常的耗時。因此我們會偵測檔案是否有變動，當文件更新後，我們才會再把文檔切分並存入。

整體流程如下：

image

我們會利用檔案的路徑與檔案內容的hash值來做是否存入的根據。檔案路徑因為資料庫本身就會存，因此實作相當容易，但hash值計算必須自己實現。

目前只有實現本地文件與github repo的策略，本地文件需要自己計算hash值如下，github有提供SHA可供比對。

import hashlib

def compute_file_hash(file_path: str):
    """Compute the SHA-256 hash of the given file.

    Args:
        file_path (str): Path to the file to hash.

    Returns:
        str: The computed hash value of the file.
    """
    hash_algo = hashlib.sha256()
    try:
        with open(file_path, "rb") as f:
            while chunk := f.read(8192):
                hash_algo.update(chunk)
    except IOError as e:
        print(f"Error reading file {file_path}: {e}")
    return hash_algo.hexdigest()

Future Works

  • Ensemble retrieval
    • 混和BM25與現有retriever
  • More document source
    • 抓取Hackmd、網站等
  • Semantic Cache
    • 利用前人問題答案當作下次生成答案的參考

Acknowledge

特別感謝BOB、小畢、Katy、jimmy與iris對本專案的貢獻。本專案是一個開源項目，歡迎有興趣的朋友加入我們的開發行列。

Last changed by 
 
jeff3071
0
81
Published on HackMD