专栏第9篇：前面的文章讲了 Agent 怎么思考、怎么调用工具、怎么用 LangGraph 组织工作流。但 Agent 回答问题需要知识——这些知识从哪来？LLM 的训练数据有截止日期，对企业私有数据更是一无所知。RAG（检索增强生成）就是解决这个问题的核心技术。今天我们从文档加载到向量存储，走完 RAG 的"离线阶段"。

目录

一、RAG 到底是什么？

很多人以为 RAG 就是"把文档丢进向量库，用户提问时检索一下，把结果塞给 LLM"。这个理解没错，但只看到了表面。

1.1 完整的 RAG 流水线

在线阶段

离线阶段

原始文档

文档加载

文本分割

向量化

向量存储

用户提问

查询向量化

相似度检索

上下文构建

LLM生成回答

返回答案+来源

RAG 分为两个阶段：

  • 离线阶段（本文重点）：文档 → 加载 → 分割 → 向量化 → 存储
  • 在线阶段（下篇重点）：用户提问 → 检索 → 上下文构建 → LLM 生成

每一步都影响最终效果：

  • 分割不好 → 关键信息被切断 → 检索不到
  • 向量模型不合适 → 语义匹配不准 → 召回率低
  • 检索数量太多 → 上下文超限 → LLM 遗漏重点
  • 检索数量太少 → 信息不足 → 回答不完整

1.2 为什么需要 RAG？

问题 不用 RAG 用 RAG
知识截止 LLM 只知道训练数据 连接实时知识库
幻觉问题 可能编造不存在的信息 基于检索的事实生成
私有数据 无法访问企业内部文档 注入私有知识库
可溯源 无法验证回答来源 返回引用文档

二、文档加载与分割：决定 RAG 效果的上限

RAG 有一句行话：Garbage in, garbage out。数据质量决定输出质量，如果输入的文档处理不好，后面再强的模型也救不回来。

2.1 文档加载

LangChain 提供了多种加载器，适配不同格式：

from langchain_community.document_loaders import TextLoader, PyPDFLoader, CSVLoader

# 文本文件
loader = TextLoader("knowledge.txt", encoding="utf-8")
docs = loader.load()

# PDF 文件
loader = PyPDFLoader("manual.pdf")
docs = loader.load()  # 每页一个 Document

# CSV 文件
loader = CSVLoader("data.csv", encoding="utf-8")
docs = loader.load()  # 每行一个 Document

2.2 文本分割：不是越细越好

加载后的文档需要切成小块，才能送入向量库。但怎么切，直接决定检索效果。

基础分割：RecursiveCharacterTextSplitter

最常用的分割器，按优先级尝试不同分隔符：

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,      # 每个块的最大字符数
    chunk_overlap=100,   # 相邻块之间的重叠字符数
    separators=["\n\n", "\n", "。", "！", "？", ".", " ", ""]  # 优先按段落，其次按句子
)

chunks = splitter.split_documents(docs)

separators 的作用：分割器不是简单按字符数硬切，而是优先在段落/句子边界处分割。只有当一个段落或句子超过 chunk_size 时，才会强制切割。所以 RecursiveCharacterTextSplitter 本身就是"按段落/句子优先"的分割器。

chunk_size 的选择：

chunk_size 优点 缺点 适用场景
小（200-500） 粒度细，匹配精准 可能切断上下文 短问答对、FAQ
中（500-1000） 平衡精准度和完整性 需要调参 通用文档
大（1000-2000） 保留完整上下文 噪声多，匹配泛 长文分析、论文

💡 问答对形式的特殊处理：如果你的知识库本身就是问答对（Q+A）格式，chunk_size 可以设得更大（如 2048），策略是"小于上限不切分"。因为问答对的语义边界天然就是 QA 对本身，一个完整的 QA 作为一个 chunk 效果更好。

举个例子：你的知识库里大部分 QA 是 500-2000 字符，只有少数超长 QA 达到 3000 字符。设 chunk_size=2048 后，500-2000 字符的 QA 都保持完整不切分，只有超过 2048 的超长 QA 才会被切分。所以 2048 不是"目标大小"，而是"允许的最大长度"。

