健兼
(LLM系列)RAG(检索增强生成)原理与实践

(LLM系列)RAG(检索增强生成)原理与实践


AI RAG 向量检索 Embedding LLM
Last updated on

RAG(检索增强生成)原理与实践

引言

在大语言模型（LLM）蓬勃发展的今天，如何让AI更准确地回答特定领域的问题成为了一个关键挑战。RAG（Retrieval-Augmented Generation，检索增强生成）技术应运而生，它通过结合外部知识库和生成模型，显著提升了AI回答的准确性和时效性。

本文将深入探讨RAG的核心原理，重点解析向量检索和上下文注入两大关键技术，并提供实践指导。

一、RAG是什么？

1.1 核心思想

RAG的核心思想非常直观：在生成答案之前，先从知识库中检索相关信息，然后将这些信息作为上下文提供给大语言模型，让模型基于这些”参考资料”来生成更准确的回答。

这就像是让AI在开卷考试而不是闭卷考试——它可以查阅资料后再作答。

1.2 为什么需要RAG？

传统LLM面临几个关键问题：

  • 知识时效性：模型的知识截止于训练时间，无法获取最新信息
  • 幻觉问题：模型可能生成看似合理但实际错误的内容
  • 专业领域知识不足：通用模型对特定领域的深度知识有限
  • 成本问题：频繁微调大模型成本高昂

RAG通过外部知识检索优雅地解决了这些问题，无需重新训练模型。

二、向量检索：RAG的核心引擎

2.1 什么是向量检索？

向量检索是RAG系统的第一步，也是最关键的一步。它的任务是从海量文档中快速找出与用户问题最相关的内容。

文本向量化

文本向量化（Embedding）是将文本转换为高维向量的过程：

"什么是机器学习？" → [0.12, -0.34, 0.56, ..., 0.89]  # 维度通常为384-1536

向量的特点：

  • 语义相似的文本，向量距离更近
  • 向量可以进行数学运算（相似度计算）
  • 降维后可视化（理解语义空间）

常用的Embedding模型

  • OpenAI text-embedding-3-small/large：性能强大，支持多语言
  • sentence-transformers：开源方案，适合中文
  • BGE系列：国内优秀的开源模型
  • m3e：专门针对中文优化

2.2 向量检索的工作流程

用户问题 → Embedding模型 → 查询向量 → 向量数据库 → Top-K 相似文档

步骤详解：

  1. 文档预处理：

    • 文档切片（Chunking）：将长文档分割成适当大小的片段（通常300-1000 tokens）
    • 向量化：使用Embedding模型将每个片段转换为向量
    • 存储：将向量及元数据存入向量数据库

  2. 查询处理：

    • 用户问题同样经过Embedding模型转换为查询向量
    • 在向量数据库中进行相似度搜索
    • 返回Top-K个最相关的文档片段

2.3 相似度计算方法

余弦相似度（最常用）

import numpy as np


def cosine_similarity(vec1, vec2):
    """计算两个向量的余弦相似度"""
    dot_product = np.dot(vec1, vec2)
    norm_product = np.linalg.norm(vec1) * np.linalg.norm(vec2)
    return dot_product / norm_product


# 示例
query_vec = np.array([0.5, 0.3, 0.8])
doc_vec = np.array([0.6, 0.2, 0.9])
similarity = cosine_similarity(query_vec, doc_vec)
print(f"相似度: {similarity:.3f}")  # 输出：0.989

优点：不受向量长度影响，只关注方向

欧氏距离

def euclidean_distance(vec1, vec2):
    """计算欧氏距离（距离越小越相似）"""
    return np.linalg.norm(vec1 - vec2)

点积

def dot_product_similarity(vec1, vec2):
    """点积相似度"""
    return np.dot(vec1, vec2)

2.4 向量数据库选择

数据库特点适用场景
Pinecone云服务，易用性强快速原型开发
Milvus开源，性能强大大规模生产环境
Weaviate支持多模态复杂查询需求
Chroma轻量级，易部署小型项目、本地开发
FAISSFacebook开源，速度快研究和实验

2.5 优化向量检索的技巧

结合关键词检索和向量检索：

# 伪代码示例
def hybrid_search(query, alpha=0.5):
    # 向量检索得分
    vector_results = vector_search(query)


    # 关键词检索得分（BM25）
    keyword_results = bm25_search(query)


    # 加权融合
    final_scores = alpha * vector_results + (1-alpha) * keyword_results
    return top_k(final_scores)

