RAG技术完全指南

检索增强生成(Retrieval-Augmented Generation, RAG)是解决LLM知识局限性的关键技术，通过引入外部知识库大幅提升AI应用的准确性和可靠性。

什么是RAG

核心概念

RAG结合了信息检索和生成式AI两大技术：

为什么需要RAG

LLM的固有局限：

❌ 知识截止日期限制
❌ 无法访问私有数据
❌ 容易产生幻觉
❌ 缺乏领域专业知识

RAG的解决方案：

✅ 实时获取最新信息
✅ 整合企业内部知识
✅ 提供可验证的信息来源
✅ 注入领域专业知识

RAG vs 微调

维度	RAG	微调
成本	低（无需重新训练）	高（需要GPU资源）
更新速度	快（更新知识库即可）	慢（需要重新训练）
知识来源	可追溯	黑盒
适用场景	知识密集型任务	特定风格/格式
技术门槛	低	高

RAG系统架构

完整工作流程

python

1# RAG系统核心流程
2class RAGSystem:
3    def __init__(self, llm, vector_store, embeddings):
4        self.llm = llm
5        self.vector_store = vector_store
6        self.embeddings = embeddings
7    
8    def query(self, question: str) -> str:
9        # 1. 问题向量化
10        query_embedding = self.embeddings.embed_query(question)
11        
12        # 2. 检索相关文档
13        relevant_docs = self.vector_store.similarity_search(
14            query_embedding, 
15            k=5
16        )
17        
18        # 3. 构建增强提示
19        context = "\n\n".join([doc.content for doc in relevant_docs])
20        prompt = f"""
21        基于以下上下文回答问题：
22        
23        上下文：
24        {context}
25        
26        问题：{question}
27        
28        请基于上下文提供准确答案，如果上下文中没有相关信息，请明确说明。
29        """
30        
31        # 4. LLM生成答案
32        answer = self.llm.generate(prompt)
33        
34        return answer

关键组件

1. 文档加载与预处理

python

1from langchain.document_loaders import (
2    TextLoader, 
3    PDFLoader, 
4    WebBaseLoader
5)
6from langchain.text_splitter import RecursiveCharacterTextSplitter
7
8# 加载文档
9loader = PDFLoader("knowledge_base.pdf")
10documents = loader.load()
11
12# 文档切分
13text_splitter = RecursiveCharacterTextSplitter(
14    chunk_size=1000,        # 每个块的大小
15    chunk_overlap=200,      # 块之间的重叠
16    separators=["\n\n", "\n", "。", "！", "？", " ", ""]
17)
18
19chunks = text_splitter.split_documents(documents)

切分策略：

策略	适用场景	优点	缺点
固定长度	通用场景	简单高效	可能割裂语义
语义切分	长文档	保持语义完整	计算成本高
结构化切分	代码/表格	保留结构信息	需要特定解析器

2. 向量化(Embedding)

python

1from langchain.embeddings import OpenAIEmbeddings, HuggingFaceEmbeddings
2
3# 使用OpenAI Embedding
4openai_embeddings = OpenAIEmbeddings(
5    model="text-embedding-3-small"
6)
7
8# 使用开源模型
9hf_embeddings = HuggingFaceEmbeddings(
10    model_name="BAAI/bge-large-zh-v1.5"  # 中文效果好
11)
12
13# 向量化文本
14text = "RAG是检索增强生成技术"
15vector = openai_embeddings.embed_query(text)
16print(f"向量维度: {len(vector)}")  # 通常1536维

主流Embedding模型对比：

模型	维度	语言	性能	成本
text-embedding-3-small	1536	多语言	优秀	低
text-embedding-3-large	3072	多语言	最佳	中
bge-large-zh-v1.5	1024	中文	优秀	免费
m3e-base	768	中文	良好	免费

3. 向量数据库

python