overlap 的作用：

文档：... 机器学习是AI的分支。深度学习是机器学习的子集。 ...

chunk_size=20, overlap=5
块1: "机器学习是AI的分支。深度学"
块2: "度学习是机器学习的子集。 ..."  ← 重叠5个字符，避免断句

overlap 保证句子不会在边界处被切断，通常设为 chunk_size 的 10%-20%。

高级分割策略

基础分割器适用于大多数场景。如果文档有特殊结构或对切分质量要求更高，可以用更精细的策略：

语义分割（Semantic Chunking）

基础分割靠分隔符规则判断边界，但有时候两个段落在语法上连续，语义却已经转换了。语义分割用 Embedding 模型来判断语义边界：

  1. 把文本按句子拆开
  2. 计算相邻句子的 Embedding 相似度
  3. 如果相似度低于阈值，就在这里切分
from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings

# 用 Embedding 模型判断语义边界
semantic_chunker = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile"  # 用百分位数作为切分阈值
)
chunks = semantic_chunker.split_text(text)
# 每个 chunk 是一个语义连贯的段落

语义分割 vs 基础分割：

维度 基础分割 语义分割
边界判断 分隔符规则（段落、句号） 模型计算相似度
成本 无额外成本 需要 Embedding 调用
效果 对结构规范的文档效果好 对语义转换隐含的文本更精准
适用 FAQ、技术文档 文章、散文、长篇论述

💡 简单判断：文档结构清晰（有标题、段落分明）→ 基础分割够用；文档是连续论述、语义边界不明显 → 考虑语义分割。

按标题分割

按 Markdown 标题切分，保留标题作为上下文：

from langchain_text_splitters import MarkdownHeaderTextSplitter

markdown_splitter = MarkdownHeaderTextSplitter(
    headers_to_split_on=[("#", "Header 1"), ("##", "Header 2")]
)
chunks = markdown_splitter.split_text(markdown_text)
# 每个 chunk 都带有所属标题的 metadata

父子分割（Parent-Child Chunking）

一种更聪明的分割策略：

  • 父块（Parent）：大块，用于检索（保留完整上下文，召回率高）
  • 子块（Child）：小块，用于向量化（粒度细，匹配精准）

检索时匹配子块，但返回父块的完整内容给 LLM：

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

# 子块：小粒度，用于向量检索
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)

# 父块：大粒度，用于返回给 LLM
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=1000)

retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=InMemoryStore(),
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)

为什么父子分割有效？

  • 小块做向量检索 → 匹配更精准（不会因为包含太多无关信息而稀释语义）
  • 大块返回给 LLM → 上下文更完整（LLM 能看到完整的段落，而不是被切断的片段）

按代码结构分割

代码文档和普通文本不同，应该按函数、类来分割：

from langchain_text_splitters import Language

# Python 代码分割
python_splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=500,
    chunk_overlap=50
)
chunks = python_splitter.split_text(python_code)
# 会优先在函数、类定义处分割

分割策略对比：

策略 原理 优势 局限 适用场景
固定长度 + 段落优先 按字符数切割，优先在段落/句子边界处分割 简单通用、无额外成本 长段落仍可能被切 通用文档、FAQ
语义分割 Embedding 模型判断语义边界，相似度低处切分 语义边界精准 需要模型调用、有成本 文章、散文、长篇论述
按标题分割 按 Markdown/HTML 标题切分 语义完整、标题作上下文 依赖文档有标题结构 技术文档、产品手册
父子分割 小块检索、大块返回 兼顾匹配精度和上下文完整性 实现复杂、存储翻倍 长文档、上下文依赖强的场景
按代码结构分割 按函数/类定义切分 保留代码逻辑完整性 仅适用于代码 API 文档、代码库

