RAG 检索增强生成

SunnyFan大约 14 分钟约 4222 字

RAG 检索增强生成

简介

RAG（Retrieval-Augmented Generation，检索增强生成）将信息检索与大语言模型结合，先从知识库中检索相关文档，再将检索结果作为上下文传给 LLM 生成回答。RAG 解决了 LLM 的知识过时、幻觉和领域知识不足等问题，是企业 AI 应用的主流架构。

RAG 的概念由 Meta AI 在 2020 年的论文 "Retrieval-Augmented Generation for Knowledge-Intensive NLP Tasks" 中首次提出。其核心动机是：纯参数化的 LLM 存在知识局限性——训练数据有截止日期、无法覆盖所有领域知识、容易产生幻觉。通过引入非参数化的检索组件，RAG 可以动态获取最新的、特定领域的知识，而不需要重新训练模型。

从系统架构的角度看，RAG 是一个典型的"检索-生成"两阶段系统：第一阶段从知识库中检索与查询相关的文档片段，第二阶段将检索结果注入 LLM 的上下文中生成回答。这种架构的优势在于：知识可以实时更新（只需更新知识库）、回答可以追溯到来源文档、成本远低于微调大模型。

特点

1.知识注入 — 外部知识库实时检索
2.减少幻觉 — 基于事实生成回答
3.实时更新 — 知识库随时更新无需重新训练
4.来源追溯 — 回答可追溯到原始文档
5.成本可控 — 无需微调大模型，使用现有 API 即可

RAG 的核心挑战

检索质量：如果检索不到相关文档，生成质量无从谈起
分块策略：文档如何分割直接影响检索精度
上下文窗口：检索结果过多会超出 LLM 的 Token 限制
延迟优化：检索+生成的总延迟需要满足用户体验
答案质量评估：如何自动化评估 RAG 系统的回答质量

RAG 基础架构

完整 RAG 流程

from openai import OpenAI
from sentence_transformers import SentenceTransformer
import numpy as np

client = OpenAI(api_key="your-api-key")
embed_model = SentenceTransformer('paraphrase-multilingual-MiniLM-L12-v2')

# 1. 文档加载和分割
documents = [
    "ASP.NET Core 的中间件管道按顺序处理 HTTP 请求，每个中间件可以执行操作并决定是否传递给下一个中间件。",
    "依赖注入是 ASP.NET Core 的核心功能，通过 IServiceCollection 注册服务，通过构造函数注入使用。",
    "Entity Framework Core 支持 Code First 模式，通过迁移命令自动生成数据库表结构。",
    "Redis 缓存支持三种模式：内存缓存、分布式缓存和 Redis 输出缓存。",
    "Docker 多阶段构建可以显著减小镜像大小，将编译环境和运行环境分离。",
]

# 2. 文档分块
def chunk_text(text, max_length=200, overlap=50):
    if len(text) <= max_length:
        return [text]
    chunks = []
    for i in range(0, len(text), max_length - overlap):
        chunk = text[i:i + max_length]
        if chunk:
            chunks.append(chunk)
    return chunks

chunks = []
for doc in documents:
    chunks.extend(chunk_text(doc))

# 3. 向量化
chunk_vectors = embed_model.encode(chunks)

# 4. 检索
def retrieve(query, chunks, vectors, top_k=3):
    query_vector = embed_model.encode([query])
    similarities = np.dot(query_vector, vectors.T)[0]
    top_indices = np.argsort(similarities)[-top_k:][::-1]
    return [(chunks[i], similarities[i]) for i in top_indices]

# 5. 生成
def rag_generate(query, chunks, vectors):
    # 检索相关文档
    results = retrieve(query, chunks, vectors, top_k=3)
    context = "\n\n".join([r[0] for r in results])

    # 构建 Prompt
    prompt = f"""根据以下参考信息回答问题。如果参考信息不足以回答，请说明。

参考信息：
{context}

问题：{query}

回答："""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3
    )
    return response.choices[0].message.content

# 使用
answer = rag_generate("ASP.NET Core 怎么处理HTTP请求？", chunks, chunk_vectors)
print(answer)

分块策略详解

