LangChain v1.3.4 笔记 - 06 RAG 检索增强生成

对应 LangChain 中的 Retrieval 内容 虽然前面通过 store | checkpoint 保证了模型具备上下文的能力。但仍然存在 消息滞后、知识缺失 的问题，因为模型的训练数据都是

中文全文翻译正在补齐，可先查看下方英文原文。

对应 LangChain 中的 Retrieval 内容

虽然前面通过 store | checkpoint 保证了模型具备上下文的能力。但仍然存在 消息滞后、知识缺失 的问题，因为模型的训练数据都是公开数据并且时效性不是当前；从而导致在回答问题时出现 幻觉

RAG 就是将模型触及不到知识作为模型的知识库每次回答问题时去检索出来，然后再回答给用户，从而在某些特定领域让模型的回答更加靠谱。

什么是 RAG

RAG（Retrieval-Augmented Generation，检索增强生成）

• 解析语料内容，将非结构化的数据转换为结构化数据 Loader
• 通过语义或者字符等手段对结构化数据进行拆分 Chunk
• 通过向量模型，将拆分后的数据解析转换成向量存入数据库中 Embedding
• 用户的输入也转为向量，做数据的相似度检索，将检索出来的数据作为上下文结合用户的输入提交给模型 Retrieval
• 模型根据上下文生成回答内容 Generation

最典型的莫过于客服系统，包括飞书开放文档的智能助手等

RAG 优点就是提高问答的可靠性和业务数据的私密性；缺点就是上下文会比较长时间比较慢，因为每次都会查都会查库，并且携带的上下午比较多

Document Loader 文档加载器

实际业务中文档的形式五花八门，LangChain 也提供了多种加载方案 加载器

需要提前安装 uv add langchain-community

from langchain_community.document_loaders import (
  TextLoader, 
  CSVLoader, 
  PyPDFLoader, 
  JSONLoader, 
  UnstructuredWordDocumentLoader, 
  UnstructuredMarkdownLoader, 
  UnstructuredHTMLLoader, 
  DirectoryLoader, 
  PythonLoader
)

所有的 Loader 都继承至 BaseLoader 都实现了 load | lazy_load 方法加载文档最终返回一个 list[Docutment] 对象包含 metadata | page_content 属性。

TextLoader

from langchain_community import TextLoader

loader = TextLoader(file_path=f"{assets_dir}/01.txt", encoding="utf-8")
rprint(loader.load())

CSVLoader

每列数据作为一个 Document 输出

from langchain_community import CSVLoader

loader = CSVLoader(file_path=f"{assets_dir}/01.csv")
rprint(loader.load())

JSONLoader

接收一个 json_schema 参数加载数据中指定数据，通过 jq 实现

from langchain_community import JSONLoader

loader = JSONLoader(
  file_path=f"{assets_dir}/01.json", 
  jq_schema=".",  
  text_content=False 
)
rprint(loader.load())

PyPDFLoader

依赖 pypdf 实现，常用参数有

• mode: single 或 page 决定加载出来文档的个数默认 page

• extraction_mode: plain 提取文本也是默认值； layout 布局感知会插入大量的空格

• password 加密 pdf 的密码

  from langchain_community import PyPDFLoader

  loader = PyPDFLoader(f"{assets_dir}/xxx.pdf", mode="single") rprint(loader.load())

UnstructuredWordDocumentLoader

提取非结构化数据依赖 unstructured 常用参数有

• mode: single 或 elements，决定按标题加载还是整个一块且，默认 single
• strategy: fast 快速模型不纠结布局，hi_res 高速模型

还需要安装 python-docx

如果运行仍然报错，参考信息继续安装内容，总结处理起来很麻烦

FileNotFoundError: soffice command was not found. Please install libreoffice on your system and try again.
- Install instructions: https://www.libreoffice.org/get-help/install-howto/
- Mac: https://formulae.brew.sh/cask/libreoffice
- Debian: https://wiki.debian.org/LibreOffice



