RAG 系统实战：从零搭建检索增强生成应用

安装依赖

创建虚拟环境后安装以下包。faiss-cpu 是纯 CPU 版本的向量相似度搜索库，无需 GPU。

# 创建并激活虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 安装核心依赖
pip install langchain langchain-openai langchain-community \
            faiss-cpu pymupdf tiktoken openai python-dotenv

如果你想完全本地运行（不依赖 OpenAI），可以改用 Ollama：

# 本地 LLM 方案
pip install langchain-ollama
# 然后本地启动：ollama run qwen2.5:7b

配置 API Key

在项目根目录创建 .env 文件，放置你的 OpenAI API Key：

# .env
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxx
# 如果用国内中转服务
# OPENAI_BASE_URL=https://your-proxy.com/v1

实现文档加载器

创建 app/document_loader.py，支持 TXT 和 PDF 两种格式：

# app/document_loader.py
from pathlib import Path
from langchain_community.document_loaders import TextLoader, PyMuPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.documents import Document
from typing import List


def load_documents(data_dir: str = "data") -> List[Document]:
    """加载 data/ 目录下的所有文档"""
    docs = []
    data_path = Path(data_dir)

    for file_path in data_path.rglob("*"):
        if file_path.suffix == ".txt":
            loader = TextLoader(str(file_path), encoding="utf-8")
            docs.extend(loader.load())
        elif file_path.suffix == ".pdf":
            loader = PyMuPDFLoader(str(file_path))
            docs.extend(loader.load())

    print(f"✅ 共加载 {len(docs)} 个文档片段")
    return docs


def split_documents(docs: List[Document]) -> List[Document]:
    """将文档分割为适合检索的小块"""
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=800,        # 每块最多 800 字符
        chunk_overlap=100,     # 相邻块重叠 100 字符（保留上下文）
        separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""],
    )
    chunks = splitter.split_documents(docs)
    print(f"✅ 分割为 {len(chunks)} 个文本块")
    return chunks

创建向量存储

创建 app/vector_store.py，封装 FAISS 索引的创建和持久化：

# app/vector_store.py
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_core.documents import Document
from typing import List
import os


def get_embeddings():
    """返回 Embedding 模型实例"""
    return OpenAIEmbeddings(
        model="text-embedding-3-small",  # 比 ada-002 更便宜且更好
        openai_api_key=os.environ.get("OPENAI_API_KEY"),
    )


def build_vector_store(chunks: List[Document], index_path: str = "faiss_index"):
    """从文本块构建 FAISS 向量索引并持久化到磁盘"""
    embeddings = get_embeddings()

    print("⏳ 正在生成向量嵌入，请稍候...")
    vector_store = FAISS.from_documents(chunks, embeddings)
    vector_store.save_local(index_path)

    print(f"✅ 向量索引已保存至 {index_path}/")
    return vector_store


def load_vector_store(index_path: str = "faiss_index"):
    """从磁盘加载已有的 FAISS 索引"""
    embeddings = get_embeddings()
    vector_store = FAISS.load_local(
        index_path,
        embeddings,
        allow_dangerous_deserialization=True,  # 本地文件信任即可
    )
    print(f"✅ 已从 {index_path}/ 加载向量索引")
    return vector_store

实现 RAG 主链路

创建 app/rag_chain.py，使用 LangChain Expression Language（LCEL）组合各组件：

# app/rag_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import FAISS
import os


# 系统 Prompt：严格限制模型只使用检索到的上下文
RAG_PROMPT = ChatPromptTemplate.from_messages([
    ("system", """你是一个专业的知识库问答助手。请严格遵守以下规则：

1. 只根据下方提供的【参考文档】来回答问题
2. 如果参考文档中没有相关信息，直接说"根据现有文档，我无法回答这个问题"
3. 引用信息时注明来源（如"根据第X部分..."）
4. 回答要简洁准确，不要过度解读文档内容

【参考文档】
{context}"""),
    ("human", "{question}"),
])


def format_docs(docs):
    """将检索到的文档列表格式化为字符串"""
    formatted = []
    for i, doc in enumerate(docs, 1):
        source = doc.metadata.get("source", "未知来源")
        formatted.append(f"[文档{i}] 来源：{source}\n{doc.page_content}")
    return "\n\n---\n\n".join(formatted)


def create_rag_chain(vector_store: FAISS, k: int = 4):
    """
    创建 RAG 链路
    k: 检索最相似的 k 个文本块
    """
    # 1. 配置检索器
    retriever = vector_store.as_retriever(
        search_type="mmr",          # MMR 算法：兼顾相关性和多样性
        search_kwargs={"k": k, "fetch_k": 20},
    )

    # 2. 配置语言模型
    llm = ChatOpenAI(
        model="gpt-4o-mini",        # 性价比最高的选择
        temperature=0,              # 设为 0：减少随机性，提升一致性
        openai_api_key=os.environ.get("OPENAI_API_KEY"),
    )

    # 3. 用 LCEL 组装链路
    rag_chain = (
        {
            "context": retriever | format_docs,   # 检索 → 格式化
            "question": RunnablePassthrough(),    # 用户问题直接传递
        }
        | RAG_PROMPT
        | llm
        | StrOutputParser()
    )

    return rag_chain, retriever

创建主程序入口

创建 main.py，整合所有模块：

# main.py
import os
from pathlib import Path
from dotenv import load_dotenv
from app.document_loader import load_documents, split_documents
from app.vector_store import build_vector_store, load_vector_store
from app.rag_chain import create_rag_chain

load_dotenv()

INDEX_PATH = "faiss_index"


def setup():
    """首次运行：构建向量索引"""
    print("🚀 首次运行，开始构建知识库...")
    docs = load_documents("data")
    chunks = split_documents(docs)
    vector_store = build_vector_store(chunks, INDEX_PATH)
    return vector_store


def main():
    # 判断是否已有索引
    if Path(INDEX_PATH).exists():
        vector_store = load_vector_store(INDEX_PATH)
    else:
        vector_store = setup()

    rag_chain, retriever = create_rag_chain(vector_store, k=4)

    print("\n✅ 知识库就绪！输入 'quit' 退出\n")
    print("=" * 50)

    while True:
        question = input("\n❓ 你的问题：").strip()
        if question.lower() in ("quit", "exit", "q"):
            break
        if not question:
            continue

        print("\n🔍 正在检索相关文档...")
        # 也可以单独查看检索到了哪些文档
        retrieved_docs = retriever.invoke(question)
        print(f"   找到 {len(retrieved_docs)} 个相关文档块")

        print("\n🤖 正在生成答案...\n")
        answer = rag_chain.invoke(question)
        print(f"📝 答案：\n{answer}")
        print("\n" + "=" * 50)


if __name__ == "__main__":
    main()