Skip to content

base

AsyncAPI

Bases: ABC

Source code in src/prompto/apis/base.py 
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
class AsyncAPI(ABC):
    def __init__(self, settings: Settings, log_file: str, *args: Any, **kwargs: Any):
        """
        Base class for asynchronous API models.

        Each subclass should implement the following methods:
        - check_environment_variables: a static method that checks
          if the required or optional environment variables are set
        - check_prompt_dict: a static method that checks if an input
          dictionary (prompt_dict) is valid
        - query: an async method that queries the API/model and
          returns the response as a completed dictionary (prompt_dict)

        Parameters
        ----------
        settings : Settings
            The settings for the pipeline/experiment
        log_file : str
            The path to the log file
        """
        self.settings = settings
        self.log_file = log_file

    @staticmethod
    def check_environment_variables() -> list[Exception]:
        """
        Method for checking the environment variables.
        Each subclass should implement this method to check if the
        required or optional environment variables are set.

        Returns
        -------
        list[Exception]
            A list of exceptions or warnings if the environment variables
            are not set

        Raises
        ------
        NotImplementedError
            If the method is not implemented by a subclass
        """
        raise NotImplementedError(
            "'check_environment_variables' method needs to be implemented by a subclass of AsyncAPI"
        )

    @staticmethod
    def check_prompt_dict(prompt_dict: dict) -> list[Exception]:
        """
        Method for checking the prompt dictionary.
        Each subclass should implement this method to check if the
        prompt dictionary is a valid input for the model.

        Parameters
        ----------
        prompt_dict : dict
            The prompt dictionary to check

        Returns
        -------
        list[Exception]
            A list of exceptions or warnings if the prompt dictionary
            is not valid

        Raises
        ------
        NotImplementedError
            If the method is not implemented by a subclass
        """
        raise NotImplementedError(
            "'check_prompt_dict' method needs to be implemented by a subclass of AsyncAPI"
        )

    async def query(
        self, prompt_dict: dict, index: int | str = "NA", *args: Any, **kwargs: Any
    ) -> dict:
        """
        Method for querying the API/model asynchronously.

        Parameters
        ----------
        prompt_dict : dict
            The prompt dictionary to use for querying the model
        index : int | str, optional
            The index of the prompt in the experiment, by default "NA"

        Returns
        -------
        dict
            Completed prompt_dict with "response" key storing the response(s)
            from the LLM

        Raises
        ------
        NotImplementedError
            If the method is not implemented by a subclass
        """
        # async method for querying the model using the prompt and other arguments
        # returns a dictionary/json object which is saved into the output jsonl
        raise NotImplementedError(
            "'query' method needs to be implemented by a subclass of AsyncAPI"
        )

__init__ 

__init__(settings: Settings, log_file: str, *args: Any, **kwargs: Any)

Base class for asynchronous API models.

Each subclass should implement the following methods: - check_environment_variables: a static method that checks if the required or optional environment variables are set - check_prompt_dict: a static method that checks if an input dictionary (prompt_dict) is valid - query: an async method that queries the API/model and returns the response as a completed dictionary (prompt_dict)

Parameters:

Name Type Description Default
settings Settings

The settings for the pipeline/experiment

 required
log_file str

The path to the log file

 required
Source code in src/prompto/apis/base.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def __init__(self, settings: Settings, log_file: str, *args: Any, **kwargs: Any):
    """
    Base class for asynchronous API models.

    Each subclass should implement the following methods:
    - check_environment_variables: a static method that checks
      if the required or optional environment variables are set
    - check_prompt_dict: a static method that checks if an input
      dictionary (prompt_dict) is valid
    - query: an async method that queries the API/model and
      returns the response as a completed dictionary (prompt_dict)

    Parameters
    ----------
    settings : Settings
        The settings for the pipeline/experiment
    log_file : str
        The path to the log file
    """
    self.settings = settings
    self.log_file = log_file

check_environment_variables staticmethod 

check_environment_variables() -> list[Exception]

Method for checking the environment variables. Each subclass should implement this method to check if the required or optional environment variables are set.

Returns:

Type Description
list[Exception]

A list of exceptions or warnings if the environment variables are not set

Raises:

Type Description
NotImplementedError

If the method is not implemented by a subclass
Source code in src/prompto/apis/base.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
@staticmethod
def check_environment_variables() -> list[Exception]:
    """
    Method for checking the environment variables.
    Each subclass should implement this method to check if the
    required or optional environment variables are set.

    Returns
    -------
    list[Exception]
        A list of exceptions or warnings if the environment variables
        are not set

    Raises
    ------
    NotImplementedError
        If the method is not implemented by a subclass
    """
    raise NotImplementedError(
        "'check_environment_variables' method needs to be implemented by a subclass of AsyncAPI"
    )

check_prompt_dict staticmethod 

check_prompt_dict(prompt_dict: dict) -> list[Exception]

Method for checking the prompt dictionary. Each subclass should implement this method to check if the prompt dictionary is a valid input for the model.

Parameters:

Name Type Description Default
prompt_dict dict

The prompt dictionary to check

 required

Returns:

Type Description
list[Exception]

A list of exceptions or warnings if the prompt dictionary is not valid

Raises:

Type Description
NotImplementedError

If the method is not implemented by a subclass
Source code in src/prompto/apis/base.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
@staticmethod
def check_prompt_dict(prompt_dict: dict) -> list[Exception]:
    """
    Method for checking the prompt dictionary.
    Each subclass should implement this method to check if the
    prompt dictionary is a valid input for the model.

    Parameters
    ----------
    prompt_dict : dict
        The prompt dictionary to check

    Returns
    -------
    list[Exception]
        A list of exceptions or warnings if the prompt dictionary
        is not valid

    Raises
    ------
    NotImplementedError
        If the method is not implemented by a subclass
    """
    raise NotImplementedError(
        "'check_prompt_dict' method needs to be implemented by a subclass of AsyncAPI"
    )

query async 

query(
    prompt_dict: dict, index: int | str = "NA", *args: Any, **kwargs: Any
) -> dict

Method for querying the API/model asynchronously.

Parameters:

Name Type Description Default
prompt_dict dict

The prompt dictionary to use for querying the model

 required
index int | str

The index of the prompt in the experiment, by default “NA”

 'NA'

Returns:

Type Description
dict

Completed prompt_dict with “response” key storing the response(s) from the LLM

Raises:

Type Description
NotImplementedError

If the method is not implemented by a subclass
Source code in src/prompto/apis/base.py 
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
async def query(
    self, prompt_dict: dict, index: int | str = "NA", *args: Any, **kwargs: Any
) -> dict:
    """
    Method for querying the API/model asynchronously.

    Parameters
    ----------
    prompt_dict : dict
        The prompt dictionary to use for querying the model
    index : int | str, optional
        The index of the prompt in the experiment, by default "NA"

    Returns
    -------
    dict
        Completed prompt_dict with "response" key storing the response(s)
        from the LLM

    Raises
    ------
    NotImplementedError
        If the method is not implemented by a subclass
    """
    # async method for querying the model using the prompt and other arguments
    # returns a dictionary/json object which is saved into the output jsonl
    raise NotImplementedError(
        "'query' method needs to be implemented by a subclass of AsyncAPI"
    )