自然语言处理（十）—— RAG 与知识增强系统

大语言模型虽然强大，但有一个致命弱点：它们的知识被"冻结"在训练数据中。当用户询问最新事件、私有文档或领域专业知识时，模型往往给出过时或错误的答案。更糟糕的是，模型还会"编造"看似合理但实际不存在的信息——这就是所谓的幻觉问题。

检索增强生成（ RAG）技术通过一个简单而有效的思路解决了这个问题：在生成答案之前，先从外部知识库中检索相关信息，然后将检索到的文档与用户查询一起输入生成模型。这样，模型就能基于真实的外部知识生成答案，而不是仅依赖训练时的记忆。

但构建一个高效的 RAG 系统并不简单。向量数据库的选择决定了检索速度和可扩展性； Embedding 模型的质量直接影响检索精度；检索策略（密集检索、稀疏检索、混合检索）需要根据数据特点精心设计；重排序技术能进一步提升结果质量；查询重写和扩展则能显著改善检索效果。本文深入解析 RAG 系统的每个组件，从原理到实现，从优化到部署，帮助读者构建生产级的 RAG 应用。

RAG 基础与架构

RAG 核心思想

RAG 的核心思想可以用一个简单的公式概括：生成 = 检索 + 增强生成。具体来说，系统首先从大规模知识库中检索与查询相关的文档片段，然后将这些文档作为"上下文"输入生成模型，让模型基于真实的外部知识生成答案。

数学表示：

RAG 将生成过程分解为两个步骤：检索和生成。对于查询， RAG 系统的输出为：

其中：

- 是检索到的文档集合（通常取 top-k，如 top-5） - 是检索概率，表示文档与查询的相关性，通常用向量相似度（如余弦相似度）计算 - 是生成概率，表示基于检索文档和查询生成答案的概率

为什么有效：

RAG 的优势在于它将"记忆"和"推理"分离：知识存储在向量数据库中，可以随时更新；推理能力由生成模型提供，两者结合既保证了知识的时效性，又避免了模型需要重新训练。更重要的是， RAG 提供了可解释性：每个答案都能追溯到具体的源文档，这对于生产应用至关重要。

RAG 架构流程

典型的 RAG 系统包含以下步骤：

文档处理：将原始文档切分、向量化并存储到向量数据库
查询处理：将用户查询转换为向量表示
检索：在向量数据库中检索相关文档
重排序：对检索结果进行精排
生成：将检索到的文档和查询输入生成模型

from langchain.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain.llms import OpenAI
from langchain.chains import RetrievalQA

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

# 2. 文档切分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = text_splitter.split_documents(documents)

# 3. 创建向量数据库
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)
vectorstore = FAISS.from_documents(chunks, embeddings)

# 4. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 5. 创建 RAG 链
llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 6. 查询
query = "什么是机器学习？"
result = qa_chain({"query": query})
print(result["result"])
print(result["source_documents"])

RAG vs 微调

方法	优势	劣势	适用场景
RAG	可动态更新知识、成本低、可解释性强	依赖检索质量、可能有幻觉	知识库频繁更新、多领域知识
微调	模型完全适应任务、性能可能更好	成本高、难以更新、可能遗忘	特定任务、知识相对稳定

向量数据库选择

向量数据库是 RAG 系统的核心组件，负责存储和检索文档向量。不同的向量数据库有不同的特点和适用场景。

FAISS（ Facebook AI Similarity Search）

FAISS 是 Facebook 开源的向量相似度搜索库，支持 CPU 和 GPU 加速。

特点： - 高性能：支持多种索引算法（ IVF 、 HNSW 、 LSH） - 内存高效：支持内存映射和量化 - 易于集成： Python API 简单易用

适用场景： - 中小规模数据集（百万级向量） - 需要快速原型开发 - 本地部署

import faiss
import numpy as np

# 创建索引
dimension = 768  # 向量维度
index = faiss.IndexFlatL2(dimension)  # L2 距离

# 添加向量
vectors = np.random.random((10000, dimension)).astype('float32')
index.add(vectors)

