RAG 整体架构——两条管道的每个环节

RAG（检索增强生成）已成为构建企业级 AI 应用的核心范式。但很多人只停留在"向量检索 + 丢给 LLM"的粗浅理解上。本文带你深入 RAG 的两大核心管道——索引管道和查询管道，逐环节拆解每个步骤的技术细节与最佳实践。

一、RAG 的本质：为什么需要两条管道？

RAG 的核心思想是：让 LLM 在生成答案之前，先去外部知识库里查资料。

这样做解决了大模型的三个根本痛点：

• 📌知识滞后：模型训练数据有截止日期，RAG 可以接入最新文档
• 📌幻觉问题：要求模型"仅凭检索结果回答"，大幅降低编造概率
• 📌私有数据：企业内部的文档、代码、规范，不需要微调就能用

但要把这件事做好，需要两条相互配合的管道：

索引管道（离线，定期运行） 文档 → 切分 → 向量化 → 存入向量数据库 ↓查询管道（在线，用户每次提问触发） 用户问题 → 查询优化 → 检索 → 重排序 → LLM 生成答案

索引管道决定"能查到什么"，查询管道决定"查得有多准"。两条管道缺一不可。

二、索引管道（Ingestion Pipeline）—— 让知识可被检索

索引管道是离线预处理阶段，目标是从原始文档中构建高质量的检索索引。

环节 1：文档加载（Document Loading）

做什么：把各种格式的文档统一加载为文本。

支持的数据源：

• 文件：PDF、Word、Markdown、HTML、TXT
• 代码：Java/Python/JS 等源码文件
• 网页：通过爬虫或 API 获取
• 数据库：MySQL、MongoDB 的查询结果
• 企业系统：Confluence、Notion、GitLab

最佳实践：

# LangChain 文档加载示例from langchain.document_loaders import ( PyPDFLoader, # PDF UnstructuredHTMLLoader, # HTML NotionDirectoryLoader, # Notion)loaders = [PyPDFLoader("手册.pdf"), UnstructuredHTMLLoader("页面.html")]documents = []for loader in loaders: documents.extend(loader.load())

⚠️坑点提醒：PDF 加载常有乱码、表格丢失、多栏排版错乱等问题，生产环境建议使用专门的 PDF 解析库（如pdfplumber或unstructured）。

环节 2：文本分块（Chunking）—— 最容易被忽视的关键步骤

为什么必须分块？

即使 Gemini 1.5 Pro 支持百万级 Token 上下文，把整本书塞进去也不现实：

1. 成本爆炸：每次查询都要消耗大量 Token
2. 中间遗忘：LLM 对超长上下文的中间部分记忆能力显著下降
3. 检索精度：用整本书去做向量相似度匹配，结果会非常模糊

分块策略全景对比：

关键参数设置：

# 递归切分示例（LangChain）from langchain.text_splitter import RecursiveCharacterTextSplittertext_splitter = RecursiveCharacterTextSplitter( separators=["\n\n", "\n", "。", ".", " ", ""], # 分隔符优先级 chunk_size=800, # 每块约 800 tokens chunk_overlap=100, # 相邻块重叠 100 tokens length_function=len,)chunks = text_splitter.split_documents(documents)

参数建议：

• chunk_size：500～1000 tokens（根据嵌入模型的最佳输入长度调整）
• chunk_overlap：50～150 tokens（太小会丢失跨块信息，太大浪费存储）

高级技巧：Parent Document Retriever（小块到大块）

# 索引时用小块（提高检索精度）# 但返回给 LLM 时用大块（保证上下文完整）from langchain.retrievers import ParentDocumentRetrieverfrom langchain.storage import InMemoryStoreretriever = ParentDocumentRetriever( vectorstore=vectorstore, # 存小块的向量库 docstore=InMemoryStore(), # 存大块的文档库 child_chunk_size=400, # 小块大小 parent_chunk_size=2000, # 大块大小)

💡核心思想：检索用小块的精确向量，生成用大块的完整上下文，一箭双雕。

环节 3：文本嵌入（Embedding）

做什么：把每个文本块转换为固定维度的向量，使其能被相似度搜索。