def explain_chunking_strategies():
    """文档分块策略详解

    分块是 RAG 中最容易被忽视但极其重要的环节。

    分块策略对比：
    1. 固定长度分块：
       - 简单但可能切断语义
       - chunk_size=512, overlap=50 是常用默认值

    2. 句子级分块：
       - 按句子边界分割，保持语义完整
       - 适合问答场景

    3. 段落级分块：
       - 按段落或标题分割
       - 适合结构化文档

    4. 语义分块：
       - 根据语义相似度分割
       - 用 embedding 相似度判断分块边界

    5. 递归字符分块（LangChain 推荐）：
       - 优先按大分隔符（\n\n）分割
       - 再按小分隔符（\n, 。）分割
       - 最后按字符分割

    分块大小的影响：
    - 太大: 检索精度低（包含太多无关信息），超出上下文限制
    - 太小: 丢失上下文，语义不完整
    - 最佳范围: 256-1024 tokens（取决于任务和模型）
    """
    print("分块策略选择建议:")
    print("  通用问答: 512 tokens, overlap=50")
    print("  精确匹配: 256 tokens, overlap=25")
    print("  摘要生成: 1024 tokens, overlap=100")
    print("  代码文档: 按函数/类分块")
    print("  法律文档: 按条款/章节分块")

explain_chunking_strategies()

Embedding 模型选择

def explain_embedding_models():
    """Embedding 模型选择指南

    Embedding 模型决定了向量检索的质量。

    中文场景推荐：
    1. bge-large-zh-v1.5 (BAAI): 中文效果最好之一
    2. text2vec-large-chinese: 中文通用
    3. paraphrase-multilingual-MiniLM-L12-v2: 多语言，体积小
    4. m3e-base: 中文开源，效果好

    英文场景推荐：
    1. text-embedding-3-large (OpenAI): 效果最好
    2. text-embedding-ada-002 (OpenAI): 性价比高
    3. all-MiniLM-L6-v2: 开源，速度快

    选择考量：
    - 准确率: OpenAI > bge > MiniLM
    - 速度: MiniLM > bge > OpenAI API
    - 成本: 开源免费 < OpenAI API
    - 维度: 影响存储和检索速度
    """
    print("Embedding 模型选择:")
    print("  中文高精度: bge-large-zh-v1.5 (1024维)")
    print("  多语言通用: multilingual-MiniLM (384维)")
    print("  追求最佳: OpenAI text-embedding-3-large")
    print("  本地部署: text2vec / m3e")

explain_embedding_models()

高级 RAG 技术

混合检索

from sklearn.feature_extraction.text import TfidfVectorizer

class HybridRetriever:
    def __init__(self, documents, embed_model):
        self.documents = documents
        self.embed_model = embed_model
        self.doc_vectors = embed_model.encode(documents)

        # BM25/TF-IDF 检索器
        self.tfidf = TfidfVectorizer(max_features=500)
        self.tfidf_matrix = self.tfidf.fit_transform(documents)

    def retrieve(self, query, top_k=3, alpha=0.5):
        # 语义检索
        query_vector = self.embed_model.encode([query])
        semantic_scores = np.dot(query_vector, self.doc_vectors.T)[0]
        semantic_scores = (semantic_scores - semantic_scores.min()) / \
                         (semantic_scores.max() - semantic_scores.min() + 1e-8)

        # 关键词检索
        query_tfidf = self.tfidf.transform([query])
        keyword_scores = np.dot(query_tfidf, self.tfidf_matrix.T).toarray()[0]
        keyword_scores = (keyword_scores - keyword_scores.min()) / \
                        (keyword_scores.max() - keyword_scores.min() + 1e-8)

        # 混合评分
        combined = alpha * semantic_scores + (1 - alpha) * keyword_scores
        top_indices = np.argsort(combined)[-top_k:][::-1]

        return [(self.documents[i], combined[i]) for i in top_indices]

# 使用
retriever = HybridRetriever(documents, embed_model)
results = retriever.retrieve("Docker 镜像优化", alpha=0.6)

重排序（Re-ranking）

