👨 作者简介：大家好，我是唐璜Taro，全栈 领域创作者
✒️ 个人主页 ：唐璜Taro
🚀 支持我：点赞👍+📝 评论 + ⭐️收藏

---

上一篇讲述了RAG（Retrieval-Augmented Generation）的理论以及应用场景，这一章节讲解RAG的核心实现。

四、核心实现

4.1 文档加载

支持多种格式的文档读取：

from langchain_community.document_loaders import (
    PyPDFLoader,        # PDF
    Docx2txtLoader,     # Word
    TextLoader,         # TXT
    CSVLoader,          # CSV
    UnstructuredMarkdownLoader,  # Markdown
)

# 加载单个文件
loader = PyPDFLoader("公司制度手册.pdf")
documents = loader.load()

# 批量加载目录下所有文件
from langchain_community.document_loaders import DirectoryLoader

loader = DirectoryLoader(
    "./docs",
    glob="**/*.pdf",
    loader_cls=PyPDFLoader,
    show_progress=True  # 显示进度条
)
documents = loader.load()

print(f"共加载 {len(documents)} 页文档")
print(f"第一页内容预览：{documents[0].page_content[:200]}")

中文文档注意：PDF 中文提取可能乱码，推荐先转成 Markdown 或 TXT。

4.2 文本分块（Chunking）

这是 RAG 中最关键的环节之一。分块质量直接影响检索效果。

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,       # 每块最大字符数
    chunk_overlap=50,     # 块之间重叠字符数
    length_function=len,
    separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
    # 中文优化：优先按段落 → 句号 → 换行 → 空格 切分
)

chunks = splitter.split_documents(documents)
print(f"共生成 {len(chunks)} 个文本块")

分块参数怎么调？

参数	值太小	值太大	推荐范围
chunk_size	语义碎片化，丢失上下文	检索不精确，噪音多	300 - 1000 字符
chunk_overlap	块之间语义断裂	重复内容多，浪费空间	50 - 150 字符

经验法则：

• 问答场景：chunk_size 小一些（300-500），检索更精确
• 总结场景：chunk_size 大一些（800-1000），保留更多上下文

4.3 向量化（Embedding）

将文本转成高维向量，用于后续相似度计算：

from langchain_community.embeddings import HuggingFaceEmbeddings

# 首次运行会自动下载模型（约 100MB）
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",
    model_kwargs={"device": "cpu"},       # 无 GPU 用 cpu
    encode_kwargs={"normalize_embeddings": True}  # 归一化，提高余弦相似度效果
)

# 测试：把一段文字转成向量
vector = embeddings.embed_query("什么是退货政策？")
print(f"向量维度：{len(vector)}")  # 512 维
print(f"前5个值：{vector[:5]}")

Embedding 模型对比

模型	维度	中文效果	大小	说明
BAAI/bge-small-zh	512	好	~100MB	推荐入门用
BAAI/bge-large-zh	1024	更好	~1.3GB	精度更高
text-embedding-3-small	1536	好	API 调用	OpenAI 付费
text-embedding-3-large	3072	很好	API 调用	OpenAI 付费，最贵

4.4 向量数据库存储

from langchain_community.vectorstores import Chroma

# 创建并持久化
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",  # 本地存储目录
    collection_name="my_knowledge"    # 集合名称
)

print(f"成功索引 {vectorstore._collection.count()} 个文本块")

# 加载已有数据库（下次运行时不需要重新建库）
# vectorstore = Chroma(
#     persist_directory="./chroma_db",
#     embedding_function=embeddings,
#     collection_name="my_knowledge"
# )

4.5 检索测试

# 基础检索
results = vectorstore.similarity_search("退货政策", k=3)

for i, doc in enumerate(results):
    print(f"\n--- 结果 {i+1} ---")
    print(f"来源：{doc.metadata.get('source', '未知')}")
    print(f"内容：{doc.page_content[:200]}")

4.6 构建 RAG 问答链

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

# 初始化大模型
llm = ChatOpenAI(
    model="deepseek-chat",
    openai_api_key="your-api-key",
    openai_api_base="https://api.deepseek.com",
    temperature=0.1,  # 低温度，回答更稳定
    max_tokens=1024
)

# 自定义 Prompt 模板
prompt_template = """你是一个专业的客服助手。请基于以下参考资料回答用户问题。

规则：
1. 只根据参考资料回答，不要编造信息
2. 如果参考资料中没有相关内容，请回答"根据现有资料，我无法回答这个问题"
3. 回答时注明信息来源

参考资料：
{context}

用户问题：{question}

回答："""

