RAG 2026 生产级工程化完全指南：从朴素检索到 Agentic RAG 的架构演进与性能优化实战

作者: 程序员茄子 | 日期: 2026-05-23 | 阅读时间: 约 45 分钟 | 字数: 约 12000 字

目录

1. 引言：为什么你的 RAG 还停留在 2023 年？
2. RAG 技术演进的五个阶段
   • 2.1 阶段一：Naive RAG（2023 年主流）
   • 2.2 阶段二：Advanced RAG（2024 年）
   • 2.3 阶段三：Modular RAG（2024-2025 年）
   • 2.4 阶段四：Agentic RAG（2025-2026 年）
   • 2.5 阶段五：Multimodal RAG（2026+ 未来）
3. 生产级 RAG 架构设计
   • 3.1 文档加载与预处理
   • 3.2 分块策略：从固定大小到语义分块
   • 3.3 Embedding 模型选型与优化
   • 3.4 向量数据库选型对比
   • 3.5 检索策略：混合检索与重排序
4. 代码实战：构建生产级 RAG 系统
   • 4.1 环境准备与依赖安装
   • 4.2 完整实现：从文档到问答
   • 4.3 Advanced RAG：加入查询重写与重排序
   • 4.4 Agentic RAG：自主决策的多跳检索
5. 性能优化：从 60% 到 94% 准确率的 11 个策略
   • 5.1 策略 1：上下文感知分块（Context-Aware Chunking）
   • 5.2 策略 2：假设文档生成（HyDE）
   • 5.3 策略 3：查询扩展（Query Expansion）
   • 5.4 策略 4：混合检索（Hybrid Search）
   • 5.5 策略 5：重排序（Reranking）
   • 5.6 策略 6：递归检索（Recursive Retrieval）
   • 5.7 策略 7：CRAG（Corrective RAG）
   • 5.8 策略 8：Self-RAG（自我反思 RAG）
   • 5.9 策略 9：GraphRAG（知识图谱增强）
   • 5.10 策略 10：多模态 RAG
   • 5.11 策略 11：在线学习与反馈循环
6. 生产环境部署与监控
   • 6.1 Docker 容器化部署
   • 6.2 性能监控与日志
   • 6.3 成本控制与缓存策略
7. 真实案例：从 60% 到 94% 准确率的企业知识库 RAG
8. 总结与展望
9. 参考资料

引言：为什么你的 RAG 还停留在 2023 年？

如果你现在还在用最朴素的"切块 → 向量化 → TopK 检索 → 生成"四步 RAG，在面试或者和同行交流时大概率会显露出知识断层。

2023 年的 RAG：把文档塞进向量库然后检索，简单粗暴，但召回率低、幻觉严重。
2024 年的 RAG：引入查询重写、混合检索、重排序，准确率提升到 70-80%。
2025-2026 年的 RAG：Agentic RAG、GraphRAG、Multi-modal RAG 成为主流，准确率突破 90%，工程化落地进入成熟期。

根据 2026 年最新调研数据：

• 企业 RAG 项目成功率：从 2023 年的 32% 提升到 2026 年的 67%
• 平均准确率：从 60% 提升到 87%（头部企业达到 94%）
• 生产部署时间：从平均 6 个月缩短到 6 周（得益于 LangChain、LlamaIndex 等框架成熟）

本文目标：带你从零构建一套生产级 RAG 系统，覆盖从 Naive RAG 到 Agentic RAG 的完整演进路径，提供可运行的代码和 11 个性能优化策略，帮你将 RAG 准确率从 60% 提升到 90%+。

RAG 技术演进的五个阶段

阶段一：Naive RAG（2023 年主流）

最简单的实现，也是大多数教程还在教的方式：

# 朴素 RAG 方法（不推荐用于生产）
def naive_rag(query: str) -> str:
    # 1. 对查询进行向量化
    query_embedding = embed(query)
    
    # 2. 查找相似片段（Top-K）
    chunks = vector_db.search(query_embedding, top_k=5)
    
    # 3. 拼接上下文并生成答案
    context = "\n".join([chunk.text for chunk in chunks])
    answer = llm.generate(f"Context: {context}\n\nQuestion: {query}")
    
    return answer

问题：

1. 切片粒度难以兼顾：切大了丢失细节，切小了丢失上下文
2. 相关性检索不精准：向量相似 ≠ 语义相关
3. 没有多轮对话记忆：每次查询独立，无法处理指代
4. 无法处理复杂查询：多跳推理、比较类问题全部失败

适用场景：Demo、原型验证、简单 FAQ 问答。

阶段二：Advanced RAG（2024 年）

引入检索前/后优化，准确率提升 30-50%：

Pre-retrieval 优化（检索前）：

• 查询重写（Query Rewriting）：将口语化查询转为规范化查询
• 假设文档生成（HyDE）：让 LLM 先生成一个假设答案，再用假设答案检索

Retrieval 优化（检索中）：

• 混合检索（Hybrid Search）：BM25（关键词）+ 向量检索（语义），取长补短
• 语义分块（Semantic Chunking）：按语义边界分块，而不是固定字符数

Post-retrieval 优化（检索后）：

• 重排序（Reranking）：用交叉编码器（Cross-Encoder）对 Top-100 结果重新打分
• 上下文压缩（Contextual Compression）：提取与查询最相关的句子，减少噪声

# Advanced RAG 伪代码
def advanced_rag(query: str) -> str:
    # 1. 查询重写
    rewritten_query = rewrite_query(query)
    
    # 2. HyDE：生成假设文档
    hypothetical_doc = llm.generate(f"Answer this: {query}")
    
    # 3. 混合检索（BM25 + 向量）
    bm25_results = bm25_search(rewritten_query, top_k=50)
    vector_results = vector_search(embed(hypothetical_doc), top_k=50)
    merged = merge_and_deduplicate(bm25_results, vector_results)
    
    # 4. 重排序（Reranker）
    reranked = reranker.rerank(rewritten_query, merged, top_k=10)
    
    # 5. 上下文压缩
    compressed_context = compress_context(reranked, query)
    
    # 6. 生成答案
    answer = llm.generate(f"Context: {compressed_context}\n\nQuestion: {query}")
    
    return answer

阶段三：Modular RAG（2024-2025 年）

模块化设计，每个步骤可插拔替换：

[查询理解] → [检索策略路由] → [多源检索] → [结果聚合] → [生成]

核心创新：

