RAG系统进阶与优化完全教程

1. RAG系统回顾与挑战分析

检索增强生成（Retrieval-Augmented Generation）已成为大语言模型落地的核心范式。基础RAG流程遵循"索引→检索→生成"三步走，但在真实业务场景中，这套简单管线暴露出诸多问题：

• 检索质量不稳定：用户查询与文档表述之间存在语义鸿沟，简单向量相似度匹配经常返回不相关内容
• 分块策略粗糙：固定窗口切分破坏语义完整性，上下文信息丢失严重
• 上下文窗口浪费：检索到的片段存在大量冗余，挤占了LLM的有效推理空间
• 无法自我纠错：当检索结果有误时，系统缺乏反思和修正能力
• 评估困难：缺少系统化的端到端评估手段，难以量化每个环节的改进效果

一个典型的简单RAG实现如下：

from langchain.vectorstores import FAISS
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 基础RAG管线
vectorstore = FAISS.load_local("my_index", OpenAIEmbeddings())
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(temperature=0),
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
)
answer = qa_chain.run("公司的退款政策是什么？")

这段代码能跑，但距离生产级还差十万八千里。以下逐步拆解每个环节的进阶优化手段。

2. 高级分块策略

分块是RAG质量的第一道关卡。目标是在保持语义完整性的同时，将文档切成检索友好的片段。

语义分块

基于嵌入向量的相似度变化来判断语义边界，而非固定字数：

import numpy as np
from langchain.text_splitter import SemanticChunker
from langchain.embeddings import OpenAIEmbeddings

embeddings = OpenAIEmbeddings()

# 基于语义断点的自适应分块
semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",  # 使用百分位数作为阈值
    breakpoint_threshold_amount=75,           # 相似度低于75%百分位时切分
    min_chunk_size=100,                       # 最小块大小
)

chunks = semantic_splitter.split_text(long_document)

自建语义分块的核心逻辑：

from sentence_transformers import SentenceTransformer
from sklearn.metrics.pairwise import cosine_similarity

model = SentenceTransformer("all-MiniLM-L6-v2")

def semantic_chunk(text: str, threshold: float = 0.5, min_size: int = 50) -> list[str]:
    """基于嵌入相似度跳跃的语义分块"""
    sentences = [s.strip() for s in text.replace("\n", " ").split(". ") if s.strip()]
    if len(sentences) <= 1:
        return [text]

    embeddings = model.encode(sentences)
    similarities = [
        cosine_similarity([embeddings[i]], [embeddings[i + 1]])[0][0]
        for i in range(len(embeddings) - 1)
    ]

    chunks, current_chunk = [], [sentences[0]]
    for i, sim in enumerate(similarities):
        if sim < threshold and len(". ".join(current_chunk)) > min_size:
            chunks.append(". ".join(current_chunk) + ".")
            current_chunk = [sentences[i + 1]]
        else:
            current_chunk.append(sentences[i + 1])

    if current_chunk:
        chunks.append(". ".join(current_chunk))
    return chunks

层次分块

先按文档结构（章节/段落）粗切，再对长段落细切，保留父子关系：

from langchain.text_splitter import RecursiveCharacterTextSplitter

class HierarchicalChunker:
    """两层分块：粗粒度(parent) + 细粒度(child)"""

    def __init__(self, parent_size=2000, child_size=400, overlap=50):
        self.parent_splitter = RecursiveCharacterTextSplitter(
            chunk_size=parent_size, chunk_overlap=100
        )
        self.child_splitter = RecursiveCharacterTextSplitter(
            chunk_size=child_size, chunk_overlap=overlap
        )

    def split(self, text: str) -> list[dict]:
        parents = self.parent_splitter.split_text(text)
        result = []
        for parent_idx, parent in enumerate(parents):
            children = self.child_splitter.split_text(parent)
            for child_idx, child in enumerate(children):
                result.append({
                    "text": child,
                    "metadata": {
                        "parent_id": parent_idx,
                        "child_id": child_idx,
                        "parent_text": parent[:200] + "...",  # 父块摘要用于重排
                    }
                })
        return result

