企業(yè) RAG 準(zhǔn)確性提升全流程指南：從數(shù)據(jù)提取到精準(zhǔn)檢索
在企業(yè)環(huán)境中，準(zhǔn)確高效地從大量非結(jié)構(gòu)化數(shù)據(jù)（如 PDF 文件）中檢索信息至關(guān)重要?；跈z索增強(qiáng)生成（RAG）的系統(tǒng)在這方面發(fā)揮著重要作用，但提升其準(zhǔn)確性是一個復(fù)雜且具有挑戰(zhàn)性的任務(wù)。本文將詳細(xì)介紹提高企業(yè) RAG 準(zhǔn)確性的逐步指南，涵蓋從數(shù)據(jù)提取到檢索的各個關(guān)鍵步驟。
1. 從 PDF 中提取知識
1.1 上傳與記錄創(chuàng)建
用戶上傳 PDF 文件（未來還將支持音頻和視頻等其他文件類型）。系統(tǒng)將文件保存到磁盤（近期計劃遷移到 AWS S3 存儲桶以更好地滿足企業(yè)需求），并在數(shù)據(jù)庫中插入記錄，創(chuàng)建處理狀態(tài)條目。使用支持多種數(shù)據(jù)類型、混合搜索和單次檢索的 SingleStore 數(shù)據(jù)庫。同時，將處理 PDF 的任務(wù)放入后臺隊列，通過 Redis 和 Celery 進(jìn)行異步處理和跟蹤。
# Pseudo-code
save_file_to_disk(pdf)
db_insert(document_record, status=”started”)
queue_processing_task(pdf)1.2 解析與分塊 PDF
打開文件并驗證大小限制和密碼保護(hù)，若文件不可讀則盡早終止處理。將文件內(nèi)容提取為文本或 Markdown 格式，可使用 Llamaparse（來自 Llamaindex）替代之前的 PyMudf，其免費版每天支持 1000 次文檔解析，并能更好地處理表格和圖像提取。分析文檔結(jié)構(gòu)（如目錄、標(biāo)題等），利用 Gemini Flash 2.0 基于語義將文本分割成有意義的塊，若語義分塊失敗，則采用簡單的分割方式，并在塊之間添加重疊部分以保持上下文連貫性。
# Pseudo-code
validate_pdf(pdf)
text = extract_text(pdf)
chunks = semantic_chunking(text) or fallback_chunking(text)
add_overlaps(chunks)1.3 生成嵌入
使用嵌入模型將每個文本塊轉(zhuǎn)換為高維向量，例如使用 OpenAI 的 large ada 模型，維度設(shè)置為 1536。將文本塊及其嵌入向量存儲在數(shù)據(jù)庫中，在 SingleStore 中，將文本塊和文本分別存儲在同一表的不同列，便于維護(hù)和檢索。
# Pseudo-code
for chunk in chunks:
vector = generate_embedding(chunk.text)
db_insert(embedding_record, vector)1.4 使用大語言模型提取實體和關(guān)系
這一步對整體準(zhǔn)確性影響很大。將語義組織好的文本塊發(fā)送給 OpenAI，通過特定的提示要求其返回每個塊中的實體和關(guān)系，包括關(guān)鍵實體（名稱、類型、描述、別名）。映射實體之間的關(guān)系，避免重復(fù)添加數(shù)據(jù)，將提取的 “知識” 存儲在結(jié)構(gòu)化表中。
# Pseudo-code
for chunk in chunks:
entities, relationships = extract_knowledge(chunk.text)
db_insert(entities)
db_insert(relationships)1.5 最終處理狀態(tài)
若所有步驟處理正確，將狀態(tài)更新為 “已完成”，以便前端隨時輪詢并顯示正確狀態(tài)；若處理失敗，則標(biāo)記為 “失敗”，并清理臨時數(shù)據(jù)。
# Pseudo-code
if success:
update_status(“completed”)
else:
update_status(“failed”)
cleanup_partial_data()2. 知識檢索（RAG 管道）
2.1 用戶查詢
用戶向系統(tǒng)提交查詢請求。
# Pseudo-code
query = get_user_query()2.2 預(yù)處理和擴(kuò)展查詢
系統(tǒng)對查詢進(jìn)行規(guī)范化處理，去除標(biāo)點符號、標(biāo)準(zhǔn)化空白字符，并使用大語言模型（如 Groq，處理速度更快）擴(kuò)展同義詞。
# Pseudo-code
query = preprocess_query(query)
expanded_query = expand_query(query)2.3 嵌入查詢和搜索向量
使用與提取時相同的 ada 模型將查詢嵌入為高維向量，在文檔嵌入數(shù)據(jù)庫中使用語義搜索（如在 SingleStore 中使用點積運算）查找最匹配的文本塊。
# Pseudo-code
query_vector = generate_embedding(expanded_query)
top_chunks = vector_search(query_vector)2.4 全文搜索
并行進(jìn)行全文搜索以補充向量搜索，在 SingleStore 中可使用 MATCH 語句實現(xiàn)。
# Pseudo-code
text_results = full_text_search(query)2.5 合并和排序結(jié)果
將向量搜索和全文搜索的結(jié)果合并，并根據(jù)相關(guān)性重新排序，可調(diào)整返回的前 k 個結(jié)果數(shù)量（如 k = 10 或更高效果更好），過濾掉低置信度的結(jié)果。
# Pseudo-code
merged_results = merge_and_rank(top_chunks, text_results)
filtered_results = filter_low_confidence(merged_results)2.6 檢索實體和關(guān)系
若檢索到的文本塊存在實體和關(guān)系，則將其包含在響應(yīng)中。
# Pseudo-code
for result in filtered_results:
entities, relationships = fetch_knowledge(result)
enrich_result(result, entities, relationships)2.7 生成最終答案
利用提示增強(qiáng)上下文，將相關(guān)數(shù)據(jù)發(fā)送給大語言模型（如 gpt3o - mini）生成最終響應(yīng)。
# Pseudo-code
final_answer = generate_llm_response(filtered_results)2.8 返回答案給用戶
系統(tǒng)將響應(yīng)作為結(jié)構(gòu)化 JSON 有效負(fù)載返回，并附帶原始數(shù)據(jù)庫搜索結(jié)果，以便在需要時進(jìn)行調(diào)試和調(diào)整。
# Pseudo-code
return_response(final_answer)3. 性能優(yōu)化與實現(xiàn)
在實踐中，發(fā)現(xiàn)檢索過程的響應(yīng)時間過長（約 8 秒），主要瓶頸在于大語言模型的調(diào)用（每次約 1.5 - 2 秒），而 SingleStore 數(shù)據(jù)庫查詢時間通常在 600 毫秒以內(nèi)。切換到 Groq 進(jìn)行部分大語言模型調(diào)用后，響應(yīng)時間縮短至 3.5 秒。為進(jìn)一步優(yōu)化，可嘗試并行調(diào)用而非串行調(diào)用。
為簡化管理和提高數(shù)據(jù)庫響應(yīng)時間，實現(xiàn)了單次檢索查詢的代碼。通過 OpenAI 的 Embeddings API 生成查詢嵌入向量，在 SingleStore 中執(zhí)行混合搜索 SQL 查詢，同時獲取文本塊、向量得分、文本得分、綜合得分以及相關(guān)的實體和關(guān)系信息。
import os
import json
import mysql.connector
from openai import OpenAI
# Define database connection parameters (assumed from env vars)
DB_CONFIG = {
"host": os.getenv("SINGLESTORE_HOST", "localhost"),
"port": int(os.getenv("SINGLESTORE_PORT", "3306")),
"user": os.getenv("SINGLESTORE_USER", "root"),
"password": os.getenv("SINGLESTORE_PASSWORD", ""),
"database": os.getenv("SINGLESTORE_DATABASE", "knowledge_graph")
}
def get_query_embedding(query: str) -> list:
"""
Generate a 1536-dimensional embedding for the query using OpenAI embeddings API.
"""
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = client.embeddings.create(
model="text-embedding-ada-002",
input=query
)
return response.data[0].embedding # Extract embedding vector
def retrieve_rag_results(query: str) -> list:
"""
Execute the hybrid search SQL query in SingleStore and return the top-ranked results.
"""
conn = mysql.connector.connect(**DB_CONFIG)
cursor = conn.cursor(dictinotallow=True)
# Generate query embedding
query_embedding = get_query_embedding(query)
embedding_str = json.dumps(query_embedding) # Convert to JSON for SQL compatibility
# Set the query embedding session variable
cursor.execute("SET @qvec = %s", (embedding_str,))
# Hybrid Search SQL Query (same as provided earlier)
sql_query = """
SELECT
d.doc_id,
d.content,
(d.embedding <*> @qvec) AS vector_score,
MATCH(TABLE Document_Embeddings) AGAINST(%s) AS text_score,
(0.7 * (d.embedding <*> @qvec) + 0.3 * MATCH(TABLE Document_Embeddings) AGAINST(%s)) AS combined_score,
JSON_AGG(DISTINCT JSON_OBJECT(
'entity_id', e.entity_id,
'name', e.name,
'description', e.description,
'category', e.category
)) AS entities,
JSON_AGG(DISTINCT JSON_OBJECT(
'relationship_id', r.relationship_id,
'source_entity_id', r.source_entity_id,
'target_entity_id', r.target_entity_id,
'relation_type', r.relation_type
)) AS relationships
FROM Document_Embeddings d
LEFT JOIN Relationships r ON r.doc_id = d.doc_id
LEFT JOIN Entities e ON e.entity_id IN (r.source_entity_id, r.target_entity_id)
WHERE MATCH(TABLE Document_Embeddings) AGAINST(%s)
GROUP BY d.doc_id, d.content, d.embedding
ORDER BY combined_score DESC
LIMIT 10;
"""
# Execute the query
cursor.execute(sql_query, (query, query, query))
results = cursor.fetchall()
cursor.close()
conn.close()
return results # Return list of retrieved documents with entities and relationships4. 經(jīng)驗教訓(xùn)與未來改進(jìn)方向
提高 RAG 準(zhǔn)確性并保持低延遲是一項艱巨的任務(wù)，尤其是在處理結(jié)構(gòu)化數(shù)據(jù)時。未來可從以下方面進(jìn)行改進(jìn)：
4.1 準(zhǔn)確性提升 - 提取階段
- 外部化和試驗實體提取提示嘗試不同的提示策略，以更準(zhǔn)確地從文本塊中提取實體。
- 在處理前總結(jié)文本塊可能對提高準(zhǔn)確性有顯著影響。
- 添加更好的失敗重試機(jī)制確保在各個處理步驟出現(xiàn)故障時能夠有效恢復(fù)。
4.2 準(zhǔn)確性提升 - 檢索階段
- 使用更好的查詢擴(kuò)展技術(shù)如自定義詞典、特定行業(yè)術(shù)語等，提高查詢的相關(guān)性。
- 微調(diào)向量與文本搜索的權(quán)重目前已在配置文件中進(jìn)行外部化設(shè)置，可進(jìn)一步優(yōu)化。
- 添加第二個大語言模型進(jìn)行重新排名但需要權(quán)衡延遲問題。
- 調(diào)整檢索窗口大小優(yōu)化召回率與相關(guān)性之間的平衡。
- 生成文本塊級別的摘要避免將原始文本直接發(fā)送給大語言模型，減少處理負(fù)擔(dān)。
通過以上逐步指南和持續(xù)改進(jìn)的方向，企業(yè)可以不斷優(yōu)化 RAG 系統(tǒng)的準(zhǔn)確性，更好地滿足在大量非結(jié)構(gòu)化數(shù)據(jù)中高效檢索信息的需求，提升業(yè)務(wù)效率和決策質(zhì)量。
