LangChain---RAG核心：向量数据库与高级检索器

• 理解向量数据库的原理、作用及其在RAG中的关键地位。
• 掌握主流本地向量数据库（如Chroma, FAISS）的部署、数据写入与查询。
• 学习如何使用VectorstoreRetriever进行基础的相似度检索。
• 深入了解并实践LangChain中几种高级检索策略（如ParentDocumentRetriever, SelfQueryRetriever, ContextualCompressionRetriever），以提升检索质量。
• 通过LCEL将检索环节无缝集成到整个RAG流程中。

1. 存储： 持久化存储通过Embedding Models生成的文本向量（以及对应的原始文本内容和元数据）。
2. 相似度搜索： 当用户提出问题时，将用户问题向量化，然后在数据库中高效地检索出与问题向量最相似的文档块向量。
3. 扩展性与性能： 针对大规模向量数据的存储和实时查询进行优化，支持高并发。

• 本地/内存型 (适合开发测试、小规模应用)：Chroma, FAISS
• 云服务/分布式型 (适合生产环境、大规模应用)：Pinecone, Weaviate, Qdrant, Milvus, Vectra 等。

from dotenv import load_dotenv
import os
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings
from langchain_core.documents import Document
from langchain_chroma import Chroma
from langchain_core.runnables import RunnableLambda
load_dotenv()


embeddings_model = OpenAIEmbeddings(
    model=os.environ.get("EMBEDDING_MODEL"),
    api_key=os.environ.get("EMBEDDING_API_KEY"),
    base_url=os.environ.get("EMBEDDING_BASE_URL"),
)


# 准备数据 (与第三期类似)
with open("docs/example.txt", "w", encoding="utf-8") as f:
    f.write("LangChain 是一个强大的框架，用于开发由大型语言模型驱动的应用程序。\n")
    f.write("作为一名LangChain教程架构师，我负责设计一套全面、深入且易于理解的LangChain系列教程。\n")
    f.write("旨在帮助读者从入门到精通，掌握LangChain的核心技术和应用。\n")
    f.write("RAG（检索增强生成）是LangChain中的一个关键应用场景。\n")
    f.write("通过RAG，我们可以将LLM与外部知识库相结合。\n")
    f.write("从而让LLM能够回答其训练数据之外的问题。\n")
    f.write("这大大扩展了LLM的应用范围，解决了幻觉和知识过时的问题。\n")
    f.write("LangSmith 是 LangChain 的一个强大工具，用于调试和评估 LLM 应用程序。\n")
    f.write("LCEL 是 LangChain Expression Language 的简称，是构建链条的首选方式。\n")
    f.write("LangGraph 则用于构建具有循环和复杂状态的 Agent。\n")
loader = TextLoader("example.txt", encoding="utf-8")
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=100,
    chunk_overlap=20,
    length_function=len
)


# 步骤 1: 加载和切分文档，得到 Document 块列表
raw_documents = loader.load()
split_documents = text_splitter.split_documents(raw_documents)
print(f"原始文档切分后得到 {len(split_documents)} 个块。")


# print(f"第一个块内容: {split_documents[0].page_content}")
# --- 2. 创建并持久化 Chroma 向量数据库 ---
# from_documents 方法会同时进行 embedding 和存储
# persist_directory 参数用于指定存储路径，这样数据就会被保存到磁盘上，下次可以直接加载
persist_directory = "./chroma_db"
# 如果目录已存在，可以先清理 (仅用于测试)
import shutil
if os.path.exists(persist_directory):
    shutil.rmtree(persist_directory)


print(f"正在创建或加载 Chroma 数据库到 '{persist_directory}'...")
vectorstore = Chroma.from_documents(
    documents=split_documents,
    embedding=embeddings_model,
    persist_directory=persist_directory
)
print("Chroma 数据库创建/加载完成并已持久化。")


# --- 3. 进行相似度检索 (直接使用向量数据库的相似度搜索方法) ---
query = "LangChain是用来做什么的？"
# similarity_search 方法会向量化查询，并在数据库中搜索最相似的文档
found_docs = vectorstore.similarity_search(query, k=2) # k=2 表示返回最相似的2个文档
print(f"\n--- 对查询 '{query}' 的检索结果 (Chroma.similarity_search) ---")
for i, doc in enumerate(found_docs):
    print(f"文档 {i+1} (来源: {doc.metadata.get('source', 'N/A')}, 页码: {doc.metadata.get('page', 'N/A')}):")
    print(doc.page_content)
    print("-" * 30)


