RAG 架构：从朴素检索到智能知识系统

01 RAG 是什么？为什么需要它？

RAG（Retrieval-Augmented Generation）是一种将外部知识检索与大语言模型生成相结合的架构模式。核心思想：与其让 LLM 死记硬背所有知识，不如在需要时动态检索相关信息，让模型基于真实数据生成回答。

为什么需要 RAG？

RAG vs 微调 vs 长上下文

02 何时用 RAG，何时不用

✅ 适合用 RAG 的场景

• 知识库问答 — 企业内部文档、FAQ、政策查询
• 客服机器人 — 基于产品文档的自动客服
• 代码助手 — 基于代码库的编程辅助（参考 Agent 记忆系统 中 Aider 的 RepoMap 方案）
• 法律/医疗 — 需要精确引用原文的专业问答
• 数据分析 — 基于结构化数据的自然语言查询
• 实时信息 — 新闻、股价、天气等需要最新数据的场景

❌ 不适合用 RAG 的场景

• 创意写作 — 不需要事实依据的场景（小说、文案）
• 简单对话 — 闲聊、情感陪伴，检索反而增加延迟
• 纯推理任务 — 数学、逻辑推理，不需要外部知识
• 知识已内化 — 模型已充分掌握的通用知识（如基础编程语法）
• 数据极度敏感 — 不允许任何外部存储/检索的场景

03 RAG Pipeline 全景

一个完整的 RAG 系统分为两大阶段：离线索引（Indexing）和在线查询（Querying）。

朴素 RAG vs 高级 RAG

04 Embedding 模型选型

Embedding 模型将文本转换为高维向量，是 RAG 的基石。选型直接影响检索质量和性能。

主流模型对比

选型决策

代码：OpenAI Embedding

from openai import OpenAI

client = OpenAI()

def embed_texts(texts: list[str], model="text-embedding-3-small") -> list[list[float]]:
    """批量生成 Embedding，支持维度裁剪"""
    response = client.embeddings.create(
        input=texts,
        model=model,
        dimensions=512  # 裁剪到 512 维，节省 2/3 存储
    )
    return [item.embedding for item in response.data]

# 单条 vs 批量
query_vec = embed_texts(["什么是 RAG？"])[0]
doc_vecs = embed_texts(["RAG 是检索增强生成", "微调是..."])

代码：本地 BGE-M3

from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel('BAAI/bge-m3', use_fp16=True)

# 支持三种检索粒度：dense / sparse / colbert
embeddings = model.encode(
    ["RAG 架构详解", "向量数据库对比"],
    return_dense=True,
    return_sparse=True,
    return_colbert_vecs=True,
)

dense_vecs = embeddings['dense_vecs']       # (2, 1024) 密集向量
sparse_vecs = embeddings['lexical_weights']  # 稀疏词权重（类似 BM25）
colbert_vecs = embeddings['colbert_vecs']    # (2, seq_len, 128) token 级向量

05 Chunking 策略：分块的艺术

Chunking 决定了检索粒度——太粗则噪声多，太细则丢失上下文。这是 RAG 系统中最被低估但对效果影响最大的环节之一。

六大分块策略

关键参数调优

代码：LangChain 递归分块

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=512,
    chunk_overlap=64,
    separators=["\n## ", "\n### ", "\n\n", "\n", "。", "！", "？", ".", " ", ""],
    length_function=lambda x: len(x),  # 或 tiktoken 计数器
)

chunks = splitter.split_text(markdown_text)

# 每个chunk自动继承元数据
print(f"生成 {len(chunks)} 个分块")
print(f"平均长度: {sum(len(c) for c in chunks) / len(chunks):.0f} 字符")

代码：语义分块

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 基于语义相似度断点切割
splitter = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",  # 百分位数断点
    breakpoint_threshold_amount=75,          # 前 75% 相似度内不断
)

chunks = splitter.split_text(long_document)

# 输出示例：
# Chunk 1: "RAG 是检索增强生成..." (语义连贯的第一段)
# Chunk 2: "Embedding 模型将文本..." (语义转换处断开)

代码：父子分块（Small-to-Big Retrieval）

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter

# 子分块器：小而精准，用于检索
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 父分块器：大而完整，用于提供上下文
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

vectorstore = Chroma(embedding_function=OpenAIEmbeddings())
docstore = InMemoryStore()  # 父文档存储

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,        # 存子分块的向量
    docstore=docstore,              # 存父分块的原文
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