自适应分块

根据内容类型动态选择分块策略：

def adaptive_chunk(text: str, content_type: str = "auto") -> list[str]:
    """根据内容类型自适应选择分块策略"""
    if content_type == "auto":
        content_type = detect_content_type(text)

    if content_type == "code":
        # 代码按函数/类切分
        return code_aware_split(text)
    elif content_type == "table":
        # 表格保持行完整性
        return table_aware_split(text)
    elif content_type == "legal":
        # 法律文档按条款切分
        return clause_based_split(text)
    else:
        # 通用文本用语义分块
        return semantic_chunk(text)

def detect_content_type(text: str) -> str:
    """简单启发式内容类型检测"""
    code_indicators = ["def ", "class ", "import ", "function ", "const ", "var "]
    if any(ind in text for ind in code_indicators):
        return "code"
    if text.count("|") > text.count("\n") * 0.5:
        return "table"
    if any(kw in text for kw in ["第.*条", "甲方", "乙方", "协议"]):
        return "legal"
    return "general"

3. 查询改写与扩展

用户的原始查询往往模糊、口语化或信息不足。查询改写在检索前"翻译"用户意图，显著提升召回率。

HyDE（假设文档嵌入）

先让LLM生成一个假设性的回答，用这个回答去检索，而非原始问题：

from langchain.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

hyde_prompt = ChatPromptTemplate.from_template(
    "请根据以下问题，写一段可能出现在技术文档中的回答（约150字）：\n\n问题：{question}"
)

def hyde_retrieve(question: str, retriever, llm=None):
    """HyDE检索：生成假设文档 → 嵌入 → 检索"""
    llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    chain = hyde_prompt | llm
    hypothetical_doc = chain.invoke({"question": question}).content

    # 用假设文档的嵌入去检索，而非原始问题
    results = retriever.invoke(hypothetical_doc)
    return results

Step-back Prompting

将具体问题抽象为更高层次的问题，扩大检索覆盖范围：

step_back_prompt = ChatPromptTemplate.from_template(
    "你是一个善于抽象思考的助手。请将以下具体问题改写为一个更通用、更高层次的问题。\n\n"
    "具体问题：{question}\n"
    "更通用的问题："
)

def step_back_retrieve(question: str, retriever, llm=None):
    """Step-back检索：抽象化问题后再检索"""
    llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0)
    chain = step_back_prompt | llm
    abstract_question = chain.invoke({"question": question}).content

    # 同时用原始问题和抽象问题检索
    original_results = retriever.invoke(question)
    abstract_results = retriever.invoke(abstract_question)

    # 合并去重
    seen = set()
    merged = []
    for doc in original_results + abstract_results:
        if doc.page_content not in seen:
            seen.add(doc.page_content)
            merged.append(doc)
    return merged[:6]  # 限制总数

Multi-query 生成

将一个问题拆分为多个不同角度的查询：

multi_query_prompt = ChatPromptTemplate.from_template(
    "你是一个查询优化助手。请将以下问题从3个不同角度改写，每行一个：\n\n"
    "原始问题：{question}\n\n"
    "改写后的查询（每行一个）："
)

def multi_query_retrieve(question: str, retriever, llm=None):
    """Multi-query检索：多角度查询 + 合并结果"""
    llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
    chain = multi_query_prompt | llm
    response = chain.invoke({"question": question}).content

    queries = [q.strip() for q in response.strip().split("\n") if q.strip()]
    all_results = []
    for q in queries:
        all_results.extend(retriever.invoke(q))

    # 基于去重的倒数排名融合（RRF）
    return reciprocal_rank_fusion(all_results)

