來自工業界的知識庫 RAG 服務(三)，FinGLM 競賽獲獎項目詳解

背景介紹

前面介紹過工業界的 RAG 服務 QAnything 和 RagFlow 的詳細設計，也介紹過來自學術界的一些優化手段。

前一陣子剛好看到智譜組織的一個金融大模型比賽 FinGLM，主要做就是 RAG 服務的競賽，深入研究了其中的幾個獲獎作品，感覺還是有不少亮點。整理一些獲獎項目的設計方案，希望對大家有所啟發。

FinGLM 比賽

FinGLM 比賽介紹

FinGLM 是基于一定數量的上市公司財報構建知識庫，使用 ChatGLM-6B 作為大模型完成知識庫問答。需要回答的問題包含三類：

初級問題，可以直接從原文中獲得信息進行回答，比如直接問特定公司某一年的研發費用，考察的是能否正確檢索到內容的能力；
中級問題，需要對原文中內容進行統計分析和關聯，比如問某公司某一年研發費用的增長率，考慮的是能否檢索到內容并進行二次加工得到結果的能力；
高級問題，安全開放的問題，比如問研發項目是否涉及國家戰略，考察的是檢索到內容并綜合處理的能力；

可以看到使用一個相對小的大模型 ChatGLM-6B，需要能準確回答上面的這些問題，對于內容的檢索以及架構精細設計要求還是很高的，直接使用最原始的 RAG 框架肯定是不夠的, 發揮不穩定的向量檢索大概率是無法幫你獲獎的。

比賽難點

我根據決賽答辯過程中的一些反饋，整理此比賽中的一些難點：

財報中包含大量的數據，摻雜文本內容與表格數據，很多精細的問題都需要依賴表格進行回答，如何進行精細的處理，保證原始文檔中的內容可以正確檢索到；
不同類型的問題需要不同的處理方案，如何區分不同的問題進行有針對性的解決；
ChatGLM-6B 模型較弱，稍微復雜的情況就無法正確處理，甚至模型的輸出就不可控了，如果保證穩定輸出正確的答案；
用戶問題與文檔使用中詞匯可能不是完全一致的，這會導致精確的檢索更加困難；

獲獎項目介紹

在天池的決賽文章可以看到最終獲獎的隊伍，本文主要想介紹的是項目是決賽獲得第三名的 “ChatGLM反卷總局” 團隊的項目，為什么沒有選擇前面的團隊的項目，主要原因是：“ChatGLM反卷總局” 是獲獎作品中性價比最高的實現方案：

其他團隊或多或少都做了模型的微調，甚至獲得第一名團隊微調了 2 到 3 個模型，“ChatGLM反卷總局” 沒有做任何的微調就取得了不錯的成績；
在原始 chatGLM 做關鍵詞提取效果很差的情況下，其他團隊都是靠微調解決，“ChatGLM反卷總局” 基于原始模型 + 正則表達式就實現了不錯的效果；
實現方案比較輕量，沒有引入太多額外的服務，生產環境可以方便應用；

項目方案詳解

方案設計

項目設計的整體流程如下所示：

請添加圖片描述

整理流程主要包含三個核心模塊：

問題分類，不同類型的問題需要采取的方案完全不同，在選擇處理方案時需要先確定是什么類型的問題，方便進行有針對性的處理；
關鍵詞抽取，因為內容的檢索主要使用關鍵詞檢索，因此需要從原始問題中提取對應的關鍵詞；
表格與文檔內容的檢索與回答，根據問題類型的不同從不同來源獲取數據，并執行必要的外部處理（主要是額外的數學計算），處理后提供給大模型得出最終結果；

問題分類

不同類型的問題存在不同的處理方案，因此在回答前需要先進行分類，分類的正確性對結果影響很大。分類一般是根據實際的測試問題進行歸納總結得出類型和判斷依據, 之后就可以自動化分類。這部分不少團隊是通過微調 chatGLM 模型實現線上分類的。

此項目是基于簡單的規則判斷，這個只能算是一個取巧方案，沒有太多可說的東西。可以簡單看看對應的實現：

if '保留' in q['question'] and com_match != '':question_type = 'calculate'
elif com_match != '' and len(year_match) > 0 and ('分析' in q['question'] or '介紹' in q['question'] or '如何' in q['question'] or '描述' in q['question'] or '是否' in q['question'] or '原因' in q['question'] or '哪些' in q['question'] or len(com_normal_keywords) > 0) and '元' not in q['question'] and len(com_info_keywords) == 0:question_type = 'com_normal'
elif com_match != '' and len(year_match) > 0:question_type = 'com_info'
elif com_match == '' and judge_tongji(q['question']):question_type = 'com_statis'
else:question_type = 'normal'q['question_type'] = question_type
if '增長率' in q['question']:q['question_type'] = 'com_info'

