背景介绍

大语言模型很强，但这东西有个致命问题：它懂的东西全是训练数据里的。你公司内部的文档、产品手册、个人笔记——这些私有数据它根本看不到。

这就是 RAG（检索增强生成）要解决的问题。简单说就是把私有数据先存到向量数据库里，用户提问的时候，系统去数据库里找相关内容，然后把找到的内容和问题一起喂给大模型，让它基于真实资料来回答。

这篇文章手把手教你在本地跑通一个知识库问答系统。技术栈是 LangChain + ChromaDB + OpenAI。整个过程数据都在本地，不用担心泄露问题。

问题描述

实际做的时候会遇到几个坑：

文档格式杂。 PDF、Word、Markdown、TXT，每种格式处理方式都不一样。你得有个统一的加载器来处理这些。

关键词搜索太蠢。 用户搜“怎么安装软件”，传统搜索可能匹配不到“安装步骤”、“部署方法”这些同义表达。向量检索通过把文本转成数学向量，能理解语义相似性，这才是正确的打开方式。

怎么在本地搭一个能用的系统。 既要保证数据安全，又要问答体验过得去，需要一套完整的技术方案。

下文会一个一个解决这些问题。

详细步骤

1. 环境准备

装依赖：

python -m venv venv
source venv/bin/activate  # Windows 下用 venv\Scripts\activate

pip install langchain langchain-openai chromadb pypdf python-docx 
pip install markdown tiktoken numpy

langchain 搭 RAG 框架；langchain-openai 连 OpenAI；chroma 存向量；pypdf 和 python-docx 分别处理 PDF 和 Word。

2. 文档加载器

不同格式用不同的加载器：

from langchain_community.document_loaders import (
    PyPDFLoader, 
    TextLoader, 
    Docx2txtLoader,
    MarkdownLoader
)
from langchain.text_splitter import RecursiveCharacterTextSplitter
from pathlib import Path
import os

class DocumentProcessor:
    def __init__(self, chunk_size=500, chunk_overlap=50):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            length_function=len,
        )
    
    def load_document(self, file_path):
        """根据文件扩展名选择加载器"""
        path = Path(file_path)
        suffix = path.suffix.lower()
        
        if suffix == ".pdf":
            loader = PyPDFLoader(file_path)
        elif suffix == ".docx":
            loader = Docx2txtLoader(file_path)
        elif suffix == ".md":
            loader = MarkdownLoader(file_path)
        elif suffix == ".txt":
            loader = TextLoader(file_path, encoding="utf-8")
        else:
            raise ValueError(f"不支持的文件类型: {suffix}")
        
        return loader.load()
    
    def process_directory(self, directory):
        """处理目录下所有文档"""
        documents = []
        for root, _, files in os.walk(directory):
            for file in files:
                file_path = os.path.join(root, file)
                try:
                    docs = self.load_document(file_path)
                    documents.extend(docs)
                    print(f"已加载: {file_path}, 页面数: {len(docs)}")
                except Exception as e:
                    print(f"加载失败 {file_path}: {e}")
        
        # 文本分块
        chunks = self.text_splitter.split_documents(documents)
        print(f"共生成 {len(chunks)} 个文本块")
        return chunks

支持 PDF、Word、Markdown、纯文本四种格式，分块处理。块大小根据实际情况调，太小丢上下文，太大引入噪声。

3. 向量存储与检索

ChromaDB 轻量级向量数据库，本地部署正合适：

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

class VectorStoreManager:
    def __init__(self, persist_directory="./chroma_db", use_openai=True):
        self.persist_directory = persist_directory
        self.use_openai = use_openai
        
        if use_openai:
            # 用 OpenAI 的 Embedding
            self.embeddings = OpenAIEmbeddings(
                model="text-embedding-3-small"
            )
        else:
            # 用本地模型，完全离线
            self.embeddings = HuggingFaceEmbeddings(
                model_name="sentence-transformers/all-MiniLM-L6-v2"
            )
        
        self.vectorstore = None
    
    def create_vectorstore(self, documents):
        """从文档创建向量存储"""
        self.vectorstore = Chroma.from_documents(
            documents=documents,
            embedding=self.embeddings,
            persist_directory=self.persist_directory
        )
        print(f"向量数据库已保存到: {self.persist_directory}")
        return self.vectorstore
    
    def load_vectorstore(self):
        """加载已存在的向量数据库"""
        self.vectorstore = Chroma(
            persist_directory=self.persist_directory,
            embedding_function=self.embeddings
        )
        return self.vectorstore
    
    def similarity_search(self, query, k=4):
        """相似性搜索"""
        if self.vectorstore is None:
            raise ValueError("向量数据库未初始化")
        
        results = self.vectorstore.similarity_search(
            query=query,
            k=k
        )
        return results
    
    def similarity_search_with_score(self, query, k=4):
        """带相关性分数的搜索"""
        if self.vectorstore is None:
            raise ValueError("向量数据库未初始化")
        
        results = self.vectorstore.similarity_search_with_score(
            query=query,
            k=k
        )
        return results