def reciprocal_rank_fusion(results: list, k: int = 60) -> list:
    """倒数排名融合算法"""
    scores = {}
    for rank, doc in enumerate(results):
        key = doc.page_content[:100]  # 用前100字符作为去重key
        scores[key] = scores.get(key, 0) + 1 / (k + rank + 1)
    # 按融合分数排序
    seen = {}
    for doc in results:
        key = doc.page_content[:100]
        if key not in seen:
            seen[key] = doc
    sorted_keys = sorted(scores, key=scores.get, reverse=True)
    return [seen[k] for k in sorted_keys if k in seen]

4. 混合检索架构

单一向量检索在精确匹配场景下表现不佳（如搜索产品编号、法律条款号）。混合检索融合稀疏（BM25）与稠密（向量）检索，再用交叉编码器精排。

from langchain.retrievers import EnsembleRetriever, BM25Retriever
from langchain_community.vectorstores import FAISS
from sentence_transformers import CrossEncoder

# 1. 构建两路检索器
bm25_retriever = BM25Retriever.from_texts(documents, k=10)
vector_retriever = FAISS.from_texts(documents, embeddings).as_retriever(search_kwargs={"k": 10})

# 2. 等权融合
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6],  # BM25权重稍低，向量检索权重稍高
)

# 3. 交叉编码器重排
reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2", max_length=512)

def hybrid_retrieve_with_rerank(query: str, top_k: int = 5) -> list[dict]:
    """混合检索 + 重排的完整管线"""
    # 第一阶段：多路召回
    candidates = ensemble_retriever.invoke(query)

    # 第二阶段：交叉编码器精排
    pairs = [(query, doc.page_content) for doc in candidates]
    scores = reranker.predict(pairs)

    # 按重排分数排序
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)

    return [
        {"text": doc.page_content, "score": float(score)}
        for doc, score in ranked[:top_k]
    ]

如果使用Cohere的重排API，集成更简单：

import cohere

co = cohere.Client("YOUR_API_KEY")

def cohere_rerank(query: str, documents: list[str], top_n: int = 5) -> list:
    response = co.rerank(
        model="rerank-v3.5",
        query=query,
        documents=documents,
        top_n=top_n,
    )
    return [
        {"text": documents[r.index], "score": r.relevance_score}
        for r in response.results
    ]

5. 上下文压缩与冗余过滤

检索到的文档片段常有大量冗余，直接塞进prompt既浪费token又干扰推理。上下文压缩提取与查询最相关的部分。

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 基于LLM的上下文压缩
compressor = LLMChainExtractor.from_llm(ChatOpenAI(model="gpt-4o-mini", temperature=0))
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10}),
)

# 每个检索到的文档只保留与查询相关的句子
compressed_docs = compression_retriever.invoke("退款流程需要多长时间？")

更高效的嵌入相似度过滤方案：

import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")

def filter_redundant_chunks(chunks: list[str], query: str,
                             similarity_threshold: float = 0.85,
                             relevance_threshold: float = 0.3) -> list[str]:
    """过滤冗余块，只保留与查询相关且彼此不重复的块"""
    chunk_embeddings = model.encode(chunks)
    query_embedding = model.encode([query])

    # 1. 过滤与查询不相关的块
    relevance_scores = cosine_similarity(query_embedding, chunk_embeddings)[0]
    relevant_mask = relevance_scores > relevance_threshold

    # 2. 在相关块中去重（保留第一个）
    filtered_chunks = []
    filtered_embeddings = []
    for i, (chunk, emb) in enumerate(zip(chunks, chunk_embeddings)):
        if not relevant_mask[i]:
            continue
        # 检查是否与已选块重复
        if filtered_embeddings:
            sims = cosine_similarity([emb], filtered_embeddings)[0]
            if max(sims) > similarity_threshold:
                continue  # 跳过冗余块
        filtered_chunks.append(chunk)
        filtered_embeddings.append(emb)

    return filtered_chunks

6. 自适应RAG与路由策略

