構建生產級RAG系統：從數據處理到智能體的全流程實踐

檢索增強生成（RAG）技術已成為打造高級知識問答系統的核心，但從原型到穩定高效的生產級系統，需突破數據處理、檢索優化、智能決策等多重挑戰。本文以某型號工業設備故障維修手冊為知識源，結合LangChain、LangGraph、RAGAS等工具，詳細拆解生產級RAG系統的全鏈路構建過程，為技術落地提供可復用的實踐方案。

本文較長，建議點贊收藏，以免遺失。更多AI大模型開發 學習視頻/籽料/面試題 都在這>>Github<< >>Gitee<<

一、整體架構：打造Agent驅動的智能RAG系統

生產級RAG絕非簡單的“檢索+生成”，而是具備自我規劃、決策與反思能力的智能體系。基于Agent的RAG架構，核心是將復雜問題拆解為可執行子任務，通過工具調用與循環優化，確保回答的準確性與完整性。

1.1 核心工作流

系統以“問題匿名化-規劃-任務執行-重新規劃-答案生成與評估”為核心鏈路，具體步驟如下：

問題匿名化：替換問題中的具體實體（如“V12型發動機”→“設備X”），消除大模型預訓練知識偏見，避免“知識污染”。
規劃：Agent根據匿名化問題生成高層次解決計劃，例如“先明確故障代碼含義→檢索對應診斷流程→提取安全操作指令”。
計劃分解與反匿名化：將高層計劃拆分為工具可執行的子任務，并恢復原始實體名稱，確保任務與實際知識源匹配。
任務處理與工具執行：任務處理器根據子任務類型，選擇合適工具（如故障摘要檢索、維修流程檢索），從不同向量數據庫獲取信息。
重新規劃：Agent評估新獲取的信息是否充足，若存在缺口則調整計劃（如補充檢索相關安全警告），形成“執行-反思-優化”的閉環。
答案生成與評估：信息充足時，結合思維鏈（CoT）推理生成答案，再通過RAGAS框架量化評估答案質量，確保無幻覺、高準確。

二、數據預處理：構建高質量知識庫的基石

高質量知識庫是RAG系統性能的核心保障。以《V12型渦輪增壓柴油發動機維修手冊》（PDF格式）為例，需經過“加載提取-多策略分塊-清洗-重構-向量化存儲”五步流程，構建多維度知識索引。

2.1 數據加載與初步提取

使用PyPDF2庫提取PDF文本，處理加密文件、文本斷裂等異常情況，確保原始數據完整性。核心代碼邏輯如下：

import PyPDF2
from pathlib import Pathdef load_and_extract_text_from_pdf(pdf_path: Path):if not pdf_path.is_file():print(f"文件未找到: {pdf_path}")return Nonetry:with open(pdf_path, 'rb') as f:reader = PyPDF2.PdfReader(f)if reader.is_encrypted:  # 處理加密文件print(f"PDF {pdf_path.name} 已加密，無法提取")return None# 提取所有頁面文本，過濾空內容pages_text = [page.extract_text() for page in reader.pages if page.extract_text()]return " ".join(pages_text)except Exception as e:print(f"提取錯誤: {e}")return None

2.2 多策略分塊：適配不同查詢需求

單一分塊策略無法應對多樣化查詢（如“故障代碼含義”“維修步驟細節”“安全警告”），需采用三種分塊方式并行處理：

邏輯分塊：基于文本結構（如故障代碼章節、關鍵指令）分塊，保留完整語義。例如用正則表達式匹配“FAULT CODE XXX”格式，拆分故障代碼章節，并提取故障代碼作為元數據：

import re
from langchain.docstore.document import Documentdef chunk_by_fault_code(text: str) -> list[Document]:# 匹配“FAULT CODE 字母數字組合”格式sections = re.split(r'(FAULT CODE\s[A-Z0-9]+.*)', text)chapters = []for i in range(1, len(sections), 2):title = sections[i].strip()content = sections[i+1].strip()# 提取故障代碼作為元數據fault_code = re.search(r'FAULT CODE\s([A-Z0-9]+)', title).group(1) if re.search(r'FAULT CODE\s([A-Z0-9]+)', title) else "UNKNOWN"chapters.append(Document(page_content=title+"\n"+content, metadata={"source": "fault_code", "fault_code": fault_code}))return chapters

關鍵指令提取：針對“WARNING”“NOTE”等關鍵字，提取安全操作、注意事項等核心信息，滿足緊急查詢需求。
傳統分塊：用RecursiveCharacterTextSplitter對無結構文本進行固定大小分塊（如1000字符/塊，200字符重疊），適配常規細節查詢。

2.3 數據清洗：消除格式噪聲

PDF提取文本常存在多余空格、換行符、單詞斷裂（如“ma- nual”）等問題，需通過正則表達式標準化：