两套方案：OpenAI 的 Embedding 效果好但要 API 密钥；HuggingFace 本地模型能离线跑，效果稍差。

4. 问答系统核心

把零件组装成完整的问答系统：

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate

class RAGQASystem:
    def __init__(
        self, 
        vectorstore,
        model_name="gpt-3.5-turbo",
        openai_api_key=None,
        temperature=0.3
    ):
        self.vectorstore = vectorstore
        
        # 初始化 LLM
        self.llm = ChatOpenAI(
            model_name=model_name,
            openai_api_key=openai_api_key,
            temperature=temperature
        )
        
        # 自定义提示词
        self.prompt_template = """使用以下上下文信息来回答问题。如果上下文中没有相关信息，直接说找不到。

上下文信息：
{context}

问题：{question}

回答："""
        
        self.qa_chain = None
    
    def build_chain(self):
        """构建问答链"""
        prompt = PromptTemplate(
            template=self.prompt_template,
            input_variables=["context", "question"]
        )
        
        self.qa_chain = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": 4}
            ),
            chain_type_kwargs={"prompt": prompt},
            return_source_documents=True
        )
        
        return self.qa_chain
    
    def ask(self, question):
        """提问"""
        if self.qa_chain is None:
            self.build_chain()
        
        result = self.qa_chain.invoke({"query": question})
        
        return {
            "answer": result["result"],
            "source_documents": result["source_documents"]
        }
    
    def ask_with_sources(self, question):
        """提问并显示来源"""
        result = self.ask(question)
        
        print(f"\n问题: {question}\n")
        print(f"回答: {result[\"answer\"]}\n")
        print("参考来源:")
        for i, doc in enumerate(result["source_documents"], 1):
            source = doc.metadata.get("source", "未知")
            print(f"  {i}. {source}")
            print(f"     内容摘要: {doc.page_content[:100]}...")
        
        return result

用了 RetrievalQA 链，自动从向量库检索相关内容，和问题一起发给 LLM。还加了返回源文档的功能，方便你核对答案。

5. 完整使用示例

import os

def main():
    # 配置 API 密钥
    os.environ["OPENAI_API_KEY"] = "your-api-key-here"
    
    # 1. 准备文档目录
    docs_directory = "./knowledge_base"
    
    # 2. 加载处理文档
    processor = DocumentProcessor(chunk_size=500, chunk_overlap=50)
    chunks = processor.process_directory(docs_directory)
    
    # 3. 创建向量数据库
    vector_manager = VectorStoreManager(
        persist_directory="./my_vectorstore",
        use_openai=True
    )
    vectorstore = vector_manager.create_vectorstore(chunks)
    
    # 4. 初始化问答系统
    qa_system = RAGQASystem(
        vectorstore=vectorstore,
        model_name="gpt-3.5-turbo",
        openai_api_key=os.environ["OPENAI_API_KEY"],
        temperature=0.3
    )
    
    # 5. 开始问答
    questions = [
        "什么是 RAG？",
        "如何安装 LangChain？",
        "向量数据库的作用是什么？"
    ]
    
    for question in questions:
        qa_system.ask_with_sources(question)
        print("-" * 50)

if __name__ == "__main__":
    main()

运行结果

跑起来之后，先看到文档加载和分块的过程：

已加载: ./knowledge_base/rag_intro.pdf, 页面数: 15
已加载: ./knowledge_base/langchain_guide.md, 页面数: 8
已加载: ./knowledge_base/vector_db.txt, 页面数: 3
共生成 127 个文本块
向量数据库已保存到: ./my_vectorstore

然后是问答输出：

问题: 什么是 RAG？

回答: RAG（检索增强生成）是一种将外部知识检索与大语言模型生成能力相结合的技术架构。它先在向量数据库中搜索与用户问题相关的文档，然后把检索到的内容作为上下文提供给 LLM，从而生成更加准确、基于事实的回答。RAG 有效解决了 LLM 知识过时和幻觉的问题。

参考来源:
  1. ./knowledge_base/rag_intro.pdf
     内容摘要: RAG（Retrieval-Augmented Generation）是一种结合检索系统和生成模型的技术...
  2. ./knowledge_base/langchain_guide.md
     内容摘要: 在 LangChain 中实现 RAG 主要涉及以下几个组件...

系统准确理解问题，从相关文档提取信息作答。参考来源显示了引用的原始文档，可以进一步核对。

总结

这篇文章搞定了这几个事：

文档处理。 PDF、Word、Markdown、纯文本都能加载，分块处理。

向量检索。 ChromaDB 存向量，语义搜索能理解“安装”和“部署”的关联。

问答生成。 GPT 模型结合检索到的上下文，生成基于事实的回答。

系统优点：数据全本地，安全性有保障；支持多格式；提示词可定制；能看到答案来源。

后续可以优化的地方：调分块大小找最优平衡；用本地 Embedding 提升速度；加对话历史支持多轮问答；对接企业认证做权限控制。

RAG 让大模型能用上私有数据，智能客服、知识管理、内容创作这些场景都能用上。随着技术发展，这东西会越来越常见。