# --- 4. 从持久化路径加载数据库 (下次运行时可以直接加载，无需重新创建) ---
print(f"\n--- 从持久化路径重新加载 Chroma 数据库 ---")
loaded_vectorstore = Chroma(
    persist_directory=persist_directory,
    embedding_function=embeddings_model # 注意：加载时也要指定embedding_function
)
reloaded_found_docs = loaded_vectorstore.similarity_search(query, k=1)
print(f"重新加载后检索结果 (部分): {reloaded_found_docs[0].page_content[:100]}...")

• Chroma.from_documents() 是最方便的创建方法，它接收切分后的 Document 列表和 Embedding 模型，自动完成向量化和存储。
• persist_directory 参数至关重要，它让 Chroma 能够将数据保存到磁盘，下次无需重新处理文档。
• vectorstore.similarity_search(query, k=N) 即可进行相似度查询，返回 k 个最相关的 Document 对象。
• Chroma(persist_directory=..., embedding_function=...) 用于从磁盘加载已存在的数据库。

# 假设我们给文档添加了更多元数据，这里修改 example.txt 并重新加载
# 为了简化，我们直接修改 split_documents，为其中一些添加 category
if len(split_documents) > 2:
    split_documents[0].metadata["category"] = "LangChain Core"
    split_documents[1].metadata["category"] = "LCEL"
    split_documents[2].metadata["category"] = "RAG"
    split_documents[3].metadata["category"] = "RAG"


persist_directory="./chroma_db_with_meta"
if os.path.exists(persist_directory):
    shutil.rmtree(persist_directory)


# 重新创建向量库 (或加载后，如果需要更新则需要重新添加)
vectorstore_with_meta = Chroma.from_documents(
    documents=split_documents,
    embedding=embeddings_model,
    persist_directory=persist_directory
)


# 使用元数据过滤进行检索
query_filtered = "LangChain的关键概念"
# filter 参数接受一个字典，定义过滤条件
# "$eq" 表示等于 (equal)，"$in" 表示包含在列表中 (in)
found_docs_filtered = vectorstore_with_meta.similarity_search(
    query_filtered, 
    k=3, 
    filter={"category": "LangChain Core"} # 仅搜索 category 为 "LangChain Core" 的文档
)
print(f"\n--- 对查询 '{query_filtered}' 的元数据过滤检索结果 (category = 'LangChain Core') ---")
for i, doc in enumerate(found_docs_filtered):
    print(f"文档 {i+1} (分类: {doc.metadata.get('category', 'N/A')}):")
    print(doc.page_content)
    print("-" * 30)


# 复杂过滤条件
found_docs_complex_filter = vectorstore_with_meta.similarity_search(
    query_filtered,
    k=3,
    filter={"$or": [{"category": "LangChain Core"}, {"category": "LCEL"}]} # 或关系
)


print(f"\n--- 对查询 '{query_filtered}' 的元数据过滤检索结果 (category = 'LangChain Core' or ''LCEL) ---")
for i, doc in enumerate(found_docs_complex_filter):
    print(f"文档 {i+1} (分类: {doc.metadata.get('category', 'N/A')}):")
    print(doc.page_content)
    print("-" * 30)
# 更多过滤条件请参考 Chroma 文档

1. VectorstoreRetriever (最基础的检索器)
• 这是最直接的 Retriever，它直接从一个 Vectorstore 实例创建。
• 它会将输入查询向量化，然后在其关联的 Vectorstore 中执行相似度搜索。
• 在LCEL链中，Retriever 通常作为字典中的一个键，负责获取上下文。

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


# 从我们上面创建的 vectorstore (或 loaded_vectorstore) 获取一个 retriever
base_retriever = vectorstore.as_retriever(search_kwargs={"k": 2}) # search_kwargs 可以传递给底层的 similarity_search


# 构建一个基础的 RAG 链 (使用 LCEL)
# 输入 {"question": "...", "context": "...(retrieved docs)..."}
rag_prompt = ChatPromptTemplate.from_template("""
请根据提供的上下文回答以下问题。
如果上下文中没有足够的信息，请说明你不知道。
问题: {question}
上下文:
{context}
""")


llm = ChatOpenAI(
    model=os.environ.get("OPENAI_MODEL"),
    temperature=0.9,
    base_url=os.environ.get("OPENAI_BASE_URL"),
    openai_api_key=os.environ.get("OPENAI_API_KEY"),
)