1. 检索策略路由：根据查询类型选择不同检索后端（向量库、知识图谱、SQL 数据库、搜索引擎）
2. 多源检索：同时查询多个数据源，聚合结果
3. 迭代检索：根据初步答案，自主决定是否需要进一步检索

代码框架（LangChain Expression Language）：

from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

# 1. 定义检索器（可插拔）
retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

# 2. 定义生成链（可定制）
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是专业的问答助手，基于以下上下文回答问题：\n{context}"),
    ("human", "{input}"),
])
document_chain = create_stuff_documents_chain(llm, prompt)

# 3. 组合成检索链
retrieval_chain = create_retrieval_chain(retriever, document_chain)

# 4. 调用
result = retrieval_chain.invoke({"input": "RAG 的最优分块策略是什么？"})

阶段四：Agentic RAG（2025-2026 年）

RAG 的革命性跃迁：让 Agent 自主决策检索策略，而不是固定流水线。

核心能力：

1. 自主规划：Agent 根据问题复杂度，决定是否需要多跳检索
2. 工具调用：Agent 可以调用搜索引擎、SQL 查询、API 获取数据
3. 自我反思：Agent 评估检索结果质量，决定是否需要重新检索
4. 多轮对话：维护对话历史，处理指代和上下文

架构图：

用户问题
  ↓
Agent（Planner）
  ↓
[工具 1：向量检索]  [工具 2：知识图谱]  [工具 3：Web 搜索]
  ↓
结果聚合与评估
  ↓
需要更多数据？ → 是 → 继续检索
  ↓ 否
生成最终答案

代码实现（使用 LangGraph）：

from langgraph.graph import StateGraph, END

# 定义状态
class AgentState(TypedDict):
    query: str
    chat_history: list
    retrieved_docs: list
    answer: str
    needs_more_docs: bool

# 定义节点
def retrieve(state: AgentState):
    """检索节点：自主决定使用哪种检索策略"""
    query = state["query"]
    
    # 让 LLM 决定检索策略
    strategy = llm.invoke(f"问题：{query}\n请选择检索策略：A.向量检索 B.知识图谱 C.混合检索")
    
    if "A" in strategy:
        docs = vector_retrieve(query)
    elif "B" in strategy:
        docs = graph_retrieve(query)
    else:
        docs = hybrid_retrieve(query)
    
    return {"retrieved_docs": docs}

def generate(state: AgentState):
    """生成节点：评估检索质量，决定是否需要更多数据"""
    docs = state["retrieved_docs"]
    query = state["query"]
    
    # 评估检索质量
    quality_score = evaluate_retrieval_quality(query, docs)
    
    if quality_score < 0.7:
        return {"needs_more_docs": True}  # 触发重新检索
    else:
        answer = llm.invoke(f"基于以下文档回答问题：\n{docs}\n\n问题：{query}")
        return {"answer": answer, "needs_more_docs": False}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("generate", generate)
workflow.add_edge("retrieve", "generate")
workflow.add_conditional_edges(
    "generate",
    lambda s: "retrieve" if s["needs_more_docs"] else END
)

阶段五：Multimodal RAG（2026+ 未来）

下一代 RAG：支持文本、图片、表格、代码混合检索。

应用场景：

• 技术文档（文本）+ 架构图（图片）+ 代码示例（代码）混合检索
• 电商客服（商品描述 + 商品图片 + 用户评论）
• 医疗诊断（病历文本 + X 光片 + 检验报表）

技术挑战：

1. 多模态 Embedding：如何统一表示不同模态的语义？
2. 跨模态检索：如何用文本查询检索相关图片？
3. 融合生成：如何让 LLM 理解并融合多模态上下文？

当前最佳方案：CLIP（图像-文本联合 Embedding）+ 各模态独立检索 + 统一排序。

生产级 RAG 架构设计

文档加载与预处理

生产环境中，文档来源多样（PDF、Word、HTML、Markdown、代码仓库），需要统一的加载接口。

推荐使用 LangChain 的 Document Loaders：

from langchain_community.document_loaders import (
    PyPDFLoader,       # PDF
    Docx2txtLoader,   # Word
    UnstructuredHTMLLoader,  # HTML
    MarkdownLoader,   # Markdown
    GitLoader,        # 代码仓库
)

# 1. 加载 PDF
pdf_loader = PyPDFLoader("technical_manual.pdf")
pdf_docs = pdf_loader.load()

# 2. 加载 Word
word_loader = Docx2txtLoader("requirements.docx")
word_docs = word_loader.load()

# 3. 加载代码仓库（Git）
git_loader = GitLoader(
    clone_url="https://github.com/your-repo.git",
    branch="main",
    file_filter=lambda file_path: file_path.endswith(".py")  # 只加载 Python 文件
)
code_docs = git_loader.load()

# 4. 统一预处理：去除噪声、标准化格式
import re

def clean_text(text: str) -> str:
    """清理文本：去除多余空白、特殊字符"""
    text = re.sub(r'\s+', ' ', text)  # 多个空白符合并为单个空格
    text = re.sub(r'[^\w\s\u4e00-\u9fff，。！？；：""''（）]', '', text)  # 保留中英文、标点
    return text.strip()

cleaned_docs = [Document(page_content=clean_text(doc.page_content), metadata=doc.metadata) for doc in pdf_docs + word_docs + code_docs]

分块策略：从固定大小到语义分块

固定大小分块（不推荐）：

from langchain_text_splitters import RecursiveCharacterTextSplitter

# 固定大小分块（500 字符，重叠 50 字符）
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", "。", " ", ""]  # 按段落、句子、词语分割
)

chunks = text_splitter.split_documents(cleaned_docs)

问题：可能把一个完整的句子或段落切成两半，丢失语义。

语义分块（推荐）：

使用 LLM 或 Embedding 模型识别语义边界：

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 语义分块：根据 Embedding 相似度动态分块
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
semantic_splitter = SemanticChunker(
    embeddings,
    breakpoint_threshold_type="percentile",  # 使用百分位数阈值
    breakpoint_threshold_amount=95  # 语义变化超过 95% 分位数时切分
)

semantic_chunks = semantic_splitter.split_documents(cleaned_docs)

进阶：上下文感知分块（Context-Aware Chunking）：

在分块时，让 LLM 为每个块生成上下文摘要，检索时同时匹配摘要和内容：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# 1. 为每个块生成上下文摘要
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个文档摘要助手。请为以下文本生成一句简短的上下文摘要（20 字以内），描述这段内容的主旨。"),
    ("human", "{text}")
])
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