# 检索时：先搜子分块 → 返回其父分块作为上下文
results = retriever.invoke("什么是 RAG？")
# 返回的是 1000-token 的父分块，但匹配的是 200-token 的子分块

06 检索策略：从向量搜索到混合检索

1. 稠密检索 (Dense Retrieval)

最基础的方式：将 query 和文档都编码为向量，用余弦相似度找最近的 K 个文档。

2. 稀疏检索 (Sparse Retrieval / BM25)

传统关键词检索：TF-IDF 的进化版，基于词频和文档频率打分。

from rank_bm25 import BM25Okapi
import jieba  # 中文分词

# 中文分词
tokenized_corpus = [list(jieba.cut(doc)) for doc in documents]
bm25 = BM25Okapi(tokenized_corpus)

# 查询
query_tokens = list(jieba.cut("RAG 架构设计"))
scores = bm25.get_scores(query_tokens)
top_k_indices = scores.argsort()[-5:][::-1]

3. 混合检索 (Hybrid Retrieval) 推荐

结合稠密检索和稀疏检索，取长补短——当前业界最佳实践。

import numpy as np
from rank_bm25 import BM25Okapi

def hybrid_search(
    query: str,
    query_embedding: np.ndarray,    # (dim,)
    doc_embeddings: np.ndarray,      # (n, dim)
    doc_texts: list[str],
    top_k: int = 10,
    dense_weight: float = 0.7,
    sparse_weight: float = 0.3,
) -> list[tuple[int, float]]:
    """混合检索：Dense + Sparse 分数融合"""

    # 1. Dense 检索（余弦相似度）
    dense_scores = np.dot(doc_embeddings, query_embedding) / (
        np.linalg.norm(doc_embeddings, axis=1) * np.linalg.norm(query_embedding)
    )

    # 2. Sparse 检索（BM25）
    tokenized = [doc.split() for doc in doc_texts]
    bm25 = BM25Okapi(tokenized)
    sparse_scores = np.array(bm25.get_scores(query.split()))

    # 3. Min-Max 归一化到 [0, 1]
    def normalize(scores):
        s_min, s_max = scores.min(), scores.max()
        return (scores - s_min) / (s_max - s_min + 1e-8)

    dense_norm = normalize(dense_scores)
    sparse_norm = normalize(sparse_scores)

    # 4. 加权融合
    combined = dense_weight * dense_norm + sparse_weight * sparse_norm

    # 5. 取 Top-K
    top_indices = combined.argsort()[-top_k:][::-1]
    return [(idx, combined[idx]) for idx in top_indices]

4. RRF (Reciprocal Rank Fusion)

比加权平均更稳健的融合策略，基于排名而非原始分数。

def reciprocal_rank_fusion(
    ranked_lists: list[list[int]],  # 多个排序列表
    k: int = 60,                    # RRF 常数，通常 60
) -> list[tuple[int, float]]:
    """RRF 融合：基于排名的融合，对分数尺度不敏感"""
    scores = {}
    for ranked_list in ranked_lists:
        for rank, doc_id in enumerate(ranked_list, start=1):
            scores[doc_id] = scores.get(doc_id, 0) + 1.0 / (k + rank)

    # 按融合分数排序
    sorted_results = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_results

# 用法：将 Dense 和 Sparse 的 Top-K 排名列表传入
dense_ranking = [3, 7, 1, 9, 5]    # Dense 检索的文档 ID 排名
sparse_ranking = [1, 5, 3, 2, 8]   # BM25 检索的文档 ID 排名
fused = reciprocal_rank_fusion([dense_ranking, sparse_ranking])

5. Query 改写与增强

将以下用户问题改写为
更适合向量检索的查询：
原问题：{query}
要求：保留核心意图，去除口语化表达，
添加同义关键词

from langchain.retrievers.multi_query import MultiQueryRetriever
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_chroma import Chroma

llm = ChatOpenAI(temperature=0)
vectorstore = Chroma(embedding_function=OpenAIEmbeddings())

# 自动生成多个查询角度
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    llm=llm,
)

# 一行代码，背后生成 3-5 个改写 query
docs = retriever.invoke("RAG 系统如何优化检索效果？")

07 Reranking：精确排序的关键

