烤点老白薯

06-21 22:29 北京体育大学测试工程师发布于北京

关注

RAG 项目各模块的详细解释和示例

以下是针对RAG项目各模块的详细展开，结合初学者需求提供实例、亮点及实践方案，帮助理解原理与落地思路：

一、RAG知识库：从概念到应用

1. 什么是RAG？为什么需要知识库？

概念：RAG（Retrieval-Augmented Generation）即检索增强生成，通过外部知识库检索相关信息，辅助大模型生成更准确、实时的回答，解决大模型“幻觉”（生成错误信息）和“记忆有限”的问题。
为什么需要？ 例：当大模型被问“2025年北京冬奥会的新增项目”时，若知识库未更新，模型可能生成错误答案；而RAG通过检索最新知识库，能准确返回信息。
作用：让大模型具备“外部记忆”，支持动态更新知识（如产品手册、法律条文、行业报告等）。

2. 知识库类型与实例

文本知识库（最常用）：例：企业内部FAQ文档、医学指南、代码文档（如Python官方文档分章节存储）。作用：通过文本分块和向量化，快速匹配用户查询。
多模态知识库（进阶）：例：图像知识库（如商品图片+描述）、音频知识库（如语音转文字后存储）。作用：支持跨模态检索（如“查找红色连衣裙的图片”），但初学者可先聚焦文本。

二、整体架构：RAG系统的“骨架”

1. 核心组件与数据流

组件拆解：输入层：用户查询（如“如何安装Python库”）。检索层：向量数据库根据查询向量检索相关文档块。生成层：大模型结合检索结果生成回答。输出层：返回答案，可能附带引用来源。
流程图例（初学者可手绘）：

用户查询 → 向量化 → 向量数据库检索（Top-K相关文档）→ 文档块拼接 → 大模型生成 → 回答

亮点：相比直接调用大模型，RAG通过“先检索后生成”，将回答准确率提升约30%（如客服场景中减少错误引导）。

2. 实战思路：以客服系统为例

场景：用户问“订单退款需要多久？”
流程：系统从知识库中检索“退款政策”相关文档块；大模型根据文档内容生成回答：“普通订单退款将在3-5个工作日到账，具体请查看订单详情页。”

三、分块和向量化：让数据“可计算”

1. 分块：解决长文本处理难题

为什么分块？ 例：若直接向量化一整本书（10万字），向量维度过高且检索精度低；分块后（如每200字一块），可精准匹配用户查询的局部内容。
分块方法与工具：按长度分块：用LangChain的RecursiveCharacterTextSplitter按字符数分割（如每500字一块）。

按语义分块：进阶可结合NLP检测段落边界（如用spaCy识别句子）。

代码示例（初学者可直接运行）：

from langchain.text_splitter import RecursiveCharacterTextSplitter
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50  # 重叠部分避免语义断裂
)
chunks = text_splitter.split_text(long_document)

2. 向量化：文本转数字的“魔法”

概念：将文本转换为向量（如1536维数组），通过余弦相似度计算文本相关性。
技术选择：基础方案：用OpenAI Embedding API（简单易用，适合快速验证）。本地方案：Hugging Face的sentence-transformers模型（如all-MiniLM-L6-v2，本地运行节省成本）。
实例：文本“Python是动态类型语言” → 向量化后得到向量A；查询“Python类型系统” → 向量化得到向量B；计算A和B的余弦相似度，若高于阈值则认为相关。

四、向量数据库：存储与检索的“大脑”

1. 为什么用向量数据库？

传统数据库vs向量数据库：传统数据库（如MySQL）：存储文本字符串，无法直接计算语义相似度。向量数据库：存储向量+元数据，支持高效语义检索（如10万条数据检索时间<1秒）。

2. 适合初学者的向量数据库

Chroma	纯本地部署，零配置，Python友好	不支持分布式，适合小规模	入门首选，本地测试数据量<10万
Pinecone	云端托管，开箱即用	按流量计费	快速搭建线上Demo
Milvus	支持大规模数据，开源	部署稍复杂	数据量>100万时考虑

3. 代码示例：用Chroma存储向量

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings

# 初始化向量化模型
embeddings = OpenAIEmbeddings()
# 创建向量数据库（本地存储在chroma_db文件夹）
vector_db = Chroma.from_texts(
    texts=chunks,  # 分块后的文本列表
    embedding=embeddings,
    persist_directory="./chroma_db"
)
vector_db.persist()  # 持久化存储，下次可直接加载