💡 选择建议：没有万能策略，根据知识库类型选。问答对用固定长度（大 chunk_size），技术文档按标题，长文档用父子分割，代码库按代码结构。也可以组合使用——先按标题分段，再对每段做固定长度切割。

2.3 元数据：别忘了给文档打标签

分割时保留元数据，后面追溯来源时才有据可查：

from langchain_core.documents import Document

doc = Document(
    page_content="Python 由 Guido 创建",
    metadata={
        "source": "Python历史",
        "page": 1,
        "doc_id": "doc_001"
    }
)

三、向量化与存储：Embedding 模型的选择

文本变成向量，才能做相似度检索。Embedding 模型的选择决定了"语义匹配"的质量。

3.1 向量化流程

from langchain_openai import OpenAIEmbeddings

# 初始化 Embedding 模型
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# 文本 → 向量
vector = embeddings.embed_query("什么是机器学习？")
# 返回 1536 维的浮点数数组

3.2 Embedding 模型选型

模型 维度 特点 适用场景
text-embedding-3-small 1536 便宜、通用 通用问答
text-embedding-3-large 3072 精度高、贵 专业领域
BGE 系列 1024 中文优化好 中文知识库
E5 系列 1024 开源、可本地部署 隐私敏感场景

💡 选型建议：中文知识库优先选 BGE 系列，英文选 OpenAI 的 embedding-3，隐私要求高选本地部署的 E5。

3.3 稠密向量 vs 稀疏向量

RAG 中常见的两种向量表示方式：

类型 生成方式 维度 特点 代表
稠密向量 Embedding 模型编码 通常 768-3072 维 捕获语义关系，浮点数数组 OpenAI Embedding、BGE
稀疏向量 关键词统计（如 TF-IDF、BM25） 和词表大小相同（几万到几十万维） 捕获词频信息，大部分为 0 BM25、TF-IDF

简单理解：

  • 稠密向量：把"北京天气"编码成 [0.12, -0.05, 0.33, ...]，语义相近的文本向量也相近
  • 稀疏向量：记录每个词出现了几次，“北京:1, 天气:1, 今天:0, …”，大部分是 0

为什么 RAG 主流用稠密向量？

  • 稠密向量能理解语义相似性（“founder” 和 “creator” 向量相近）
  • 稀疏向量只能精确匹配关键词（搜"创始人"找不到"creator"）

但稀疏向量（BM25）也有优势：精确匹配能力强、计算快、不需要 GPU。所以生产环境常见做法是两者结合（混合检索），下篇会详细讲。

3.4 向量检索的原理：余弦相似度

向量库检索的核心是相似度计算。最常用的方法是余弦相似度：

两个向量的夹角越小 → 余弦值越接近 1 → 越相似
两个向量的夹角越大 → 余弦值越接近 0 → 越不相似

import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 查询向量
query_vec = embeddings.embed_query("Python 创始人")
# 文档向量
doc_vec = embeddings.embed_query("Guido 创建了 Python")

similarity = cosine_similarity(query_vec, doc_vec)
print(f"相似度：{similarity:.4f}")  # 值越接近 1 越相似

除了余弦相似度，还有其它方法：

方法 计算方式 特点 适用场景
余弦相似度 向量夹角的余弦值 不受向量长度影响，只关注方向 最常用，语义检索默认选择
欧氏距离 向量空间中的直线距离 距离越小越相似，受向量长度影响 图像检索、聚类分析
点积 向量对应元素相乘后求和 计算最简单，同时受方向和长度影响 需要区分向量"强度"的场景

💡 为什么余弦相似度最常用？ Embedding 模型输出的向量已经被归一化到单位长度，此时余弦相似度和点积结果等价。而余弦相似度的物理意义更直观（夹角越小越相似），所以成了默认选择。

3.5 向量存储

文本变成向量后，需要存起来供后续检索。为什么不能直接存原始文本、用的时候再向量化？

不行，太慢了。 假设你有 1 万个文档，每次用户提问都要把 1 万个文档全部跑一遍 Embedding，再逐个比相似度——这个过程可能要几分钟。所以必须预先算好向量，存进向量库，检索时只需把用户的问题向量化一次，然后从库里找最近的邻居。

