RAG 技术系列（上篇）：从零搭建与混合检索

本系列文章基于 LangChain v1.2 生态，通过 9 个循序渐进的完整代码示例，从最基础的 RAG 实现一路演进到企业级多源文档检索方案。上篇聚焦 RAG 的核心流水线与混合检索策略。

---

一、为什么需要 RAG？

大语言模型（LLM）虽然能力强大，但存在两个根本性限制：

1. 知识截止日期：模型训练完成后，无法获知训练数据之后的新信息。
2. 幻觉问题：模型在面对不确定的问题时可能编造看似合理但实际错误的内容。

RAG（Retrieval-Augmented Generation，检索增强生成） 正是为解决这些问题而生。它的核心思想简单而有效：在 LLM 生成回答之前，先从外部知识库中检索相关文档，将检索结果作为上下文注入给模型，让模型"有据可依"地回答问题。

一个完整的 RAG 流水线包含以下环节：

文档加载 → 文本分割 → 向量嵌入 → 向量存储 → 检索召回 → 增强生成

下面我们通过代码一步步构建并优化这套流水线。

---

二、环境准备与依赖安装

2.1 基础环境

本项目使用 Python 3.13+，以 uv 作为包管理工具。以下是本篇文章涉及的核心依赖：

# pyproject.toml (上篇所需依赖)
dependencies = [
    "bs4>=0.0.2",                    # HTML 解析
    "chromadb>=1.5.9",              # 向量数据库底层
    "langchain>=1.2.17",            # LangChain 核心
    "langchain-chroma>=1.1.0",      # Chroma 向量库集成
    "langchain-community>=0.4.1",   # 社区组件（BM25、文档加载器等）
    "langchain-openai>=1.2.1",      # OpenAI 兼容模型接口
    "langchain-text-splitters>=1.1.2", # 文本分割器
    "langgraph>=1.1.10",            # Agent 图引擎
    "rank-bm25>=0.2.2",             # BM25 关键词检索算法
    "python-dotenv",                 # 环境变量管理（隐式依赖）
]

安装命令：

uv sync

2.2 环境变量配置

在项目根目录创建 .env 文件，配置 LLM 和 Embedding 模型的 API 连接信息：

# 大语言模型配置
AL_MODEL_NAME=deepseek-chat        # 替换为你使用的模型名称
AL_BASE_URL=https://api.deepseek.com  # API 地址
AL_API_KEY=sk-your-api-key         # API 密钥

# 嵌入模型配置
AL_EMMODEL_NAME=text-embedding-3-small  # 嵌入模型名称

说明：本项目使用 OpenAI 兼容接口，ChatOpenAI 和 OpenAIEmbeddings 均可对接任何兼容 OpenAI API 协议的服务（如 DeepSeek、阿里百炼、本地 Ollama 等），只需修改 base_url 和 model 参数即可。

---

三、第一版：最简 RAG 实现——打好地基

main.py 是本系列最基础的 RAG 实现，它用不到 110 行代码完整展示了 RAG 全流程。我们从这段代码中可以看到 RAG 系统的骨架结构。

3.1 整体架构

                   ┌──────────────┐
                   │   WebBase    │
                   │   Loader     │  网页抓取 + BS4 过滤
                   └──────┬───────┘
                          │ 原始文档
                          ▼
                   ┌──────────────┐
                   │ Recursive    │
                   │ Character    │  文本分割 (chunk_size=1000, overlap=200)
                   │ Splitter     │
                   └──────┬───────┘
                          │ 文档块
                          ▼
                   ┌──────────────┐
                   │  OpenAI      │
                   │  Embeddings  │  向量化
                   └──────┬───────┘
                          │ 向量
                          ▼
                   ┌──────────────┐
                   │   Chroma     │
                   │  VectorStore │  持久化存储 + 相似度检索
                   └──────┬───────┘
                          │ Top-K 文档
                          ▼
                   ┌──────────────┐
                   │   Agent      │
                   │ (LLM + Tool) │  理解问题 → 调用检索 → 生成回答
                   └──────────────┘

3.2 关键代码逐段解析

(1) 文档加载

import bs4
from langchain_community.document_loaders import WebBaseLoader