1from langchain.vectorstores import Chroma, Pinecone, Weaviate
2
3# 使用Chroma (本地开发)
4vectorstore = Chroma.from_documents(
5    documents=chunks,
6    embedding=openai_embeddings,
7    persist_directory="./chroma_db"
8)
9
10# 使用Pinecone (生产环境)
11import pinecone
12pinecone.init(api_key="your-api-key", environment="us-west1-gcp")
13
14vectorstore = Pinecone.from_documents(
15    documents=chunks,
16    embedding=openai_embeddings,
17    index_name="rag-knowledge-base"
18)
19
20# 相似度搜索
21results = vectorstore.similarity_search(
22    query="什么是RAG",
23    k=5  # 返回前5个最相关的文档
24)

向量数据库选型：

数据库	特点	适用场景	部署方式
Chroma	轻量级，易用	开发测试	本地/内存
Pinecone	云原生，高性能	生产环境	托管服务
Weaviate	功能丰富	企业级应用	自托管/云
Milvus	开源，高扩展性	大规模数据	自托管
Qdrant	快速，Rust实现	高性能需求	自托管/云

RAG实现方案

基础RAG实现

python

1from langchain.chains import RetrievalQA
2from langchain.llms import OpenAI
3from langchain.vectorstores import Chroma
4from langchain.embeddings import OpenAIEmbeddings
5
6# 初始化组件
7embeddings = OpenAIEmbeddings()
8vectorstore = Chroma(
9    persist_directory="./db",
10    embedding_function=embeddings
11)
12llm = OpenAI(temperature=0)
13
14# 构建RAG链
15qa_chain = RetrievalQA.from_chain_type(
16    llm=llm,
17    chain_type="stuff",
18    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
19    return_source_documents=True
20)
21
22# 查询
23result = qa_chain({"query": "什么是向量数据库？"})
24print(f"答案: {result['result']}")
25print(f"来源: {result['source_documents']}")

高级RAG模式

1. 多查询RAG

python

1from langchain.retrievers import MultiQueryRetriever
2
3# 从不同角度生成多个查询
4multi_query_retriever = MultiQueryRetriever.from_llm(
5    retriever=vectorstore.as_retriever(),
6    llm=llm
7)
8
9# 自动生成3-5个相关查询并检索
10results = multi_query_retriever.get_relevant_documents(
11    "如何优化RAG性能？"
12)

2. 上下文压缩RAG

python

1from langchain.retrievers import ContextualCompressionRetriever
2from langchain.retrievers.document_compressors import LLMChainExtractor
3
4# 使用LLM提取最相关的片段
5compressor = LLMChainExtractor.from_llm(llm)
6compression_retriever = ContextualCompressionRetriever(
7    base_compressor=compressor,
8    base_retriever=vectorstore.as_retriever()
9)
10
11# 只返回压缩后的相关内容
12compressed_docs = compression_retriever.get_relevant_documents(
13    "RAG的优势是什么？"
14)

3. 混合检索(Hybrid Search)

python

1from langchain.retrievers import EnsembleRetriever
2from langchain.retrievers import BM25Retriever
3
4# 关键词检索
5bm25_retriever = BM25Retriever.from_documents(chunks)
6bm25_retriever.k = 3
7
8# 向量检索
9vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
10
11# 混合检索（结合关键词和语义）
12ensemble_retriever = EnsembleRetriever(
13    retrievers=[bm25_retriever, vector_retriever],
14    weights=[0.4, 0.6]  # 关键词40%，语义60%
15)
16
17docs = ensemble_retriever.get_relevant_documents("RAG实现方案")

4. Self-Query检索

python

1from langchain.retrievers.self_query.base import SelfQueryRetriever
2from langchain.chains.query_constructor.base import AttributeInfo
3
4# 定义文档元数据
5metadata_field_info = [
6    AttributeInfo(
7        name="source",
8        description="文档来源",
9        type="string"
10    ),
11    AttributeInfo(
12        name="date",
13        description="文档日期",
14        type="string"
15    ),
16]
17
18# 自动从用户查询中提取过滤条件
19self_query_retriever = SelfQueryRetriever.from_llm(
20    llm=llm,
21    vectorstore=vectorstore,
22    document_contents="技术文档",
23    metadata_field_info=metadata_field_info
24)
25
26# "找2024年发布的关于RAG的文档" -> 自动添加date过滤
27docs = self_query_retriever.get_relevant_documents(
28    "2024年的RAG技术文档"
29)

