/ RAG  向量数据库  LLM  Embedding  检索增强生成  AI工程  大语言模型  知识库 

RAG 工程实践全指南：从向量检索到生产部署


封面

一、RAG 技术概述与应用场景

RAG（Retrieval-Augmented Generation，检索增强生成）是当前大语言模型落地最主流的工程方案之一。它通过在推理阶段动态检索外部知识库，将相关文档片段注入 Prompt，从而显著提升 LLM 在垂直领域的准确性和时效性。

相比纯粹的 Fine-Tuning，RAG 具有以下核心优势：

  • 知识可更新：只需更新向量库，无需重新训练模型

  • 成本更低：省去昂贵的模型微调费用

  • 可解释性强：检索到的来源文档可直接展示给用户

  • 幻觉更少：基于真实文档生成，减少 LLM 编造内容

典型应用场景包括：企业内部知识库问答、代码文档助手、法律合规咨询、医疗知识检索等。

二、文档预处理与切片策略

RAG 的质量很大程度上取决于文档预处理阶段。原始文档（PDF、Word、Markdown、HTML）需经过以下流程：

  • 格式解析：使用 PyMuPDF、pdfplumber、unstructured 等库提取文本

  • 文本清洗：去除多余空白、特殊字符、页眉页脚

  • 切片（Chunking）：将长文本拆分为语义完整的小块

切片策略对检索效果影响极大，常见策略对比：

  • 固定大小切片：简单，但可能切断语义。推荐 chunk_size=512，overlap=64

  • 递归字符切片：LangChain 的 RecursiveCharacterTextSplitter，按段落→句子→词语递归切分

  • 语义切片：基于句子 Embedding 相似度动态切分，效果最好但计算成本高

  • 文档结构切片：利用 Markdown 标题、HTML 标签等结构信息切分

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n\n", "\n", "。", "！", "？", " ", ""]
)

docs = splitter.create_documents([raw_text])
print(f"切片数量: {len(docs)}")

三、Embedding 模型选型与向量化

Embedding 模型将文本转换为稠密向量，是 RAG 系统的核心组件。选型时需考虑：向量维度、最大 token 长度、多语言支持、推理速度和成本。

主流 Embedding 模型对比：

  • text-embedding-3-small（OpenAI）：1536 维，8192 token，性价比高，适合英文场景

  • text-embedding-3-large（OpenAI）：3072 维，效果更好，成本较高

  • BGE-M3（BAAI）：开源，1024 维，支持中英双语，推荐用于中文 RAG

  • Jina Embeddings v3：8192 token 超长上下文，支持多任务

  • text2vec-large-chinese：专为中文优化，本地部署友好

from sentence_transformers import SentenceTransformer

# 使用 BGE-M3 进行中文 Embedding
model = SentenceTransformer("BAAI/bge-m3")

texts = ["RAG技术原理", "向量数据库选型", "检索增强生成实践"]
embeddings = model.encode(texts, normalize_embeddings=True)
print(f"向量维度: {embeddings.shape}")  # (3, 1024)

四、向量数据库选型与实践

向量数据库负责存储和检索 Embedding，是 RAG 的"记忆存储"。主流方案对比：

  • Chroma：轻量级，开源，支持本地和客户端模式，适合原型开发

  • Milvus：高性能分布式，支持十亿级向量，适合生产环境大规模部署

  • Pinecone：全托管云服务，开箱即用，按用量计费

  • Weaviate：支持混合检索（向量+关键词），内置多租户

  • Qdrant：Rust 编写，高性能，支持 payload 过滤

  • pgvector：PostgreSQL 扩展，适合已有 PG 基础设施的团队

import chromadb
from chromadb.utils import embedding_functions

# 创建本地持久化 Chroma 集合
client = chromadb.PersistentClient(path="./chroma_db")

ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-m3"
)

collection = client.get_or_create_collection(
    name="knowledge_base",
    embedding_function=ef,
    metadata={"hnsw:space": "cosine"}
)

# 批量插入文档
collection.add(
    documents=[doc.page_content for doc in docs],
    metadatas=[{"source": doc.metadata.get("source", "")} for doc in docs],
    ids=[f"doc_{i}" for i in range(len(docs))]
)

# 语义检索
results = collection.query(
    query_texts=["如何优化RAG检索效果"],
    n_results=5
)