技巧2：重排序（Reranking）

使用更强大的模型对初步检索结果重新排序：

def rerank(query, initial_results):
    """使用交叉编码器重排序"""
    cross_encoder = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')


    pairs = [(query, doc) for doc in initial_results]
    scores = cross_encoder.predict(pairs)


    # 按新得分重新排序
    return sort_by_scores(initial_results, scores)

技巧3：查询扩展

扩展用户查询以提高召回率：

def query_expansion(query):
    """生成查询的多个变体"""
    expanded_queries = [
        query,
        f"关于{query}的详细解释",
        f"{query}是什么意思",
        f"如何理解{query}"
    ]
    return expanded_queries

三、上下文注入：让LLM”看见”外部知识

3.1 上下文注入的原理

上下文注入是将检索到的文档作为提示（Prompt）的一部分，提供给LLM。这个过程就像给AI提供”参考资料”。

基本结构

系统指令 + 检索到的上下文 + 用户问题 → LLM → 生成答案

3.2 Prompt工程最佳实践

模板示例1：基础RAG Prompt

def create_rag_prompt(query, context_docs):
    prompt = f"""你是一个专业的AI助手。请基于以下参考资料回答用户的问题。


参考资料：
{format_context(context_docs)}


重要提示：
1. 只基于上述参考资料回答问题
2. 如果参考资料中没有相关信息，请明确说明
3. 引用参考资料时请注明来源


用户问题：{query}


请提供准确、详细的回答："""


    return prompt


def format_context(docs):
    """格式化上下文文档"""
    formatted = []
    for i, doc in enumerate(docs, 1):
        formatted.append(f"[文档{i}]\n{doc['content']}\n来源：{doc['source']}\n")
    return "\n".join(formatted)

模板示例2：带引用的高级Prompt

def create_advanced_rag_prompt(query, context_docs):
    prompt = f"""# 角色
你是一个严谨的知识问答助手。


# 任务
基于提供的参考资料回答用户问题，并标注信息来源。


# 参考资料
{format_numbered_context(context_docs)}


# 回答要求
1. **准确性**：确保答案完全基于参考资料
2. **引用标注**：使用[1][2]标注信息来源
3. **完整性**：综合所有相关资料给出全面回答
4. **诚实性**：如果资料不足，明确说明局限性


# 用户问题
{query}


# 你的回答
"""
    return prompt


def format_numbered_context(docs):
    """带编号的上下文格式化"""
    formatted = []
    for i, doc in enumerate(docs, 1):
        formatted.append(f"[{i}] {doc['content']}\n(来源: {doc['source']})\n")
    return "\n".join(formatted)

3.3 上下文窗口管理

问题：上下文过长

当检索到的文档过多或过长时，可能超出LLM的上下文窗口限制。

解决方案

方案1：智能截断

def truncate_context(docs, max_tokens=2000):
    """智能截断上下文"""
    truncated = []
    current_tokens = 0


    for doc in docs:
        doc_tokens = count_tokens(doc['content'])
        if current_tokens + doc_tokens <= max_tokens:
            truncated.append(doc)
            current_tokens += doc_tokens
        else:
            # 截断最后一个文档
            remaining = max_tokens - current_tokens
            doc['content'] = truncate_to_tokens(doc['content'], remaining)
            truncated.append(doc)
            break


    return truncated

方案2：分层检索

def hierarchical_retrieval(query, k1=10, k2=3):
    """两阶段检索：先召回，再精选"""
    # 第一阶段：快速召回更多文档
    candidates = vector_search(query, top_k=k1)


    # 第二阶段：使用更强模型精选最相关的
    final_docs = rerank(query, candidates, top_k=k2)


    return final_docs

方案3：文档摘要

async def summarize_docs(docs, llm):
    """对长文档进行摘要"""
    summaries = []
    for doc in docs:
        if len(doc['content']) > 1000:
            summary = await llm.summarize(doc['content'])
            doc['content'] = summary
        summaries.append(doc)
    return summaries

3.4 上下文质量优化

技巧1：去重

def deduplicate_docs(docs, similarity_threshold=0.9):
    """移除相似度过高的重复文档"""
    unique_docs = []
    for doc in docs:
        is_duplicate = False
        for existing in unique_docs:
            if cosine_similarity(doc['embedding'], existing['embedding']) > similarity_threshold:
                is_duplicate = True
                break
        if not is_duplicate:
            unique_docs.append(doc)
    return unique_docs