# 核心 LCEL RAG 链
# 1. 接收 {question} 作为输入
# 2. RunnableParallel 会并行处理 "context" (通过 retriever 获取) 和 "question" (直接透传)
# 3. 结果合并为 {"context": List[Document], "question": str}
# 4. prompt 接收这个字典，格式化
# 5. LLM 生成答案
# 6. OutputParser 解析
basic_rag_chain = (
    {"context": base_retriever, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | StrOutputParser()
)
print("\n--- 基础 RAG 链示例 (使用 VectorstoreRetriever) ---")
query_basic = "LangChain是什么？"
response_basic = basic_rag_chain.invoke(query_basic)
print(f"问题: {query_basic}")
print(f"回答: {response_basic}")

• 上下文冗余： 检索到的块虽然相关，但包含大量不必要的细节。
• 语义断裂： 为了满足 chunk_size，一个完整的语义单元被切断。
• 查询模糊： 用户查询本身不够明确，导致检索效果不佳。

1. ParentDocumentRetriever：平衡粒度与上下文
• 问题： 为了精确检索，我们希望 chunk_size 小；但为了提供完整上下文给LLM，又希望 chunk_size 大。
• 解决方案： 存储两种粒度的文档。
• 小块 (Child Chunks)： 用于进行向量化和检索，确保精确匹配。
• 大块 (Parent Documents)： 原始的、更大的完整语义单元。
• 检索时，先用小块进行相似度搜索，找到相关的小块ID，然后通过这些ID获取对应的大块作为最终上下文。

from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore # 内存存储，也可以用 Redis, MongoDB 等
from langchain.text_splitter import RecursiveCharacterTextSplitter


# 1. 定义大块和小块切分器
parent_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=0) # 大块
child_splitter = RecursiveCharacterTextSplitter(chunk_size=100, chunk_overlap=20) # 小块 (用于检索)


# 2. 定义向量存储 (用于小块) 和文档存储 (用于大块，通过ID映射)
vectorstore_for_child = Chroma(
    embedding_function=embeddings_model,
    persist_directory="./chroma_db_child_chunks"
)
doc_store = InMemoryStore() # 存储大块原始文档


# 3. 创建 ParentDocumentRetriever
parent_document_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore_for_child,
    docstore=doc_store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter,
)


# 4. 添加文档
# 注意：这里直接 add_documents，ParentDocumentRetriever 会自动处理切分和存储
print("\n--- ParentDocumentRetriever 示例 (添加文档) ---")
parent_document_retriever.add_documents(raw_documents) # raw_documents 是未切分的原始文档
print("文档已添加到 ParentDocumentRetriever。")


# 5. 进行检索
query_parent = "什么是LangGraph，它和LCEL有什么区别？"
retrieved_parent_docs = parent_document_retriever.invoke(query_parent)
print(f"\n--- 对查询 '{query_parent}' 的 ParentDocumentRetriever 检索结果 ---")
for i, doc in enumerate(retrieved_parent_docs):
    print(f"文档 {i+1} (长度: {len(doc.page_content)}):")
    print(doc.page_content[:200] + "...") # 打印部分内容，通常会比小块大
    print("-" * 30)
# 你会发现这里的文档块长度更长，因为它们是从大块中取出的，包含了更多上下文。

1. SelfQueryRetriever：让LLM生成结构化查询
• 问题： 用户查询可能是自然语言，包含语义和元数据过滤意图，但 similarity_search 只能做语义搜索，无法理解过滤条件。
• 解决方案： 让LLM解析用户的自然语言查询，从中提取出：
• query(查询字符串)： 用于语义搜索的部分。
• filter(过滤条件)： 基于元数据的结构化过滤条件。
• 这需要LLM具备一定的指令遵循和结构化输出能力，以及对元数据字段的理解。

from langchain.chains.query_constructor.base import AttributeInfo
from langchain.retrievers.self_query.base import SelfQueryRetriever


# 定义我们的文档元数据信息
# 这是关键！让LLM知道有哪些元数据字段以及它们的含义
document_content_description = "关于LangChain框架、RAG、LangSmith、LCEL和LangGraph的文档。"
metadata_field_info = [
    AttributeInfo(
        name="category",
        description="文档内容所属的类别，例如 'LangChain Core', 'RAG', 'LangSmith', 'LCEL', 'LangGraph'。",
        type="string",
    ),
    AttributeInfo(
        name="source",
        description="文档的来源文件名称，例如 'example.txt'。",
        type="string",
    ),
]


# 从我们之前创建的 vectorstore_with_meta 获取 SelfQueryRetriever
# 注意：SelfQueryRetriever 需要一个 LLM 来解析查询
self_query_retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore_with_meta, # 带有元数据的向量库
    document_contents=document_content_description,
    metadata_field_info=metadata_field_info,
)
print("\n--- SelfQueryRetriever 示例 ---")
query_self = "关于RAG的关键应用，只从RAG相关的文档中搜索。"
retrieved_self_docs = self_query_retriever.invoke(query_self)
print(f"对查询 '{query_self}' 的 SelfQueryRetriever 检索结果:")
for i, doc in enumerate(retrieved_self_docs):
    print(f"文档 {i+1} (分类: {doc.metadata.get('category', 'N/A')}):")
    print(doc.page_content)
    print("-" * 30)


