Repository: phidatahq/phidata
File: libs/agno/agno/knowledge/knowledge.py (Python)

Code before:

        # If no metadata filters tracked yet, all keys are considered invalid
        if valid_metadata_filters is None or not valid_metadata_filters:
            invalid_keys = list(filters.keys())
            log_warning(f"No valid metadata filters tracked yet. All filter keys considered invalid: {invalid_keys}")
            return {}, invalid_keys

        for key, value in filters.items():
            # Handle both normal keys and prefixed keys like meta_data.key
            base_key = key.split(".")[-1] if "." in key else key
            if base_key in valid_metadata_filters or key in valid_metadata_filters:
                valid_filters[key] = value
            else:
                invalid_keys.append(key)
                log_warning(f"Invalid filter key: {key} - not present in knowledge base")

        return valid_filters, invalid_keys

    def validate_filters(self, filters: Optional[Dict[str, Any]]) -> Tuple[Dict[str, Any], List[str]]:
        valid_filters_from_db = self.get_valid_filters()

        valid_filters, invalid_keys = self._validate_filters(filters, valid_filters_from_db)

        return valid_filters, invalid_keys

    async def async_validate_filters(self, filters: Optional[Dict[str, Any]]) -> Tuple[Dict[str, Any], List[str]]:
        valid_filters_from_db = await self.async_get_valid_filters()

        valid_filters, invalid_keys = self._validate_filters(filters, valid_filters_from_db)

        return valid_filters, invalid_keys

    def remove_vector_by_id(self, id: str) -> bool:
        from agno.vectordb import VectorDb

        self.vector_db = cast(VectorDb, self.vector_db)
        if self.vector_db is None:
            log_warning("No vector DB provided")
            return False
        return self.vector_db.delete_by_id(id)

    def remove_vectors_by_name(self, name: str) -> bool:
        from agno.vectordb import VectorDb

        self.vector_db = cast(VectorDb, self.vector_db)
        if self.vector_db is None:
            log_warning("No vector DB provided")
            return False
        return self.vector_db.delete_by_name(name)

    def remove_vectors_by_metadata(self, metadata: Dict[str, Any]) -> bool:

Review comment (suggestion):
```suggestion
    async def async_validate_filters(self, filters: Optional[Dict[str, Any]]) -> Tuple[Dict[str, Any], List[str]]:
        """Return a tuple containing a dict with all valid filters and a list of invalid filter keys"""
```

docstring as it's public

Code after:
                valid_filters.update(content.metadata.keys())

        return valid_filters

    def _validate_filters(
        self, filters: Union[Dict[str, Any], List[FilterExpr]], valid_metadata_filters: Set[str]
    ) -> Tuple[Union[Dict[str, Any], List[FilterExpr]], List[str]]:
        if not filters:
            return {}, []

        valid_filters: Union[Dict[str, Any], List[FilterExpr]] = {}
        invalid_keys = []

        if isinstance(filters, dict):
            # If no metadata filters tracked yet, all keys are considered invalid
            if valid_metadata_filters is None or not valid_metadata_filters:
                invalid_keys = list(filters.keys())
                log_warning(
                    f"No valid metadata filters tracked yet. All filter keys considered invalid: {invalid_keys}"
                )
                return {}, invalid_keys

            for key, value in filters.items():
                # Handle both normal keys and prefixed keys like meta_data.key
                base_key = key.split(".")[-1] if "." in key else key
                if base_key in valid_metadata_filters or key in valid_metadata_filters:
                    valid_filters[key] = value  # type: ignore
                else:
                    invalid_keys.append(key)
                    log_warning(f"Invalid filter key: {key} - not present in knowledge base")

        elif isinstance(filters, List):
            # Validate that list contains FilterExpr instances
            for i, filter_item in enumerate(filters):
                if not isinstance(filter_item, FilterExpr):
                    log_warning(
                        f"Invalid filter at index {i}: expected FilterExpr instance, "
                        f"got {type(filter_item).__name__}. "
                        f"Use filter expressions like EQ('key', 'value'), IN('key', [values]), "
                        f"AND(...), OR(...), NOT(...) from agno.filters"
                    )
            # Filter expressions are already validated, return empty dict/list
            # The actual filtering happens in the vector_db layer
            return filters, []

        return valid_filters, invalid_keys

    def validate_filters(
        self, filters: Union[Dict[str, Any], List[FilterExpr]]
    ) -> Tuple[Union[Dict[str, Any], List[FilterExpr]], List[str]]:
        valid_filters_from_db = self.get_valid_filters()