def add_contextual_metadata(doc: Document) -> Document:
    """为文档块添加上下文摘要"""
    summary = llm.invoke(prompt.format_messages(text=doc.page_content)).content
    doc.metadata["context_summary"] = summary
    return doc

chunks_with_context = [add_contextual_metadata(chunk) for chunk in semantic_chunks]

# 2. 检索时，同时匹配摘要和内容
def retrieve_with_context(query: str, chunks: list, top_k: int = 5):
    """检索时同时考虑块内容和上下文摘要"""
    query_embedding = embed(query)
    
    results = []
    for chunk in chunks:
        # 计算查询与块内容的相似度
        content_sim = cosine_similarity(query_embedding, embed(chunk.page_content))
        # 计算查询与上下文摘要的相似度
        summary_sim = cosine_similarity(query_embedding, embed(chunk.metadata["context_summary"]))
        # 加权平均
        final_score = 0.7 * content_sim + 0.3 * summary_sim
        results.append((chunk, final_score))
    
    # 排序并返回 Top-K
    results.sort(key=lambda x: x[1], reverse=True)
    return [chunk for chunk, score in results[:top_k]]

Embedding 模型选型与优化

2026 年主流 Embedding 模型对比：

选型建议：

• 英文场景：OpenAI text-embedding-3-small（性价比最高）
• 中文场景：BGE-M3 或 ChatGLM4-Embedding（本地部署，无隐私风险）
• 代码检索：Jina Embeddings v3（专门针对代码优化）
• 多语言：Cohere Embed v4 或 Jina v3

代码示例：使用 BGE-M3（开源中文 Embedding）：

from sentence_transformers import SentenceTransformer

# 1. 加载模型（本地）
model = SentenceTransformer('BAAI/bge-m3', device='cuda')  # 使用 GPU 加速

# 2. 编码文档
documents = [chunk.page_content for chunk in chunks_with_context]
doc_embeddings = model.encode(documents, batch_size=32, normalize_embeddings=True)

# 3. 编码查询
query = "RAG 的最优分块策略是什么？"
query_embedding = model.encode([query], normalize_embeddings=True)[0]

# 4. 计算相似度
from sklearn.metrics.pairwise import cosine_similarity
similarities = cosine_similarity([query_embedding], doc_embeddings)[0]

# 5. 获取 Top-K
top_k_indices = similarities.argsort()[-5:][::-1]
retrieved_chunks = [chunks_with_context[i] for i in top_k_indices]

向量数据库选型对比

2026 年主流向量数据库对比：

选型建议：

• 原型开发：Chroma（最简单）
• 已有 PostgreSQL：PGVector（零成本迁移）
• 生产环境（大规模）：Qdrant 或 Milvus
• 不想运维：Pinecone（全托管）
• GraphRAG：Weaviate（原生支持知识图谱）

代码示例：使用 PGVector（PostgreSQL 扩展）：

from langchain_community.vectorstores import PGVector
from langchain_openai import OpenAIEmbeddings

# 1. 连接 PostgreSQL（需先安装 pgvector 扩展）
vectordb = PGVector(
    connection_string="postgresql://user:password@localhost:5432/rag_db",
    embedding_function=OpenAIEmbeddings(),
    collection_name="my_documents"
)

# 2. 插入文档（自动向量化）
vectordb.add_documents(chunks_with_context)

# 3. 相似度检索
query = "RAG 的最优分块策略是什么？"
retrieved_docs = vectordb.similarity_search(query, k=5)

# 4. 混合检索（向量 + BM25）
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

bm25_retriever = BM25Retriever.from_documents(chunks_with_context)
vector_retriever = vectordb.as_retriever(search_kwargs={"k": 10})

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.3, 0.7]  # 向量检索权重更高
)

retrieved_docs = ensemble_retriever.get_relevant_documents(query)

检索策略：混合检索与重排序

混合检索（Hybrid Search）：结合稀疏检索（BM25）和稠密检索（向量），取长补短。

代码实现：

from rank_bm25 import BM25Okapi
import numpy as np

class HybridRetriever:
    def __init__(self, chunks: list, embeddings, weight_bm25=0.3, weight_vector=0.7):
        self.chunks = chunks
        self.embeddings = embeddings
        self.weight_bm25 = weight_bm25
        self.weight_vector = weight_vector
        
        # 初始化 BM25
        tokenized_corpus = [chunk.page_content.split() for chunk in chunks]
        self.bm25 = BM25Okapi(tokenized_corpus)
        
        # 预计算所有文档的向量
        self.doc_embeddings = embeddings.embed_documents([chunk.page_content for chunk in chunks])
    
    def retrieve(self, query: str, top_k: int = 10):
        # 1. BM25 检索
        tokenized_query = query.split()
        bm25_scores = self.bm25.get_scores(tokenized_query)
        bm25_top_k = np.argsort(bm25_scores)[-top_k:][::-1]
        
        # 2. 向量检索
        query_embedding = self.embeddings.embed_query(query)
        vector_scores = cosine_similarity([query_embedding], self.doc_embeddings)[0]
        vector_top_k = np.argsort(vector_scores)[-top_k:][::-1]
        
        # 3. 融合（RRF: Reciprocal Rank Fusion）
        rrf_scores = {}
        for rank, doc_idx in enumerate(bm25_top_k):
            rrf_scores[doc_idx] = rrf_scores.get(doc_idx, 0) + 1 / (rank + 1 + 60)  # 60 是常量
        for rank, doc_idx in enumerate(vector_top_k):
            rrf_scores[doc_idx] = rrf_scores.get(doc_idx, 0) + 1 / (rank + 1 + 60)
        
        # 4. 排序并返回 Top-K
        sorted_docs = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)
        return [(self.chunks[idx], score) for idx, score in sorted_docs[:top_k]]

重排序（Reranking）：用更精确但更慢的模型对 Top-100 结果重新打分。

推荐使用 Cohere Reranker 或 BGE Reranker：

from langchain_community.document_compressors import CohereRerank
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
from langchain_community.document_transformers import LongContextReorder

# 1. 使用 Cohere Reranker（需要 API Key）
compressor = CohereRerank(
    cohere_api_key="your-cohere-api-key",
    model="rerank-english-v3.0",  # 或 rerank-multilingual-v3.0（多语言）
    top_n=10  # 最终返回 10 个结果
)

# 2. 构建压缩管道（重排序 + 长上下文重排序）
pipeline = DocumentCompressorPipeline(
    transformers=[compressor, LongContextReorder()]  # LongContextReorder 把最相关的放最后（避免 Lost in the Middle）
)

# 3. 使用压缩器包装检索器
from langchain.retrievers import ContextualCompressionRetriever

