LangChain新版本中SelfQueryRetriever是否废弃？替代工具及元数据查询方法

一、SelfQueryRetriever状态确认 #

SelfQueryRetriever确实在LangChain v0.1及以上版本中被废弃，官方已从正式文档中移除该组件，不再维护更新。

二、针对source元数据的替代实现方案 #

你的场景是基于source元数据（文档名称）筛选特定文档进行问答，以下是两种实用的替代方法：

方法1：结构化输出解析器提取元数据+向量检索器过滤 #

先通过大模型解析用户查询中的目标文档名称，再用该条件过滤向量库中的文档，最后完成问答。

示例代码：

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import Chroma
from langchain_core.runnables import RunnablePassthrough

# 定义结构化输出的数据模型
class DocumentQuery(BaseModel):
    source: str = Field(description="需要查询的文档名称，比如document_1、document_5")
    question: str = Field(description="针对该文档的具体问题")

# 初始化输出解析器
parser = JsonOutputParser(pydantic_object=DocumentQuery)

# 构建提示词
prompt = ChatPromptTemplate.from_messages([
    ("system", "从用户查询中提取目标文档名称（source）和具体问题（question），严格按指定格式输出JSON。"),
    ("user", "{query}\n{format_instructions}")
]).partial(format_instructions=parser.get_format_instructions())

# 初始化大模型
llm = ChatOpenAI(model="gpt-3.5-turbo")

# 构建解析链
parse_chain = prompt | llm | parser

# 假设已初始化向量存储（Chroma示例）
vectorstore = Chroma(persist_directory="./chroma_db", embedding_function=your_embedding)

# 定义检索+问答链
def retrieve_and_answer(parsed_query):
    # 过滤出指定source的文档
    retriever = vectorstore.as_retriever(search_kwargs={"filter": {"source": parsed_query["source"]}})
    # 构建问答提示
    qa_prompt = ChatPromptTemplate.from_messages([
        ("system", "根据提供的文档片段回答问题：\n{context}"),
        ("user", "{question}")
    ])
    qa_chain = (
        {"context": retriever, "question": RunnablePassthrough()}
        | qa_prompt
        | llm
    )
    return qa_chain.invoke(parsed_query["question"])

# 调用示例
user_query = "document_1有哪些条款？"
parsed_result = parse_chain.invoke({"query": user_query})
answer = retrieve_and_answer(parsed_result)
print(answer.content)

方法2：FilterRetriever直接过滤 #

如果查询中source格式固定（比如都是document_数字），可直接通过字符串匹配提取source，再用FilterRetriever过滤：

from langchain_core.retrievers import FilterRetriever

# 提取source的简单函数（适用于格式固定的场景）
def extract_source(query):
    import re
    match = re.search(r'document_\d+', query)
    return match.group() if match else None

# 构建过滤检索器
retriever = FilterRetriever(
    vectorstore=vectorstore,
    filter=lambda doc: doc.metadata["source"] == extract_source(user_query)
)

# 后续问答链同方法1

三、适配你的查询场景 #

针对给出的示例查询：

• 对于“document_1有哪些条款？”：提取source为document_1，过滤向量库中所有该来源的文档，再基于这些文档回答条款相关问题
• 对于“给出document_5中的总金额。”：提取source为document_5，过滤后检索相关片段，再提取总金额信息

内容的提问来源于stack exchange，提问作者Augusto Firmo