query_self_2 = "LangGraph是什么？它的来源文件是哪个？" # 假设我们知道是 example.txt
retrieved_self_docs_2 = self_query_retriever.invoke(query_self_2)
print(f"\n对查询 '{query_self_2}' 的 SelfQueryRetriever 检索结果:")
for i, doc in enumerate(retrieved_self_docs_2):
    print(f"文档 {i+1} (分类: {doc.metadata.get('category', 'N/A')}, 来源: {doc.metadata.get('source', 'N/A')}):")
    print(doc.page_content)
    print("-" * 30)

1. ContextualCompressionRetriever：检索后精简上下文
• 问题： 基础检索器返回的文档块可能包含大量与问题不直接相关的冗余信息，浪费Token。
• 解决方案： 在检索到文档块之后，使用一个 BaseLLMCompressor（通常是LLM）对这些块进行“压缩”或“精简”，只保留与查询最相关的部分。
• LLMChainExtractor： 常用的一种压缩器，它会用LLM提取每个文档块中最相关的句子或片段。

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_core.prompts import PromptTemplate
# 1. 定义一个基础检索器 (比如 VectorstoreRetriever)
base_retriever_for_compression = vectorstore.as_retriever(search_kwargs={"k": 5}) # 先多检索一些
# 2. 定义一个 LLMChainExtractor (压缩器)
# 它内部会使用一个LLM来判断哪些内容是相关的
# llm.temperature = 0.0
compressor = LLMChainExtractor.from_llm(llm)
# 3. 创建 ContextualCompressionRetriever
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=base_retriever_for_compression # 传入基础检索器
)
print("\n--- ContextualCompressionRetriever 示例 ---")
query_compression = "LangChain的调试工具叫什么？"
retrieved_compressed_docs = compression_retriever.invoke(query_compression)
# 这里如何输出得都是OUTPUT，可考虑分块优化
print(f"对查询 '{query_compression}' 的 ContextualCompressionRetriever 检索结果:")
for i, doc in enumerate(retrieved_compressed_docs):
    print(f"文档 {i+1} (长度: {len(doc.page_content)}):")
    print(doc.page_content) # 打印被压缩后的内容
    print("-" * 30)
# 对比一下，如果用 base_retriever_for_compression.invoke(query_compression)
# 你会发现原始文档块可能更长，包含更多不直接相关的信息。

• MultiVectorRetriever： 为同一个文档创建多种不同类型的向量（如摘要的向量、内容的向量），在检索时使用最合适的向量。
• EnsembleRetriever： 组合多个不同的检索器（如一个向量检索器和一个关键词检索器），然后融合它们的检索结果。
• Reranking (重排序)： 检索到初步结果后，使用一个独立的Reranking模型（通常是交叉编码器）对这些结果进行二次排序，进一步提升相关性。这不是一个 Retriever 本身，而是 Retriever 的下游优化步骤。

# 假设我们选择使用 self_query_retriever 作为我们的高级检索器
# 当然，你也可以替换成 parent_document_retriever 或 compression_retriever
# 核心 LCEL RAG 链 (与基础 RAG 链类似，只是替换了 retriever)
advanced_rag_chain = (
    {"context": self_query_retriever, "question": RunnablePassthrough()}
    | rag_prompt # 沿用之前的 RAG 提示模板
    | llm
    | StrOutputParser()
)
print("\n--- 高级 RAG 链示例 (使用 SelfQueryRetriever) ---")
query_advanced = "请告诉我关于LCEL的定义和特点，并且仅从LCEL相关的文档中提取。"
response_advanced = advanced_rag_chain.invoke(query_advanced)
print(f"问题: {query_advanced}")
print(f"回答: {response_advanced}")
query_advanced_2 = "LangChain是什么？它在哪个文件中？"
response_advanced_2 = advanced_rag_chain.invoke(query_advanced_2)
print(f"\n问题: {query_advanced_2}")
print(f"回答: {response_advanced_2}")

• 无论你选择哪种 Retriever，只要它符合 BaseRetriever 接口（即能够接受字符串查询并返回 List[Document]），就可以直接放入 LCEL 链中，作为 context 的来源。
• RunnablePassthrough() 依然用于将原始用户问题传递给 rag_prompt 的 question 变量。

• 向量数据库： 理解了其原理，并学会了使用 Chroma 进行向量的存储、持久化和基于元数据的过滤查询。
• 基础检索器： 掌握了 VectorstoreRetriever 的基本用法和LCEL集成。
• 高级检索策略： 深入了解并实践了 ParentDocumentRetriever (平衡粒度与上下文)、 SelfQueryRetriever (LLM生成结构化查询) 和 ContextualCompressionRetriever (检索后精简)。