RAG调优实战:12个让检索增强生成从"能用"到"好用"的关键策略

你: 费了九牛二虎之力搭建了一个RAG系统
老板: 这检索结果怎么这么离谱?
你: (╯°□°)╯︵ ┻━┻ 明明demo挺好的啊!

import re
from unidecode import unidecode
 
def clean_text(text):
    """数据清洗示例"""
    # 1. 移除多余空白字符
    text = re.sub(r'\s+', ' ', text)
 
    # 2. 修正特殊字符编码
    text = text.encode('utf-8', errors='ignore').decode('utf-8')
 
    # 3. 统一标点符号
    text = text.replace('…', '...')
 
    # 4. 移除冗余信息
    text = re.sub(r'Page \d+ of \d+', '', text)  # 移除页码
 
    return text.strip()
 
# 使用示例
raw_text = "这是一段   测试文本…    Page 1 of 10"
clean = clean_text(raw_text)
print(clean)  # "这是一段 测试文本..."

from langchain.text_splitter import RecursiveCharacterTextSplitter
 
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,        # 每块大小
    chunk_overlap=50,      # 块间重叠
    separators=["\n\n", "\n", "。", " ", ""]  # 分割符优先级
)
 
chunks = text_splitter.split_text(long_document)

from langchain.text_splitter import (
    PythonCodeTextSplitter,      # 代码文件
    MarkdownTextSplitter,        # Markdown文档
    RecursiveCharacterTextSplitter  # 通用文本
)
 
# 示例:为代码文件使用专用分割器
code_splitter = PythonCodeTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
 
code_chunks = code_splitter.split_text(python_code)

[块1: 第1-5段]
           [块2: 第4-8段]  ← 注意overlap
                      [块3: 第7-11段]

from sentence_transformers import SentenceTransformer
from sentence_transformers import InputExample, losses
from torch.utils.data import DataLoader
 
# 准备训练数据
train_examples = [
    InputExample(texts=['查询1', '相关文档1']),
    InputExample(texts=['查询2', '相关文档2']),
]
 
# 加载基础模型
model = SentenceTransformer('all-MiniLM-L6-v2')
 
# 定义损失函数
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
train_loss = losses.MultipleNegativesRankingLoss(model)
 
# 微调
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=3
)

from langchain.vectorstores import Weaviate
from langchain.embeddings import OpenAIEmbeddings
 
# 为文档添加元数据
documents = [
    {
        "content": "Kubernetes是容器编排系统...",
        "metadata": {
            "source": "k8s-official-docs",
            "category": "infrastructure",
            "date": "2025-01-01",
            "section": "入门指南",
            "version": "v1.28"
        }
    }
]
 
# 存储时保留元数据
vectorstore = Weaviate.from_documents(
    documents=documents,
    embedding=OpenAIEmbeddings(),
    metadatas=[doc["metadata"] for doc in documents]
)
 
# 检索时使用元数据过滤
results = vectorstore.similarity_search(
    query="如何部署K8s?",
    filter={"category": "infrastructure", "version": "v1.28"},
    k=3
)

知识库结构:
├── 技术文档/        ← 索引1: tech_docs
├── 产品手册/        ← 索引2: product_manuals
├── API参考/         ← 索引3: api_reference
└── 常见问题/        ← 索引4: faq

from langchain.vectorstores import Weaviate
 
# 创建多个索引
tech_docs_index = Weaviate(..., index_name="tech_docs")
api_index = Weaviate(..., index_name="api_reference")
faq_index = Weaviate(..., index_name="faq")
 
# 智能路由函数
def route_query(query: str):
    """根据查询内容路由到合适的索引"""
    if "API" in query or "接口" in query:
        return api_index
    elif "?" in query or "如何" in query:
        return faq_index
    else:
        return tech_docs_index
 
# 使用路由
query = "如何调用登录API?"
selected_index = route_query(query)
results = selected_index.similarity_search(query)

KNN: 精确但慢 (O(n))
ANN: 近似但快 (O(log n))
 
精度损失: 通常<5%
速度提升: 10-100倍

import weaviate
 
client = weaviate.Client("http://localhost:8080")
 
# 创建collection时配置HNSW参数
client.schema.create_class({
    "class": "Document",
    "vectorIndexConfig": {
        "ef": 64,                  # 搜索时的候选数量
        "efConstruction": 128,     # 构建时的候选数量
        "maxConnections": 64       # 每个节点的最大连接数
    }
})

用户问题: "这玩意儿咋用啊?"
→ 太口语化,embedding效果差
 
用户问题: "介绍一下Docker和K8s的区别,以及它们各自的优缺点和适用场景"
→ 太复杂,一次检索很难覆盖所有方面

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
 
# 查询重写Prompt
rewrite_prompt = ChatPromptTemplate.from_template(
    """将下面的口语化问题改写为更正式、更适合检索的表达:
 
原始问题: {query}
 
改写后的问题:"""
)
 
llm = ChatOpenAI(temperature=0)
 