不同类型的查询适合不同的检索策略。自适应RAG根据查询特征动态选择处理路径。

from enum import Enum

class QueryRoute(Enum):
    VECTOR_SEARCH = "vector"      # 语义相似查询
    KEYWORD_SEARCH = "keyword"    # 精确匹配查询
    WEB_SEARCH = "web"            # 需要实时信息
    DIRECT_ANSWER = "direct"      # 无需检索（常识性问题）

def route_query(question: str, llm=None) -> QueryRoute:
    """基于查询特征的智能路由"""
    llm = llm or ChatOpenAI(model="gpt-4o-mini", temperature=0)

    routing_prompt = f"""分析以下问题，判断最适合的处理方式：
1. vector - 需要在内部知识库中语义搜索
2. keyword - 包含精确的产品编号、条款号等关键词
3. web - 需要最新/实时信息（如天气、股价）
4. direct - 通用常识，无需检索

问题：{question}
只输出路由名称（vector/keyword/web/direct）："""

    route_str = llm.invoke(routing_prompt).content.strip().lower()
    return QueryRoute(route_str) if route_str in [r.value for r in QueryRoute] else QueryRoute.VECTOR_SEARCH

def adaptive_rag(question: str) -> str:
    """自适应RAG主流程"""
    route = route_query(question)

    if route == QueryRoute.DIRECT_ANSWER:
        return llm.invoke(question).content
    elif route == QueryRoute.KEYWORD_SEARCH:
        docs = bm25_retriever.invoke(question)
    elif route == QueryRoute.WEB_SEARCH:
        web_results = tavily_search(question)  # 假设有Tavily搜索工具
        docs = web_results
    else:
        docs = hybrid_retrieve_with_rerank(question)

    return generate_answer(question, docs)

7. 纠正性RAG（CRAG）与自反思RAG

纠正性RAG

CRAG在检索后增加一个"质量评估"步骤，判断检索结果是否足够好，决定是否回退到网络搜索：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

class CorrectiveRAG:
    def __init__(self, retriever, web_search_fn, llm):
        self.retriever = retriever
        self.web_search = web_search_fn
        self.llm = llm

    def evaluate_docs(self, question: str, docs: list) -> str:
        """评估检索文档的质量"""
        docs_text = "\n\n".join([d.page_content for d in docs[:3]])
        prompt = ChatPromptTemplate.from_template(
            "请评估以下文档是否能回答用户问题。只输出 correct / incorrect / ambiguous\n\n"
            "问题：{question}\n文档：{docs}\n评估结果："
        )
        chain = prompt | self.llm | StrOutputParser()
        return chain.invoke({"question": question, "docs": docs_text}).strip().lower()

    def refine_docs(self, question: str, docs: list) -> list:
        """精炼文档，提取有用部分"""
        docs_text = "\n\n".join([d.page_content for d in docs])
        prompt = ChatPromptTemplate.from_template(
            "从以下文档中提取与问题相关的关键信息，删除无关内容：\n\n"
            "问题：{question}\n文档：{docs}\n精炼后的内容："
        )
        chain = prompt | self.llm | StrOutputParser()
        refined = chain.invoke({"question": question, "docs": docs_text})
        return [type("Doc", (), {"page_content": refined})()]

    def run(self, question: str) -> str:
        docs = self.retriever.invoke(question)
        evaluation = self.evaluate_docs(question, docs)

        if evaluation == "correct":
            context = "\n\n".join([d.page_content for d in docs])
        elif evaluation == "incorrect":
            # 检索失败，回退到网络搜索
            web_docs = self.web_search(question)
            context = web_docs
        else:
            # 模棱两可，精炼现有文档 + 网络搜索补充
            refined = self.refine_docs(question, docs)
            web_docs = self.web_search(question)
            context = refined[0].page_content + "\n\n" + web_docs

        prompt = f"基于以下信息回答问题：\n\n{context}\n\n问题：{question}\n回答："
        return self.llm.invoke(prompt).content

