Skip to content

OpenAI

This module implements a wrapper for OpenAI completion models, using BaseLLM as a base class.

OpenAI

Bases: BaseLLM

A class for interacting with the OpenAI completion API.

Inherits from BaseLLM.

Uses the specified OpenAI model and parameters for interacting with the OpenAI API.

Parameters:

Name Type Description Default
model str

The name of the OpenAI model to use.

 'text-davinci-003'
temperature float

The temperature to use for text generation.

 0.7
max_tokens int

The maximum number of tokens to generate.

 500
max_retries int

The maximum number of retries for generating tokens.

 3
api_key str

The API key to use for interacting with the OpenAI API.

 required

Attributes:

Name Type Description
temperature float

The temperature to use for text generation.
max_tokens int

The maximum number of tokens to generate.
max_retries int

The maximum number of retries for generating tokens.
Source code in llmflows/llms/openai.py 
 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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
class OpenAI(BaseLLM):
    """
    A class for interacting with the OpenAI completion API.

    Inherits from BaseLLM.

    Uses the specified OpenAI model and parameters for interacting with the OpenAI API.

    Args:
        model (str): The name of the OpenAI model to use.
        temperature (float): The temperature to use for text generation.
        max_tokens (int): The maximum number of tokens to generate.
        max_retries (int): The maximum number of retries for generating tokens.
        api_key (str): The API key to use for interacting with the OpenAI API.

    Attributes:
        temperature (float): The temperature to use for text generation.
        max_tokens (int): The maximum number of tokens to generate.
        max_retries (int): The maximum number of retries for generating tokens.
    """

    def __init__(
        self,
        api_key: str,
        model: str = "text-davinci-003",
        temperature: float = 0.7,
        max_tokens: int = 500,
        max_retries: int = 3,
    ):
        super().__init__(model)
        self.temperature = temperature
        self.max_tokens = max_tokens
        self.max_retries = max_retries
        self._api_key = api_key
        if not self._api_key:
            raise ValueError("You must provide OpenAI API key")
        openai.api_key = self._api_key

    def _format_results(self, model_outputs, retries) -> tuple[str, dict, dict]:
        """
        Formats results after generation.

        Args:
            model_outputs: Raw output after model generation.
            retries (int): Number of retries taken for successful generation.

        Returns:
            A tuple containing the generated text, the raw response data, and the
                model configuration.
        """
        text_result = model_outputs.choices[0]["text"]

        call_data = {
            "raw_outputs": model_outputs,
            "retries": retries,
        }

        model_config = {
            "model_name": self.model,
            "temperature": self.temperature,
            "max_tokens": self.max_tokens,
        }

        return text_result, call_data, model_config

    def generate(self, prompt: str) -> tuple[str, dict, dict]:
        """
        Generates text from a given prompt using OpenAI API.

        Args:
            prompt (str): Text prompt for generation.

        Returns:
            A tuple containing the generated text, the raw response data, and the
                model configuration.
        """

        if not isinstance(prompt, str):
            raise TypeError("Prompt must be a string")

        completion, retries = call_with_retry(
            func=openai.Completion.create,
            exceptions_to_retry=(
                APIError,
                Timeout,
                RateLimitError,
                APIConnectionError,
                ServiceUnavailableError,
            ),
            max_retries=self.max_retries,
            model=self.model,
            prompt=prompt,
            max_tokens=self.max_tokens,
            temperature=self.temperature,
        )

        return self._format_results(completion, retries)

    async def generate_async(self, prompt: str) -> tuple[str, dict, dict]:
        """
        Generates text from a given prompt using OpenAI API asynchronously.

        Args:
            prompt (str): Text prompt for generation.

        Returns:
            A tuple containing the generated text, the raw response data, and the
                model configuration.
        """

        if not isinstance(prompt, str):
            raise TypeError("Prompt must be a string")

        completion, retries = await async_call_with_retry(
            async_func=openai.Completion.acreate,
            exceptions_to_retry=(
                APIError,
                Timeout,
                RateLimitError,
                APIConnectionError,
                ServiceUnavailableError,
            ),
            max_retries=self.max_retries,
            model=self.model,
            prompt=prompt,
            max_tokens=self.max_tokens,
            temperature=self.temperature,
        )

        return self._format_results(completion, retries)

generate(prompt)

Generates text from a given prompt using OpenAI API.

Parameters:

Name Type Description Default
prompt str

Text prompt for generation.

 required

Returns:

Type Description
tuple[str, dict, dict]

A tuple containing the generated text, the raw response data, and the model configuration.
Source code in llmflows/llms/openai.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def generate(self, prompt: str) -> tuple[str, dict, dict]:
    """
    Generates text from a given prompt using OpenAI API.

    Args:
        prompt (str): Text prompt for generation.

    Returns:
        A tuple containing the generated text, the raw response data, and the
            model configuration.
    """

    if not isinstance(prompt, str):
        raise TypeError("Prompt must be a string")

    completion, retries = call_with_retry(
        func=openai.Completion.create,
        exceptions_to_retry=(
            APIError,
            Timeout,
            RateLimitError,
            APIConnectionError,
            ServiceUnavailableError,
        ),
        max_retries=self.max_retries,
        model=self.model,
        prompt=prompt,
        max_tokens=self.max_tokens,
        temperature=self.temperature,
    )

    return self._format_results(completion, retries)

generate_async(prompt) async

Generates text from a given prompt using OpenAI API asynchronously.

Parameters:

Name Type Description Default
prompt str

Text prompt for generation.

 required

Returns:

Type Description
tuple[str, dict, dict]

A tuple containing the generated text, the raw response data, and the model configuration.
Source code in llmflows/llms/openai.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
async def generate_async(self, prompt: str) -> tuple[str, dict, dict]:
    """
    Generates text from a given prompt using OpenAI API asynchronously.

    Args:
        prompt (str): Text prompt for generation.

    Returns:
        A tuple containing the generated text, the raw response data, and the
            model configuration.
    """

    if not isinstance(prompt, str):
        raise TypeError("Prompt must be a string")

    completion, retries = await async_call_with_retry(
        async_func=openai.Completion.acreate,
        exceptions_to_retry=(
            APIError,
            Timeout,
            RateLimitError,
            APIConnectionError,
            ServiceUnavailableError,
        ),
        max_retries=self.max_retries,
        model=self.model,
        prompt=prompt,
        max_tokens=self.max_tokens,
        temperature=self.temperature,
    )

    return self._format_results(completion, retries)