def rag_with_rerank(query, retriever, client, top_k=3, rerank_top=2):
    # 1. 初步检索（多召回）
    initial_results = retriever.retrieve(query, top_k=top_k * 3)

    # 2. LLM 重排序
    rerank_prompt = f"""对以下文档按与查询的相关性打分（1-10）：

查询：{query}

文档：
"""
    for i, (doc, score) in enumerate(initial_results):
        rerank_prompt += f"\n[{i+1}] {doc}"

    rerank_prompt += "\n\n请按格式输出：编号:分数，例如 1:8"

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": rerank_prompt}],
        temperature=0.1
    )

    # 3. 选择最相关的文档
    # 解析重排序结果，选择 top rerank_top 个
    context = "\n\n".join([doc for doc, _ in initial_results[:rerank_top]])

    # 4. 生成最终答案
    final_prompt = f"参考信息：\n{context}\n\n问题：{query}\n\n回答："
    result = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": final_prompt}]
    )
    return result.choices[0].message.content

RAG 评估

def explain_rag_evaluation():
    """RAG 系统评估方法

    RAG 评估需要分别评估检索和生成两个阶段：

    检索评估：
    1. Recall@K: 相关文档是否被召回
    2. Precision@K: 召回的文档中相关文档的比例
    3. MRR (Mean Reciprocal Rank): 第一个相关文档的排名倒数均值
    4. NDCG: 考虑位置权重的归一化折损累计增益

    生成评估：
    1. Faithfulness: 回答是否忠实于检索到的上下文
    2. Answer Relevance: 回答是否与问题相关
    3. Context Relevance: 检索到的上下文是否与问题相关
    4. 有参考答案时: BLEU、ROUGE、BERTScore

    端到端评估：
    1. RAGAS 框架: 自动化 RAG 评估
    2. 人工评估: 最可靠但成本高
    3. A/B 测试: 线上对比不同版本
    """
    print("RAG 评估框架:")
    print("  RAGAS: 开源自动化 RAG 评估")
    print("  Trulens: RAG 可观测性平台")
    print("  人工评估: 100-500 条标注样本")
    print("  评估维度: Faithfulness + Relevance + Completeness")

explain_rag_evaluation()

Query 改写与扩展

def explain_query_transformation():
    """查询改写和扩展

    原始查询可能表述不清晰、缺少关键信息或过于宽泛。
    通过查询改写可以提升检索质量。

    方法：
    1. LLM 改写: 用 LLM 将模糊查询改写为更明确的查询
    2. 查询扩展: 添加同义词、相关术语
    3. 子查询分解: 将复杂查询分解为多个子查询
    4. HyDE: 先用 LLM 生成假设性答案，用答案做检索
    5. 多轮对话: 利用对话历史补充上下文
    """
    print("查询改写策略:")
    print("  1. LLM 改写: '怎么部署' -> 'ASP.NET Core 应用部署到 Docker 的步骤'")
    print("  2. HyDE: 先生成假设答案，再用答案做检索")
    print("  3. 子查询分解: '性能优化' -> ['缓存优化', '数据库优化', '代码优化']")
    print("  4. 对话历史: 结合前几轮对话补充查询上下文")

explain_query_transformation()

RAG 架构模式

模式	说明	适用场景
Naive RAG	基础检索+生成	简单问答
Advanced RAG	混合检索+重排序	高精度需求
Modular RAG	可替换的模块化组件	灵活定制
Agentic RAG	Agent 决策检索策略	复杂推理
Graph RAG	知识图谱增强	实体关系丰富
Multi-modal RAG	图文混合检索	含图片文档

优点

1.知识实时 — 知识库更新即可使用
2.减少幻觉 — 基于检索事实生成
3.成本较低 — 无需微调大模型
4.可追溯 — 引用来源文档

缺点

1.检索质量 — 检索不准导致回答偏差
2.上下文限制 — 检索文档过多超出 Token 限制
3.分块策略 — 文档分割粒度影响效果
4.延迟增加 — 检索+生成两步处理

总结

RAG 核心流程：文档分块 -> 向量化存储 -> 检索相关片段 -> 注入 LLM Prompt -> 生成回答。基础 RAG 用语义检索（embedding + 余弦相似度），进阶用混合检索（语义+关键词）+ 重排序。关键优化点：分块策略（chunk size/overlap）、检索策略（混合检索）、重排序（LLM/Cross-encoder）。

关键知识点

