---
# System prepended metadata

title: RAG技術實戰 - 以COSCUP問答bot為例
tags: [RAG]

---

# RAG實戰 - 以COSCUP問答bot為例

###### tags: `RAG`

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


## 目標

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

![image](https://hackmd.io/_uploads/B1vj612q0.png)
*bot使用範例*

---

本文大致分為三個部分

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


## RAG背景介紹

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

以下為一個經典的RAG流程
![image](https://hackmd.io/_uploads/Hki6Cy3cR.png)

*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](https://python.langchain.com/v0.1/docs/get_started/quickstart/)，並且可以快速的與langsmith結合，langsmith能提供清楚明瞭的log。


**缺點**
由於高度抽象化導致後續維護變複雜，許多文章也在討論是否需要使用langchain，如[這篇文章](https://www.octomind.dev/blog/why-we-no-longer-use-langchain-for-building-our-ai-agents)。

## 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

```python
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()
```

:::info
本地文件與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

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

```python
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"
- " "

```python
from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(chunk_size = 65, chunk_overlap=0)
```

### Document Specific Splitting

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

```python
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](https://hackmd.io/_uploads/BJrFTgnc0.png)
*[Source](https://github.com/FullStackRetrieval-com/RetrievalTutorials/blob/main/tutorials/LevelsOfTextSplitting/5_Levels_Of_Text_Splitting.ipynb)*


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

### Agentic Splitting

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

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

## Embedding model

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

由於我們的文件大多為繁體中文，因此我們從[C_MTEB](https://github.com/FlagOpen/FlagEmbedding/tree/master/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檢索結果



:::info
我們採用的指標為Recall，測試找到多少**正確**出處，並且為了貼近實際的使用情境，一道問題可能會對應多個出處。
![image](https://hackmd.io/_uploads/rykEHc25C.png)
*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。

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

## MultiQuery retriever

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

![image](https://hackmd.io/_uploads/Sks0oF39C.png)
*https://medium.com/@kbdhunga/advanced-rag-multi-query-retriever-approach-ad8cd0ea0f5b*

此方法在langchain已經有實作。
[docs](https://python.langchain.com/v0.1/docs/modules/data_connection/retrievers/MultiQueryRetriever/)

---

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

**multiqueryChain**
```python
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

```python
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**

```python
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
> 
```python
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](https://hackmd.io/_uploads/rJ9_-93qC.png)


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

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

```python
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對本專案的貢獻。本專案是一個開源項目，歡迎有興趣的朋友加入我們的開發行列。

- [專案連結](https://github.com/COSCUP/polyhistor)
- [專案討論頻道](https://chat.coscup.org/coscup/channels/project-polyhistor)