沒有捷徑:RAG入門不推薦直接使用成熟框架
春節期間我在 Github 開源的 RAG 項目目前已經攢了 134 個 Star,盲猜可能也是因為最開始用的就是 Ollama 本地部署 DeepSeek-r1:7b 的方案,年后當本地部署知識庫和 deepseek火了起來之后,被動蹭了一波流量。
1、為什么重復造輪子?
但是,在過去的一個月時間里也收到了很多網友的私信,詢問關于為什么市面上已經有了類似 AnythingLLM、Cherry Studio、Dify、RAGFlow 等成熟的開源框架,還要重復造輪子去編一個不是很好用的 RAG 項目。
當然與此同時,也有很多網友在私信或者評論區中反饋上手調試過我這個簡單的開源項目后,再去用其他框架更加得心應手了。
1.1 開箱即用的”不友好“
其實我從 24 年 6 月份開始,就逐一深度試用了上述常見的幾個開源項目,但一段時間之后明顯發現,作為一個剛入門的人來說,雖然 AnyThingLLM、RAGFlow 等成熟框架提供了便捷的"開箱即用"體驗,但直接使用這些工具會讓人陷入"知其然,而不知其所以然"的困境。
1.2 從零構建再到框架應用
換句話說,就像學編程不應該從框架開始,而是應該從基礎語法入手一樣。學習 RAG 技術實測也同樣適合先構建基礎認知框架,再應用封裝工具。這不僅是技能學習的路徑,更多也是培養解決問題能力的過程。
2、項目特點與局限性
2.1 項目優勢
我這個開源項目采用簡潔的代碼展示了 RAG 的完整流程,通過親手調試這些組件,可以:
更有助于建立具象認知
比如直觀理解文本分塊如何影響語義完整性,檢索策略如何決定召回質量,以及重排序如何提升最終回答準確度
掌握核心決策點
親身體驗不同 chunk_size 對檢索效果的影響,感受不同嵌入模型的語義表達差異,理解為什么相同的 RAG 系統在不同場景下表現迥異
培養調試直覺
當回答質量不理想時,可以通過控制變量法能相對準確判斷是分塊策略不當,還是檢索精度不足,或是提示工程欠缺
掌握這些基礎后,再轉向 RAGFlow 或其他框架時,眾多的配置選項就不再是很抽象的參數,再做 API 調優或者二次開發也會變得有的放矢。從而深度適配業務場景時,也就可以針對性地做調整框架配置,而不是非盲目嘗試。
2.2 項目局限性
分塊策略
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=200,
chunk_overlap=20
)
chunks = text_splitter.split_text(text)
使用 LangChain 的 RecursiveCharacterTextSplitter
塊大小:每塊 200 字符(較小,一個字符相當于 1 個漢字)
重疊度:20 字符
特點:小塊大小有利于精準定位信息,但可能損失上下文連貫性
檢索策略
results = COLLECTION.query(
query_embeddings=query_embedding,
n_results=5,
include=['documents', 'metadatas']
)
純語義檢索(沒有混合 BM25 等關鍵詞檢索)
未使用過濾條件或高級查詢功能
重排策略
代碼中沒有顯式的重排序步驟,但設計了一些相關邏輯:
# 檢測矛盾
conflict_detected = detect_conflicts(sources_for_conflict_detection)
# 獲取可信源
if conflict_detected:
credible_sources = [s for s in sources_for_conflict_detection
if s['type'] == 'web' and evaluate_source_credibility(s) > 0.7]
通過 evaluate_source_credibility 函數對信息源進行可信度評分
在檢測到沖突時,優先考慮可信度高的來源
3、進階版本
前幾天開通知識星球后,有些有一定實踐經驗的網友過來交流一些技術細節時發現,基礎版本的開源項目已經不能滿足他們當前的需求。
我花了半天時間重構了遍代碼,相比上一版多了600行,也是敲到手酸。這個進階版屬于更符合 RAG 系統最佳實踐的完整版本,一共有10 個大類的優化要點,下述展示會按照對最終回答質量的影響程度排序。需要說明的是,優先實施以下前 3-5 項將能帶來最顯著的效果提升。
注:下述代碼優化示例是針對前述我提到的自己開源項目而言
3.1 分塊策略優化(最高優先級)
200 字符的塊過小,無法包含足夠上下文
可能導致語義割裂和信息碎片化
改進方案:
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=800, # 增大到800-1000字符
chunk_overlap=150, # 增加重疊以保持連貫性
separators=["\n\n", "\n", "。", ",", " "] # 優先在自然段落分割
)
預期效果:更連貫的上下文,減少信息丟失,提高回答質量和相關性。
3.2 混合檢索策略
當前問題:
純語義檢索可能忽略關鍵詞匹配
容易出現語義漂移
改進方案:
# 引入BM25關鍵詞檢索
from rank_bm25 import BM25Okapi
tokenized_corpus = [doc.split() for doc in corpus]
bm25 = BM25Okapi(tokenized_corpus)
# 混合檢索結果
semantic_results = COLLECTION.query(query_embeddings, n_results=7)
bm25_results = bm25.get_top_n(query.split(), corpus, n=7)
combined_results = hybrid_merge(semantic_results, bm25_results)
預期效果:提高檢索準確性,尤其對事實性和專業術語的問題。
3.3 重排序機制
當前問題:
缺乏真正的重排序機制
相關度評分單一依賴向量相似度
改進方案:
# 方案A: 使用交叉編碼器重排序
from sentence_transformers import CrossEncoder
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
candidate_pairs = [[query, doc] for doc in retrieved_docs]
scores = reranker.predict(candidate_pairs)
reranked_results = [doc for _, doc in sorted(zip(scores, retrieved_docs), reverse=True)]
# 方案B: 使用LLM進行相關性評分
def llm_rerank(query, docs, llm_client):
results = []
for doc in docs:
prompt = f"問題: {query}\n文檔: {doc}\n這個文檔與問題的相關性是否高? 0-10分打分:"
score = float(llm_client.generate(prompt).strip())
results.append((score, doc))
return [doc for _, doc in sorted(results, reverse=True)]
預期效果:更精準的結果排序,將最相關的內容放在前面,顯著提高答案質量。
3.4 遞歸檢索與迭代查詢
當前問題:
單次檢索可能不足以回答復雜問題
缺乏根據初步檢索結果調整查詢的機制
改進方案:
def recursive_retrieval(initial_query, max_iteratinotallow=3):
query = initial_query
all_contexts = []
for i in range(max_iterations):
# 當前查詢的檢索結果
current_results = retrieve_documents(query)
all_contexts.extend(current_results)
# 使用LLM分析還需要查詢什么
next_query_prompt = f"""
基于原始問題: {initial_query}
以及已檢索信息: {summarize(current_results)}
是否需要進一步查詢? 如果需要,請提供新的查詢問題:
"""
next_query = llm_client.generate(next_query_prompt)
if "不需要" in next_query or i == max_iterations-1:
break
query = next_query
return all_contexts
預期效果:能夠處理多跳推理問題,循序漸進獲取所需信息。
3.5 上下文壓縮與總結
當前問題:
檢索文檔過長時浪費 token
包含過多無關信息
改進方案:
def compress_context(query, documents, max_tokens=2000):
compressed_docs = []
for doc in documents:
# 方案A: 使用map-reduce模式總結
summary_prompt = f"原文: {doc}\n請提取與問題'{query}'最相關的信息,總結在100詞以內:"
compressed = llm_client.generate(summary_prompt)
compressed_docs.append(compressed)
# 方案B: 使用抽取式摘要選擇關鍵句子
# sentences = split_into_sentences(doc)
# scores = sentence_similarity(query, sentences)
# top_sentences = [s for s, _ in sorted(zip(sentences, scores), key=lambda x: x[1], reverse=True)[:5]]
# compressed_docs.append(" ".join(top_sentences))
return compressed_docs
預期效果:減少無關信息干擾,降低 token 消耗,提高模型對關鍵信息的關注度。
3.6 提示工程優化
當前問題:
現有提示模板相對簡單
缺乏針對不同問題類型的專門指導
改進方案:
def create_advanced_prompt(query, context, question_type):
if question_type == "factual":
template = """
你是一個精確的事實回答助手。以下是與問題相關的信息:
{context}
問題: {query}
請基于以上信息提供精確的事實回答。如信息不足,請明確指出。請在回答末尾標明信息來源。
"""
elif question_type == "analytical":
template = """
你是一個分析型助手。以下是相關參考信息:
{context}
問題: {query}
請分析以上信息,提供深入見解。注意分析信息的一致性與可靠性,標明不同來源間的差異。最后給出綜合結論并注明信息來源。
"""
# 更多問題類型...
# 添加思維鏈指導
template += """
思考步驟:
1. 提取問題關鍵點
2. 識別相關文檔中最有價值的信息
3. 對比不同來源信息
4. 形成清晰、全面的回答
"""
return template.format(cnotallow=context, query=query)
預期效果:更加結構化和針對性的回答,提高回答質量和可信度。
3.7 元數據增強與過濾
當前問題:
元數據利用不足
缺乏基于元數據的預過濾和后過濾
改進方案:
# 豐富元數據
metadatas = [{
"source": file_name,
"doc_id": doc_id,
"date_processed": datetime.now().isoformat(),
"chunk_index": i,
"total_chunks": len(chunks),
"document_type": detect_document_type(text),
"language": detect_language(chunk),
"entities": extract_entities(chunk)
} for i, chunk in enumerate(chunks)]
# 檢索時使用元數據過濾
results = COLLECTION.query(
query_embeddings=query_embedding,
n_results=10,
where={"document_type": {"$in": ["report", "research_paper"]}},
include=['documents', 'metadatas']
)
預期效果:更精準的檢索結果篩選,減少噪音,提高答案質量。
3.8 向量化模型升級
當前問題:
all-MiniLM-L6-v2 維度較低(384 維)
中文表示能力有限
改進方案:
# 升級為更強大的雙語模型
from sentence_transformers import SentenceTransformer
EMBED_MODEL = SentenceTransformer('BAAI/bge-large-zh-v1.5') # 1024維,中文效果更好
# 或考慮OpenAI嵌入模型
import openai
embeddings = openai.Embedding.create(
input=chunks,
model="text-embedding-ada-002"
)
預期效果:提高語義理解能力和檢索精度,尤其是中文文檔。
3.9 評估與反饋機制
當前問題:
缺乏回答質量評估
無法迭代改進
改進方案:
def evaluate_response(query, response, retrieved_docs):
# LLM自評估
evaluation_prompt = f"""
問題: {query}
回答: {response}
請評估這個回答的質量(1-10分),考慮以下標準:
1. 準確性: 回答是否符合檢索文檔的事實?
2. 完整性: 回答是否全面覆蓋問題的各個方面?
3. 相關性: 回答是否切中問題要點?
4. 一致性: 回答內部是否存在矛盾?
請詳細說明評分理由。
"""
feedback = llm_client.generate(evaluation_prompt)
# 存儲評估結果用于系統優化
save_evaluation_result(query, response, retrieved_docs, feedback)
return feedback
預期效果:建立持續評估和改進機制,累積數據用于系統優化。
3.10 緩存與預計算策略
當前問題:
重復問題重復計算
實時響應不夠快
改進方案:
# 問題-檢索結果緩存
import hashlib
import pickle
from functools import lru_cache
@lru_cache(maxsize=100)
def cached_retrieval(query):
# 計算查詢哈希
query_hash = hashlib.md5(query.encode()).hexdigest()
cache_file = f"cache/{query_hash}.pkl"
# 檢查緩存
try:
with open(cache_file, 'rb') as f:
return pickle.load(f)
except FileNotFoundError:
# 執行檢索
results = perform_retrieval(query)
# 保存到緩存
os.makedirs("cache", exist_ok=True)
with open(cache_file, 'wb') as f:
pickle.dump(results, f)
return results
4、寫在最后
升級版本的項目源碼在目前開源版本的基礎上,主要完成了前三個方面(分塊策略、混合檢索策略、重排序機制)的代碼升級,源碼發布在了知識星球內,供有一定實踐經驗的盆友測試。剩余其他 7 個維度的示例代碼,后續會持續發布,并結合真實案例進行演示。
另外有個彩蛋是,為了讓大家更加可視化的了解整個調優過程,我在代碼升級的同時做了 UI 優化,相比開源項目新增了一個”分塊可視化“的選項卡,其中直觀展示了當前代碼中的核心模型和方法,以及分塊之后的實際內容預覽,對調試過程會更加便利些。