Langchain模块-DataCollection组件-Retriever检索器

 2025/12/11 

用以查询嵌入到向量数据库中的数据

MultiQueryRetriever请求拆分多索引：将问题拆分出不同的问答
Contextual compression上下文压缩：对每个文件块进行摘要，方便于定位
self-querying元数据过滤器：将请求过滤成请求+过滤器，进行查询
时间加权向量存储检索器：加上时间因素

请求拆分多索引MultiQueryRetriever

通过请求拆分多索，生成多维度的查询输入，其本身是一种增强了的检索器，可以被RetrievalQAWithSourcesChain所引用

# MultiQueryRetriever请求拆分多索引
from langchain.vectorstores import Chroma
from langchain.document_loaders import WebBaseLoader
from langchain_huggingface import HuggingFaceEmbeddings
from dotenv import load_dotenv
from langchain.text_splitter import RecursiveCharacterTextSplitter

load_dotenv()
loader = WebBaseLoader("https://chatgptzhanghao.com/")
data = loader.load()

# 切割文档
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size = 500,
    chunk_overlap = 20
)
splits = text_splitter.split_documents(data)

# 又使用HuggingFaceEmbeddings了，之前报错访问不上，是因为网站被墙了，需要梯子才能上去
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-mpnet-base-v2",
    # sentence-transformers/all-mpnet-base-v2 或 sentence-transformers/all-MiniLM-L6-v2
    model_kwargs={'device': 'cpu'}
)
# 初始化向量数据库
db = Chroma.from_documents(splits,embeddings)

from langchain_deepseek import ChatDeepSeek
from langchain.retrievers.multi_query import MultiQueryRetriever
llm = ChatDeepSeek(model="deepseek-chat")

# 创建多维度提问查询
retriever = MultiQueryRetriever.from_llm(
    retriever=db.as_retriever(),
    llm = llm
)

question = "什么是GPT?"

import logging
logging.basicConfig()
logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)

response = retriever.get_relevant_documents(query=question)
print(len(response))
# print(response)

# INFO:langchain.retrievers.multi_query:Generated queries: ['1. 请解释GPT的基本概念和工作原理。', '2. GPT模型的定义及其主要应用领域有哪些？', '3. 人工智能中的GPT指的是什么，它有哪些技术特点？']
# 5

# 可以从日志看出来，这里将我的问题已经拆分出了三个问题

上下文压缩compress

传统的retriever查询时，会将相关块全部返回，但是一个文件块往往有几百的chunk_size大小，内容非常多，导致查询不精准

上下文压缩索引器本身也是利用LLM，给每个文件块生成摘要

from langchain.text_splitter import CharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain.document_loaders import TextLoader

# 文档加载和分割
documents = TextLoader('story.txt').load()
text_splitter = CharacterTextSplitter(chunk_size=500,chunk_overlap=0)
texts = text_splitter.split_documents(documents)

# 创建db与检索器
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2",model_kwargs={'device': 'cpu'})
retriever = FAISS.from_documents(texts,embeddings).as_retriever()

# 此时的检索器会返回大段文字
print(retriever.invoke("火星地下城怎么样了")[0])
# page_content='在遥远的未来，人类已突破星际航行的限制，却因资源争夺陷入新的危机。年轻的天文学家艾琳在废弃的月球观测站发现了一份古老的星图，记载着传说中能平衡宇宙能量的"星核"坐标。然而，这份星图被加密成三段，分别藏在三个不同文明的遗迹中：
# 艾琳与机器人伙伴K-7潜入火星废弃的地下城，这里曾是机械文明的科研中心。他们破解了由齿轮和激光组成的谜题，获得第一段星图碎片，但触发警报后被迫逃离，碎片被神秘组织"黑曜石"抢走。
# 为追回碎片，艾琳前往木卫二的液态海洋基地。这里曾是生物文明培育共生生命的实验室，她通过与发光水母般的生物建立精神链接，解读出第二段星图，却因能量波动导致基地崩塌，碎片再次遗失。
# 最终，艾琳来到漂浮在土星环中的古老空间站。这里曾是能量文明操控宇宙能量的核心，她必须同时平衡三股相反的能量流才能获取最后一段星图。成功瞬间，空间站开始解体，而"黑曜石"已布下陷阱……
# 当艾琳集齐所有碎片时，发现星图指向的并非星核，而是人类自身——过度依赖外部能量导致文明失衡。真正的解决方案是关闭所有星际引擎，让宇宙自然修复。最终，艾琳选择销毁星图，与K-7驾驶飞船驶向未知，留下悬念：人类能否学会与宇宙共存？' metadata={'source': 'story.txt'}

from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")

# 压缩器
compressor = LLMChainExtractor.from_llm(llm)
# 创建压缩索引器
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=retriever
)

compressd_docs = compression_retriever.invoke("火星地下城怎么样了")
print(compressd_docs)

