手搓RAG新增功能：遞歸檢索與迭代查詢+重回成熟框架API
在上那篇提到的我手搓的那個 RAG 項目新增功能中，漏掉了遞歸檢索與迭代查詢，這篇補上（源碼見知識星球）。經過初步調試對召回效果有明顯提升，這種方法解決了傳統 RAG 的幾個關鍵問題：
- 處理復雜多步驟問題：通過多次迭代，分解復雜問題
- 信息不足的補充：當初始檢索結果不足以回答問題時，自動生成補充查詢
- 多角度信息收集：能夠從不同角度收集相關信息
1、遞歸檢索具體實現
遞歸檢索函數（recursive_retrieval）
（支持最多三次迭代查詢）
每次迭代使用混合檢索（向量檢索+BM25）獲取信息
使用 LLM 分析當前檢索結果，判斷是否需要進一步查詢
如果需要，LLM 會生成新的查詢問題，用于下一輪檢索
換句話說，遞歸檢索的工作原理可以理解為"先檢索-后思考-再檢索"的過程，模擬了人解決問題的方式：先獲取一些信息，思考下是否足夠，如果不夠則繼續查找更多相關信息?？傊玫慕Y果不是一蹴而就的。
為了更直觀的讓大家了解這個遞歸檢索的過程，貼幾張本地測試圖片，供參考。需要說明的是，目前這個版本中還沒做多模態的預處理和圖片的召回，后續結合具體案例更新在知識星球中。
2、核心組件全解析
下面再對整個代碼的核心組件做進一步的解析，大家可以在這個基礎上按照業務需求進一步優化：
2.1文檔處理和向量化
文本提?。菏褂?pdfminer.high_level.extract_text_to_fp 從 PDF 提取文本內容
文本分塊：使用 RecursiveCharacterTextSplitter 將長文本分割成更小的塊（在代碼中塊大小設為 400 字符，重疊為 40 字符）
向量化：使用 all-MiniLM-L6-v2 模型為每個文本塊生成嵌入向量
存儲：將文本塊、向量和元數據存儲到 ChromaDB 向量數據庫
輔助索引：同時構建 BM25Okapi 索引用于基于關鍵詞的檢索
分塊粒度和嵌入質量直接影響檢索性能，這部分大家需要結合自己的文檔結構特點自行調整，上述列出的做法僅供參考。
2.2混合檢索機制
結合了兩種不同的檢索方法：
語義向量檢索：基于嵌入向量的相似度搜索，能夠捕捉語義關系
BM25 關鍵詞檢索：基于詞頻的經典檢索算法，專注于關鍵詞匹配
混合策略：通過 hybrid_merge 函數融合兩種檢索結果，使用α參數(0.7)控制兩種方法的權重
這種混合策略結合了語義理解和關鍵詞匹配的優勢，提高了檢索的綜合表現。
2.3重排序機制
檢索到初步結果后，使用更精確的模型進行重排序：
交叉編碼器重排序：使用 CrossEncoder 模型，能夠同時考慮查詢和文檔內容進行更精確的相關性評分
LLM 重排序：可選使用 LLM 對查詢和文檔的相關性進行評分，利用大模型的理解能力
緩存機制：使用@lru_cache 減少重復計算，提高效率
重排序步驟使檢索結果更符合用戶的實際需求，大幅提升了檢索質量。
2.4遞歸檢索與迭代查詢
這是新增的一個功能，也是實測下來對最終效果有明顯提升的一點，由 recursive_retrieval 函數實現：
初始檢索：使用原始查詢進行首輪檢索
結果分析：使用 LLM 分析當前檢索結果是否充分回答問題
查詢改寫：如果信息不足，LLM 會生成一個新的、更具體或從不同角度的查詢
迭代過程：使用新查詢繼續檢索，直到獲取足夠信息或達到最大迭代次數
這個機制解決了單次檢索的局限性，能夠處理復雜問題、多步驟推理，以及需要從多個角度收集信息的場景。
2.5生成回答
系統支持兩種回答生成方式：
本地 Ollama 模型：使用 deepseek-r1:1.5b 或 deepseek-r1:7b 模型在本地生成回答
SiliconFlow API：調用云端 API 使用更強大的模型生成回答
https://cloud.siliconflow.cn/i/cmXGS5IJ
思維鏈處理：支持分離思考過程，以可折疊的形式展示給用戶
之所以選擇新增一個商業api選項，也是因為我自己的電腦跑本地模型是在太卡，問了一些復雜問題容易觸發超時機制。當然另外還有個重要原因是，在針對不同核心組件做調優時，保持 chat 模型的水準個人實踐下來也很重要，否則就變成了過度雕花。
3、優化方向參考
因為只是方便大家練手，大家感興趣的可以在目前代碼基礎上做如下優化：
3.1文檔格式支持擴展
當前系統僅支持 PDF 文件，可擴展支持：
# 支持更多文檔格式
def extract_text_from_file(file_path):
"""根據文件類型選擇適當的提取方法"""
ext = os.path.splitext(file_path)[1].lower()
if ext == '.pdf':
return extract_text_from_pdf(file_path)
elif ext == '.docx':
return extract_text_from_docx(file_path)
elif ext == '.pptx':
return extract_text_from_pptx(file_path)
elif ext in ['.txt', '.md', '.py', '.java', '.js', '.html', '.css']:
with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
return f.read()
elif ext in ['.csv', '.xlsx', '.xls']:
return extract_text_from_spreadsheet(file_path)
else:
raise ValueError(f"不支持的文件類型: {ext}")3.2高級分塊策略
可以實現更智能的分塊策略：
語義分塊：基于段落、章節或自然語義邊界進行分塊
結構感知分塊：識別文檔結構（標題、列表、表格等）
多粒度分塊：同時維護不同粒度的塊，靈活應對不同類型的查詢
# 多粒度分塊示例
def create_multi_granularity_chunks(text):
"""創建多粒度分塊"""
# 大塊(1000字符)：適合概述類問題
large_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000, chunk_overlap=100
)
# 中塊(400字符)：平衡粒度
medium_splitter = RecursiveCharacterTextSplitter(
chunk_size=400, chunk_overlap=40
)
# 小塊(150字符)：適合精確問答
small_splitter = RecursiveCharacterTextSplitter(
chunk_size=150, chunk_overlap=20
)
large_chunks = large_splitter.split_text(text)
medium_chunks = medium_splitter.split_text(text)
small_chunks = small_splitter.split_text(text)
return {
"large": large_chunks,
"medium": medium_chunks,
"small": small_chunks
}3.3嵌入模型優化
當前系統使用 all-MiniLM-L6-v2，可以考慮：
更強大的多語言模型：如 multilingual-e5-large 提升中文理解能力
領域特定微調：針對特定領域數據微調嵌入模型
多模型集成：使用多個嵌入模型并融合結果
# 多模型嵌入示例
def generate_multi_model_embeddings(text):
"""使用多個模型生成嵌入并融合"""
model1 = SentenceTransformer('all-MiniLM-L6-v2')
model2 = SentenceTransformer('paraphrase-multilingual-mpnet-base-v2')
# 生成嵌入
emb1 = model1.encode(text)
emb2 = model2.encode(text)
# 簡單融合策略：歸一化后拼接
emb1_norm = emb1 / np.linalg.norm(emb1)
emb2_norm = emb2 / np.linalg.norm(emb2)
# 返回融合嵌入
return np.concatenate([emb1_norm, emb2_norm])3.4檢索與重排序改進
可以考慮以下改進：
查詢擴展：使用同義詞擴展或關鍵詞擴展增強原始查詢
密集檢索與稀疏檢索結合：更高級的混合檢索方法
上下文感知重排序：考慮已檢索文檔間的關系進行重排序
多輪對話記憶：在多輪對話中利用歷史上下文改進檢索
# 查詢擴展示例
def expand_query(query):
"""使用LLM擴展查詢，生成多個角度的查詢變體"""
prompt = f"""
為以下查詢生成3個不同角度的變體，以提高檢索召回率：
原始查詢: {query}
變體應當包含同義詞替換、關鍵詞擴展或問題重構。
僅返回三個變體，每行一個，不要有額外解釋。
"""
response = call_llm_api(prompt)
variants = [line.strip() for line in response.strip().split('\n')]
return [query] + variants # 返回原始查詢和變體3.5其他可能的擴展方向
結果驗證機制：使用外部知識或規則驗證生成結果的準確性
用戶反饋學習：根據用戶反饋調整檢索策略和參數
多模態支持：處理圖像、表格等非文本內容
層次化索引：構建文檔的層次化表示，從概覽到細節
個性化調整：根據用戶歷史查詢和偏好調整檢索策略
4、RAG 學習路線全梳理
最后回到學習路線上來，最開始手搓這個 demo，也是自己為了更好的從底層理解 RAG 系統，不曾想收到了很多正反饋。
https://github.com/weiwill88/Local_Pdf_Chat_RAG
另外正如上篇所說，直接使用封裝度高的框架容易讓人"知其然不知其所以然"。如果你已經掌握了基礎原理，就可以考慮進一步探索 RAGFlow 或者 LlamaIndex 等成熟框架的 Python API，充分利用成熟框架提供的生產力優勢。按照個人近幾個月的學習和實踐經驗，整體學習路線參考如下：
4.1從自建到框架的過渡期
對比學習：將手搓實現的組件與成熟框架中對應功能進行比較
框架源碼閱讀：嘗試閱讀 LlamaIndex 或 RAGFlow 等框架的核心源碼，看看他們如何實現相同功能
增量替換：逐步用框架組件替換自建組件，對比性能和結果差異
4.2成熟框架深入學習
從示例入手：優先研究官方提供的示例代碼和教程
構建微型項目：使用框架 API 實現小型但完整的應用
實驗各種組件：測試不同的檢索器、嵌入模型、重排序策略等
4.3高級應用與定制
探索高級功能：如多模態 RAG、Agent 集成、自定義索引等
性能優化：學習如何調整參數提高檢索質量和速度
領域適配：根據特定行業或領域需求定制 RAG 系統
5、Python API vs Web 界面能做什么？
公眾號之前有篇文章詳細介紹了 RAGFlow 官方 Python API 各個模塊使用，后續有些盆友私信我問到，這些 api 除了可以批量的按照自定義方式處理文件外，還可以干啥？結合近期學習和實踐，提供以下五個方向供大家參考，具體參考示例代碼：
5.1系統集成能力
# 與現有系統集成示例
from ragflow import RAGPipeline
import your_company_database
# 從公司數據庫獲取文檔
documents = your_company_database.get_latest_documents()
# 使用RAGFlow處理
pipeline = RAGPipeline()
for doc in documents:
pipeline.add_document(doc)
# 將結果同步回公司系統
processed_results = pipeline.get_indexed_status()
your_company_database.update_document_status(processed_results)5.2定制化檢索策略
# 自定義混合檢索策略
from ragflow import RetrievalPipeline
from ragflow.retrievers import VectorRetriever, KeywordRetriever
# 創建自定義的檢索器組合
def custom_hybrid_retriever(query, top_k=10):
# 使用向量檢索獲取語義相關結果
semantic_results = vector_retriever.retrieve(query, top_k=top_k)
# 使用關鍵詞檢索獲取精確匹配
keyword_results = keyword_retriever.retrieve(query, top_k=top_k)
# 自定義融合策略（比如考慮文檔及時性等）
results = custom_merge_function(semantic_results, keyword_results,
recency_weight=0.2)
return results5.3高級處理流程自動化
# 創建復雜的數據處理和RAG工作流
from ragflow import DocumentProcessor, Indexer, QueryEngine
import schedule
import time
def daily_knowledge_update():
# 從多個來源獲取新文檔
new_docs = fetch_documents_from_sources()
# 文檔預處理（去重、格式轉換等）
processed_docs = DocumentProcessor().process_batch(new_docs)
# 增量更新索引
indexer = Indexer()
indexer.update_incrementally(processed_docs)
# 生成日報
report = generate_daily_summary()
send_email(report)
# 設置定時任務
schedule.every().day.at("01:00").do(daily_knowledge_update)
while True:
schedule.run_pending()
time.sleep(60)5.4自定義評估和優化
# 構建RAG系統評估框架
from ragflow import RAGEvaluator
import matplotlib.pyplot as plt
# 準備測試數據集
test_queries = load_test_queries()
ground_truth = load_ground_truth_answers()
# 測試不同的配置
configurations = [
{"embeddings": "minilm", "chunk_size": 200, "retriever": "hybrid"},
{"embeddings": "e5-large", "chunk_size": 500, "retriever": "vector"},
{"embeddings": "bge-large", "chunk_size": 300, "retriever": "hybrid"}
]
results = {}
for config in configurations:
# 構建并測試每種配置
rag_system = build_rag_with_config(config)
eval_results = RAGEvaluator.evaluate(
rag_system, test_queries, ground_truth
)
results[str(config)] = eval_results
# 可視化不同配置的性能比較
plot_evaluation_results(results)5.5領域特定的增強
# 醫療領域RAG增強示例
from ragflow import RAGPipeline
from custom_medical_modules import MedicalTermNormalizer, DiseaseEntityExtractor
# 創建領域特定的文檔處理器
medical_processor = DocumentProcessor()
medical_processor.add_transformer(MedicalTermNormalizer())
medical_processor.add_transformer(DiseaseEntityExtractor())
# 構建醫療特定的RAG系統
medical_rag = RAGPipeline(
document_processor=medical_processor,
retrieval_augmentors=[
CitationRetriever(), # 添加醫學文獻引用
EvidenceLevelClassifier() # 分類證據等級
],
response_formatters=[
MedicalDisclaimerFormatter(), # 添加醫療免責聲明
CitationFormatter() # 格式化引用
]
)當然，還有結合多個框架使用的示例，我在Medium看到一個博主使用 LlamaIndex 整合多源數據并構建索引，然后通過 RAGFlow 的深度解析和重排序優化結果。最后利用 RAGFlow 的引用功能增強答案可信度，同時調用 LlamaIndex 的代理進行動態數據補充。不過，我還沒試過，感興趣的可以測試下看看。
RAGFlow 等框架的實際業務價值，不同實踐經驗的盆友可能有不同的體會，個人總結其實就是一句話，使用這些框架的 Python API 可以讓我們快速構建原型并迭代改進，比從 0-1 開發節省了大量時間。從而專注于解決業務問題，而非技術細節。
