Retrieval

Retieval

一、检索体系全景

LangChain 的检索模块不只是"查向量库"，而是一套完整的文档处理流水线：

原始数据（PDF/网页/数据库...）
        │
        ▼
  Document Loaders      ← 加载原始文档
        │
        ▼
  Text Splitters        ← 切分成小块
        │
        ▼
  Embedding Models      ← 向量化
        │
        ▼
  Vector Stores         ← 存储向量
        │
        ▼
  Retrievers            ← 检索接口
        │
        ▼
      RAG Chain         ← 组合 LLM 生成答案

每一层都可以单独替换，整体是插件化设计。

---

二、Document Loaders：加载数据

LangChain内置了许多种类的文档加载器
文档加载器均继承于BaseLoader类
返回Document类型的对象
load方法一次性批量加载（返回list内含Document对象），如内容过多可能list太大，出现内存溢出问题
lazy_load方法会得到生成器对象，可用for循环依次获取单个Document对象，适用于大文档避免内存存不下。

所有 Loader 都返回 List[Document]，每个 Document 包含 page_content（文本）和 metadata（元信息）。

2.1 常用 Loader

LangChain内置了许多文档加载器，详细参见官方文档：
Document Loaders官方文档：

PyPDFLoader

LangChain内支持许多PDF的加载器，我们选择其中的PyPDFLoader使用。
PyPDFLoader加载器，依赖PyPDF库，所以，需要安装它：
pip install pypdf
PyPDFLoader使用还是比较简单的，如下代码即可快速加载PDF中的文字内容了：

from langchain_community.document_loaders import PyPDFLoader

loader = PyPDFLoader(
    file_path="",   # 文件路径必填
    mode='page',    # 读取模式，可选page（按页面划分不同Document）和single（单个Document）
    password='password', # 文件密码
)
docs = loader.load()

# # PDF
# from langchain_community.document_loaders import PyPDFLoader
# loader = PyPDFLoader("report.pdf")
# docs = loader.load()
# # metadata 自动包含：{"source": "report.pdf", "page": 0}

CSVLoader

# CSV
from langchain_community.document_loaders import CSVLoader
loader = CSVLoader("data.csv", source_column="url")  # 指定哪列作为 source
docs = loader.load()
# loader = CSVLoader(
#     file_path="./xxx.csv",
#     csv_args={
#         "delimiter": ",",   # 指定分隔符
#         "quotechar": '"'     # 指定字符串的引号包裹
#         # 字段列表（无表头使用，有表头勿用会读取首行做为数据）
#         "fieldnames": ["name", "age", "gender"],
#     },
# )

data = loader.load()

JSONLoader

JSONLoader用于将JSON数据加载为Document类型对象。
使用JSONLoader需要额外安装： pip install jq
jq是一个跨平台的json解析工具，LangChain底层对JSON的解析就是基于jq工具实现的。
将JSON数据的信息抽取出来，封装为Document对象，抽取的时候依赖jq_schema语法。

{
  "name": "周杰轮",
  "age": 11,
  "hobby": ["唱", "跳", "RAP"],
  "other": {
    "addr": "深圳",
    "tel": "12332112321"
  }
}

.表示整个JSON对象（根）
[]表示数组
.name表示抽取周杰轮
.hobby表示抽取爱好数组
.hobby[1]或.hobby.[1]表示抽取跳
.other.addr表示抽取地址深圳
.[].得到3个字典
.[].name 表示抽取全部的name，即得到3个name信息

from langchain_community.document_loaders import JSONLoader
loader = JSONLoader(
    file_path="xxx.json",   # 文件路径
    jq_schema=".",          # jq schema语法
    text_content=False,     # 抽取的是否是字符串，默认True
    json_lines=True,        # 是否是JsonLines文件（每一行都是JSON的文件）
)

# 如下是一个典型的JsonLines文件
{"name": "周杰轮", "age": 11, "gender": "男"}
{"name": "蔡依临", "age": 12, "gender": "女"}
{"name": "王力鸿", "age": 11, "gender": "男"}