# [Document(metadata={'source': 'story.txt'}, page_content='艾琳与机器人伙伴K-7潜入火星废弃的地下城，这里曾是机械文明的科研中心。他们破解了由齿轮和激光组成的谜题，获得第一段星图碎片，但触发警报后被迫逃离，碎片被神秘组织"黑曜石"抢走。')]

元数据过滤器self-querying（需要openai

自查询检索器是一种利用llm链编写自我查询的检索器，比如”我想看一部昆汀导演的电影”，检索器会将我的问题分解为一个查询和一个过滤器，查询”电影”，过滤”导演是昆汀”

需要使用langchain.retrievers.self_query.base.SelfQueryRetriever类，这个类支持openai+chroma，但是我因为使用的是HuggingFace免费的嵌入式模型，需要使用Weaviate+HuggingFace

官方文档：https://docs.weaviate.io/weaviate/quickstart?import=vectorization#create-a-collection

本例是失败了，因为weaviate最新的4.x的用法与langchain的Weaviate不兼容，运行到后面就报错了

pip install weaviate-client[agent]
docker run -itd -p 8080:8080 -p 50051:50051 cr.weaviate.io/semitechnologies/weaviate:1.34.4

from langchain.schema import Document  
from langchain.embeddings import HuggingFaceEmbeddings  
from langchain.document_loaders import TextLoader
import weaviate

# 准备向量数据库
# 初始化Weaviate客户端
client = weaviate.connect_to_local(
    host="localhost",
    port=8080,
    grpc_port=50051,
    skip_init_checks=True
)
collection_name = "Movie" # 创建集合

embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2",model_kwargs={'device': 'cpu'})
docs = TextLoader("movie_document.txt").load()

import lark
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo
from langchain.vectorstores import Weaviate

# 使用AttributeInfo定义元数据结构,告诉自查询检索器你的元数据中有哪些字段
metadata_field_info = [
    AttributeInfo(
        name="title",
        description="电影的名称",
        type="string",
    ),
    AttributeInfo(
        name="year",
        description="电影的上映年份",
        type="integer",
    ),
    AttributeInfo(
        name="rating",
        description="电影的评分，从0到10",
        type="float",
    ),
    AttributeInfo(
        name="director",
        description="电影的导演",
        type="string",
    ),
    AttributeInfo(
        name="genre",
        description="电影的类型，如：动作、犯罪、西部等",
        type="string",
    ),
]
document_content_description = "电影摘要"
index_name = "Movie"  # 索引名称
text_key = "content"  # 存储文档文本的字段名
attributes = [info.name for info in metadata_field_info]

# 初始化Weaviate向量存储
db = Weaviate(
    client=client,
    index_name=index_name,
    text_key=text_key,
    embedding=embeddings,
    attributes=attributes  # 指定要查询的属性
)

db.add_documents(docs) # 添加文档

load_dotenv()
llm = ChatDeepSeek(model="deepseek-chat")

# 创建查询器，将请求进行拆分并筛选
retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=db,
    document_contents=document_content_description,
    metadata_field_info=metadata_field_info,
    verbose=True,
)

print(retriever.invoke("哪些电影和犯罪有关？"))

时间加权向量存储检索器

TimeWeightedVectorStoreRetriever不仅考虑语义相似度，还考虑时间因素。最近访问过的文档会被优先返回。

语义越接近，衰减率越低，过去的小时数越小，越先检索到

import faiss
from datetime import datetime,timedelta
from langchain.docstore import InMemoryDocstore
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.retrievers import TimeWeightedVectorStoreRetriever
from langchain.schema import Document
from langchain.vectorstores import FAISS

# 初始化HuggingFace嵌入模型
embeddings = HuggingFaceEmbeddings(model_name="sentence-transformers/all-mpnet-base-v2",model_kwargs={'device': 'cpu'})
embedding_size = 768

# IndexFlatL2是简单的暴力搜索索引，适合小规模数据
index = faiss.IndexFlatL2(embedding_size)

# 创建FAISS向量存储
vectorstore = FAISS(embeddings.embed_query,index,InMemoryDocstore({}),{})

# 创建时间加权的向量检索器，时间衰减率
retriever = TimeWeightedVectorStoreRetriever(vectorstore=vectorstore,decay_rate=.00000001,k=1)

# 计算昨天的日期时间
yesterday = datetime.now() - timedelta(days=1)
# 向检索器添加文档
retriever.add_documents([Document(page_content="hello,world",metadata={"last_accessed_at": yesterday})])
retriever.add_documents([Document(page_content="hello,food")])

print(retriever.invoke("hello"))

# 如果衰减率很小，就会搜索出hello,world
# [Document(metadata={'last_accessed_at': datetime.datetime(2025, 12, 11, 11, 27, 34, 763851), 'created_at': datetime.datetime(2025, 12, 11, 11, 27, 34, 390809), 'buffer_idx': 0}, page_content='hello,world')]

# 如果衰减率比较大，就会搜索出hello,food，因为hello,world是昨天的数据