五、搜索和代码实现：从查询到结果

1. 检索核心：如何找到“最相关”的文档？

技术步骤：查询向量化 → 2. 计算与数据库中所有向量的相似度 → 3. 按相似度排序，取Top-K结果。
优化技巧：关键词过滤：先通过关键词筛选初步缩小范围（如查询“Python安装”时，先过滤包含“Python”的文档块）。结果重排序：用大模型对检索结果再排序（进阶，提升准确性）。

2. 实战代码：用LangChain实现检索

from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainCompressor
from langchain.chat_models import ChatOpenAI

# 加载已保存的向量数据库
embeddings = OpenAIEmbeddings()
vector_db = Chroma(
    persist_directory="./chroma_db", 
    embedding_function=embeddings
)

# 创建检索器（带压缩优化，减少冗余信息）
llm = ChatOpenAI(temperature=0)
compressor = LLMChainCompressor.from_llm(llm)
retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vector_db.as_retriever()
)

# 执行查询
query = "如何解决Python模块导入错误？"
docs = retriever.get_relevant_documents(query)
print(f"检索到{len(docs)}篇相关文档，前1篇内容：{docs[0].page_content[:200]}...")

六、DeepSeek部署：本地运行大模型

1. 为什么选择本地部署？

优势：数据安全（本地处理，不泄露到云端）；成本低（一次性下载模型，无需按token付费）；断网可用（适合企业内网场景）。

2. 本地部署步骤（以DeepSeek-LLM为例）

环境准备： Python 3.8+，显卡NVIDIA GPU（至少8GB显存，或用CPU模式但速度慢）。安装依赖：pip install transformers accelerate deepseek-coder。
下载模型：小模型：deepseek-llm-7b-chat（约13GB，适合本地测试），用Hugging Face的from_pretrained加载。
简单调用示例：

from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline

model_name = "deepseek-ai/deepseek-llm-7b-chat"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(
    model_name,
    device_map="auto",  # 自动分配到GPU
    trust_remote_code=True
)
generator = pipeline("text-generation", model=model, tokenizer=tokenizer)

# 结合RAG检索结果生成回答
rag_response = generator(
    f"根据以下文档回答问题：{docs[0].page_content}\n问题：{query}",
    max_new_tokens=200
)
print(rag_response[0]["generated_text"])

3. 初学者注意事项

内存不足解决方案：启用gradient_checkpointing减少显存占用；选择更小的模型（如deepseek-llm-3b）。

七、调用CMD终端&搭建网站：让RAG“看得见摸得着”

1. CMD调用：命令行快速测试

脚本示例（保存为rag_query.py）：

import sys
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 加载向量数据库和LLM
embeddings = OpenAIEmbeddings()
vector_db = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
llm = ChatOpenAI(temperature=0)

# 创建RAG链
rag_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vector_db.as_retriever()
)

# 从命令行获取查询
query = sys.argv[1]
result = rag_chain.run(query)
print(result)

命令行调用：python rag_query.py "Python如何读取CSV文件"

2. 搭建网站：用Streamlit快速实现UI

优势：Streamlit无需前端知识，几行代码生成交互界面。
代码示例（保存为app.py）：

import streamlit as st
from langchain.vectorstores import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA

# 页面设置
st.title("RAG知识库问答系统")

# 加载模型（仅首次运行时加载，后续缓存）
@st.cache_resource
def load_rag_chain():
    embeddings = OpenAIEmbeddings()
    vector_db = Chroma(persist_directory="./chroma_db", embedding_function=embeddings)
    llm = ChatOpenAI(temperature=0)
    return RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=vector_db.as_retriever())

rag_chain = load_rag_chain()

# 用户输入框
query = st.text_input("请输入问题：", placeholder="例如：如何安装Python库？")
if query:
    with st.spinner("正在检索和生成回答..."):
        result = rag_chain.run(query)
        st.success(result)
        
        # 显示检索到的文档（可选）
        st.expander("查看参考文档").write(rag_chain.combine_documents_chain.llm_chain.prompt.input_variables)

运行方式：streamlit run app.py，浏览器打开http://localhost:8501即可使用。

