RAG技术深度解析与实践：让LLM拥有实时知识库

# 安装LangChain（RAG开发框架）
pip install langchain
# 安装嵌入模型依赖
pip install sentence-transformers
# 安装向量数据库Chroma
pip install chromadb
# 安装OpenAI SDK（调用GPT模型）
pip install openai
# 安装文档加载工具（用于加载文本文件）
pip install python-dotenv

import os
from dotenv import load_dotenv
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import Chroma
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import RetrievalQA

# ---------------------- 1. 环境配置 ----------------------
# 加载环境变量（需在.env文件中配置OPENAI_API_KEY）
load_dotenv()
openai_api_key = os.getenv("OPENAI_API_KEY")

# 配置组件参数
# 文档分割参数：片段长度500字符，重叠度50字符（保证上下文连贯性）
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50
# 嵌入模型：使用sentence-transformers的轻量模型
EMBEDDING_MODEL_NAME = "all-MiniLM-L6-v2"
# 向量数据库存储路径
VECTOR_DB_PATH = "./chroma_python_knowledge"
# LLM模型：GPT-3.5-turbo
LLM_MODEL_NAME = "gpt-3.5-turbo"

# ---------------------- 2. 数据准备：构建知识库 ----------------------
def build_knowledge_base(document_paths):
    """
    加载文档、分割、编码并存储到向量数据库
    :param document_paths: 文档路径列表（本示例使用.txt文档）
    :return: 向量数据库检索器
    """
    # 1. 加载文档（此处简化处理，加载.txt文本）
    texts = []
    for path in document_paths:
        with open(path, "r", encoding="utf-8") as f:
            texts.append(f.read())
    
    # 2. 文档分割
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        length_function=len
    )
    splits = text_splitter.split_text("\n\n".join(texts))  # 合并所有文本后分割
    
    # 3. 初始化嵌入模型
    embeddings = HuggingFaceEmbeddings(model_name=EMBEDDING_MODEL_NAME)
    
    # 4. 构建并存储向量数据库
    db = Chroma.from_texts(
        texts=splits,
        embedding=embeddings,
        persist_directory=VECTOR_DB_PATH
    )
    db.persist()  # 持久化存储
    
    # 5. 创建检索器（返回Top 3最相似片段）
    retriever = db.as_retriever(search_kwargs={"k": 3})
    return retriever

# ---------------------- 3. 构建RAG生成链 ----------------------
def build_rag_chain(retriever):
    """
    构建检索-生成链
    :param retriever: 向量数据库检索器
    :return: RAG链
    """
    # 定义Prompt模板：将检索到的上下文和用户问题组合
    prompt_template = """
    你是一个Python学习助手，仅基于提供的上下文信息回答用户问题。
    如果上下文没有相关信息，直接说明“没有找到相关知识点”，不要编造内容。
    
    上下文信息：
    {context}
    
    用户问题：
    {question}
    
    回答：
    """
    
    # 初始化Prompt模板
    prompt = PromptTemplate(
        template=prompt_template,
        input_variables=["context", "question"]
    )
    
    # 初始化LLM
    llm = ChatOpenAI(
        model_name=LLM_MODEL_NAME,
        api_key=openai_api_key,
        temperature=0.3  # 降低随机性，保证回答准确性
    )
    
    # 构建检索-生成链
    rag_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",  # 将所有检索到的上下文塞入Prompt
        retriever=retriever,
        chain_type_kwargs={"prompt": prompt},
        return_source_documents=True  # 返回检索到的原始上下文（用于验证）
    )
    return rag_chain

# ---------------------- 4. 核心查询函数 ----------------------
def rag_query(rag_chain, question):
    """
    执行RAG查询
    :param rag_chain: RAG链
    :param question: 用户问题
    :return: 生成的回答和检索到的上下文
    """
    result = rag_chain({"query": question})
    answer = result["result"]
    source_documents = result["source_documents"]
    return answer, source_documents

# ---------------------- 5. 测试运行 ----------------------
if __name__ == "__main__":
    # 准备测试文档（需提前创建，示例文档内容为Python基础知识点）
    test_documents = ["./python_basics.txt"]  # 文档内容示例：Python变量、函数、列表操作等基础知识点
    
    # 步骤1：构建知识库
    print("正在构建Python学习知识库...")
    retriever = build_knowledge_base(test_documents)
    
    # 步骤2：构建RAG链
    print("正在初始化RAG生成链...")
    rag_chain = build_rag_chain(retriever)
    
    # 步骤3：测试查询
    test_questions = [
        "Python中如何定义函数？",
        "列表和元组的区别是什么？",
        "Python的装饰器有什么作用？"
    ]
    
    for question in test_questions:
        print(f"\n用户问题：{question}")
        answer, sources = rag_query(rag_chain, question)
        print(f"回答：{answer}")
        print("检索到的相关知识点：")
        for i, source in enumerate(sources, 1):
            print(f"  {i}. {source.page_content}")