RAG 的核心不只是向量检索，而是分块、召回、重排和答案生成的协同。
分块策略对检索质量的影响往往比 Embedding 模型更大。
混合检索（语义+关键词）通常比纯语义检索效果更好。
评估 RAG 时要分别评估检索质量和生成质量。

项目落地视角

先建立评估集，再对分块策略、检索方式和重排策略做实验。
给数据来源、Prompt 模板、Embedding 版本、评估集和实验结果做版本管理。
上线前准备兜底策略，例如拒答、回退、人工审核或缓存降级。

常见误区

只换模型，不分析是分块还是召回出了问题。
不建评估集就频繁调参。
忽略检索延迟对用户体验的影响。
使用不合适的 Embedding 模型（如用英文模型处理中文）。

进阶路线

学习混合检索（语义 + BM25）和重排序技术。
探索 Graph RAG 和 Agentic RAG。
掌握 RAGAS 评估框架和自动化评估流程。
了解向量数据库的高级功能（过滤、混合搜索）。

适用场景

企业知识库问答、技术文档检索。
智能客服、内部知识助手。
法律、医疗、金融等专业领域的知识问答。

落地建议

先用小数据集（50-100 条）验证 RAG 管线。
系统测试不同的分块大小和 overlap。
使用混合检索提升召回率。
建立评估集，追踪 Faithfulness 和 Relevance。

排错清单

检查分块是否过碎或过大。
检查 Embedding 模型是否与文档语言匹配。
检查检索 Top-K 数量和相似度阈值。
检查 Prompt 是否明确要求基于上下文回答。

复盘问题

当前 RAG 系统的检索召回率是多少？
分块策略是否经过系统性的消融实验？
是否有评估集来追踪系统效果的变化？

RAG 与 LLM 的 Prompt 工程结合

def build_rag_prompt(query: str, context: str, history: list = None) -> str:
    """构建高质量 RAG Prompt

    Prompt 设计要点：
    1. 明确角色和任务边界
    2. 要求基于上下文回答，并标注来源
    3. 明确拒绝机制（上下文不足时）
    4. 控制输出格式
    """
    system_prompt = """你是一个知识问答助手。请严格基于以下参考信息回答用户问题。

规则：
1. 只使用参考信息中的内容回答，不要编造信息
2. 如果参考信息不足以回答问题，请明确说明
3. 回答时标注信息来源（如 [来源1]、[来源2]）
4. 如果参考信息之间有矛盾，列出不同观点"""

    context_section = "参考信息：\n"
    for i, chunk in enumerate(context.split("\n\n")):
        context_section += f"[来源{i+1}] {chunk}\n\n"

    user_prompt = f"{context_section}\n问题：{query}\n\n回答："

    return system_prompt, user_prompt

# 带来源追溯的 RAG
def rag_with_sources(query, chunks, vectors, embed_model, client):
    results = retrieve(query, chunks, vectors, top_k=3)
    context = "\n\n".join([r[0] for r in results])
    sources = [{"text": r[0][:100], "score": r[1]} for r in results]

    system_prompt, user_prompt = build_rag_prompt(query, context)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.3
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": sources,
        "model": "gpt-4o-mini",
        "chunks_used": len(results)
    }

文档预处理与清洗

import re

def preprocess_document(text: str) -> str:
    """文档预处理

    预处理步骤：
    1. 去除多余空白和特殊字符
    2. 修正常见编码问题
    3. 去除页眉页脚（通常包含页码、公司名等噪声）
    4. 合并被分割的段落
    5. 标准化特殊格式（表格、代码块）
    """
    # 去除多余空白
    text = re.sub(r'\n{3,}', '\n\n', text)
    text = re.sub(r' {2,}', ' ', text)

    # 去除常见噪声模式
    text = re.sub(r'第\s*\d+\s*页.*?\n', '', text)  # 页码
    text = re.sub(r'机密.*?\n', '', text)  # 机密标记

    # 合并被分割的行（中英文行末无标点）
    text = re.sub(r'([^\n。.!?])\n(?=[^\n\s])', r'\1', text)

    # 保留代码块标记
    text = re.sub(r'```\w*\n', '```code\n', text)

    return text.strip()

