from .base import Base
from .document_qa import DocumentQA
from .code_generator import CodeGenerator
from .document_classifier import DocumentClassifier

promptmeteo/__init__.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

import os
import tarfile
import tempfile
import json
from abc import ABC
from inspect import signature
from typing import List
from typing import Dict
from typing import Optional

try:
    from typing import Self
except ImportError:
    from typing_extensions import Self

from .tasks import Task
from .tasks import TaskBuilder
from .tools import add_docstring_from


class Base(ABC):

    """
    'Sun is setting on the New Republic. It's time for the ResistencIA to rise'

                                         - Padme Amidala, mother of Leia -
    """
    SELECTOR_TYPE: str = ''

    def __init__(
        self,
        language: str,
        model_name: str,
        model_provider_name: str,
        model_provider_token: Optional[str] = None,
        model_params: Optional[Dict] = None,
        prompt_domain: Optional[str] = "",
        prompt_labels: List[str] = None,
        prompt_detail: Optional[str] = None,
        selector_k: int = 10,
        selector_algorithm: str = "relevance",
        verbose: bool = False,
        **kwargs
    ) -> None:
        """
        Prompmeteo is tool, powered by LLMs, which is able to solve NLP models
        such as text classification and Named Entity Recognition. Its interface
        is similar to a conventional ML model, which allows Prometeo to be used
        in a MLOps pipeline easily.

        Parameters
        ----------

        task_type : str

        model_name : str

        model_provider_name : str

        model_provider_token : Optional[str]

        prompt_domain : str

        prompt_labels : List[str]

        prompt_task_info : Optional[str]

        selector_k : int

        selector_algorithm : str

        verbose : bool


        Returns
        -------

        None
        """

        _init_params: dict = {}
        _local = locals()

        for param in signature(self.__class__).parameters:
            if 'kwargs' in param:
                continue
            _init_params[param] = _local.get(param, kwargs.get(param))

        for param in signature(Base).parameters:
            if 'kwargs' in param:
                continue
            _init_params[param] = _local.get(param, kwargs.get(param))

        self._init_params: dict = _init_params

        self.language: str = language
        self.model_name: str = model_name
        self.model_provider_name: str = model_provider_name
        self.model_provider_token: Optional[str] = model_provider_token
        self.model_params: Dict = model_params or {}
        self.prompt_domain: Optional[str] = prompt_domain
        self.prompt_labels: List[str] = prompt_labels or []
        self.prompt_detail: Optional[str] = prompt_detail
        self._selector_k: int = selector_k
        self._selector_algorithm: str = selector_algorithm
        self.verbose: bool = verbose

        self._builder = None
        self._is_trained = False

    @property
    def builder(
        self,
    ) -> TaskBuilder:
        """
        Get TaskBuilder.
        """
        return self._builder

    @property
    def task(
        self,
    ) -> Task:
        """
        Get Task.
        """
        return self._builder.task

    @property
    def is_trained(
        self,
    ) -> bool:
        """
        Check if model intance is trained.
        """
        return self._is_trained

    def predict(
        self,
        examples: List[str],
    ) -> List[str]:
        """
        Predicts over new text samples.

        Parameters
        ----------

        examples : List[str]

        Returns
        -------

        List[str]

        """

        if not isinstance(examples, list):
            raise ValueError(
                f"{self.__class__.__name__} error in function `predict()`. "
                f"Arguments `examples` and `annotations` are expected to be "
                f"of type `List[str]`. Instead they got: `{type(examples)}`"
            )

        if not all([isinstance(val, str) for val in examples]):
            raise ValueError(
                f"{self.__class__.__name__} error in function `predict()`. "
                f"Arguments `examples` are expected to be of type "
                f"  `List[str]`. Some values seem no to be of type `str`."
            )

        results = []
        for example in examples:
            results.append(self.task.run(example))

        return results

    def save_model(
        self,
        model_path: str,
    ) -> Self:
        """
        Saves the training result of the model in disk.

        Parameters
        ----------

        model_path : str


        Returns
        -------

        self : Promptmeteo

        """

        model_dir = os.path.dirname(model_path)
        model_name = os.path.basename(model_path)

        if not self.is_trained:
            raise RuntimeError(
                f"{self.__class__.__name__} error in `save_model()`. "
                f"You are trying to save a model that has non been trained. "
                f"Please, call `train()` function before"
            )

        if not model_name.endswith(".meteo"):
            raise ValueError(
                f"{self.__class__.__name__} error in `save_model()`. "
                f'model_path="{model_path}" has a bad model name extension. '
                f"Model name must end with `.meteo` (i.e. `./model.meteo`)"
            )

        if model_dir != "" and not os.path.exists(model_dir):
            raise ValueError(
                f"{self.__class__.__name__} error in `save_model()`. "
                f"directory {model_dir} does not exists."
            )

        with tempfile.TemporaryDirectory() as tmp:
            tmp_path = os.path.join(tmp, model_name.replace(".meteo", ""))
            self.task.selector.vectorstore.save_local(tmp_path)

            init_tmp_path = f"{tmp_path}.init"
            with open(init_tmp_path, mode='w') as f:
                json.dump(self._init_params, f)

            with tarfile.open(model_path, mode="w:gz") as tar:
                tar.add(tmp_path, arcname=model_name)
                tar.add(init_tmp_path, arcname=os.path.basename(init_tmp_path))

        return self

    @classmethod
    def load_model(
        cls,
        model_path: str,
    ) -> Self:
        """
        Loads a model artifact to make new predictions.

        Parameters
        ----------

        model_path : str


        Returns
        -------

        self : Promptmeteo

        """

        model_dir = os.path.dirname(model_path)
        model_name = os.path.basename(model_path)

        if not model_name.endswith(".meteo"):
            raise ValueError(
                f"{cls.__name__} error in `load_model()`. "
                f'model_path="{model_path}" has a bad model name extension. '
                f"Model name must end with `.meteo` (i.e. `./model.meteo`)"
            )

        if not os.path.exists(model_path):
            raise ValueError(
                f"{cls.__name__} error in `load_model()`. "
                f"directory {model_dir} does not exists."
            )

        with tempfile.TemporaryDirectory() as tmp:
            with tarfile.open(model_path, "r:gz") as tar:
                tar.extractall(tmp)

            init_tmp_path = os.path.join(tmp, f"{os.path.splitext(model_name)[0]}.init")
            with open(init_tmp_path) as f:
                self = cls(**json.load(f))

            self.builder.build_selector_by_load(
                model_path=os.path.join(tmp, model_name),
                selector_type=self.SELECTOR_TYPE,
                selector_k=self._selector_k,
                selector_algorithm=self._selector_algorithm,
            )

            self._is_trained = True

        return self


class BaseSupervised(Base):

    """
    Model Inferface for supervised training tasks.
    """

    SELECTOR_TYPE = "supervised"

    @add_docstring_from(Base.__init__)
    def __init__(
        self,
        **kwargs,
    ) -> None:
        """
        """

        super(BaseSupervised, self).__init__(**kwargs)

    def train(
        self,
        examples: List[str],
        annotations: List[str],
    ) -> Self:
        """
        Trains the model given examples and its annotations. The training
        process create a vector store with all the training texts in memory

        Parameters
        ----------

        examples : List[str]

        annotations : List[str]


        Returns
        -------

        self

        """

        if not isinstance(examples, list) or not isinstance(annotations, list):
            raise ValueError(
                f"Arguments `examples` and `annotations` are expected to be of "
                f"type `List[str]`. Instead they got: examples {type(examples)}"
                f" annotations {type(annotations)}."
            )

        if not all([isinstance(val, str) for val in examples]) or not all(
            [isinstance(val, str) for val in annotations]
        ):
            raise ValueError(
                f"{self.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` and `annotations` are expected to be of "
                f"type `List[str]`. Some values seem no to be of type `str`."
            )

        if len(examples) != len(annotations):
            raise ValueError(
                f"{self.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` and `annotations` are expected to have "
                f"the same length. examples=({len(examples)},) annotations= "
                f"({len(annotations)},)"
            )

        if self.prompt_labels:
            for idx, annotation in enumerate(annotations):
                if annotation not in self.prompt_labels:
                    raise ValueError(
                        f"{self.__class__.__name__} error in `train()`. "
                        f"`annotation value in item {idx}: `{annotation}`"
                        f"is not in the expected values: {self.prompt_labels}"
                    )

        self.builder.build_selector_by_train(
            examples=examples,
            annotations=annotations,
            selector_k=self._selector_k,
            selector_type=self.SELECTOR_TYPE,
            selector_algorithm=self._selector_algorithm,
        )

        self._is_trained = True

        return self


class BaseUnsupervised(Base):
    """
    Model Inferface for unsupervised training tasks.
    """

    SELECTOR_TYPE = "unsupervised"

    @add_docstring_from(Base.__init__)
    def __init__(
        self,
        **kwargs,
    ) -> None:
        if "prompt_labels" in kwargs:
            raise ValueError(
                f"{self.__class__.__name__} can not be inicializated with the "
                f"argument `prompt_labels`."
            )

        super(BaseUnsupervised, self).__init__(**kwargs)

    def train(
        self,
        examples: List[str],
    ) -> Self:
        """
        Trains the model given examples and its annotations. The training
        process create a vector store with all the training texts in memory

        Parameters
        ----------

        examples : List[str]


        Returns
        -------

        self

        """

        if not isinstance(examples, list):
            raise ValueError(
                f"{self.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` are expected to be of type `List[str]`."
                f"Instead they got: examples {type(examples)}"
            )

        if not all([isinstance(val, str) for val in examples]):
            raise ValueError(
                f"{self.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` are expected to be of type `List[str]`."
                f"Some values seem no to be of type `str`."
            )

        self.builder.build_selector_by_train(
            examples=examples,
            annotations=None,
            selector_k=self._selector_k,
            selector_type=self.SELECTOR_TYPE,
            selector_algorithm=self._selector_algorithm,
        )

        self._is_trained = True

        return self

promptmeteo/base.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from .tasks import TaskTypes
from .tasks import TaskBuilder
from .base import BaseSupervised
from .tools import add_docstring_from


class CodeGenerator(BaseSupervised):

    """
    Code Generator Task.
    """

    @add_docstring_from(BaseSupervised.__init__)
    def __init__(
        self,
        **kwargs,
    ) -> None:
        """
        Example
        -------

        >>> from promptmeteo import CodeGenerator

        >>> model = CodeGenerator(
        >>>     language='en',
        >>>     prompt_domain='python',
        >>>     model_provider_name='openai',
        >>>     model_name='text-davinci-003',
        >>>     model_provider_token=model_token,
        >>>     prompt_detail=[
        >>>         "add docstring in function definitions",
        >>>         "add argumment typing annotations"]
        >>>     )

        >>> pred=model.predict(['A function that receives the argument `foo`
        >>>     and prints it.'])

        def print_foo(foo: str):
            '''Prints the argument passed into the function'''
            print(foo)

        """

        super(CodeGenerator, self).__init__(**kwargs)

        #        if self.prompt_detail is None:
        #            raise ValueError(
        #                f"{self.__class__.__name__} error in initialization. "
        #                f"argument `prompt_detail` can not be None."
        #            )

        task_type = TaskTypes.CODE_GENERATION.value

        self._builder = TaskBuilder(
            language=self.language,
            task_type=task_type,
            verbose=self.verbose,
        )

        # Build model
        self._builder.build_model(
            model_name=self.model_name,
            model_provider_name=self.model_provider_name,
            model_provider_token=self.model_provider_token,
            model_params=self.model_params,
        )

        # Build prompt
        self._builder.build_prompt(
            model_name=self.model_name,
            prompt_domain=self.prompt_domain,
            prompt_labels=self.prompt_labels,
            prompt_detail=self.prompt_detail,
        )

        # Build parser
        self._builder.build_parser(
            prompt_labels=self.prompt_labels,
        )