RAG性能优化

1. 检索质量优化

优化Chunk策略

python

1# 方案1: 小块+父块引用
2from langchain.retrievers import ParentDocumentRetriever
3from langchain.storage import InMemoryStore
4
5# 小块用于检索（精确匹配）
6child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)
7
8# 大块用于上下文（完整信息）
9parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
10
11store = InMemoryStore()
12retriever = ParentDocumentRetriever(
13    vectorstore=vectorstore,
14    docstore=store,
15    child_splitter=child_splitter,
16    parent_splitter=parent_splitter
17)

重排序(Reranking)

python

1from langchain.retrievers import ContextualCompressionRetriever
2from langchain.retrievers.document_compressors import CohereRerank
3
4# 使用Cohere Rerank API
5compressor = CohereRerank(api_key="your-api-key", top_n=3)
6
7compression_retriever = ContextualCompressionRetriever(
8    base_compressor=compressor,
9    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 10})
10)
11
12# 先检索10个，然后重排序选出最相关的3个
13docs = compression_retriever.get_relevant_documents("RAG优化技巧")

2. 生成质量优化

提示词模板优化

python

1from langchain.prompts import PromptTemplate
2
3# 优化后的提示词模板
4template = """
5你是一位专业的技术助手。请基于以下参考资料回答用户问题。
6
7参考资料：
8{context}
9
10用户问题：{question}
11
12回答要求：
131. 优先使用参考资料中的信息
142. 如果参考资料不足，明确说明
153. 引用具体的来源
164. 回答要准确、简洁、专业
17
18你的回答：
19"""
20
21prompt = PromptTemplate(
22    template=template,
23    input_variables=["context", "question"]
24)

答案验证与引用

python

1def rag_with_citation(question: str) -> dict:
2    # 检索文档
3    docs = vectorstore.similarity_search(question, k=3)
4    
5    # 构建带编号的上下文
6    context_with_citations = ""
7    for i, doc in enumerate(docs, 1):
8        context_with_citations += f"\n[{i}] {doc.page_content}\n"
9    
10    # 要求LLM引用来源
11    prompt = f"""
12    基于以下资料回答问题，并在答案中标注引用来源（如[1]、[2]）。
13    
14    资料：
15    {context_with_citations}
16    
17    问题：{question}
18    """
19    
20    answer = llm.generate(prompt)
21    
22    return {
23        "answer": answer,
24        "sources": [{"id": i+1, "content": doc.page_content, "metadata": doc.metadata} 
25                   for i, doc in enumerate(docs)]
26    }

3. 性能优化

缓存策略

python

1from functools import lru_cache
2import hashlib
3
4class CachedRAG:
5    def __init__(self, rag_system):
6        self.rag = rag_system
7        self.cache = {}
8    
9    def query(self, question: str) -> str:
10        # 问题指纹
11        cache_key = hashlib.md5(question.encode()).hexdigest()
12        
13        if cache_key in self.cache:
14            print("命中缓存")
15            return self.cache[cache_key]
16        
17        # 执行RAG
18        answer = self.rag.query(question)
19        
20        # 缓存结果
21        self.cache[cache_key] = answer
22        
23        return answer

批量处理

python

1async def batch_rag_query(questions: list) -> list:
2    import asyncio
3    
4    async def process_single(q):
5        return await rag_system.aquery(q)
6    
7    # 并发处理多个查询
8    tasks = [process_single(q) for q in questions]
9    results = await asyncio.gather(*tasks)
10    
11    return results

完整RAG项目示例

构建企业知识库问答系统

python