初始检索（Recall 阶段）求全不求精——宁可多返回一些相关文档。Reranking（Precision 阶段）对候选文档精确排序，大幅提升最终质量。

主流 Reranker 对比

from FlagEmbedding import FlagReranker

reranker = FlagReranker('BAAI/bge-reranker-v2-m3', use_fp16=True)

# 对 (query, doc) 对打分
pairs = [
    ["什么是 RAG？", "RAG 是检索增强生成，结合外部知识检索与 LLM 生成"],
    ["什么是 RAG？", "微调是通过训练数据调整模型参数的方法"],
    ["什么是 RAG？", "RAG 系统的 chunking 策略对效果影响很大"],
]

scores = reranker.compute_score(pairs)
# 输出: [0.92, 0.15, 0.78]  → 第 1 个最相关

# 实际使用：对检索结果重排
def rerank_results(query: str, docs: list[str], top_k: int = 5) -> list[int]:
    pairs = [[query, doc] for doc in docs]
    scores = reranker.compute_score(pairs)
    ranked = sorted(range(len(docs)), key=lambda i: scores[i], reverse=True)
    return ranked[:top_k]

import cohere

co = cohere.Client("your-api-key")

results = co.rerank(
    model="rerank-v3.5",
    query="RAG 架构最佳实践",
    documents=[
        "RAG 是检索增强生成...",
        "微调需要大量标注数据...",
        "混合检索结合语义和关键词匹配...",
    ],
    top_n=3,
)

for r in results.results:
    print(f"Rank {r.index}: score={r.relevance_score:.3f}")

08 生成阶段：上下文组装与幻觉控制

上下文组装策略

Prompt 模板设计

RAG_PROMPT = """你是一个知识助手。根据以下检索到的文档片段回答用户问题。

## 规则
1. 仅根据提供的文档内容回答，不要使用外部知识
2. 如果文档中没有相关信息，明确说"根据现有文档无法回答"
3. 每个论点必须引用来源，格式：[来源编号]
4. 不要编造文档中没有的信息

## 检索到的文档
{context}

## 用户问题
{query}

## 回答（引用来源，如不确定请说明）"""

def build_context(documents: list[dict]) -> str:
    """组装上下文，标注来源"""
    parts = []
    for i, doc in enumerate(documents, 1):
        source = doc.get('source', '未知来源')
        parts.append(f"[{i}] 来源: {source}\n{doc['text']}")
    return "\n\n".join(parts)

幻觉检测与控制

09 进阶架构：Agentic RAG / Graph RAG / Modular RAG

当朴素 RAG 不够用时，需要更智能的架构。2024-2025 年最重要的三个进化方向：

Agentic RAG：让 Agent 自主决定如何检索

核心思想：检索不再是固定 pipeline，而是 Agent 的一个工具。Agent 可以决定：

• 是否需要检索 — 简单问题直接回答，复杂问题再检索
• 检索什么 — 选择合适的知识库/检索策略
• 检索够不够 — 评估检索结果质量，决定是否需要更多检索
• 如何组合 — 从不同源检索后综合分析

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# 定义多种检索工具
@tool
def search_internal_docs(query: str) -> str:
    """搜索企业内部文档库"""
    # 向量检索 + BM25 混合检索
    results = hybrid_retriever.invoke(query)
    return format_results(results)

@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    # 调用搜索 API
    return web_search_api(query)

@tool
def search_codebase(query: str) -> str:
    """搜索代码仓库"""
    # 代码专用检索
    return code_search(query)

@tool
def evaluate_sufficiency(context: str, query: str) -> str:
    """评估当前上下文是否足够回答问题"""
    prompt = f"给定上下文是否足以回答问题？回答 YES 或 NO。\n上下文: {context}\n问题: {query}"
    return ChatOpenAI(temperature=0).invoke(prompt).content

# Agent 自主决定使用哪些工具、是否需要更多检索
agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4o"),
    tools=[search_internal_docs, search_web, search_codebase, evaluate_sufficiency],
)

# Agent 可能的执行路径：
# 1. 搜索内部文档 → 评估不足 → 搜索网页 → 生成回答
# 2. 搜索代码库 → 评估足够 → 直接生成
# 3. 不检索 → 直接回答（简单问题）
result = agent.invoke({"messages": [("user", "我们公司最新发布的 API 有哪些变更？")]})

Graph RAG：用知识图谱补充关联关系