promptmeteo/code_generator.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from typing import List

try:
    from typing import Self
except ImportError:
    from typing_extensions import Self

from .tasks import TaskTypes
from .tasks import TaskBuilder
from .base import BaseSupervised
from .tools import add_docstring_from


class DocumentClassifier(BaseSupervised):

    """
    DocumentClassifier Task
    """

    @add_docstring_from(BaseSupervised.__init__)
    def __init__(
        self,
        **kwargs,
    ) -> None:
        """
        Example
        -------

        >>> from promptmeteo import DocumentClassifier

        >>> clf = DocumentClassifier(
        >>>     model_provider_name='hf_pipeline',
        >>>     model_name='google/flan-t5-small',
        >>>     prompt_labels=['positive','negative','neutral'],
        >>> )

        >>> clf.train(
        >>>     examples = ['estoy feliz', 'me da igual', 'no me gusta'],
        >>>     annotations = ['positive', 'neutral', 'negative']
        >>> )

        >>> clf.predict(['que guay!!'])

        [['positive']]

        """

        super(DocumentClassifier, self).__init__(**kwargs)

        task_type = TaskTypes.CLASSIFICATION.value

        self._builder = TaskBuilder(
            language=self.language,
            task_type=task_type,
            verbose=self.verbose,
        )

        # Build model
        self._builder.build_model(
            model_name=self.model_name,
            model_provider_name=self.model_provider_name,
            model_provider_token=self.model_provider_token,
            model_params=self.model_params,
        )

        # Build prompt
        self._builder.build_prompt(
            model_name=self.model_name,
            prompt_domain=self.prompt_domain,
            prompt_labels=self.prompt_labels,
            prompt_detail=self.prompt_detail,
        )

        # Build parser
        self._builder.build_parser(
            prompt_labels=self.prompt_labels,
        )

    @add_docstring_from(BaseSupervised.train)
    def train(
        self,
        examples: List[str],
        annotations: List[str],
    ) -> Self:
        """ """

        super(DocumentClassifier, self).train(
            examples=examples, annotations=annotations
        )

        if not self.prompt_labels:
            self.prompt_labels = list(set(annotations))
            self._builder.build_parser(
                prompt_labels=self.prompt_labels,
            )

        return self

promptmeteo/document_classifier.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from .tasks import TaskTypes
from .tasks import TaskBuilder
from .base import BaseUnsupervised
from .tools import add_docstring_from


class DocumentQA(BaseUnsupervised):

    """
    Question Answering over Documents Task
    """

    @add_docstring_from(BaseUnsupervised.__init__)
    def __init__(
        self,
        **kwargs,
    ) -> None:
        """
        Example
        -------

        >>> from promptmeteo import DocumentQA

        >>> clf = DocumentQA(
        >>>     language='en',
        >>>     model_provider_name='hf_pipeline',
        >>>     model_name='google/flan-t5-small',
        >>> )

        >>> clf.train(
        >>>     examples = [
        >>>     "The rain in spain is always in plain",
        >>>     "The logarithm's limit is the limit's logarithm",
        >>>     "To punish oppresors is clementy. To forgive them is cruelty"],
        >>> )

        >>> clf.predict(['How is the rain in spain?'])

        [['in plain']]
        """

        super(DocumentQA, self).__init__(**kwargs)

        task_type = TaskTypes.QA.value

        self._builder = TaskBuilder(
            language=self.language,
            task_type=task_type,
            verbose=self.verbose,
        )

        # Build model
        self._builder.build_model(
            model_name=self.model_name,
            model_provider_name=self.model_provider_name,
            model_provider_token=self.model_provider_token,
            model_params=self.model_params,
        )

        # Build prompt
        self._builder.build_prompt(
            model_name=self.model_name,
            prompt_domain=self.prompt_domain,
            prompt_labels=self.prompt_labels,
            prompt_detail=self.prompt_detail,
        )

        # Build parser
        self._builder.build_parser(
            prompt_labels=self.prompt_labels,
        )

promptmeteo/document_qa.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from enum import Enum
from typing import Dict

from .base import BaseModel
from .openai import OpenAILLM
from .fake_llm import FakeLLM
from .hf_hub_api import HFHubApiLLM
from .hf_pipeline import HFPipelineLLM
from .google_vertexai import GoogleVertexAILLM


class ModelProvider(str, Enum):

    """
    LLM providers currently supported by Promptmeteo
    """

    PROVIDER_0: str = "fake-llm"
    PROVIDER_1: str = "openai"
    PROVIDER_2: str = "hf_hub_api"
    PROVIDER_3: str = "hf_pipeline"
    PROVIDER_4: str = "google-vertexai"



class ModelFactory:

    """
    The ModelFactory class is used to create a BaseModel object from the given
    configuration.
    """

    MAPPING = {
        ModelProvider.PROVIDER_0: FakeLLM,
        ModelProvider.PROVIDER_1: OpenAILLM,
        ModelProvider.PROVIDER_2: HFHubApiLLM,
        ModelProvider.PROVIDER_3: HFPipelineLLM,
        ModelProvider.PROVIDER_3: GoogleVertexAILLM,
    }

    @classmethod
    def factory_method(
        cls,
        model_name: str,
        model_provider_name: str,
        model_provider_token: str,
        model_params: Dict,
    ) -> BaseModel:
        """
        Returns a BaseModel object configured with the settings found in the
        provided parameters.
        """
        model_cls = cls.MAPPING.get(model_provider_name)

        if model_provider_name == ModelProvider.PROVIDER_0.value:
            model_cls = FakeLLM

        elif model_provider_name == ModelProvider.PROVIDER_1.value:
            model_cls = OpenAILLM

        elif model_provider_name == ModelProvider.PROVIDER_2.value:
            model_cls = HFHubApiLLM

        elif model_provider_name == ModelProvider.PROVIDER_3.value:
            model_cls = HFPipelineLLM
        
        elif model_provider_name == ModelProvider.PROVIDER_4.value:
            model_cls = GoogleVertexAILLM

        else:
            raise ValueError(
                f"{cls.__name__} error in `factory_method()`. "
                f"{model_provider_name} is not in the list of supported "
                f"providers: {[i.value for i in ModelProvider]}"
            )

        return model_cls(
            model_name=model_name,
            model_params=model_params,
            model_provider_token=model_provider_token,
        )

promptmeteo/models/__init__.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from abc import ABC
from langchain.llms.base import BaseLLM
from langchain.embeddings.base import Embeddings


class BaseModel(ABC):

    """
    Model Interface.
    """

    def __init__(self):
        self._llm: BaseLLM = None
        self._embeddings: Embeddings = None

    @property
    def llm(
        self,
    ) -> BaseLLM:
        """Get Model LLM."""
        return self._llm

    @property
    def embeddings(
        self,
    ) -> Embeddings:
        """Get Model Embeddings."""
        return self._embeddings

    def run(
        self,
        sample: str,
    ) -> str:
        """
        Executes the model LLM and return its prediction.
        """

        try:
            return self._llm(prompt=sample)

        except Exception as error:
            raise RuntimeError(
                f'Error generating from LLM: with sample "{sample}"'
            ) from error

promptmeteo/models/base.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from enum import Enum
from typing import Any
from typing import List
from typing import Dict
from typing import Mapping
from typing import Optional

from langchain.callbacks.manager import AsyncCallbackManagerForLLMRun
from langchain.llms.base import LLM
from langchain.embeddings import FakeEmbeddings

from .base import BaseModel