from langchain_community import UnstructuredWordDocumentLoader

loader = UnstructuredWordDocumentLoader(
  f"{assets_dir}/01.doc", 
  mode="elements", 
  strategy="fast"
)
rprint(loader.load())

UnstructuredMarkdownLoader

依赖 markdown

from langchain_community import UnstructuredMarkdownLoader

loader = UnstructuredMarkdownLoader(
  f"{assets_dir}/01.md",
  mode="single" 
)
rprint(loader.load())

UnstructuredHTMLLoader

from langchain_community import UnstructuredHTMLLoader

loader = UnstructuredHTMLLoader(
  f"{assets_dir}/01.html",
  mode="single" 
)
rprint(loader.load())

DirectoryLoader、PythonLoader

from langchain_community import DirectoryLoader, PythonLoader

loader = DirectoryLoader(
  f"{assets_dir}",
  glob="*", 
  use_multithreading=True, 
  show_progress=True, 
  loader_cls=PythonLoader 
)
rprint(loader.load())

Text Splitter 文本切割

太大的 Document 会影响模型的判断增加不必要的消耗，根据策略将语料进一步拆分为一个个的 Chunk 可以减少内容噪音和减少 token 消耗。

• 根据句子 按照自然语句的边界切分，保证语义的完整性
• 固定字符数 可能会把语义给搞坏掉
• 固定字符数 + 重叠窗口 切分的前后各多保留一些内容，避免切分关键内容，确保信息的连贯性
• 递归字符 通过递归字符的方式，动态确认切分点，可以根据文档的复杂度和信息密度调整 chunk 的大小
• 根据语义内容 保证相关内容的语义和完整性，不过需要额外引入一个模型去切分

4、5切分方案比较推荐，5 不能保证 chunk 的大小，切效率不高。 切分方案在线测试

LangChain 切分方案实例，切分类都基于 TextSplitter 实现，需要额外安装 uv add langchain-text-splitters

常用参数

• chunk_size 默认 4000，切分的大小
• chunk_overlap 默认 200，chunk 之间的重叠的字符数
• length_function 默认 len 函数 测量给定块长度的函数
• keep_separator 默认 false，可选 bool | 'start' | 'end' 是否保留分隔符及应该放在哪个位置
• add_start_index 默认 false 是否在元数据中包含索引
• strip_whitespace 默认 true 是否去除开头和结尾的空格

常用方法

def split_text(self, text: str) -> list[str]:

def create_documents(
    self, texts: list[str], metadatas: list[dict[Any, Any]] | None = None
) -> list[Document]:

def split_documents(self, documents: Iterable[Document]) -> list[Document]:

CharacterTextSplitter 字符切分

separator 存在优先原则，当指定分隔符时，会优先考虑分隔符切割其实才是 chunk_size 避免句子中间硬性切断，chunk_size 过小会导致 chunk_overlap 失效

额外参数 separator 分隔符，默认 \n\n

from langchain_text_splitters import CharacterTextSplitter

loader = TextLoader(file_path=f"{assets_dir}/01.txt", encoding="utf-8")
splitter = CharacterTextSplitter(
  chunk_size=50, 
  chunk_overlap=5, 
  separator="" 
)

texts = splitter.split_text(loader.load()[0].page_content)
rprint(texts)

RecursiveCharacterTextSplitter 递归字符切分

遵循先拆分后合并的方案

拆分 先按照分隔符列表的顺序，从第一个分隔符开始切分，切分后如果大于 chunk_size 就继续按照第二个分隔符继续拆分，直到所有内容都小于 chunk_size

合并 遍历 chunk_list，从 当前 chunk + 1 开始进行合并