# WebBaseLoader：一站式网页加载器，内置 HTTP 请求 + HTML 解析
loader = WebBaseLoader(
    web_path="https://pixle.blog.csdn.net/article/details/156417028?spm=1001.2014.3001.5502",
    # bs4.SoupStrainer：精确过滤 DOM 节点，只提取正文内容区域
    # 排除导航栏、侧边栏、广告等噪音，提升后续检索质量
    bs_kwargs={"parse_only": bs4.SoupStrainer(class_=("blog-content-box"))},
)

这里有两个值得注意的细节：

• WebBaseLoader：LangChain 提供的一站式网页加载器，内部完成 HTTP 请求和 HTML 解析。
• bs4.SoupStrainer：仅提取 class="blog-content-box" 的 DOM 元素，过滤掉导航栏、侧边栏、广告等噪音内容。这在实际项目中至关重要——垃圾进垃圾出，文档清洗直接影响检索质量。

(2) 嵌入模型初始化

from langchain_openai import OpenAIEmbeddings

# OpenAI 兼容接口：可对接 DeepSeek、阿里百炼、Ollama 等任意兼容服务
embedding = OpenAIEmbeddings(
    model=os.getenv("AL_EMMODEL_NAME"),       # 从环境变量读取，避免硬编码
    base_url=os.getenv("AL_BASE_URL"),         # API 地址
    api_key=os.getenv("AL_API_KEY"),           # API 密钥
    check_embedding_ctx_length=False,          # 关闭 token 长度检查，兼容不同厂商
    chunk_size=10,                              # 每批处理 10 个文本块，控制并发
)

几个关键参数说明：

参数	作用
check_embedding_ctx_length=False	避免因不同 API 厂商的上下文长度检查逻辑差异导致报错
chunk_size=10	控制并发批次大小，对 API 限流友好；网络不稳定时可调小
model / base_url / api_key	从环境变量读取，避免硬编码敏感信息

(3) 文本分割与向量库构建

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma

# 持久化策略：检测本地已有向量库则直接加载，避免重复抓取和索引
if os.path.exists("./chroma_db") and os.listdir("./chroma_db"):
    vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embedding)
    print("已加载已有向量库")
else:
    # 首次运行：从网页抓取并建立索引
    docs = loader.load()

    # RecursiveCharacterTextSplitter：递归语义分割器
    # 按分隔符优先级依次尝试，优先在自然断点处切割
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000,       # 每个块最大 1000 字符
        chunk_overlap=200,     # 相邻块重叠 200 字符，防止关键信息被边界切断
        add_start_index=True,  # 元数据中记录原文起始位置，便于溯源
        # 中文友好分隔符：段落 → 换行 → 句子标点 → 字符
        separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
    )
    all_splits = text_splitter.split_documents(docs)  # 执行切分

    # 从切分块创建 Chroma 向量库并持久化到磁盘
    vectorstore = Chroma.from_documents(
        documents=all_splits, embedding=embedding, persist_directory="./chroma_db"
    )

核心设计决策：

1. chunk_size=1000 与 chunk_overlap=200：每个文本块最多 1000 字符，相邻块重叠 200 字符。重叠的意义在于防止关键信息恰好落在两个块的边界被切断。20% 的重叠率是一个经验性的平衡点——太小则语义断裂风险高，太大则冗余增加存储和检索成本。

2. 中文友好的分隔符优先级：["\n\n", "\n", "。", "！", "？", "；", " ", ""]——先按段落切，再按句子切，最后到单个字符。这个顺序有效避免了在句子中间切断导致的语义碎片化。

3. 持久化策略：首次运行创建向量库并落盘到 ./chroma_db，后续运行直接加载。这避免了每次重启都重新抓取和索引，在生产环境中是标配。

4. add_start_index=True：在元数据中记录每个块在原文中的起始位置，方便溯源。

(4) 检索工具封装

from langchain_core.tools import create_retriever_tool

# 将向量库封装为检索器，每次取最相关的 5 个文档片段
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# 将检索器包装为 Agent 可调用的 Tool
# Agent 根据 description 判断何时调用、工具能做什么
retriever_tool = create_retriever_tool(
    retriever=retriever,
    name="langchain_docs",    # 工具唯一标识，Agent 通过它引用工具
    description=(
        "在已加载的技术文档知识库中进行语义搜索，返回与用户问题最相关的文本片段。"
        "当用户询问具体技术细节、配置方法或参考文档内容时，必须优先使用此工具。"
    ),
)