1import os
2from langchain.document_loaders import DirectoryLoader, TextLoader
3from langchain.text_splitter import RecursiveCharacterTextSplitter
4from langchain.embeddings import OpenAIEmbeddings
5from langchain.vectorstores import Chroma
6from langchain.chains import RetrievalQA
7from langchain.llms import OpenAI
8
9class EnterpriseKnowledgeBase:
10    def __init__(self, docs_dir: str, persist_dir: str = "./kb_db"):
11        self.docs_dir = docs_dir
12        self.persist_dir = persist_dir
13        self.vectorstore = None
14        self.qa_chain = None
15    
16    def build_knowledge_base(self):
17        """构建知识库"""
18        print("📚 加载文档...")
19        loader = DirectoryLoader(
20            self.docs_dir,
21            glob="**/*.md",
22            loader_cls=TextLoader
23        )
24        documents = loader.load()
25        print(f"加载了 {len(documents)} 个文档")
26        
27        print("✂️ 切分文档...")
28        text_splitter = RecursiveCharacterTextSplitter(
29            chunk_size=1000,
30            chunk_overlap=200,
31            separators=["\n## ", "\n### ", "\n\n", "\n", "。", " "]
32        )
33        chunks = text_splitter.split_documents(documents)
34        print(f"切分为 {len(chunks)} 个片段")
35        
36        print("🔢 向量化并存储...")
37        embeddings = OpenAIEmbeddings()
38        self.vectorstore = Chroma.from_documents(
39            documents=chunks,
40            embedding=embeddings,
41            persist_directory=self.persist_dir
42        )
43        self.vectorstore.persist()
44        print("✅ 知识库构建完成！")
45    
46    def load_knowledge_base(self):
47        """加载已有知识库"""
48        embeddings = OpenAIEmbeddings()
49        self.vectorstore = Chroma(
50            persist_directory=self.persist_dir,
51            embedding_function=embeddings
52        )
53    
54    def setup_qa_chain(self):
55        """设置问答链"""
56        llm = OpenAI(temperature=0, model="gpt-3.5-turbo-instruct")
57        
58        self.qa_chain = RetrievalQA.from_chain_type(
59            llm=llm,
60            chain_type="stuff",
61            retriever=self.vectorstore.as_retriever(
62                search_kwargs={"k": 5}
63            ),
64            return_source_documents=True
65        )
66    
67    def query(self, question: str) -> dict:
68        """查询知识库"""
69        if not self.qa_chain:
70            self.setup_qa_chain()
71        
72        result = self.qa_chain({"query": question})
73        
74        return {
75            "answer": result["result"],
76            "sources": [
77                {
78                    "content": doc.page_content[:200] + "...",
79                    "source": doc.metadata.get("source", "Unknown")
80                }
81                for doc in result["source_documents"]
82            ]
83        }
84    
85    def interactive_mode(self):
86        """交互式问答"""
87        print("\n💬 知识库问答系统启动！输入'exit'退出。\n")
88        
89        while True:
90            question = input("问题: ")
91            if question.lower() == 'exit':
92                break
93            
94            result = self.query(question)
95            
96            print(f"\n答案: {result['answer']}\n")
97            print("来源:")
98            for i, source in enumerate(result['sources'], 1):
99                print(f"  [{i}] {source['source']}")
100            print()
101
102# 使用示例
103if __name__ == "__main__":
104    kb = EnterpriseKnowledgeBase(docs_dir="./company_docs")
105    
106    # 首次构建
107    # kb.build_knowledge_base()
108    
109    # 后续使用
110    kb.load_knowledge_base()
111    kb.interactive_mode()

RAG评估与监控

评估指标

python

1from ragas import evaluate
2from ragas.metrics import (
3    faithfulness,          # 忠实度（答案是否基于上下文）
4    answer_relevancy,      # 答案相关性
5    context_precision,     # 上下文精确度
6    context_recall        # 上下文召回率
7)
8
9# 评估数据集
10eval_dataset = {
11    "question": ["什么是RAG?", "RAG有哪些优点?"],
12    "answer": ["...", "..."],
13    "contexts": [[...], [...]],
14    "ground_truth": ["...", "..."]
15}
16
17# 运行评估
18results = evaluate(
19    dataset=eval_dataset,
20    metrics=[
21        faithfulness,
22        answer_relevancy,
23        context_precision,
24        context_recall
25    ]
26)
27
28print(results)