# 示例
original_query = "这玩意儿咋用啊?"
rewritten = llm(rewrite_prompt.format(query=original_query))
# 输出: "如何使用这个工具?"

# HyDE Prompt
hyde_prompt = ChatPromptTemplate.from_template(
    """请为以下问题生成一个详细的假设性答案:
 
问题: {query}
 
假设性答案:"""
)
 
# 生成假设答案
query = "Kubernetes的核心组件有哪些?"
hypothetical_answer = llm(hyde_prompt.format(query=query))
 
# 用假设答案 + 原始查询一起检索
combined_query = f"{query}\n{hypothetical_answer}"
results = vectorstore.similarity_search(combined_query)

# 子查询分解
decompose_prompt = ChatPromptTemplate.from_template(
    """将以下复杂问题分解为3-5个简单的子问题:
 
复杂问题: {query}
 
子问题列表:
1."""
)
 
complex_query = "介绍Docker和K8s的区别、优缺点和适用场景"
sub_queries = llm(decompose_prompt.format(query=complex_query))
 
# 输出:
# 1. Docker是什么?
# 2. Kubernetes是什么?
# 3. Docker和Kubernetes有什么区别?
# 4. Docker的优缺点是什么?
# 5. Kubernetes的优缺点是什么?
 
# 分别检索后合并结果
all_results = []
for sub_q in sub_queries:
    results = vectorstore.similarity_search(sub_q, k=2)
    all_results.extend(results)

# 纯语义搜索
results = vectorstore.similarity_search(
    query="机器学习",
    k=5
)
 
# 混合搜索(语义 + 关键词)
from langchain.retrievers import BM25Retriever, EnsembleRetriever
 
bm25_retriever = BM25Retriever.from_documents(documents)
vector_retriever = vectorstore.as_retriever()
 
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # alpha=0.7表示70%语义,30%关键词
)
 
results = ensemble_retriever.get_relevant_documents("机器学习")

# 调整返回结果数量
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 5}  # 返回Top 5结果
)

k=3-5:  适合问答场景,上下文窗口小
k=5-10: 适合摘要场景,需要更多信息
k=10+:  配合重排序使用,先召回再精排

# 选择相似度计算方法
vectorstore = Weaviate.from_documents(
    documents=docs,
    embedding=embeddings,
    distance_metric="cosine"  # cosine, dot_product, euclidean
)

理想做法:
检索用小块(精准定位) → 生成用大块(完整上下文)

# 示例:检索到第5句后,返回第3-7句
retrieved_sentence_id = 5
window_size = 2
 
context = sentences[
    retrieved_sentence_id - window_size :
    retrieved_sentence_id + window_size + 1
]

文档: [句1] [句2] [句3] [句4] [句5] [句6] [句7] [句8]
 
检索命中: [句5] ✅
 
实际返回: [句3] [句4] [句5] [句6] [句7]  ← 带窗口

文档树:
章节1
  ├── 小节1.1
  │    ├── 段落1.1.1
  │    └── 段落1.1.2
  └── 小节1.2
       ├── 段落1.2.1
       └── 段落1.2.2

# 如果检索到的多个段落属于同一小节,则合并返回整个小节
 
检索结果:
- 段落1.1.1 ✅
- 段落1.1.2 ✅
 
自动合并 → 返回整个"小节1.1"

查询: "如何优化Docker镜像大小?"
 
检索结果(按相似度):
1. "Docker镜像的构建过程..." ← 相似但不太相关
2. "优化Docker镜像的5个技巧..." ← 高度相关 ✅
3. "Docker镜像的安全性..." ← 相似但不相关

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank
 
# 初始检索器
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 20})
 
# Cohere重排序模型
compressor = CohereRerank(
    model="rerank-english-v2.0",
    top_n=5  # 从20个结果中重排后选Top 5
)
 
# 组合检索器
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever
)
 
# 使用
docs = compression_retriever.get_relevant_documents(
    "如何优化Docker镜像大小?"
)

1. 初始检索: Top 20 (召回率优先)
      ↓
2. 重排序模型: 计算相关性分数
      ↓
3. 精选输出: Top 5 (精准度优先)

微调LLM: 调整模型风格、指令遵循能力
   +
RAG系统: 提供最新知识和事实信息
   =
最优方案

prompt_template = """你是一个专业的技术问答助手。
 
重要指示:
1. 请**仅基于**下面提供的上下文信息回答问题
2. 如果上下文中没有相关信息,明确说"根据提供的信息无法回答"
3. 回答时请引用上下文中的具体内容
4. 保持简洁,最多3-4句话
 
上下文信息:
{context}
 
用户问题: {question}
 
你的回答:"""

❌ 差的Prompt: "回答问题"
 
✅ 好的Prompt: "基于上下文回答,并说明你的答案来自哪一段"