# 搜索
query_vector = np.random.random((1, dimension)).astype('float32')
k = 5  # 返回 top-k
distances, indices = index.search(query_vector, k)

print(f"Top {k} similar vectors:")
for i, idx in enumerate(indices[0]):
    print(f"Rank {i+1}: Index {idx}, Distance {distances[0][i]}")

# 使用 IVF 索引（更快，但需要训练）
nlist = 100  # 聚类中心数
quantizer = faiss.IndexFlatL2(dimension)
index_ivf = faiss.IndexIVFFlat(quantizer, dimension, nlist)
index_ivf.train(vectors)  # 训练索引
index_ivf.add(vectors)
index_ivf.nprobe = 10  # 搜索时检查的聚类数

Milvus

Milvus 是云原生的向量数据库，支持分布式部署和水平扩展。

特点： - 分布式架构：支持集群部署 - 高可用性：支持数据复制和故障恢复 - 丰富的功能：支持标量过滤、时间序列等

适用场景： - 大规模数据集（千万级以上） - 生产环境部署 - 需要高可用性和可扩展性

from pymilvus import connections, Collection, FieldSchema, CollectionSchema, DataType

# 连接 Milvus
connections.connect(
    alias="default",
    host="localhost",
    port="19530"
)

# 定义 schema
fields = [
    FieldSchema(name="id", dtype=DataType.INT64, is_primary=True, auto_id=True),
    FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=1000),
    FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=768)
]

schema = CollectionSchema(fields, "RAG collection")
collection = Collection("rag_collection", schema)

# 创建索引
index_params = {
    "metric_type": "L2",
    "index_type": "IVF_FLAT",
    "params": {"nlist": 1024}
}
collection.create_index("embedding", index_params)

# 插入数据
data = [
    ["Document 1", "Document 2", "Document 3"],
    [[0.1] * 768, [0.2] * 768, [0.3] * 768]  # 示例向量
]
collection.insert(data)
collection.load()

# 搜索
search_params = {"metric_type": "L2", "params": {"nprobe": 10 }}
results = collection.search(
    data=[[0.15] * 768],  # 查询向量
    anns_field="embedding",
    param=search_params,
    limit=5
)

Pinecone

Pinecone 是托管的向量数据库服务，无需管理基础设施。

特点： - 完全托管：无需管理服务器 - 自动扩展：根据负载自动调整 - 简单易用： RESTful API

适用场景： - 快速上线 - 中小规模应用 - 不想管理基础设施

import pinecone

# 初始化
pinecone.init(api_key="your-api-key", environment="us-west1-gcp")

# 创建索引
index_name = "rag-index"
if index_name not in pinecone.list_indexes():
    pinecone.create_index(
        index_name,
        dimension=768,
        metric="cosine"
    )

# 连接索引
index = pinecone.Index(index_name)

# 插入向量
vectors = [
    ("vec1", [0.1] * 768, {"text": "Document 1"}),
    ("vec2", [0.2] * 768, {"text": "Document 2"})
]
index.upsert(vectors=vectors)

# 搜索
query_vector = [0.15] * 768
results = index.query(
    vector=query_vector,
    top_k=5,
    include_metadata=True
)

Chroma

Chroma 是轻量级的向量数据库，专注于易用性和开发体验。

特点： - 轻量级：资源占用小 - 易用性： API 设计简洁 - 灵活性：支持多种部署方式

适用场景： - 开发测试 - 小规模应用 - 快速原型

import chromadb
from chromadb.config import Settings

# 创建客户端
client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

# 创建集合
collection = client.create_collection(
    name="rag_collection",
    metadata={"hnsw:space": "cosine"}
)

# 添加文档
collection.add(
    documents=["Document 1", "Document 2", "Document 3"],
    embeddings=[[0.1] * 768, [0.2] * 768, [0.3] * 768],
    ids=["id1", "id2", "id3"],
    metadatas=[{"source": "doc1"}, {"source": "doc2"}, {"source": "doc3"}]
)

# 查询
results = collection.query(
    query_embeddings=[[0.15] * 768],
    n_results=5
)

向量数据库对比

