Adding language detection to Query in order to filter Retriever results by language

[tjakk399]

SOLVED (probably, not thoroughly tested, but works for now for me)

Goal: Extend the query with language detection in order to filter the retriever results by document metadata (see https://docs.haystack.deepset.ai/docs/metadata_filtering).

Solution: Add a custom node in the pipeline (see LanguageDetectorNode in the discussion above), as documented here: https://docs.haystack.deepset.ai/docs/custom_nodes. Also modify RetrieverBase.

This solution hacks on the original source code. It is probably more elegant to subclass your preferred retriever class.

See also discussion in #3475

Pipeline definition:

pipelines:
  - name: query    # a sample extractive-qa Pipeline
    nodes:
      - name: LanguageDetector
        inputs: [Query]
      - name: Retriever
        inputs: [LanguageDetector]
      - name: Reader
        inputs: [Retriever]
...

Custom node:

import langdetect
from typing import Tuple, Dict, Any, Optional, List, Union

from haystack.nodes.base import BaseComponent, Document, MultiLabel
from haystack.errors import HaystackError, PipelineError


def _language(string: str) -> str:
    try:
        return langdetect.detect(string)
    except Exception as e:
        return "UNK"


class LanguageDetector(BaseComponent):
    outgoing_edges: int = 1

    def run(
            self,
            root_node: str,
            query: Optional[str] = None,
            params: Optional[Dict] = None,
            documents: Optional[List[Document]] = None,
    ) -> Tuple[Dict, str]:
        if root_node == "Query":
            if not query:
                raise HaystackError(
                    "Must provide a 'query' parameter for nodes in pipelines where Query is the root node."
                )

            return {
                       'root_node': root_node,
                       'query': query,
                       '_meta': {
                           'filters': {
                               '_language': _language(query)
                           }
                       }
                   }, 'output_1'

        elif root_node == "File":
            for document in documents:
                document.meta["_language"] = _language(document.content)

            return {
                       'documents': documents
                   }, 'output_1'
        else:
            raise PipelineError(f"Invalid root_node '{root_node}'.")

    def run_batch(
            self,
            queries: Optional[Union[str, List[str]]] = None,
            file_paths: Optional[List[str]] = None,
            labels: Optional[Union[MultiLabel, List[MultiLabel]]] = None,
            documents: Optional[Union[List[Document], List[List[Document]]]] = None,
            meta: Optional[Union[Dict[str, Any], List[Dict[str, Any]]]] = None,
            params: Optional[dict] = None,
            debug: Optional[bool] = None,
    ):
        raise NotImplementedError()  # Or you can implement it, if you want to use Pipeline.run_batch()

Note the '_meta' field in the return dictionary from run(). (The field name is arbitrary.) This dictionary must be referenced by the run() function in haystack.nodes.retriever.base.BaseRetriever as follows:

    def run(  # type: ignore
        self,
        root_node: str,
        query: Optional[str] = None,
        filters: Optional[dict] = None,
        top_k: Optional[int] = None,
        documents: Optional[List[Document]] = None,
        index: Optional[str] = None,
        headers: Optional[Dict[str, str]] = None,
        scale_score: bool = None,
        _meta: Optional[dict] = None,  # NEW
    ):
        if root_node == "Query":
            if not query:
                raise HaystackError(
                    "Must provide a 'query' parameter for retrievers in pipelines where Query is the root node."
                )

            # NEW
            # Join filters from original query and from prior node.
            # The former overwrites the latter.
            if _meta and 'filters' in _meta:
                filters = dict(
                    **_meta['filters'],
                    **filters
                )
            logger.info("Meta: " + str(_meta))
            logger.info("Filters: " + str(filters))
            logger.info("Query: " + query)

            self.query_count += 1
            run_query_timed = self.timing(self.run_query, "query_time")
            output, stream = run_query_timed(
                query=query, filters=filters, top_k=top_k, index=index, headers=headers, scale_score=scale_score
            )
        elif root_node == "File":
            self.index_count += len(documents) if documents else 0
            run_indexing = self.timing(self.run_indexing, "index_time")
            output, stream = run_indexing(documents=documents)
        else:
            raise PipelineError(f"Invalid root_node '{root_node}'.")
        return output, stream

Both filters from query and the prior node are joined. The query filters overwrite the prior node filters.

This solution can be applied to e.g. entity extraction as well.

[ZanSara]

Hey @kaumanns sorry for the late reply. I'm not too sure what was your final goal but I think it was to filter the results by language?

Did you have a look at https://docs.haystack.deepset.ai/docs/metadata_filtering? It might be way easier than this and require no hacking of the source. If that's not sufficient, let's rather have an elegant solution added to Haystack! PRs are always welcome 😊

[tjakk399]

@ZanSara In order to provide a PR, I would have to understand how to properly merge filters, i.e. the filter from the search request and the filter generated from a node within the pipeline. Since filters obey a particular grammar (as shown in https://docs.haystack.deepset.ai/docs/metadata_filtering), this may be quite complicated.

If you know how to do this or if you could give me a hint, I can whip up a PR.

[ZanSara]

You should be able to access the content of the params parameter through the signature of run():

haystack/haystack/pipelines/base.py

Line 453 in 9f4a9a7

def run( # type: ignore

params will be a dictionary that contains a filters key if filters are given to the pipeline. Nodes can do anything with that value, so you should be able to merge the filters there. Do you think that would cover your case?

[ZanSara]

About the merging of the filters themselves, would that work to simply AND or OR the existing filters with the ones introduced by the node?

[tjakk399]

You should be able to access the content of the params parameter through the signature of run():

haystack/haystack/pipelines/base.py

Line 453 in 9f4a9a7

def run( # type: ignore

params will be a dictionary that contains a filters key if filters are given to the pipeline. Nodes can do anything with that value, so you should be able to merge the filters there. Do you think that would cover your case?

You mean by returning from the node's run() as such?

return {
   'root_node': root_node,
   'query': query,
   'params': { 
       'filters': _merge(newfilters, params['filters']) 
   },
}, 'output_1'

I tried that hoping that the Retriever would pick up, but it seems the returned params key is ignored (or overwritten by the original one).

Which is weird because some other key (e.g. _meta in my hack) is respected and passed through to the downstream Retriever node. So I figured there is some special logic dealing with params and I shouldn't touch it. There be dragons.

That's why I modified the Retriever to perform the filters merge. But yes, the logic doesn't belong there. The custom node should control such shenanigans with the request.

[ZanSara]

Wow, really odd. I should really check this one out, I believe param should be treated normally... Thank you for the attempt!