企業 RAG 準確性提升全流程指南：從數據提取到精準檢索
在企業環境中，準確高效地從大量非結構化數據（如 PDF 文件）中檢索信息至關重要。基于檢索增強生成（RAG）的系統在這方面發揮著重要作用，但提升其準確性是一個復雜且具有挑戰性的任務。本文將詳細介紹提高企業 RAG 準確性的逐步指南，涵蓋從數據提取到檢索的各個關鍵步驟。
1. 從 PDF 中提取知識
1.1 上傳與記錄創建
用戶上傳 PDF 文件（未來還將支持音頻和視頻等其他文件類型）。系統將文件保存到磁盤（近期計劃遷移到 AWS S3 存儲桶以更好地滿足企業需求），并在數據庫中插入記錄，創建處理狀態條目。使用支持多種數據類型、混合搜索和單次檢索的 SingleStore 數據庫。同時，將處理 PDF 的任務放入后臺隊列，通過 Redis 和 Celery 進行異步處理和跟蹤。
# Pseudo-code
save_file_to_disk(pdf)
db_insert(document_record, status=”started”)
queue_processing_task(pdf)1.2 解析與分塊 PDF
打開文件并驗證大小限制和密碼保護，若文件不可讀則盡早終止處理。將文件內容提取為文本或 Markdown 格式，可使用 Llamaparse（來自 Llamaindex）替代之前的 PyMudf，其免費版每天支持 1000 次文檔解析，并能更好地處理表格和圖像提取。分析文檔結構（如目錄、標題等），利用 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 生成嵌入
使用嵌入模型將每個文本塊轉換為高維向量，例如使用 OpenAI 的 large ada 模型，維度設置為 1536。將文本塊及其嵌入向量存儲在數據庫中，在 SingleStore 中，將文本塊和文本分別存儲在同一表的不同列，便于維護和檢索。
# Pseudo-code
for chunk in chunks:
vector = generate_embedding(chunk.text)
db_insert(embedding_record, vector)1.4 使用大語言模型提取實體和關系
這一步對整體準確性影響很大。將語義組織好的文本塊發送給 OpenAI，通過特定的提示要求其返回每個塊中的實體和關系，包括關鍵實體（名稱、類型、描述、別名）。映射實體之間的關系，避免重復添加數據，將提取的 “知識” 存儲在結構化表中。
# Pseudo-code
for chunk in chunks:
entities, relationships = extract_knowledge(chunk.text)
db_insert(entities)
db_insert(relationships)1.5 最終處理狀態
若所有步驟處理正確，將狀態更新為 “已完成”，以便前端隨時輪詢并顯示正確狀態；若處理失敗，則標記為 “失敗”，并清理臨時數據。
# Pseudo-code
if success:
update_status(“completed”)
else:
update_status(“failed”)
cleanup_partial_data()2. 知識檢索（RAG 管道）
2.1 用戶查詢
用戶向系統提交查詢請求。
# Pseudo-code
query = get_user_query()2.2 預處理和擴展查詢
系統對查詢進行規范化處理，去除標點符號、標準化空白字符，并使用大語言模型（如 Groq，處理速度更快）擴展同義詞。
# Pseudo-code
query = preprocess_query(query)
expanded_query = expand_query(query)2.3 嵌入查詢和搜索向量
使用與提取時相同的 ada 模型將查詢嵌入為高維向量，在文檔嵌入數據庫中使用語義搜索（如在 SingleStore 中使用點積運算）查找最匹配的文本塊。
# Pseudo-code
query_vector = generate_embedding(expanded_query)
top_chunks = vector_search(query_vector)2.4 全文搜索
并行進行全文搜索以補充向量搜索，在 SingleStore 中可使用 MATCH 語句實現。
# Pseudo-code
text_results = full_text_search(query)2.5 合并和排序結果
將向量搜索和全文搜索的結果合并，并根據相關性重新排序，可調整返回的前 k 個結果數量（如 k = 10 或更高效果更好），過濾掉低置信度的結果。
# Pseudo-code
merged_results = merge_and_rank(top_chunks, text_results)
filtered_results = filter_low_confidence(merged_results)2.6 檢索實體和關系
若檢索到的文本塊存在實體和關系，則將其包含在響應中。
# Pseudo-code
for result in filtered_results:
entities, relationships = fetch_knowledge(result)
enrich_result(result, entities, relationships)2.7 生成最終答案
利用提示增強上下文，將相關數據發送給大語言模型（如 gpt3o - mini）生成最終響應。
# Pseudo-code
final_answer = generate_llm_response(filtered_results)2.8 返回答案給用戶
系統將響應作為結構化 JSON 有效負載返回，并附帶原始數據庫搜索結果，以便在需要時進行調試和調整。
# Pseudo-code
return_response(final_answer)3. 性能優化與實現
在實踐中，發現檢索過程的響應時間過長（約 8 秒），主要瓶頸在于大語言模型的調用（每次約 1.5 - 2 秒），而 SingleStore 數據庫查詢時間通常在 600 毫秒以內。切換到 Groq 進行部分大語言模型調用后，響應時間縮短至 3.5 秒。為進一步優化，可嘗試并行調用而非串行調用。
為簡化管理和提高數據庫響應時間，實現了單次檢索查詢的代碼。通過 OpenAI 的 Embeddings API 生成查詢嵌入向量，在 SingleStore 中執行混合搜索 SQL 查詢，同時獲取文本塊、向量得分、文本得分、綜合得分以及相關的實體和關系信息。
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. 經驗教訓與未來改進方向
提高 RAG 準確性并保持低延遲是一項艱巨的任務，尤其是在處理結構化數據時。未來可從以下方面進行改進：
4.1 準確性提升 - 提取階段
- 外部化和試驗實體提取提示嘗試不同的提示策略，以更準確地從文本塊中提取實體。
- 在處理前總結文本塊可能對提高準確性有顯著影響。
- 添加更好的失敗重試機制確保在各個處理步驟出現故障時能夠有效恢復。
4.2 準確性提升 - 檢索階段
- 使用更好的查詢擴展技術如自定義詞典、特定行業術語等，提高查詢的相關性。
- 微調向量與文本搜索的權重目前已在配置文件中進行外部化設置，可進一步優化。
- 添加第二個大語言模型進行重新排名但需要權衡延遲問題。
- 調整檢索窗口大小優化召回率與相關性之間的平衡。
- 生成文本塊級別的摘要避免將原始文本直接發送給大語言模型，減少處理負擔。
通過以上逐步指南和持續改進的方向，企業可以不斷優化 RAG 系統的準確性，更好地滿足在大量非結構化數據中高效檢索信息的需求，提升業務效率和決策質量。