数据库	规模	部署方式	特点	适用场景
FAISS	百万级	本地	高性能、易用	开发、中小规模
Milvus	千万级+	分布式	可扩展、高可用	生产、大规模
Pinecone	百万级	托管	简单、无需运维	快速上线
Chroma	十万级	本地/云	轻量、易用	开发、小规模

Embedding 模型对比

Embedding 模型的质量直接影响检索效果。不同的模型有不同的特点和适用场景。

通用 Embedding 模型

OpenAI text-embedding-ada-002： - 维度： 1536 - 优势：性能优秀、多语言支持 - 劣势：需要 API 调用、有成本

sentence-transformers： - 开源模型集合 - 优势：免费、可本地部署、性能好 - 常用模型： - all-MiniLM-L6-v2：快速、轻量 - all-mpnet-base-v2：性能更好 - multi-qa-mpnet-base：针对问答优化

from sentence_transformers import SentenceTransformer
import numpy as np

# 加载模型
model = SentenceTransformer('all-mpnet-base-v2')

# 编码文本
texts = [
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络",
    "自然语言处理处理文本数据"
]
embeddings = model.encode(texts)

print(f"Embedding shape: {embeddings.shape}")
print(f"Embedding dimension: {embeddings.shape[1]}")

# 计算相似度
from sklearn.metrics.pairwise import cosine_similarity
similarity_matrix = cosine_similarity(embeddings)
print("Similarity matrix:")
print(similarity_matrix)

领域特定 Embedding

对于特定领域，可以使用领域数据微调 Embedding 模型。

from sentence_transformers import SentenceTransformer, InputExample, losses
from torch.utils.data import DataLoader

# 加载基础模型
model = SentenceTransformer('all-mpnet-base-v2')

# 准备训练数据
train_examples = [
    InputExample(texts=["机器学习", "ML"]),
    InputExample(texts=["深度学习", "DL"]),
    InputExample(texts=["自然语言处理", "NLP"])
]

# 定义损失函数（对比学习）
train_dataloader = DataLoader(train_examples, shuffle=True, batch_size=16)
train_loss = losses.CosineSimilarityLoss(model)

# 微调
model.fit(
    train_objectives=[(train_dataloader, train_loss)],
    epochs=10,
    warmup_steps=100
)

# 保存模型
model.save('./domain-specific-embedding')

Embedding 模型选择指南

模型类型	维度	速度	精度	适用场景
text-embedding-ada-002	1536	中	高	生产环境、多语言
all-MiniLM-L6-v2	384	快	中	快速原型、资源受限
all-mpnet-base-v2	768	中	高	平衡性能和速度
multi-qa-mpnet-base	768	中	高	问答任务

检索策略优化

检索是 RAG 系统的核心环节，检索质量直接影响最终答案的准确性。不同的检索策略有不同的优势和适用场景，理解它们的差异对于构建高效的 RAG 系统至关重要。

Dense Retrieval（密集检索）

Dense Retrieval 通过 Embedding 模型将查询和文档都转换为高维向量，然后计算向量相似度（通常是余弦相似度）来检索相关文档。这是目前最主流的检索方法。

工作原理：

密集检索的核心假设是：语义相似的文本在向量空间中应该距离较近。通过在大规模文本对上训练的 Embedding 模型（如 sentence-transformers），系统能够将语义信息编码到向量表示中。检索时，计算查询向量与所有文档向量的相似度，选择相似度最高的 top-k 个文档。

优势： - 语义理解能力强：能够理解同义词、近义词和语义相似的概念 - 跨语言能力：多语言 Embedding 模型支持跨语言检索 - 实现简单：只需向量相似度计算，无需复杂的特征工程

劣势： - 对精确关键词匹配不敏感：如果查询包含特定术语（如产品名称、代码标识符），可能检索不到精确匹配的文档 - 计算成本：需要为所有文档计算 Embedding，大规模场景下成本较高 - 领域适应性：通用 Embedding 模型在特定领域可能表现不佳

def dense_retrieval(query_embedding, vectorstore, top_k=5):
    """密集检索"""
    # 计算相似度
    similarities = vectorstore.similarity_search_with_score(
        query_embedding,
        k=top_k
    )
    
    return similarities

