如何在LangChain的PGVector中筛选最新文件的查询结果

核心逻辑 #

文件以timestamp_file_name格式存储，需从文件名提取时间戳作为元数据存入PGVector，查询时优先筛选时间戳最大的文档（即最新上传文件）的匹配内容。

步骤1：存储文档时注入时间戳元数据 #

上传并加载文档时，从文件名解析时间戳，将其作为元数据附加到每个Document对象中：

import os
from langchain.document_loaders import PyPDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter

def load_and_store_document(auth_id, openai_key, file_path):
    # 从文件名提取时间戳
    filename = os.path.basename(file_path)
    timestamp = int(filename.split("_")[0])
    
    # 加载并拆分文档
    loader = PyPDFLoader(file_path)
    documents = loader.load()
    text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
    split_docs = text_splitter.split_documents(documents)
    
    # 为每个文档块添加时间戳元数据
    for doc in split_docs:
        doc.metadata.update({"timestamp": timestamp, "source": filename})
    
    # 存入PGVector
    vector_store = get_vector_store(auth_id, openai_key)
    vector_store.add_documents(split_docs)

步骤2：PGVector检索时筛选最新文件 #

有两种实现方式：

方式一：先查询最大时间戳，再过滤检索 #

先通过PGVector的底层连接查询当前集合下的最大时间戳，再以此为条件筛选文档：

import psycopg2
from psycopg2.extras import RealDictCursor

def get_latest_timestamp(auth_id):
    conn = psycopg2.connect(CONNECTION_STRING)
    cursor = conn.cursor(cursor_factory=RealDictCursor)
    # 查询当前collection的最大时间戳
    cursor.execute("""
        SELECT MAX(metadata->>'timestamp') as max_ts 
        FROM langchain_pg_embedding 
        WHERE collection_id = (SELECT uuid FROM langchain_pg_collection WHERE name = %s)
    """, (auth_id,))
    result = cursor.fetchone()
    conn.close()
    return int(result['max_ts']) if result['max_ts'] else None

# 获取最新时间戳
latest_ts = get_latest_timestamp(auth_id)

# 构建检索器：仅筛选最新时间戳的文档
retriever = vector_store.as_retriever(
    search_type=SEARCH_TYPE,
    search_kwargs={
        **SEARCH_KWARGS,
        "filter": {"timestamp": latest_ts}
    }
)

方式二：检索后过滤最新文件内容 #

若需先获取所有匹配文档再筛选，可在返回结果后按时间戳分组，保留最新文件的内容：

chain_response = chain(user_input)

# 提取来源文档，按文件分组并筛选最新文件
source_docs = chain_response['source_documents']
# 按文件名分组，记录每个文件的时间戳
file_ts_map = {}
for doc in source_docs:
    filename = doc.metadata['source']
    ts = doc.metadata['timestamp']
    if filename not in file_ts_map or ts > file_ts_map[filename]:
        file_ts_map[filename] = ts

# 找到时间戳最大的文件
latest_file = max(file_ts_map, key=file_ts_map.get)
# 过滤出仅来自最新文件的文档
latest_docs = [doc for doc in source_docs if doc.metadata['source'] == latest_file]

# 重新生成基于最新文件的回答
from langchain.chains.qa_with_sources import load_qa_with_sources_chain

refined_chain = load_qa_with_sources_chain(llm, chain_type="stuff")
refined_response = refined_chain({
    "input_documents": latest_docs,
    "question": user_input
}, return_only_outputs=True)

步骤3：调整RetrievalQA链配置 #

将上述构建的带筛选条件的检索器替换原链中的检索器：

chain = RetrievalQAWithSourcesChain.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,  # 使用带最新时间戳筛选的检索器
    chain_type_kwargs=chain_type_kwargs,
    return_source_documents=True,
    memory=window_memory
)
chain_response = chain(user_input)

内容的提问来源于stack exchange，提问作者Andrea Bency