diff --git a/ovos_plugin_manager/templates/solvers.py b/ovos_plugin_manager/templates/solvers.py index b2bd725f..8ca7195d 100644 --- a/ovos_plugin_manager/templates/solvers.py +++ b/ovos_plugin_manager/templates/solvers.py @@ -49,7 +49,7 @@ def func_wrapper(*args, **kwargs): return func_decorator -def auto_translate_inputs(translate_keys: List[str]): +def auto_translate(translate_keys: List[str]): """ Decorator to ensure all kwargs in 'translate_keys' are translated to self.default_lang. data returned by the decorated function will be translated back to original language NOTE: not meant to be used outside solver plugins""" @@ -118,6 +118,19 @@ def func_wrapper(*args, **kwargs): return func_decorator +def _call_with_sanitized_kwargs(func, *args, lang: Optional[str] = None): + # Inspect the function signature to ensure it has both 'lang' and 'context' parameters + params = inspect.signature(func).parameters + kwargs = {} + if "lang" in params: + # new style - only lang is passed + kwargs["lang"] = lang + elif "context" in kwargs: + # old style - when plugins received context only + kwargs["context"]["lang"] = lang + return func(*args, **kwargs) + + class AbstractSolver: """Base class for solvers that perform various NLP tasks.""" @@ -262,94 +275,77 @@ def __init__(self, config: Optional[Dict] = None, # plugin methods to override @abc.abstractmethod - def get_spoken_answer(self, query: str, - context: Optional[Dict] = None, - lang: Optional[str] = None) -> str: + def get_spoken_answer(self, query: str, lang: Optional[str] = None) -> str: """ Obtain the spoken answer for a given query. :param query: The query text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: The spoken answer as a text response. """ raise NotImplementedError @_deprecate_context2lang() - def stream_utterances(self, query: str, - context: Optional[Dict] = None, - lang: Optional[str] = None) -> Iterable[str]: + def stream_utterances(self, query: str, lang: Optional[str] = None) -> Iterable[str]: """ Stream utterances for the given query as they become available. :param query: The query text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: An iterable of utterances. """ - ans = _call_with_sanitized_kwargs(self.get_spoken_answer, query, - lang=lang, context=context) + ans = _call_with_sanitized_kwargs(self.get_spoken_answer, query, lang=lang) for utt in self.sentence_split(ans): yield utt @_deprecate_context2lang() - def get_data(self, query: str, context: Optional[Dict] = None, - lang: Optional[str] = None) -> Optional[dict]: + def get_data(self, query: str, lang: Optional[str] = None) -> Optional[dict]: """ Retrieve data for the given query. :param query: The query text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: A dictionary containing the answer. """ - return {"answer": _call_with_sanitized_kwargs(self.get_spoken_answer, query, - lang=lang, context=context)(query, **kwargs)} + return {"answer": _call_with_sanitized_kwargs(self.get_spoken_answer, query, lang=lang)} @_deprecate_context2lang() - def get_image(self, query: str, context: Optional[Dict] = None, - lang: Optional[str] = None) -> Optional[str]: + def get_image(self, query: str, lang: Optional[str] = None) -> Optional[str]: """ Get the path or URL to an image associated with the query. :param query: The query text - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: The path or URL to a single image. """ return None @_deprecate_context2lang() - def get_expanded_answer(self, query: str, context: Optional[Dict] = None, - lang: Optional[str] = None) -> List[dict]: + def get_expanded_answer(self, query: str, lang: Optional[str] = None) -> List[dict]: """ Get an expanded list of steps to elaborate on the answer. :param query: The query text - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: A list of dictionaries with each step containing a title, summary, and optional image. """ return [{"title": query, - "summary": _call_with_sanitized_kwargs(self.get_spoken_answer, query, - lang=lang, context=context), - "img": _call_with_sanitized_kwargs(self.get_image, query, - lang=lang, context=context)}] + "summary": _call_with_sanitized_kwargs(self.get_spoken_answer, query, lang=lang), + "img": _call_with_sanitized_kwargs(self.get_image, query, lang=lang)}] # user facing methods @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["query"]) - def search(self, query: str, context: Optional[Dict] = None, lang: Optional[str] = None) -> dict: + @auto_translate(translate_keys=["query"]) + def search(self, query: str, lang: Optional[str] = None) -> dict: """ Perform a search with automatic translation and caching. - NOTE: "query" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "query" automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param query: The query text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: The data dictionary retrieved from the cache or computed anew. """ @@ -359,8 +355,7 @@ def search(self, query: str, context: Optional[Dict] = None, lang: Optional[str] else: # search data try: - data = _call_with_sanitized_kwargs(self.get_data, query, - lang=lang, context=context) + data = _call_with_sanitized_kwargs(self.get_data, query, lang=lang) except: return {} @@ -371,37 +366,34 @@ def search(self, query: str, context: Optional[Dict] = None, lang: Optional[str] return data @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["query"]) - def visual_answer(self, query: str, context: Optional[Dict] = None, lang: Optional[str] = None) -> str: + @auto_translate(translate_keys=["query"]) + def visual_answer(self, query: str, lang: Optional[str] = None) -> str: """ Retrieve the image associated with the query with automatic translation and caching. - NOTE: "query" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "query" automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param query: The query text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: The path or URL to the image. """ - return _call_with_sanitized_kwargs(self.get_image, query, - lang=lang, context=context) + return _call_with_sanitized_kwargs(self.get_image, query, lang=lang) @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["query"]) - def spoken_answer(self, query: str, context: Optional[Dict] = None, lang: Optional[str] = None) -> str: + @auto_translate(translate_keys=["query"]) + def spoken_answer(self, query: str, lang: Optional[str] = None) -> str: """ Retrieve the spoken answer for the query with automatic translation and caching. - NOTE: "query" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "query" automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param query: The query text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: The spoken answer as a text response. """ @@ -411,8 +403,7 @@ def spoken_answer(self, query: str, context: Optional[Dict] = None, lang: Option summary = self.spoken_cache[query] else: - summary = _call_with_sanitized_kwargs(self.get_spoken_answer, query, - lang=lang, context=context) + summary = _call_with_sanitized_kwargs(self.get_spoken_answer, query, lang=lang) # save to cache if self.enable_cache: self.spoken_cache[query] = summary @@ -420,31 +411,26 @@ def spoken_answer(self, query: str, context: Optional[Dict] = None, lang: Option return summary @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["query"]) - def long_answer(self, query: str, context: Optional[Dict] = None, - lang: Optional[str] = None) -> List[dict]: + @auto_translate(translate_keys=["query"]) + def long_answer(self, query: str, lang: Optional[str] = None) -> List[dict]: """ Retrieve a detailed list of steps to expand the answer. - NOTE: "query" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "query" automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param query: The query text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: A list of steps to elaborate on the answer, with each step containing a title, summary, and optional image. """ - steps = _call_with_sanitized_kwargs(self.get_expanded_answer, query, - lang=lang, context=context) + steps = _call_with_sanitized_kwargs(self.get_expanded_answer, query, lang=lang) # use spoken_answer as last resort if not steps: - summary = _call_with_sanitized_kwargs(self.get_spoken_answer, query, - lang=lang, context=context) + summary = _call_with_sanitized_kwargs(self.get_spoken_answer, query, lang=lang) if summary: - img = _call_with_sanitized_kwargs(self.get_image, query, - lang=lang, context=context) + img = _call_with_sanitized_kwargs(self.get_image, query, lang=lang) steps = [{"title": query, "summary": step0, "img": img} for step0 in self.sentence_split(summary, -1)] return steps @@ -458,13 +444,11 @@ class TldrSolver(AbstractSolver): @abc.abstractmethod def get_tldr(self, document: str, - context: Optional[Dict] = None, lang: Optional[str] = None) -> str: """ Summarize the provided document. :param document: The text of the document to summarize, assured to be in the default language. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: A summary of the provided document. """ @@ -473,37 +457,22 @@ def get_tldr(self, document: str, # user facing methods @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["document"]) - def tldr(self, document: str, - context: Optional[Dict] = None, - lang: Optional[str] = None) -> str: + @auto_translate(translate_keys=["document"]) + def tldr(self, document: str, lang: Optional[str] = None) -> str: """ Summarize the provided document with automatic translation and caching if needed. - NOTE: "document" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "document" automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param document: The text of the document to summarize. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: A summary of the provided document. """ # summarize - return _call_with_sanitized_kwargs(self.get_tldr, document, - lang=lang, context=context) - - -def _call_with_sanitized_kwargs(func, *args, context: Optional[Dict] = None, lang: Optional[str] = None): - # Inspect the function signature to ensure it has both 'lang' and 'context' parameters - params = inspect.signature(func).parameters - kwargs = {} - if "context" in params: - kwargs["context"] = context - if "lang" in params: - kwargs["lang"] = lang - return func(*args, **kwargs) + return _call_with_sanitized_kwargs(self.get_tldr, document, lang=lang) class EvidenceSolver(AbstractSolver): @@ -514,14 +483,12 @@ class EvidenceSolver(AbstractSolver): @abc.abstractmethod def get_best_passage(self, evidence: str, question: str, - context: Optional[Dict] = None, lang: Optional[str] = None) -> str: """ Extract the best passage from evidence that answers the given question. :param evidence: The text containing the evidence, assured to be in the default language. :param question: The question to answer, assured to be in the default language. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: The passage from the evidence that best answers the question. """ @@ -529,27 +496,24 @@ def get_best_passage(self, evidence: str, question: str, # user facing methods @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["evidence", "question"]) + @auto_translate(translate_keys=["evidence", "question"]) def extract_answer(self, evidence: str, question: str, - context: Optional[Dict] = None, lang: Optional[str] = None) -> str: """ Extract the best passage from evidence that answers the question with automatic translation and caching if needed. - NOTE: "evidence" and "question" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "evidence" and "question" are automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param evidence: The text containing the evidence. :param question: The question to answer. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: The passage from the evidence that answers the question. """ # extract answer from doc - return self.get_best_passage(evidence, question, - context=context, lang=lang) + return self.get_best_passage(evidence, question, lang=lang) class MultipleChoiceSolver(AbstractSolver): @@ -565,69 +529,41 @@ class MultipleChoiceSolver(AbstractSolver): # plugins in the wild missing this method # @abc.abstractmethod def rerank(self, query: str, options: List[str], - context: Optional[Dict] = None, lang: Optional[str] = None) -> List[Tuple[float, str]]: """ Rank the provided options based on the query. :param query: The query text, assured to be in the default language. :param options: A list of answer options, each assured to be in the default language. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: A list of tuples where each tuple contains a score and the corresponding option text, sorted by score. """ raise NotImplementedError @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["query", "options"]) + @auto_translate(translate_keys=["query", "options"]) def select_answer(self, query: str, options: List[str], - context: Optional[Dict] = None, lang: Optional[str] = None, return_index: bool = False) -> Union[str, int]: """ Select the best answer from the provided options based on the query with automatic translation and caching if needed. - NOTE: "query" and "options" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "query" and "options" are automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param query: The query text. :param options: A list of answer options. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :param return_index: If True, return the index of the best option; otherwise, return the best option text. :return: The best answer from the options list, or the index of the best option if `return_index` is True. """ - best = self.rerank(query, options, - context=context, lang=lang)[0][1] + best = self.rerank(query, options, lang=lang)[0][1] if return_index: return options.index(best) return best - # user facing methods - - @_deprecate_context2lang() - def solve(self, query: str, options: List[str], - context: Optional[Dict] = None, - lang: Optional[str] = None) -> str: - """ - Solve the multiple-choice question by selecting the best answer from the options with automatic translation and caching if needed. - - :param query: The question text. - :param options: A list of multiple-choice options. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) - :param lang: Optional language code. - :return: The best answer from the provided options. - """ - # select best answer - # NOTE: use index so we return exactly the source text - # there may have been an auto-translation step in self.select_answer - idx = self.select_answer(query, options, - context=context, lang=lang, - return_index=True) - return options[idx] - class EntailmentSolver(AbstractSolver): """ select best answer from question + multiple choice @@ -637,14 +573,12 @@ class EntailmentSolver(AbstractSolver): @abc.abstractmethod def check_entailment(self, premise: str, hypothesis: str, - context: Optional[Dict] = None, lang: Optional[str] = None) -> bool: """ Check if the premise entails the hypothesis. :param premise: The premise text, assured to be in the default language. :param hypothesis: The hypothesis text, assured to be in the default language. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: True if the premise entails the hypothesis; False otherwise. """ @@ -652,24 +586,21 @@ def check_entailment(self, premise: str, hypothesis: str, # user facing methods @_deprecate_context2lang() - @auto_translate_inputs(translate_keys=["premise", "hypothesis"]) + @auto_translate(translate_keys=["premise", "hypothesis"]) def entails(self, premise: str, hypothesis: str, - context: Optional[Dict] = None, lang: Optional[str] = None) -> bool: """ Determine if the premise entails the hypothesis with automatic translation and caching if needed. - NOTE: "premise" and "hypothesis" assured to be in self.supported_langs, - otherwise they are automatically translated to self.default_lang. + NOTE: "lang" assured to be in self.supported_langs, + otherwise "premise" and "hypothesis" are automatically translated to self.default_lang. If translations happens, the returned value of this method will also be automatically translated back :param premise: The premise text. :param hypothesis: The hypothesis text. - :param context: Optional context dictionary. (Deprecated, not used in the new implementation) :param lang: Optional language code. :return: True if the premise entails the hypothesis; False otherwise. """ # check for entailment - return self.check_entailment(premise, hypothesis, - context=context, lang=lang) + return self.check_entailment(premise, hypothesis, lang=lang)