Sparse Retrieval（稀疏检索）

Sparse Retrieval 使用关键词匹配（如 BM25）进行检索。

优势： - 关键词匹配精确 - 对精确术语敏感 - 不需要 Embedding 模型

劣势： - 无法理解语义 - 对同义词不敏感

from rank_bm25 import BM25Okapi
import jieba

def sparse_retrieval(query, documents, top_k=5):
    """稀疏检索（ BM25）"""
    # 分词
    tokenized_docs = [jieba.lcut(doc) for doc in documents]
    tokenized_query = jieba.lcut(query)
    
    # 创建 BM25 索引
    bm25 = BM25Okapi(tokenized_docs)
    
    # 检索
    scores = bm25.get_scores(tokenized_query)
    top_indices = np.argsort(scores)[::-1][:top_k]
    
    return [(documents[i], scores[i]) for i in top_indices]

Hybrid Retrieval（混合检索）

Dense 和 Sparse 检索各有优势，混合检索通过结合两者来获得最佳效果。实际应用中，混合检索通常能比单一方法提升 10-30% 的检索精度。

为什么需要混合检索：

Dense 检索擅长语义理解，但可能遗漏精确匹配
Sparse 检索擅长关键词匹配，但无法理解语义
混合检索结合两者优势，既保证语义相关性，又确保精确匹配不被遗漏

融合策略：

RRF（ Reciprocal Rank Fusion）：最常用的融合方法，对两种检索结果的排名进行加权合并。 RRF 分数计算公式为：

其中是所有检索方法的结果集合，是文档在方法中的排名，是平滑参数（通常为 60）。

加权融合：对两种检索结果的相似度分数进行加权求和，需要根据数据特点调整权重（如 dense:sparse = 0.7:0.3）。
重排序融合：先用两种方法各检索 top-k 个候选（如各检索 20 个），合并去重后得到候选集，再用 Cross-Encoder 重排序模型对所有候选进行精排，选择最终的 top-k 。

def hybrid_retrieval(query, query_embedding, vectorstore, documents, top_k=5):
    """混合检索"""
    # Dense 检索
    dense_results = dense_retrieval(query_embedding, vectorstore, top_k=top_k*2)
    dense_scores = {doc: score for doc, score in dense_results}
    
    # Sparse 检索
    sparse_results = sparse_retrieval(query, documents, top_k=top_k*2)
    sparse_scores = {doc: score for doc, score in sparse_results}
    
    # RRF 融合
    def rrf_score(doc, rank, k=60):
        return 1 / (k + rank)
    
    # 计算 RRF 分数
    combined_scores = {}
    for rank, (doc, _) in enumerate(dense_results, 1):
        combined_scores[doc] = combined_scores.get(doc, 0) + rrf_score(doc, rank)
    
    for rank, (doc, _) in enumerate(sparse_results, 1):
        combined_scores[doc] = combined_scores.get(doc, 0) + rrf_score(doc, rank)
    
    # 排序并返回 top-k
    sorted_docs = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_docs[:top_k]

Reranking 技术

Reranking 对初始检索结果进行精排，提高最终结果的质量。

Cross-Encoder Reranking

Cross-Encoder 将查询和文档一起输入模型，计算相关性分数。

优势： - 精度高 - 能够理解查询-文档交互

劣势： - 计算成本高（不能预先计算） - 速度慢

from sentence_transformers import CrossEncoder

# 加载 Cross-Encoder 模型
reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

def rerank(query, documents, top_k=5):
    """重排序"""
    # 构建查询-文档对
    pairs = [[query, doc] for doc in documents]
    
    # 计算相关性分数
    scores = reranker.predict(pairs)
    
    # 排序
    ranked_indices = np.argsort(scores)[::-1]
    ranked_docs = [documents[i] for i in ranked_indices[:top_k]]
    
    return ranked_docs

多阶段检索

多阶段检索结合快速检索和精确重排序：

第一阶段：使用快速方法（ Dense/Sparse）检索大量候选（如 100 个）
第二阶段：使用重排序模型对候选进行精排（如 top-5）

