rag-implementation

RAG 实现

掌握检索增强生成（RAG），构建基于外部知识源提供准确、有据可依回答的大模型应用。

适用场景

• 基于专有文档构建问答系统
• 创建能够提供最新、准确事实信息的聊天机器人
• 使用自然语言查询实现语义搜索
• 通过提供有据可依的回答减少幻觉
• 让大模型能够访问特定领域的知识
• 构建文档助手
• 创建带有来源引用的研究工具

不适用场景

• 只需要纯生成式写作，无需检索
• 数据集太小，不值得使用嵌入
• 无法安全存储或处理源数据

说明

1. 明确语料库、更新频率和评估目标。
2. 根据规模选择嵌入模型和向量存储。
3. 构建包含重排序的摄取、分块和检索流程。
4. 使用有据可依的问答指标进行评估，并监控偏移。

安全

• 对敏感数据进行脱敏，并实施访问控制。
• 如果受到限制，请勿在回答中暴露源文档。

核心组件

1. 向量数据库

用途：高效存储和检索文档嵌入

选项：

• Pinecone：托管式、可扩展、查询速度快
• Weaviate：开源、混合搜索
• Milvus：高性能、可本地部署
• Chroma：轻量、易用
• Qdrant：搜索快速，支持筛选
• FAISS：Meta 的库，可本地部署

2. 嵌入

用途：将文本转换为用于相似度搜索的数值向量

模型：

• text-embedding-ada-002（OpenAI）：通用型，1536 维
• all-MiniLM-L6-v2（Sentence Transformers）：快速、轻量
• e5-large-v2：高质量、多语言
• Instructor：针对任务提供指令
• bge-large-en-v1.5：SOTA 性能

3. 检索策略

方法：

• 稠密检索：通过嵌入计算语义相似度
• 稀疏检索：关键词匹配（BM25、TF-IDF）
• 混合搜索：结合稠密检索和稀疏检索
• 多查询：生成多个查询变体
• HyDE：生成假设文档

4. 重排序

用途：通过重新排列结果来提升检索质量

方法：

• Cross-Encoder：基于 BERT 的重排序
• Cohere Rerank：基于 API 的重排序
• 最大边际相关性（MMR）：兼顾多样性和相关性
• 基于大模型：使用大模型评估相关性

快速入门

from langchain.document_loaders import DirectoryLoader
from langchain.text_splitters import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI

# 1. Load documents
loader = DirectoryLoader('./docs', glob="**/*.txt")
documents = loader.load()

# 2. Split into chunks
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len
)
chunks = text_splitter.split_documents(documents)

# 3. Create embeddings and vector store
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. Create retrieval chain
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(),
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

# 5. Query
result = qa_chain({"query": "What are the main features?"})
print(result['result'])
print(result['source_documents'])

高级 RAG 模式

模式 1：混合搜索

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# Sparse retriever (BM25)
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# Dense retriever (embeddings)
embedding_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# Combine with weights
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, embedding_retriever],
    weights=[0.3, 0.7]
)

模式 2：多查询检索

from langchain.retrievers.multi_query import MultiQueryRetriever

# Generate multiple query perspectives
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=OpenAI()
)

# Single query → multiple variations → combined results
results = retriever.get_relevant_documents("What is the main topic?")

模式 3：上下文压缩

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever()
)

# Returns only relevant parts of documents
compressed_docs = compression_retriever.get_relevant_documents("query")

模式 4：父文档检索器

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# Store for parent documents
store = InMemoryStore()

# Small chunks for retrieval, large chunks for context
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

文档分块策略

递归字符文本拆分器

from langchain.text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    length_function=len,
    separators=["\n\n", "\n", " ", ""]  # Try these in order
)

基于 Token 的拆分

from langchain.text_splitters import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=512,
    chunk_overlap=50
)

语义分块

from langchain.text_splitters import SemanticChunker

splitter = SemanticChunker(
    embeddings=OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile"
)

Markdown 标题拆分器

from langchain.text_splitters import MarkdownHeaderTextSplitter

headers_to_split_on = [
    ("#", "Header 1"),
    ("##", "Header 2"),
    ("###", "Header 3"),
]

splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers_to_split_on)

向量存储配置

Pinecone

import pinecone
from langchain.vectorstores import Pinecone

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