简单说：向量库 = 预计算 + 索引加速。预计算解决了"每次都要重新算"的问题，索引加速解决了"逐个比对太慢"的问题。常见的两种索引：

索引类型 原理 特点
HNSW 构建多层近邻图，逐层缩小搜索范围 精度高、构建快、内存占用较大
IVF 把向量空间划分成多个区域，先定位区域再搜索 内存友好、适合超大规模

索引的作用就像图书馆的分类标签：不需要逐本书翻找，先按类别定位，只在相关区域内搜索，把检索从 O(N) 降到 O(logN)。

from langchain_chroma import Chroma

# 创建向量库（这一步会把所有文档向量化并构建索引）
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    collection_name="my_knowledge"
)

# 作为检索器使用
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

不同向量库怎么选？

向量库 特点 默认索引类型 适用场景
Chroma 轻量、本地、无需服务器 HNSW 开发调试、小规模
Qdrant Rust 编写、单节点性能极高、部署简单、分布式可选 HNSW 中小规模到大规模、追求性能
Milvus 云原生、功能最全、专为超大规模设计、需 K8s HNSW / IVF（可配置） 企业级、十亿级向量、有运维团队
Pinecone 全托管、即开即用 专有图索引（类似 HNSW） 快速上线、不想自己运维

四、基础检索：从向量库中找到相关文档

向量库存好了，怎么检索？最基础的方式就是相似度检索——把用户问题也变成向量，和库里的文档向量比相似度，返回最像的 Top-K。

# 默认的相似度检索
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 检索
docs = retriever.invoke("Python 是谁创建的？")
# 返回最相似的 4 个文档块

这就是 RAG 检索的最基础形式。但生产环境通常不够用——下篇我们会讲 BM25、混合检索、查询拓展、重排序等高级策略。

五、完整代码：一个基础 RAG 应用

把前面的知识串起来，实现一个基础但完整的 RAG 应用：

from typing import List, Dict, Optional
from langchain_community.document_loaders import TextLoader, PyPDFLoader, CSVLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_chroma import Chroma
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

class RAGApplication:
    """基础 RAG 应用"""
    
    def __init__(self, chunk_size: int = 500, chunk_overlap: int = 100):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.llm = create_llm_instance()
        self.embeddings = create_embedding_instance()
        self.text_splitter = RecursiveCharacterTextSplitter(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            separators=["\n\n", "\n", ".", " ", ""]
        )
        self.vectorstore = None
        self.documents: Dict[str, dict] = {}
    
    def add_document(self, file_path: str, doc_id: Optional[str] = None) -> bool:
        """添加文档到知识库"""
        doc_id = doc_id or os.path.basename(file_path)
        file_extension = os.path.splitext(file_path)[1].lower()
        
        if file_extension == ".txt":
            loader = TextLoader(file_path, encoding="utf-8")
        elif file_extension == ".pdf":
            loader = PyPDFLoader(file_path)
        elif file_extension == ".csv":
            loader = CSVLoader(file_path, encoding="utf-8")
        else:
            return False
        
        documents = loader.load()
        for doc in documents:
            doc.metadata["doc_id"] = doc_id
        
        chunks = self.text_splitter.split_documents(documents)
        
        if self.vectorstore is None:
            self.vectorstore = Chroma(
                embedding_function=self.embeddings,
                collection_name="rag_app"
            )
        self.vectorstore.add_documents(chunks)
        
        self.documents[doc_id] = {
            "file_path": file_path,
            "chunks_count": len(chunks)
        }
        return True
    
    def query(self, question: str, k: int = 4) -> dict:
        """查询知识库"""
        if not self.vectorstore:
            return {"answer": "请先添加文档", "sources": [], "confidence": 0.0}
        
        # 检索
        retriever = self.vectorstore.as_retriever(search_kwargs={"k": k})
        docs = retriever.invoke(question)
        
        if not docs:
            return {"answer": "未找到相关信息", "sources": [], "confidence": 0.0}
        
        # 构建上下文
        context = "\n\n".join(f"[文档 {i+1}]\n{doc.page_content}" 
                              for i, doc in enumerate(docs))
        
        # 生成回答
        prompt = ChatPromptTemplate.from_template("""
你是一个专业的问答助手。基于以下参考文档回答问题。

参考文档：
{context}

用户问题：{question}

要求：
1. 基于提供的参考文档回答
2. 如果文档中没有相关信息，请明确说明
3. 回答要准确、简洁

回答：""")
        
        chain = prompt | self.llm | StrOutputParser()
        answer = chain.invoke({"context": context, "question": question})
        
        # 来源信息
        sources = [
            {
                "content": doc.page_content[:150] + "...",
                "doc_id": doc.metadata.get("doc_id", "unknown")
            }
            for doc in docs
        ]
        
        return {
            "answer": answer,
            "sources": sources,
            "confidence": min(len(docs) / k, 1.0)
        }