def multi_stage_retrieval(query, query_embedding, vectorstore, top_k=5):
    """多阶段检索"""
    # 第一阶段：快速检索大量候选
    candidates = dense_retrieval(query_embedding, vectorstore, top_k=100)
    candidate_docs = [doc for doc, _ in candidates]
    
    # 第二阶段：重排序
    reranked_docs = rerank(query, candidate_docs, top_k=top_k)
    
    return reranked_docs

Query 重写与扩展

查询优化可以提高检索效果，包括查询重写、查询扩展和查询分解。

Query Rewriting（查询重写）

查询重写将用户查询转换为更适合检索的形式。

方法： 1. 同义词扩展：添加同义词 2. 查询补全：补全不完整的查询 3. 查询简化：去除冗余词

from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

def rewrite_query(original_query, llm):
    """查询重写"""
    prompt = PromptTemplate(
        input_variables=["query"],
        template="""
        将以下查询重写为更适合信息检索的形式，保持原意但使用更精确的术语：
        
        原始查询：{query}
        
        重写后的查询：
        """
    )
    
    rewritten = llm(prompt.format(query=original_query))
    return rewritten.strip()

Query Expansion（查询扩展）

查询扩展添加相关术语和概念。

def expand_query(query, llm):
    """查询扩展"""
    prompt = f"""
    为以下查询生成相关的关键词和同义词，用于信息检索：
    
    查询：{query}
    
    相关关键词（用逗号分隔）：
    """
    
    expanded_terms = llm(prompt).strip().split(',')
    expanded_query = query + " " + " ".join(expanded_terms)
    
    return expanded_query

Query Decomposition（查询分解）

对于复杂查询，可以分解为多个子查询。

def decompose_query(query, llm):
    """查询分解"""
    prompt = f"""
    将以下复杂查询分解为多个简单的子查询：
    
    复杂查询：{query}
    
    子查询（每行一个）：
    """
    
    subqueries = llm(prompt).strip().split('\n')
    return [q.strip() for q in subqueries if q.strip()]

实战：构建企业级 RAG 系统

使用 LangChain 构建 RAG

LangChain 提供了完整的 RAG 工具链。

from langchain.document_loaders import DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
from langchain.prompts import PromptTemplate

# 1. 加载文档
loader = DirectoryLoader("./documents", glob="*.txt")
documents = loader.load()

# 2. 文档切分
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)
chunks = text_splitter.split_documents(documents)

# 3. 创建 Embedding
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-mpnet-base-v2"
)

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

# 5. 自定义 Prompt
prompt_template = """使用以下上下文回答问题。如果你不知道答案，就说不知道，不要编造答案。

上下文：{context}

问题：{question}

答案："""

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

# 6. 创建 RAG 链
llm = OpenAI(temperature=0)
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True
)

# 7. 查询
query = "什么是 RAG？"
result = qa_chain({"query": query})
print(f"答案：{result['result']}")
print(f"来源文档：{result['source_documents']}")

使用 LlamaIndex 构建 RAG

LlamaIndex 专注于 LLM 应用的数据层。

from llama_index import VectorStoreIndex, SimpleDirectoryReader, ServiceContext
from llama_index.embeddings import HuggingFaceEmbedding
from llama_index.llms import OpenAI

# 1. 加载文档
documents = SimpleDirectoryReader("./documents").load_data()

# 2. 创建服务上下文
embed_model = HuggingFaceEmbedding(
    model_name="sentence-transformers/all-mpnet-base-v2"
)
llm = OpenAI(temperature=0)

service_context = ServiceContext.from_defaults(
    llm=llm,
    embed_model=embed_model,
    chunk_size=1000,
    chunk_overlap=200
)

# 3. 创建索引
index = VectorStoreIndex.from_documents(
    documents,
    service_context=service_context
)

# 4. 创建查询引擎
query_engine = index.as_query_engine(
    similarity_top_k=5,
    response_mode="compact"
)

# 5. 查询
response = query_engine.query("什么是 RAG？")
print(response)
print(response.source_nodes)

高级 RAG 模式

Parent-Child Retrieval： - 存储时：将文档切分为小块（子块） - 检索时：检索子块，但返回父块（包含更多上下文）