核心公式：
v _ c = E ( c ) , v _ q = E ( q ) v\_c = E(c), \quad v\_q = E(q)v_c=E(c),v_q=E(q)

其中E EE是嵌入模型，c cc是文本块，q qq是查询。

主流嵌入模型对比（2025 推荐）：

⚠️ 关键原则：

• 查询和文档必须使用同一个嵌入模型
• 不同模型生成的向量在不同向量空间，无法直接比较相似度

代码示例：

from langchain.embeddings import OpenAIEmbeddingsembeddings = OpenAIEmbeddings( model="text-embedding-3-small", dimensions=1536 # 可以指定降维，节省存储)# 对文本块进行向量化vectors = embeddings.embed_documents([chunk.page_content for chunk in chunks])

环节 4：向量存储与索引（Vector Storage & Indexing）

做什么：把向量和对应的文本块存入向量数据库，并建立高效索引。

主流向量数据库对比：

向量索引算法对比：

HNSW 关键参数：

# Qdrant 创建索引示例client.create_collection( collection_name="docs", vectors_config=models.VectorParams( size=1536, distance=models.Distance.COSINE, ), hnsw_config=models.HnswConfigDiff( m=16, # 每个节点最大出度，越大召回率越高，内存消耗越大 ef_construct=100, # 构建索引时的候选列表大小，越大质量越高，构建越慢 ))# 搜索时results = client.search( collection_name="docs", query_vector=query_vector, limit=5, search_params=models.SearchParams( hnsw_ef=50, # 搜索时的候选列表大小，越大召回率越高，搜索越慢 ))

💡参数调优经验：m=16, ef_construct=100是通用起点；追求更高召回率可设m=32, ef_construct=200；资源受限时可设m=8。

存储时要保留的元数据：

# 每个向量条目应包含的元数据{ "vector": [0.123, 0.456, ...], "text": "原始文本块内容", "metadata": { "source": "用户手册.pdf", # 来源文档 "page": 42, # 页码 "chunk_id": "chunk_123", # 块 ID "last_updated": "2025-01-15", # 最后更新时间 "category": "安装指南", # 自定义分类 }}

元数据可以用于检索时的过滤（如：只搜索 2025 年之后的文档）和加权（如：优先展示官方文档而非论坛回复）。

环节 5（可选）：元数据增强与预处理

Contextual Retrieval（Anthropic 提出）：

在嵌入之前，用 LLM 为每个文本块生成一段上下文说明（50～100 tokens），拼接到原文本后再做嵌入：

# 为每个块生成上下文说明def add_contextual_explanation(chunk, full_doc): prompt = f""" 以下是文档中的一个片段，以及完整文档的内容。 请为这个片段写一段简短说明，描述它在完整文档中的上下文背景。 完整文档（节选）：{full_doc[:2000]} 文档片段：{chunk} 上下文说明： """ return client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}] ).choices[0].message.content# 存储时：上下文说明 + 原文本enhanced_text = f"{contextual_explanation}\n\n{original_chunk}"

Anthropic 的实验表明，这种方法能将检索失败率降低 67%。

三、查询管道（Query Pipeline）—— 让检索更精准

查询管道是在线服务阶段，每次用户提问都会完整执行一遍。

环节 1：查询处理与优化（Query Processing）

很多用户的问题并不直接适合检索（口语化、指代模糊、过于简略），这一步的目标是把用户的"人话"转成更适合检索的"检索语言"。

1.1 查询重写（Query Rewriting）

# 将口语化查询改写为更规范的形式# 原查询："那个啥来着，就是那个跑得快的框架"# 重写后："高性能 Web 框架 对比"rewrite_prompt = f"""请将以下用户查询改写为更适合向量检索的规范化查询。保留所有关键信息，去除口语化表达，补充可能的同义词。原始查询：{user_query}规范化查询："""

1.2 查询扩展（Query Expansion）

让 LLM 生成多个查询变体，从不同角度去检索：

# 生成 3 个不同角度的查询变体expansion_prompt = f"""用户原始查询：{user_query}请从以下 3 个不同角度，生成查询变体：1. 使用更专业的术语2. 使用更通俗的表达3. 分解为多个子问题输出格式：每个变体一行"""# 用每个变体分别检索，合并结果去重

1.3 HyDE（假设性文档嵌入）

这是目前效果最显著的查询优化技术之一：

# 步骤 1：让 LLM 根据查询生成一个"假设性答案"hypothetical_doc = llm.invoke(f"请回答以下问题：{user_query}")# 步骤 2：用这个假设性答案的向量去检索（而不是用原查询的向量）# 原理：假设性答案的语义比原查询更接近真实相关文档hypothetical_vector = embeddings.embed_query(hypothetical_doc)results = vectorstore.similarity_search_by_vector(hypothetical_vector, k=5)

为什么 HyDE 有效？

• 原查询：“RAG 怎么优化？”（很短，语义稀疏）
• 假设性答案：包含"检索、分块、重排序、混合搜索"等关键词（语义丰富）
• 向量空间里，假设性答案的向量更接近真实相关文档

1.4 查询路由（Query Routing）

根据查询类型，路由到不同的数据源或检索策略：

def route_query(query: str) -> str: """简单的关键词路由示例""" if any(kw in query for kw in ["比较", "对比", "vs", "区别"]): return "vector_search" # 需要广泛背景，用向量搜索 elif any(kw in query for kw in ["最新", "2025", "近期"]): return "keyword_search" # 有时间敏感性，用关键词精确匹配 elif any(kw in query for kw in ["代码", "实现", "示例"]): return "code_search" # 代码相关，用专用代码检索器 else: return "hybrid_search" # 默认混合搜索

环节 2：检索（Retrieval）

这是查询管道的核心环节。

2.1 查询向量化

# 使用与索引阶段完全相同的嵌入模型！query_vector = embeddings.embed_query(user_query)

2.2 向量相似度计算

# 向量数据库中的 Top-K 检索results = vectorstore.similarity_search_by_vector( embedding=query_vector, k=10, # 先取 10 个候选，后续重排序会进一步筛选)

2.3 全文检索（BM25）—— 向量检索的重要补充

向量检索擅长"语义理解"（同义词、近义词），但在精确关键词匹配上不如传统全文检索。

BM25 算法：
score ( Q , D ) = ∑ _ q _ i ∈ Q IDF ( q _ i ) × f ( q _ i , D ) × ( k _ 1 + 1 ) f ( q _ i , D ) + k _ 1 × ( 1 − b + b × ∣ D ∣ avgdl ) \text{score}(Q, D) = \sum\_{q\_i \in Q} \text{IDF}(q\_i) \times \frac{f(q\_i, D) \times (k\_1 + 1)}{f(q\_i, D) + k\_1 \times (1 - b + b \times \frac{|D|}{\text{avgdl}})}score(Q,D)=∑_q_i∈QIDF(q_i)×f(q_i,D)+k_1×(1−b+b×avgdl∣D∣​)f(q_i,D)×(k_1+1)​

from rank_bm25 import BM25Okapi# 对文本块建立 BM25 索引tokenized_corpus = [list(jieba.cut(doc.page_content)) for doc in all_chunks]bm25 = BM25Okapi(tokenized_corpus)# 查询tokenized_query = list(jieba.cut(user_query))scores = bm25.get_scores(tokenized_query)top_k_docs = sorted(zip(scores, all_chunks), reverse=True)[:10]

2.4 混合检索（Hybrid Search）—— 最佳实践是两者结合

向量检索 → 候选集 A（语义相似） ∪（取并集或 RRF 融合）BM25检索 → 候选集 B（关键词匹配） ↓ 最终候选集

RRF（倒数排序融合）：

RRFScore ( D ) = ∑ _ r ∈ systems 1 k _ r r f + rank _ r ( D ) \text{RRFScore}(D) = \sum\_{r \in \text{systems}} \frac{1}{k\_{rrf} + \text{rank}\_r(D)}RRFScore(D)=∑_r∈systemsk_rrf+rank_r(D)1​

• k _ r r f k\_{rrf}k_rrf通常取60
• 不依赖各系统的原始得分（得分范围可能差异很大），只依赖排名

def rrf_fusion(vector_results, bm25_results, k=60): """RRF 融合排序""" scores = {} # 向量检索结果 for rank, doc in enumerate(vector_results, 1): doc_id = doc.metadata["chunk_id"] scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank) # BM25 检索结果 for rank, doc in enumerate(bm25_results, 1): doc_id = doc.metadata["chunk_id"] scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank) # 按融合得分排序 return sorted(scores.items(), key=lambda x: x[1], reverse=True)

环节 3：重排序（Reranking）—— 性价比最高的优化手段

为什么需要重排序？

召回阶段（向量检索/BM25）为了速度，使用了近似算法，精度有限。
重排序阶段对少量候选文档（如 Top 20）进行精细化排序，显著提升最终质量。

重排序方法对比：

实战：使用 bge-reranker 进行重排序：

from sentence_transformers import CrossEncoder# 加载重排序模型（中文场景推荐 bge-reranker）reranker = CrossEncoder('BAAI/bge-reranker-large')# 对候选文档进行重排序query_doc_pairs = [[user_query, doc.page_content] for doc in candidate_docs]scores = reranker.predict(query_doc_pairs)# 按重排序分数重新排序reranked_docs = [ doc for _, doc in sorted( zip(scores, candidate_docs), key=lambda x: x[0], reverse=True )][:5] # 只保留 Top 5

💡经验值：召回阶段取 Top 20～50，重排序后取 Top 5～10 给 LLM，性价比最高。

环节 4：上下文压缩与过滤（Context Compression）

检索到的文档可能包含冗余信息或无关内容，直接全部塞给 LLM 会：

1. 浪费 Token（成本↑）
2. 引入噪声（质量↓）
3. 超出上下文窗口（报错）

4.1 上下文压缩（Context Compression）

LLMLingua：使用小型 LLM 识别并移除提示中的冗余 Token：

# 使用 LLMLingua 压缩检索结果from llmlingua import PromptCompressorcompressor = PromptCompressor( model_name="microsoft/llmlingua-2-bert-large-chinese", use_llmlingua2=True, # 使用 LLMLingua-2 算法)compressed_context = compressor.compress_prompt( context, rate=0.5, # 压缩率 50% force_tokens=["重要", "关键", "核心"], # 这些 Token 不被压缩)

4.2 上下文过滤

# 简单的相关性过滤：去掉重排序分数低于阈值的文档FILTER_THRESHOLD = 0.3filtered_docs = [ doc for doc, score in zip(reranked_docs, rerank_scores) if score > FILTER_THRESHOLD]# 如果过滤后没有文档了，诚实告知用户if not filtered_docs: return "抱歉，我在知识库中没有找到与您问题相关的信息。"

环节 5：提示构建（Prompt Engineering）

将用户查询和检索到的上下文组装成最终的提示，送给 LLM 生成答案。

标准 RAG 提示模板：

RAG_PROMPT_TEMPLATE = """你是一个专业的技术助手。请**严格根据以下提供的参考文档**来回答用户的问题。规则：1. 如果参考文档中没有相关信息，请明确回答"根据现有资料，无法回答此问题"2. 不要编造任何信息3. 回答时请引用信息来源的文档编号4. 使用清晰的结构化格式（列表、表格等）呈现信息———— 参考文档开始 ————{documents}———— 参考文档结束 ————用户问题：{question}请给出详细、准确的回答："""

动态组装提示：

def build_rag_prompt(query: str, docs: list, max_tokens: int = 4000): """组装 RAG 提示，自动管理上下文窗口""" context_parts = [] total_tokens = 0 # 按相关性从高到低，依次加入上下文 for i, doc in enumerate(docs): doc_tokens = len(doc.page_content) // 2 # 粗略估算 Token 数 if total_tokens + doc_tokens > max_tokens: break context_parts.append(f"[文档 {i+1}]（来源：{doc.metadata['source']}）\n{doc.page_content}") total_tokens += doc_tokens documents_text = "\n\n".join(context_parts) return RAG_PROMPT_TEMPLATE.format( documents=documents_text, question=query )

环节 6：LLM 推理与生成（LLM Inference & Generation）

最终环节，将组装好的提示送给 LLM，生成答案。

def generate_answer(query: str, docs: list) -> str: prompt = build_rag_prompt(query, docs) response = client.chat.completions.create( model="gpt-4o", # 或 DeepSeek-V2、Qwen 等 messages=[ {"role": "system", "content": "你是专业的技术助手，严格根据参考文档回答问题。"}, {"role": "user", "content": prompt} ], temperature=0.1, # RAG 场景建议低温，减少随机性 max_tokens=2000, ) return response.choices[0].message.content

进阶：自适应检索（Self-RAG）

不是所有问题都需要检索——“你好”、"谢谢"这类问题直接回答即可，不需要浪费 RAG 流程。

Self-RAG的思路是让 LLM自己判断是否需要检索：

# Self-RAG 简化版实现def self_rag_answer(query: str) -> str: # 第一步：让 LLM 判断是否需要检索 judge_prompt = f""" 用户问题：{query} 请判断以下问题是否需要查阅外部资料才能准确回答。 只需要回答 YES 或 NO。 """ needs_retrieval = llm.invoke(judge_prompt).strip().upper() == "YES" # 第二步：根据判断结果决定流程 if needs_retrieval: docs = retrieve(query) return generate_answer(query, docs) else: return llm.invoke(query) # 直接回答，跳过检索

四、两条管道的完整数据流

【索引管道 - 离线】原始文档 ↓ Document Loading（加载）原始文本 ↓ Chunking（分块）文本块（500～1000 tokens） ↓ Embedding（向量化）向量 + 文本块 ↓ Vector Storage（存储）向量数据库（可被高效检索）【查询管道 - 在线】用户查询 ↓ Query Processing（查询优化）优化后的查询（可能多个变体） ↓ Retrieval（检索）候选文档集（Top 20～50） ↓ Reranking（重排序）精排后的文档（Top 5～10） ↓ Context Compression（压缩过滤）最终上下文（fit 进 LLM 窗口） ↓ Prompt Engineering（提示构建）最终提示 ↓ LLM Generation（生成答案）最终答案 → 返回用户

五、每个环节的常见坑与最佳实践

六、如何评估你的 RAG 系统？

RAG 评估框架（Ragas）核心指标：

from ragas import evaluatefrom ragas.metrics import ( faithfulness, answer_relevancy, context_precision, context_recall,)# 用标注好的测试集评估results = evaluate( dataset=test_dataset, metrics=[faithfulness, answer_relevancy, context_precision, context_recall],)print(results)

七、总结

RAG 看似简单，实则每个环节都有深度可挖：

索引管道决定知识库的质量上限 ↓查询管道决定每次查询的体验下限

快速提升 RAG 效果的"三板斧"（按投入产出比排序）：

1. 🥇加入重排序（Reranker）—— 几乎零成本，效果立竿见影
2. 🥈混合检索（向量 + BM25 + RRF）—— 解决单一检索的盲区
3. 🥉查询优化（至少加入 HyDE）—— 让用户的"烂查询"也能搜到好结果

RAG 不是"接入向量数据库就完事"——它是一条有很多环节的流水线，最弱的那个环节决定了整个系统的天花板。

学AI大模型的正确顺序，千万不要搞错了

🤔2026年AI风口已来！各行各业的AI渗透肉眼可见，超多公司要么转型做AI相关产品，要么高薪挖AI技术人才，机遇直接摆在眼前！

有往AI方向发展，或者本身有后端编程基础的朋友，直接冲AI大模型应用开发转岗超合适！

就算暂时不打算转岗，了解大模型、RAG、Prompt、Agent这些热门概念，能上手做简单项目，也绝对是求职加分王🔋

📝给大家整理了超全最新的AI大模型应用开发学习清单和资料，手把手帮你快速入门！👇👇

学习路线:

✅大模型基础认知—大模型核心原理、发展历程、主流模型（GPT、文心一言等）特点解析
✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑
✅开发基础能力—Python进阶、API接口调用、大模型开发框架（LangChain等）实操
✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用
✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代
✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经

以上6大模块，看似清晰好上手，实则每个部分都有扎实的核心内容需要吃透！

我把大模型的学习全流程已经整理📚好了！抓住AI时代风口，轻松解锁职业新可能，希望大家都能把握机遇，实现薪资/职业跃迁～

这份完整版的大模型 AI 学习资料已经上传CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】