监控系统

python

1import time
2from datetime import datetime
3
4class RAGMonitor:
5    def __init__(self):
6        self.metrics = []
7    
8    def log_query(self, question, answer, latency, sources):
9        """记录查询指标"""
10        self.metrics.append({
11            "timestamp": datetime.now(),
12            "question": question,
13            "answer_length": len(answer),
14            "latency": latency,
15            "num_sources": len(sources),
16        })
17    
18    def get_stats(self):
19        """获取统计数据"""
20        if not self.metrics:
21            return {}
22        
23        latencies = [m["latency"] for m in self.metrics]
24        
25        return {
26            "total_queries": len(self.metrics),
27            "avg_latency": sum(latencies) / len(latencies),
28            "max_latency": max(latencies),
29            "min_latency": min(latencies)
30        }
31
32# 使用
33monitor = RAGMonitor()
34
35def monitored_query(question):
36    start_time = time.time()
37    result = rag_system.query(question)
38    latency = time.time() - start_time
39    
40    monitor.log_query(
41        question=question,
42        answer=result["answer"],
43        latency=latency,
44        sources=result["sources"]
45    )
46    
47    return result

RAG最佳实践

1. 文档准备

✅ 做：

清理文档格式（去除无关内容）
保留文档结构（标题、章节）
添加元数据（日期、作者、标签）
使用语义化的切分策略

❌ 不要做：

直接使用原始文档
忽略文档质量
使用过大或过小的chunk

2. 检索优化

✅ 做：

使用混合检索（关键词+语义）
实施重排序
调整top-k参数
考虑使用过滤条件

❌ 不要做：

只依赖单一检索方法
忽略检索质量
返回过多或过少文档

3. 提示词设计

✅ 做：

明确指示LLM使用上下文
要求引用来源
处理"不知道"的情况
结构化输出格式

❌ 不要做：

使用模糊的指令
让LLM过度推理
忽略引用来源

常见问题解决

Q1: 检索结果不相关

解决方案：

python

1# 1. 检查embedding质量
2# 测试embedding相似度
3test_query = "RAG技术"
4test_doc = "检索增强生成"
5
6query_vec = embeddings.embed_query(test_query)
7doc_vec = embeddings.embed_documents([test_doc])[0]
8
9similarity = cosine_similarity([query_vec], [doc_vec])
10print(f"相似度: {similarity}")
11
12# 2. 尝试不同的embedding模型
13# 中文建议使用: bge-large-zh-v1.5 或 m3e-base
14
15# 3. 优化文档切分
16# 增加chunk_overlap，使用语义切分

Q2: 答案产生幻觉

解决方案：

python

1# 1. 强化提示词约束
2strict_prompt = """
3严格基于以下上下文回答问题。如果上下文中没有相关信息，
4请明确回答"根据提供的资料，我无法回答这个问题"。
5绝对不要编造信息。
6
7上下文：{context}
8
9问题：{question}
10"""
11
12# 2. 降低temperature
13llm = OpenAI(temperature=0)  # 更确定性的输出
14
15# 3. 实施答案验证
16def verify_answer(answer, contexts):
17    """验证答案是否来自上下文"""
18    verification_prompt = f"""
19    答案：{answer}
20    上下文：{contexts}
21    
22    答案中的信息是否全部来自上下文？回答Yes或No。
23    """
24    return llm.generate(verification_prompt)

Q3: 响应速度慢

解决方案：

python

1# 1. 使用更快的embedding模型
2fast_embeddings = HuggingFaceEmbeddings(
3    model_name="sentence-transformers/all-MiniLM-L6-v2"
4)
5
6# 2. 减少检索数量
7retriever = vectorstore.as_retriever(search_kwargs={"k": 3})  # 从5减到3
8
9# 3. 使用更快的LLM
10llm = ChatOpenAI(model="gpt-3.5-turbo")  # 比gpt-4快很多
11
12# 4. 实施缓存
13from langchain.cache import InMemoryCache
14langchain.llm_cache = InMemoryCache()