使用示例：

app = RAGApplication(chunk_size=500, chunk_overlap=100)

# 添加文档
app.add_document("product_manual.pdf", "产品手册")
app.add_document("user_guide.txt", "用户指南")

# 查询
result = app.query("产品有哪些特点？")
print(f"回答：{result['answer']}")
print(f"来源：{result['sources']}")

六、总结

本文聚焦 RAG 的"离线阶段"——从文档加载到向量存储：

  1. RAG 的本质：不是向量库+LLM 的简单组合，而是从文档加载到答案生成的完整流水线
  2. 文档加载与分割：
    • 基础分割：RecursiveCharacterTextSplitter，chunk_size 和 overlap 的选择
    • 高级分割：语义分割（模型判断边界）、按标题分割、父子分割（Parent-Child）、按代码结构分割
    • 问答对形式：设大 chunk_size 做安全阀，尽量保持 QA 完整性
  3. 向量化与存储：
    • Embedding 模型决定语义匹配质量，中文知识库优先选 BGE 系列
    • 稠密向量捕获语义，稀疏向量捕获词频，生产环境通常两者结合
    • 向量库用 HNSW/IVF 索引加速检索，Qdrant 是性能和部署的平衡之选
  4. 基础检索：相似度检索是最基础的形式，生产环境需要更高级的策略
  5. 完整代码：一个支持多文档管理、多格式加载、来源追溯的基础 RAG 应用

💡 优化口诀：加载要完整、分割要合理、向量要精准、存储要选对。

参考资源：

  • LangChain Documentation: RAG
  • LangChain Text Splitters
  • OpenAI Embeddings Documentation
# AI
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

RDK X5 从零上手：烧录系统镜像 + Wi-Fi 配网 + SSH 远程登录 完整教程

摘要：本文提供RDK X5开发板从零开始的完整环境搭建指南，覆盖系统镜像选择（Server/Desktop版）、使用RDK Studio或Rufus工具烧录Micro SD卡、通过115200波特率串口首次登录、命令行/Wi-Fi配网（nmcli/srpi-config）、SSH远程连接等关键步骤。特别强调必须使用5V/5A电源适配器，避免USB供电不足导致重启。教程包含硬件清单、详细操作命令和常

avatar AtomGit开源社区
cover

基于SpringBoot+Vue的火锅店管理系统管理系统设计与实现【Java+MySQL+MyBatis完整源码】

avatar AtomGit开源社区

Java开发工具指南：提高开发效率的利器

Maven 采用约定优于配置的原则，通过 pom.xml 文件定义项目的结构和依赖关系，简化了项目的构建过程。IntelliJ IDEA 是由 JetBrains 公司开发的一款功能全面的 Java 集成开发环境（IDE），被广泛认为是目前最优秀的 Java IDE 之一。例如，通过安装 Mylyn 插件，可以实现任务导向的开发，提高工作效率；它支持多种语言和框架，通过插件可以实现代码补全、调试、

avatar AtomGit开源社区