def clean_text(text: str) -> str:# 合并多換行符為單個，修復斷裂單詞，統一空格text = re.sub(r'\n\s*\n', '\n', text)  # 合并空行text = re.sub(r'(\w)-\n(\w)', r'\1\2', text)  # 修復“ma- nual”→“manual”text = text.replace('\n', ' ')  # 換行符轉空格text = re.sub(r' +', ' ', text)  # 多空格轉單個return text.strip()

2.4 數據重構：生成章節摘要

對故障章節等長文本，用LLM生成濃縮摘要，減少向量化噪聲。以Ollama（qwen2:1.5b模型）為例，核心邏輯如下：

from langchain.chains.summarize import load_summarize_chain
from langchain_community.chat_models.ollama import ChatOllamadef generate_chapter_summaries(documents: list[Document]):llm = ChatOllama(model="qwen2:1.5b", temperature=0)# 定義摘要Prompt，聚焦癥狀、診斷步驟、操作動作prompt = """請總結以下維修章節，重點包含故障癥狀、診斷步驟、執行動作，供技術人員直接使用：章節內容：{text}詳細摘要："""chain = load_summarize_chain(llm, chain_type="stuff", prompt=prompt)summaries = []for doc in documents:result = chain.invoke([doc])summaries.append(Document(page_content=result["output_text"], metadata=doc.metadata))return summaries

2.5 向量化與存儲：構建多源向量庫

將“傳統分塊文本”“章節摘要”“關鍵指令”分別向量化，用FAISS存儲為三個獨立向量庫，供Agent按需檢索。以HuggingFace的BAAI/bge-small-en-v1.5模型為例：

from langchain_community.vectorstores import FAISS
from langchain_community.embeddings.huggingface import HuggingFaceEmbeddingsdef build_vector_stores(data_dict: dict):# 初始化嵌入模型embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-small-en-v1.5",model_kwargs={"device": "cpu"},encode_kwargs={"normalize_embeddings": True})# 為每種數據類型創建FAISS庫vector_stores = {}for name, docs in data_dict.items():vector_store = FAISS.from_documents(docs, embeddings)vector_store.save_local(f"./vector_stores/{name}")vector_stores[name] = vector_storereturn vector_stores# 示例：構建三個向量庫
data = {"manual_splits": 傳統分塊文檔列表,"chapter_summaries": 章節摘要文檔列表,"key_instructions": 關鍵指令文檔列表
}
vector_stores = build_vector_stores(data)

三、核心RAG流程：構建LangGraph智能體

基于LangGraph構建Agent，實現“規劃-執行-反思”的閉環邏輯，核心包括規劃器、任務處理器、檢索子圖、幻覺抑制四大模塊。

3.1 規劃與任務處理：Agent的決策核心

規劃器（Planner）：根據問題生成初始計劃，例如針對“如何處理故障代碼P0128”，生成計劃“1. 檢索P0128含義→2. 提取診斷步驟→3. 確認安全操作”。
重新規劃器（Re-planner）：根據檢索結果調整計劃，若未找到安全操作指令，則補充“檢索WARNING類型文檔”步驟。
任務處理器（Task Handler）：將子任務映射到具體工具，例如“檢索P0128含義”→調用“chapter_summaries”向量庫，核心代碼邏輯如下：

from pydantic import BaseModel
from langchain_core.prompts import ChatPromptTemplate# 定義任務處理器輸出結構
class TaskOutput(BaseModel):query: str  # 檢索查詢詞tool: str  # 工具類型（如retrieve_summaries）curr_context: str  # 當前上下文def task_handler(question: str, curr_task: str, aggregated_context: str):llm = ChatOllama(model="qwen2:1.5b", temperature=0, format="json")# Prompt定義工具映射規則prompt = """你是工業維修RAG的任務處理器，需選擇工具執行當前任務：可用工具：- retrieve_summaries：查詢故障摘要（如故障代碼含義）- retrieve_chunks：查詢詳細步驟（如維修流程）- retrieve_instructions：查詢安全指令（如WARNING）- answer_question：信息充足時直接生成答案原始問題：{question}當前上下文：{aggregated_context}當前任務：{curr_task}請返回JSON格式：{{"query": "檢索詞", "tool": "工具名", "curr_context": "當前上下文"}}"""prompt_template = ChatPromptTemplate.from_template(prompt)chain = prompt_template | llm.with_structured_output(TaskOutput)return chain.invoke({"question": question, "curr_task": curr_task, "aggregated_context": aggregated_context})

3.2 檢索子圖：模塊化檢索與蒸餾

為每種檢索工具構建獨立LangGraph子圖，包含“檢索-蒸餾-驗證”三步，確保檢索結果準確。以“摘要檢索”為例，子圖邏輯如下：