from langchain.text_splitter import RecursiveCharacterTextSplitter

# 创建父子文档分割器
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=500)

# 分割文档
parent_docs = parent_splitter.split_documents(documents)
child_docs = []

for parent in parent_docs:
    children = child_splitter.split_documents([parent])
    for child in children:
        child.metadata["parent_id"] = parent.metadata.get("id")
    child_docs.extend(children)

# 存储子块（用于检索）
vectorstore = Chroma.from_documents(child_docs, embeddings)

# 检索时返回父块
def retrieve_with_parent(query, vectorstore):
    # 检索子块
    child_results = vectorstore.similarity_search(query, k=5)
    
    # 获取父块
    parent_ids = set([r.metadata["parent_id"] for r in child_results])
    parent_docs = [doc for doc in parent_docs if doc.metadata.get("id") in parent_ids]
    
    return parent_docs

Self-RAG： - 使用 LLM 判断是否需要检索 - 对检索结果进行批判性评估 - 根据评估结果决定是否使用检索信息

def self_rag(query, llm, vectorstore):
    """Self-RAG 实现"""
    # 1. 判断是否需要检索
    need_retrieval_prompt = f"""
    判断以下查询是否需要检索外部知识库：
    
    查询：{query}
    
    如果需要检索，回答"是"，否则回答"否"。
    """
    
    need_retrieval = llm(need_retrieval_prompt).strip()
    
    if "是" in need_retrieval or "yes" in need_retrieval.lower():
        # 2. 检索
        docs = vectorstore.similarity_search(query, k=5)
        
        # 3. 评估检索结果
        evaluation_prompt = f"""
        评估以下检索结果与查询的相关性：
        
        查询：{query}
        
        检索结果：
        {chr(10).join([f"{i+1}. {doc.page_content[:200]}" for i, doc in enumerate(docs)])}
        
        对每个结果给出相关性评分（ 1-5），并说明原因。
        """
        
        evaluation = llm(evaluation_prompt)
        
        # 4. 生成答案
        answer_prompt = f"""
        基于以下检索结果回答问题：
        
        查询：{query}
        
        检索结果：
        {chr(10).join([doc.page_content for doc in docs])}
        
        相关性评估：
        {evaluation}
        
        答案：
        """
        
        answer = llm(answer_prompt)
        return answer, docs
    else:
        # 直接生成（不需要检索）
        answer = llm(f"回答以下问题：{query}")
        return answer, []

❓ Q&A: RAG 常见问题

Q1: RAG 和微调有什么区别？什么时候用 RAG？

A: - RAG：动态检索外部知识，适合知识库频繁更新、需要访问最新信息、多领域知识的场景 - 微调：将知识编码到模型参数中，适合特定任务、知识相对稳定、需要极致性能的场景 - 选择：如果知识需要频繁更新或涉及私有数据，选择 RAG；如果任务固定且性能要求高，考虑微调

Q2: 如何选择向量数据库？

A: 选择取决于： - 数据规模：百万级用 FAISS，千万级+用 Milvus - 部署方式：本地用 FAISS/Chroma，云部署用 Milvus/Pinecone - 运维能力：不想管理基础设施用 Pinecone，有运维团队用 Milvus - 开发阶段：快速原型用 FAISS/Chroma，生产环境用 Milvus

Q3: Embedding 模型如何选择？

A: - 通用场景：all-mpnet-base-v2 或 text-embedding-ada-002 - 资源受限：all-MiniLM-L6-v2（速度快、维度低） - 问答任务：multi-qa-mpnet-base - 多语言：paraphrase-multilingual-mpnet-base-v2 - 领域特定：使用领域数据微调通用模型

Q4: Dense Retrieval 和 Sparse Retrieval 如何选择？

A: - Dense Retrieval：适合语义理解、同义词匹配、概念检索 - Sparse Retrieval：适合精确关键词匹配、术语检索 - 推荐：使用 Hybrid Retrieval，结合两者优势

Q5: 如何提高检索精度？