create_retriever_tool 将检索器包装为 Agent 可调用的 Tool。Agent 会根据 description 中的指令判断何时应该调用它。这段描述相当于给 Agent 的"工具使用说明书"——写得越清晰，Agent 调用工具的时机就越准确。

(5) Agent 创建与问答循环

from langchain.agents import create_agent
from langchain_core.messages import HumanMessage

# LLM 初始化：同样使用 OpenAI 兼容接口
llm = ChatOpenAI(
    model=os.getenv("AL_MODEL_NAME"),   # 从环境变量读取模型名
    base_url=os.getenv("AL_BASE_URL"),
    api_key=os.getenv("AL_API_KEY"),
)

# create_agent：将 LLM 与工具绑定，创建可自主决策调用工具的智能体
agent = create_agent(
    model=llm,
    tools=[retriever_tool],             # Agent 可调用的工具列表
    # system_prompt 定义 Agent 的行为规范：何时检索、如何回答、怎么处理边界
    system_prompt=(
        "你是一个专业的技术助手。你有权使用搜索工具查询知识库。回答规则：\n"
        "1. 当用户问题需要参考具体技术文档、配置细节、代码示例等时，请先调用 langchain_docs 工具获取相关内容；\n"
        "2. 对于问候、闲聊、天气、通用常识等与知识库无关的问题，不需要调用工具，直接根据常识回答即可；"
        "3. 若调用工具后获得内容,严格根据搜索结果回答，不要添加额外推测或外部知识；\n"
        "4. 如果搜索结果中找不到相关信息，直接回复"知识库中未找到相关内容"；\n"
        "5. 回答使用中文，力求精简：只给出最核心的步骤、配置或结论，避免长段解释或重复内容；\n"
        "6. 如果有多个相关结果，选择最直接匹配的内容进行总结。"
    ),
)

# 交互式问答循环
while True:
    question = input("问题：")           # 终端读取用户输入
    if question in ["stop", "quit"]:     # 退出条件
        break
    # agent.invoke：自动完成"理解问题 → 决定是否调用工具 → 生成回答"全流程
    response = agent.invoke({"messages": [HumanMessage(question)]})
    # 提取最后一条消息（Agent 的最终回复）
    final_answer = response["messages"][-1].content
    print("答案：", final_answer, end="\n")

System Prompt 的设计是关键。它定义了 Agent 的行为边界：

• 规则 1：明确工具调用条件——"技术细节、配置方法、代码示例"时触发检索。
• 规则 2：避免滥用工具——闲聊不需要检索，直接回答即可，节省 API 调用成本。
• 规则 3：防止幻觉——严格基于检索结果回答，不允许模型自由发挥。
• 规则 4：诚实告知——搜不到就说搜不到，不要强答。

3.3 第一版小结

这是 RAG 的"Hello World"，它跑通了完整链路，但检索策略存在明显局限：

• 纯向量检索：依赖语义相似度匹配，对精确关键词（如函数名、配置项、错误码）的召回能力较弱。
• 单一检索源：没有候选结果的质量控制机制，Top-K 结果的质量直接影响最终回答。

下一节我们引入 BM25 来解决这些问题。

---

四、第二版：混合检索——语义 + 关键词双引擎

main2.py 在第一版基础上引入了 BM25 关键词检索 和 加权融合机制，将向量检索的语义理解能力与 BM25 的精确匹配能力结合在一起。

4.1 BM25 是什么？

BM25（Best Match 25） 是信息检索领域最经典的算法之一，属于"稀疏检索"（Sparse Retrieval）范畴。它的核心逻辑是统计查询词在文档中的 词频（TF） 和 逆文档频率（IDF），计算查询与文档的相关性得分。

与向量检索的核心区别：

维度	向量检索（Dense）	BM25（Sparse）
原理	语义相似度（余弦距离）	关键词匹配统计
优势	理解同义词、近义表达	精确匹配专有名词、代码
劣势	对生僻术语召回弱	不理解语义变体
典型场景	“怎么配置数据库” ↔ “如何设置 DB”	搜索 “max_connections” 参数

两者的优劣势恰好互补——这正是混合检索的理论基础。

4.2 实现对比

第一版中，检索器是单一的向量检索：

# main.py - 单向量检索
retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

