前言

大模型的知识截止于训练数据的时间点，而且不知道你公司内部的产品文档、技术手册、规章制度。

RAG（Retrieval-Augmented Generation，检索增强生成）解决了这个问题：先把你的文档”喂”给系统，用户提问时，系统先检索最相关的文档片段，再把片段和问题一起发给大模型。这样模型就能基于你的私有知识来回答了。

这篇文章用最小的代码量，跑通一个完整的 RAG 流程。

一、RAG 的核心流程

1
2
3
4
5
文档 → 切片 → Embedding → 向量数据库
                              ↓ (用户提问时)
用户提问 → 向量化 → 相似度检索 → Top-K 文档片段
                                      ↓
                          拼进 Prompt → 发给 LLM → 返回答案

四个关键步骤：

  1. 加载文档：读入 PDF、TXT、Markdown、网页等
  2. 文本切片：大文档切成小段落（chunk）
  3. 向量化存储：把每个 chunk 转成向量，存入向量数据库
  4. 检索 + 生成：查询时检索最相关的 chunk，拼进 prompt 发给 LLM

二、安装依赖

1
2
pip install langchain langchain-community langchain-openai
pip install chromadb pypdf unstructured

三、加载文档

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from langchain_community.document_loaders import (
    TextLoader,           # .txt
    PyPDFLoader,          # .pdf
    UnstructuredMarkdownLoader,  # .md
    CSVLoader,            # .csv
)

# 加载单个文件
loader = TextLoader("docs/产品手册.txt", encoding="utf-8")
documents = loader.load()
print(f"加载了 {len(documents)} 个文档")

# 加载整个目录
from langchain_community.document_loaders import DirectoryLoader

loader = DirectoryLoader("docs/", glob="**/*.md", loader_cls=UnstructuredMarkdownLoader)
documents = loader.load()
print(f"共加载 {len(documents)} 个文档")

四、文本切片

文档太长不能直接塞给 Embedding 模型，需要切成小段落（chunk）。

1
2
3
4
5
6
7
8
9
10
11
12
13
from langchain.text_splitter import RecursiveCharacterTextSplitter

text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=800,         # 每块最多 800 个字符
    chunk_overlap=150,      # 相邻块重叠 150 个字符（避免关键信息被切断）
    separators=["\n\n", "\n", "。", "，", " ", ""]  # 优先按段落切，再按句子切
)

chunks = text_splitter.split_documents(documents)
print(f"切成了 {len(chunks)} 个文本块")
for i, chunk in enumerate(chunks[:3]):
    print(f"--- Chunk {i} ({len(chunk.page_content)} 字符) ---")
    print(chunk.page_content[:200])

切片参数调优

参数 太小 太大 推荐范围
chunk_size 缺少上下文，回答碎片化 检索精度下降，噪音多 500~1500
chunk_overlap 冗余多，存储和检索开销大 关键信息在边界被切断 chunk_size 的 15~25%

五、向量化与存储

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma

# 使用 OpenAI 的 Embedding 模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    openai_api_key="sk-xxx"
)

# 如果 OpenAI 不可用，可以换成国产模型
# pip install langchain-huggingface
# from langchain_huggingface import HuggingFaceEmbeddings
# embeddings = HuggingFaceEmbeddings(
#     model_name="BAAI/bge-small-zh-v1.5",  # 智源中文 Embedding
#     model_kwargs={"device": "cpu"}
# )

# 向量化所有 chunk 并存入 ChromaDB
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db"   # 持久化到磁盘
)
print(f"向量数据库已创建，共 {len(chunks)} 条记录")

ChromaDB 是一个轻量级的向量数据库，适合单机使用。数据存在 ./chroma_db/ 目录，下次启动可以直接加载：

1
2
3
4
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings
)

六、检索 + 回答

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA

# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-4o-mini",     # 或换成 deepseek-chat / qwen-turbo
    temperature=0.3,
    openai_api_key="sk-xxx"
)