base_retriever = vectordb.as_retriever(search_kwargs={"k": 50})  # 先检索 50 个
compression_retriever = ContextualCompressionRetriever(
    base_compressor=pipeline,
    base_retriever=base_retriever
)

# 4. 检索（自动重排序）
query = "RAG 的最优分块策略是什么？"
retrieved_docs = compression_retriever.get_relevant_documents(query)

代码实战：构建生产级 RAG 系统

环境准备与依赖安装

# 创建虚拟环境
python3 -m venv rag-env
source rag-env/bin/activate  # Linux/Mac
# 或 rag-env\Scripts\activate  # Windows

# 安装依赖
pip install \
    langchain==0.3.0 \
    langchain-community==0.3.0 \
    langchain-openai==0.2.0 \
    langchain-postgres==0.1.0 \
    sentence-transformers==3.0.0 \
    rank-bm25==0.2.2 \
    scikit-learn==1.5.0 \
    cohere==5.0.0 \
    transformers==4.44.0 \
    torch==2.4.0 \
    pgvector==0.3.0 \
    qdrant-client==1.12.0 \
    langgraph==0.2.0 \
    langchain-experimental==0.3.0

完整实现：从文档到问答

完整代码示例（Production-Ready）：

import os
from typing import List, Tuple
from langchain_community.document_loaders import PyPDFLoader, Docx2txtLoader, MarkdownLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import PGVector
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever
from langchain_community.document_compressors import CohereRerank
from langchain.retrievers.document_compressors import DocumentCompressorPipeline
from langchain.retrievers import ContextualCompressionRetriever

# ==================== 配置 ====================
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
COHERE_API_KEY = os.getenv("COHERE_API_KEY")
POSTGRES_CONNECTION = "postgresql://user:password@localhost:5432/rag_db"

# ==================== 1. 文档加载 ====================
def load_documents(doc_dir: str) -> List:
    """加载目录下所有文档"""
    documents = []
    
    for file in os.listdir(doc_dir):
        file_path = os.path.join(doc_dir, file)
        if file.endswith(".pdf"):
            loader = PyPDFLoader(file_path)
        elif file.endswith(".docx"):
            loader = Docx2txtLoader(file_path)
        elif file.endswith(".md"):
            loader = MarkdownLoader(file_path)
        else:
            continue
        
        documents.extend(loader.load())
    
    return documents

# ==================== 2. 文档分块 ====================
def split_documents(documents: List, chunk_size: int = 800, chunk_overlap: int = 100) -> List:
    """分块（固定大小 + 递归分割）"""
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", "。", "；", " ", ""]
    )
    return text_splitter.split_documents(documents)

# ==================== 3. 向量化并存储 ====================
def store_in_vector_db(chunks: List, collection_name: str = "my_documents"):
    """存储到 PGVector"""
    embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
    
    vectordb = PGVector.from_documents(
        documents=chunks,
        embedding=embeddings,
        connection_string=POSTGRES_CONNECTION,
        collection_name=collection_name,
        use_jsonb=True  # 使用 JSONB 存储元数据（更快的过滤）
    )
    
    return vectordb

# ==================== 4. 构建检索器（混合检索 + 重排序） ====================
def build_retriever(vectordb, chunks: List):
    """构建检索器：混合检索 + 重排序"""
    # 4.1 向量检索
    vector_retriever = vectordb.as_retriever(search_kwargs={"k": 50})
    
    # 4.2 BM25 检索
    bm25_retriever = BM25Retriever.from_documents(chunks)
    bm25_retriever.k = 50
    
    # 4.3 混合检索（Ensemble）
    ensemble_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=[0.7, 0.3]
    )
    
    # 4.4 重排序（Cohere Reranker）
    compressor = CohereRerank(
        cohere_api_key=COHERE_API_KEY,
        model="rerank-multilingual-v3.0",
        top_n=10
    )
    pipeline = DocumentCompressorPipeline(transformers=[compressor])
    
    compression_retriever = ContextualCompressionRetriever(
        base_compressor=pipeline,
        base_retriever=ensemble_retriever
    )
    
    return compression_retriever

# ==================== 5. 构建 RAG 链 ====================
def build_rag_chain(retriever):
    """构建 RAG 链"""
    # 5.1 定义 Prompt
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是专业的技术文档助手。基于以下上下文回答问题。
        
规则：
1. 只在上下文包含答案时回答，否则说"我无法从提供的文档中找到答案"。
2. 回答时引用上下文的具体部分（用 [来源: 文件名, 第X页] 标注）。
3. 如果问题不明确，先澄清用户意图。
4. 回答要详细、准确，使用技术术语。

上下文：
{context}"""),
        ("human", "{input}")
    ])
    
    # 5.2 定义 LLM
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    
    # 5.3 构建链
    document_chain = create_stuff_documents_chain(llm, prompt)
    retrieval_chain = create_retrieval_chain(retriever, document_chain)
    
    return retrieval_chain

# ==================== 6. 主函数 ====================
def main():
    # 1. 加载文档
    print("正在加载文档...")
    documents = load_documents("./docs")
    print(f"已加载 {len(documents)} 个文档")
    
    # 2. 分块
    print("正在分块...")
    chunks = split_documents(documents)
    print(f"已生成 {len(chunks)} 个块")
    
    # 3. 存储到向量数据库
    print("正在向量化并存储...")
    vectordb = store_in_vector_db(chunks)
    print("存储完成")
    
    # 4. 构建检索器
    print("正在构建检索器...")
    retriever = build_retriever(vectordb, chunks)
    print("检索器构建完成")
    
    # 5. 构建 RAG 链
    print("正在构建 RAG 链...")
    rag_chain = build_rag_chain(retriever)
    print("RAG 系统就绪！")
    
    # 6. 交互式问答
    while True:
        query = input("\n请输入问题（输入 'exit' 退出）: ")
        if query.lower() == "exit":
            break
        
        # 调用 RAG 链
        result = rag_chain.invoke({"input": query})
        
        print("\n答案：")
        print(result["answer"])
        print("\n来源文档：")
        for i, doc in enumerate(result["context"]):
            print(f"{i+1}. {doc.metadata.get('source', '未知来源')}")

if __name__ == "__main__":
    main()

Advanced RAG：加入查询重写与 HyDE

改进点：

1. 查询重写：将口语化查询转为规范化查询
2. HyDE：生成假设答案，用假设答案检索（提高召回率）

代码实现：

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# ==================== 查询重写 ====================
def rewrite_query(original_query: str) -> str:
    """查询重写：将口语化查询转为规范化查询"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是查询优化助手。请将用户的问题改写为更适合文档检索的规范化查询。保持原意，但使用更正式的术语。"),
        ("human", "原问题：{query}\n\n规范化查询：")
    ])
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    return llm.invoke(prompt.format_messages(query=original_query)).content