技巧2：相关性过滤

def filter_by_relevance(docs, min_score=0.7):
    """过滤掉相关性低的文档"""
    return [doc for doc in docs if doc['score'] >= min_score]

技巧3：多样性采样

def diversify_results(docs, top_k=5):
    """确保结果的多样性"""
    selected = [docs[0]]  # 选择最相关的


    for doc in docs[1:]:
        if len(selected) >= top_k:
            break


        # 计算与已选文档的最大相似度
        max_sim = max([cosine_similarity(doc['embedding'], s['embedding'])
                       for s in selected])


        # 如果不太相似，则添加
        if max_sim < 0.85:
            selected.append(doc)


    return selected

四、完整RAG系统实现

4.1 系统架构

┌─────────────┐
│  用户查询   │
└──────┬──────┘
       │
       ▼
┌─────────────────┐
│  查询处理模块   │ ← 查询改写、扩展
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  向量检索引擎   │ ← 向量数据库
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  重排序模块     │ ← 提高精确度
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  上下文构建     │ ← Prompt工程
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  LLM生成        │ ← 生成答案
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  后处理与验证   │ ← 事实检查
└──────┬──────────┘
       │
       ▼
┌─────────────────┐
│  返回结果       │
└─────────────────┘

4.2 Python实现示例

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA


class RAGSystem:
    def __init__(self, documents):
        """初始化RAG系统"""
        # 1. 文档处理
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=500,
            chunk_overlap=50,
            separators=["\n\n", "\n", "。", "！", "？", ".", "!", "?"]
        )


        # 2. Embedding模型
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")


        # 3. 向量数据库
        self.vectorstore = self._build_vectorstore(documents)


        # 4. LLM
        self.llm = OpenAI(temperature=0)


        # 5. 检索器
        self.retriever = self.vectorstore.as_retriever(
            search_type="mmr",  # 最大边际相关性
            search_kwargs={
                "k": 4,
                "fetch_k": 20,
                "lambda_mult": 0.5
            }
        )


    def _build_vectorstore(self, documents):
        """构建向量存储"""
        # 切分文档
        chunks = self.text_splitter.split_documents(documents)


        # 创建向量数据库
        vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory="./chroma_db"
        )


        return vectorstore


    def query(self, question):
        """执行RAG查询"""
        # 创建问答链
        qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.retriever,
            return_source_documents=True,
            chain_type_kwargs={
                "prompt": self._create_prompt()
            }
        )


        # 执行查询
        result = qa_chain({"query": question})


        return {
            "answer": result["result"],
            "sources": result["source_documents"]
        }


    def _create_prompt(self):
        """创建Prompt模板"""
        from langchain.prompts import PromptTemplate


        template = """基于以下参考资料回答问题。如果资料中没有答案，请说"我不知道"。


参考资料：
{context}


问题：{question}


详细回答："""


        return PromptTemplate(
            template=template,
            input_variables=["context", "question"]
        )


# 使用示例
from langchain.document_loaders import TextLoader


# 加载文档
loader = TextLoader("knowledge_base.txt")
documents = loader.load()


# 创建RAG系统
rag = RAGSystem(documents)


# 查询
result = rag.query("什么是机器学习？")
print(f"回答：{result['answer']}")
print(f"参考文档数量：{len(result['sources'])}")

4.3 高级优化：多查询RAG

class AdvancedRAG:
    def multi_query_retrieval(self, question):
        """生成多个查询角度"""
        # 使用LLM生成问题的不同表述
        variations = self.llm.generate_variations(question, num=3)


        all_docs = []
        for variation in variations:
            docs = self.retriever.get_relevant_documents(variation)
            all_docs.extend(docs)


        # 去重和排序
        unique_docs = self.deduplicate(all_docs)
        ranked_docs = self.rerank(question, unique_docs)


        return ranked_docs[:5]


    def self_query_with_metadata(self, question):
        """基于元数据的自查询"""
        # 从问题中提取过滤条件
        metadata_filter = self.extract_metadata_filter(question)


        # 在向量搜索中应用过滤
        docs = self.vectorstore.similarity_search(
            question,
            filter=metadata_filter,
            k=5
        )


        return docs

五、实践案例与应用场景

5.1 企业知识库问答