# 创建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",      # stuff: 把所有检索结果拼进 prompt
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),  # 返回最相似的 4 块
    return_source_documents=True  # 返回引用来源
)

# 提问
question = "我们的产品支持哪些支付方式？"
result = qa_chain.invoke({"query": question})

print("回答:", result["result"])
print("\n引用来源:")
for doc in result["source_documents"]:
    print(f"  - {doc.page_content[:80]}...")

七、一个完整的 RAG 工具类

把上面的流程封装成一个可复用的类：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
from langchain_community.document_loaders import DirectoryLoader, UnstructuredMarkdownLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain.chains import RetrievalQA
import os

class RAG:
    def __init__(self, docs_dir="./docs", db_dir="./chroma_db"):
        self.docs_dir = docs_dir
        self.db_dir = db_dir
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
        
        # 如果向量库已存在，直接加载；否则创建
        if os.path.exists(db_dir) and os.listdir(db_dir):
            self.vectorstore = Chroma(
                persist_directory=db_dir,
                embedding_function=self.embeddings
            )
        else:
            self.vectorstore = None
    
    def build_index(self):
        """构建/重建向量索引"""
        loader = DirectoryLoader(self.docs_dir, glob="**/*.md",
                                 loader_cls=UnstructuredMarkdownLoader)
        documents = loader.load()
        
        text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=800, chunk_overlap=150
        )
        chunks = text_splitter.split_documents(documents)
        
        self.vectorstore = Chroma.from_documents(
            documents=chunks,
            embedding=self.embeddings,
            persist_directory=self.db_dir
        )
        print(f"索引构建完成: {len(chunks)} 个文本块")
    
    def ask(self, question, k=4):
        """提问"""
        if self.vectorstore is None:
            raise ValueError("请先调用 build_index()")
        
        qa = RetrievalQA.from_chain_type(
            llm=self.llm,
            chain_type="stuff",
            retriever=self.vectorstore.as_retriever(
                search_kwargs={"k": k}
            ),
            return_source_documents=True
        )
        result = qa.invoke({"query": question})
        return {
            "answer": result["result"],
            "sources": [doc.page_content[:100] for doc in result["source_documents"]]
        }


# ===== 使用 =====
rag = RAG(docs_dir="./my_knowledge_base")
rag.build_index()

print(rag.ask("公司年假政策是什么？")["answer"])

# 每次有新文档，重新 build_index 即可

八、效果调优经验

检索效果不好？

  • 调整 chunk_size 和 chunk_overlap
  • 换更强的 Embedding 模型（如 bge-large-zh-v1.5 替换 bge-small）
  • 尝试多种检索策略混合（关键词 BM25 + 向量检索）
  • 对文档做预处理：去掉页眉页脚、表格转文字

回答效果不好？

  • 调低 temperature（0.1~0.3），减少模型”自由发挥”
  • 在 system prompt 里强调”如果文档中没有相关信息，就说不知道”
  • 增加 k 值（返回更多 chunk，但同时 prompt 会变长）
  • 考虑对检索到的 chunk 做重排序（reranker 模型）

效果不错但太慢？

  • 换用更小的 LLM（GPT-4o-mini 替代 GPT-4o）
  • 用 GPU 跑 Embedding 模型
  • 减少每次检索的 chunk 数量

九、更高级的 RAG

这篇文章走通了最基础的 RAG 流程。生产环境中还会用到：

  • Reranker：对检索结果做二次排序，提高相关性
  • 多轮对话 RAG：支持追问，需要处理对话历史和上下文
  • 混合检索：向量检索 + 关键词检索，互补不足
  • 图谱 RAG：不是检索文本块，而是构建知识图谱后推理

以上跑通了一整个 RAG 管道。把 ./my_knowledge_base/ 里的文件换成你自己的文档，就能用了。