第二版改造为混合检索：

# main2.py - 混合检索
from langchain_community.retrievers import BM25Retriever
from langchain_classic.retrievers import EnsembleRetriever

vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})

bm25_retriever = BM25Retriever.from_documents(documents=all_splits, k=10)

retrievers = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4],  # 向量权重 60%，BM25 权重 40%
    k=10,
)

4.3 EnsembleRetriever 的工作原理

EnsembleRetriever 的融合策略基于 RRF（Reciprocal Rank Fusion，倒数排名融合）：

1. 向量检索器返回 Top-10 结果（带相似度分数）。
2. BM25 检索器返回 Top-10 结果（带关键词相关性分数）。
3. 对两组结果做分数归一化，按 weights 权重加权求和。
4. 取融合后得分最高的 Top-K 个文档返回。

向量检索分数 × 0.6 + BM25 分数 × 0.4 = 最终分数

权重设置需要根据实际场景调优：

• 语义理解型任务（如长文档问答、摘要生成）：向量权重应更高（0.7~0.8）。
• 精确匹配型任务（如代码搜索、配置项查询）：BM25 权重应更高（0.5~0.6）。
• 0.6 vs 0.4 是一个通用平衡点，对大多数技术文档问答都适用。

4.4 恢复文档列表的关键细节

在从已有向量库恢复时，第二版增加了一个关键操作：

# main2.py - 从 Chroma 恢复文档供 BM25 使用
if os.path.exists("./chroma_db") and os.listdir("./chroma_db"):
    vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embedding)
    existing = vectorstore.get(include=["metadatas", "documents"])
    all_splits = [
        Document(page_content=doc, metadata=meta)
        for doc, meta in zip(existing["documents"], existing["metadatas"])
    ]
    print("已加载已有向量库")

BM25 需要原始文档文本构建倒排索引，但 Chroma 持久化的是向量数据。因此需要从 Chroma 中取出 documents 和 metadatas，重新构造 Document 对象列表。这保证了二次启动时 BM25 也能正常工作。

4.5 完整代码

以下是 main2.py 的完整实现（107 行），为节省篇幅，与 main.py 相同的部分已在上文解析：

"""RAG agent 混合检索版"""

import os
import bs4
from dotenv import load_dotenv                    # 加载 .env 环境变量
from langchain.agents import create_agent          # Agent 创建
from langchain_chroma import Chroma                # Chroma 向量数据库
from langchain_community.document_loaders import WebBaseLoader  # 网页加载器
from langchain_community.retrievers import BM25Retriever         # BM25 关键词检索
from langchain_classic.retrievers import EnsembleRetriever       # 混合检索器
from langchain_core.documents import Document       # 文档对象
from langchain_core.messages import HumanMessage    # 用户消息
from langchain_core.tools import create_retriever_tool  # 检索工具包装
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter

# 加载 .env 中的 API 配置
load_dotenv()

# 网页文档加载，BS4 过滤正文区域
loader = WebBaseLoader(
    web_path="https://pixle.blog.csdn.net/article/details/156417028?spm=1001.2014.3001.5502",
    bs_kwargs={"parse_only": bs4.SoupStrainer(class_=("blog-content-box"))},
)

# 向量嵌入模型：OpenAI 兼容接口
embedding = OpenAIEmbeddings(
    model=os.getenv("AL_EMMODEL_NAME"),
    base_url=os.getenv("AL_BASE_URL"),
    api_key=os.getenv("AL_API_KEY"),
    check_embedding_ctx_length=False,   # 兼容不同厂商的上下文检查差异
    chunk_size=10,                       # 每批处理 10 个文档
)

# ===== 向量数据库初始化（带持久化） =====
if os.path.exists("./chroma_db") and os.listdir("./chroma_db"):
    # 已有向量库：直接加载
    vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=embedding)
    # 从 Chroma 恢复文档列表，供 BM25 构建索引
    existing = vectorstore.get(include=["metadatas", "documents"])
    all_splits = [
        Document(page_content=doc, metadata=meta)
        for doc, meta in zip(existing["documents"], existing["metadatas"])
    ]
    print("已加载已有向量库")