# ==================== HyDE（假设文档生成） ====================
def generate_hypothetical_document(query: str) -> str:
    """HyDE：让 LLM 生成一个假设答案，用于检索"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是知识库问答助手。请根据用户的问题，生成一段可能出现在文档中的答案（200 字以内）。如果不确定，生成一个合理的假设。"),
        ("human", "问题：{query}\n\n假设答案：")
    ])
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
    return llm.invoke(prompt.format_messages(query=query)).content

# ==================== 改进版检索（加入 HyDE） ====================
def retrieve_with_hyde(query: str, vectordb, top_k: int = 10):
    """使用 HyDE 检索：用假设答案的向量检索"""
    # 1. 生成假设答案
    hypothetical_doc = generate_hypothetical_document(query)
    print(f"假设答案：{hypothetical_doc[:100]}...")
    
    # 2. 用假设答案的向量检索
    query_embedding = vectordb.embedding_function.embed_query(hypothetical_doc)
    results = vectordb.similarity_search_by_vector(query_embedding, k=top_k)
    
    return results

# ==================== 集成到 RAG 链 ====================
def build_advanced_rag_chain(retriever):
    """构建 Advanced RAG 链（加入查询重写和 HyDE）"""
    # 1. 查询重写 + HyDE
    def preprocess_query(inputs: dict) -> dict:
        original_query = inputs["input"]
        # 重写查询
        rewritten_query = rewrite_query(original_query)
        print(f"原始查询：{original_query}")
        print(f"重写查询：{rewritten_query}")
        return {"input": rewritten_query, "original_input": original_query}
    
    # 2. 定义 Prompt（使用重写后的查询）
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是专业的技术文档助手。基于以下上下文回答问题。\n\n上下文：\n{context}"),
        ("human", "{input}")
    ])
    
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    document_chain = create_stuff_documents_chain(llm, prompt)
    
    # 3. 构建链（加入预处理）
    from langchain.chains import TransformChain
    transform_chain = TransformChain(
        input_variables=["input"],
        output_variables=["input", "original_input"],
        transform=preprocess_query
    )
    
    # 4. 组合链
    from langchain.chains import SequentialChain
    retrieval_chain = create_retrieval_chain(retriever, document_chain)
    full_chain = SequentialChain(
        chains=[transform_chain, retrieval_chain],
        input_variables=["input"],
        output_variables=["answer", "context"]
    )
    
    return full_chain

Agentic RAG：自主决策的多跳检索

使用 LangGraph 实现 Agentic RAG：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator

# ==================== 定义状态 ====================
class AgentState(TypedDict):
    query: str
    chat_history: list
    retrieved_docs: list
    answer: str
    needs_more_docs: bool
    retry_count: int

# ==================== 定义节点 ====================
def should_retry(state: AgentState) -> str:
    """决定是否继续检索"""
    if state["needs_more_docs"] and state["retry_count"] < 3:
        return "retrieve"
    else:
        return END

def retrieve(state: AgentState):
    """检索节点：自主决定使用哪种检索策略"""
    query = state["query"]
    chat_history = state["chat_history"]
    
    # 让 LLM 决定检索策略
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是检索策略助手。根据用户的问题，选择最合适的检索策略：
A. 向量检索（适合语义相似问题）
B. 关键词检索（适合精确匹配术语）
C. 混合检索（适合大多数问题）
D. 知识图谱检索（适合需要推理的问题）

只返回字母，不要解释。"""),
        ("human", "问题：{query}\n\n检索策略：")
    ])
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    strategy = llm.invoke(prompt.format_messages(query=query)).content.strip()
    
    # 执行检索
    if "A" in strategy:
        docs = vector_retrieve(query)
    elif "B" in strategy:
        docs = keyword_retrieve(query)
    elif "D" in strategy:
        docs = graph_retrieve(query)
    else:  # 默认混合检索
        docs = hybrid_retrieve(query)
    
    # 更新状态
    return {
        "retrieved_docs": state["retrieved_docs"] + docs,
        "retry_count": state["retry_count"] + 1
    }

def generate(state: AgentState):
    """生成节点：评估检索质量，决定是否需要更多数据"""
    docs = state["retrieved_docs"]
    query = state["query"]
    
    # 评估检索质量（让 LLM 评估）
    eval_prompt = ChatPromptTemplate.from_messages([
        ("system", """你是检索质量评估助手。根据以下检索到的文档，评估它们是否能够回答用户的问题。
        
评分标准（0-10分）：
- 10分：文档完全包含答案
- 7-9分：文档包含部分答案，需要推理
- 4-6分：文档相关但不直接包含答案
- 0-3分：文档不相关

只返回数字分数，不要解释。"""),
        ("human", "问题：{query}\n\n检索到的文档：\n{docs}\n\n评分：")
    ])
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    quality_score = int(llm.invoke(eval_prompt.format_messages(
        query=query,
        docs="\n\n".join([doc.page_content[:200] for doc in docs])
    )).content.strip())
    
    print(f"检索质量评分：{quality_score}/10")
    
    # 决定是否需要更多文档
    if quality_score < 7 and state["retry_count"] < 3:
        return {"needs_more_docs": True}
    else:
        # 生成答案
        answer_prompt = ChatPromptTemplate.from_messages([
            ("system", "你是专业的技术文档助手。基于以下上下文回答问题。\n\n上下文：\n{context}"),
            ("human", "{input}")
        ])
        answer_llm = ChatOpenAI(model="gpt-4o", temperature=0)
        answer = answer_llm.invoke(answer_prompt.format_messages(
            context="\n\n".join([doc.page_content for doc in docs]),
            input=query
        )).content
        
        return {"answer": answer, "needs_more_docs": False}

# ==================== 构建 Agentic RAG 图 ====================
def build_agentic_rag():
    """构建 Agentic RAG（使用 LangGraph）"""
    workflow = StateGraph(AgentState)
    
    # 添加节点
    workflow.add_node("retrieve", retrieve)
    workflow.add_node("generate", generate)
    
    # 添加边
    workflow.set_entry_point("retrieve")
    workflow.add_edge("retrieve", "generate")
    workflow.add_conditional_edges(
        "generate",
        should_retry,
        {
            "retrieve": "retrieve",
            END: END
        }
    )
    
    # 编译图
    app = workflow.compile()
    return app