class FakeStaticLLM(LLM):

    """
    Fake Static LLM wrapper for testing purposes.
    """

    response: str = "positive"

    @property
    def _llm_type(
        self,
    ) -> str:
        """
        Return type of llm.
        """

        return "fake-static"

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[Any] = None,
        **kwargs: Any,
    ) -> str:
        """
        Return static response.
        """

        return self.response

    @property
    def _identifying_params(
        self,
    ) -> Mapping[str, Any]:
        return {}

    async def _acall(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        return self._call(prompt, stop, run_manager, **kwargs)


class FakePromptCopyLLM(LLM):

    """
    Fake Prompt Copy LLM wrapper for testing purposes.
    """

    @property
    def _llm_type(
        self,
    ) -> str:
        """
        Return type of llm.
        """

        return "fake-prompt-copy"

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[Any] = None,
        **kwargs: Any,
    ) -> str:
        """
        Return prompt.
        """

        return prompt

    @property
    def _identifying_params(
        self,
    ) -> Mapping[str, Any]:
        return {}

    async def _acall(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        return self._call(prompt, stop, run_manager, **kwargs)


class FakeListLLM(LLM):

    """
    Fake LLM wrapper for testing purposes.
    """

    responses: List = ["uno", "dos", "tres"]
    i: int = 0

    @property
    def _llm_type(
        self,
    ) -> str:
        """
        Return type of llm.
        """

        return "fake-list"

    def _call(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[Any] = None,
        **kwargs: Any,
    ) -> str:
        """
        First try to lookup in queries, else return 'foo' or 'bar'.
        """

        response = self.responses[self.i]
        self.i += 1

        return response

    @property
    def _identifying_params(
        self,
    ) -> Mapping[str, Any]:
        return {}

    async def _acall(
        self,
        prompt: str,
        stop: Optional[List[str]] = None,
        run_manager: Optional[AsyncCallbackManagerForLLMRun] = None,
        **kwargs: Any,
    ) -> str:
        return self._call(prompt, stop, run_manager, **kwargs)


class ModelTypes(Enum):

    """
    FakeLLM Model Types.
    """

    MODEL_1: str = "fake-static"
    MODEL_2: str = "fake-prompt_copy"
    MODEL_3: str = "fake-list"


class FakeLLM(BaseModel):
    """
    Fake LLM class.
    """

    LLM_MAPPING: Dict[str, LLM] = {
        ModelTypes.MODEL_1.value: FakeStaticLLM,
        ModelTypes.MODEL_2.value: FakePromptCopyLLM,
        ModelTypes.MODEL_3.value: FakeListLLM,
    }

    def __init__(
        self,
        model_name: Optional[str] = "",
        model_params: Optional[Dict] = None,
        model_provider_token: Optional[str] = "",
    ) -> None:
        super(FakeLLM, self).__init__()
        self.model_params = model_params
        self.model_provider_token = model_provider_token

        self._embeddings = FakeEmbeddings(size=64)
        if model_name in self.LLM_MAPPING:
            self._llm = self.LLM_MAPPING[model_name]()
        else:
            raise ValueError(
                f"{self.__class__.__name__} error creating object. "
                f"{model_name} is not in the list of supported FakeLLMS: "
                f"{[i.value for i in ModelTypes]}"
            )

promptmeteo/models/fake_llm.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

import os
from enum import Enum
from typing import Dict
from typing import Optional

from langchain.llms import VertexAI
from langchain.embeddings import VertexAIEmbeddings

from .base import BaseModel


class ModelTypes(str, Enum):

    """
    Enum of available model types.
    """

    TextBison: str = "text-bison" #latest version
    TextBison001: str = "text-bison@001" 
    TextBison32k: str = "text-bison-32k" # latest version of text-bison with 32k token input

    @classmethod
    def has_value(
        cls,
        value: str,
    ) -> bool:
        """
        Checks if the value is in the enum or not.
        """

        return value in cls._value2member_map_


class ModelParams(Enum):

    """
    Model Parameters.
    """

    class TextBison001:

        """
        Default parameters for text-bison model.
        """

        model_task: str = "text-bison@001"
        model_kwargs = {"temperature": 0.4, "max_tokens": 256, "max_retries": 3}
    
    class TextBison:

        """
        Default parameters for text-bison model in their lastest version
        """

        model_task: str = "text-bison"
        model_kwargs = {"temperature": 0.4, "max_tokens": 256, "max_retries": 3}
    
    class TextBison32k:

        """
        Default parameters for text-bison-32 model in their lastest version
        """

        model_task: str = "text-bison-32k"
        model_kwargs = {"temperature": 0.4, "max_tokens": 256, "max_retries": 3}


class GoogleVertexAILLM(BaseModel):

    """
    Google VertexAI LLM model.
    """

    def __init__(
        self,
        model_name: Optional[str] = "",
        model_params: Optional[Dict] = None,
        model_provider_token: Optional[str] = "",
        model_provider_project: Optional[str] = None,
    ) -> None:
        """
        Make predictions using a model from Google Vertex AI.
        It will use the os environment called GOOGLE_PROJECT_ID for instance the LLM
        """

        if not ModelTypes.has_value(model_name):
            raise ValueError(
                f"`model_name`={model_name} not in supported model names: "
                f"{[i.name for i in ModelTypes]}"
            )
        super(GoogleVertexAILLM, self).__init__()

        self._llm = VertexAI(
            model_name=model_name,
            project=model_provider_project or os.environ.get("GOOGLE_CLOUD_PROJECT_ID"),
        )

        self._embeddings = VertexAIEmbeddings()

        if not model_params:
            model_params = ModelParams[ModelTypes(model_name).name].value
        self.model_params = model_params

promptmeteo/models/google_vertexai.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from enum import Enum
from typing import Optional

from langchain import HuggingFaceHub
from langchain.embeddings import HuggingFaceHubEmbeddings

from .base import BaseModel


class ModelTypes(str, Enum):

    """
    Enum of available model types.
    """

    FlanT5Xxl: str = "google/flan-t5-xxl"

    @classmethod
    def has_value(
        cls,
        value: str,
    ) -> bool:
        """
        Checks if the value is in the enum or not.
        """

        return value in cls._value2member_map_


class ModelParams(Enum):

    """
    Model Parameters Enum
    """

    class FlanT5Xxl:

        """
        Flan-t5-xxl default params
        """

        model_task = "text2text-generation"
        model_kwargs = {"temperature": 0.9, "max_length": 64}


class HFHubApiLLM(BaseModel):

    """
    HuggingFace API call.
    """

    def __init__(
        self,
        model_name: Optional[str] = "",
        model_params: Optional[ModelParams] = None,
        model_provider_token: Optional[str] = "",
    ) -> None:
        """
        Make predictions using a model from HuggingFace using the API.
        """

        if ModelTypes.has_value(model_name):
            model_params = ModelParams[ModelTypes(model_name).name].value
        elif not model_params:
            raise ValueError(
                f"`model_name`={model_name} not in supported model names: "
                f"{[i.value for i in ModelTypes]}"
            )
        elif not hasattr(model_params, "model_kwargs"):
            raise ValueError("invalid model_params")

        super(HFHubApiLLM, self).__init__()

        self._llm = HuggingFaceHub(
            repo_id=model_name,
            huggingfacehub_api_token=model_provider_token,
            model_kwargs=model_params.model_kwargs,
        )

        self._embeddings = HuggingFaceHubEmbeddings(
            repo_id="sentence-transformers/all-MiniLM-L6-v2",
            huggingfacehub_api_token=model_provider_token,
        )

promptmeteo/models/hf_hub_api.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

import os
from enum import Enum
from typing import Optional

from langchain import HuggingFacePipeline
from langchain.embeddings import HuggingFaceEmbeddings

from .base import BaseModel


class ModelTypes(str, Enum):

    """
    Enum of available model types.
    """

    MODEL_1 = "google/flan-t5-small"

    @classmethod
    def has_value(cls, value):
        """
        Return whether the value is in the class or not.
        """

        return value in cls._value2member_map_


class ModelParams(Enum):

    """
    Model Parameters.
    """

    class MODEL_1:

        """
        Parameters Model 1.
        """

        model_path = "/home/models/flan-t5-small"
        model_task = "text2text-generation"
        model_kwargs = {"temperature": 0.0, "max_length": 64}


class HFPipelineLLM(BaseModel):

    """
    HuggingFace Local Pipeline.
    """

    def __init__(
        self,
        model_name: Optional[str] = "",
        model_params: Optional[ModelParams] = None,
        model_provider_token: Optional[str] = "",
    ) -> None:
        """
        Make predictions using a model from HuggingFace locally.
        """

        if ModelTypes.has_value(model_name):
            model_params = ModelParams[ModelTypes(model_name).name].value
        elif not model_params:
            raise ValueError(
                f"`model_name`={model_name} not in supported model names: "
                f"{[i.value for i in ModelTypes]}"
            )
        elif any(
            [
                not hasattr(model_params, "model_path"),
                not hasattr(model_params, "model_task"),
                not hasattr(model_params, "model_kwargs"),
            ]
        ):
            raise ValueError("Invalid model_params")

        super(HFPipelineLLM, self).__init__()

        if os.path.exists(model_params.model_path):
            model_name = model_params.model_path

        self._llm = HuggingFacePipeline.from_model_id(
            model_id=model_name,
            task=model_params.model_task,
            model_kwargs=model_params.model_kwargs,
        )

        embedding_name = "sentence-transformers/all-MiniLM-L6-v2"
        if os.path.exists("/home/models/all-MiniLM-L6-v2"):
            embedding_name = "/home/models/all-MiniLM-L6-v2"

        self._embeddings = HuggingFaceEmbeddings(model_name=embedding_name)

        self.model_provider_token = model_provider_token

promptmeteo/models/hf_pipeline.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

import os
from enum import Enum
from typing import Dict
from typing import Optional

from langchain.llms import OpenAI
from langchain.embeddings import OpenAIEmbeddings

from .base import BaseModel


class ModelTypes(str, Enum):

    """
    Enum of available model types.
    """

    TextDavinci003: str = "text-davinci-003"

    @classmethod
    def has_value(
        cls,
        value: str,
    ) -> bool:
        """
        Checks if the value is in the enum or not.
        """

        return value in cls._value2member_map_


class ModelParams(Enum):

    """
    Model Parameters.
    """

    class TextDavinci003:

        """
        Default parameters for TextDavinci003 model.
        """

        model_task: str = "text2text-generation"
        model_kwargs = {"temperature": 0.7, "max_tokens": 256, "max_retries": 3}


class OpenAILLM(BaseModel):

    """
    OpenAI LLM model.
    """

    def __init__(
        self,
        model_name: Optional[str] = "",
        model_params: Optional[Dict] = None,
        model_provider_token: Optional[str] = "",
    ) -> None:
        """
        Make predictions using a model from OpenAI.
        It will use the os environment called OPENAI_ORGANIZATION for instance the LLM
        """

        if not ModelTypes.has_value(model_name):
            raise ValueError(
                f"`model_name`={model_name} not in supported model names: "
                f"{[i.name for i in ModelTypes]}"
            )
        super(OpenAILLM, self).__init__()

        self._llm = OpenAI(
            model_name=model_name,
            openai_api_key=model_provider_token,
            openai_organization=os.environ.get("OPENAI_ORGANIZATION", ""),
        )

        self._embeddings = OpenAIEmbeddings(openai_api_key=model_provider_token)

        if not model_params:
            model_params = ModelParams[ModelTypes(model_name).name].value
        self.model_params = model_params

promptmeteo/models/openai.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from enum import Enum
from typing import List

from .base import BaseParser
from .dummy_parser import DummyParser
from .classification_parser import ClassificationParser


class ParserTypes(str, Enum):

    """
    Enum of availables parsers.
    """

    PARSER_1: str = "classification"
    PARSER_2: str = "ner"
    PARSER_3: str = "qa"
    PARSER_4: str = "code-generation"


class ParserFactory:

    """
    Factory of Parsers.
    """

    @classmethod
    def factory_method(
        cls,
        task_type: str,
        prompt_labels: List[str],
    ):
        """
        Returns and instance of a BaseParser object depending on the
        `task_type`.
        """

        if task_type == ParserTypes.PARSER_1.value:
            parser_cls = ClassificationParser

        elif task_type == ParserTypes.PARSER_2.value:
            parser_cls = DummyParser

        elif task_type == ParserTypes.PARSER_3.value:
            parser_cls = DummyParser

        elif task_type == ParserTypes.PARSER_4.value:
            parser_cls = DummyParser

        else:
            raise ValueError(
                f"`{cls.__name__}` error in function `factory_method()`: "
                f"{task_type} is not in the list of supported providers: "
                f"{[i.value for i in ParserTypes]}"
            )

        return parser_cls(
            prompt_labels=prompt_labels,
        )

promptmeteo/parsers/__init__.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from abc import ABC
from abc import abstractmethod

from typing import List


class BaseParser(ABC):

    """
    Parser class interface.
    """

    def __init__(
        self,
        prompt_labels: List[str],
    ) -> None:
        self._labels = prompt_labels
        self._labels_separator = ","
        self._chain_of_thoughts = True

    @abstractmethod
    def run(
        self,
        text: str,
    ) -> List[str]:
        """
        Given a response string from an LLM, returns the response expected for
        the task.
        """

        raise NotImplementedError

promptmeteo/parsers/base.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from typing import List

from .base import BaseParser


class ClassificationParser(BaseParser):

    """
    Parser for the classification task.
    """

    def run(
        self,
        text: str,
    ) -> List[str]:
        """
        Given a response string from an LLM, returns the response expected for
        the task.
        """

        text = self._preprocess(text)

        result = [
            label for label in self._labels if label.lower() in text.split()
        ]

        return result if result else [""]

    def _preprocess(
        self,
        text: str,
    ) -> str:
        """
        Preprocess output string before parsing result to solve common mistakes
        such as end-of-line presence and beginning and finishing with empty
        space.
        """

        return " ".join(
            text.lower()
            .replace("\n", " ")
            .strip()
            .split(self._labels_separator)
        )

promptmeteo/parsers/classification_parser.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from typing import List

from .base import BaseParser


class DummyParser(BaseParser):

    """
    Dummy parser, returns what it receives.
    """

    def run(
        self,
        text: str,
    ) -> List[str]:
        """
        Given a response string from an LLM, returns the response expected for the task.
        """

        return [text]

promptmeteo/parsers/dummy_parser.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

import os
from enum import Enum
from typing import List

from .base import BasePrompt


module_dir = os.path.abspath(os.path.join(__file__, os.path.pardir))


def get_files_taxonomy(sep: str = "_"):
    """
    Convert a list of prompt files with the naming convetion of
    `{model_name}_{language}_{task}.prompt into a dictorionary version.
    Given the following folder structure:

        ./prompts/
          ├── __init__.py
          ├── base.py
          ├── google-flan-t5-small_es_classification.prompt
          ├── google-flan-t5-small_es_code-generation.prompt
          ├── google-flan-t5-small_es_ner.prompt
          ├── text-davinci-003_es_classification.prompt
          ├── text-davinci-003_es_code-generation.prompt
          └── text-davinci-003_es_ner.prompt

    The function returns:

        {'google-flan-t5-small': {
             'es': {
                 'classification': {},
                 'ner': {}
                 }
              },
         'text-davinci-003': {
             'es': {
                 'classification': {},
                 'code-generation': {},
                 'ner': {}
                 }
              }
        }

    """

    try:
        prompt_files = [
            os.path.join(path, name.replace(".prompt", "")).replace(
                module_dir + "/", ""
            )
            for path, _, files in os.walk(module_dir)
            for name in files
            if name.endswith(".prompt")
        ]

        taxonomy = {}
        for prompt_file in prompt_files:
            tmp = taxonomy
            for field in prompt_file.split(sep):
                tmp = tmp.setdefault(field, {})

    except Exception as error:
        raise RuntimeError(
            "Problems with the prompts file structure in directory "
            "promptmeteo/prompts. The expected naming for the prompts "
            "is `<model_name>_<language>_<task>.prompt`."
        ) from error

    return taxonomy


class PromptFactory:

    """
    Factory of Prompts
    """

    @classmethod
    def factory_method(
        cls,
        language: str,
        task_type: str,
        model_name: str,
        prompt_domain: str,
        prompt_labels: List[str],
        prompt_detail: str,
    ):
        """
        Returns and instance of a BasePrompt object depending on the
        `task_type`.
        """

        _model_name = model_name.replace("/", "-")

        taxonomy = get_files_taxonomy()

        if _model_name not in taxonomy:
            raise ValueError(
                f"`{cls.__name__}` class in function `factory_method()`. "
                f"{model_name} has not a prompt file created. Available model "
                f"prompts are: {list(taxonomy)}"
            )

        if language not in taxonomy[_model_name]:
            raise ValueError(
                f"`{cls.__name__}` class in function `factory_method()`. "
                f"{model_name} has not a prompt file created for the language "
                f"{language}. Available languages for {model_name}: "
                f"{list(taxonomy[model_name])}"
            )

        if task_type not in taxonomy[_model_name][language]:
            raise ValueError(
                f"`{cls.__name__}` class in function `factory_method()`. "
                f"{model_name}  in {language} has not a prompt file created"
                f"for the task {task_type}. Available tasks are: "
                f"{list(taxonomy[model_name][language])}"
            )

        prompt_cls = cls.build_class(language, task_type, _model_name)

        return prompt_cls(
            prompt_domain=prompt_domain,
            prompt_labels=prompt_labels,
            prompt_detail=prompt_detail,
        )

    @staticmethod
    def build_class(
        language: str,
        task_type: str,
        _model_name: str,
    ) -> BasePrompt:
        """
        Creates a class dinamically that inherits from `BasePrompt` given
        arguments configuracion. This new class has included its prompt from
        the .prompt file.

        Example
        -------

        >>> PromptFactory.build_class(
        >>>     language='es',
        >>>     task_type='ner',
        >>>     model_name='text-davinci-003')
        >>> )

        <class '__main__.Textdavinci003SpNer'>
        """

        prompt_class_name = "".join(
            [
                _model_name.replace("-", "").capitalize(),
                language.capitalize(),
                task_type.replace("-", "").capitalize(),
            ]
        )

        prompt_cls = type(
            prompt_class_name, (BasePrompt,), {"__init__": BasePrompt.__init__}
        )

        file_name = f"{_model_name}_{language}_{task_type}.prompt"
        with open(os.path.join(module_dir, file_name), encoding="utf-8") as fin:
            prompt_cls.read_prompt(fin.read())

        return prompt_cls

promptmeteo/prompts/__init__.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from abc import ABC
from typing import List

import yaml
from langchain import PromptTemplate
from langchain.prompts.pipeline import PipelinePromptTemplate


class BasePrompt(ABC):

    """
    Prompt class interface.
    """

    PROMPT_EXAMPLE = """
        TEMPLATE:
            "Here you exaplain the task.
            {__PROMPT_DOMAIN__}
            {__PROMPT_LABELS__}

            {__CHAIN_THOUGHT__}
            {__ANSWER_FORMAT__}"

        PROMPT_DOMAIN:
            "Here you explain the {__DOMAIN__} from the texts."

        PROMPT_LABELS:
            "Here you give the {__LABELS__} if required."

        PROMPT_DETAIL:
            "Here you can give some {__DETAIL__}"

        CHAIN_THOUGHT:
            "Explain your answer step by step."

        ANSWER_FORMAT:
            "Response just with the asnwer."
    """

    def __init__(
        self,
        prompt_domain: str = "",
        prompt_labels: str = "",
        prompt_detail: str = "",
    ) -> None:
        """
        Build a Prompt object string given a concrete especification.
        """

        self._prompt_domain = prompt_domain
        self._prompt_labels = prompt_labels
        self._prompt_detail = prompt_detail

    @property
    def domain(
        self,
    ) -> str:
        """Prompt Domain."""
        return self._prompt_domain

    @property
    def labels(
        self,
    ) -> List[str]:
        """Prompt Labels."""
        return [self._prompt_labels]

    @property
    def template(
        self,
    ) -> str:
        """Prompt Template."""
        return self.run().format()

    @classmethod
    def read_prompt(
        cls,
        prompt_text: str,
    ) -> None:
        """
        Reads a Promptmeteo prompt string to build the Task Prompt. Promptmeteo
        prompts are expected to follow the following template:


        Parameters
        ----------

        prompt_text : str


        Returns
        -------

        Self

        """

        try:
            prompt = yaml.load(prompt_text, Loader=yaml.FullLoader)

        except Exception as error:
            raise ValueError(
                f"`{cls.__name__}` error in function `read_prompt()`. "
                f"The expected string input should be like:\n\n"
                f"{cls.PROMPT_EXAMPLE}\n\n{error}"
            ) from error

        try:
            cls.TEMPLATE = prompt["TEMPLATE"]
            cls.PROMPT_DOMAIN = prompt["PROMPT_DOMAIN"]
            cls.PROMPT_LABELS = prompt["PROMPT_LABELS"]
            cls.PROMPT_DETAIL = prompt["PROMPT_DETAIL"]
            cls.ANSWER_FORMAT = prompt["ANSWER_FORMAT"]
            cls.CHAIN_THOUGHT = prompt["CHAIN_THOUGHT"]

            cls.PROMPT_EXAMPLE = prompt_text

        except Exception as error:
            raise ValueError(
                f"`{cls.__name__}` error `read_prompt()`. The expected keys "
                f"are {yaml.load(cls.PROMPT_EXAMPLE, Loader=yaml.FullLoader)}"
            ) from error

    def run(
        self,
    ) -> PromptTemplate:
        """
        Returns the prompt template for the current task.
        """

        # Labels
        prompt_labels = (
            ", ".join(self._prompt_labels)
            if isinstance(self._prompt_labels, list)
            else self._prompt_detail
        )
        prompt_labels = (
            self.PROMPT_LABELS.format(__LABELS__=prompt_labels)
            if self._prompt_labels
            else ""
        )

        # Domain
        prompt_domain = (
            self.PROMPT_DOMAIN.format(__DOMAIN__=self._prompt_domain)
            if self._prompt_domain
            else ""
        )

        # Detail
        prompt_detail = (
            "\n - ".join([""] + self._prompt_detail)
            if isinstance(self._prompt_detail, list)
            else self._prompt_detail
        )
        prompt_detail = (
            self.PROMPT_DETAIL.format(__DETAIL__=prompt_detail)
            if self._prompt_detail
            else ""
        )

        # Answer format
        answer_format = self.ANSWER_FORMAT

        # Chain of thoughts
        chain_thought = self.CHAIN_THOUGHT

        return PipelinePromptTemplate(
            final_prompt=PromptTemplate.from_template(self.TEMPLATE),
            pipeline_prompts=[
                (
                    "__PROMPT_DOMAIN__",
                    PromptTemplate.from_template(prompt_domain),
                ),
                (
                    "__PROMPT_LABELS__",
                    PromptTemplate.from_template(prompt_labels),
                ),
                (
                    "__PROMPT_DETAIL__",
                    PromptTemplate.from_template(prompt_detail),
                ),
                (
                    "__ANSWER_FORMAT__",
                    PromptTemplate.from_template(answer_format),
                ),
                (
                    "__CHAIN_THOUGHT__",
                    PromptTemplate.from_template(chain_thought),
                ),
            ],
        )

promptmeteo/prompts/base.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

import os
from enum import Enum
from typing import List
from typing import Dict

from langchain.embeddings.base import Embeddings

from .base import BaseSelector
from .base import BaseSelectorSupervised
from .base import BaseSelectorUnsupervised


class SelectorTypes(str, Enum):

    """
    Enum with the avaialable selector types.
    """

    SUPERVISED: str = "supervised"
    UNSUPERVISED: str = "unsupervised"


class SelectorFactory:

    """
    Factory of Selectors
    """

    @classmethod
    def factory_method(
        cls,
        language: str,
        embeddings: Embeddings,
        selector_k: int,
        selector_type: str,
        selector_algorithm: str,
    ) -> BaseSelector:
        """
        Returns and instance of a BaseSelector object depending on the
        `selector_algorithm`.
        """

        if selector_type == SelectorTypes.SUPERVISED.value:
            selector_cls = BaseSelectorSupervised

        elif selector_type == SelectorTypes.UNSUPERVISED.value:
            selector_cls = BaseSelectorUnsupervised

        else:
            raise ValueError(
                f"`{cls.__name__}` error in `factory_method()` . "
                f"{selector_type} is not in the list of supported "
                f"providers: {[i.value for i in SelectorTypes]}"
            )

        return selector_cls(
            language=language,
            embeddings=embeddings,
            selector_k=selector_k,
            selector_algorithm=selector_algorithm,
        )

promptmeteo/selector/__init__.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from abc import ABC
from enum import Enum
from typing import List
from typing import Optional

try:
    from typing import Self
except ImportError:
    from typing_extensions import Self

from langchain.vectorstores import FAISS
from langchain.embeddings.base import Embeddings

from langchain.prompts import PromptTemplate
from langchain.prompts import FewShotPromptTemplate
from langchain.prompts.example_selector import (
    SemanticSimilarityExampleSelector,
    MaxMarginalRelevanceExampleSelector,
)


class SelectorAlgorithms(str, Enum):

    """
    Enum with the avaialable selector algorithms.
    """

    RELEVANCE: str = "relevance"
    SIMILARITY: str = "similarity"


class BaseSelector(ABC):

    """
    Base Selector Interface
    """

    SELECTOR = None

    def __init__(
        self,
        language: str,
        embeddings: Embeddings,
        selector_k: int,
        selector_algorithm: str,
    ) -> None:
        self._language = language
        self._embeddings = embeddings
        self._selector_k = selector_k

        if selector_algorithm == SelectorAlgorithms.SIMILARITY.value:
            self.selector = SemanticSimilarityExampleSelector

        elif selector_algorithm == SelectorAlgorithms.RELEVANCE.value:
            self.selector = MaxMarginalRelevanceExampleSelector

        else:
            raise ValueError(
                f"`{self.__class__.__name__}` error in __init__. "
                f"`selector_algorithm` value `{selector_algorithm}` is not in "
                f"the available values:{[i.value for i in SelectorAlgorithms]}"
            )

    @property
    def vectorstore(self):
        """
        Selector Vectorstore.
        """
        return self._selector.vectorstore

    @property
    def template(
        self,
    ) -> str:
        """
        Selector Template
        """
        return self.run().format(__INPUT__="{__INPUT__}")

    def load_example_selector(
        self,
        model_path: str,
    ) -> Self:
        """
        Load a vectorstore database from a disk file
        """

        vectorstore = FAISS.load_local(
            model_path,
            self._embeddings,
        )

        self._selector = self.selector(
            vectorstore=vectorstore,
            k=self._selector_k,
        )

        return self


class BaseSelectorSupervised(BaseSelector):
    def train(
        self,
        examples: List[str],
        annotations: List[str],
    ) -> Self:
        """
        Creates the vectorstor with the training samples.
        """

        examples = [
            {"__INPUT__": example, "__OUTPUT__": annotation}
            for example, annotation in zip(examples, annotations)
        ]

        self._selector = self.selector.from_examples(
            examples=examples,
            embeddings=self._embeddings,
            vectorstore_cls=FAISS,
            k=self._selector_k,
        )

        return self

    def run(
        self,
    ) -> FewShotPromptTemplate:
        """
        Creates the FewShotPromptTemplate from the samples of the vectorstore.
        """

        if self._selector is None:
            raise RuntimeError(
                f"`{self.__class__.__name__}` object has no vector store "
                f"created when executing `run()` method. You should call "
                f"method `load_example_selector()` `train()` befoto create "
                f"a vector store before."
            )

        template = "{__INPUT__}\n{__OUTPUT__}"

        if self._language == "es":
            template = "EJEMPLO: {__INPUT__}\nRESPUESTA: {__OUTPUT__}"

        if self._language == "en":
            template = "INPUT: {__INPUT__}\nOUTPUT: {__OUTPUT__}"

        example_prompt = PromptTemplate(
            input_variables=["__INPUT__", "__OUTPUT__"],
            template=template,
        )

        return FewShotPromptTemplate(
            example_selector=self._selector,
            example_prompt=example_prompt,
            suffix=template.format(__INPUT__="{__INPUT__}", __OUTPUT__=""),
            input_variables=["__INPUT__"],
        )


class BaseSelectorUnsupervised(BaseSelector):
    def train(
        self,
        examples: List[str],
        annotations: Optional[List[str]],
    ) -> Self:
        """
        Creates the vectorstor with the training samples.
        """

        if annotations is not None:
            raise ValueError(
                f"{self.__class__.__name__} error in __init__. "
                f"`annotations` value has to be `None` for this "
                f"king of task."
            )

        examples = [{"__INPUT__": example} for example in examples]

        self._selector = self.selector.from_examples(
            examples=examples,
            embeddings=self._embeddings,
            vectorstore_cls=FAISS,
            k=1,
        )

        return self

    def run(
        self,
    ) -> FewShotPromptTemplate:
        """
        Creates the FewShotPromptTemplate from the samples of the vectorstore.
        """

        if self._selector is None:
            raise RuntimeError(
                f"`{self.__class__.__name__}` object has no vector store "
                f"created when executing `run()` method. You should call "
                f"method `load_example_selector()` `train()` befoto create "
                f"a vector store before."
            )

        example_prompt = PromptTemplate(
            input_variables=["__INPUT__"],
            template="\n{__INPUT__}\n",
        )

        return FewShotPromptTemplate(
            example_selector=self._selector,
            example_prompt=example_prompt,
            suffix="Question: {__INPUT__}",
            input_variables=["__INPUT__"],
        )

promptmeteo/selector/base.py
paradigmadigital/Promptmeteo
from .task import Task
from .task_builder import TaskTypes
from .task_builder import TaskBuilder

promptmeteo/tasks/__init__.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from langchain.prompts.prompt import PromptTemplate
from langchain.prompts.pipeline import PipelinePromptTemplate

from ..models import BaseModel
from ..prompts import BasePrompt
from ..parsers import BaseParser
from ..selector import BaseSelector


class Task:

    """
    Base Task interface.
    """

    def __init__(
        self,
        language: str,
        task_type: str,
        verbose: bool = False,
    ):
        self._model = None
        self._parser = None
        self._prompt = None
        self._selector = None
        self._verbose = verbose
        self._language = language
        self._task_type = task_type

    # Getters
    @property
    def language(
        self,
    ) -> str:
        """
        Get Task Language.
        """
        return self._language

    @property
    def task_type(
        self,
    ) -> str:
        """
        Get Task type.
        """
        return self._task_type

    @property
    def prompt(
        self,
    ) -> BasePrompt:
        """
        Get Task Prompt.
        """
        return self._prompt

    @property
    def model(
        self,
    ) -> BaseModel:
        """
        Get Task Model.
        """
        return self._model

    @property
    def selector(
        self,
    ) -> BaseSelector:
        """
        Get Task Selector.
        """
        return self._selector

    @property
    def parser(
        self,
    ) -> BaseParser:
        """
        Get Task Parser.
        """
        return self._parser

    # Setters
    @prompt.setter
    def prompt(
        self,
        prompt: BasePrompt,
    ) -> None:
        """
        Set Task Prompt.
        """
        self._prompt = prompt

    @model.setter
    def model(
        self,
        model: BaseModel,
    ) -> None:
        """
        Set Task Model.
        """
        self._model = model

    @selector.setter
    def selector(
        self,
        selector: BaseSelector,
    ) -> None:
        """
        Set Task Selector.
        """
        self._selector = selector

    @parser.setter
    def parser(
        self,
        parser: BaseParser,
    ) -> None:
        """
        Task Parser
        """
        self._parser = parser

    def _get_prompt(
        self,
        example: str,
    ) -> str:
        """
        Create a PipelinePromptTemplate by merging the PromptTemplate and the
        FewShotPromptTemplate.
        """

        intro_prompt = self.prompt.run()

        no_examples_prompt = PromptTemplate.from_template("{__INPUT__}")

        if self._language == "es":
            no_examples_prompt = PromptTemplate.from_template(
                "Texto de entrada: {__INPUT__}"
            )

        if self._language == "en":
            no_examples_prompt = PromptTemplate.from_template(
                "Input text: {__INPUT__}"
            )

        examples_prompt = (
            self.selector.run() if self.selector else no_examples_prompt
        )

        return PipelinePromptTemplate(
            final_prompt=PromptTemplate.from_template(
                """
                {__INSTRUCTION__}

                {__EXAMPLES__}
                """.replace(
                    " " * 4, ""
                )
                .replace("\n\n", "|")
                .replace("\n", " ")
                .replace("|", "\n\n")
            ),
            pipeline_prompts=[
                ("__INSTRUCTION__", intro_prompt),
                ("__EXAMPLES__", examples_prompt),
            ],
        ).format(__INPUT__=example)

    def run(
        self,
        example: str,
    ) -> str:
        """
        Given a text sample, return the text predicted by Promptmeteo.
        """

        prompt = self._get_prompt(example)
        output = self.model.run(prompt)
        result = self.parser.run(output)

        if self._verbose:
            print("\n\nPROMPT INPUT\n\n", prompt)
            print("\n\nMODEL OUTPUT\n\n", output)
            print("\n\nPARSE RESULT\n\n", result)

        return result

promptmeteo/tasks/task.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.

from enum import Enum
from typing import List
from typing import Dict
from typing import Optional

try:
    from typing import Self
except ImportError:
    from typing_extensions import Self

from .task import Task
from ..models import ModelFactory
from ..prompts import PromptFactory
from ..parsers import ParserFactory
from ..selector import SelectorFactory


class TaskTypes(str, Enum):

    """
    Enum with all the available task types
    """

    QA: str = "qa"
    CLASSIFICATION: str = "classification"
    CODE_GENERATION: str = "code-generation"


class TaskBuilder:

    """
    Builder of Tasks.
    """

    def __init__(
        self,
        language: str,
        task_type: str,
        verbose: bool = False,
    ) -> None:
        self._task = Task(
            language=language,
            task_type=task_type,
            verbose=verbose,
        )

    @property
    def task(
        self,
    ) -> Task:
        """Task to built."""
        return self._task

    def build_prompt(
        self,
        model_name: str,
        prompt_domain: str,
        prompt_labels: List[str],
        prompt_detail: str,
    ) -> Self:
        """
        Builds a prompt for the task.
        """

        self._task.prompt = PromptFactory.factory_method(
            language=self._task.language,
            task_type=self._task.task_type,
            model_name=model_name,
            prompt_domain=prompt_domain,
            prompt_labels=prompt_labels,
            prompt_detail=prompt_detail,
        )

        return self

    def build_selector_by_train(
        self,
        examples: List[str],
        annotations: List[str],
        selector_k: int,
        selector_type: str,
        selector_algorithm: str,
    ) -> Self:
        """
        Builds a the selector for the task by training a new selector.
        """

        if not self._task.model:
            raise RuntimeError(
                "Selector algorithm is trying yo be built but there is no"
                "LLM model or embeddings loaded. You need to call function"
                "`build_model()` before calling `build_selector()`."
            )

        if not self._task.model.embeddings:
            raise RuntimeError(
                "Selector algorithm is trying yo be built but there is no"
                "embeddigns for model {self._task._model.__name__}."
            )

        embeddings = self._task.model.embeddings

        self._task.selector = SelectorFactory.factory_method(
            language=self._task.language,
            embeddings=embeddings,
            selector_k=selector_k,
            selector_type=selector_type,
            selector_algorithm=selector_algorithm,
        ).train(
            examples=examples,
            annotations=annotations,
        )

        return self

    def build_model(
        self,
        model_name: str = "",
        model_provider_name: str = "",
        model_provider_token: Optional[str] = "",
        model_params: Dict = None,
    ) -> Self:
        """
        Builds a model for the task.
        """

        self._task.model = ModelFactory.factory_method(
            model_name=model_name,
            model_provider_name=model_provider_name,
            model_provider_token=model_provider_token,
            model_params=model_params or {},
        )

        return self

    def build_parser(
        self,
        prompt_labels: List[str],
    ) -> Self:
        """
        Builds a the parser for the task.
        """

        self._task.parser = ParserFactory.factory_method(
            task_type=self._task.task_type,
            prompt_labels=prompt_labels,
        )

        return self

    def build_selector_by_load(
        self,
        model_path: str,
        selector_k: int,
        selector_type: str,
        selector_algorithm: str,
    ) -> Self:
        """
        Builds the selector for the task by loading a pretrained selector.
        """

        if not self._task.model:
            raise RuntimeError(
                "Selector algorithm is trying to be built but there is no"
                "LLM model or embeddings loaded. You need to call function"
                "`build_model()` before calling `load_selector()`."
            )

        if not self._task.model.embeddings:
            raise RuntimeError(
                "Selector algorithm is trying to be built but there is no"
                "embeddigns for model {self._task._model.__name__}."
            )

        embeddings = self._task.model.embeddings

        self._task.selector = SelectorFactory.factory_method(
            language=self._task.language,
            embeddings=embeddings,
            selector_k=selector_k,
            selector_type=selector_type,
            selector_algorithm=selector_algorithm,
        ).load_example_selector(model_path=model_path)

        return self

promptmeteo/tasks/task_builder.py
paradigmadigital/Promptmeteo
#!/usr/bin/python3

#  Copyright (c) 2023 Paradigma Digital S.L.

#  Permission is hereby granted, free of charge, to any person obtaining a copy
#  of this software and associated documentation files (the "Software"), to deal
#  in the Software without restriction, including without limitation the rights
#  to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
#  copies of the Software, and to permit persons to whom the Software is
#  furnished to do so, subject to the following conditions:

#  The above copyright notice and this permission notice shall be included in
#  all copies or substantial portions of the Software.

#  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
#  IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
#  FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
#  AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
#  LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
#  OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
#  THE SOFTWARE.


def add_docstring_from(parent_function):
    """
    Decorator function used to concatenate a docstring from another function
    at the beginning.

    Example
    -------

    >>> def foo():
    >>>     '''documentation for foo'''
    >>>     pass

    >>> @add_docstring_from(foo)
    >>> def bar():
    >>>     '''additional notes for bar'''
    >>>     pass

    >>> print(bar.__doc__)
    documentation for foo

    additional notes for bar
    """

    def decorator(inherit_function):
        parent_docstring = (
            parent_function.__doc__ if parent_function.__doc__ else ""
        )

        inherit_docstring = (
            inherit_function.__doc__ if inherit_function.__doc__ else ""
        )

        inherit_function.__doc__ = parent_docstring + "\n" + inherit_docstring

        return inherit_function

    return decorator

promptmeteo/tools.py
paradigmadigital/Promptmeteo
import os
import tempfile

import pytest

from promptmeteo.tasks import Task
from promptmeteo.tasks import TaskBuilder
from promptmeteo.base import BaseSupervised


class TestBaseSupervised:
    def test_minimal_init(self):
        """
        Tests the exepected behaviour in a normal init.
        """

        model = BaseSupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
        )

    def test_init_with_arguments(self):
        """
        Tests the exepected behaviour in a normal init.
        """

        model = BaseSupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            model_provider_token="",
            prompt_domain="reseñas",
            prompt_labels=["positive", "negative", "neutral"],
            prompt_detail="""
            Asume que estamos usando estas categorías con su significado
            subjetivo habitual: algo positivo puede describirse como bueno,
            de buena calidad, deseable, útil y satisfactorio; algo negativo
            puede describirse como malo, de mala calidad, indeseable, inútil
            o insatisfactorio; neutro es la clase que asignaremos a todo lo
            que no sea positivo o negativo.""",
            selector_k=10,
            selector_algorithm="mmr",
            verbose=True,
        )

    def test_wrong_predict(self):
        """
        Test that the model perform predictions by calling the method
        `predict()`
        """

        with pytest.raises(ValueError):
            pred = BaseSupervised(
                language="es",
                model_provider_name="fake-llm",
                model_name="fake-static",
            ).predict("Wrong type this is expected to be a list")

            assert error.value.args[0] == (
                f"BaseSupervised error in function `predict()`. "
                f"Arguments `examples` and `annotations` are expected to be of "
                f"type `List[str]`. Instead they got: `{type(examples)}`"
            )

        with pytest.raises(ValueError):
            pred = BaseSupervised(
                language="es",
                model_provider_name="fake-llm",
                model_name="fake-static",
            ).predict([1, 2, 3])

            assert error.value.args[0] == (
                f"BaseSupervised error in function `predict()`. "
                f"Arguments `examples` are expected to be of type "
                f"`List[str]`. Some values seem no to be of type `str`."
            )

    def test_wrong_train(self):
        """
        Test that the task property form BaseSupervised is trained after
        calling the method `fit()`
        """

        model = BaseSupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            prompt_labels=["1", "0"],
        )

        with pytest.raises(ValueError) as error:
            model.train(
                examples="Wrong type this is expected to be a list",
                annotations="Wrong type this is expected to be a list",
            )

            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` are expected to be of type `List[str]`."
                f"Instead they got: examples {type(examples)}"
            )

        with pytest.raises(ValueError) as error:
            model.train(
                examples=["wrong element type: int", 1],
                annotations=["wrong element type: boo", True],
            )

            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` are expected to be of type `List[str]`."
                f"Some values seem no to be of type `str`."
            )

        with pytest.raises(ValueError) as error:
            model.train(
                examples=["list", "with", "four", "items"],
                annotations=["list", "with", "three"],
            )

            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` and `annotations` are expected to have "
                f"the same length. examples=({len(examples)},) annotations= "
                f"({len(annotations)},)"
            )

        with pytest.raises(ValueError) as error:
            model.train(
                examples=["text", "text", "text"],
                annotations=["1", "0", "WRONG_LABEL"],
            )

            assert error.value.args[0] == (
                f"{self.__class__.__name__} error in `train()`. "
                f"`annotation value in item {idx}: `{annotation}`"
                f"is not in the expected values: {self.prompt_labels}"
            )

    def test_load_model(self):
        model = BaseSupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        )

        # Use wrong file extension
        with pytest.raises(ValueError) as error:
            model_path = "model.WRONG_EXTENSION"
            model.load_model(model_path)
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `load_model()`. "
                f'model_path="{model_path}" has a bad model name extension. '
                f"Model name must end with `.meteo` (i.e. `./model.meteo`)"
            )

        # Use non existing file path
        with pytest.raises(ValueError) as error:
            model_dir = "WRONG_DIR_PATH"
            model.load_model(os.path.join(model_dir, "model.meteo"))
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `load_model()`. "
                f"directory {model_dir} does not exists."
            )

    def test_save_model(self):
        model = BaseSupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        )

        # Fake train
        model._is_trained = True

        # Try to save before training
        with pytest.raises(Exception) as error:
            model.save_model("model.meteo")
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `save_model()`. "
                f"You are trying to save a model that has non been trained. "
                f"Please, call `train()` function before"
            )

        # Use wrong file extension
        with pytest.raises(ValueError) as error:
            model_path = "model.WRONG_EXTENSION"
            model.save_model(model_path)
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `save_model()`. "
                f'model_path="{model_path}" has a bad model name extension. '
                f"Model name must end with `.meteo` (i.e. `./model.meteo`)"
            )

        # Use non existing file path
        with pytest.raises(ValueError) as error:
            model_dir = "WRONG_DIR_PATH"
            model.save_model(os.path.join(model_dir, "model.meteo"))
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `save_model()`. "
                f"directory {model_dir} does not exists."
            )