檢索節點：調用“chapter_summaries”向量庫，獲取相關文檔。
蒸餾節點：過濾無關信息（如排除與故障代碼無關的摘要）。
驗證節點：檢查蒸餾后內容是否忠于原文，若存在偏差則重新過濾。

核心代碼框架如下：

from langgraph.graph import StateGraph, END
from typing import TypedDict# 定義子圖狀態
class RetrievalState(TypedDict):query: strretrieved_docs: list[Document]distilled_docs: list[Document]# 構建檢索子圖的高階函數
def build_retrieval_subgraph(retriever):graph = StateGraph(RetrievalState)# 1. 檢索節點：調用向量庫def retrieve_node(state):docs = retriever.invoke(state["query"])return {"retrieved_docs": docs}# 2. 蒸餾節點：過濾無關文檔def distill_node(state):# 模擬：保留與query匹配度高的文檔（實際用LLM判斷）distilled = [doc for doc in state["retrieved_docs"] if state["query"] in doc.page_content]return {"distilled_docs": distilled}# 3. 驗證節點：判斷是否忠于原文（實際用LLM評估）def is_grounded(state):return "grounded" if len(state["distilled_docs"]) > 0 else "not_grounded"# 組裝子圖graph.add_node("retrieve", retrieve_node)graph.add_node("distill", distill_node)graph.set_entry_point("retrieve")graph.add_edge("retrieve", "distill")# 條件邊：驗證通過則結束，否則重新蒸餾graph.add_conditional_edges("distill", is_grounded, {"grounded": END, "not_grounded": "distill"})return graph.compile()

3.3 思維鏈與幻覺抑制：提升答案可信度

思維鏈（CoT）推理：在答案生成階段，引導LLM輸出“推理過程+最終答案”，例如：“1. 故障代碼P0128表示冷卻液節溫器故障→2. 診斷步驟：先檢查溫度傳感器讀數→3. 維修動作：若讀數異常，更換節溫器”。
幻覺抑制子圖：生成答案后，用LLM評估“答案是否基于檢索上下文”，若存在幻覺（如編造未提及的工具），則回退至“重新檢索”步驟。核心邏輯如下：

def hallucination_check(answer: str, context: list[Document]) -> bool:llm = ChatOllama(model="qwen2:1.5b", temperature=0)prompt = """判斷以下答案是否完全基于提供的上下文，無編造信息：答案：{answer}上下文：{context}若完全基于上下文，返回True；否則返回False。"""result = llm.invoke(prompt.format(answer=answer, context="\n".join([c.page_content for c in context])))return "true" in result.lower()

四、實戰測試與評估：驗證系統性能

通過“案例測試+RAGAS量化評估”，確保系統在準確性、抗幻覺性、實用性上達標。

4.1 案例測試：驗證核心能力

信息不存在測試：提問“如何更換V12發動機火花塞？”（柴油發動機無火花塞）。系統應檢索后返回“未找到相關信息，且該設備為柴油機，無需更換火花塞”，驗證抗幻覺能力。
復雜推理測試：提問“發動機過熱且冷卻液位正常，應檢查哪些部件？”。系統需推理：“1. 排除冷卻液泄漏→2. 檢索過熱相關故障→3. 確定冷卻風扇、節溫器為關鍵部件”，驗證多步推理能力。
CoT生成測試：提問“如何處理故障代碼P0128？”。系統需輸出“推理過程+步驟”，例如：“1. P0128含義：冷卻液節溫器故障→2. 診斷：檢查溫度傳感器讀數→3. 維修：更換節溫器→4. 安全警告：需冷卻發動機后操作”，驗證深度解釋能力。

4.2 RAGAS量化評估：多維度打分

使用RAGAS框架，從5個核心指標評估系統性能，示例代碼如下：

from ragas import evaluate
from ragas.metrics import answer_correctness, faithfulness, answer_relevancy# 準備評估數據集（問題、生成答案、檢索上下文、標準答案）
data = {"question": ["V12發動機燃油噴射系統標準壓力是多少？", "拆卸機油濾清器需什么工具？"],"answer": ["燃油噴射系統標準壓力為2000 bar", "需36mm套筒扳手"],"contexts": [["5.2節：燃油噴射系統標準壓力2000 bar，高壓易致損壞"],["3章12頁：拆卸機油濾清器需36mm套筒扳手、接油盤"]],"ground_truth": ["標準壓力2000 bar", "需36mm套筒扳手"]
}
dataset = Dataset.from_dict(data)# 配置RAGAS評估模型
ragas_llm = LangchainLLM(llm=ChatOllama(model="qwen2:1.5b"))
faithfulness.llm = ragas_llm
answer_correctness.llm = ragas_llm# 運行評估
results = evaluate(dataset=dataset,metrics=[answer_correctness, faithfulness, answer_relevancy]
)
# 輸出結果（DataFrame格式）
print(results.to_pandas())