PROMPT = PromptTemplate(
    template=prompt_template,
    input_variables=["context", "question"]
)

# 创建检索器
retriever = vectorstore.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 3}  # 检索最相似的 3 个文本块
)

# 构建 RAG 链
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",           # 把所有检索结果拼接后一次性发送
    retriever=retriever,
    chain_type_kwargs={"prompt": PROMPT},
    return_source_documents=True   # 返回来源文档，方便溯源
)

# 提问
result = qa_chain.invoke({"query": "退货需要满足什么条件？"})
print("回答：", result["result"])
print("\n来源：")
for doc in result["source_documents"]:
    print(f"  - {doc.metadata.get('source', '未知')}")

4.7 运行效果示例

用户提问：退货需要满足什么条件？

回答：根据公司退货政策，退货需要满足以下条件：
1. 商品签收后 7 天内可申请退货
2. 商品需保持原包装完好，不影响二次销售
3. 食品、贴身衣物等特殊商品不支持退货
4. 需提供订单号和购买凭证

来源：
  - docs/售后服务政策.pdf (第3页)
  - docs/常见问题FAQ.txt

五、完整项目结构

rag-knowledge-base/
├── docs/                    # 原始文档目录
│   ├── 售后服务政策.pdf
│   ├── 产品使用手册.docx
│   └── 常见问题FAQ.txt
├── chroma_db/               # 向量数据库（自动生成）
├── build_index.py           # 建库脚本
├── query.py                 # 问答脚本
├── config.py                # 配置文件
└── requirements.txt

config.py — 统一配置

import os

# API 配置
DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "your-api-key")
DEEPSEEK_BASE_URL = "https://api.deepseek.com"

# Embedding 配置
EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5"

# 分块配置
CHUNK_SIZE = 500
CHUNK_OVERLAP = 50

# 检索配置
TOP_K = 3

# 向量数据库配置
CHROMA_DIR = "./chroma_db"
COLLECTION_NAME = "my_knowledge"

# 文档目录
DOCS_DIR = "./docs"

build_index.py — 一键建库

from langchain_community.document_loaders import DirectoryLoader, TextLoader, PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from config import *

def build():
    print("1/4 加载文档...")
    loaders = [
        DirectoryLoader(DOCS_DIR, glob="**/*.txt", loader_cls=TextLoader, show_progress=True),
        DirectoryLoader(DOCS_DIR, glob="**/*.pdf", loader_cls=PyPDFLoader, show_progress=True),
    ]
    documents = []
    for loader in loaders:
        documents.extend(loader.load())
    print(f"  加载了 {len(documents)} 个文档")

    print("2/4 分块处理...")
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=CHUNK_SIZE,
        chunk_overlap=CHUNK_OVERLAP,
        separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""]
    )
    chunks = splitter.split_documents(documents)
    print(f"  生成 {len(chunks)} 个文本块")

    print("3/4 向量化...")
    embeddings = HuggingFaceEmbeddings(
        model_name=EMBEDDING_MODEL,
        model_kwargs={"device": "cpu"},
        encode_kwargs={"normalize_embeddings": True}
    )

    print("4/4 存入数据库...")
    vectorstore = Chroma.from_documents(
        documents=chunks,
        embedding=embeddings,
        persist_directory=CHROMA_DIR,
        collection_name=COLLECTION_NAME
    )
    print(f"完成！共索引 {vectorstore._collection.count()} 个文本块")

if __name__ == "__main__":
    build()

query.py — 问答入口

from langchain_openai import ChatOpenAI
from langchain.chains import RetrievalQA
from langchain.prompts import PromptTemplate
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_community.vectorstores import Chroma
from config import *

def create_qa_chain():
    embeddings = HuggingFaceEmbeddings(
        model_name=EMBEDDING_MODEL,
        model_kwargs={"device": "cpu"},
        encode_kwargs={"normalize_embeddings": True}
    )

    vectorstore = Chroma(
        persist_directory=CHROMA_DIR,
        embedding_function=embeddings,
        collection_name=COLLECTION_NAME
    )

    llm = ChatOpenAI(
        model="deepseek-chat",
        openai_api_key=DEEPSEEK_API_KEY,
        openai_api_base=DEEPSEEK_BASE_URL,
        temperature=0.1
    )

    prompt = PromptTemplate(
        template="""基于以下参考资料回答问题。如果资料中没有相关内容，请回答"无法回答"。

参考资料：{context}

问题：{question}

回答：""",
        input_variables=["context", "question"]
    )

    return RetrievalQA.from_chain_type(
        llm=llm,
        retriever=vectorstore.as_retriever(search_kwargs={"k": TOP_K}),
        chain_type="stuff",
        chain_type_kwargs={"prompt": prompt},
        return_source_documents=True
    )

