【大模型工程实践③】RAG 基础架构与完整实现

【大模型工程实践③】RAG 基础架构与完整实现：从0到1跑通作者：AI学习者 | 来源：大模型工程实践学习系列 | 更新：2026年3月【理论要点速览】学习本篇前，建议先掌握以下核心理论（点击跳转）：① 为什么需要RAG？② RAG vs Fine-tuning vs Long Context的决策框架③ Naive RAG → Advanced RAG → Modular RAG的架构演进④ Self-RAG的理论基础⑤ 检索-生成协同的理论基础前言RAG（Retrieval-Augmented Generation，检索增强生成）是目前大模型工程落地最成熟、最广泛使用的架构。不夸张地说：不懂 RAG，就等于不懂大模型工程。2025-2026年 RAG 的最新进展RAG 领域在过去一年发生了重大演变：① 工具生态的爆发RAGFlow（开源，阿里开发）：企业级RAG平台，支持复杂文档解析、多模态检索Dify：低代码RAG应用构建平台，已成为开发者首选LlamaIndex：从单纯的索引库演进为完整的RAG编排框架LangChain：从链式调用升级到LangGraph（有向图编排）② 架构范式的转变Native RAG → Modular RAG：从简单的检索+生成，演进到模块化、可组合的架构查询改写（Query Rewriting）多路召回（Multi-path Retrieval）融合排序（Fusion Ranking）自适应检索（Adaptive Retrieval）Self-RAG 的兴起：模型自主决定何时检索、检索什么、如何利用检索结果相比传统RAG，准确率提升15-25%减少不必要的检索调用，降低成本③ Embedding 模型的升级BGE-M3（2024年底发布）：支持稠密+稀疏向量混合检索，中英文双语，成为业界标准本地部署成为主流：FlagEmbedding库使得BGE-M3本地部署成本极低多语言、多粒度检索：从单一向量到多种检索方式的融合④ 国产LLM的成熟通义千问（Qwen）：API稳定，成本低，中文理解能力强DeepSeek：开源模型质量突破，本地部署成本大幅下降零一万物（Yi）：长上下文支持，适合RAG场景1. RAG 为什么存在？1.1 大模型的知识困境大模型有两大硬伤：① 知识过时GPT-4（2023年4月训练） → 无法知道2024年的事情 你问它最新的技术进展 → 它会编造（幻觉）② 知识盲区你公司的内部数据 → 大模型从未见过 你的私有文档 → 大模型无法回答1.2 RAG 的解决思路“不要让模型记住一切，让模型学会查找。”用户问题 → 查询改写 → 多路召回 → 融合排序 → 上下文组装 → LLM生成回答本质：把"知识的存储"和"知识的推理"分开：检索系统负责找相关知识（可以实时更新，支持多种检索方式）LLM 负责理解和生成（利用强大的推理能力）2. 现代 RAG 完整架构图┌────────────────────────────────────────────────────────────────────────┐ │ 现代 Modular RAG 系统 │ │ │ │ ┌──────────────┐ ┌──────────────────────┐ │ │ │ 文档输入 │ │ 用户问题 │ │ │ └──────┬───────┘ └────────┬─────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────┐ 写入流 ┌──────────────────────┐ │ │ │ 文档加载 │ ──────────────────────▶ │ 查询改写/扩展 │ │ │ │ (多格式支持) │ │ (Query Rewriting) │ │ │ └──────┬───────┘ └────────┬─────────────┘ │ │ │ │ │ │ ▼ ▼ │ │ ┌──────────────┐ ┌──────────────────────┐ │ │ │ 文本分块 │ │ 多路召回 │ │ │ │ (智能分块) │ │ (Multi-path) │ │ │ └──────┬───────┘ │ ├─ 稠密向量检索 │ │ │ │ │ ├─ 稀疏向量检索 │ │ │ ▼ │ ├─ 关键词检索 │ │ │ ┌──────────────┐ │ └─ 混合检索 │ │ │ │ Embedding │ └────────┬─────────────┘ │ │ │ (BGE-M3) │ ────────┐ │ │ │ │ 稠密+稀疏向量 │ │ ▼ │ │ └──────┬───────┘ │ ┌──────────────────────┐ │ │ │ │ │ 融合排序 │ │ │ ▼ ▼ │ (Fusion Ranking) │ │ │ ┌──────────────┐ ┌──────────────┐ │ ├─ RRF算法 │ │ │ │ 向量数据库 │◀─┤ 相似度计算 │ │ ├─ 学习排序 │ │ │ │ (FAISS/Qdrant)│ └──────────────┘ │ └─ 重排序模型 │ │ │ └──────┬───────┘ └────────┬─────────────┘ │ │ │ │ │ │ └───────────────────────┬───────────────────┘ │ │ ▼ │ │ ┌──────────────────────┐ │ │ │ 上下文组装 │ │ │ │ (Context Assembly) │ │ │ └────────┬─────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────────┐ │ │ │ Self-RAG 决策 │ │ │ │ (是否需要检索？) │ │ │ └────────┬─────────────┘ │ │ │ │ │ ▼ │ │ ┌──────────────────────┐ │ │ │ LLM 生成回答 │ │ │ │ (支持流式输出) │ │ │ └──────────────────────┘ │ │ │ └────────────────────────────────────────────────────────────────────────┘3. 环境准备与镜像配置3.1 关键依赖安装为什么需要这些依赖？库作用选型理由langchainRAG 流程抽象统一文档加载、分割、检索接口faiss-cpu/faiss-gpu向量索引高效最近邻搜索，支持百万级向量sentence-transformersEmbedding 模型生成文档和查询的语义向量transformersLLM 调用与各类大模型交互的统一接口版本注意：Windows 推荐faiss-cpu；如使用 CUDA 用faiss-gpu。conda 安装示例（推荐）：conda install faiss-cpu langchain langchain-community \ sentence-transformers transformers -c conda-forge pip install langchain-openai # 如果使用 OpenAI API# 核心RAG框架pipinstalllangchain langchain-community langchain-openai# Embedding模型（本地）pipinstallFlagEmbedding# 向量数据库（选择一个或多个）pipinstallfaiss-cpu# 轻量级，适合原型pipinstallqdrant-client# 生产级，支持持久化pipinstallchromadb# 简单易用，适合快速原型# 文档处理pipinstallpypdf# PDF解析pipinstallpython-docx# Word文档pipinstallbeautifulsoup4# HTML解析pipinstallmarkdown# Markdown解析# 工具库pipinstalltiktoken# Token计数pipinstallrequests# HTTP请求3.2 镜像源配置（必须在import之前）为什么镜像必须在 import 前配置？transformers/sentence-transformers在首次 import 时自动下载模型权重如果没有提前设置镜像，国内网络访问huggingface.co会超时设置HF_ENDPOINT环境变量可以在代码层面指定镜像地址，无需手动改 pip 源镜像检测逻辑：依次尝试hf-mirror.com→hf.duiniucdn.com→huggingface.co用urllib做 5 秒超时探测，可访问则选中均不可用时继续运行（后续 import 可能失败，给出明确错误提示）其他可选镜像（补充）：腾讯云：https://hf.duiniucdn.com（推荐国内）火山引擎：https://hf-mirror.com（备选）""" 重要：镜像配置必须在所有huggingface/transformers相关import之前执行 这样可以加速模型下载，避免网络超时 """importos# ==== Mirror config (must be before import) ====HF_MIRRORS=["https://hf-mirror.com","https://hf.duiniucdn.com","https://huggingface.co",]for_minHF_MIRRORS:os.environ["HF_ENDPOINT"]=_mtry:importurllib.request urllib.request.urlopen(_m,timeout=5)print(f"Mirror available:{_m}")breakexceptException:print(f"Mirror failed:{_m}")else:print("WARNING: all mirrors failed")os.environ["HF_HUB_CACHE"]="C:/Users/OMEN/.hf_cache"os.environ["HF_HOME"]="C:/Users/OMEN/.hf_home"os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"]="1"importosimportsys# ============ 镜像源配置（第一步，必须最先执行） ============# HuggingFace 镜像（用于下载BGE-M3等模型）os.environ["HF_ENDPOINT"]="https://hf-mirror.com"# 国际源# 或使用国内镜像（如果国际源慢）：# os.environ["HF_ENDPOINT"] = "https://hf-mirror.com"# 通义千问 API Key（如果使用国产LLM）os.environ["DASHSCOPE_API_KEY"]="your-dashscope-api-key"# OpenAI API Key（如果使用GPT）os.environ["OPENAI_API_KEY"]="your-openai-api-key"# ============ 现在才能import相关库 ============fromlangchain_openaiimportOpenAIEmbeddings,ChatOpenAIfromlangchain_community.vectorstoresimportFAISSfromlangchain.text_splitterimportRecursiveCharacterTextSplitterfromlangchain.document_loadersimportPyPDFLoader,TextLoaderfromFlagEmbeddingimportFlagModelimporttiktoken4. 完整代码实现4.1 第一步：文档加载（多格式支持）文档加载在 RAG 中的地位：文档加载是 RAG 流水线的入口，负责将原始文件（PDF/Word/Markdown/TXT）解析为纯文本加载质量直接影响后续分块和检索的效果：噪声内容会被一并灌入 EmbeddingLangChain 文档加载器的设计思路：每个加载器只负责一种格式，实现单一职责PyPDFLoader：逐页读取 PDF，保留页码信息（用于溯源）TextLoader：最简单，直接读纯文本文件UnstructuredMarkdownLoader：解析 Markdown 结构（标题、列表、代码块）UnstructuredWordDocumentLoader：解析 Word 文档，提取正文段落最佳实践：加载后建议打印page_content确认格式是否符合预期PDF 扫描件（图片）需要先用 OCR 处理，纯 PyPDFLoader 无法提取文字importosfrompathlibimportPathfromlangchain.document_loadersimport(PyPDFLoader,TextLoader,UnstructuredMarkdownLoader,UnstructuredWordDocumentLoader)defload_documents(file_path:str)-list:""" 根据文件类型加载文档 支持格式：PDF、TXT、MD、DOCX """file_path=Path(file_path)ifnotfile_path.exists():raiseFileNotFoundError(f"文件不存在:{file_path}")suffix=file_path.suffix.lower()try:ifsuffix==".pdf":loader=PyPDFLoader(str(file_path))elifsuffix==".txt":loader=TextLoader(str(file_path),encoding="utf-8")elifsuffix==".md":loader=UnstructuredMarkdownLoader(str(file_path))elifsuffix==".docx":loader=UnstructuredWordDocumentLoader(str(file_path))else:raiseValueError(f"不支持的文件类型:{suffix}")documents=loader.load()# 添加文件名到metadatafordocindocuments:doc.metadata["source_file"]=file_path.namereturndocumentsexceptExceptionase:print(f"加载文档失败:{e}")raise# 示例if__name__=="__main__":docs=load_documents("data/产品手册.pdf")print(f"✓ 加载了{len(docs)}个文档块")print(f"✓ 内容预览:{docs[0].page_content[:200]}...")4.2 第二步：文本分块（智能分块策略）分块是 RAG 中最影响召回质量的环节之一：模型对每个 chunk 生成一个向量；chunk 太大则语义稀释、容易被截断chunk 太小则语义不完整、丢失上下文常用分块策略对比：策略原理优点缺点固定长度按 token 数直接切简单快速语义可能被截断递归字符分割按\n\n→\n→.优先级递归切尽量保持段落完整参数敏感语义分割用模型判断句子边界语义最完整速度慢父子块大块存父、小块检索引、召回后挂载父兼顾精度和上下文实现复杂关键参数：chunk_size：每块 token 数，推荐 256~512（中文约 200~400 字）chunk_overlap：块间重叠 token 数，推荐 chunk_size 的 10~20%，避免边界截断丢失语义fromlangchain.text_splitterimportRecursiveCharacterTextSplitterimporttiktokendefchunk_documents(documents:list,chunk_size:int=512,chunk_overlap:int=100,use_token_count:bool=True)-list:""" 将文档分割成小块（智能分块） Args: documents: 原始文档列表 chunk_size: 每个块的大小（字符数或token数） chunk_overlap: 块之间的重叠（防止边界信息丢失） use_token_count: 是否使用token计数（更精确） Returns: 分块后的文档列表 说明： - chunk_size=512 对应约150-200个token - chunk_overlap=100 保证相邻块有约30%的重叠 - 对于中文文本，建议chunk_size=400-600 """# 如果使用token计数，需要转换ifuse_token_count:try:encoding=tiktoken.get_encoding("cl100k_base")deftoken_length(text):returnlen(encoding.encode(text))length_function=token_lengthexcept:# 如果tiktoken失败，回退到字符计数length_function=lenelse:length_function=len# 递归分块：优先按段落、句子、词语切割text_splitter=RecursiveCharacterTextSplitter(chunk_size=chunk_size,chunk_overlap=chunk_overlap,length_function=length_function,separators=["\n\n",# 段落分隔"\n",# 行分隔"。",# 中文句号"！",# 中文感叹号"？",# 中文问号"；",# 中文分号" ",# 空格importosimportos# ==== Mirror config (must be before import) ====HF_MIRRORS=["https://hf-mirror.com","https://hf.duiniucdn.com","https://huggingface.co",]for_minHF_MIRRORS:os.environ["HF_ENDPOINT"]=_mtry:importurllib.request urllib.request.urlopen(_m,timeout=5)print(f"Mirror available:{_m}")breakexceptException:print(f"Mirror failed:{_m}")else:print("WARNING: all mirrors failed")os.environ["HF_HUB_CACHE"]="C:/Users/OMEN/.hf_cache"os.environ["HF_HOME"]="C:/Users/OMEN/.hf_home"os.environ["HF_HUB_DISABLE_SYMLINKS_WARNING"]="1"fromFlagEmbeddingimportFlagModelfromlangchain_community.vectorstoresimportFAISSfromlangchain.embeddings.baseimportEmbeddingsfromtypingimportListclassBGEEmbeddings(Embeddings):""" 使用 BGE-M3 模型的 Embedding 类 支持稠密向量 + 稀疏向量混合检索 """def__init__(self,model_name:str="BAAI/bge-m3",use_fp16:bool=True,device:str="cuda"):""" Args: model_name: 模型名称（默认BGE-M3） use_fp16: 是否使用半精度（节省显存） device: 设备（cuda/cpu） """self.model=FlagModel(model_name,query_instruction_for_retrieval="为这个句子生成表示以用于检索相关文章：",use_fp16=use_fp16,device=device)defembed_documents(self,texts:List[str])-List[List[float