多模態RAG賽題實戰之策略優化--Datawhale AI夏令營

科大訊飛AI大賽（多模態RAG方向） - Datawhale

項目流程圖

1、升級數據解析方案：從 fitz?到 MinerU

PyMuPDF（fitz）是基于規則的方式提取pdf里面的數據；MinerU是基于深度學習模型通過把PDF內的頁面看成是圖片進行各種檢測，識別的方式提取。

（1）識別表格 ：將表格轉化為結構化的Markdown或JSON格式。

（2）提取圖片 ：對文檔中的圖片進行識別。

（3）圖片描述 ：（可選）調用多模態模型為提取出的圖片生成文字說明。

這會為后續的RAG流程提供包含表格和圖片信息的、更豐富且更精確的上下文，是解決多模態問題的關鍵舉措。

基礎方案所使用的 fitz 工具僅能提取文本，會遺漏表格、圖片等關鍵信息。
MinerU 的優勢：對PDF進行深度的版面分析，除了能更精準地提取文本塊外，還具備以下功能：
因此轉而使用 mineru_pipeline_all.py 腳本，具體操作如下：

?# 建議在GPU環境下運行，執行完大約需要1.5h
python mineru_pipeline_all.py

報錯：系統網絡無法訪問外網 huggingface.co，導致 Mineru 無法下載所需的 PDF 處理模型。

解決方法：在 mineru_pipeline_all.py 文件開頭添加環境變量設置：

os.environ['MINERU_MODEL_SOURCE'] = "modelscope"

這將使 Mineru 從 ModelScope（阿里云模型庫）下載模型，避免因網絡問題無法訪問 Hugging Face，下載到的是OpenDataLab的MinerU2.0-2505-0.9B模型。

mineru_pipeline_all.py 文件中關鍵的有兩個函數依次執行：

1、parse_all_pdfs 函數——分析 PDF 的版面布局，識別出里面的文本、標題、表格和圖片，然后把這些識別出的所有內容元素，連同它們的類型、位置、層級等信息，都存進一個名為 _content_list.json 的文件里。

2、process_all_pdfs_to_page_json 函數——讀取 _content_list.json ，先按頁碼把內容分好組，然后逐個處理每一頁里的內容項，里面內嵌了一個item_to_markdown 函數，這個函數是一個轉換器，它根據內容項的類型（ text、table、image）來決定如何轉換成MarkDown格式，而且代碼會檢查圖片本身有沒有自帶的文字描述（ caption ），如果沒有，并且我們允許進行視覺分析（ enable_image_caption=True ），它就會調用一個多模態大模型（代碼里指定的是 Qwen/Qwen2.5-VL-32B-Instruct ）來給這張圖片生成一段描述。

MinerU的輸出：會是三種類型——text、table、image的集合（相比于fitz則完全是text，對圖表也是提取text，完全丟失了圖的表意）

    {"type": "text","text": "分析師： 彭波  \nE-mail： pengbo@yongxingsec.com  \nSAC編號： S1760524100001  \n分析師： 陳燦  \nE-mail： chencan2@yongxingsec.com  \nSAC編號： S1760525010002  \n相關報告：  \n《伏美替尼持續放量，適應癥拓展仍  \n有空間》2025 年 05 月 06 日","page_idx": 0},{"type": "table","img_path": "images/e615166fd385400608ed9069eb20088bb78ce2c5de7fad66da7072570597386f.jpg","table_caption": [],"table_footnote": [],"table_body": "<table><tr><td colspan=\"2\">基本數據</td></tr><tr><td>07月04日收盤價(元)</td><td>94.61</td></tr><tr><td>12mthA股價格區間(元)</td><td>39.82-99.99</td></tr><tr><td>總股本(百萬股)</td><td>450.00</td></tr><tr><td>無限售A股/總股本</td><td>100.00%</td></tr><tr><td>流通市值 (億元)</td><td>425.75</td></tr></table>","page_idx": 0},{"type": "image","img_path": "images/cefb4046fc651be250334d71e02a2c9289c2dc5420d575183f79eb329cdb68d5.jpg","image_caption": ["最近一年股票與滬深 300比較","資料來源：Wind，甬興證券研究所"],"image_footnote": [],"page_idx": 0},

table：

image：原圖和提取的圖

但是目前mineru只是提取出來了圖片，還需要對圖片進行進一步的融合，只是簡單加入圖片的描述信息還有有比較大的局限性的。

2、升級分塊策略

目前的分塊策略：

按頁來分塊（每一頁都是一個知識塊，可以直接用于后續的向量化和索引），每一個pdf文件按頁來排序，每一頁的內容包含text、table、image，上一個pdf最后一頁結束之后，便是下一個pdf的第一頁，以此類推直到最后一個pdf的最后一頁。

上述分塊方式存在缺點：按“頁”分塊過于粗暴，一個完整的表格或邏輯段落可能被硬生生切開，或者說當本來應檢索的信息分布于前后兩頁之中時，便破壞了信息的上下文完整性。

優化分塊策略：

有了 MinerU 精細化的解析結果，我們可以對圖片進行進一步的內容解釋，添加圖片的描述信息。

后續涉及對圖像描述信息的融合處理。

3、引入重排模型

在終端下載BAAI的bge-reranker-v2-m3重排模型：

# 先下載 lfs
git lfs install
git clone https://www.modelscope.cn/BAAI/bge-reranker-v2-m3.git

加載重排模型：

# 初始化 FlagReranker（加載一次就行）
local_model_path = "./bge-reranker-v2-m3"  # 替換為你的下載模型路徑
self.reranker = FlagReranker(local_model_path,use_fp16=True # 沒 GPU 用 "cpu"
)