2.2 懒加载（大文件场景）

# load() 一次性加载所有到内存
# lazy_load() 返回生成器，逐条处理，适合大文件
for doc in loader.lazy_load():
    print(doc.page_content[:100])

三、Text Loader和文本分割器

2.3 TextLoader

TextLoader用于将文本文件加载为Document类型对象。

from langchain_community.document_loaders import TextLoader
loader = TextLoader("xxx.txt", encoding="utf-8")
docs = loader.load()
print(docs)
print(len(docs))	# 结果为1

LLM 有 context window 限制，长文档必须切分。切分策略直接影响检索质量。

3.1 递归字符切分（默认首选）

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,        # 每块最多 1000 字符
    chunk_overlap=200,      # 相邻块重叠 200 字符（防止语义被截断）
    separators=["\n\n", "\n", "。", "，", " ", ""],  # 优先按段落 → 句子 → 词 切分
    # 字符统计依据（函数）
    length_function=len
)

chunks = splitter.split_documents(docs)
print(f"切分后：{len(chunks)} 块")
print(chunks[0].metadata)  # {"source": "report.pdf", "page": 0} 自动继承

💡 chunk_overlap 的作用：如果一个概念正好跨在两块的边界，重叠区域能保证它在某一块中完整出现，避免语义断裂。

3.2 按 Token 切分（精确控制）

from langchain_text_splitters import TokenTextSplitter

# 按实际 token 数切分，避免超出模型限制
splitter = TokenTextSplitter(
    chunk_size=512,      # 每块最多 512 token
    chunk_overlap=50,
)

3.3 Markdown / 代码感知切分

from langchain_text_splitters import MarkdownHeaderTextSplitter

# 按标题层级切分，metadata 自动记录所属章节
headers = [
    ("#", "h1"),
    ("##", "h2"),
    ("###", "h3"),
]
splitter = MarkdownHeaderTextSplitter(headers_to_split_on=headers)
chunks = splitter.split_text(markdown_text)
# chunks[0].metadata → {"h1": "章节名", "h2": "小节名"}

# 代码按函数/类切分
from langchain_text_splitters import Language, RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter.from_language(
    language=Language.PYTHON,
    chunk_size=1000,
    chunk_overlap=100,
)

⚠️ 切分原则：chunk 太大 → 检索精度低，噪音多；chunk 太小 → 语义不完整，上下文丢失。通常 500-1500 字符是合理范围，具体要根据文档类型调整。

---

四、Embeddings：向量化

from langchain_openai import OpenAIEmbeddings
from langchain_community.embeddings import HuggingFaceEmbeddings

# OpenAI（效果好，收费）
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# HuggingFace 本地模型（免费，适合中文）
embeddings = HuggingFaceEmbeddings(
    model_name="BAAI/bge-small-zh-v1.5",  # 中文效果好的模型
    model_kwargs={"device": "cpu"},
    encode_kwargs={"normalize_embeddings": True},
)

# 手动测试向量化
vector = embeddings.embed_query("向量数据库是什么？")
print(len(vector))   # → 1536（OpenAI）或 512（bge-small）

---

五、Vector Stores：向量存储

5.1 常用向量库对比

向量库	特点	适用场景
Chroma	本地文件存储，零配置	开发测试、小项目
FAISS	Meta 出品，纯内存，速度极快	百万级向量，无需持久化
Pinecone	全托管云服务	生产环境，省运维
Weaviate	支持混合搜索（向量+关键词）	需要混合检索场景
pgvector	PostgreSQL 插件	已有 PG 数据库的项目

5.2 统一的增删查接口

from langchain_chroma import Chroma

# 首次：创建并存入文档
vectorstore = Chroma.from_documents(
    documents=chunks,
    embedding=embeddings,
    persist_directory="./chroma_db",  # 持久化路径
    collection_name="my_docs",
)

# 之后：加载已有库
vectorstore = Chroma(
    persist_directory="./chroma_db",
    embedding_function=embeddings,
    collection_name="my_docs",
)