def chunk_by_structure(text: str, max_length: int = 500, overlap: int = 50) -> list[str]:
    """按结构分块（优先按标题、段落分割）

    优于固定长度分块：
    - 保持语义完整性
    - 保留文档结构信息
    - 适合 Markdown、HTML 等结构化文档
    """
    # 按标题分割
    sections = re.split(r'\n(#{1,4}\s+.+)\n', text)

    chunks = []
    current_chunk = ""

    for i, section in enumerate(sections):
        if re.match(r'^#{1,4}\s+', section):
            # 这是标题，作为新块的开始
            if current_chunk.strip():
                chunks.append(current_chunk.strip())
            current_chunk = section + "\n"
        else:
            current_chunk += section + "\n"

            # 如果当前块超过最大长度，按段落分割
            while len(current_chunk) > max_length:
                # 找到最近的段落边界
                split_pos = current_chunk.rfind('\n\n', 0, max_length)
                if split_pos == -1:
                    split_pos = current_chunk.rfind('\n', 0, max_length)
                if split_pos == -1:
                    split_pos = max_length

                chunks.append(current_chunk[:split_pos].strip())
                current_chunk = current_chunk[split_pos - overlap:]

    if current_chunk.strip():
        chunks.append(current_chunk.strip())

    return chunks

# 分块质量评估
def evaluate_chunking(chunks: list[str]) -> dict:
    """评估分块质量"""
    lengths = [len(c) for c in chunks]
    return {
        "chunk_count": len(chunks),
        "avg_length": sum(lengths) / len(lengths),
        "min_length": min(lengths),
        "max_length": max(lengths),
        "empty_chunks": sum(1 for c in chunks if not c.strip()),
        "very_short": sum(1 for l in lengths if l < 50),  # 过短可能丢失上下文
        "very_long": sum(1 for l in lengths if l > 1000),  # 过长影响检索精度
    }

RAG 系统性能优化

def explain_rag_performance():
    """RAG 系统性能优化

    端到端延迟分解：
    1. Embedding 延迟（查询向量化）：50-200ms
    2. 向量检索延迟：5-50ms
    3. Prompt 构建：1-5ms
    4. LLM 生成延迟：500-3000ms
    5. 总延迟：600-3500ms

    优化策略：

    1. 检索层优化：
       - 使用更小的 embedding 模型（如 MiniLM 替代 large 模型）
       - 预计算查询 embedding（缓存高频查询）
       - 调整 top_k 和 score_threshold 减少上下文量
       - 使用 HNSW 参数调优减少检索时间

    2. 生成层优化：
       - 使用更快的模型（gpt-4o-mini 替代 gpt-4o）
       - 控制 max_tokens 限制生成长度
       - 使用流式输出降低首字延迟
       - 缓存高频问答对

    3. 架构优化：
       - 检索和生成并行化（预取可能相关的文档）
       - 使用异步处理（先返回检索结果，再生成回答）
       - 多级缓存（embedding 缓存 + 结果缓存）

    4. 成本优化：
       - 缓存 embedding 结果（相同文本不重复计算）
       - 使用开源 embedding 模型替代 API
       - 小模型优先，大模型兜底
       - 控制 RAG 的 Token 消耗（精简上下文）
    """
    print("性能优化优先级:")
    print("  1. 减少 LLM Token 消耗（最直接影响成本）")
    print("  2. 缓存高频查询的 embedding 和回答")
    print("  3. 使用更小更快的 embedding 模型")
    print("  4. 优化分块策略提高检索精度")

explain_rag_performance()

流式 RAG 输出

from openai import OpenAI

client = OpenAI(api_key="your-api-key")

def rag_stream(query, chunks, vectors, embed_model):
    """流式 RAG — 逐 token 输出，降低首字延迟"""
    results = retrieve(query, chunks, vectors, top_k=3)
    context = "\n\n".join([r[0] for r in results])

    prompt = f"""参考信息：
{context}

问题：{query}

回答："""

    stream = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        stream=True  # 启用流式输出
    )

    # 逐 token 输出
    for chunk in stream:
        if chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)
    print()

# 流式输出的优势：
# 1. 首字延迟从 2-3 秒降到 200-500ms
# 2. 用户体验更好（逐步看到内容）
# 3. 可以实现"取消生成"功能
# 4. 适合长回答场景