Passing filter through ConversationalRetrievalChain to the underlying vector store

[iurshina]

Feature request

It seems that right now there is no way to pass the filter dynamically on the call of the ConversationalRetrievalChain, the filter can only be specified in the retriever when it's created and used for all the searches.

    qa = ConversationalRetrievalChainPassArgs.from_llm(
        OpenAI(..),
        VectorStoreRetriever(vectorstore=db, search_kwargs={"filter": {"source": "my.pdf"}}))

I had to extend in my code both ConversationalRetrievalChain and VectorStoreRetriever just to pass the filter to the vectore store like this:

    qa = ConversationalRetrievalChainPassArgs.from_llm(
        OpenAI(...),
        VectorStoreRetrieverWithFiltering(vectorstore=db),
)
    result = qa({"question": query, "chat_history": [], "filter":  {"source": "my.pdf"}}})

Is it really the case or am I missing something? And if it the case, shouldn't it be fixed? (I could provide a pull request)

Motivation

To be able to filter dynamically in ConversationalRetrievalChain since vector stores allow this.

Your contribution

I can provide a PR if my understanding is confirmed.

[dosubot[bot]]

🤖

Thank you for your interest in contributing to LangChain! Your understanding is correct, currently there is no way to pass the filter dynamically on the call of the ConversationalRetrievalChain. It's great to hear that you are willing to provide a pull request to address this issue.

Please go ahead and create the pull request with your proposed solution. The maintainers will review it and decide if it should be merged. Your contribution will be greatly appreciated!

---

This response is meant to be useful and save you time. It is not meant to be a precise solution, but rather a starting point for your own research.

Help me be more useful! Please leave a 👍 if this is helpful and 👎 if it is irrelevant.

If you want to continue the conversation, start your reply with @dosu-bot.

[fdurant]

Hi @iurshina , I had a similar issue and worked around it as follows:

mychains.py

from langchain.chains import ConversationalRetrievalChain
from pydantic import Extra

class MyConversationalRetrievalChain(ConversationalRetrievalChain):
    class Config:
        extra = (
            Extra.ignore
        )  # THIS IS NEEDED TO ALLOW FOR ARBITRARY KEYWORD ARGUMENTS IN THE CONSTRUCTOR
        arbitrary_types_allowed = True
        allow_population_by_field_name = True

session.py

import weaviate
from langchain.memory import ConversationBufferMemory
from langchain.retrievers import WeaviateHybridSearchRetriever
from config import (index_name_vectorstore, text_key_vectorstore, url_vectorstore)
from langchain.vectorstores import Weaviate

memory = ConversationBufferMemory(memory_key="chat_history", input_key="question", return_messages=True)
db_client = weaviate.Client(url_vectorstore)
db = Weaviate(
    client=db_client,
    index_name=index_name_vectorstore,
    text_key=text_key_vectorstore,
)
retriever = WeaviateHybridSearchRetriever(
    client=db_client,
    index_name=index_name_vectorstore,
    text_key=text_key_vectorstore,
    alpha=0.8,
    k=5,
    attributes=[],
)
where_filter = {"path": "nr_words", "operator": "GreaterThanEqual", "valueInt": 3}
retriever_kwargs = {"where_filter": where_filter}

myapp.py

from session import memory, retriever, retriever_kwargs

prompt_template = "..."

conversation = MyConversationalRetrievalChain.from_llm(
    llm=llm,
    retriever=retriever,
    memory=memory,
    verbose=True,
    combine_docs_chain_kwargs={"prompt": prompt_template},
    return_source_documents=False,
    **retriever_kwargs, # NEWLY ADDED
)

I hope this can serve as inspiration for a fix. Happy to help in testing/debugging your fix on my code.

[iurshina]

Hi fdurant, Thank you for the input. But if I'm not mistaken, you pass the filter when you create the chain and then it's fixed? I want to pass the filer when I call the chain: conversation(...).

I ended up doing it like this in my code:

class ConversationalRetrievalChainPassArgs(ConversationalRetrievalChain):

    def _get_docs(self, question: str, inputs: Dict[str, Any], *,
        run_manager: CallbackManagerForChainRun) -> List[Document]:
        """Get docs."""
        docs = self.retriever._get_relevant_documents(
            question, inputs['filter']
        )
        return self._reduce_tokens_below_limit(docs)