进阶话题

RAG + Agent

python

1from langchain.agents import initialize_agent, Tool
2
3# 将RAG作为工具集成到Agent
4rag_tool = Tool(
5    name="KnowledgeBase",
6    func=rag_system.query,
7    description="用于查询企业知识库的工具"
8)
9
10agent = initialize_agent(
11    tools=[rag_tool],
12    llm=llm,
13    agent="zero-shot-react-description"
14)
15
16# Agent可以自主决定是否使用知识库
17response = agent.run("公司的休假政策是什么？")

GraphRAG

python

1# 使用知识图谱增强RAG
2from langchain.graphs import Neo4jGraph
3
4graph = Neo4jGraph(
5    url="bolt://localhost:7687",
6    username="neo4j",
7    password="password"
8)
9
10# 图谱检索
11def graph_rag_query(question):
12    # 1. 从问题提取实体
13    entities = extract_entities(question)
14    
15    # 2. 图谱查询
16    graph_results = graph.query(
17        f"MATCH (n)-[r]->(m) WHERE n.name IN {entities} RETURN n, r, m"
18    )
19    
20    # 3. 结合向量检索
21    vector_results = vectorstore.similarity_search(question)
22    
23    # 4. 融合结果
24    combined_context = merge_results(graph_results, vector_results)
25    
26    # 5. LLM生成
27    return llm.generate_with_context(question, combined_context)

总结

RAG技术是构建智能问答系统的核心技术，通过本指南你已经掌握：

✅ RAG的原理和架构
✅ 完整的实现流程
✅ 性能优化策略
✅ 实战项目开发
✅ 常见问题解决

下一步学习

LangChain框架 - 快速构建RAG应用
向量数据库深入 - 掌握向量检索技术
LLM基础 - 理解大语言模型

参考资源

forum

评论区 / Comments

share分享文章 / Share

什么是RAG​

核心概念​

为什么需要RAG​

RAG vs 微调​

RAG系统架构​

完整工作流程​

关键组件​

1. 文档加载与预处理​

2. 向量化(Embedding)​

3. 向量数据库​

RAG实现方案​

基础RAG实现​

高级RAG模式​

1. 多查询RAG​

2. 上下文压缩RAG​

3. 混合检索(Hybrid Search)​

4. Self-Query检索​

RAG性能优化​

1. 检索质量优化​

优化Chunk策略​

重排序(Reranking)​

2. 生成质量优化​

提示词模板优化​

答案验证与引用​

3. 性能优化​

缓存策略​

批量处理​

完整RAG项目示例​

构建企业知识库问答系统​

RAG评估与监控​

评估指标​

监控系统​

RAG最佳实践​

1. 文档准备​

2. 检索优化​

3. 提示词设计​

常见问题解决​

Q1: 检索结果不相关​

Q2: 答案产生幻觉​

Q3: 响应速度慢​

进阶话题​

RAG + Agent​

GraphRAG​

总结​

参考资源​

评论区 / Comments

什么是RAG

核心概念

为什么需要RAG

RAG vs 微调

RAG系统架构

完整工作流程

关键组件

1. 文档加载与预处理

2. 向量化(Embedding)

3. 向量数据库

RAG实现方案

基础RAG实现

高级RAG模式

1. 多查询RAG

2. 上下文压缩RAG

3. 混合检索(Hybrid Search)

4. Self-Query检索

RAG性能优化

1. 检索质量优化

优化Chunk策略

重排序(Reranking)

2. 生成质量优化

提示词模板优化

答案验证与引用

3. 性能优化

缓存策略

批量处理

完整RAG项目示例

构建企业知识库问答系统

RAG评估与监控

评估指标

监控系统

RAG最佳实践

1. 文档准备

2. 检索优化

3. 提示词设计

常见问题解决

Q1: 检索结果不相关

Q2: 答案产生幻觉

Q3: 响应速度慢

进阶话题

RAG + Agent

GraphRAG

总结

参考资源