向量检索的局限：只能找到语义相似的文档，无法理解实体间的关联关系。Graph RAG 通过知识图谱解决这一问题。

# Microsoft GraphRAG 流程（简化）
# pip install graphrag

"""
1. 索引阶段（离线）:
   - 文档分块
   - LLM 提取实体和关系 → 知识图谱
   - Leiden 社区检测 → 层级社区
   - 为每个社区生成摘要

2. 查询阶段（在线）:
   - Local Search: 找相关实体 → 图遍历 → 拼接邻域
   - Global Search: 找相关社区 → 汇总社区摘要
   - Drift Search: 结合 local + global
"""

# 配置文件 settings.yaml
entity_extraction:
  prompt: "提取文本中的实体和关系..."
  max_gleanings: 1  # 补充提取轮次

community_reports:
  prompt: "为以下实体群组生成社区摘要..."
  max_length: 2000
  max_levels: 3  # 社区层级深度

# 运行索引
# graphrag index --root ./workspace

# 查询
# graphrag query --root ./workspace --method local "公司的核心产品有哪些？"
# graphrag query --root ./workspace --method global "整体技术趋势是什么？"

三种 RAG 架构对比

10 RAG 评估：如何知道你的 RAG 行不行

RAGAS 评估框架

RAGAS (Retrieval Augmented Generation Assessment) 是最广泛使用的 RAG 评估框架，提供 7+ 核心指标：

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

# 准备评估数据（需要人工标注的问答对）
eval_data = {
    "question": ["什么是 RAG？", "RAG 的优势是什么？"],
    "answer": ["RAG 是检索增强生成...", "RAG 的优势包括实时性、准确性..."],
    "contexts": [
        ["RAG 是检索增强生成，结合检索和生成..."],
        ["RAG 的优势：实时知识更新、减少幻觉..."],
    ],
    "ground_truth": ["RAG 是将外部检索与LLM生成结合的架构", "实时更新、减少幻觉、可追溯来源"],
}

dataset = Dataset.from_dict(eval_data)

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

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

评估最佳实践

11 生产级 RAG：索引管理、增量更新、成本优化

索引更新策略

import hashlib
from datetime import datetime

class IncrementalIndexer:
    """增量索引器：只处理变更的文档"""
    
    def __init__(self, vector_store, doc_registry):
        self.vector_store = vector_store
        self.registry = doc_registry  # 文档元数据注册表
    
    def compute_hash(self, content: str) -> str:
        return hashlib.sha256(content.encode()).hexdigest()
    
    async def sync_documents(self, documents: list[dict]):
        """同步文档：新增、更新、删除"""
        current_hashes = {doc['id']: self.compute_hash(doc['content']) for doc in documents}
        stored_hashes = self.registry.get_all_hashes()
        
        to_add = []      # 新增
        to_update = []   # 内容变更
        to_delete = []   # 已删除
        
        for doc_id, content_hash in current_hashes.items():
            if doc_id not in stored_hashes:
                to_add.append(doc_id)
            elif stored_hashes[doc_id] != content_hash:
                to_update.append(doc_id)
        
        for doc_id in stored_hashes:
            if doc_id not in current_hashes:
                to_delete.append(doc_id)
        
        # 执行变更
        for doc_id in to_add:
            doc = next(d for d in documents if d['id'] == doc_id)
            await self._index_document(doc)
        
        for doc_id in to_update:
            await self._remove_from_index(doc_id)
            doc = next(d for d in documents if d['id'] == doc_id)
            await self._index_document(doc)
        
        for doc_id in to_delete:
            await self._remove_from_index(doc_id)
        
        # 更新注册表
        self.registry.update_hashes(current_hashes)
        
        return {"added": len(to_add), "updated": len(to_update), "deleted": len(to_delete)}

成本优化

import hashlib
import json
from datetime import timedelta

class QueryCache:
    """缓存热门 Query 的检索结果，避免重复检索"""
    
    def __init__(self, redis_client, ttl=timedelta(hours=24)):
        self.redis = redis_client
        self.ttl = ttl
    
    def _cache_key(self, query: str, top_k: int) -> str:
        # 对 query 做语义归一化后哈希
        normalized = query.strip().lower()
        content_hash = hashlib.md5(normalized.encode()).hexdigest()
        return f"rag:cache:{content_hash}:{top_k}"
    
    async def get(self, query: str, top_k: int) -> list[dict] | None:
        key = self._cache_key(query, top_k)
        cached = await self.redis.get(key)
        if cached:
            return json.loads(cached)
        return None
    
    async def set(self, query: str, top_k: int, results: list[dict]):
        key = self._cache_key(query, top_k)
        await self.redis.set(key, json.dumps(results), ex=self.ttl)