class VectorStoreRetrieverWithFiltering(VectorStoreRetriever):

    def _get_relevant_documents(self, query: str, filter: dict) -> List[Document]:
        if self.search_type == "similarity":
            docs = self.vectorstore.similarity_search(query, filter=filter, **self.search_kwargs)
        ...
        return docs

It allows me to pass the filter to the chain and then to the db. It just seems to be a very trivial case not worth of two inheritances by a user when everything is there and can be just take from the inputs :)

[pooriaarab]

@iurshina any idea on how to do this dynamic filtering based on conversational Retrieval Chain in Flowise?

How can I translate your python langchain code to Flowise basically?

e.g. Can we dynamically create metadata filters? Meaning for a conversational retrieval tool, intead of manually assigning metadata filters and description (e.g. search and return anything related to michigan workers, use this metadata filter {state: michigan, category: workers} from this pinecone index), the agent would correctly identify which state and category it is (e.g. search and return anything related to {state} {category})?

If I manually create 50 * 20 (e.g. 50 coutries, 20 categories) retrieval tools with metadata filters, there's a high chance i might miss something (e.g. a particular metadata filter). I want the chain to dynamically identify the metadata filter

[MissArtemis]

@iurshina

Hi, I meet the same problem. Could u please provide more details about how to do with it? I use VectorStoreRetrieverMemory + LLMChain

[felixthekraut]

This would be a useful feature.

[isabellaaquino]

Also looking for a similar functionality. I'm currently trying to search the same set of information in different documents, and filtering by the source usually returns better answers

[Reios0]

Hi @iurshina, just wondering if you are working on the PR? If not, then I would love to try my hands on it!

[iurshina]

Hi @Reios0, no, I am not working on this, I was waiting for an approval from the maintainers to start. Feel free to work on it :)

[Kick28]

@dosu-bot Hi! I encountered the same problem, so your code was helpful but I'm confused about how to integrate a new filter argument into the chain.

I would appreciate it if someone could help me resolve this issue. Custom chain and retriever:

class RetrievalQAFilter(RetrievalQA):
    def _get_docs(
            self,
            question: str,
            filter: Dict[str, str] = None,
            *,
            run_manager: CallbackManagerForChainRun
    ) -> List[Document]:
        '''Get docs.'''
        if filter is not None:
            return self.retriever._get_relevant_documents(
                question, filter
            )
        else:
            return self.retriever._get_relevant_documents(
                question
            )

class VectorStoreRetrieverFilter(VectorStoreRetriever):
    def _get_relevant_documents(
        self,
        query: str,
        filter: Dict[str, str] = None
    ) -> List[Document]:
        if filter is not None:
            if self.search_type == "similarity":
                docs = self.vectorstore.similarity_search(query, filter=filter, **self.search_kwargs)
        else:
            if self.search_type == "similarity":
                docs = self.vectorstore.similarity_search(query, **self.search_kwargs)
        return docs

Then, I initialize RetrievalQA Chain like this:

chain_type_kwargs = {"prompt": retrieval_prompt}

qa = RetrievalQAFilter.from_chain_type(
            llm=<some_llm>,
            chain_type="stuff",
            retriever=VectorStoreRetrieverFilter(vectorstore=FAISS.from_documents(loader.load(), embedding_engine)),
            chain_type_kwargs=chain_type_kwargs
)

And finally, I call the chain to filter documents by the title key. context = qa(<user_query>, {"title": "<some_name>"})

But the problem is the filter argument is not passed to the _get_docs(). How can I fix it? Thanks in advance!

[mandgie]

Any update on this feature? Would be really helpful.

[Reios0]

Hi @mandgie, I haven't been able to dedicate as much time as I wanted for this issue, but I should be able to push out a PR next week!

[Kick28]

@mandgie @Reios0 Well, I solved the issue for the RetrievalQA - we need to override _call() method as well. I just need to solve some bug - somehow one of the documents is not retrieving, even though it exists and the metadata is correct.

[Kick28]