def main():
    qa = create_qa_chain()
    print("知识库问答系统已启动，输入 quit 退出\n")

    while True:
        question = input("请输入问题：").strip()
        if question.lower() in ("quit", "exit", "q"):
            break
        if not question:
            continue

        result = qa.invoke({"query": question})
        print(f"\n回答：{result['result']}")
        print("来源：", [d.metadata.get("source", "") for d in result["source_documents"]])
        print()

if __name__ == "__main__":
    main()

六、优化技巧

6.1 混合检索（Hybrid Search）

单一向量检索可能漏掉关键词精确匹配的结果。混合检索结合语义检索 + 关键词检索：

from langchain.retrievers import EnsembleRetriever
from langchain_community.retrievers import BM25Retriever

# BM25 关键词检索
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 3

# 向量语义检索
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 3})

# 混合检索（各占 50% 权重）
hybrid_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6]  # BM25 占 40%，向量检索占 60%
)

6.2 Rerank 重排序

检索后对结果重新排序，提高相关性：

# pip install sentence-transformers
from langchain.retrievers import ContextualCompressionRetriever
from langchain_cohere import CohereRerank

# 使用 Cohere Rerank（需要 API Key）
reranker = CohereRerank(model="rerank-multilingual-v3.0", top_n=3)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=reranker,
    base_retriever=vector_retriever
)

6.3 多轮对话

from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationalRetrievalChain

memory = ConversationBufferWindowMemory(
    k=5,  # 保留最近 5 轮对话
    memory_key="chat_history",
    return_messages=True
)

qa_chain = ConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=vectorstore.as_retriever(),
    memory=memory
)

6.4 分块优化策略

策略	方法	适用场景
按语义分块	用 NLP 模型判断语义边界	文章、报告
按固定长度	RecursiveCharacterTextSplitter	通用场景
按文档结构	按标题/章节切分	Markdown、技术文档
递归分块	先大块再小块	长文档

七、常见问题排查

Q1: 检索结果不相关

• 检查 chunk_size 是否合适，太大容易混入噪音
• 尝试不同的 Embedding 模型
• 增加 chunk_overlap 减少语义断裂

Q2: 回答总是说"无法回答"

• 降低 Prompt 中的限制性描述
• 增大 Top_K 值（比如从 3 改到 5）
• 检查文档是否正确加载和分块

Q3: 响应速度慢

• Embedding 模型换用更小的（如 bge-small）
• 使用 GPU 加速：model_kwargs={"device": "cuda"}
• 向量数据库换用 Milvus 等高性能方案

Q4: 中文 PDF 乱码

# 用 OCR 方案
pip install rapidocr-onnxruntime

from langchain_community.document_loaders import PDFPlumberLoader
# 或使用 PaddleOCR 等工具预处理

八、进阶方向

方向	说明
Web UI	用 Streamlit / Gradio 做可视化界面
增量更新	文档变更后只更新变化的部分，不重建全量索引
多模态 RAG	支持图片、表格的检索
Agent + RAG	让模型自主决定是否需要检索、检索什么
生产级部署	Milvus + FastAPI + Redis 缓存

九、总结

RAG 搭建的核心流程：

文档 → 分块 → Embedding → 向量数据库
                              ↓
用户提问 → Embedding → 相似度检索 → Top-K 文档
                                      ↓
                          Prompt(问题 + 文档) → 大模型 → 回答

---

学习路线建议

• 先跑通 — 用 LangChain + Chroma + DeepSeek 跑一个最小 demo
• 优化检索 — 试不同的 chunk_size、overlap、Top-K 值
• 混合检索 — 关键词检索（BM25）+ 向量检索结合，效果更好
• 进阶 — 多轮对话、引用来源标注、Rerank 重排序

---

入门建议：
先把最小 demo 跑通，再逐步优化分块策略、检索方式和 Prompt 模板。RAG 的效果 80% 取决于数据处理和检索质量，而不是模型本身。