Implement filter by metadata post retrieval

[rahulkrprajapati]

what I meant is implement a metadata filter to filter based on this:

MetadataFilters(
                        filters=[
                            ExactMatchFilter(key="namespace", value=self.namespace)
                        ]
                    ),

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = self.fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

[rahulkrprajapati]

@dosu

what I meant is implement a metadata filter to filter based on this:

MetadataFilters(
                        filters=[
                            ExactMatchFilter(key="namespace", value=self.namespace)
                        ]
                    ),

    def process_streamlit_fetch_query_results(self, query="", verbose=False):
        try:
            query_bundle = QueryBundle(query_str=query)
            retrived_nodes = self.fusion_retriever.retrieve(query_bundle)
            recency_nodes = self.recency_postprocessor.postprocess_nodes(
                retrived_nodes, query_bundle=query_bundle
            )
            rerank_nodes = self.postprocessor.postprocess_nodes(
                nodes=recency_nodes, query_bundle=query_bundle
            )
            px.active_session().url
            return rerank_nodes
        except Exception as e:
            traceback.print_exc()
            raise Exception(f"An error occurred retrieving: {e}")

[dosubot[bot]]

To implement a metadata filter in the process_streamlit_fetch_query_results function, you can modify the function to include the MetadataFilters with an ExactMatchFilter for the "namespace" key. Here is how you can do it:

from llama_index.core.vector_stores import ExactMatchFilter, MetadataFilters

def process_streamlit_fetch_query_results(self, query="", verbose=False):
    try:
        query_bundle = QueryBundle(query_str=query)
        
        # Apply metadata filter
        metadata_filter = MetadataFilters(
            filters=[ExactMatchFilter(key="namespace", value=self.namespace)]
        )
        
        # Retrieve nodes with the applied filter
        retrived_nodes = self.fusion_retriever.retrieve(query_bundle, filters=metadata_filter)
        
        recency_nodes = self.recency_postprocessor.postprocess_nodes(
            retrived_nodes, query_bundle=query_bundle
        )
        rerank_nodes = self.postprocessor.postprocess_nodes(
            nodes=recency_nodes, query_bundle=query_bundle
        )
        px.active_session().url
        return rerank_nodes
    except Exception as e:
        traceback.print_exc()
        raise Exception(f"An error occurred retrieving: {e}")

This code snippet sets up a metadata filter that matches the "namespace" key exactly to the provided self.namespace value. The retrieve method of fusion_retriever is then called with this filter to ensure that only documents with the specified "namespace" metadata are considered in the query results ^[1].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu

raise TypeError(

TypeError: got an unexpected keyword argument 'filters'

[dosubot[bot]]

The TypeError indicates that the retrieve method of the fusion_retriever does not accept a filters argument. To resolve this, you need to apply the metadata filter in a different way. Here is the corrected function:

from llama_index.core.vector_stores import ExactMatchFilter, MetadataFilters

def process_streamlit_fetch_query_results(self, query="", verbose=False):
    try:
        query_bundle = QueryBundle(query_str=query)
        
        # Retrieve nodes without filters
        retrived_nodes = self.fusion_retriever.retrieve(query_bundle)
        
        # Apply metadata filter after retrieval
        metadata_filter = MetadataFilters(
            filters=[ExactMatchFilter(key="namespace", value=self.namespace)]
        )
        filtered_nodes = [node for node in retrived_nodes if metadata_filter.apply(node.metadata)]
        
        recency_nodes = self.recency_postprocessor.postprocess_nodes(
            filtered_nodes, query_bundle=query_bundle
        )
        rerank_nodes = self.postprocessor.postprocess_nodes(
            nodes=recency_nodes, query_bundle=query_bundle
        )
        px.active_session().url
        return rerank_nodes
    except Exception as e:
        traceback.print_exc()
        raise Exception(f"An error occurred retrieving: {e}")

This code retrieves the nodes first and then applies the metadata filter to the retrieved nodes. This way, you avoid passing the filters argument directly to the retrieve method, which does not accept it ^[1].

To continue talking to Dosu, mention @dosu.

[rahulkrprajapati]

@dosu

raise Exception(f"An error occurred retrieving: {e}")

Exception: An error occurred retrieving: 'MetadataFilters' object has no attribute 'apply

[dosubot[bot]]

The MetadataFilters object does not have an apply method. Instead, you should use the _create_odata_filter method to generate a filter string and apply it to your query. Here is the corrected function:

from llama_index.core.vector_stores import ExactMatchFilter, MetadataFilters

def process_streamlit_fetch_query_results(self, query="", verbose=False):
    try:
        query_bundle = QueryBundle(query_str=query)
        
        # Retrieve nodes without filters
        retrived_nodes = self.fusion_retriever.retrieve(query_bundle)
        
        # Apply metadata filter after retrieval
        metadata_filter = MetadataFilters(
            filters=[ExactMatchFilter(key="namespace", value=self.namespace)]
        )
        
        # Create OData filter string
        filter_fn = self._create_odata_filter(metadata_filter)
        
        # Filter nodes based on the OData filter string
        filtered_nodes = [node for node in retrived_nodes if filter_fn(node.metadata)]
        
        recency_nodes = self.recency_postprocessor.postprocess_nodes(
            filtered_nodes, query_bundle=query_bundle
        )
        rerank_nodes = self.postprocessor.postprocess_nodes(
            nodes=recency_nodes, query_bundle=query_bundle
        )
        px.active_session().url
        return rerank_nodes
    except Exception as e:
        traceback.print_exc()
        raise Exception(f"An error occurred retrieving: {e}")

Ensure you have the _create_odata_filter method implemented in your class:

def _create_odata_filter(self, metadata_filters: MetadataFilters) -> callable:
    """Generate a filter function using supplied metadata filters."""
    def filter_fn(metadata):
        for f in metadata_filters.legacy_filters():
            if f.key not in metadata or metadata[f.key] != f.value:
                return False
        return True
    return filter_fn

This code retrieves the nodes first and then applies the metadata filter using a filter function generated by _create_odata_filter ^[1].

To continue talking to Dosu, mention @dosu.