A: 多管齐下： 1. 优化 Embedding：使用更好的模型或领域微调 2. 改进切分策略：根据文档特点选择合适的切分方法 3. 使用 Reranking： Cross-Encoder 重排序 4. 查询优化：查询重写、扩展、分解 5. 多阶段检索：先粗排再精排

Q6: RAG 系统出现幻觉怎么办？

A: 1. 提高检索质量：确保检索到的文档与查询相关 2. Prompt 设计：明确要求模型基于检索内容回答，不知道就说不知道 3. 结果验证：对关键信息进行事实核查 4. 使用 Self-RAG：让模型评估检索结果的相关性 5. 置信度评分：对生成结果给出置信度，低置信度时提示用户

Q7: 如何处理长文档？

A: 1. Parent-Child Retrieval：检索小块，返回大块 2. 滑动窗口：检索时包含相邻块 3. 文档摘要：对长文档生成摘要，检索摘要 4. 层次检索：先检索章节，再检索具体内容

Q8: RAG 系统的延迟如何优化？

A: 1. 异步检索：并行检索多个查询 2. 缓存：缓存常见查询的结果 3. 索引优化：使用更快的索引（如 HNSW） 4. 批量处理：批量处理多个查询 5. 模型优化：使用更快的 Embedding 和生成模型

Q9: 如何评估 RAG 系统性能？

A: 评估指标： - 检索指标： Recall@K 、 MRR（ Mean Reciprocal Rank）、 NDCG - 生成指标： BLEU 、 ROUGE 、 BERTScore 、人工评估 - 端到端指标：答案准确性、相关性、完整性 - 系统指标：延迟、吞吐量、成本

def evaluate_rag_system(qa_chain, test_set):
    """评估 RAG 系统"""
    results = []
    
    for query, ground_truth in test_set:
        # 生成答案
        result = qa_chain({"query": query})
        answer = result["result"]
        
        # 计算指标
        from rouge_score import rouge_scorer
        scorer = rouge_scorer.RougeScorer(['rouge1', 'rougeL'], use_stemmer=True)
        scores = scorer.score(ground_truth, answer)
        
        results.append({
            "query": query,
            "answer": answer,
            "ground_truth": ground_truth,
            "rouge1": scores["rouge1"].fmeasure,
            "rougeL": scores["rougeL"].fmeasure
        })
    
    # 计算平均分数
    avg_rouge1 = sum([r["rouge1"] for r in results]) / len(results)
    avg_rougeL = sum([r["rougeL"] for r in results]) / len(results)
    
    return {
        "avg_rouge1": avg_rouge1,
        "avg_rougeL": avg_rougeL,
        "results": results
    }

Q10: 如何构建多轮对话的 RAG 系统？

A: 1. 上下文管理：维护对话历史 2. 查询重写：将当前查询与历史结合 3. 上下文检索：检索时考虑对话上下文 4. 记忆机制：区分短期记忆（当前对话）和长期记忆（知识库）

class ConversationalRAG:
    def __init__(self, qa_chain):
        self.qa_chain = qa_chain
        self.conversation_history = []
    
    def chat(self, query):
        # 1. 构建带上下文的查询
        context_query = self._build_contextual_query(query)
        
        # 2. 检索和生成
        result = self.qa_chain({"query": context_query})
        
        # 3. 更新历史
        self.conversation_history.append({
            "user": query,
            "assistant": result["result"]
        })
        
        return result["result"]
    
    def _build_contextual_query(self, current_query):
        """构建带上下文的查询"""
        if not self.conversation_history:
            return current_query
        
        # 添加最近几轮对话
        recent_history = self.conversation_history[-3:]
        context = "\n".join([
            f"用户：{h['user']}\n 助手：{h['assistant']}"
            for h in recent_history
        ])
        
        return f"""
        对话历史：
        {context}
        
        当前问题：{current_query}
        
        请基于对话历史回答当前问题。
        """

RAG 技术为大语言模型提供了访问外部知识的能力，是构建知识增强 AI 系统的关键技术。一个优秀的 RAG 系统需要精心设计各个组件，从向量数据库选择到检索策略优化，从查询处理到结果生成。在实际应用中，需要根据具体需求选择合适的组件和技术，不断优化和迭代，才能构建出高效、准确的 RAG 系统。