五、检索策略与重排序优化

基础向量检索（ANN 近似最近邻）并非总是最优，高质量 RAG 系统通常组合多种检索策略：

  • 混合检索（Hybrid Search）：向量检索 + BM25 关键词检索，通过 RRF（倒数排名融合）合并结果

  • 多查询检索：用 LLM 将原始问题扩展为多个子查询，提升召回率

  • HyDE（假设文档嵌入）：先让 LLM 生成一个假设答案，用假设答案的向量去检索

  • 重排序（Reranking）：使用 Cross-Encoder 模型对检索结果精细排序

from sentence_transformers import CrossEncoder

# 使用 BGE-Reranker 对检索结果重排
reranker = CrossEncoder("BAAI/bge-reranker-v2-m3")

query = "RAG系统如何减少幻觉？"
candidates = [doc["document"] for doc in initial_results]

# 计算 query-document 相关性分数
pairs = [[query, doc] for doc in candidates]
scores = reranker.predict(pairs)

# 按分数重排
ranked = sorted(zip(scores, candidates), reverse=True)
top_docs = [doc for _, doc in ranked[:3]]

六、RAG Pipeline 完整实现

将以上组件串联，构建一个完整的 RAG Pipeline：

from openai import OpenAI

client = OpenAI()

def rag_query(question: str, collection, reranker, top_k=5) -> str:
    # 1. 检索相关文档
    results = collection.query(
        query_texts=[question],
        n_results=top_k * 2  # 多检索一些供重排
    )
    candidates = results["documents"][0]
    
    # 2. 重排序
    pairs = [[question, doc] for doc in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(scores, candidates), reverse=True)
    context_docs = [doc for _, doc in ranked[:top_k]]
    
    # 3. 构建 Prompt
    context = "\n\n---\n\n".join(context_docs)
    prompt = f"""请根据以下参考资料回答问题。如果资料中没有相关信息，请说明无法从资料中找到答案。

参考资料：
{context}

问题：{question}

回答："""
    
    # 4. 调用 LLM 生成答案
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    
    return response.choices[0].message.content

# 使用示例
answer = rag_query("企业部署RAG系统需要注意哪些问题？", collection, reranker)
print(answer)

七、RAG 系统评估方法

科学评估 RAG 系统是工程化落地的关键。主要评估维度：

  • 检索质量：Recall@K（召回率）、MRR（平均倒数排名）、NDCG

  • 生成质量：忠实度（Faithfulness）、答案相关性（Answer Relevancy）、上下文精确率

  • 端到端质量：RAGAS 框架综合评分

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall

# 准备评估数据集
from datasets import Dataset

eval_data = {
    "question": ["RAG和Fine-Tuning有什么区别？"],
    "answer": [generated_answer],
    "contexts": [retrieved_docs],
    "ground_truth": ["RAG通过检索增强，Fine-Tuning通过训练更新模型参数..."]
}

dataset = Dataset.from_dict(eval_data)

# 运行评估
result = evaluate(
    dataset,
    metrics=[faithfulness, answer_relevancy, context_recall]
)
print(result)
# {'faithfulness': 0.92, 'answer_relevancy': 0.88, 'context_recall': 0.85}

建议在上线前构建一个包含 100-500 条标注问答对的评估集，持续监控各项指标的变化趋势，以此指导系统迭代优化。

八、生产部署最佳实践

将 RAG 系统推向生产环境时，还需关注以下工程实践：

  • 异步索引：文档更新通过消息队列异步处理，避免阻塞主流程

  • 缓存策略：对高频查询的 Embedding 和检索结果进行 Redis 缓存

  • 权限控制：多租户场景下使用 metadata 过滤实现文档级权限隔离

  • 监控告警：接入 Prometheus + Grafana，监控检索延迟、LLM 调用耗时、错误率

  • 版本管理：Embedding 模型升级时需重建索引，做好版本兼容

  • 安全防护：防范 Prompt 注入攻击，对用户输入进行校验和过滤

随着技术的不断演进，RAG 与 Agent、知识图谱的深度融合将带来更强大的 AI 应用能力。掌握 RAG 工程实践，是每位 AI 工程师走向生产级系统开发的必经之路。

发布评论

热门评论区：