12 方案对比与选型决策树

向量数据库对比

RAG 框架对比

选型决策树

13 完整代码示例

生产级 RAG 系统（FastAPI + LangChain + Qdrant）

"""
生产级 RAG 服务：FastAPI + LangChain + Qdrant + BGE-M3 + Reranker
功能：文档上传、索引管理、混合检索、Reranking、流式生成
"""

from fastapi import FastAPI, UploadFile, Query
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_qdrant import Qdrant
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever
from FlagEmbedding import FlagReranker
from qdrant_client import QdrantClient
import asyncio

app = FastAPI(title="RAG Service")

# ===== 初始化 =====
COLLECTION = "knowledge_base"

qdrant_client = QdrantClient(host="localhost", port=6333)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small", dimensions=512)
reranker = FlagReranker("BAAI/bge-reranker-v2-m3", use_fp16=True)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, streaming=True)

vector_store = Qdrant(
    client=qdrant_client,
    collection_name=COLLECTION,
    embeddings=embeddings,
)

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

# ===== 文档上传与索引 =====
class IndexRequest(BaseModel):
    content: str
    source: str
    metadata: dict = {}

@app.post("/index")
async def index_document(req: IndexRequest):
    """索引单个文档"""
    chunks = splitter.split_text(req.content)
    
    metadatas = [
        {"source": req.source, "chunk_index": i, **req.metadata}
        for i in range(len(chunks))
    ]
    
    await asyncio.to_thread(
        vector_store.add_texts,
        texts=chunks,
        metadatas=metadatas,
    )
    
    return {"status": "indexed", "chunks": len(chunks)}

# ===== 混合检索 + Reranking =====
async def hybrid_retrieve(query: str, top_k: int = 20) -> list[dict]:
    """混合检索：向量 + BM25 + Reranking"""
    
    # 1. 向量检索
    vector_retriever = vector_store.as_retriever(
        search_kwargs={"k": top_k}
    )
    
    # 2. BM25 检索
    all_docs = vector_store.similarity_search("", k=10000)  # 获取全部文档
    bm25_retriever = BM25Retriever.from_documents(all_docs)
    bm25_retriever.k = top_k
    
    # 3. 混合检索
    ensemble_retriever = EnsembleRetriever(
        retrievers=[vector_retriever, bm25_retriever],
        weights=[0.7, 0.3],
    )
    candidates = ensemble_retriever.invoke(query)
    
    # 4. Reranking
    pairs = [[query, doc.page_content] for doc in candidates]
    scores = await asyncio.to_thread(reranker.compute_score, pairs)
    
    # 5. 按重排分数排序
    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True,
    )
    
    return [
        {
            "content": doc.page_content,
            "score": score,
            "metadata": doc.metadata,
        }
        for doc, score in ranked[:5]  # 最终取 Top-5
    ]

# ===== 生成回答 =====
RAG_PROMPT = """基于以下检索到的文档回答问题。

文档：
{context}

问题：{query}

要求：
1. 仅基于文档内容回答
2. 引用来源 [编号]
3. 不确定时说明"""

@app.get("/query")
async def query_rag(q: str = Query(...)):
    """RAG 问答"""
    # 检索
    results = await hybrid_retrieve(q)
    
    # 组装上下文
    context_parts = [
        f"[{i+1}] 来源: {r['metadata'].get('source', '未知')}\n{r['content']}"
        for i, r in enumerate(results)
    ]
    context = "\n\n".join(context_parts)
    
    # 流式生成
    prompt = RAG_PROMPT.format(context=context, query=q)
    
    async def stream():
        async for chunk in llm.astream(prompt):
            yield chunk.content
    
    return StreamingResponse(stream(), media_type="text/plain")

@app.get("/health")
async def health():
    return {"status": "ok"}

14 替代方案

当 RAG 不是最佳选择时

15 链接回 Agent 研究

RAG 是 Agent 记忆系统的核心组件。以下是与 Agent 研究的关联：