如何为基于OpenSearch的SelfQueryRetriever添加额外查询过滤器？

针对你用OpenSearch向量存储实现SelfQueryRetriever时，需要追加UI筛选、角色限制这类固定/动态过滤器的需求，这里有几个实用的实现方式：

1. 继承SelfQueryRetriever重写核心方法 #

直接继承原Retriever类，在生成原始过滤器后手动合并额外条件，逻辑清晰且长期可用：

from langchain.retrievers.self_query.base import SelfQueryRetriever

class CustomSelfQueryRetriever(SelfQueryRetriever):
    def _get_relevant_documents(self, query: str, *, run_manager):
        # 获取Retriever自动生成的原始查询结果和过滤器
        super()._get_relevant_documents(query, run_manager=run_manager)
        structured_query = self._query_constructor.invoke(query)
        original_filter = structured_query.filter

        # 定义你的额外过滤器（示例：角色限制+UI价格筛选）
        extra_filter = {
            "bool": {
                "must": [
                    {"term": {"user_role": "editor"}},
                    {"range": {"price": {"lte": 200}}}
                ]
            }
        }

        # 合并过滤器：原始过滤器和额外条件用AND逻辑组合
        if original_filter:
            combined_filter = {
                "bool": {"must": [original_filter, extra_filter]}
            }
        else:
            combined_filter = extra_filter

        # 用合并后的过滤器重新执行查询
        return self.vector_store.similarity_search(
            query,
            k=self.k,
            filter=combined_filter
        )

这个方案适合权限控制这类需要全局生效的固定过滤器，继承后直接替换原Retriever即可。

2. 封装调用函数动态传入过滤器 #

如果需要根据不同场景动态传参（比如UI的实时筛选条件），可以封装一个调用函数，在每次查询时合并过滤器：

def retrieve_with_custom_filters(retriever, user_query, extra_filter):
    # 让QueryConstructor生成原始过滤器
    structured_query = retriever._query_constructor.invoke(user_query)
    original_filter = structured_query.filter

    # 合并过滤器逻辑
    if original_filter:
        combined_filter = {"bool": {"must": [original_filter, extra_filter]}}
    else:
        combined_filter = extra_filter

    # 执行合并后的查询
    return retriever.vector_store.similarity_search(
        user_query,
        k=retriever.k,
        filter=combined_filter
    )

# 使用示例：传入UI选择的分类筛选
ui_filter = {"term": {"category": "electronics"}}
docs = retrieve_with_custom_filters(your_selfquery_retriever, "蓝牙音箱推荐", ui_filter)

这种方式灵活性最高，适合需要动态变更过滤条件的场景。

3. 优化提示词强制生成固定过滤器 #

如果你不想修改代码逻辑，可以在QueryConstructor的提示词末尾追加固定要求，让LLM自动生成包含额外条件的过滤器：

from langchain.chains.query_constructor.base import AttributeInfo

# 原属性定义不变
metadata_field_info = [
    AttributeInfo(
        name="category",
        description="产品分类，比如电子产品、家居用品",
        type="string"
    ),
    # 其他属性...
]

# 修改提示词，追加固定要求
prompt = """根据用户查询生成对应的查询语句和元数据过滤器，注意：无论用户查询内容是什么，必须始终添加过滤器：user_role = 当前用户的角色，以及category = UI选中的分类"""

# 构造QueryConstructor时传入修改后的提示词
query_constructor = self_query_constructor.from_llm(
    your_llm,
    your_vector_store,
    metadata_field_info,
    prompt=prompt
)

这个方案适合过滤器规则相对固定的场景，但灵活性较差，动态变更条件时需要重新生成提示词。

内容的提问来源于stack exchange，提问作者Luis Leal