# ==================== 使用 Agentic RAG ====================
def main_agentic():
    """使用 Agentic RAG"""
    # 构建 Agent
    agentic_rag = build_agentic_rag()
    
    # 交互式问答
    while True:
        query = input("\n请输入问题（输入 'exit' 退出）: ")
        if query.lower() == "exit":
            break
        
        # 初始化状态
        initial_state = {
            "query": query,
            "chat_history": [],
            "retrieved_docs": [],
            "answer": "",
            "needs_more_docs": False,
            "retry_count": 0
        }
        
        # 运行 Agent
        result = agentic_rag.invoke(initial_state)
        
        print("\n答案：")
        print(result["answer"])
        print(f"\n检索轮数：{result['retry_count']}")

if __name__ == "__main__":
    main_agentic()

性能优化：从 60% 到 94% 准确率的 11 个策略

根据 2026 年最新研究和企业实践，以下是提升 RAG 准确率最有效的 11 个策略（按效果排序）：

策略 1：上下文感知分块（Context-Aware Chunking）

效果：准确率提升 8-12%

原理：在分块时，让 LLM 为每个块生成上下文摘要，检索时同时匹配摘要和内容。

代码：见"分块策略"章节。

策略 2：假设文档生成（HyDE）

效果：准确率提升 10-15%

原理：让 LLM 先生成一个假设答案，再用假设答案的向量检索（假设答案的向量比原查询更接近目标文档）。

代码：见"Advanced RAG"章节。

策略 3：查询扩展（Query Expansion）

效果：准确率提升 5-10%

原理：让 LLM 生成多个相似查询，分别检索，然后聚合结果。

def expand_query(query: str, num_expansions: int = 3) -> List[str]:
    """查询扩展：生成多个相似查询"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""你是查询扩展助手。请根据用户的问题，生成 {num_expansions} 个相似的查询（用不同的术语表达同样的意思）。
        
只返回查询列表，每行一个，不要编号。"""),
        ("human", "原问题：{query}\n\n扩展查询：")
    ])
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)
    result = llm.invoke(prompt.format_messages(query=query)).content
    
    expansions = [line.strip() for line in result.split("\n") if line.strip()]
    return [query] + expansions  # 包含原查询

策略 4：混合检索（Hybrid Search）

效果：准确率提升 12-18%

原理：结合 BM25（关键词）和向量检索（语义），使用 RRF（Reciprocal Rank Fusion）融合。

代码：见"检索策略"章节。

策略 5：重排序（Reranking）

效果：准确率提升 15-20%

原理：用交叉编码器（Cross-Encoder）对 Top-100 结果重新打分，精确排序。

推荐使用 Cohere Reranker 或 BGE Reranker。

from sentence_transformers import CrossEncoder

# 使用 BGE Reranker（开源，中文效果好）
reranker = CrossEncoder('BAAI/bge-reranker-v1.5', device='cuda')

def rerank(query: str, docs: List, top_k: int = 10) -> List:
    """重排序"""
    # 构造输入对
    pairs = [[query, doc.page_content] for doc in docs]
    
    # 计算相关性分数
    scores = reranker.predict(pairs)
    
    # 排序
    scored_docs = list(zip(docs, scores))
    scored_docs.sort(key=lambda x: x[1], reverse=True)
    
    return [doc for doc, score in scored_docs[:top_k]]

策略 6：递归检索（Recursive Retrieval）

效果：准确率提升 8-10%（对长文档效果显著）

原理：先检索文档摘要，再检索相关章节（两阶段检索）。

def recursive_retrieve(query: str, doc_summaries: List, full_docs: List, top_k: int = 5):
    """递归检索：先检索摘要，再检索全文"""
    # 1. 检索摘要（第一阶段）
    summary_embeddings = embeddings.embed_documents([summary for summary, _ in doc_summaries])
    query_embedding = embeddings.embed_query(query)
    summary_scores = cosine_similarity([query_embedding], summary_embeddings)[0]
    top_summary_indices = np.argsort(summary_scores)[-top_k:][::-1]
    
    # 2. 检索相关全文（第二阶段）
    relevant_docs = [full_docs[i] for i in top_summary_indices]
    full_text_embeddings = embeddings.embed_documents([doc.page_content for doc in relevant_docs])
    full_text_scores = cosine_similarity([query_embedding], full_text_embeddings)[0]
    top_full_text_indices = np.argsort(full_text_scores)[-top_k:][::-1]
    
    return [relevant_docs[i] for i in top_full_text_indices]

策略 7：CRAG（Corrective RAG）

效果：准确率提升 10-15%

原理：检索后评估文档质量，如果质量低，则触发网页搜索补充知识。

def crag_retrieve(query: str, retrieved_docs: List) -> List:
    """CRAG：纠正性检索"""
    # 1. 评估检索质量
    quality_scores = []
    for doc in retrieved_docs:
        score = evaluate_doc_relevance(query, doc.page_content)
        quality_scores.append(score)
    
    # 2. 如果平均质量低，触发网页搜索
    if np.mean(quality_scores) < 0.6:
        print("检索质量低，触发网页搜索...")
        web_results = web_search(query)  # 调用搜索引擎 API
        retrieved_docs.extend(web_results)
    
    return retrieved_docs

策略 8：Self-RAG（自我反思 RAG）

效果：准确率提升 12-18%

原理：让 LLM 在生成过程中自我反思，决定是否需要检索更多信息。

实现：需要在生成时插入特殊的反思 Token（如 [Retrieve]、[NoRetrieve]、[Continue]），较复杂。建议使用 LangChain 的 SelfQueryRetriever。

策略 9：GraphRAG（知识图谱增强）

效果：准确率提升 15-25%（对多跳推理问题效果显著）

原理：从文档中抽取实体和关系，构建知识图谱，检索时结合图遍历。

代码实现（使用 Neo4j + LangChain）：

from langchain_community.graphs import Neo4jGraph
from langchain_community.chains.graph_qa.base import GraphQAChain

# 1. 构建知识图谱（从文档中抽取实体和关系）
graph = Neo4jGraph(
    url="bolt://localhost:7687",
    username="neo4j",
    password="password"
)

