# RAG實戰 - 以COSCUP問答bot為例
###### tags: `RAG`
本文章會分享筆者在參與COSCUP的問答bot的開發經驗。
## 目標
COSCUP是由台灣開放原始碼社群推動的年度研討會，每年都有眾多志工投入籌備工作。由於COSCUP擁有豐富的文件，但志工們經常需要翻閱過往資料，這為工作增加了負擔。本專案旨在減輕志工的工作量，利用RAG（Retrieval Augmented Generation）技術打造一個AI bot，以便將部分文件查閱工作轉化為與bot互動的形式。
*bot使用範例*
---
本文大致分為三個部分
1. 背景介紹
2. 我們採用的方法
3. 未來想用但還沒實現的方法
## RAG背景介紹
RAG (Retrieval-Augmented Generation)旨在提供LLM生成回答的方法，使其能夠在回答問題時，不僅依賴於模型本身的知識，還可以檢索相關的外部資料，提供更準確且相關性更高的答案。相比單純使用LLM生成答案，RAG能夠有效降低模型的幻覺（hallucination）問題，並提升回答的可靠性。
以下為一個經典的RAG流程
*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。
*[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，測試找到多少**正確**出處，並且為了貼近實際的使用情境，一道問題可能會對應多個出處。
*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，保證檢索結果的多樣性。
*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再存入資料庫內，會非常的耗時。因此我們會偵測檔案是否有變動，當文件更新後，我們才會再把文檔切分並存入。
整體流程如下：
我們會利用檔案的路徑與檔案內容的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)
Sign in with Wallet
Connect another wallet