So, if anyone is trying to solve this issue for RetrievalQA, here is the solution: We should override the next methods.

from typing import Any, List, Dict, Union
from langchain.callbacks.manager import CallbackManagerForChainRun
from langchain.chains import RetrievalQA
from langchain.docstore.document import Document
from langchain.schema import Document
from langchain.vectorstores.base import VectorStoreRetriever
import inspect

class RetrievalQAFilter(RetrievalQA):
    '''Custom RetrievalQA Chain with filter functionality.'''
    def _get_docs(
        self,
        question: str,
        filter: Dict[str, Any] = None,
        *,
        run_manager: CallbackManagerForChainRun
    ) -> List[Document]:
        '''Overrided get docs.'''
        # Check if filter is provided
        if filter is not None:
            return self.retriever._get_relevant_documents(question, filter)
        else:
            return self.retriever._get_relevant_documents(question)

    def _call(
        self,
        inputs: Dict[str, Any],
        run_manager: CallbackManagerForChainRun = None
    ) -> Dict[str, Any]:
        '''Overrided call method so we can provide filter.'''
        _run_manager = run_manager or CallbackManagerForChainRun.get_noop_manager()
        question = inputs[self.input_key]
        # Get filter conditions
        filter = inputs['filter']
        accepts_run_manager = (
            'run_manager' in inspect.signature(self._get_docs).parameters
        )
        if accepts_run_manager:
            docs = self._get_docs(question, filter, run_manager=_run_manager)
        else:
            docs = self._get_docs(question, filter)
        answer = self.combine_documents_chain.run(
            input_documents=docs, question=question, callbacks=_run_manager.get_child()
        )

        if self.return_source_documents:
            return {self.output_key: answer, "source_documents": docs}
        else:
            return {self.output_key: answer}

class VectorStoreRetrieverFilter(VectorStoreRetriever):
    '''Custom vectorstore retriever with filter functionality.'''
    def _get_relevant_documents(
        self,
        query: str,
        filter: Dict[str, Any] = None
    ) -> List[Document]:
        '''Overrided method with filter functionality.'''
        # Filter is provided
        if filter is not None:
            # Only similarity search is implemented for now
            if self.search_type == "similarity":
                docs = self.vectorstore.similarity_search(query, filter=filter, **self.search_kwargs, fetch_k=30)
        else:
            # Filter is not provided
            if self.search_type == "similarity":
                docs = self.vectorstore.similarity_search(query, **self.search_kwargs)
        return docs

The we can initialize our chain and execute it like this:

loader = <some_loader>
index = VectorStoreRetrieverFilter(vectorstore=FAISS.from_documents(loader.load(), <embedding_engine>))
qa = RetrievalQAFilter.from_chain_type(
            llm=<llm>,
            chain_type="stuff",
            retriever=index,
            chain_type_kwargs=<some_kwargs>
)
filter = {"<metadata>": <value>}
context = qa({"query": last_comment, "filter": filter})

Also, if you are using FAISS be aware that this VectorDB doesn't support native filtering, so we need first to fetch some documents and then filter them using the provided filter. I encountered this issue yesterday, where some of my documents with the same metadata tag were not retrieving. We should increase fetch_k to gather more documents and filter them.

I hope this will be helpful :) (idk, but it seems like ConversationalRetrievalChain should have the same logic).

[christopherwerner]

PR and Workaround:

I think I’ve found the neatest way to add this enhancement in the current code flow, taking advantage of as much of the current behavior as possible, which I've submitted in a PR: https://github.com/langchain-ai/langchain/pull/13352

My PR lets you pass search_kwargs into the inputs, including, say, a filter, such that they are passed down to the vector store and merged with any set during construction of the retriever. There's an example in the documentation update included in the PR.

Workaround: In the meantime, for a workaround, you can use the construction-time search_kwargs on the retriever, and simply update them (change the values in the existing dict) just before you query.

This is hacky, and not threadsafe, so if you have a multithreaded application where the same retriever is used for concurrent queries with different filters - you'd need to add some primitive locking. But it's good enough for simple cases and POCs (until/unless my PR is accepted).

Here's how I'm doing it:

In my class that invokes the retriever I create an instance variable that holds a dict for search_kwargs

      self._search_kwargs = {}