tests/test_base_supervised.py
paradigmadigital/Promptmeteo
import os
import tempfile

import pytest

from promptmeteo.tasks import Task
from promptmeteo.tasks import TaskBuilder
from promptmeteo.base import BaseUnsupervised


class TestBaseUnsupervised:
    def test_minimal_init(self):
        """
        Tests the exepected behaviour in a normal init.
        """

        model = BaseUnsupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
        )

    def test_init_with_arguments(self):
        """
        Tests the exepected behaviour in a normal init.
        """

        model = BaseUnsupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            model_provider_token="",
            prompt_domain="reseñas",
            prompt_detail="""
            Asume que estamos usando estas categorías con su significado
            subjetivo habitual: algo positivo puede describirse como bueno,
            de buena calidad, deseable, útil y satisfactorio; algo negativo
            puede describirse como malo, de mala calidad, indeseable, inútil
            o insatisfactorio; neutro es la clase que asignaremos a todo lo
            que no sea positivo o negativo.""",
            selector_k=10,
            selector_algorithm="mmr",
            verbose=True,
        )

        with pytest.raises(ValueError) as error:
            model = BaseUnsupervised(
                language="es",
                model_provider_name="fake-llm",
                model_name="fake-static",
                prompt_labels=["positive", "negative", "neutral"],
            )

            assert error.value.args[0] == (
                f"BaseUnsupervised can not be inicializated with "
                f"the argument `prompt_labels`."
            )

    def test_wrong_predict(self):
        """
        Test that the model perform predictions by calling the method
        `predict()`
        """

        with pytest.raises(ValueError) as error:
            pred = BaseUnsupervised(
                language="es",
                model_provider_name="fake-llm",
                model_name="fake-static",
            ).predict("Wrong type this is expected to be a list")

            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in function `predict()`. "
                f"Arguments `examples` and `annotations` are expected to be of "
                f"type `List[str]`. Instead they got: `{type(examples)}`"
            )

        with pytest.raises(ValueError) as error:
            pred = BaseUnsupervised(
                language="es",
                model_provider_name="fake-llm",
                model_name="fake-static",
            ).predict([1, 2, 3])

            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in function `predict()`. "
                f"Arguments `examples` are expected to be of type "
                f"`List[str]`. Some values seem no to be of type `str`."
            )

    def test_wrong_train(self):
        """
        Test that the task property form BaseUnsupervised is trained after
        calling the method `fit()`
        """

        model = BaseUnsupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
        )

        with pytest.raises(ValueError) as error:
            model.train(
                examples="Wrong type this is expected to be a list",
            )

            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` are expected to be of type `List[str]`."
                f"Instead they got: examples {type(examples)}"
            )

        with pytest.raises(ValueError) as error:
            model.train(
                examples=["wrong element type: int", 1],
            )

            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in function `train()`. "
                f"Arguments `examples` are expected to be of type `List[str]`."
                f"Some values seem no to be of type `str`."
            )

        with pytest.raises(TypeError) as error:
            model.train(
                examples=["1", "2", "3"],
                annotations=["one", "two", "three"],
            )

            assert error.value.args[0] == (
                f"TypeError: BaseUnsupervised.train() got an unexpected "
                f"keyword argument 'annotations'"
            )

    def test_load_model(self):
        model = BaseUnsupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        )

        # Use wrong file extension
        with pytest.raises(ValueError) as error:
            model_path = "model.WRONG_EXTENSION"
            model.load_model(model_path)
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `load_model()`. "
                f'model_path="{model_path}" has a bad model name extension. '
                f"Model name must end with `.meteo` (i.e. `./model.meteo`)"
            )

        # Use non existing file path
        with pytest.raises(ValueError) as error:
            model_dir = "WRONG_DIR_PATH"
            model.load_model(os.path.join(model_dir, "model.meteo"))
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `load_model()`. "
                f"directory {model_dir} does not exists."
            )

    def test_save_model(self):
        model = BaseUnsupervised(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        )

        # Fake train
        model._is_trained = True

        # Try to save before training
        with pytest.raises(Exception) as error:
            model.save_model("model.meteo")
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `save_model()`. "
                f"You are trying to save a model that has non been trained. "
                f"Please, call `train()` function before"
            )

        # Use wrong file extension
        with pytest.raises(ValueError) as error:
            model_path = "model.WRONG_EXTENSION"
            model.save_model(model_path)
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `save_model()`. "
                f'model_path="{model_path}" has a bad model name extension. '
                f"Model name must end with `.meteo` (i.e. `./model.meteo`)"
            )

        # Use non existing file path
        with pytest.raises(ValueError) as error:
            model_dir = "WRONG_DIR_PATH"
            model.save_model(os.path.join(model_dir, "model.meteo"))
            assert error.value.args[0] == (
                f"{model.__class__.__name__} error in `save_model()`. "
                f"directory {model_dir} does not exists."
            )

tests/test_base_unsupervised.py
paradigmadigital/Promptmeteo
import os
import tempfile

import pytest

from promptmeteo import CodeGenerator


class TestCodeGenerator:
    def test_correct_train(self):
        """
        Test that the task property form CodeGenerator is trained
        after calling the method `fit()`
        """

        model = CodeGenerator(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            prompt_detail=["add docstring in function definitions"],
        )

        model = model.train(
            examples=["Function with the argument `foo` that prints it"],
            annotations=[
                '''
                def print_foo(foo: str):
                """This function prints the value passed in as an argument.

                Args:        foo (str): The value to be printed.
                """
                print(foo)'''.replace(
                    " " * 16, ""
                )
            ],
        )

        assert model
        assert model.is_trained

    def test_predict_with_train(self):
        """
        Test that the model perform predictions by calling the method
        `predict()`
        """

        model = CodeGenerator(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            prompt_detail=["add docstring in function definitions"],
        )

        model = model.train(
            examples=["Function with the argument `foo` that prints it"],
            annotations=[
                '''
                def print_foo(foo: str):
                """This function prints the value passed in as an argument.

                Args:        foo (str): The value to be printed.
                """
                print(foo)'''.replace(
                    " " * 16, ""
                )
            ],
        )

        pred = model.predict(["positive"])

        assert pred == [["positive"]]

    def test_save_model(self):
        model = CodeGenerator(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        )

        model.train(
            examples=["Function with the argument `foo` that prints it"],
            annotations=[
                '''
                def print_foo(foo: str):
                """This function prints the value passed in as an argument.

                Args:        foo (str): The value to be printed.
                """
                print(foo)'''.replace(
                    " " * 16, ""
                )
            ],
        )

        with tempfile.TemporaryDirectory() as tmp:
            model.save_model(os.path.join(tmp, "model.meteo"))
            assert os.path.exists(os.path.join(tmp, "model.meteo"))

    def test_load_model(self):
        model = CodeGenerator(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        ).train(
            examples=["Function with the argument `foo` that prints it"],
            annotations=[
                '''
                def print_foo(foo: str):
                """This function prints the value passed in as an argument.

                Args:        foo (str): The value to be printed.
                """
                print(foo)'''.replace(
                    " " * 16, ""
                )
            ],
        )

        with tempfile.TemporaryDirectory() as tmp:
            model.save_model(os.path.join(tmp, "model.meteo"))
            model.load_model(os.path.join(tmp, "model.meteo"))

tests/test_code_generator.py
paradigmadigital/Promptmeteo
import os
import tarfile
import tempfile

import pytest

from promptmeteo import DocumentClassifier


class TestDocumentClassifier:
    def test_correct_train(self):
        """
        Test that the task property form DocumentClassifier is trained
        after calling the method `fit()`
        """

        model = DocumentClassifier(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
        )

        model = model.train(
            examples=["estoy feliz", "me da igual", "no me gusta"],
            annotations=["positive", "neutral", "negative"],
        )

        assert model
        assert model.is_trained
        assert model.prompt_labels == list(
            set(["positive", "neutral", "negative"])
        )

    def test_predict_with_train(self):
        """
        Test that the model perform predictions by calling the method
        `predict()`
        """

        model = DocumentClassifier(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
        )

        model = model.train(
            examples=["estoy feliz", "me da igual", "no me gusta"],
            annotations=["positive", "neutral", "negative"],
        )

        pred = model.predict(["positive"])

        assert pred == [["positive"]]

    def test_save_model(self):
        model = DocumentClassifier(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        )

        model.train(
            examples=["estoy feliz", "me da igual", "no me gusta"],
            annotations=["positivo", "neutral", "negativo"],
        )

        with tempfile.TemporaryDirectory() as tmp:
            model.save_model(os.path.join(tmp, "model.meteo"))
            assert os.path.exists(os.path.join(tmp, "model.meteo"))
            tar = tarfile.open(os.path.join(tmp, "model.meteo"), "r:gz")
            items = [t.path for t in list(tar)]
            assert 'model.meteo' in items
            assert 'model.init' in items

    def test_load_model(self):
        model = DocumentClassifier(
            language="es",
            model_provider_name="fake-llm",
            model_name="fake-static",
            verbose=True,
        ).train(
            examples=["estoy feliz", "me da igual", "no me gusta"],
            annotations=["positivo", "neutral", "negativo"],
        )

        with tempfile.TemporaryDirectory() as tmp:
            model.save_model(os.path.join(tmp, "model.meteo"))
            load_model = DocumentClassifier.load_model(os.path.join(tmp, "model.meteo"))

            assert load_model.language == model.language
            assert load_model.model_provider_name == model.model_provider_name
            assert load_model.model_name == model.model_name
            assert load_model.verbose == model.verbose

tests/test_document_classifier.py
paradigmadigital/Promptmeteo
from unittest.mock import MagicMock

import pytest

from promptmeteo.models import ModelFactory
from promptmeteo.models import ModelProvider


class TestModels:
    def test_model_factory(self):
        invalid_provider = (
            "ModelFactory error in `factory_method()`. "
            "WRONG_PROVIDER_NAME is not in the list of supported "
            f"providers: {[i.value for i in ModelProvider]}"
        )
        for model_provider_name in ModelProvider:
            with pytest.raises(ValueError) as error:
                ModelFactory.factory_method(
                    model_name="TEST_NAME",
                    model_provider_name=model_provider_name.value,
                    model_provider_token="TEST_TOKEN",
                    model_params={"TEST_PARAMS": True},
                )

            assert error.value.args[0] != invalid_provider

        with pytest.raises(ValueError) as error:
            ModelFactory.factory_method(
                model_name="TEST_NAME",
                model_provider_name="WRONG_PROVIDER_NAME",
                model_provider_token="TEST_TOKEN",
                model_params={"TEST_PARAMS": True},
            )

        assert error.value.args[0] == invalid_provider

    def test_model_openai(self):
        from promptmeteo.models.openai import ModelTypes
        from promptmeteo.models.openai import OpenAILLM

        for model_name in ModelTypes:
            OpenAILLM(
                model_name=model_name.value,
                model_params={},
                model_provider_token="TEST_TOKEN",
            )

        with pytest.raises(ValueError) as error:
            OpenAILLM(
                model_name="WRONG_NAME",
                model_params={},
                model_provider_token="TEST_TOKEN",
            )
        invalid_provider = (
            "`model_name`=WRONG_NAME not in supported model names: "
            f"{[i.name for i in ModelTypes]}"
        )
        assert error.value.args[0] == invalid_provider

    def test_model_fakellm(self):
        from promptmeteo.models.fake_llm import ModelTypes
        from promptmeteo.models.fake_llm import FakeLLM

        for model_name in ModelTypes:
            FakeLLM(
                model_name=model_name.value,
                model_params={},
                model_provider_token="TEST_TOKEN",
            )

        with pytest.raises(ValueError) as error:
            FakeLLM(
                model_name="WRONG_NAME",
                model_params={},
                model_provider_token="TEST_TOKEN",
            )
        invalid_provider = (
            "FakeLLM error creating object. "
            "WRONG_NAME is not in the list of supported FakeLLMS: "
            f"{[i.value for i in ModelTypes]}"
        )
        assert error.value.args[0] == invalid_provider

    def test_model_hf_hub_api(self, mocker):
        from promptmeteo.models.hf_hub_api import ModelTypes
        from promptmeteo.models.hf_hub_api import HFHubApiLLM

        mock = MagicMock()
        mocker.patch(
            "huggingface_hub.inference_api.InferenceApi.__call__", mock
        )

        for model_name in ModelTypes:
            HFHubApiLLM(
                model_name=model_name.value,
                model_params={},
                model_provider_token="TEST_TOKEN",
            )

        with pytest.raises(ValueError) as error:
            HFHubApiLLM(
                model_name="WRONG_NAME",
                model_params={},
                model_provider_token="TEST_TOKEN",
            )

        invalid_provider = (
            "`model_name`=WRONG_NAME not in supported model names: "
            f"{[i.value for i in ModelTypes]}"
        )
        assert error.value.args[0] == invalid_provider

    def test_model_hf_pipeline(self):
        from promptmeteo.models.hf_pipeline import ModelTypes
        from promptmeteo.models.hf_pipeline import HFPipelineLLM

        with pytest.raises(ValueError) as error:
            HFPipelineLLM(
                model_name="WRONG_NAME",
                model_params={},
                model_provider_token="TEST_TOKEN",
            )

        invalid_provider = (
            "`model_name`=WRONG_NAME not in supported model names: "
            f"{[i.value for i in ModelTypes]}"
        )
        assert error.value.args[0] == invalid_provider

tests/test_models.py
paradigmadigital/Promptmeteo
import pytest

from promptmeteo.parsers import ParserTypes
from promptmeteo.parsers import ParserFactory
from promptmeteo.parsers.dummy_parser import DummyParser
from promptmeteo.parsers.classification_parser import ClassificationParser


class Testparsers:
    def test_parser_factory(self):
        for parser in ParserTypes:
            ParserFactory.factory_method(
                task_type=parser.value,
                prompt_labels=["true", "false"],
            )

        with pytest.raises(ValueError):
            ParserFactory.factory_method(
                task_type="WRONG PARSER TYPE",
                prompt_labels=["true", "false"],
            )

    def test_dummy_parser(self):
        parser = DummyParser(
            prompt_labels=["true", "false"],
        )

        assert ["blabla"] == parser.run("blabla")

    def test_classification_parser(self):
        parser = ClassificationParser(
            prompt_labels=["true", "false"],
        )

        assert [""] == parser.run("blabla")
        assert ["true"] == parser.run("true")
        assert ["true"] == parser.run("True")
        assert ["true"] == parser.run("blabla, true, blabla")
        assert ["true", "false"] == parser.run("true, false")

tests/test_parsers.py
paradigmadigital/Promptmeteo
import os
import yaml
import pytest
from tests.tools import DictionaryChecker

from promptmeteo.prompts import BasePrompt
from promptmeteo.prompts import PromptFactory
from promptmeteo.prompts import module_dir


class TestPrompts:
    def test_prompt_factory(self):
        for model_name, language, task_type in [
            ["fake-static", "es", "ner"],
            ["fake-static", "es", "classification"],
        ]:
            prompt = PromptFactory.factory_method(
                language=language,
                task_type=task_type,
                model_name=model_name,
                prompt_domain="TEST DOMAIN",
                prompt_labels=["true", "false"],
                prompt_detail="TEST PROMPT DETAIL",
            )

    def test_prompt_factory_exception(self):
        with pytest.raises(ValueError):
            PromptFactory.factory_method(
                language="WRONG LANGUAGE",  # WRONG
                task_type="ner",
                model_name="fake-static",
                prompt_domain="TEST DOMAIN",
                prompt_labels=["true", "false"],
                prompt_detail="TEST PROMPT DETAIL",
            )

        with pytest.raises(ValueError):
            PromptFactory.factory_method(
                language="es",
                task_type="WRONG TASK TYPE",  # WRONG
                model_name="fake-static",
                prompt_domain="TEST DOMAIN",
                prompt_labels=["true", "false"],
                prompt_detail="TEST PROMPT DETAIL",
            )

        with pytest.raises(ValueError):
            PromptFactory.factory_method(
                language="es",
                task_type="ner",
                model_name="WRONG MODEL NAME",  # WRONG
                prompt_domain="TEST DOMAIN",
                prompt_labels=["true", "false"],
                prompt_detail="TEST PROMPT DETAIL",
            )

    def test_base_prompt(self):
        """
        Test BasePrompt behavior.
        """

        BasePrompt.read_prompt(BasePrompt.PROMPT_EXAMPLE)

        BasePrompt(
            prompt_domain="TEST DOMAIN",
            prompt_labels=["true", "false"],
            prompt_detail="TEST PROMPT DETAIL",
        )

    def test_base_prompt_exception(self):
        """
        Tests load prompt text format.
        """

        with pytest.raises(ValueError) as error:
            BasePrompt.read_prompt(
                """
                WRONG_TEMPLATE_KEY:
                    wrong template key

                ANOTHER_WRONG_TEMPLATE_KEY:
                    wrong template key again
                """
            )

        assert error.value.args[0] == (
            f"`BasePrompt` error `read_prompt()`. "
            f"The expected keys are {yaml.load(BasePrompt.PROMPT_EXAMPLE, Loader=yaml.FullLoader)}"
        )

    def test_prompts_naming(self):
        """
        Test that the hierrachycal structure of the prompts is correct.
        """

        prompt_files = [
            file_name
            for file_name in os.listdir(module_dir)
            if file_name.endswith(".prompt")
        ]

        assert all([len(file.split("_")) == 3 for file in prompt_files])

    def test_prompts_format(self):
        """
        Test that all the prompts from the project follow the expected
        structure.
        """

        params = [
            file_name.replace(".prompt", "").split("_")
            for file_name in os.listdir(module_dir)
            if file_name.endswith(".prompt")
        ]

        for model_name, language, task_type in params:
            prompt = PromptFactory.factory_method(
                language=language,
                task_type=task_type,
                model_name=model_name,
                prompt_domain="",
                prompt_labels=["0", "1"],
                prompt_detail="",
            )

    def test_prompts_spelling(self):
        """
        Test that all the prompts are written correctly.
        """

        params = [
            file_name.replace(".prompt", "").split("_")
            for file_name in os.listdir(module_dir)
            if file_name.endswith(".prompt")
        ]

        for model_name, language, task_type in params:
            text = PromptFactory.factory_method(
                language=language,
                task_type=task_type,
                model_name=model_name,
                prompt_domain="",
                prompt_labels=["0", "1"],
                prompt_detail="",
            ).template

            check = DictionaryChecker(language)

            for symbol in """!()-[]{};:'"\\,<>./?@#$%^&*_~""":
                text = text.replace(symbol, " ")

            for word in text.split():
                if not check(word):
                    raise Exception(
                        f"Error in {model_name}_{language}_{task_type}.prompt "
                        f"the word `{word}` is not in the dictionary"
                    )

tests/test_prompts.py
paradigmadigital/Promptmeteo
import pytest

from langchain.embeddings import FakeEmbeddings

from promptmeteo.selector import SelectorTypes
from promptmeteo.selector import SelectorFactory
from promptmeteo.selector.base import SelectorAlgorithms
from promptmeteo.selector.base import BaseSelectorSupervised
from promptmeteo.selector.base import BaseSelectorUnsupervised


class TestSelectors:
    def test_selector_factory(self):
        for selector_type in SelectorTypes:
            for selector_algorithm in SelectorAlgorithms:
                SelectorFactory.factory_method(
                    language="es",
                    selector_k=1,
                    selector_type=selector_type.value,
                    selector_algorithm=selector_algorithm.value,
                    embeddings=FakeEmbeddings(size=64),
                )

        with pytest.raises(ValueError) as error:
            SelectorFactory.factory_method(
                language="es",
                selector_k=1,
                selector_type="WRONG_SELECTOR_TYPE",
                selector_algorithm=selector_algorithm.value,
                embeddings=FakeEmbeddings(size=64),
            )

        assert error.value.args[0] == (
            f"`SelectorFactory` error in `factory_method()` . "
            f"WRONG_SELECTOR_TYPE is not in the list of supported "
            f"providers: {[i.value for i in SelectorTypes]}"
        )

        with pytest.raises(ValueError) as error:
            selector = SelectorFactory.factory_method(
                language="es",
                selector_k=1,
                selector_type=selector_type.value,
                selector_algorithm="WRONG_ALGORITHM_NAME",
                embeddings=FakeEmbeddings(size=64),
            )

            assert error.value.args[0] == (
                f"{selector.__class__.__name__} error in __init__. "
                f"`selector_algorithm` value `WRONG_ALGORITHM_NAME` is not in "
                f"the available values: {[i.value for i in SelectorAlgorithms]}"
            )

    def test_supervised_selector(self):
        for selector_algorithm in SelectorAlgorithms:
            selector = BaseSelectorSupervised(
                language="es",
                embeddings=FakeEmbeddings(size=64),
                selector_k=1,
                selector_algorithm=selector_algorithm,
            )

            assert selector.selector is not None

            selector.train(
                examples=["TEST_EXAMPLE"], annotations=["TEST_ANNOTATION"]
            )

            assert (
                selector.template
                == """
                EJEMPLO: TEST_EXAMPLE
                RESPUESTA: TEST_ANNOTATION

                EJEMPLO: {__INPUT__}
                RESPUESTA: """.replace(
                    " " * 4, ""
                )[
                    1:
                ]
            )

tests/test_selectors.py
paradigmadigital/Promptmeteo
import pytest

from promptmeteo.models import BaseModel
from promptmeteo.tasks import TaskBuilder
from promptmeteo.selector import SelectorTypes
from promptmeteo.selector.base import SelectorAlgorithms

task_types = ["classification", "qa"]


class TestTaskBuilder:
    def test_build_model(self):
        for task_type in task_types:
            task_builder = TaskBuilder(
                language="es",
                task_type=task_type,
            ).build_model(
                model_name="fake-static",
                model_provider_name="fake-llm",
                model_provider_token="",
            )

            assert task_builder.task.model is not None
            assert isinstance(task_builder.task.model, BaseModel)

    def test_build_prompt(self):
        for task_type in task_types:
            task_builder = TaskBuilder(
                language="es",
                task_type=task_type,
            ).build_prompt(
                model_name="fake-static",
                prompt_domain="TEST_DOMAIN",
                prompt_labels=["LABELS_1", "LABEL_2"],
                prompt_detail="TEST_DETAIL",
            )

            assert task_builder.task.prompt is not None

    def test_selector_prompt(self):
        for task_type in task_types:
            with pytest.raises(Exception):
                TaskBuilder(
                    language="es",
                    task_type=task_type,
                ).build_selector_by_train(
                    examples=["estoy feliz", "me da igual", "no me gusta"],
                    annotations=["positive", "neutral", "negative"],
                    selector_k=10,
                    selector_type="supervised",
                    selector_algorithm="mmr",
                )

        for task_type in task_types:
            for selector_algorithm in SelectorAlgorithms:
                task_builder = TaskBuilder(
                    language="es",
                    task_type=task_type,
                ).build_model(
                    model_name="fake-static",
                    model_provider_name="fake-llm",
                    model_provider_token="",
                )

                # SUPERVISED TASK
                selector_type = SelectorTypes.SUPERVISED.value
                task_builder.build_selector_by_train(
                    examples=["text1", "text2", "text3"],
                    annotations=["1", "0", "1"],
                    selector_k=10,
                    selector_type=selector_type,
                    selector_algorithm=selector_algorithm.value,
                )

                # UNSUPERVISED TASK
                selector_type = SelectorTypes.UNSUPERVISED.value
                task_builder.build_selector_by_train(
                    examples=["text1", "text2", "text3"],
                    annotations=None,
                    selector_k=10,
                    selector_type=selector_type,
                    selector_algorithm=selector_algorithm.value,
                )

            assert task_builder.task.selector is not None

tests/test_task.py
paradigmadigital/Promptmeteo
from .dictionary_checker import DictionaryChecker

tests/tools/__init__.py
paradigmadigital/Promptmeteo
import os
import json
import gzip


module_dir = os.path.abspath(os.path.join(__file__, os.path.pardir))


class DictionaryChecker:
    def __init__(self, language: str):
        try:
            file_name = f"{language}.json.gz"
            with gzip.open(os.path.join(module_dir, file_name)) as fin:
                self.lexicon = json.loads(fin.read())

        except FileNotFoundError:
            raise NotImplementedError(
                f"{self.__class__.__name__} error. Language `{language}`"
                f"is not implemented. Availables languages: sp, en"
            )

    def __call__(self, word: str):
        return word.lower() in self.lexicon or word.isnumeric()

tests/tools/dictionary_checker.py
paradigmadigital/Promptmeteo