02.数据检索增强RAG
# 01.数据检索增强RAG
# 1、RAG数据检索增强
RAG =
将企业知识库转化为可搜索的向量库
+ 搭配 LLM 生成答案的机制- ① 向量化存储(把知识文档转为可搜索的向量)
- ② 检索 + LLM问答(智能使用这些向量)
阶段 你做了什么 技术关键词 向量化阶段 把企业文档转成 Embedding 向量 文本切分、Embedding 检索阶段 把用户的问题向量与文档向量比相似度 Top-K 向量搜索 生成阶段 用 LLM 综合文档片段 + 问题生成回答 Prompt + LLM
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from dotenv import load_dotenv
import os
# 加载环境变量
load_dotenv()
# 初始化 DeepSeek 模型
model = ChatOpenAI(
api_key=os.getenv("DEEPSEEK_API_KEY"),
base_url="https://api.deepseek.com/v1",
model="deepseek-chat",
temperature=0.7,
)
# 1. 准备知识库文档 (这里用简单示例代替)
documents = [
"量子计算利用量子比特(qubit)存储信息,与传统比特不同,量子比特可以同时处于0和1的叠加态",
"量子纠缠是量子计算的重要特性,允许量子比特之间存在强关联",
"量子门是量子计算中的基本操作,类似于传统计算中的逻辑门"
]
# 2. 创建嵌入模型和向量数据库
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
vector_db = FAISS.from_texts(documents, embeddings)
# 3. 定义RAG提示模板
prompt_template = """根据以下上下文信息回答问题。如果你不确定答案,就说你不知道。
上下文: {context}
问题: {question}
"""
prompt = ChatPromptTemplate.from_template(prompt_template)
# 4. 创建RAG链
def format_docs(docs):
return "\n\n".join(doc.page_content for doc in docs)
rag_chain = (
{"context": lambda x: vector_db.similarity_search(x["question"]), "question": lambda x: x["question"]}
| {"context": format_docs, "question": lambda x: x["question"]}
| prompt
| model
)
# 5. 使用RAG提问
question = "量子比特和传统比特有什么区别?"
response = rag_chain.invoke({"question": question})
print(response.content)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
# 2、加载知识库文档
# 1)支持多种文档格式
from langchain_community.document_loaders import (
TextLoader, # 文本文件
PyPDFLoader, # PDF
Docx2txtLoader, # Word
WebBaseLoader # 网页
)
# 示例:加载不同格式的文档
loaders = {
".txt": TextLoader,
".pdf": PyPDFLoader,
".docx": Docx2txtLoader
}
def load_documents(file_path):
ext = os.path.splitext(file_path)[-1].lower()
if ext in loaders:
return loaders[ext](file_path).load()
elif file_path.startswith("http"):
return WebBaseLoader(file_path).load()
else:
raise ValueError(f"Unsupported file type: {ext}")
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
# 2)文档预处理
from langchain_text_splitters import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每个分块500字符
chunk_overlap=50, # 分块间重叠50字符
length_function=len,
is_separator_regex=False
)
def preprocess_docs(docs):
# 分割长文档
splits = text_splitter.split_documents(docs)
# 移除多余空格/换行
cleaned = [doc.page_content.replace("\n", " ").strip() for doc in splits]
# 过滤短文本
return [doc for doc in splits if len(doc.page_content) > 30]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
# 3、检索算法设置
# 1)向量检索配置
from langchain_community.vectorstores import FAISS
from langchain.embeddings import HuggingFaceEmbeddings
# 优化后的嵌入模型配置
embeddings = HuggingFaceEmbeddings(
model_name="all-MiniLM-L6-v2",
model_kwargs={"device": "cpu"}, # GPU可用时改为 "cuda"
encode_kwargs={
"normalize_embeddings": True, # 归一化提升相似度计算精度
"batch_size": 32 # 批量处理效率更高
}
)
# 创建带元数据的向量库
vector_db = FAISS.from_documents(
documents=processed_docs,
embedding=embeddings,
metadatas=[{"source": f"doc_{i}"} for i in range(len(processed_docs))]
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# 2)混合检索策略
from langchain.retrievers import BM25Retriever, EnsembleRetriever
# 1. 向量检索
vector_retriever = vector_db.as_retriever(
search_type="mmr", # 最大边际相关性算法
search_kwargs={
"k": 5, # 返回5个结果
"lambda_mult": 0.25 # 多样性控制参数
}
)
# 2. 关键词检索 (BM25)
bm25_retriever = BM25Retriever.from_documents(processed_docs)
bm25_retriever.k = 3
# 3. 组合检索器
ensemble_retriever = EnsembleRetriever(
retrievers=[bm25_retriever, vector_retriever],
weights=[0.4, 0.6] # 权重分配
)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
# 3)检索结果重排序
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
# 使用LLM精炼检索结果
compressor = LLMChainExtractor.from_llm(model)
compression_retriever = ContextualCompressionRetriever(
base_compressor=compressor,
base_retriever=ensemble_retriever
)
1
2
3
4
5
6
7
8
9
2
3
4
5
6
7
8
9
# 4)完整工作流示例
# 1. 加载并预处理文档
doc_paths = ["data/article.pdf", "data/notes.txt"]
all_docs = []
for path in doc_paths:
all_docs.extend(load_documents(path))
processed_docs = preprocess_docs(all_docs)
# 2. 构建检索系统
vector_db = FAISS.from_documents(processed_docs, embeddings)
retriever = setup_retriever(vector_db, processed_docs) # 使用前面定义的混合检索
# 3. 构建RAG链
prompt_template = """基于以下上下文给出专业回答:
{context}
问题:{question}
要求:
1. 如果上下文不相关,回答"根据现有信息无法确定"
2. 保持回答简洁专业
"""
prompt = ChatPromptTemplate.from_template(prompt_template)
rag_chain = (
{"context": compression_retriever, "question": RunnablePassthrough()}
| prompt
| model
)
# 4. 执行查询
question = "量子比特的叠加态如何表示?"
result = rag_chain.invoke(question)
print(result.content)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
# 02.解析PDF文档为知识库
# 1、解析PDF为知识库
处理 PDF 文档构建知识库是 RAG 系统的关键环节
包含 PDF 解析、文本处理、向量化存储和检索优化
数据流转流程
关键组件说明
文本分块 将长文档分割为适合模型处理的片段 LangChain TextSplitter 嵌入模型 将文本转化为高维向量(通常768/1024维) HuggingFace Embeddings, OpenAI 向量数据库 高效存储和检索向量 FAISS, Chroma, Pinecone 相似度算法 计算向量间的相似度 余弦相似度/Cosine, L2距离
Q1:为什么不需要传统数据库?
向量搜索的特殊性:传统B+树索引不适合高维向量相似度计算
性能考量:专用向量索引(如FAISS)针对SIMD指令优化,比SQL查询快100倍+
Q3:如何处理大规模数据?
- <1M 文档,FAISS + 单机内存
- 1M-100M,FAISS + 内存映射文件
- >100M,分布式方案(Milvus/Pinecone)
from langchain_community.vectorstores import FAISS
from langchain.embeddings import HuggingFaceBgeEmbeddings
import os
def build_knowledge_base(pdf_folder):
"""从PDF文件夹构建知识库"""
all_docs = []
# 1. 遍历PDF文件
for file in os.listdir(pdf_folder):
if file.endswith(".pdf"):
file_path = os.path.join(pdf_folder, file)
# 2. 解析PDF
elements, raw_text = parse_pdf(file_path)
processed_elements = process_tables(elements)
# 3. 分块处理
chunks = semantic_chunking(raw_text)
chunks.extend([Document(page_content=e.content) for e in processed_elements])
# 4. 添加元数据
final_docs = add_metadata(chunks, file)
all_docs.extend(final_docs)
# 5. 构建向量库
embeddings = HuggingFaceBgeEmbeddings(
model_name="BAAI/bge-small-zh-v1.5", # 中文优化模型
encode_kwargs={"normalize_embeddings": True}
)
# 6. 创建带元数据的向量存储
vector_db = FAISS.from_documents(
documents=all_docs,
embedding=embeddings,
metadatas=[doc.metadata for doc in all_docs]
)
# 7. 保存索引
vector_db.save_local("pdf_knowledge_base")
return vector_db
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41