RAG 进阶 - 企业级知识库实战 / Advanced RAG for Enterprise Knowledge Bases
📅 创建时间：2026-05-08 🏷️ 标签：#RAG进阶 #混合检索 #重排序 #Chunk策略 #RAG评估 📚 前置知识：[[03-rag-basics]]
📋 本章目标
- 掌握文档 Chunk 策略的选择与优化
- 理解混合检索（向量 + 关键词）
- 掌握重排序（Re-ranking）技术
- 理解 Query 改写与扩展
- 掌握 RAG + Agent 的结合方法
- 了解 RAG 效果的评估指标
第1部分：Chunk 策略优化
1.1 为什么 Chunk 策略很重要？
┌─────────────────────────────────────────────────────────────┐
│ Chunk 的重要性 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Chunk 太小： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ "年假" ← 单独一个块，缺少上下文 │ │
│ │ → 检索到了但不知道是什么意思 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Chunk 太大： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ [整个公司手册 500 页] ← 一个巨大的块 │ │
│ │ → 检索时"整块"匹配，不够精准 │ │
│ │ → 塞给 LLM 时超出上下文 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 理想的 Chunk： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 每个块是一个完整的语义单元 │ │
│ │ "年假政策：工作满1年享受5天年假，满3年享受10天..." │ │
│ │ → 有头有尾，语义完整 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘1.2 常见 Chunk 策略
策略1：固定大小分块（Fixed Size Chunking）
python
from langchain.text_splitter import CharacterTextSplitter
# 最简单，按字符数分割
splitter = CharacterTextSplitter(
chunk_size=500, # 每块 500 字符
chunk_overlap=50, # 块之间重叠 50 字符（保留上下文）
separator="\n" # 尽量在换行符处分割
)
chunks = splitter.split_text(long_document)┌─────────────────────────────────────────────────────────────┐
│ 固定大小分块示意 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 原文： │
│ [AAA BBB CCC DDD EEE FFF GGG HHH III JJJ KKK] │
│ │
│ chunk_size=5, chunk_overlap=2 │
│ │
│ Chunk 1: [AAA BBB CCC DDD EEE] │
│ Chunk 2: [DDD EEE FFF GGG HHH] ← DDD EEE 与 Chunk1 重叠 │
│ Chunk 3: [GGG HHH III JJJ KKK] │
│ │
│ 优点：简单可控 │
│ 缺点：可能在句子中间断开 │
│ │
└─────────────────────────────────────────────────────────────┘策略2：递归分块（Recursive Chunking）
python
from langchain.text_splitter import RecursiveCharacterTextSplitter
# 按层级尝试不同分隔符
splitter = RecursiveCharacterTextSplitter(
chunk_size=500,
chunk_overlap=50,
# 按优先级尝试分隔符：段落 → 句子 → 词
separators=["\n\n", "\n", "。", "！", "？", ". ", " ", ""]
)
chunks = splitter.split_text(document)策略3：语义分块（Semantic Chunking）
python
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings
# 按语义相似度分割
splitter = SemanticChunker(
embeddings=OpenAIEmbeddings(),
breakpoint_threshold_amount=0.95 # 相似度突变时分割
)
chunks = splitter.split_text(document)策略4：按文档结构分块
python
from langchain_community.document_loaders import UnstructuredHTMLLoader
# 从 HTML 中提取标题和内容
loader = UnstructuredHTMLLoader("doc.html")
docs = loader.load()
# 按 <h1>, <h2> 等结构分割
from langchain.text_splitter import MarkdownTextSplitter
splitter = MarkdownTextSplitter(chunk_size=500)
chunks = splitter.split_documents(docs)1.3 Chunk 策略选择指南
┌─────────────────────────────────────────────────────────────┐
│ Chunk 策略选择指南 │
├─────────────────────────────────────────────────────────────┤
│ │
│ │ 文档类型 │ 推荐策略 │ chunk_size │
│ ├────────────────┼────────────────────┼─────────────────────┤
│ │ 规范文档 │ 语义分块 / 标题分块 │ 300-500 │
│ │ 技术文档 │ 标题分块 │ 500-800 │
│ │ 聊天记录 │ 固定大小分块 │ 500左右 │
│ │ 代码 │ 按函数/类分块 │ 取决于函数大小 │
│ │ 书籍/长文 │ 章节 + 语义分块 │ 500-1000 │
│ │
│ 通用建议： │
│ • chunk_overlap 通常设为 chunk_size 的 10-20% │
│ • 检索场景偏小（300-500），生成场景可偏大（500-1000） │
│ • 记得用 metadata 保留来源信息 │
│ │
└─────────────────────────────────────────────────────────────┘第2部分：混合检索
2.1 为什么需要混合检索？
┌─────────────────────────────────────────────────────────────┐
│ 单一检索的局限 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 向量检索的问题： │
│ 用户问："张三和李四的矛盾" │
│ → "矛盾"语义匹配到了"王五和陈六打架" │
│ → 但用户想找的是"矛盾纠纷调解"的文档 │
│ → 用词不同，向量相似度高但语义不匹配 │
│ │
│ 关键词检索的问题： │
│ 用户问："AI Agent 最新进展" │
│ → 文档写的是"人工智能智能体" │
│ → "Agent" 和"智能体"含义相同但文字不同 │
│ → 关键词匹配失败 │
│ │
└─────────────────────────────────────────────────────────────┘解决方案：向量 + 关键词 = 混合检索
┌─────────────────────────────────────────────────────────────┐
│ 混合检索示意 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户问题："公司年假政策" │
│ │
│ ┌──────────────────┐ ┌──────────────────┐ │
│ │ 向量检索 │ │ 关键词检索 │ │
│ │ 语义相似度匹配 │ │ BM25 / TF-IDF │ │
│ │ │ │ │ │
│ │ "年假安排" 0.92 │ │ "年假" score 15 │ │
│ │ "请假流程" 0.85 │ │ "政策" score 12 │ │
│ │ "团建" 0.30 │ │ "团建" score 3 │ │
│ └────────┬─────────┘ └────────┬─────────┘ │
│ │ │ │
│ └───────────┬────────────┘ │
│ ↓ │
│ ┌────────────────┐ │
│ │ 融合结果 │ │
│ │ 权重加权 │ │
│ │ RRF / 分数加总 │ │
│ └────────┬───────┘ │
│ ↓ │
│ 最终排序：[年假安排, 请假流程, ...] │
│ │
└─────────────────────────────────────────────────────────────┘2.2 混合检索实现
python
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
# 向量检索器
vectorstore = Chroma(persist_directory="./vector_db")
vector_retriever = vectorstore.as_retriever(
search_kwargs={"k": 5}
)
# 关键词检索器（需要先对文本进行 tokenize）
texts = ["文档1内容...", "文档2内容...", ...]
bm25_retriever = BM25Retriever.from_texts(
texts,
embeddings=OpenAIEmbeddings()
)
bm25_retriever.k = 5
# 混合检索器
ensemble_retriever = EnsembleRetriever(
retrievers=[vector_retriever, bm25_retriever],
weights=[0.5, 0.5] # 各占 50%
)
# 使用
results = ensemble_retriever.invoke("公司年假政策")2.3 RRF 融合算法
RRF（Reciprocal Rank Fusion）= 简单有效的多检索结果融合
python
def rrf_fusion(results_list, k=60):
"""RRF 融合多个检索结果"""
scores = {}
for results in results_list:
for rank, doc in enumerate(results):
doc_id = doc["id"]
# RRF 公式：1 / (k + rank)
score = 1 / (k + rank)
scores[doc_id] = scores.get(doc_id, 0) + score
# 按分数排序
sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [doc_id for doc_id, _ in sorted_docs]┌─────────────────────────────────────────────────────────────┐
│ RRF 示例 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 向量检索排名：[Doc_A, Doc_B, Doc_C] │
│ 关键词检索排名：[Doc_B, Doc_D, Doc_A] │
│ │
│ k=60 计算 RRF 分数： │
│ Doc_A: 1/(60+0) + 1/(60+2) = 0.0161 + 0.0161 = 0.0322 │
│ Doc_B: 1/(60+1) + 1/(60+0) = 0.0164 + 0.0167 = 0.0331 │
│ Doc_C: 1/(60+2) + 0 = 0.0161 │
│ Doc_D: 0 + 1/(60+1) = 0.0164 │
│ │
│ 最终排名：Doc_B > Doc_A > Doc_C > Doc_D │
│ │
└─────────────────────────────────────────────────────────────┘第3部分：重排序（Re-ranking）
3.1 什么是 Re-ranking？
┌─────────────────────────────────────────────────────────────┐
│ 两阶段检索示意 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 阶段1：向量检索（召回） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 从 100 万文档中快速召回 Top 100 相关文档 │ │
│ │ 目标：不全漏，但可能不精准 │ │
│ │ 方法：向量相似度（ANN 近似最近邻） │ │
│ └─────────────────────────────────────────────────────┘ │
│ ↓ │
│ 阶段2：重排序（精排） │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ 对 Top 100 精细排序，取 Top 5 │ │
│ │ 目标：精准排序 │ │
│ │ 方法：Cross-Encoder（更准确但更慢） │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘3.2 Cross-Encoder vs Bi-Encoder
┌─────────────────────────────────────────────────────────────┐
│ Bi-Encoder vs Cross-Encoder │
├─────────────────────────────────────────────────────────────┤
│ │
│ Bi-Encoder（向量检索）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Query → Encoder → Vector_Q │ │
│ │ Doc → Encoder → Vector_D │ │
│ │ Sim(V_Q, V_D) = 相似度 │ │
│ │ │ │
│ │ 优点：Doc 向量可预计算，检索快（毫秒级） │ │
│ │ 缺点：Query 和 Doc 独立编码，交互不足 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ Cross-Encoder（重排序）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ [Query; Doc] → Encoder → Score │ │
│ │ │ │
│ │ Query 和 Doc 一起编码，充分交互 │ │
│ │ │ │
│ │ 优点：准确性高，能捕捉细粒度匹配 │ │
│ │ 缺点：慢，需要对每个 Doc 单独计算 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 最佳实践：向量检索快速召回 → Cross-Encoder 精细排序 │
│ │
└─────────────────────────────────────────────────────────────┘3.3 Re-ranking 实现
python
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank
# 或使用 sentence-transformers
# 使用 Cohere 的 Rerank 模型
reranker = CohereRerank(
model="rerank-multilingual-v2.0",
top_n=5 # 返回 Top 5
)
# 包装向量检索器
compression_retriever = ContextualCompressionRetriever(
base_compressor=reranker,
base_retriever=vector_retriever
)
# 使用
results = compression_retriever.invoke("公司年假有多少天？")第4部分：Query 改写
4.1 Query 改写的必要性
┌─────────────────────────────────────────────────────────────┐
│ Query 改写的价值 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户原始问题："我想休年假要怎么弄" │
│ │
│ 问题： │
│ • 口语化表达 │
│ • 可能用了公司内部的叫法（"年假" = "带薪休假"） │
│ • 没有明确说要查文档 │
│ │
│ Query 改写后： │
│ "公司的带薪休假政策、申请流程、计算方式" │
│ │
│ → 更适合检索 │
│ → 提高召回率 │
│ │
└─────────────────────────────────────────────────────────────┘4.2 Query 改写技术
技术1：HyDE（Hypothetical Document Embeddings）
python
def hyde_rewrite(query, llm):
"""HyDE：让 LLM 先写一个假设答案，再用答案检索"""
# Step 1: 生成假设答案
prompt = f"""基于以下问题，生成一个假设的答案片段：
问题：{query}
要求：
1. 假设这个答案来自公司内部文档
2. 包含具体的政策条款、流程步骤
3. 写 2-3 句话即可
假设答案："""
hypothetical_answer = llm.invoke(prompt)
# Step 2: 用假设答案检索
results = vectorstore.similarity_search(hypothetical_answer, k=5)
return results技术2：Query 扩展
python
def expand_query(query, llm):
"""Query 扩展：生成多个相关查询"""
prompt = f"""将以下用户问题改写成 3 个不同的检索查询：
原问题：{query}
要求：
1. 保留原意
2. 用不同的表述方式
3. 包含可能的同义词
查询1：
查询2：
查询3："""
expanded = llm.invoke(prompt).content
queries = parse_queries(expanded) # 解析出多个查询
# 并行检索
all_results = []
for q in queries:
results = vectorstore.similarity_search(q, k=3)
all_results.extend(results)
return deduplicate(all_results) # 去重第5部分：RAG + Agent
5.1 Agent 何时需要 RAG？
┌─────────────────────────────────────────────────────────────┐
│ RAG + Agent 示意 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 用户问："公司去年 Q3 的营收是多少？" │
│ │
│ 普通 Agent： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ → LLM："我的知识截止到 2024 年 12 月， │ │
│ │ 无法获取实时财务数据" │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ RAG + Agent： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Step 1: Agent 决定需要查财务数据 │ │
│ │ Step 2: 调用 RAG 检索公司财务报告 │ │
│ │ Step 3: 获取 "Q3 营收 = 1.2 亿" │ │
│ │ Step 4: Agent 整合回答："Q3 营收 1.2 亿，增长 15%" │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘5.2 RAG + Agent 实现
python
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.tools import Tool
# RAG 工具
def rag_retriever(query):
"""RAG 检索工具"""
docs = vectorstore.similarity_search(query, k=3)
return "\n\n".join([doc.page_content for doc in docs])
# 注册为工具
tools = [
Tool(
name="company_knowledge",
func=rag_retriever,
description="""当用户问及公司政策、流程、文档相关内容时使用。
输入应该是用户的具体问题。
返回相关文档片段。"""
),
# ... 其他工具
]
# 创建 Agent
agent = create_openai_functions_agent(llm, tools, system_prompt)
executor = AgentExecutor(agent=agent, tools=tools)
# Agent 会自动决定何时调用 RAG
result = executor.invoke({
"input": "我的年假怎么计算？入职两年了"
})5.3 自主判断是否检索
python
system_prompt = """你是一个公司内部助手。
你有以下工具可用：
- company_knowledge：查询公司政策、文档
- calculator：进行计算
- email_sender：发送邮件
决策规则：
1. 如果用户问的是公司政策/文档 → 必须调用 company_knowledge
2. 如果用户问的是计算问题 → 调用 calculator
3. 如果用户只是闲聊 → 直接回答
每次收到问题，先判断是否需要检索，然后执行。"""第6部分：RAG 评估
6.1 评估指标
┌─────────────────────────────────────────────────────────────┐
│ RAG 评估指标 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 检索阶段（Retrieval）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Hit Rate：正确的文档有没有被检索到 │ │
│ │ MRR（Mean Reciprocal Rank）：正确文档排第几 │ │
│ │ NDCG：综合考虑相关性和排序 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
│ 生成阶段（Generation）： │
│ ┌─────────────────────────────────────────────────────┐ │
│ │ Faithfulness：回答是否忠实于检索到的内容 │ │
│ │ Answer Relevancy：回答和问题的相关性 │ │
│ │ Context Precision：上下文排序是否正确 │ │
│ └─────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────┘6.2 RAGAS 评估框架
python
from ragas import EvaluationDataset
from ragas.metrics import (
faithfulness,
answer_relevancy,
context_precision,
context_recall
)
from ragas import evaluate
# 准备评估数据
eval_dataset = EvaluationDataset.from_dict({
"user_input": ["问题1", "问题2", ...],
"retrieved_contexts": [[doc1, doc2], [doc3, doc4], ...],
"response": ["回答1", "回答2", ...],
"reference": ["标准答案1", "标准答案2", ...]
})
# 运行评估
result = evaluate(
dataset=eval_dataset,
metrics=[
faithfulness,
answer_relevancy,
context_precision,
context_recall
]
)
print(result)核心总结
总结1：Chunk 策略
| 策略 | 适用场景 |
|---|---|
| 固定大小 | 通用场景，简单可控 |
| 递归分块 | 尽量在语义边界分割 |
| 语义分块 | 需要完整语义单元 |
| 按标题分块 | 结构化文档（Markdown/HTML） |
总结2：混合检索
向量检索（语义） + 关键词检索（精确） → RRF 融合
→ 两者取长补短，召回率和精确率兼顾总结3：两阶段检索
阶段1：向量检索快速召回 Top-N
阶段2：Cross-Encoder 精细排序 Top-K（K << N）章节测试
测试1：Chunk 策略
固定大小分块的优缺点是什么？
测试2：混合检索
为什么需要混合检索？向量检索和关键词检索各有什么局限性？
测试3：Re-ranking
Bi-Encoder 和 Cross-Encoder 的区别是什么？为什么要结合使用？
测试4：Query 改写
HyDE 的核心思想是什么？
测试5：RAG 评估
Hit Rate 和 MRR 分别衡量什么？
参考答案
测试1答案
答案：
- 优点：简单可控，分割均匀，适合大多数场景
- 缺点：可能在句子中间断开，破坏语义完整性
测试2答案
答案：
- 向量检索局限：依赖语义相似度，用词不同时可能匹配不到（如"年假"vs"带薪休假"）
- 关键词检索局限：无法处理同义词和语义关联（如"Agent"vs"智能体"）
- 混合检索：结合两者优势，用关键词确保精确匹配，用向量确保语义相关
测试3答案
答案：
- Bi-Encoder：Query 和 Doc 分别编码，Doc 向量可预计算，检索快但交互不足
- Cross-Encoder：Query 和 Doc 一起编码，充分交互，准确但慢
- 结合使用：先用 Bi-Encoder 快速召回（如 Top 100），再用 Cross-Encoder 精细排序（如 Top 5）
测试4答案
答案：HyDE 让 LLM 先根据问题生成一个假设的答案文档，然后用这个假设答案去检索。这样可以利用 LLM 的推理能力生成更接近真实文档风格的检索词。
测试5答案
答案：
- Hit Rate：衡量检索"有没有找到"——正确文档是否出现在结果集中
- MRR（Mean Reciprocal Rank）：衡量检索"找到得多快"——正确文档排在第几位，越靠前分数越高
相关笔记
- [[03-rag-basics]] - RAG 基础概念
- [[05-agent-workflow]] - RAG + Agent 的具体应用
- [[04-memory-management]] - 记忆与知识库的结合
下一步学习
- [ ] 阅读 08 - 真实应用场景
学习状态：🟡 开始学习