场景：企业内部有大量文档（产品手册、政策文档、FAQ等）

实现要点：

  • 文档分类和元数据管理
  • 权限控制
  • 定期更新向量库
# 示例：企业知识库RAG
class EnterpriseRAG:
    def __init__(self):
        self.vectorstore = Chroma(
            collection_name="company_docs",
            embedding_function=embeddings
        )


    def add_document(self, doc, metadata):
        """添加文档并包含元数据"""
        chunks = self.split_document(doc)


        for chunk in chunks:
            self.vectorstore.add_texts(
                texts=[chunk],
                metadatas=[{
                    "department": metadata["department"],
                    "doc_type": metadata["doc_type"],
                    "last_updated": metadata["date"],
                    "access_level": metadata["access_level"]
                }]
            )


    def query_with_access_control(self, question, user_level):
        """带权限控制的查询"""
        results = self.vectorstore.similarity_search(
            question,
            filter={"access_level": {"$lte": user_level}},
            k=5
        )
        return results

5.2 客服智能问答

场景：自动回答客户常见问题

实现要点：

  • 快速响应时间
  • 多轮对话上下文管理
  • 答案质量监控

5.3 学术研究助手

场景：帮助研究人员查找和总结文献

实现要点：

  • 支持PDF解析
  • 引用管理
  • 多模态检索（文本+图表）

六、评估与优化

6.1 评估指标

检索质量指标

def calculate_retrieval_metrics(retrieved_docs, relevant_docs):
    """计算检索指标"""
    retrieved_ids = set([doc['id'] for doc in retrieved_docs])
    relevant_ids = set([doc['id'] for doc in relevant_docs])


    # 召回率 (Recall)
    recall = len(retrieved_ids & relevant_ids) / len(relevant_ids)


    # 精确率 (Precision)
    precision = len(retrieved_ids & relevant_ids) / len(retrieved_ids)


    # F1分数
    f1 = 2 * (precision * recall) / (precision + recall)


    # MRR (Mean Reciprocal Rank)
    for i, doc in enumerate(retrieved_docs, 1):
        if doc['id'] in relevant_ids:
            mrr = 1 / i
            break


    return {
        "recall": recall,
        "precision": precision,
        "f1": f1,
        "mrr": mrr
    }

生成质量指标

  • 答案准确性：与标准答案的相似度
  • 幻觉率：生成内容中不基于参考资料的比例
  • 完整性：是否完整回答了问题
  • 引用准确性：引用是否正确

6.2 常见问题与解决方案

问题原因解决方案
检索不到相关文档Embedding模型不合适更换或微调Embedding模型
答案包含幻觉上下文不足或Prompt不当优化Prompt，增加”仅基于资料回答”约束
响应速度慢检索或生成耗时长使用更快的向量数据库，减少检索文档数
答案质量不稳定检索结果质量波动增加重排序步骤，提高检索精确度

6.3 持续优化策略

  1. A/B测试：对比不同检索策略和Prompt的效果
  2. 用户反馈循环：收集用户评价，优化系统
  3. 定期评估：建立测试集，定期评估系统性能
  4. 模型更新：跟踪最新的Embedding和LLM模型

七、未来趋势与展望

7.1 多模态RAG

支持图像、音频等多种模态的检索和生成。

7.2 自适应RAG

根据问题类型自动选择最佳检索策略。

7.3 知识图谱增强

结合结构化知识图谱提升推理能力。

7.4 实时RAG

支持流式检索和增量生成，提升用户体验。

总结

RAG技术通过向量检索和上下文注入两大核心机制，成功地将外部知识与大语言模型结合，显著提升了AI系统的准确性和实用性。

关键要点回顾

  1. 向量检索是基础：选择合适的Embedding模型和向量数据库至关重要
  2. 上下文注入是关键：精心设计的Prompt能大幅提升答案质量
  3. 优化是持续的：通过混合检索、重排序、元数据过滤等技术不断改进
  4. 评估要全面：关注检索和生成两个阶段的指标

实践建议

  • 从简单开始：先实现基础RAG，再逐步优化
  • 重视数据质量：高质量的文档是RAG成功的前提
  • 持续迭代：基于用户反馈和评估结果不断改进
  • 选择合适的工具栈：根据实际需求选择Embedding模型、向量数据库和LLM

RAG技术正在快速发展，掌握其原理与实践，将帮助你构建更智能、更可靠的AI应用。