代碼有點糙，應該是根據實際問題總結出來的。但是在實際生產環境中，如果問題分類可以通過簡單規則直接區分出來，確實沒必要做大模型的微調或復雜的意圖識別了。

關鍵詞抽取

常規的 RAG 服務一般是基于向量進行檢索，但是此項目主要使用的是關鍵詞進行檢索，支持了關鍵詞 + 向量的混合檢索，貌似決賽還因為鏡像過大最終沒有使用向量檢索，因此主要依賴的就是關鍵詞檢索。

而關鍵詞抽取的效果直接影響關鍵詞檢索的效果，因此這一步相當重要，此項目使用的兩種方案：

正則關鍵詞抽取；
基于 LLM 的關鍵詞抽取

同一內容在問題中的使用的關鍵詞可能與文檔中使用的關鍵詞存在一些差異，因此還需要進行關鍵詞的泛化。

正則關鍵詞提取

正則關鍵詞的提取完全是通過已有的關鍵詞表進行匹配得到的，效果優劣與原始的關鍵詞表的覆蓋范圍有很大關系。具體的實現簡化如下：

# 各個來源的關鍵詞列表匯總find_keywords = list(attr_mapping_title.keys() | gongshi_mapping.keys() | com_normal_attr_mapping_title.keys())
find_keywords.sort(key=len, reverse=True)# 構建正則表達式，可以匹配關鍵詞表中存在的關鍵詞attr_regex_pattern = r'(?:' + '|'.join(find_keywords) + r')'
attr_regex = re.compile(attr_regex_pattern, re.IGNORECASE)# 從原始問題 q 中提取關鍵詞，并實施去重attr_match = attr_regex.findall(q['question'])
attr_match.extend(attr_regex.findall(q['question'].replace('的','')))
keywords = list(set(attr_match))

這個是通過簡單的匹配覆蓋常規的關鍵詞提取，主打的就是快，即使無法命中也有 LLM 提取關鍵詞兜底。

基于 LLM 的關鍵詞提取

常規的 ChatGLM-6B 模型較小，抽取關鍵詞的效果很可能不佳。常規方案是構造訓練數據進行微調，但是此項目用了一種有意思的方案：

項目使用 Few Shot 去提升 ChatGLM-6B 的提取關鍵詞的效果，與常規 Few Shot 不同在于，構造的 Few Shot 是放在歷史聊天記錄中的，而不是拼接在 prompt 中的。

對應的實現簡化后如下所示：

cls_history = [("現在你需要幫我完成信息抽取的任務，你需要幫我抽取出句子中三元組，如果沒找到對應的值，則設為空，并按照JSON的格式輸出", '好的，請輸入您的句子。'),("<year><company>電子信箱是什么?\n\n提取上述句子中的關鍵詞，并按照json輸出。", '{"關鍵詞":["電子信箱"]}'),("根據<year>的年報數據，<company>的公允價值變動收益是多少元?\n\n提取上述句子中的關鍵詞，并按照json輸出。",'{"關鍵詞":["公允價值變動收益"]}'),("<company>在<year>的博士及以上人員數量是多少?\n\n提取上述句子中的關鍵詞，并按照json輸出。",'{"關鍵詞":["博士及以上人員數量"]}'),("<company><year>年銷售費用和管理費用分別是多少元?\n\n提取上述句子中的關鍵詞，并按照json輸出。",'{"關鍵詞":["銷售費用","管理費用"]}'),("<company><year>的衍生金融資產和其他非流動金融資產分別是多少元？\n\n提取上述句子中的關鍵詞，并按照json輸出。",'{"關鍵詞":["衍生金融資產","其他非流動金融資產"]}'),...
]prompt = f'{question}\n\n提取上述句子中的關鍵詞，并按照json輸出。'response, history = model.chat(tokenizer, prompt, history=cls_history, top_p=0.7, temperature=1.0)

通過這種方案，最終大模型提取關鍵詞的表現更穩定，下面的團隊給出的對比情況：

在這里插入圖片描述

關鍵詞泛化

因為原始問題中的關鍵詞與實際文檔中的關鍵詞不是完全一致的，此時就無法正確匹配到文檔中內容，因此需要執行必要的泛化，從而提升匹配的概率。

項目是通過 fuzzywuzzy 實現的，此項目目前已經遷移至 thefuzz 了，思路是通過 Levenshtein_distance 實現模糊的字符串匹配，其實就是基于編輯距離確定字符串的相似度。

項目中的關鍵詞泛化的實現如下所示：

