Skip to content

AzureOpenAI

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

AzureOpenAI

Bases: BaseLLM

A class for interacting with a Azure OpenAI deployment throught the OpenAI Completions API.

Inherits from BaseLLM.

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

Parameters:

Name Type Description Default
deployment_name str

The name of the Azure deployment to use.

 required
azure_openai_endpoint str

The Azure OpenAI endpoint to use.

 required
azure_api_version str

The Azure OpenAI API version to use.

 '2023-05-15'
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/azure_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
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
class AzureOpenAI(BaseLLM):
    """
    A class for interacting with a Azure OpenAI deployment throught the OpenAI
    Completions API.

    Inherits from BaseLLM.

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

    Args:
        deployment_name (str): The name of the Azure deployment to use.
        azure_openai_endpoint (str): The Azure OpenAI endpoint to use.
        azure_api_version (str): The Azure OpenAI API version 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,
        deployment_name: str,
        azure_openai_endpoint: str,
        azure_api_version: str = "2023-05-15",
        temperature: float = 0.7,
        max_tokens: int = 500,
        max_retries: int = 3,
    ):
        super().__init__(model=deployment_name)
        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")

        self._engine = deployment_name
        if not self._engine:
            raise ValueError("You must provide Azure OpenAI deployment name")

        self._azure_openai_endpoint = azure_openai_endpoint
        if not self._azure_openai_endpoint:
            raise ValueError("You must provide Azure OpenAI endpoint")

        openai.api_key = self._api_key
        openai.api_base = self._azure_openai_endpoint
        openai.api_type = "azure"
        openai.api_version = azure_api_version

    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,
            ),
            engine=self._engine,
            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,
            ),
            engine=self._engine,
            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/azure_openai.py 
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
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,
        ),
        engine=self._engine,
        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/azure_openai.py 
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
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,
        ),
        engine=self._engine,
        max_retries=self.max_retries,
        model=self.model,
        prompt=prompt,
        max_tokens=self.max_tokens,
        temperature=self.temperature,
    )

    return self._format_results(completion, retries)