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)

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}")

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]

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))]
)

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]  # 权重分配
)

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
)

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)

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

2、知识库文档生成多向量检索器实现精准检索