RAG实现Skill rag-implementation

---

名称: rag-implementation 描述: 使用向量数据库和语义搜索为LLM应用程序构建检索增强生成（RAG）系统。适用于实现知识增强AI、构建文档问答系统或集成LLM与外部知识库。

RAG实现

掌握检索增强生成（RAG）以构建LLM应用程序，利用外部知识源提供准确、基于事实的回答。

何时使用此技能

• 基于专有文档构建问答系统
• 创建包含当前、事实信息的聊天机器人
• 使用自然语言查询实现语义搜索
• 通过基于事实的回答减少幻觉
• 使LLM能够访问领域特定知识
• 构建文档助手
• 创建具有源引用的研究工具

核心组件

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. 重新排序

目的: 通过重新排序结果提高检索质量

方法:

• 交叉编码器: 基于BERT的重新排序
• Cohere Rerank: 基于API的重新排序
• 最大边际相关性（MMR）: 多样性 + 相关性
• 基于LLM: 使用LLM评分相关性

快速开始

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. 加载文档
loader = DirectoryLoader('./docs', 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. 创建嵌入和向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. 创建检索链
qa_chain = RetrievalQA.from_chain_type(
    llm=OpenAI(),
    chain_type="stuff",
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    return_source_documents=True
)

# 5. 查询
result = qa_chain({"query": "主要特性是什么？"})
print(result['result'])
print(result['source_documents'])

高级RAG模式

模式1: 混合搜索

from langchain.retrievers import BM25Retriever, EnsembleRetriever

# 稀疏检索器（BM25）
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

# 密集检索器（嵌入）
embedding_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 使用权重组合
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, embedding_retriever],
    weights=[0.3, 0.7]
)

模式2: 多查询检索

from langchain.retrievers.multi_query import MultiQueryRetriever

# 生成多个查询视角
retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=OpenAI()
)

# 单个查询 → 多个变体 → 组合结果
results = retriever.get_relevant_documents("主要主题是什么？")

模式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()
)

# 仅返回文档的相关部分
compressed_docs = compression_retriever.get_relevant_documents("查询")

模式4: 父文档检索器

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

# 父文档存储
store = InMemoryStore()

# 检索用小块，上下文用大块
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=["

", "
", " ", ""]  # 按此顺序尝试
)

基于令牌的分割

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 = [
    ("#", "标题1"),
    ("##", "标题2"),
    ("###", "标题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. 元数据过滤

# 索引时添加元数据
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)

# 检索时过滤
results = vectorstore.similarity_search(
    "query",
    filter={"category": "technical"},
    k=5
)

2. 最大边际相关性

# 平衡相关性与多样性
results = vectorstore.max_marginal_relevance_search(
    "query",
    k=5,
    fetch_k=20,  # 获取20个，返回前5个多样化
    lambda_mult=0.5  # 0=最大多样性，1=最大相关性
)

3. 使用交叉编码器重新排序

from sentence_transformers import CrossEncoder

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

# 获取初始结果
candidates = vectorstore.similarity_search("query", k=20)

# 重新排序
pairs = [[query, doc.page_content] for doc in candidates]
scores = reranker.predict(pairs)

# 按分数排序并取前k个
reranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)[:5]

RAG提示工程

上下文提示

prompt_template = """使用以下上下文回答问题。如果无法基于上下文回答，请说“我没有足够的信息”。

上下文:
{context}

问题: {question}

回答:"""

带引用

prompt_template = """基于以下上下文回答问题。包括引用，使用[1]、[2]等。

上下文:
{context}

问题: {question}

回答（带引用）:"""

带置信度

prompt_template = """使用上下文回答问题。为您的答案提供置信度分数（0-100%）。

上下文:
{context}

问题: {question}

回答:
置信度:"""

评估指标

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

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

        # 检查答案是否匹配预期
        accuracy = calculate_accuracy(result['result'], test['expected'])
        metrics['accuracy'].append(accuracy)

        # 检查是否检索到相关文档
        retrieval_quality = evaluate_retrieved_docs(
            result['source_documents'],
            test['relevant_docs']
        )
        metrics['retrieval_quality'].append(retrieval_quality)

        # 检查答案是否基于上下文
        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令牌）
2. 重叠: 使用10-20%重叠以保留边界上下文
3. 元数据: 包括来源、页面、时间戳，用于过滤和调试
4. 混合搜索: 结合语义和关键词搜索以获得最佳结果
5. 重新排序: 使用交叉编码器改进顶部结果
6. 引用: 始终返回源文档以增加透明度
7. 评估: 持续测试检索质量和答案准确性
8. 监控: 在生产中跟踪检索指标

常见问题

• 检索差: 检查嵌入质量、分块大小、查询公式化
• 不相关结果: 添加元数据过滤、使用混合搜索、重新排序
• 缺少信息: 确保文档被正确索引
• 查询慢: 优化向量存储、使用缓存、减少k
• 幻觉: 改进基于事实的提示、添加验证步骤