prompt_with_examples = """你是技术问答助手。参考以下示例回答:
 
示例1:
问题: Docker和VM的区别?
上下文: Docker使用容器技术...VM使用虚拟机...
回答: 根据上下文,Docker使用容器技术共享主机内核,而VM使用虚拟机技术有独立内核。Docker更轻量,VM隔离性更好。
 
示例2:
问题: 如何扩展Kubernetes集群?
上下文: [无相关信息]
回答: 抱歉,提供的上下文中没有关于Kubernetes集群扩展的信息。
 
现在请回答:
问题: {question}
上下文: {context}
回答:"""

上下文结构:
[文档1] ← LLM注意力高
[文档2]
[文档3] ← 关键信息在这里,但LLM可能忽略!
[文档4]
[文档5] ← LLM注意力高

# 策略1: 限制上下文数量
retriever = vectorstore.as_retriever(
    search_kwargs={"k": 3}  # 不要超过5个
)
 
# 策略2: 重排序,把最相关的放在开头和结尾
def reorder_documents(docs):
    """将最相关的文档放在开头和结尾"""
    if len(docs) <= 2:
        return docs
 
    # 最相关的放开头
    # 次相关的放结尾
    # 其余放中间
    return [docs[0]] + docs[2:-1] + [docs[1]]

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Weaviate
from langchain.chat_models import ChatOpenAI
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import CohereRerank
from langchain.prompts import ChatPromptTemplate
from langchain.schema.runnable import RunnablePassthrough
from langchain.schema.output_parser import StrOutputParser
 
# 1. 数据清洗和分块
from langchain.text_splitter import RecursiveCharacterTextSplitter
 
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", " "]
)
 
chunks = text_splitter.split_documents(documents)
 
# 2. 向量化和存储(使用高质量嵌入模型)
embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
vectorstore = Weaviate.from_documents(
    documents=chunks,
    embedding=embeddings,
    metadatas=[chunk.metadata for chunk in chunks]
)
 
# 3. 混合检索 + 重排序
base_retriever = vectorstore.as_retriever(
    search_type="mmr",  # 使用MMR提高多样性
    search_kwargs={
        "k": 20,
        "fetch_k": 50,
        "lambda_mult": 0.7
    }
)
 
# 重排序
reranker = CohereRerank(top_n=5)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=base_retriever
)
 
# 4. 优化的Prompt
prompt = ChatPromptTemplate.from_template("""你是一个专业的技术问答助手。
 
重要规则:
1. 仅基于下面的上下文信息回答
2. 如果信息不足,明确说明
3. 引用具体的上下文内容
4. 保持简洁(3-4句话)
 
上下文:
{context}
 
问题: {question}
 
回答:""")
 
# 5. LLM
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
 
# 6. 构建RAG链
rag_chain = (
    {
        "context": compression_retriever,
        "question": RunnablePassthrough()
    }
    | prompt
    | llm
    | StrOutputParser()
)
 
# 使用
response = rag_chain.invoke("如何优化Docker镜像大小?")
print(response)

import mlflow
 
# 开启实验跟踪
mlflow.set_experiment("rag-optimization")
 
with mlflow.start_run():
    # 记录参数
    mlflow.log_params({
        "chunk_size": 500,
        "chunk_overlap": 50,
        "embedding_model": "text-embedding-3-large",
        "retrieval_k": 20,
        "rerank_top_n": 5,
        "llm_model": "gpt-4-turbo",
        "llm_temperature": 0
    })
 
    # 运行RAG系统
    results = evaluate_rag_system(test_queries)
 
    # 记录指标
    mlflow.log_metrics({
        "avg_retrieval_precision": results['precision'],
        "avg_answer_accuracy": results['accuracy'],
        "avg_latency_ms": results['latency']
    })

✅ 数据清洗
   → 确保数据整洁、正确、完整
 
✅ 分块
   → chunk_size: 200-1000(根据场景)
   → chunk_overlap: 10-20%
   → 选择合适的分割器
 
✅ 嵌入模型
   → 参考MTEB排行榜
   → 考虑微调(可提升5-10%)
 
✅ 元数据
   → 添加时间、来源、分类等标签
   → 支持后续过滤
 
✅ 多索引
   → 为不同类型数据创建独立索引
   → 实现智能路由
 
✅ 索引算法
   → HNSW, IVF等ANN算法
   → 通常使用默认配置即可

✅ 查询转换
   → 查询重写
   → HyDE(假设性文档嵌入)
   → 子查询分解
 
✅ 检索参数
   → 语义搜索 vs 混合搜索(alpha参数)
   → Top-K数量(3-20)
   → 相似度度量(根据embedding模型)
 
✅ 高级检索策略
   → 句子窗口检索
   → 自动合并检索
 
✅ 重排序模型
   → Cohere Rerank等
   → 先召回(k=20),再精排(top_n=5)
 
✅ LLM微调
   → 调整风格和指令遵循
   → 与RAG结合使用
 
✅ Prompt工程
   → 强制基于上下文回答
   → 使用Few-Shot示例
   → 注意"中间迷失"问题