自反思RAG（Self-RAG）

Self-RAG通过生成特殊的"反思token"来让模型自我评估：

class SelfRAG:
    """自反思RAG：模型在生成过程中自主决定是否需要检索、评估相关性"""

    def __init__(self, retriever, llm):
        self.retriever = retriever
        self.llm = llm

    def generate_with_reflection(self, question: str, max_iterations: int = 3) -> str:
        context_pool = []
        best_answer = ""

        for iteration in range(max_iterations):
            # 拼接已有上下文
            context = "\n\n".join(context_pool) if context_pool else "无相关文档"

            prompt = f"""基于以下上下文回答问题。如果信息不足，请输出 [RETRIEVE] 请求更多信息。

上下文：{context}

问题：{question}

回答（如果需要更多信息请以 [RETRIEVE] 开头）："""

            response = self.llm.invoke(prompt).content

            if response.startswith("[RETRIEVE]"):
                # 模型认为信息不足，追加检索
                new_docs = self.retriever.invoke(question)
                for doc in new_docs:
                    if doc.page_content not in context_pool:
                        context_pool.append(doc.page_content)
            else:
                # 评估回答质量
                eval_prompt = f"评估以下回答是否准确、完整、有依据（输出 score 1-10）：\n{response}"
                score = int(self.llm.invoke(eval_prompt).content.strip())
                if score >= 7:
                    return response
                best_answer = response  # 保留最佳答案

        return best_answer or response

8. GraphRAG与知识图谱增强检索

传统RAG只处理"相似片段"，但无法回答需要跨文档推理的问题（如"公司所有高管之间的关系"）。GraphRAG通过知识图谱补全这一短板。

import networkx as nx
from langchain_community.graphs import Neo4jGraph
from langchain.chains import GraphCypherQAChain
from langchain_openai import ChatOpenAI

# 1. 构建知识图谱
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password",
)

# 2. 从文档中提取实体和关系
extraction_prompt = """从以下文本中提取实体和关系，以JSON格式输出：
格式：[{"subject": "实体A", "relation": "关系", "object": "实体B"}]

文本：{text}"""

def extract_and_store(text: str, llm):
    """提取实体关系并存入图数据库"""
    response = llm.invoke(extraction_prompt.format(text=text))
    relations = json.loads(response.content)

    for rel in relations:
        graph.query(
            f"""
            MERGE (a:Entity {{name: $subject}})
            MERGE (b:Entity {{name: $object}})
            MERGE (a)-[r:RELATION {{type: $relation}}]->(b)
            """,
            params=rel,
        )

# 3. 图检索 + 向量检索的混合模式
def graph_enhanced_retrieve(question: str, vector_retriever, graph_qa_chain):
    """图谱增强检索"""
    # 向量检索获取相关文档
    vector_docs = vector_retriever.invoke(question)

    # 图谱检索获取结构化知识
    graph_result = graph_qa_chain.invoke({"query": question})

    # 合并两种来源
    combined_context = (
        "=== 文档检索结果 ===\n"
        + "\n\n".join([d.page_content for d in vector_docs])
        + "\n\n=== 知识图谱查询结果 ===\n"
        + graph_result["result"]
    )
    return combined_context

对于Microsoft的GraphRAG实现，使用社区摘要进行全局查询：

# 使用 graphrag 库（微软开源实现）
# pip install graphrag

# 索引构建（命令行）
# python -m graphrag.index --root ./rag_project

# 查询模式
from graphrag.query.cli import run_local_search, run_global_search

# 局部搜索：针对具体实体
local_result = run_local_search(
    root_dir="./rag_project",
    query="张三在公司担任什么职务？",
)

# 全局搜索：需要跨社区的综合理解
global_result = run_global_search(
    root_dir="./rag_project",
    query="公司各部门之间的协作关系是怎样的？",
)

9. 多模态RAG