# 增加文档
vectorstore.add_documents(new_chunks)

# 按 metadata 过滤删除
vectorstore.delete(where={"source": "old_report.pdf"})

# 相似度搜索
results = vectorstore.similarity_search("什么是向量数据库", k=4)

# 带分数的搜索（分数越低越相似，L2 距离）
results = vectorstore.similarity_search_with_score("什么是向量数据库", k=4)
for doc, score in results:
    print(f"相似度分数: {score:.4f} | {doc.page_content[:50]}")

5.3 带 metadata 过滤的检索

# 只在特定来源里检索
results = vectorstore.similarity_search(
    "产品价格是多少",
    k=3,
    filter={"source": "price_list.pdf"}  # 只检索价格手册
)

# 多条件过滤（不同向量库语法略有差异）
results = vectorstore.similarity_search(
    query,
    filter={"$and": [{"source": "manual.pdf"}, {"page": {"$gte": 10}}]}
)

---

六、Retrievers：检索接口

Retriever 是 LangChain 统一的检索抽象，输入字符串，输出 List[Document]，是 LCEL 管道中的标准组件。

6.1 从向量库创建 Retriever

# 最基本：相似度检索
retriever = vectorstore.as_retriever(
    search_type="similarity",  # 默认
    search_kwargs={"k": 4},
)

# MMR（最大边际相关）：减少结果重复，提升多样性
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={
        "k": 4,
        "fetch_k": 20,      # 先取 20 个候选，再从中选 4 个最多样的
        "lambda_mult": 0.7, # 0=最多样, 1=最相关，0.5~0.7 是常用值
    }
)

# 相似度分数阈值：低于阈值的结果不返回
retriever = vectorstore.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"score_threshold": 0.75, "k": 4},
)

# 直接调用
docs = retriever.invoke("向量数据库和关系型数据库的区别")

6.2 MultiQueryRetriever：多角度查询

单一查询可能命中偏差，用 LLM 自动生成多个相关查询，合并结果。

from langchain.retrievers import MultiQueryRetriever

retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(),
    llm=llm,
)

# 内部自动生成类似这些查询：
# 原始：  "向量数据库的优缺点"
# 生成1： "向量数据库的优势是什么"
# 生成2： "向量数据库有哪些缺点和局限性"
# 生成3： "向量数据库与传统数据库相比如何"
# → 对每个查询分别检索，结果去重合并
docs = retriever.invoke("向量数据库的优缺点")

6.3 ContextualCompressionRetriever：压缩冗余内容

检索到的 chunk 往往包含大量与问题无关的内容，压缩检索器会让 LLM 过滤掉噪音。

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor

# 压缩器：从 chunk 中只提取与问题相关的句子
compressor = LLMChainExtractor.from_llm(llm)

compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=vectorstore.as_retriever(search_kwargs={"k": 6}),
)

# 先检索 6 个，再压缩提取最相关的部分
docs = compression_retriever.invoke("产品的退款政策是什么？")

6.4 EnsembleRetriever：混合检索（向量 + 关键词）

纯向量检索对精确关键词（人名、型号等）效果差，结合 BM25 关键词检索互补。

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

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

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

# 混合：各取权重（权重之和须为 1）
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6],  # 关键词 40%，向量 60%
)

docs = ensemble_retriever.invoke("iPhone 15 Pro Max 的电池容量")
# → BM25 擅长命中 "iPhone 15 Pro Max" 精确词，向量检索补充语义相关内容

6.5 ParentDocumentRetriever：父子文档检索

用小 chunk 检索（精度高），但返回大 chunk 给 LLM（上下文全）。

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

# 子 chunk：小，用于精确检索
child_splitter = RecursiveCharacterTextSplitter(chunk_size=200)
# 父 chunk：大，用于提供完整上下文
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)

docstore = InMemoryStore()  # 存储父文档（生产用 RedisStore 等）

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

retriever.add_documents(docs)

# 检索时：用小 chunk 向量匹配 → 返回对应的大 chunk
results = retriever.invoke("产品的安装步骤")
# 返回包含安装步骤的完整段落，而不是一个残缺的小片段