八、Python代码整合：从模块到完整系统

1. 完整RAG系统代码框架（初学者可分步实现）

# 0. 导入必要库
import os
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings  # 或本地模型
from langchain.vectorstores import Chroma
from langchain.document_loaders import DirectoryLoader  # 加载文件夹中文档
from langchain.chat_models import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

# 1. 加载文档（以加载本地文件夹为例）
def load_documents(directory):
    loader = DirectoryLoader(directory, glob="**/*.txt")  # 可扩展为PDF/Word等
    documents = loader.load()
    print(f"成功加载{len(documents)}篇文档")
    return documents

# 2. 分块处理
def split_documents(documents, chunk_size=500, chunk_overlap=50):
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap
    )
    chunks = text_splitter.split_documents(documents)
    print(f"分块后得到{len(chunks)}个文本块")
    return chunks

# 3. 构建向量数据库（首次运行时执行）
def build_vector_db(chunks, embeddings, persist_dir="./chroma_db"):
    vector_db = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=persist_dir
    )
    vector_db.persist()
    print("向量数据库构建完成并持久化")
    return vector_db

# 4. 加载已存在的向量数据库（非首次运行时）
def load_vector_db(embeddings, persist_dir="./chroma_db"):
    if os.path.exists(persist_dir):
        vector_db = Chroma(
            persist_directory=persist_dir,
            embedding_function=embeddings
        )
        print("成功加载已有向量数据库")
        return vector_db
    else:
        print("向量数据库不存在，请先构建")
        return None

# 5. 创建RAG问答链
def create_rag_chain(vector_db, llm):
    # 自定义提示词（提升回答质量）
    prompt_template = """
    你是一个专业的问答助手，请根据以下提供的参考文档回答用户的问题。
    如果参考文档中没有相关信息，你可以说“我无法从知识库中找到相关信息”。
    
    参考文档：{context}
    问题：{question}
    回答：
    """
    PROMPT = PromptTemplate(
        template=prompt_template,
        input_variables=["context", "question"]
    )
    
    rag_chain = RetrievalQA.from_chain_type(
        llm=llm,
        chain_type="stuff",
        retriever=vector_db.as_retriever(),
        chain_type_kwargs={"prompt": PROMPT}
    )
    return rag_chain

# 6. 主函数（整合全流程）
def main():
    # 初始化向量化模型（可替换为本地模型）
    embeddings = OpenAIEmbeddings()
    # 初始化LLM
    llm = ChatOpenAI(temperature=0, model_name="gpt-3.5-turbo")  # 可替换为DeepSeek等本地模型
    
    # 检查是否已构建向量数据库
    vector_db = load_vector_db(embeddings)
    if vector_db is None:
        # 首次运行，加载并分块文档
        documents = load_documents("./knowledge_docs")  # 替换为你的文档路径
        chunks = split_documents(documents)
        vector_db = build_vector_db(chunks, embeddings)
    
    # 创建RAG链
    rag_chain = create_rag_chain(vector_db, llm)
    
    # 交互式问答
    while True:
        query = input("\n请输入问题（输入'退出'结束）：")
        if query.lower() == "退出":
            break
        result = rag_chain.run(query)
        print("回答：", result)

if __name__ == "__main__":
    main()

2. 代码亮点与初学者实践建议

模块化设计：将流程拆分为加载、分块、向量化等独立函数，便于调试和扩展。
持久化存储：向量数据库仅需构建一次，后续直接加载，节省时间。
自定义提示词：通过prompt_template引导大模型更准确地利用检索结果，减少幻觉。
实践路径：先在本地用小数据集（如10篇文档）跑通全流程；替换为OpenAI API快速验证效果；熟练后尝试本地部署DeepSeek模型，实现完全本地化的RAG系统。

九、初学者落地RAG的核心思路

从小做起：先构建一个针对特定领域的小知识库（如个人学习笔记），再扩展到企业级应用。
工具优先：用LangChain等框架封装底层逻辑，聚焦业务场景而非底层实现。
关注效果：重点优化“检索准确率”（如调整分块大小、向量模型）和“回答相关性”（通过提示词工程）。
常见问题解决：检索结果无关：检查分块是否过短/过长，或向量化模型是否匹配领域（如法律文档用专业领域Embedding）。回答重复：在提示词中加入“避免重复”“简洁回答”等指令。