Using metadata filter with Pinecone retriever

[Abe410]

Hi all

I have created a chatbot which uses multiple tools to get an answer for a question. One of the tools queries a Pinecone index to get an answer. The structure of the chain is as follows:

def text_retrieval_chain():
    # main retrieval chain class
    class RetrievalChain:
        def __init__(self, llm, retriever):
            self.chain = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=retriever)

        def run(self, prompt):
            response = self.chain(prompt)
            return response['result']

    # vectorstore
    index_name = "testing"
    vectorstore = Pinecone.from_existing_index(
        index_name=index_name, namespace="test", embedding=embeddings
    )
    retriever = whitepaper_vectorstore.as_retriever(search_kwargs={"k": 4})
    return RetrievalChain(llm=tool_llm, retriever=retriever)

Now, in testing and development, I know I can introduce a metadata filter as follows:

retriever = vectorstore.as_retriever(search_kwargs={"k": 1, 
                                                    'filter': {'unixtimestamp':{'$gt': 1677296101}}
                                                   }
                                    )

But how do I do it so that the filter value is always a variable, and is determined by the agent within the chain?

Following is the general structure of the code:

from langchain import SQLDatabase, SQLDatabaseChain
from langchain.prompts.prompt import PromptTemplate
from langchain.chat_models import ChatOpenAI
from langchain.memory import ConversationBufferWindowMemory
from langchain.chains import ConversationChain, LLMChain
from langchain.embeddings import OpenAIEmbeddings
from langchain.chains import RetrievalQAWithSourcesChain, RetrievalQA
from langchain.vectorstores import Pinecone
from langchain.agents import Agent, Tool, AgentType, AgentOutputParser, AgentExecutor, initialize_agent
from langchain.agents.conversational.output_parser import ConvoOutputParser
from langchain.agents.conversational.prompt import SUFFIX
from pydantic import Field
from langchain.tools.base import BaseTool
from langchain.base_language import BaseLanguageModel
from langchain.callbacks.base import BaseCallbackManager

def text_retrieval_chain():
    # main retrieval chain class
    class RetrievalChain:
        def __init__(self, llm, retriever):
            self.chain = RetrievalQA.from_chain_type(llm=llm, chain_type="stuff", retriever=retriever)

        def run(self, prompt):
            response = self.chain(prompt)
            return response['result']

    # vectorstore
    index_name = "testing"
    vectorstore = Pinecone.from_existing_index(
        index_name=index_name, namespace="articlefeeds", embedding=embeddings
    )
    retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
    return RetrievalChain(llm=tool_llm, retriever=retriever)


# retrieval chain for article/RSS feed
text_retrieval_chain = text_retrieval_chain()
# Create the tools list
tools = [
    Tool(
        name="Text Retrieval",
        func=text_retrieval_chain.run,
        description="Useful for when you need to find textual knowledge .",
    ),
]

# create agent
agent = ConversationalAgent.from_llm_and_tools(llm=llm, tools=tools)

agent_memory = ConversationBufferWindowMemory(k=1, memory_key="chat_history", return_messages=True)
agent_executor = AgentExecutor.from_agent_and_tools(agent=agent, tools=tools, memory=agent_memory, verbose=verbose)

answer = agent_executor.run(question)

For example, if I type "What is the latest news of the market?", the agent should apply a filter in the retriever as below, but how?

retriever = vectorstore.as_retriever(search_kwargs={"k": 1, 
                                                    'filter': {'unixtimestamp':{'$gt': 1677296101}}
                                                   }
                                    )

[dosubot[bot]]

🤖

Hi,

Based on your requirement, you want to dynamically apply a filter value that is determined by the agent within the chain. To achieve this, you can modify the RetrievalChain class and the PineconeTranslator class in your code.

Firstly, you need to add a method in the PineconeTranslator class that accepts the filter value as an argument and applies it to the structured query. Here's how you can do it:

class PineconeTranslator(Visitor):
    # ... existing code ...

    def apply_filter(self, filter_value):
        self.filter_value = filter_value

    def visit_structured_query(self, structured_query: StructuredQuery) -> Tuple[str, dict]:
        if structured_query.filter is None:
            kwargs = {}
        else:
            # Use the filter value set by the agent if it exists
            if hasattr(self, 'filter_value'):
                kwargs = {"filter": self.filter_value}
            else:
                kwargs = {"filter": structured_query.filter.accept(self)}
        return structured_query.query, kwargs

In this modification, the apply_filter method is added to the PineconeTranslator class. This method accepts a filter value and sets it as an attribute of the class instance. Then, in the visit_structured_query method, we check if the filter_value attribute exists. If it does, we use it as the filter for the structured query. If it doesn't, we use the existing filter from the structured query. This allows the agent to dynamically apply a filter value within the chain.

Next, you need to modify the RetrievalChain class to accept a dynamic filter value for the Pinecone retriever. Here is how you can do it:

class PineconeHybridSearchRetriever(BaseRetriever):
    """`Pinecone Hybrid Search` retriever."""

    # ... existing attributes ...

    filter_value: Optional[StructuredQuery] = None
    """Filter value for the Pinecone retriever."""

    # ... existing methods ...

    def _get_relevant_documents(
        self, query: str, *, run_manager: CallbackManagerForRetrieverRun
    ) -> List[Document]:
        # ... existing code ...

        # translate the filter value to a valid Pinecone filter
        translator = PineconeTranslator()
        query, kwargs = translator.visit_structured_query(self.filter_value)

        # query pinecone with the query parameters
        result = self.index.query(
            vector=dense_vec,
            sparse_vector=sparse_vec,
            top_k=self.top_k,
            include_metadata=True,
            **kwargs,  # use the translated filter value here
        )

        # ... existing code ...

In this modification, a new attribute filter_value is added to the PineconeHybridSearchRetriever class. This attribute is then translated to a valid Pinecone filter using the PineconeTranslator class and used as the filter argument in the index.query method. If filter_value is None, no filter will be applied. If it is a StructuredQuery, it will be translated to a valid Pinecone filter and used for the Pinecone retriever.

I hope this helps! Let me know if you have any other questions.

Sources

• can't seem to add filters in ConversationalRetrievalChain
• libs/langchain/langchain/retrievers/self_query/pinecone.py
• libs/langchain/langchain/retrievers/pinecone_hybrid_search.py

---

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.