Then I create the chain as usual, but I pass my dict instance into the method that constructs the retriever (I've prepared my llm, vector_store, etc, earlier, and I'm holding this qa_chain on another instance variable. The key is the line where the retriever is set)

        self.qa_chain = ConversationalRetrievalChain.from_llm(
            some_llm,
            retriever=some_vector_store.as_retriever(search_kwargs=self._search_kwargs),
            return_source_documents=True
            memory=some_memory,
        )

Then before each call to the chain with a new query, I set the filter on the search_kwargs dict:

    where_filter = { "author": "paulgraham" }
    self._retriever.search_kwargs["filter"] = where_filter
    qa_response = qa_chain({"question": query, "chat_history": history})

(I actually build the where_filter from other params, but for simplicity I'm showing a simple where filter that works in Chroma and matches my example in the documentation I added to the PR)

Warning: Again, if another thread came through using the same qa_chain, I'd have a problem here (which my PR would fix - or at least, improve)

[dosubot[bot]]

Hi, @iurshina,

I'm helping the LangChain team manage their backlog and am marking this issue as stale. From what I understand, the issue you opened requests the ability to dynamically pass a filter through ConversationalRetrievalChain to the underlying vector store. There have been discussions and potential solutions provided by other users, and it seems that has submitted a pull request to address the enhancement, indicating that there is a resolution in progress.

Could you please confirm if this issue is still relevant to the latest version of the LangChain repository? If it is, please let the LangChain team know by commenting on the issue. Otherwise, feel free to close the issue yourself, or the issue will be automatically closed in 7 days.

Thank you for your understanding and contributions to LangChain!

[Natasha-Databricks]

Hello, which PR has fixed the issue? The PR above is still open. I am looking for a solution.

[mkhludnev]

Hello, llm-noob in da house :

retriever = vector_store.as_retriever()
retriever = retriever.configurable_fields(search_kwargs=ConfigurableField(id="search_kwargs",
                                                                                   name="search_kwargs",
                                                                                   description='eg {"filter":{"collection":"foo"}}'))

chain = RunnableLambda(itemgetter("question")) |  retriever  | _combine_documents  ...

return chain.with_config(config={"configurable":{"search_kwargs": {"filter": {"collection": coll_name}}}})
              .invoke(dict(question.qstr))                                                                               

can this be accepted as a solution?

for the ref https://python.langchain.com/docs/expression_language/primitives/configure/

right, that's what's suggested in https://python.langchain.com/docs/use_cases/question_answering/per_user/

[Gauravmahapatrocdm]

Hi

christopherwerner

Can you please let me know how can we implement this for Azure search vector store and Conversational Retrieval Chain, i tried with this but it is throwing errors. I want to filter out based on the source document or filename, how can i do that?Please find below my code.

self.vector_store: VectorStore = AzureSearch(azure_cognitive_search_name=self.vector_store_address,azure_cognitive_search_key=self.vector_store_password, index_name=self.index_name, embedding_function=self.embeddings.embed_query) if vector_store is None else vector_store

def get_semantic_answer_lang_chain(self, question, chat_history): question_generator = LLMChain(llm=self.llm, prompt=CONDENSE_QUESTION_PROMPT, verbose=False) doc_chain = load_qa_with_sources_chain(self.llm, chain_type="stuff", verbose=False, prompt=self.prompt) chain = ConversationalRetrievalChain( retriever=self.vector_store.as_retriever(), question_generator=question_generator, combine_docs_chain=doc_chain, return_source_documents=True,

top_k_docs_for_context= self.k

    )
    result = chain({"question": question, "chat_history": chat_history})
    sources = "\n".join(set(map(lambda x: x.metadata["source"], result['source_documents'])))

    container_sas = self.blob_client.get_container_sas()
    contextDict ={}
    for res in result['source_documents']:
        source_key = self.filter_sourcesLinks(res.metadata['source'].replace('_SAS_TOKEN_PLACEHOLDER_', container_sas)).replace('\n', '').replace(' ', '')
        if source_key not in contextDict:
            contextDict[source_key] = []
        myPageContent = self.clean_encoding(res.page_content)
        contextDict[source_key].append(myPageContent)

    result['answer'] = result['answer'].split('SOURCES:')[0].split('Sources:')[0].split('SOURCE:')[0].split('Source:')[0]
    result['answer'] = self.clean_encoding(result['answer'])
    sources = sources.replace('_SAS_TOKEN_PLACEHOLDER_', container_sas)
    sources = self.filter_sourcesLinks(sources)

    return question, result['answer'], contextDict, sources

Here is the result how it looks.

{'question': 'hi', 'chat_history': [], 'answer': 'I\'m sorry, but I cannot provide any information in response to the greeting "hi." Please ask a specific question about the content provided.', 'source_documents': [Document(page_content=' Vietnam era.............', metadata={'source': 'https://iplanrequirementsstr.blob.core.windows.net/documents/converted/20.pdf.txt', 'chunk': 12, 'key': 'doc:embeddings:fad9b60e69ca07458b190d0483f6675a54ec186e', 'filename': 'converted/20.pdf.txt'}),

I only want results from this converted/20.pdf.txt and not from other documents how to filter results of the retriever based on that?

[ksmin23]

Hi, @iurshina,

I wrote a sample code of passing prefilter dynamically based on two examples in LangChain documentation: (1) https://python.langchain.com/docs/use_cases/question_answering/hybrid/ (2) https://python.langchain.com/docs/use_cases/question_answering/per_user/

This example might not be what you want, but I hope it will be helpful.

In order to pass prefilter when invoking LangChain chains, First, I created configurable_retriever like this:

vectorstore = OpenSearchVectorSearch.from_documents( ... )

retriever = vectorstore.as_retriever()

configurable_retriever = retriever.configurable_fields(
    search_kwargs=ConfigurableField(
        id="search_kwargs",
        name="Search Kwargs",
        description="The search kwargs to use",
    )
)

Then, I made rag_chain from configurable_retriever.

rag_chain = (
    {"context": configurable_retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

Now, we can pass pre_filter when call rag_chain.invoke function.

# passing prefilter dynamically
pre_filter = {
    "match": {
        "text": "LLM"
    }
}

config = {
    "configurable": {
        "search_kwargs": {
            "k": 5,
            "pre_filter": pre_filter,
            "search_type": "script_scoring"
        }
    }
}

rag_chain.invoke("What is Task Decomposition?", config=config)

Here is the full source code.

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import (
  ChatPromptTemplate,
  MessagesPlaceholder
)
from langchain_core.runnables import (
    ConfigurableField,
    RunnableBinding,
    RunnableLambda,
    RunnablePassthrough
)
from langchain_core.messages import (
  HumanMessage,
  AIMessage
)

from langchain_community.embeddings import BedrockEmbeddings
from langchain_community.vectorstores import OpenSearchVectorSearch
from langchain_aws import ChatBedrock as BedrockChat

from langchain.chains import create_history_aware_retriever
from langchain.chains import create_retrieval_chain
from langchain.chains.combine_documents import create_stuff_documents_chain

import boto3

aws_region = boto3.Session().region_name
embeddings = BedrockEmbeddings(
    model_id='amazon.titan-embed-text-v1',
    region_name=aws_region
)

all_splits = [Document(...), ...] # List of Document

opensearch_domain_endpoint = 'https://{domain-name}-xxxxxx.{region}.es.amazonaws.com' # Amazon OpenSearch Endpoint
creds = {'username': ..., 'password': ...}
http_auth = (creds['username'], creds['password'])

opensearch_index_name = 'rag_embeddings'

vectorstore = OpenSearchVectorSearch.from_documents(
    documents=all_splits,
    embedding=embeddings,
    index_name=opensearch_index_name,
    opensearch_url=opensearch_domain_endpoint,
    http_auth=http_auth
)

retriever = vectorstore.as_retriever()

template = """Answer the question based only on the following context:
{context}
Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

configurable_retriever = retriever.configurable_fields(
    search_kwargs=ConfigurableField(
        id="search_kwargs",
        name="Search Kwargs",
        description="The search kwargs to use",
    )
)

BEDROCK_MODEL_ID = "anthropic.claude-3-sonnet-20240229-v1:0"

llm = BedrockChat(
    model_id=BEDROCK_MODEL_ID,
    region_name=aws_region,
    model_kwargs={
        "max_tokens": 512,
        "temperature": 0,
        "top_p": 0.9
    }
)

rag_chain = (
    {"context": configurable_retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# passing prefilter dynamically
pre_filter = {
    "match": {
        "text": "LLM"
    }
}

config = {
    "configurable": {
        "search_kwargs": {
            "k": 5,
            "pre_filter": pre_filter,
            "search_type": "script_scoring"
        }
    }
}

rag_chain.invoke("What is Task Decomposition?", config=config)

#----- RAG with Chat History -----

template = """Answer the question based only on the following context:
{context}
Question: {question}
"""

prompt = ChatPromptTemplate.from_template(template)

configurable_retriever = retriever.configurable_fields(
    search_kwargs=ConfigurableField(
        id="search_kwargs",
        name="Search Kwargs",
        description="The search kwargs to use",
    )
)

contextualize_q_system_prompt = """Given a chat history and the latest user question \
which might reference context in the chat history, formulate a standalone question \
which can be understood without the chat history. Do NOT answer the question, \
just reformulate it if needed and otherwise return it as is."""

contextualize_q_system_prompt = ChatPromptTemplate.from_messages([
    ("system", contextualize_q_system_prompt),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

history_aware_configurable_retriever = create_history_aware_retriever(
    llm, configurable_retriever, contextualize_q_system_prompt
)

qa_system_prompt = """You are an assistant for question-answering tasks. \
Use the following pieces of retrieved context to answer the question. \
If you don't know the answer, just say that you don't know. \
Use three sentences maximum and keep the answer concise.\

{context}"""

qa_prompt = ChatPromptTemplate.from_messages([
    ("system", qa_system_prompt),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}")
])

question_answer_chain = create_stuff_documents_chain(llm, qa_prompt)

configurable_rag_chain = create_retrieval_chain(history_aware_configurable_retriever, question_answer_chain)

chat_history = []

pre_filter = {
    "match": {
        "text": "LLM"
    }
}

config = {
    "configurable": {
        "search_kwargs": {
            "k": 5,
            "pre_filter": pre_filter,
            "search_type": "script_scoring",
        }
    }
}

question = "What is Task Decomposition?"
ai_msg_1 = configurable_rag_chain.invoke({"input": question, "chat_history": chat_history})
chat_history.extend([HumanMessage(content=question), AIMessage(content=ai_msg_1['answer'])])

second_question = "What are common ways of using it?"
ai_msg_2 = configurable_rag_chain.invoke({"input": second_question, "chat_history": chat_history}, config=config)

print(ai_msg_1['answer'])
print('\n\n------------')
print(ai_msg_2['answer'])

# check if pre_filter is applied for retrieving
for document in ai_msg_2['context']:
    print('LLM' in document.page_content)

The versions of LangChain packages to test the code, here is

!pip list | grep -E -w "boto3|langchain|langchainhub|beautifulsoup4"
--------------------------------------------------------------------
beautifulsoup4                       4.12.3
boto3                                1.34.84
langchain                            0.1.16
langchain-aws                        0.1.0
langchain-community                  0.0.34
langchain-core                       0.1.46
langchain-text-splitters             0.0.1
langchainhub                         0.1.15

[Rennoveuser]

Hi, everyone! Hope you’re all doing well!

I’m encountering an issue with LangChain and could use some assistance. I’m trying to create a filter using embeddings metadata and then retrieve the data.

Vector store: FAISS Error log:

•   Instance of VectorStore expected (type=type_error.arbitrary_type; expected_arbitrary_type=VectorStore)
•   ERROR:root:Error processing request: 1 validation error for VectorStoreRetrieverFilter

I’m not sure why, but my vector store isn’t being recognized as a valid instance in VectorStoreRetrieverFilter. Any help would be greatly appreciated!

Feel free to tweak it further if needed!

retriever = VectorStoreRetrieverFilter( vectorstore=vector_store, search_type=get_config('retrieval_search_type'), search_kwargs=search_filters )

@ksmin23 tks for sharing your code and Lang Chain docs.