# 2. 从文档中抽取实体和关系（使用 LLM）
def extract_entities_and_relations(doc: Document) -> Tuple[List, List]:
    """抽取实体和关系"""
    prompt = ChatPromptTemplate.from_messages([
        ("system", """你是信息抽取助手。请从以下文档中抽取实体和关系，输出 JSON 格式：
        
{
    "entities": [{"name": "实体名", "type": "实体类型"}, ...],
    "relations": [{"head": "头实体", "relation": "关系", "tail": "尾实体"}, ...]
}

实体类型：人物、组织、技术、概念、事件、地点
关系类型：属于、使用、导致、依赖、实现、替代"""),
        ("human", "文档：{text}\n\n抽取结果：")
    ])
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    result = llm.invoke(prompt.format_messages(text=doc.page_content)).content
    
    import json
    data = json.loads(result)
    return data["entities"], data["relations"]

# 3. 将实体和关系存入 Neo4j
for chunk in chunks:
    entities, relations = extract_entities_and_relations(chunk)
    
    # 存储实体（节点）
    for entity in entities:
        graph.query(
            "MERGE (e:Entity {name: $name}) SET e.type = $type",
            {"name": entity["name"], "type": entity["type"]}
        )
    
    # 存储关系（边）
    for rel in relations:
        graph.query(
            """MATCH (head:Entity {name: $head_name})
MATCH (tail:Entity {name: $tail_name})
MERGE (head)-[:RELATION {type: $relation}]->(tail)""",
            {"head_name": rel["head"], "tail_name": rel["tail"], "relation": rel["relation"]}
        )

# 4. GraphRAG 检索
def graph_rag_retrieve(query: str) -> str:
    """GraphRAG 检索：结合知识图谱和向量检索"""
    # 4.1 从查询中识别实体
    entities = identify_entities_in_query(query)
    
    # 4.2 在图中遍历相关节点
    for entity in entities:
        related_nodes = graph.query(
            """MATCH (e:Entity {name: $name})-[r:RELATION*1..3]-(related)
RETURN related.name, related.type, r.type""",
            {"name": entity}
        )
    
    # 4.3 将图结构转化为自然语言上下文
    graph_context = convert_graph_to_text(related_nodes)
    
    # 4.4 向量检索（补充）
    vector_results = vector_retrieve(query)
    
    # 4.5 合并上下文并生成答案
    context = graph_context + "\n\n" + "\n\n".join([doc.page_content for doc in vector_results])
    answer = llm.invoke(f"基于以下上下文回答问题：\n{context}\n\n问题：{query}")
    
    return answer

策略 10：多模态 RAG

效果：准确率提升 20-30%（对包含图表、代码的文档效果显著）

原理：支持文本、图片、表格、代码混合检索。

实现：使用 CLIP（图像-文本联合 Embedding）+ 各模态独立检索 + 统一排序。

from transformers import CLIPProcessor, CLIPModel
import torch

# 1. 加载 CLIP 模型（支持图像和文本联合 Embedding）
clip_model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def multimodal_embed(text: str = None, image = None):
    """多模态嵌入：支持文本和图像"""
    inputs = clip_processor(text=text, images=image, return_tensors="pt", padding=True)
    outputs = clip_model(**inputs)
    
    if text and image:
        return outputs.text_embeds, outputs.image_embeds
    elif text:
        return outputs.text_embeds
    else:
        return outputs.image_embeds

# 2. 为文档中的图片生成嵌入
def process_document_with_images(doc: Document):
    """处理包含图片的文档"""
    # 假设文档中有图片路径
    images = extract_images_from_doc(doc)  # 自定义函数：从 PDF/Word 中提取图片
    
    image_embeddings = []
    for img in images:
        img_embedding = multimodal_embed(image=img)
        image_embeddings.append(img_embedding)
    
    # 文本嵌入
    text_embedding = multimodal_embed(text=doc.page_content)
    
    return {
        "text_embedding": text_embedding,
        "image_embeddings": image_embeddings,
        "text": doc.page_content,
        "images": images
    }

# 3. 多模态检索
def multimodal_retrieve(query: str, docs_with_images: List, top_k: int = 5):
    """多模态检索：支持文本查询检索相关文本和图片"""
    query_embedding = multimodal_embed(text=query)
    
    results = []
    for doc in docs_with_images:
        # 计算文本相似度
        text_sim = cosine_similarity(query_embedding, doc["text_embedding"])
        
        # 计算图片相似度（如果有的话）
        max_image_sim = 0
        for img_embedding in doc["image_embeddings"]:
            img_sim = cosine_similarity(query_embedding, img_embedding)
            max_image_sim = max(max_image_sim, img_sim)
        
        # 综合分数（文本权重 0.6，图片权重 0.4）
        final_score = 0.6 * text_sim + 0.4 * max_image_sim
        results.append((doc, final_score))
    
    # 排序并返回 Top-K
    results.sort(key=lambda x: x[1], reverse=True)
    return [doc for doc, score in results[:top_k]]

策略 11：在线学习与反馈循环

效果：准确率持续提升（长期使用可达 95%+）

原理：收集用户反馈（点赞/点踩），微调 Embedding 模型或 LLM。

from langchain_community.feedback import HumanFeedbackBrightness

# 1. 收集用户反馈
feedback_store = []

def collect_feedback(query: str, answer: str, retrieved_docs: List, rating: int):
    """收集用户反馈（rating: 1-5 分）"""
    feedback_store.append({
        "query": query,
        "answer": answer,
        "retrieved_docs": retrieved_docs,
        "rating": rating,
        "timestamp": time.time()
    })

# 2. 基于反馈微调 Embedding 模型（使用对比学习）
def finetune_embeddings(feedback_store: List, epochs: int = 3):
    """微调 Embedding 模型：让用户点赞的查询-文档对更相似"""
    from sentence_transformers import InputExample, losses
    
    # 构造训练数据
    train_examples = []
    for feedback in feedback_store:
        if feedback["rating"] >= 4:  # 好评：查询和文档应该相似
            for doc in feedback["retrieved_docs"]:
                train_examples.append(InputExample(
                    texts=[feedback["query"], doc.page_content],
                    label=1.0  # 相似度标签
                ))
        elif feedback["rating"] <= 2:  # 差评：查询和文档应该不相似
            for doc in feedback["retrieved_docs"]:
                train_examples.append(InputExample(
                    texts=[feedback["query"], doc.page_content],
                    label=0.0  # 不相似度标签
                ))
    
    # 微调（使用对比损失）
    model = SentenceTransformer('BAAI/bge-m3')
    train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
    train_loss = losses.CosineSimilarityLoss(model)
    
    model.fit(
        train_objectives=[(train_dataloader, train_loss)],
        epochs=epochs,
        warmup_steps=100
    )
    
    # 保存微调后的模型
    model.save("./finetuned_embeddings")