当知识库包含图表、架构图、流程图等视觉内容时，纯文本RAG力不从心。多模态RAG同时处理图文。

from langchain_experimental.open_clip import OpenCLIPEmbeddings
from langchain.vectorstores import FAISS
from PIL import Image

# 1. 多模态嵌入
clip_embeddings = OpenCLIPEmbeddings(model_name="ViT-B-32", checkpoint="laion2b_s34b_b79k")

# 2. 索引图片（图片路径列表）
image_paths = ["diagrams/arch1.png", "charts/revenue.png", "flowcharts/process.png"]
image_metadatas = [{"source": p} for p in image_paths]
vectorstore = FAISS.from_texts(
    texts=image_paths,  # CLIP会将路径视为图片
    embedding=clip_embeddings,
    metadatas=image_metadatas,
)

# 3. 图文混合检索
def multimodal_retrieve(query: str, text_retriever, image_vectorstore, k: int = 3):
    """同时检索文本和图片"""
    text_docs = text_retriever.invoke(query)
    image_docs = image_vectorstore.similarity_search(query, k=k)

    # 将图片转为base64供多模态LLM使用
    import base64
    multimodal_context = []

    for doc in text_docs:
        multimodal_context.append({"type": "text", "text": doc.page_content})

    for img_doc in image_docs:
        img_path = img_doc.page_content
        with open(img_path, "rb") as f:
            img_b64 = base64.b64encode(f.read()).decode()
        multimodal_context.append({
            "type": "image_url",
            "image_url": {"url": f"data:image/png;base64,{img_b64}"},
        })

    return multimodal_context

使用GPT-4o进行多模态RAG生成：

from openai import OpenAI

client = OpenAI()

def multimodal_generate(query: str, context: list[dict]) -> str:
    """多模态RAG生成"""
    messages = [
        {"role": "system", "content": "你是一个技术助手，根据提供的文档和图片回答问题。"},
        {"role": "user", "content": [
            {"type": "text", "text": f"参考以下资料回答问题：\n\n"},
            *context,
            {"type": "text", "text": f"\n\n问题：{query}"},
        ]},
    ]

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=2000,
    )
    return response.choices[0].message.content

10. RAG系统评估框架

RAGAS评估

RAGAS提供四个核心指标：忠实度（Faithfulness）、答案相关性（Answer Relevancy）、上下文精度（Context Precision）、上下文召回率（Context Recall）。

from ragas import evaluate
from ragas.metrics import (
    faithfulness,
    answer_relevancy,
    context_precision,
    context_recall,
)
from datasets import Dataset

# 准备评估数据
eval_data = {
    "question": [
        "公司的退款政策是什么？",
        "如何重置密码？",
    ],
    "answer": [
        "公司提供30天无理由退款...",  # RAG系统生成的答案
        "您可以在设置页面点击重置密码...",
    ],
    "contexts": [
        ["退款政策：购买后30天内可申请全额退款...", "特殊情况需联系客服..."],
        ["密码重置：进入个人设置 > 安全 > 修改密码...", "忘记密码请点击登录页面的忘记密码链接..."],
    ],
    "ground_truth": [
        "购买后30天内支持无理由退款，特殊情况需联系客服处理",
        "在设置页面的安全选项中可以重置密码，也可以通过登录页的忘记密码链接重置",
    ],
}

dataset = Dataset.from_dict(eval_data)

# 运行评估
results = evaluate(
    dataset=dataset,
    metrics=[faithfulness, answer_relevancy, context_precision, context_recall],
)

print(results)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 'context_precision': 0.85, 'context_recall': 0.90}

DeepEval评估

DeepEval提供更细粒度的评估和CI/CD集成：

from deepeval import assert_test
from deepeval.metrics import (
    FaithfulnessMetric,
    AnswerRelevancyMetric,
    ContextualPrecisionMetric,
    ContextualRecallMetric,
)
from deepeval.test_case import LLMTestCase