index = pinecone.Index("your-index-name")

vectorstore = Pinecone(index, embeddings.embed_query, "text")

Weaviate

import weaviate
from langchain.vectorstores import Weaviate

client = weaviate.Client("http://localhost:8080")

vectorstore = Weaviate(client, "Document", "content", embeddings)

Chroma（本地）

from langchain.vectorstores import Chroma

vectorstore = Chroma(
    collection_name="my_collection",
    embedding_function=embeddings,
    persist_directory="./chroma_db"
)

检索优化

1. 元数据筛选

# Add metadata during indexing
chunks_with_metadata = []
for i, chunk in enumerate(chunks):
    chunk.metadata = {
        "source": chunk.metadata.get("source"),
        "page": i,
        "category": determine_category(chunk.page_content)
    }
    chunks_with_metadata.append(chunk)

# Filter during retrieval
results = vectorstore.similarity_search(
    "query",
    filter={"category": "technical"},
    k=5
)

2. 最大边际相关性

# Balance relevance with diversity
results = vectorstore.max_marginal_relevance_search(
    "query",
    k=5,
    fetch_k=20,  # Fetch 20, return top 5 diverse
    lambda_mult=0.5  # 0=max diversity, 1=max relevance
)

3. 使用 Cross-Encoder 进行重排序

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')

# Get initial results
candidates = vectorstore.similarity_search("query", k=20)

# Rerank
pairs = [[query, doc.page_content] for doc in candidates]
scores = reranker.predict(pairs)

# Sort by score and take top k
reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)[:5]

RAG 提示词工程

上下文提示词

prompt_template = """Use the following context to answer the question. If you cannot answer based on the context, say "I don't have enough information."

Context:
{context}

Question: {question}

Answer:"""

包含引用

prompt_template = """Answer the question based on the context below. Include citations using [1], [2], etc.

Context:
{context}

Question: {question}

Answer (with citations):"""

包含置信度

prompt_template = """Answer the question using the context. Provide a confidence score (0-100%) for your answer.

Context:
{context}

Question: {question}

Answer:
Confidence:"""

评估指标

def evaluate_rag_system(qa_chain, test_cases):
    metrics = {
        'accuracy': [],
        'retrieval_quality': [],
        'groundedness': []
    }

    for test in test_cases:
        result = qa_chain({"query": test['question']})

        # Check if answer matches expected
        accuracy = calculate_accuracy(result['result'], test['expected'])
        metrics['accuracy'].append(accuracy)

        # Check if relevant docs were retrieved
        retrieval_quality = evaluate_retrieved_docs(
            result['source_documents'],
            test['relevant_docs']
        )
        metrics['retrieval_quality'].append(retrieval_quality)

        # Check if answer is grounded in context
        groundedness = check_groundedness(
            result['result'],
            result['source_documents']
        )
        metrics['groundedness'].append(groundedness)

    return {k: sum(v)/len(v) for k, v in metrics.items()}

资源

• references/vector-databases.md：向量数据库详细对比
• references/embeddings.md：嵌入模型选择指南
• references/retrieval-strategies.md：高级检索技术
• references/reranking.md：重排序方法及其适用场景
• references/context-window.md：管理上下文限制
• assets/vector-store-config.yaml：配置模板
• assets/retriever-pipeline.py：完整的 RAG 流水线
• assets/embedding-models.md：模型对比和基准测试

最佳实践

1. 分块大小：在上下文和具体程度之间取得平衡（500-1000 个 token）
2. 重叠：使用 10-20% 的重叠，以保留边界处的上下文
3. 元数据：包含来源、页面和时间戳，以便筛选和调试
4. 混合搜索：结合语义搜索和关键词搜索，以获得最佳结果
5. 重排序：使用交叉编码器改进排名靠前的结果
6. 引用：始终返回来源文档，以提高透明度
7. 评估：持续测试检索质量和回答准确性
8. 监控：跟踪生产环境中的检索指标

常见问题

• 检索效果差：检查嵌入质量、分块大小和查询表述
• 结果不相关：添加元数据筛选、使用混合搜索并重新排序
• 信息缺失：确保文档已正确编入索引
• 查询缓慢：优化向量存储、使用缓存并减小 k 值
• 幻觉：改进 grounding 提示词，并添加验证步骤