from fuzzywuzzy import processdef find_best_match_new(question, mapping_list, threshold=40):# 與系統中關鍵詞列表通過編輯距離進行匹配matches = process.extract(question, mapping_list)# 使用閾值進行過濾，并按照相似度進行排序best_matches = [match for match in matches if match[1] >= threshold]best_matches = sorted(best_matches, key=lambda x: (-x[1], -len(x[0])))# 返回最匹配的內容和得分match_score = 0total_q = ""if len(best_matches) > 0:total_q = best_matches[0][0]match_score = best_matches[0][1]return total_q, match_score

數據庫與文檔內容的檢索與回答

表格數據檢索與回答

表格數據是在預處理階段提取出來，保存在 excel 文件中，初始化時轉換為 pandas 對象進行檢索。

通過前面的關鍵詞提取與泛化后，得到的關鍵詞與文檔中的關鍵詞就保持一致了，此時通過關鍵詞就可以準確地匹配所需的內容。這部分就可以看到精確匹配的優勢所在，處理得當的情況下結果相對準確。

除了簡單直接匹配表格數據的情況，實際問題中還存在需要通過公式計算最終結果的情況，是否能直接提供原始數據給 chatGLM-6B 進行計算呢，這個就有點強 GPT 所難了，實際測試往往容易出錯。

目前是通過外部提取所需的原始后直接調用 Python 進行計算，具體的實現如下所示：


financial_formulas = {"研發經費與利潤比值": (["研發費用", "凈利潤"], "研發費用 / 凈利潤"),"企業研發經費占費用": (["銷售費用", "財務費用", "管理費用", "研發費用"], "研發費用 / (研發費用+管理費用+財務費用+銷售費用)"),"研發人員占職工": (["研發人員", "職工人數"], "研發人員 / 職工人數"),"碩士及以上學歷人員占職工": (["研發人員", "職工人數"], "研發人員 / 職工人數"),"流動比率": (["流動資產合計", "流動負債合計"], "流動資產合計 / 流動負債合計"),"速動比率": (["流動資產合計", "存貨", "流動負債合計"], "(流動資產合計 - 存貨) / 流動負債合計"),"碩士及以上人員占職工": (["碩士以上人數", "職工人數"], "碩士以上人數 / 職工人數"),"研發經費與營業收入比值": (["研發費用", "營業收入"], "研發費用 / 營業收入"),...
}# 根據關鍵詞獲得所需的公式項formula_data = financial_formulas[index_name]
data_values = {}# 獲取計算所需的原始數據finance_data = dq.get_financial_data(year_, stock_name)# 將原始數據與公式所需的元素組合為鍵值對for field in formula_data[0]:value = finance_data.get(field)data_values[field] = value# 獲取對應的公式計算字符串formula_str = formula_data[1]
calculation_str = formula_str.replace(" ", "")
for key, value in data_values.items():calculation_str = calculation_str.replace(key, str(value))# 調用 Python 執行計算result_value = eval(calculation_str)

文本數據檢索與回答

對于文本檢索，大量的信息會存在文本的各級標題中，而常規的文件分片中除了與標題直接相連的分片，其他分片中標題的信息是缺失的，效果類似如下所似：
在這里插入圖片描述

為了解決這個問題，項目會通過正則表達式識別標題行，之后通過堆棧記錄標題層級結構，在各個分片中增加標題信息，實現流程如下所示：

在這里插入圖片描述
通過上面的修復，最終各個分片中都會包含各級標題的信息，檢索更容易命中。修復后分片效果如下所示：

在這里插入圖片描述
分片的數據是保存在 ES 中的，這樣就可以比較方便的實現關鍵詞檢索，實際檢索時關鍵詞命中標題或文本內容都能被正確召回。效果如下所示：

實際的文本召回實現就相對簡單了，只是 ES 客戶端的簡單調用：

def get_context(company, final_year, query, size=3, es_index="tianchi", keyword="", recall_titles={}, title_keyword=""):force_search_body = {"size": size,"_source": ["texts"],"query": {"bool": {"filter": [{"term": {"companys": company}},{"terms": {"year": final_year}},]}},}# 構造 query 檢索force_search_body["query"]["bool"]["should"] = [{"match": {"texts": {"query": query}}}]# 額外的關鍵詞檢索增強if len(keyword) > 0:force_search_body["query"]["bool"]["should"].append({"match_phrase": {"texts": {"query": keyword}}})force_search_body["query"]["bool"]["filter"].append({"terms": {"titles_cut.keyword": recall_titles[keyword]}})if len(title_keyword) > 0:force_search_body["query"]["bool"]["should"].append({"match": {"titles_cut": {"query": title_keyword}}})# ES 檢索調用search_result = es.search(index=es_index, body=force_search_body)hits = search_result["hits"]["hits"]recall_texts = [hit["_source"]["texts"] for hit in hits]return recall_texts