• 无 overlap
  • 如果 当前块大小 + 下一个碎片大小 ≤ chunk_size → 合并，继续
  • 如果 当前块大小 + 下一个碎片大小 > chunk_size → 保存当前块，以下一个碎片作为新块的起点，继续遍历
• 有 overlap 逻辑同上，但 "保存当前块" 时
  • 从当前块的末尾取出 chunk_overlap 长度的内容，新块 = overlap 内容 + 下一个碎片
  • 如果 新块大小 > chunk_size → 递归切分下一个碎片继续遍历

常用参数

• separator 默认值 ["\n\n", "\n", " ", ""]

• keep_separator 默认 True 是否保留句尾标点（如 ……），避免切割后丢失语气和逻辑

  from langchain_text_splitters import RecursiveCharacterTextSplitter

  loader = TextLoader(file_path=f"{assets_dir}/01.txt", encoding="utf-8") splitter = RecursiveCharacterTextSplitter( chunk_size=20, chunk_overlap=10, separators=["...", "\n\n", "\n", ""], keep_separator=True )

  texts = splitter.split_text(loader.load()[0].page_content)

代码切分

from langchain_text_splitters import RecursiveCharacterTextSplitter, Language




RecursiveCharacterTextSplitter.from_language(
  language=Language.PYTHON,
  chunk_size=50,
  chunk_overlap=0
)

TokenTextSplitter

encoding_name 是指定 token 编码器，将文本切分为 token 序列 tiktoken -> model.py 中有很多类型支持

from langchain_text_splitters import TokenTextSplitter

docs = TextLoader(file_path=f"{assets_dir}/01.txt", encoding="utf-8").load()
splitter = TokenTextSplitter(encoding_name="o200k_base", chunk_size=20, chunk_overlap=0)
rprint(splitter.split_documents(docs))

CharacterTextSplitter 也支持使用 token 切分

splitter = CharacterTextSplitter.from_tiktoken_encoder(
  encoding_name="o200k_base", 
  chunk_size=20, 
  chunk_overlap=0, 
  separator="。"
)
splitter.split_documents(docs)

SemanticChunker 语义切割

需要安装 uv add langchain-experimental 需要使用向量模型将文本转换为向量，当发现前后语义差异过大时就切分

from langchain_experimental.text_splitter import SemanticChunker
from langchain.embeddings import init_embeddings

embed_model = init_embeddings("ollama:qwen3-embedding:0.6b")

docs = TextLoader(file_path=f"{assets_dir}/01.txt", encoding="utf-8").load()
splitter = SemanticChunker(
  
  embeddings=embed_model, 
  
  
  breakpoint_threshold_type="percentile", 
  breakpoint_threshold_amount=65.0,
  sentence_split_regex=r"(?<=[。？！])\s+" 
)
splitter.split_documents(docs)

HTMLHeaderTextSplitter

from langchain_text_splitters import HTMLHeaderTextSplitter

docs = TextLoader(file_path=f"{assets_dir}/01.html", encoding="utf-8").load()
splitter = HTMLHeaderTextSplitter(headers_to_split_on=[("h1", "切分的依据会放进 metadata 中"),])
splitter.split_text(docs[0].page_content)

MarkdownTextSplitter

from langchain_text_splitters import MarkdownTextSplitter

docs = TextLoader(file_path=f"{assets_dir}/01.md", encoding="utf-8").load()
splitter = MarkdownTextSplitter(chunk_size=50, chunk_overlap=0)
splitter.split_documents(docs)

向量模型 init_embeddings

通过向量模型将文本等内容转换为向量；不同的模型有不同的维度，比如 qwen3-embeddings 最大支持 4096，可以理解拆分的维度越高，匹配时得到的信息越精准

定义模型的方式和 init_chat_model 一致

from langchain.embeddings import init_embeddings



embed_model = init_embeddings(
  "ollama:qwen3-embedding:0.6b",
  dimensions=1024 
)

提供两个方法 embed_query | embed_documents 分别是字符串向量化和列表字符串向量化，返回小数列表。