# 定义测试用例
test_case = LLMTestCase(
    input="公司的退款政策是什么？",
    actual_output="公司提供30天无理由退款服务。",
    retrieval_context=["退款政策：购买后30天内可申请全额退款。"],
    expected_output="购买后30天内支持无理由退款。",
)

# 运行评估（可集成到pytest）
def test_rag_quality():
    faithfulness = FaithfulnessMetric(threshold=0.8)
    relevancy = AnswerRelevancyMetric(threshold=0.7)
    precision = ContextualPrecisionMetric(threshold=0.7)
    recall = ContextualRecallMetric(threshold=0.7)

    assert_test(test_case, [faithfulness, relevancy, precision, recall])

批量评估管线：

from deepeval import evaluate as deepeval_evaluate
from deepeval.metrics import GEval

# 自定义评估维度
coherence_metric = GEval(
    name="连贯性",
    criteria="评估回答的逻辑是否连贯、表述是否清晰",
    evaluation_params=["actual_output"],
    threshold=0.7,
)

# 批量评估
test_cases = [create_test_case(q, a, ctx) for q, a, ctx in zip(questions, answers, contexts)]
results = deepeval_evaluate(test_cases=test_cases, metrics=[faithfulness, coherence_metric])

11. 生产级RAG系统架构与调优

完整架构

一个生产级RAG系统包含以下层次：

用户查询
  │
  ▼
┌─────────────────────────────────────────┐
│  查询理解层（意图识别/改写/扩展）         │
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│  路由层（自适应策略选择）                 │
├─────────┬───────────┬───────────────────┤
│ 向量检索 │ BM25检索  │ 网络搜索/图谱检索  │
├─────────┴───────────┴───────────────────┤
│  重排层（交叉编码器/Cohere Rerank）       │
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│  后处理层（压缩/去重/过滤）               │
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│  生成层（带引用的LLM生成）                │
└────────────────┬────────────────────────┘
                 ▼
┌─────────────────────────────────────────┐
│  质量保障层（幻觉检测/事实核查）           │
└─────────────────────────────────────────┘

生产环境配置示例

from dataclasses import dataclass, field

@dataclass
class RAGConfig:
    """生产级RAG配置"""
    # 分块
    chunk_size: int = 512
    chunk_overlap: int = 64
    use_semantic_chunking: bool = True

    # 检索
    bm25_weight: float = 0.3
    vector_weight: float = 0.7
    initial_k: int = 20       # 初次召回数量
    rerank_top_k: int = 5     # 重排后保留数量

    # 查询改写
    enable_hyde: bool = True
    enable_multi_query: bool = False  # 成本较高，按需开启

    # 上下文压缩
    enable_compression: bool = True
    max_context_tokens: int = 3000

    # 质量保障
    enable_hallucination_check: bool = True
    min_relevance_score: float = 0.5

    # 缓存
    enable_cache: bool = True
    cache_ttl_seconds: int = 3600

调优要点

1. 分块大小：从512 token起步，根据评估结果微调。太大则检索精度下降，太小则上下文不完整
2. 召回数量：初召回建议10-20，重排后保留3-6。初召回太少会漏掉相关内容，太多则重排成本高
3. 混合检索权重：如果查询多为精确匹配（产品号、人名），提高BM25权重；如果多为模糊语义查询，提高向量权重
4. 重排模型选择：ms-marco-MiniLM-L-6-v2适合低延迟场景；bge-reranker-v2-m3精度更高但较慢
5. 嵌入模型：中文场景推荐bge-large-zh-v1.5或text-embedding-3-small（OpenAI），英文场景用text-embedding-3-large
6. 监控指标：关注检索延迟P99、端到端准确率、用户满意度反馈，定期回流评估

生产部署时，建议用Redis缓存高频查询结果，用异步管线处理批量索引更新，并设置降级策略——当检索质量过低时回退到纯LLM生成。