---

七、组装 RAG Chain

把所有组件用 LCEL 串联成完整的 RAG 应用。

7.1 基础 RAG

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough

def format_docs(docs: list) -> str:
    return "\n\n---\n\n".join(doc.page_content for doc in docs)

prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一个知识库问答助手。根据下面提供的上下文回答问题。
    如果上下文中没有相关信息，请直接说"我在知识库中没有找到相关信息"，不要编造答案。
    上下文：{context}"""),

    ("human", "{question}"),
])

rag_chain = (
    {
        "context":  retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | prompt
    | llm
    | StrOutputParser()
)

answer = rag_chain.invoke("向量数据库是什么？")

7.2 带来源引用的 RAG

from langchain_core.runnables import RunnableParallel

# 同时返回答案和原始文档来源
rag_chain_with_source = RunnableParallel({
    "answer": rag_chain,
    "sources": retriever,  # 同时返回检索到的原始文档
})

result = rag_chain_with_source.invoke("向量数据库是什么？")
print(result["answer"])
for doc in result["sources"]:
    print(f"来源：{doc.metadata.get('source')} 第{doc.metadata.get('page', '?')}页")

7.3 带记忆的对话 RAG

from langchain_core.prompts import MessagesPlaceholder
from langchain_core.messages import HumanMessage, AIMessage

# 用 LLM 将历史对话和新问题整合成独立问题
contextualize_prompt = ChatPromptTemplate.from_messages([
    ("system", """根据聊天历史和最新问题，将问题改写为一个独立的、不依赖历史上下文就能理解的问题。
    如果问题本身已经完整，直接返回原问题即可。"""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

# 将问题"独立化"（解决指代问题，如"它是什么"→"向量数据库是什么"）
contextualize_chain = contextualize_prompt | llm | StrOutputParser()

def get_contextualized_question(input: dict) -> str:
    if input.get("chat_history"):
        return contextualize_chain.invoke(input)
    return input["input"]

# 完整对话 RAG prompt
qa_prompt = ChatPromptTemplate.from_messages([
    ("system", "根据下面的上下文回答问题。\n\n上下文：{context}"),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
])

conversational_rag = (
    RunnablePassthrough.assign(
        context=RunnableLambda(get_contextualized_question) | retriever | format_docs
    )
    | qa_prompt
    | llm
    | StrOutputParser()
)

# 多轮对话
history = []
def chat(question: str) -> str:
    answer = conversational_rag.invoke({
        "input": question,
        "chat_history": history,
    })
    history.append(HumanMessage(content=question))
    history.append(AIMessage(content=answer))
    return answer

chat("向量数据库是什么？")
chat("它和传统数据库相比有什么优势？")  # "它" 会被自动解析为"向量数据库"

---

八、检索质量优化

问题	原因	解决方案
检索到不相关内容	chunk 太大，语义混杂	缩小 chunk_size，用压缩检索器
精确词汇检索失败（人名、型号）	纯向量检索	加 BM25，用 EnsembleRetriever
检索到的内容上下文不完整	chunk 太小	用 ParentDocumentRetriever
多个 chunk 内容重复	语义过于相似	改用 MMR 检索类型
问题和文档语言/措辞差距大	语义 gap	用 MultiQueryRetriever 多角度查询
答案"幻觉"严重	prompt 没有约束	明确指令"没有相关信息就说不知道"

---

九、完整流水线总结

原始文档
  │ Document Loaders（PDF/网页/DB...）
  ▼
List[Document]（含 metadata）
  │ Text Splitters（递归/Token/Markdown...）
  ▼
小 chunks
  │ Embeddings（OpenAI/BGE/HuggingFace...）
  ▼
向量
  │ Vector Stores（Chroma/FAISS/Pinecone...）
  ▼
向量库
  │ Retrievers（相似度/MMR/混合/父子...）
  ▼
List[Document]（检索结果）
  │ format_docs + prompt
  ▼
RAG Chain（LCEL 管道）→ 最终答案

---