如何为ConversationalRetrievalChain动态添加Chroma向量库元数据过滤？

问题背景 #

我尝试为底层的Chroma向量库添加元数据过滤，初始化代码如下：

db = Chroma.from_documents(texts, embeddings)

静态设置过滤条件的方式可以正常生效，但需要为每个过滤条件创建单独的链，不够灵活：

qa = ConversationalRetrievalChain.from_llm(
    OpenAI(openai_api_key=get_random_key(OPENAI_API_KEY_POOL), cache=True, temperature=0),
    VectorStoreRetriever(vectorstore=db, search_kwargs={"filter": {"source": "data/my.pdf"}}),
    verbose=True,
    return_source_documents=True) 
result = qa({"question": query, "chat_history": []})

但尝试向已创建的链动态传入filter参数时，没有任何效果，返回的仍是数据库中所有文档的结果：

qa = ConversationalRetrievalChain.from_llm(
    OpenAI(openai_api_key=get_random_key(OPENAI_API_KEY_POOL), cache=True, temperature=0),
    VectorStoreRetriever(vectorstore=db),
    verbose=True,
    return_source_documents=True)

filter = {'source': 'my.pdf'}

result = qa({"question": query, "chat_history": [], "filter": filter})

想知道是否遗漏了配置，或者必须扩展LangChain现有类/修改源码才能实现动态过滤？

解决方案 #

默认的ConversationalRetrievalChain不会自动处理输入中的filter参数，要实现动态过滤，有三种可行方式：

方式1：动态修改Retriever的search_kwargs #

每次调用前直接修改Retriever的search_kwargs属性，无需重新创建链：

qa = ConversationalRetrievalChain.from_llm(
    OpenAI(openai_api_key=get_random_key(OPENAI_API_KEY_POOL), cache=True, temperature=0),
    VectorStoreRetriever(vectorstore=db),
    verbose=True,
    return_source_documents=True)

# 动态设置过滤条件
qa.retriever.search_kwargs["filter"] = {'source': 'my.pdf'}
result = qa({"question": query, "chat_history": []})

# 切换其他过滤条件
qa.retriever.search_kwargs["filter"] = {'source': 'another.pdf'}
result2 = qa({"question": query2, "chat_history": []})

方式2：自定义扩展ConversationalRetrievalChain #

如果需要更优雅地通过输入参数传递过滤条件，可以自定义链类，让它支持从输入中读取filter并应用到检索器：

from langchain.chains.conversational_retrieval.base import ConversationalRetrievalChain

class DynamicFilterConversationalRetrievalChain(ConversationalRetrievalChain):
    def _get_relevant_documents(self, question: str, *, run_manager, **kwargs) -> list:
        # 从kwargs中获取filter参数
        filter_param = kwargs.get("filter")
        if filter_param:
            # 临时修改search_kwargs
            original_filter = self.retriever.search_kwargs.get("filter")
            self.retriever.search_kwargs["filter"] = filter_param
            docs = super()._get_relevant_documents(question, run_manager=run_manager, **kwargs)
            # 恢复原过滤条件
            if original_filter is not None:
                self.retriever.search_kwargs["filter"] = original_filter
            else:
                del self.retriever.search_kwargs["filter"]
            return docs
        return super()._get_relevant_documents(question, run_manager=run_manager, **kwargs)

# 使用自定义链
qa = DynamicFilterConversationalRetrievalChain.from_llm(
    OpenAI(openai_api_key=get_random_key(OPENAI_API_KEY_POOL), cache=True, temperature=0),
    VectorStoreRetriever(vectorstore=db),
    verbose=True,
    return_source_documents=True)

# 动态传入filter参数
filter = {'source': 'my.pdf'}
result = qa({"question": query, "chat_history": [], "filter": filter})

方式3：使用RunnablePassthrough构建自定义链（LangChain v0.1+） #

如果使用LangChain v0.1及以上版本，可以用Runnable组合的方式更灵活地处理动态参数：

from langchain_core.runnables import RunnablePassthrough
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain
from langchain_core.prompts import ChatPromptTemplate

# 创建检索器
retriever = VectorStoreRetriever(vectorstore=db)

# 构建prompt
prompt = ChatPromptTemplate.from_messages([
    ("system", "根据提供的文档回答问题：\n{context}"),
    ("human", "{question}")
])

# 创建文档处理链
document_chain = create_stuff_documents_chain(OpenAI(cache=True, temperature=0), prompt)

# 构建支持动态filter的检索链
def retrieve_with_filter(inputs):
    # 从输入中获取filter
    filter_param = inputs.get("filter")
    # 动态设置检索器的过滤条件
    retriever.search_kwargs["filter"] = filter_param
    return retriever.invoke(inputs["question"])

retrieval_chain = (
    RunnablePassthrough.assign(context=retrieve_with_filter)
    | document_chain
)

# 调用链
result = retrieval_chain.invoke({
    "question": query,
    "filter": {'source': 'my.pdf'}
})

内容的提问来源于stack exchange，提问作者Anastasiia Iurshina