else:
    # 首次运行：抓取网页 → 切分 → 嵌入 → 持久化
    docs = loader.load()
    text_splitter = RecursiveCharacterTextSplitter(
        chunk_size=1000, chunk_overlap=200, add_start_index=True,
        separators=["\n\n", "\n", "。", "！", "？", "；", " ", ""],
    )
    all_splits = text_splitter.split_documents(docs)
    vectorstore = Chroma.from_documents(
        documents=all_splits, embedding=embedding, persist_directory="./chroma_db"
    )

# ===== 核心变化：双检索引擎混合检索 =====
# 向量检索（语义匹配）
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 10})
# BM25 关键词检索（精确匹配）
bm25_retriever = BM25Retriever.from_documents(documents=all_splits, k=10)
# EnsembleRetriever：RRF 倒数排名融合，将两种检索结果加权合并
retrievers = EnsembleRetriever(
    retrievers=[vector_retriever, bm25_retriever],
    weights=[0.6, 0.4],  # 向量权重 60%，BM25 权重 40%
    k=10,                 # 融合后返回 Top-10
)

# 包装为 Agent 工具
retriever_tool = create_retriever_tool(
    retriever=retrievers,
    name="langchain_docs",
    description=(
        "在已加载的技术文档知识库中进行语义搜索，返回与用户问题最相关的文本片段。"
        "当用户询问具体技术细节、配置方法或参考文档内容时，必须优先使用此工具。"
    ),
)

# 大语言模型
llm = ChatOpenAI(
    model=os.getenv("AL_MODEL_NAME"),
    base_url=os.getenv("AL_BASE_URL"),
    api_key=os.getenv("AL_API_KEY"),
)

# 创建 Agent：LLM + 检索工具 + 行为规范
agent = create_agent(
    model=llm,
    tools=[retriever_tool],
    system_prompt=(
        "你是一个专业的技术助手。你有权使用搜索工具查询知识库。回答规则：\n"
        "1. 当用户问题需要参考具体技术文档、配置细节、代码示例等时，请先调用 langchain_docs 工具获取相关内容；\n"
        "2. 对于问候、闲聊、天气、通用常识等与知识库无关的问题，不需要调用工具，直接根据常识回答即可；"
        "3. 若调用工具后获得内容,严格根据搜索结果回答，不要添加额外推测或外部知识；\n"
        "4. 如果搜索结果中找不到相关信息，直接回复"知识库中未找到相关内容"；\n"
        "5. 回答使用中文，力求精简：只给出最核心的步骤、配置或结论，避免长段解释或重复内容；\n"
        "6. 如果有多个相关结果，选择最直接匹配的内容进行总结。"
    ),
)

# 交互式问答循环
while True:
    question = input("问题：")
    if question in ["stop", "quit"]:     # 输入 stop/quit 退出
        break
    response = agent.invoke({"messages": [HumanMessage(question)]})
    final_answer = response["messages"][-1].content
    print("答案：", final_answer, end="\n")

---

五、main.py vs main2.py：对比总结

维度	main.py（纯向量）	main2.py（混合检索）
检索器数量	1（向量）	2（向量 + BM25）
召回机制	单一语义匹配	语义 + 关键词双通路
专有名词召回	较弱	强（BM25 精准命中）
同义词理解	强	弱（需要靠向量通路补足）
代码行数	107	123
额外依赖	无	rank-bm25, langchain-classic
适用场景	通用文档问答	技术文档、配置查询、代码搜索

---

六、上篇小结

本文我们从零构建了两个 RAG 系统：

1. 最简版（main.py）：跑通了文档加载 → 分割 → 嵌入 → 存储 → 检索 → 生成的完整链路，是整个系列的基石。
2. 混合检索版（main2.py）：引入 BM25 关键词检索与 EnsembleRetriever 加权融合，弥补了纯向量检索在精确匹配上的短板。

但目前的方案仍有两个明显的可优化空间：

• 文档块粒度固定：1000 字符一刀切，小块检索精确但丢失上下文，大块上下文完整但检索精度降低。能不能兼得？
• 无召回质量控制：Top-K 结果直接送入 LLM，缺乏对相关性排序的精细化调整。排名第 10 的文档可能比第 1 的更有用，但 LLM 无法区分。

下一篇文章我们将逐一解决这两个问题，引入父子文档索引和**重排序（Reranking）**机制。

---

下一篇：RAG 技术系列（中篇）：检索质量深度优化——父子文档、重排序与多格式文档