召回+重排實現代碼：取的是先召回后重排得到的Top-k個chunks，代替原來的直接取的Top-k個chunks。

        # 1?? 向量粗召回 15 個q_emb = self.embedding_model.embed_text(question)retrieved_chunks = self.vector_store.search(q_emb, top_k=15)if not retrieved_chunks:return {"question": question,"answer": "","filename": "","page": "","retrieval_chunks": []}# 2?? 用 FlagReranker 精排pairs = [[question, chunk['content']] for chunk in retrieved_chunks]scores = self.reranker.compute_score(pairs)  # 返回每個pair的相關性分數# 綁定分數for i, sc in enumerate(scores):retrieved_chunks[i]['score'] = sc# 按分數排序，取 top_kreranked_chunks = sorted(retrieved_chunks, key=lambda x: x['score'], reverse=True)[:top_k]# 3?? 拼接上下文context = "\n".join([f"[文件名]{c['metadata']['file_name']} [頁碼]{c['metadata']['page']}\n{c['content']}"for c in reranked_chunks])# 4?? 構造 Promptprompt = (f"你是一名專業的金融分析助手，請根據以下檢索到的內容回答用戶問題。\n"f"請嚴格按照如下JSON格式輸出：\n"f'{{"answer": "你的簡潔回答", "filename": "來源文件名", "page": "來源頁碼"}}'"\n"f"檢索內容：\n{context}\n\n問題：{question}\n"f"請確保輸出內容為合法JSON字符串，不要輸出多余內容。")# 5?? 調用大模型client = OpenAI(api_key=qwen_api_key, base_url=qwen_base_url)completion = client.chat.completions.create(model=qwen_model,messages=[{"role": "system", "content": "你是一名專業的金融分析助手。"},{"role": "user", "content": prompt}],temperature=0.2,max_tokens=1024)

4、升級索引策略?

多路召回與融合：

除了原先的基于向量的語義檢索——使用 embedding 模型來查找意思相近的chunk之外，另外再引入一種基于關鍵詞的檢索方法——BM25 算法，它擅長匹配問題中出現的具體詞語，即要將chunk中的content內容的每一個單詞/字給分出來，再去做匹配。

step1：下載BM25算法庫和中文分詞器

uv pip install rank_bm25, jieba

step2：在SimpleVectorStore類中新增 BM25 算法關鍵詞檢索函數，利用中文分詞器（jieba）去對中文句子進行分詞

class SimpleVectorStore:def __init__(self):self.embeddings = []self.chunks = []# --- 新增 ---self.bm25 = None  # BM25 模型self.tokenized_chunks = []  # 預先分好詞的文本def add_chunks(self, chunks: List[Dict[str, Any]], embeddings: List[List[float]]):self.chunks.extend(chunks)self.embeddings.extend(embeddings)# --- 新增：構建 BM25 ---# 使用 jieba 精確分詞self.tokenized_chunks = [list(jieba.cut_for_search(c['content']))  # 搜索引擎模式，速度快for c in self.chunks]self.bm25 = BM25Okapi(self.tokenized_chunks)def search(self, query_embedding: List[float], top_k: int = 3) -> List[Dict[str, Any]]:from numpy import dotfrom numpy.linalg import normimport numpy as npif not self.embeddings:return []emb_matrix = np.array(self.embeddings)query_emb = np.array(query_embedding)sims = emb_matrix @ query_emb / (norm(emb_matrix, axis=1) * norm(query_emb) + 1e-8)idxs = sims.argsort()[::-1][:top_k]return [self.chunks[i] for i in idxs]# --- 新增：bm25檢索 ---def search_bm25(self, query: str, top_k: int = 3) -> List[Dict[str, Any]]:if not self.bm25:return []# 同樣用 jieba 分詞tokens = list(jieba.cut_for_search(query))scores = self.bm25.get_scores(tokens)idxs = scores.argsort()[::-1][:top_k]return [self.chunks[i] for i in idxs]

step3：在 SimpleRAG 類中新增混合檢索接口

class SimpleRAG:def __init__(self, chunk_json_path: str, model_path: str = None, batch_size: int = 32):self.loader = PageChunkLoader(chunk_json_path)self.embedding_model = EmbeddingModel(batch_size=batch_size)self.vector_store = SimpleVectorStore()self.memory = ConversationBufferMemory(return_messages=True)def search_hybrid(self, question: str, top_k_vec: int = 10, top_k_bm25: int = 10) -> List[Dict[str, Any]]:"""混合檢索：向量 + BM25，各取 top_k，合并去重后返回"""# 向量檢索q_emb = self.embedding_model.embed_text(question)vec_results = self.vector_store.search(q_emb, top_k=top_k_vec)# BM25 檢索bm25_results = self.vector_store.search_bm25(question, top_k=top_k_bm25)# 合并去重（保持順序）seen = set()merged = []for chunk in vec_results + bm25_results:cid = (chunk['metadata']['file_name'], chunk['metadata']['page'], chunk['content'])if cid not in seen:seen.add(cid)merged.append(chunk)return merged

step4：函數應用，修改原來search方法為混合檢索

# chunks = self.vector_store.search(q_emb, top_k)
# 2. 混合檢索
chunks = self.search_hybrid(rewritten_question)

5、反思重寫：

我們甚至可以考慮讓RAG系統擁有自我修正的能力。

具體來說，就是讓系統在檢索一次之后，能自己判斷一下找到的上下文夠不夠回答問題。

如果不夠，它可以自己生成一個新的、更具體的查詢語句，再次進行檢索，把兩次的結果合在一起再生成答案。

這會讓整個問答過程更動態一些。