雖然 ChatGPT 已經有相當豐富的知識含量，但還是難免會產生一些事實錯誤或偏差。為了解決這個問題，結合檢索模型 (Retrieval Model) 的做法相當受歡迎。檢索一詞在前幾天的文章不斷被提及，也可見其重要性。所謂的檢索模型就像是一個搜尋引擎，系統根據使用者的問題去尋找相關的文章，並將這些文章一起放進模型輸入裡面，然後要求語言模型根據文章回答問題。
語言模型看著文章回答問題，這種情況是否很眼熟呢？沒錯，就是我們從小到大常常在考場經歷的「閱讀測驗」，在機器學習裡面也被稱為機器閱讀理解 (Machine Reading Comprehension, MRC) 的問題。今天將會來講生成式資訊檢索的基本介紹，並且透過 OpenAI Embedding API & Chat API 來搭建一個 Latex 論文問答機器人的應用。
(Powered By Microsoft Designer)
生成式資訊檢索 (Generative Information Retrieval) 是指透過文字生成模型來進行資訊檢索，主要著重在如何提高檢索的準確度，是一種資訊檢索的分支做法。生成式資訊檢索分成封閉式 (Close Domain) 與開放式 (Open Domain) 兩種，以下介紹兩者的區別。
封閉生成式資訊檢索 (Close Domain Generative IR) 指的是將一個特定領域的知識，例如法律、醫學的相關文章作為訓練資料，將這些訓練資料放進文字生成模型裡面做訓練，並要求模型根據使用者的問題來回答。這樣的做法其實相當直覺，現在 ChatGPT 的知識性問答能力，就像是一種非常大型且泛用的封閉式資訊檢索系統。
但這種系統也有很多問題，首先是模型經常出現「幻覺」，也就是回答錯誤的狀況，俗稱「一本正經的胡言亂語」。而且模型的解釋性並不好，當模型產生錯誤的答案時，會很直接的認為「也許是訓練資料有問題吧！」然後我們就把幾十幾百 GB 的文本資料打開，開始尋找到底是哪份資料出了問題。
除錯的過程就是一個字，痛苦 🤮
其次是這類系統難以將生成的答案與參考的來源做配對，也就是說系統通常不知道這個答案到底來自哪篇文章或哪個網頁，因此實務上，使用者也較難驗證答案的正確性。最後就是模型的知識受限於訓練資料的範圍，即便是特定領域的知識，也會有需要更新資料的問題。那模型要怎麼更新呢？在更新的過程中，以前訓練進去的資訊會被遺忘嗎？這些都是封閉生成式資訊檢索會遇到的挑戰。
開放生成式資訊檢索 (Open Domain Generative IR) 則是挑戰多個領域的知識問答，通常會結合檢索模型或搜尋引擎來回答問題，也就是筆者在簡介介紹的做法。這類做法的關鍵有兩個：
第一點通常由 IR 系統來解決，系統的基本檢索能力、檢索的粒度、是否能跨語言檢索等，都會大幅影響整個問答系統的效果。其次是檢索的速度如何、會不會很佔用記憶體等等，這些相對實務上的問題。第二點則是考驗文字生成模型的能力，模型能夠理解問題嗎？模型知道可以回答問題的文章段落在哪裡嗎？模型能夠產生回答問題的格式嗎？如果資訊不足，模型知道拒絕回答嗎？還是也會亂答一通呢？
在 ChatGPT 問世之前，第一點已經有了非常多很好的解法。例如各種 Embedding 模型像是 Google 的 Universal Sentence Encoder 與微軟的 E5 等等，不僅效果很好，模型權重也有開源，使用方法也滿簡單的。而在 ChatGPT 問世之後，第二點問題也迎刃而解。其強大的上下文理解能力與自然流暢的文字生成，使開放生成式資訊檢索再度成為知識問答的新寵兒。
接下來，我們以 Latex 論文閱讀為題目，設計一個問答機器人。在 arXiv 上面有相當大量的論文文本，部份論文甚至會上傳 Latex 原始碼，可以在論文介紹頁面的右上角點擊 "Other Formats" 查看是否有提供：
如果有提供的話，則會有個 Download Source 的連結可以按：
下載下來通常會是個沒有副檔名的檔案，自己手動加上 .tar.gz 即可。這裡使用 GPT-4 的論文進行示範，以下所有程式碼皆放在 GitHub 上。
註：筆者使用的開發環境為 Ubuntu 22.04 + Python 3.10 版本。
GPT-4 論文的 Latex 原文有近三萬多個 Tokens，無論是基於模型輸入大小的考量，還是荷包的考量，都不太可能將整份文章直接放進模型裡面當成輸入。因此我們會在索引階段 (Index Phase) 將文章切成一塊一塊的區塊 (Chunk)，我們將這些區塊索引起來，在使用階段時方便搜尋。
首先，我們只需要讀取 Latex 文件 (.tex) 即可：
import os
def iter_tex(data_dir):
for dir_path, _, file_list in os.walk(data_dir):
for file_name in file_list:
if not file_name.endswith(".tex"):
continue
full_path = os.path.join(dir_path, file_name)
yield full_path
讀取每份文件的內容並且開始切割，其實這個切 Chunk 的步驟有相當多細節可以講究，但這邊我們先簡單依照換行符號進行切割即可，並且將排版用的雙空格換成單空格：
def get_segments(full_path):
with open(full_path, "rt", encoding="UTF-8") as fp:
text = fp.read().strip()
while " " in text:
text = text.replace(" ", " ")
return text.split("\n")
接著我們使用 tiktoken 套件提供的 Tokenizer 來計算每個 Chunk 有多少 Tokens。因為在 GPT-4 的論文裡面，剛好用到了 Tokenizer 內建的 Special Token，如果直接 Encode 會跳錯誤，所以我們要設定 disallowed_special=() 的參數，可以透過以下程式碼比較差異：
import tiktoken
tk = tiktoken.get_encoding("cl100k_base")
print(tk.encode("<|endofprompt|>"))
# ValueError: Encountered text corresponding to disallowed special token.
print(tk.encode("<|endofprompt|>", disallowed_special=()))
# 當作一般文字來編碼 - [27, 91, 408, 1073, 41681, 91, 29]
print(tk.encode("<|endofprompt|>", allowed_special="all"))
# 當作特殊 Token 來編碼 - [100276]
這個機制是為了避免使用者透過 Prompt Injection 來「越獄」語言模型的設計，因為語言模型通常都對這種 Special Token 非常敏感，當 Special Token 的出現不正常時，就有可能破壞語言模型原本的行為。若非善意使用者掌握了這個弱點，就有可能對模型做一些壞壞的事情，因此妥善處理輸入文本的 Special Token 是相當重要的。
接著，我們撰寫計算文本 Token 數量的函式：
def calc_tokens(tk: tiktoken.Encoding, seg: str):
# disallowed_special=() 會將整份文本都當成一般文字
# 不會將任何 Token 當成 Special Token
tokens = tk.encode(seg, disallowed_special=())
return len(tokens)
到目前為止，我們取得了按照換行切開的 Chunk 與各自的 Token 數量：
for full_path in iter_tex(target_dir):
segments = get_segments(full_path)
segments = [[calc_tokens(tk, seg), seg] for seg in segments]
若我們觀察此 segments 變數，會發現裡面的段落非常零散！作者幾乎三不五時就有一個換行，每個 Chunk 可能不足 100 個 Tokens 在裡面。如此零散的 Chunk 會造成上下文語意被嚴重截斷，效果通常不會很好。因此我們這裡需要做一些合併處理，將零散的小 Chunk 合併為一個大的 Chunk，但又不會大到模型塞不下。
那我們該如何決定一個 Chunk 有多大呢？這便涉及到成本的考量。首先筆者預計使用 gpt-3.5-turbo 模型，因此輸入上限不會超過 4097 個 Tokens。但因為筆者的魔法小卡承載不住這 4000 個 Tokens 的重量，我希望每個 Request 平均只需要消耗 2000 個 Tokens 就好。
接著我們開始分配這 2000 個 Token 要怎麼用：
在這些條件下，我們推估 Chunk Size 為 300 左右，這些設定可以根據模型能力與可負擔的成本而有變化。另外有些模型雖然可以支援到很高的輸入上限，但是太多或太長的 Chunk 模型未必處理的來，有時反而會因為雜訊太多而影響回答的品質！因此這也是需要去實驗與考量的一點。
接下來我們就按照剛剛算出來的 Chunk Size 來設計區塊合併的演算法：
def process_segments(segments: list[tuple[int, str]], chunk_size):
print(f"Original Segments: {len(segments)}")
i = 0
while i + 1 < len(segments):
# 取得當前 Chunk 與下個 Chunk 的長度與內容
seg1_len, seg1_txt = segments[i]
seg2_len, seg2_txt = segments[i + 1]
# 若兩個 Chunk 長度相加小於 chunk_size 則合併
if seg1_len + seg2_len < chunk_size:
segments[i][0] = seg1_len + seg2_len
segments[i][1] = seg1_txt + "\n" + seg2_txt
segments.pop(i + 1) # 移除已被合併的 Chunk
# 若 Chunk Size 超過上限則開始處理下一個
else:
i += 1
print(f"Processed Segments: {len(segments)}")
return [seg[1].strip() for seg in segments]
合併完 Chunk 之後，可以將 Chunk 拿出來觀察是否正確：
# 用來取得隨機檔名
from tempfile import NamedTemporaryFile as NTF
def dump_segments(segments):
with NTF("wt", dir=".", delete=False) as fp:
print(fp.name)
for i, seg in enumerate(segments):
fp.write(f"=== Chunk {i} Begin ===\n")
fp.write(f"{seg}\n")
fp.write(f"=== Chunk {i} End ===\n\n")
千萬不要小看這個觀察 Chunk 的動作，有時候在那邊看模型怎麼都吐不出正確答案，回頭一看才發現 Chunk 根本沒切好之類的。所以至少要看個一眼，確認格式正常沒有變亂碼怎樣的，再進行下一步。
然後我們就可以將這些切好的 Chunks 往 OpenAI Embedding API 身上全部砸下去！
import numpy as np
def create_embeddings(chunks):
resp = openai.Embedding.create(
model="text-embedding-ada-002",
input=chunks,
)
embs = [item["embedding"] for item in resp["data"]]
embs = np.array(embs)
print(f"Embedding Shape: {embs.shape}")
return embs
其實實務上是不建議這樣做，只是因為筆者在發懶 ((躺
還記得我們提過 OpenAI API 的 Rate Limit 嗎？其中 Ada Embedding API 有三十五萬個 Tokens 的每分鐘上限，因此發送 Request 之前最好還是檢查一下並分次發送。但因為 GPT-4 論文只有三萬多個 Tokens，所以一次全部發出去是沒問題的。
最後，將 Chunks 與 Embeddings 存起來，完成我們的索引階段：
import json
def dump_data(chunks, embs, data_dir):
with open(f"{data_dir}/chunks.json", "wt", encoding="UTF-8") as fp:
json.dump(chunks, fp, ensure_ascii=False)
np.save(f"{data_dir}/embs.npy", embs)
因為文章是固定的，所以切 Chunk 以及取 Embedding 的動作就不需要再做一次。這邊我們只簡單將 Chunks 用 JSON 格式存起來，Embeddings 的部份也是用 Numpy 隨便存。但這種做法是相對沒有效率的，實務上建議使用一些資料庫系統來管理這些資料。
以上索引階段的完整程式碼放在此連結。
完成索引後，就可以開始查詢階段 (Query Phase) 的應用。首先將方才存下來的 Chunks 與 Embeddings 讀取出來，並且透過 Faiss 套件建立 Embedding 的索引。
pip install faiss-cpu
Faiss 是 Facebook 製作的一個向量搜尋引擎，可以很有效率的處理大規模向量搜尋的問題，並且用法相當簡單：
import json
import numpy as np
from faiss import IndexFlatL2
def load_data(chunk_path, emb_path):
with open(chunk_path, "rt", encoding="UTF-8") as fp:
chunks = json.load(fp)
key_emb: np.ndarray = np.load(emb_path)
vectors = IndexFlatL2(key_emb.shape[1])
vectors.add(key_emb)
return chunks, vectors
我們讀出來的 key_emb 第一個維度是資料量，第二個維度是 Embedding Size，需要先透過 Embedding Size 來初始化 faiss 的 IndexFlatL2 類別，然後將這些 Embedding 加進去做索引。
接下來，我們透過 Embedding API 取得使用者輸入的 Embedding：
def get_query_emb(query_text: str):
resp = openai.Embedding.create(
model="text-embedding-ada-002",
input=[query_text],
)
query_emb = resp["data"][0]["embedding"]
return np.array([query_emb])
然後根據 Query Embedding 尋找前五個相似的 Embeddings，並根據他們的 Index 來建立我們的 Prompt，這個過程可以透過 IndexFlatL2.search 輕鬆完成：
def build_prompt(q_emb, q_text, vectors: IndexFlatL2, value_chunks):
dist, index = vectors.search(q_emb, k=5)
print(f"Distance: {dist}")
print(f"Indices: {index}")
prompts = [value_chunks[i] for i in reversed(index[0])]
prompts.append(f"問題：{q_text}")
return "\n\n".join(prompts)
在 IndexFlatL2.search 裡面指定 k=5 代表我們只需要取前五名相似的向量。理論上，最有可能包含答案的文章，通常是相似度最高（也就是距離最短）的文章。因此會讓最相近的文章最靠近問題，所以這裡使用了 reversed 將 Top-K Chunks 反過來排。最後將使用者的問題放在 Prompt 最底下，所有 Chunks 之間以雙換行隔開，完成完整的 Prompt 建構。
最後把這個 Prompt 整份往 ChatGPT API 身上砸下去！
def create_chat(prompt):
sys_prompt = "你現在是個專業的文件檢索問答機器人，請根據文件內容的資訊回答問題。"
return openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[
{"role": "system", "content": sys_prompt},
{"role": "user", "content": prompt},
],
stream=True,
)
將取得的 Response 用串流的方式輸出：
def stream_response(response):
for resp in response:
try:
token = resp["choices"][0]["delta"]["content"]
print(end=token, flush=True)
except:
pass
print()
然後開始欣賞產生的結果：
好欸，看起來結果挺不錯的！以上查詢階段的完整程式碼放在此連結。
接下來，我們對這個系統做一些簡單的分析，瞭解這種做法的優缺點為何。首先是文章內包含的資訊：
此對話連結為直接詢問 ChatGPT (GPT-3.5) 的回答，在沒有文件支援的情況下，ChatGPT 直接否定了 GPT-4 有理解圖片的能力。
那這個系統可以回答沒有提到的資訊嗎？
在沒有搜索到結果或者沒有提及該資訊的情況下，模型不會開始胡亂回答也是個重要的能力。
這樣的系統還有一個很大的侷限，就是我們沒有加入太多 Chunk 的 Metadata 在裡面，例如這個 Chunk 來自哪份文件？位在文件的哪個位置？所以如果使用者詢問與位置相關的問題，就會無法回答：
其次是 Latex 為自動編排章節，所以透過 Latex 原始碼也無從得知章節次序：
以上兩個問題必須透過更多工程上的做法來解決，例如讓 Chunk 附帶 Metadata 資訊，告知 ChatGPT 這個 Chunk 的所在位置，而且也要能讓 Embedding 查詢時可以查到這個段落。後者則可能要加入 Latex Parser 來推論章節的次序是什麼，並且一同加入 Metadata 裡面。
筆者也嘗試分析了一篇比較新的論文 (LongLoRA) 以確保 ChatGPT 完全沒有看過這篇論文，詳細的回答可以參考這份連結。
以上所有程式碼與結果紀錄皆放在 GitHub 上。
今天探討了生成式 IR 的原理，瞭解現在 IR 系統如何與 LLM 做結合，並且簡單實做了一個 Open Domain 的論文問答系統。即便是個簡單的實做，也有相當理想的效果，可以說 ChatGPT 大幅降低了這類問答系統的門檻。但也因為這個系統相對簡單，有許多細節沒有處理，因此也有不少侷限，例如無法詢問頁數資訊、章節資訊相關的問題等等。但至少開發者可以專注在優化檢索的環節，而不再需要煩惱生成的部份了。
歡樂的時光總是過的特別快，按照筆者的規劃，這是最後一篇會用到 ChatGPT API 的文章了。接下來會開始談論 Local LLM 的部份，因為當 LLM 談到商用時，隱私與資安將會是非常大的問題，無論是客戶還是法規都對這塊有很大的限制。像 OpenAI API 這種雲端服務，對小型開發者來說非常友善，但是對商業用戶來說就不太可行了。因此將 LLM 部署到本地機器勢必是個必須採取的手段，但也有許多技術上的問題需要克服。
因為未來的文章將會著重在 Local LLM 上，例如各種推論框架、量化技術和訓練方法等，所以就在這邊跟 OpenAI API 說聲再會啦～
iThome鐵人賽
看影片追技術