构建生产级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())