BaseRetriever kwargs

[Pieter-JanCnuddeE61]

class UserScoringRetrieverMSLLCEL(BaseRetriever):

    vectorstore: VectorStore
    username: str
    metadata_filters: str

    def clean_metadata(self, doc):
        """
        Cleans the metadata of a document.

        Parameters:
            doc (object): The document object.

        Returns:
            dict: A dictionary containing the cleaned metadata.

        """

        metadata = doc.metadata

        # page_number = re.search(r'pagenumber_(\d+)', metadata["chunk"])
        
        print(metadata["title"], metadata["@search.score"])
        year_doc = int(datetime.strptime(metadata["year"], '%Y-%m-%dT%H:%M:%SZ').year)
        score_divider_recency = (datetime.today().year - year_doc)*2
        if metadata["document_type"] == "PI":
            score_multiplier_doctype = 2
        elif metadata["document_type"] == "Manuscript":
            score_multiplier_doctype = 1.8
        elif metadata["document_type"] == "Poster":
            score_multiplier_doctype = 1.5
        else:
            score_multiplier_doctype = 1

        return {
            "chunk_id": metadata["chunk_id"],
            "file_id": metadata["title"], 
            "source": metadata["title"] + "_page=" + metadata["page_number_start"], 
            "page_number": metadata["page_number_start"],  
            "document_title": metadata["document_title"],
            "year": metadata["year"],
            "journal": metadata["Journal"],
            "lead_author": metadata["Lead_Author"],
            "dataset": metadata["Dataset"],
            "molecule": metadata["Molecule"],
            "indication": metadata["Indication"],
            "document_region": metadata["document_region"],
            "document_type": metadata["document_type"],
            "hyperlink": metadata["hyperlink"],
            "orig_score": metadata["@search.score"],
            "calc_score": (metadata["@search.score"] * score_multiplier_doctype) / (score_divider_recency +20)
            }
    
    
    def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun, metadata_filters:str
        ) -> List[Document]:
        
        docs = []
        load_dotenv()
        admins = os.getenv('ADMINS_MSL', '')
        admins_list = admins.split(',')
        is_admin = self.username.split('@')[0] in admins_list

        if not is_admin:
            print("User is not an admin, no access to documents.")
            return docs
        
        # Convert JSON string to Python dictionary
        metadata_filters = json.loads(metadata_filters)["metadata_filters"]

        # Ensure metadata_filters is a valid string
        if metadata_filters:
            # Convert JSON string to Python dictionary
            try:
                metadata_filters_dict = json.loads(metadata_filters)["metadata_filters"]
            except (json.JSONDecodeError, KeyError) as e:
                print(f"Error loading metadata filters: {e}")
                return docs  # Return empty docs on error

            # Construct the filter query
            filter_conditions = []
            for key, value in metadata_filters_dict.items():
                if value:  # Only process keys with non-empty values
                    if isinstance(value, list):
                        values_str = ','.join(value)
                        filter_conditions.append(f"search.in({key}, '{values_str}', ',')")
                    else:
                        filter_conditions.append(f"search.in({key}, '{value}', ',')")

            filter_query = " and ".join(filter_conditions)
            print(f"following filter_query used: {filter_query}")


        try:
            search_results = self.vectorstore.similarity_search(query, search_type="hybrid", k=NUMBER_OF_CHUNKS_TO_RETURN_MSL, filters=filter_query) 
        except Exception as e:
            print(f"Error during similarity search: {e}")
            return docs

I made the above custom retriever based on the langchain BaseRetriever. This retriever is than used in the following chain:

retriever = UserScoringRetrieverMSLLCEL(vectorstore=vector_store, username=username, metadata_filters=metadata_filters)

history_aware_retriever = create_history_aware_retriever(llm, retriever, contextualize_q_prompt)

The issue is that I cannot pass the metadata_filters to the retriever. How can I pass this metadata_filters as an argument to make sure I can filter in my similarity search?

[dosubot[bot]]

To pass metadata_filters as an argument to your custom retriever and ensure filtering in the similarity search, you need to make sure that the metadata_filters parameter is correctly handled and passed to the similarity_search method.

Here's a refined approach:

1. Ensure metadata_filters is a dictionary: Convert the metadata_filters string to a dictionary before using it.

2. Pass the filter correctly: Use the filter parameter in the similarity_search method.

Here's how you can modify your code:

def _get_relevant_documents(self, query: str, *, run_manager: CallbackManagerForRetrieverRun, metadata_filters: str) -> List[Document]:
    docs = []
    load_dotenv()
    admins = os.getenv('ADMINS_MSL', '')
    admins_list = admins.split(',')
    is_admin = self.username.split('@')[0] in admins_list

    if not is_admin:
        print("User is not an admin, no access to documents.")
        return docs

    # Convert JSON string to Python dictionary
    try:
        metadata_filters_dict = json.loads(metadata_filters)["metadata_filters"]
    except (json.JSONDecodeError, KeyError) as e:
        print(f"Error loading metadata filters: {e}")
        return docs  # Return empty docs on error

    # Construct the filter query
    filter_conditions = {}
    for key, value in metadata_filters_dict.items():
        if value:  # Only process keys with non-empty values
            filter_conditions[key] = value

    try:
        search_results = self.vectorstore.similarity_search(
            query,
            search_type="hybrid",
            k=NUMBER_OF_CHUNKS_TO_RETURN_MSL,
            filter=filter_conditions
        )
    except Exception as e:
        print(f"Error during similarity search: {e}")
        return docs

    return search_results

Key Points:

• Filter Construction: Ensure filter_conditions is a dictionary that matches the expected format for the similarity_search method.
• Error Handling: Handle JSON decoding errors gracefully to avoid runtime issues.

This approach should allow you to pass metadata_filters effectively and perform the desired filtering in your similarity search.

To continue talking to Dosu, mention @dosu.

[Pieter-JanCnuddeE61]

@dosu This is not exactly the issue I'm facing. The issue is that the langchain BaseRetriever does not seem to accept the metadata_filters as an argument. The metadata_filters never arrive in the retriever class and cannot be used. The error I'm getting is as follows:

UserScoringRetrieverMSLLCEL._get_relevant_documents() missing 1 required keyword-only argument: 'metadata_filters'

[dosubot[bot]]

To resolve the issue with metadata_filters not being passed correctly, you need to ensure that the metadata_filters argument is included when calling the _get_relevant_documents method. Here's how you can address this:

1. Modify the Call to _get_relevant_documents: Ensure that metadata_filters is passed as a keyword argument when invoking _get_relevant_documents.

2. Update the Retriever Initialization: Make sure that metadata_filters is included in the initialization and usage of your retriever.

Here's an example of how you might adjust your code:

# When initializing the retriever
retriever = UserScoringRetrieverMSLLCEL(
    vectorstore=vector_store,
    username=username,
    metadata_filters=metadata_filters
)

# Ensure that when you call the method, you pass metadata_filters
def some_function():
    # Assuming run_manager is defined
    documents = retriever._get_relevant_documents(
        query="your query here",
        run_manager=run_manager,
        metadata_filters=metadata_filters
    )

Key Points:

• Keyword Argument: Ensure metadata_filters is passed as a keyword argument in the method call.
• Consistent Usage: Verify that metadata_filters is consistently used throughout your code where needed.

This should help ensure that metadata_filters is correctly passed and used in your retriever.

To continue talking to Dosu, mention @dosu.

[Pieter-JanCnuddeE61]

@dosu How do I implement this in my chain I have in place? The chain looks like this:

history_aware_retriever = create_history_aware_retriever(llm, retriever, contextualize_q_prompt)

[dosubot[bot]]

To implement the passing of metadata_filters in your create_history_aware_retriever chain, you can modify the chain setup to ensure metadata_filters are included. Here's how you can do it:

1. Modify the Chain Setup: Ensure that metadata_filters are passed to the retriever within the chain.

2. Use a Wrapper or Adapter: If the create_history_aware_retriever function doesn't directly support passing additional arguments, you might need to wrap your retriever to include metadata_filters.

Here's an example approach:

# Define a wrapper function to include metadata_filters
def retriever_with_filters(query, run_manager, metadata_filters):
    return retriever._get_relevant_documents(
        query=query,
        run_manager=run_manager,
        metadata_filters=metadata_filters
    )

# Create the history aware retriever chain
history_aware_retriever = create_history_aware_retriever(
    llm,
    retriever_with_filters,
    contextualize_q_prompt
)

Key Points:

• Wrapper Function: Use a wrapper to ensure metadata_filters are passed correctly.
• Chain Integration: Integrate the wrapper into your chain setup.

This approach should help you pass metadata_filters effectively within your chain.

To continue talking to Dosu, mention @dosu.