生产环境部署与监控

Docker 容器化部署

Dockerfile：

FROM python:3.11-slim

WORKDIR /app

# 安装依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# 复制代码
COPY . .

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

docker-compose.yml（包含 PostgreSQL + PGVector + RAG API）：

version: '3.8'

services:
  postgres:
    image: ankane/pgvector:latest
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
      POSTGRES_DB: rag_db
    ports:
      - "5432:5432"
    volumes:
      - postgres_data:/var/lib/postgresql/data

  rag-api:
    build: .
    ports:
      - "8000:8000"
    environment:
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      COHERE_API_KEY: ${COHERE_API_KEY}
      POSTGRES_CONNECTION: postgresql://user:password@postgres:5432/rag_db
    depends_on:
      - postgres
    volumes:
      - ./docs:/app/docs

volumes:
  postgres_data:

性能监控与日志

使用 Prometheus + Grafana 监控：

from prometheus_client import Counter, Histogram, start_http_server
import time

# 定义指标
REQUEST_COUNT = Counter('rag_requests_total', 'Total RAG requests')
REQUEST_LATENCY = Histogram('rag_request_latency_seconds', 'RAG request latency')
RETRIEVAL_ACCURACY = Counter('rag_retrieval_accuracy', 'Retrieval accuracy (relevant docs / total docs)')

# 在 RAG 链中记录指标
def monitored_rag_chain(query: str):
    REQUEST_COUNT.inc()
    
    start_time = time.time()
    
    # 执行 RAG
    result = rag_chain.invoke({"input": query})
    
    # 记录延迟
    latency = time.time() - start_time
    REQUEST_LATENCY.observe(latency)
    
    # 记录准确率（需要用户反馈或自动评估）
    relevant_docs = len([doc for doc in result["context"] if is_relevant(doc, query)])
    RETRIEVAL_ACCURACY.inc(relevant_docs / len(result["context"]))
    
    return result

# 启动 Prometheus metrics 服务器
start_http_server(8001)  # metrics 暴露在 http://localhost:8001/metrics

成本控制与缓存策略

成本优化策略：

1. Embedding 缓存：缓存已编码的文档向量（避免重复编码）
2. LLM 响应缓存：缓存相同查询的 LLM 回答（使用 Redis）
3. 批量调用：批量编码文档（减少 API 调用次数）
4. 使用更便宜的模型：检索用小型 Embedding 模型，生成用 GPT-4o-mini（而不是 GPT-4o）

import hashlib
import json
import redis

# 1. Embedding 缓存
redis_client = redis.Redis(host='localhost', port=6379, db=0)

def cached_embed(text: str, cache_key: str = None):
    """缓存 Embedding 结果"""
    if not cache_key:
        cache_key = hashlib.md5(text.encode()).hexdigest()
    
    # 检查缓存
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # 计算 Embedding
    embedding = embeddings.embed_query(text)
    
    # 存入缓存（过期时间 24 小时）
    redis_client.setex(cache_key, 86400, json.dumps(embedding))
    
    return embedding

# 2. LLM 响应缓存
def cached_llm_invoke(prompt: str):
    """缓存 LLM 响应"""
    cache_key = hashlib.md5(prompt.encode()).hexdigest()
    
    # 检查缓存
    cached = redis_client.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # 调用 LLM
    response = llm.invoke(prompt)
    
    # 存入缓存（过期时间 1 小时）
    redis_client.setex(cache_key, 3600, json.dumps(response.content))
    
    return response.content

真实案例：从 60% 到 94% 准确率的企业知识库 RAG

背景：某金融企业使用 RAG 构建内部知识库问答系统（包含监管文件、产品手册、历史工单）。

初始状态（Naive RAG）：

• 准确率：60%
• 主要问题：
  1. 分块不合理（固定 500 字符，经常把表格切成两半）
  2. 只使用向量检索（很多监管条款是精确匹配，向量检索效果差）
  3. 没有重排序（Top-5 结果中有 2-3 个不相关）

优化过程：

最终状态（Agentic RAG + GraphRAG）：

• 准确率：94%
• 平均响应时间：2.3 秒
• 成本：$0.05/查询（使用 GPT-4o-mini + 本地 BGE-M3 Embedding）

关键经验：

1. 语义分块是基础（分块不合理，后面怎么优化都无效）
2. 混合检索效果最好（BM25 + 向量，取长补短）
3. 重排序是性价比最高的优化（只需调用一次 Reranker API，准确率提升 8%）
4. GraphRAG 对多跳推理问题效果显著（如"哪些监管文件提到了 XXX 产品的风险？"）

总结与展望

本文回顾

本文系统讲解了 RAG 技术从 2023 年到 2026 年的演进路径，提供了生产级 RAG 系统的完整实现代码，并总结了 11 个性能优化策略（从 60% 到 94% 准确率）。

核心要点：

1. Naive RAG 已过时：2026 年生产环境必须使用 Advanced RAG 或 Agentic RAG。
2. 语义分块 + 混合检索 + 重排序是基准配置（准确率 80%+）。
3. Agentic RAG 是未来（让 Agent 自主决策检索策略）。
4. GraphRAG 对多跳推理问题效果显著（准确率提升 15-25%）。
5. 在线学习是持续提升准确率的关键（收集用户反馈，微调模型）。

未来展望（2026+）

1. Multi-modal RAG：支持文本、图片、表格、代码混合检索（CLIP + 各模态独立检索）。
2. LongContext RAG：利用 Gemini 2.5（200 万 token 上下文）或 Claude Sonnet 4.6（100 万 token），减少分块和检索的复杂性。
3. 端到端优化：联合训练 Retrieval 和 Generation（而不是分开优化）。
4. 个性化 RAG：根据用户画像和历史行为，个性化排序检索结果。

参考资料

1. LangChain 官方文档: https://python.langchain.com/docs/use_cases/question_answering/
2. LlamaIndex 官方文档: https://docs.llamaindex.ai/
3. RAG 论文精选: https://github.com/thunlp/RAGPaper
4. BGE-M3 Embedding 模型: https://huggingface.co/BAAI/bge-m3
5. Cohere Reranker API: https://cohere.com/rerank
6. GraphRAG 论文: https://arxiv.org/abs/2404.16130
7. Self-RAG 论文: https://arxiv.org/abs/2310.11511
8. CRAG 论文: https://arxiv.org/abs/2401.15884

版权声明：本文由程序员茄子原创，转载请注明出处。

更新日志：

• 2026-05-23：初始版本发布