text = "What was the name mentioned in the conversation?"
vector = embed_model.embed_query(text)

print(len(vector), vector[:5])

text = ["What was the name mentioned in the conversation?"]
vectors = embed_model.embed_documents(text)

print(len(vectors[0]), vectors[0][:5])

向量数据库 Milvus

世面很多支持向量的数据库 FAISS | Chroma | Milvus | Pgvector | Redis | Elasticsearch | Pinecone

Milvus 比较纯粹，专门为了向量存储设计云原生数据库，轻量级高并发都能用 官网

个人测试可以使用 lite 版本 无需安装多余的服务 uv add "pymilvus[milvus-lite]" 类似于 SQLLite

与关系型数据库不同的是 Database -> Collection -> Partition -> Entity

• Collection 集合，与 mongoDB 类似也可以看做是 table
• Partition 分区，Collection 的子集，不是必须手动创建，每个 Collection 都有默认
• Entity 数据，可以看做 table 中的 row

DDL 数据库操作

初始化及数据库的操作

• list_databases 列出所有的数据库，默认会有一个 ['default']

• create_database 创建数据库

• drop_database 删除数据库

• use_database 切换数据库

  from pymilvus import MilvusClient

  client = MilvusClient("./milvus-demo.db") db_name = "rag_vector"

  if db_name not in client.list_databases(): client.create_database(db_name=db_name)

  client.drop_database(db_name=db_name) client.use_database(db_name=db_name)

集合操作

• list_collections 列表所有集合

• create_collection 创建集合

• has_collection 是否包含某个集合

• drop_collection 删除集合

• describe_collection 查看集合元数据

• get_collection_stats 获取集合统计信息

• load_collection 加载集合到内容，不然无法做后续的 curd

  col_name = "docs"

  if col_name not in client.list_collections():

  client.create_collection( collection_name=col_name, dimension=1024, metric_type="COSINE" )

  if client.has_collection(collection_name=col_name): client.drop_collection(collection_name=col_name)

  metadata = client.describe_collection(collection_name=col_name)

DML 数据库操作

创建数据有 insert | upsert 后者再插入同样数据时会先删除旧的再插入新的

texts = [
  "LangChain 是一个用于构建 LLM 应用的开发框架。",
  "RAG 的核心是先检索相关知识，再让大模型生成答案。",
  "Milvus 全球领先的开源向量数据库项目之一"
]


vectors = embed_model.embed_documents(texts)
data = [
  { 
    "id": i, 
    "vector": vectors[i], 
    "text": texts[i] 
  } for i in range(len(texts))
]

client.upsert(collection_name=col_name, data=data)


client.flush(collection_name=col_name)

stats = client.get_collection_stats(collection_name=col_name)

DQL 数据库操作

查询可以使用 get | search | query | query_iterator，共同支持 output_fields 参数

• get 用于 ids 查询

• query 支持 ids | filter

• query_iterator 支持 filter | limit | batch_size

• search 支持 filter | limit | data

  client.load_collection(collection_name=col_name)

  result: list[dict] = client.query( collection_name=col_name, filter="id == 1", output_fields=["*"], ids=["1"], ) for i in result: print(i["id"], i["vector"][:2], i["text"])

  result: QueryIterator = client.query_iterator( collection_name=col_name, output_fields=["text"], limit=2, batch_size=1000, filter="" ) print(result.next())

  result: list[dict] = client.get( collection_name=col_name, output_fields=["*"], ids=["1"], )

  data: list[float] = embed_model.embed_query("什么是向量数据库？") result: list[list[dict]] = client.search( collection_name=col_name, data=[data], output_fields=["text"], limit=2, filter="" )

删除使用 delete 方法，支持 ids | filter 参数

client.delete(
  collection_name=col_name,
  ids=["1"]
)
client.query(collection_name=col_name, output_fields=["text"])