Skip to content

openai_llm ¤

LLM implementation for OpenAI

Classes:

Name Description
AzureOpenAILLM

Azure OpenAI LLM implementation that uses openai sdk to make predictions with Azure's OpenAI.
BaseOpenAILLM

OpenAI LLM implementation that uses openai sdk to make predictions.
OpenAIError

Generic OpenAI error class when working with OpenAI provider.
OpenAILLM

OpenAI LLM implementation that uses openai sdk to make predictions.
OpenAILLMParams

OpenAI LLM Params when running execution

Functions:

Name Description
handle_streaming_response

Accumulate chunk deltas into a full response. Returns the full message.

AzureOpenAILLM ¤

AzureOpenAILLM(
    azure_openai_key: str,
    azure_openai_api_base: str,
    model: str,
    api_version: str = None,
    headers: dict = None,
    timeout: int = None,
    stream: bool = None,
    request_timeout: int = None,
)

Bases: BaseOpenAILLM

Azure OpenAI LLM implementation that uses openai sdk to make predictions with Azure's OpenAI. Args: azure_openai_key: Azure OpenAI API key azure_openai_api_base: Azure OpenAI endpoint model: Deployment name for the model in Azure api_version: Azure API version headers: Headers to use for the request timeout: Timeout to use for the request stream: Stream to use for the request request_timeout: Request timeout to use for the request

Methods:

Name Description
predict

Predicts the next message using OpenAI

Attributes:

Name Type Description
streaming bool

Returns whether the LLM is streaming or not
Source code in src/declarai/operators/openai_operators/openai_llm.py 
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def __init__(
    self,
    azure_openai_key: str,
    azure_openai_api_base: str,
    model: str,
    api_version: str = None,
    headers: dict = None,
    timeout: int = None,
    stream: bool = None,
    request_timeout: int = None,
):
    model = model or DEPLOYMENT_NAME
    api_key = azure_openai_key or AZURE_OPENAI_KEY
    api_version = api_version or AZURE_API_VERSION
    api_base = azure_openai_api_base or AZURE_OPENAI_API_BASE

    super().__init__(
        api_key,
        "azure",
        model,
        headers,
        timeout,
        stream,
        request_timeout,
        engine=model,
        api_version=api_version,
        api_base=api_base,
    )

streaming property ¤

streaming: bool

Returns whether the LLM is streaming or not Returns: bool: True if the LLM is streaming, False otherwise

predict ¤

predict(
    messages: List[Message],
    model: str = None,
    temperature: float = 0,
    max_tokens: int = 3000,
    top_p: float = 1,
    frequency_penalty: int = 0,
    presence_penalty: int = 0,
    stream: bool = None,
) -> Union[Iterator[LLMResponse], LLMResponse]

Predicts the next message using OpenAI Args: stream: if to stream the response messages: List of messages that are used as context for the prediction model: the model to use for the prediction temperature: the temperature to use for the prediction max_tokens: the maximum number of tokens to use for the prediction top_p: the top p to use for the prediction frequency_penalty: the frequency penalty to use for the prediction presence_penalty: the presence penalty to use for the prediction

Returns:

Name Type Description
LLMResponse Union[Iterator[LLMResponse], LLMResponse]

The response from the LLM
Source code in src/declarai/operators/openai_operators/openai_llm.py 
 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
def predict(
    self,
    messages: List[Message],
    model: str = None,
    temperature: float = 0,
    max_tokens: int = 3000,
    top_p: float = 1,
    frequency_penalty: int = 0,
    presence_penalty: int = 0,
    stream: bool = None,
) -> Union[Iterator[LLMResponse], LLMResponse]:
    """
    Predicts the next message using OpenAI
    Args:
        stream: if to stream the response
        messages: List of messages that are used as context for the prediction
        model: the model to use for the prediction
        temperature: the temperature to use for the prediction
        max_tokens: the maximum number of tokens to use for the prediction
        top_p: the top p to use for the prediction
        frequency_penalty: the frequency penalty to use for the prediction
        presence_penalty: the presence penalty to use for the prediction

    Returns:
        LLMResponse: The response from the LLM

    """
    if stream is None:
        stream = self.stream
    openai_messages = [{"role": m.role, "content": m.message} for m in messages]
    res = self.openai.ChatCompletion.create(
        model=model or self.model,
        messages=openai_messages,
        temperature=temperature,
        max_tokens=max_tokens,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        api_key=self.api_key,
        api_type=self.api_type,
        stream=stream,
        **self._kwargs,
    )

    if stream:
        return handle_streaming_response(res)

    else:
        return LLMResponse(
            response=res.choices[0]["message"]["content"],
            model=res.model,
            prompt_tokens=res["usage"]["prompt_tokens"],
            completion_tokens=res["usage"]["completion_tokens"],
            total_tokens=res["usage"]["total_tokens"],
            raw_response=res.to_dict_recursive(),
        )

BaseOpenAILLM ¤

BaseOpenAILLM(
    api_key: str,
    api_type: str,
    model_name: str,
    headers: dict = None,
    timeout: int = None,
    stream: bool = None,
    request_timeout: int = None,
    **kwargs: int
)

Bases: BaseLLM

OpenAI LLM implementation that uses openai sdk to make predictions. Args: openai_token: OpenAI API key model: OpenAI model name Attributes: openai (openai): OpenAI SDK model (str): OpenAI model name

Methods:

Name Description
predict

Predicts the next message using OpenAI

Attributes:

Name Type Description
streaming bool

Returns whether the LLM is streaming or not
Source code in src/declarai/operators/openai_operators/openai_llm.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def __init__(
    self,
    api_key: str,
    api_type: str,
    model_name: str,
    headers: dict = None,
    timeout: int = None,
    stream: bool = None,
    request_timeout: int = None,
    **kwargs,
):
    self._kwargs = {
        "headers": headers,
        "timeout": timeout,
        "request_timeout": request_timeout,
        **kwargs,
    }
    self.openai = openai
    self.api_key = api_key
    self.api_type = api_type
    self.stream = stream
    self.model = model_name

streaming property ¤

streaming: bool

Returns whether the LLM is streaming or not Returns: bool: True if the LLM is streaming, False otherwise

predict ¤

predict(
    messages: List[Message],
    model: str = None,
    temperature: float = 0,
    max_tokens: int = 3000,
    top_p: float = 1,
    frequency_penalty: int = 0,
    presence_penalty: int = 0,
    stream: bool = None,
) -> Union[Iterator[LLMResponse], LLMResponse]

Predicts the next message using OpenAI Args: stream: if to stream the response messages: List of messages that are used as context for the prediction model: the model to use for the prediction temperature: the temperature to use for the prediction max_tokens: the maximum number of tokens to use for the prediction top_p: the top p to use for the prediction frequency_penalty: the frequency penalty to use for the prediction presence_penalty: the presence penalty to use for the prediction

Returns:

Name Type Description
LLMResponse Union[Iterator[LLMResponse], LLMResponse]

The response from the LLM
Source code in src/declarai/operators/openai_operators/openai_llm.py 
 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
def predict(
    self,
    messages: List[Message],
    model: str = None,
    temperature: float = 0,
    max_tokens: int = 3000,
    top_p: float = 1,
    frequency_penalty: int = 0,
    presence_penalty: int = 0,
    stream: bool = None,
) -> Union[Iterator[LLMResponse], LLMResponse]:
    """
    Predicts the next message using OpenAI
    Args:
        stream: if to stream the response
        messages: List of messages that are used as context for the prediction
        model: the model to use for the prediction
        temperature: the temperature to use for the prediction
        max_tokens: the maximum number of tokens to use for the prediction
        top_p: the top p to use for the prediction
        frequency_penalty: the frequency penalty to use for the prediction
        presence_penalty: the presence penalty to use for the prediction

    Returns:
        LLMResponse: The response from the LLM

    """
    if stream is None:
        stream = self.stream
    openai_messages = [{"role": m.role, "content": m.message} for m in messages]
    res = self.openai.ChatCompletion.create(
        model=model or self.model,
        messages=openai_messages,
        temperature=temperature,
        max_tokens=max_tokens,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        api_key=self.api_key,
        api_type=self.api_type,
        stream=stream,
        **self._kwargs,
    )

    if stream:
        return handle_streaming_response(res)

    else:
        return LLMResponse(
            response=res.choices[0]["message"]["content"],
            model=res.model,
            prompt_tokens=res["usage"]["prompt_tokens"],
            completion_tokens=res["usage"]["completion_tokens"],
            total_tokens=res["usage"]["total_tokens"],
            raw_response=res.to_dict_recursive(),
        )

OpenAIError ¤

Bases: Exception

Generic OpenAI error class when working with OpenAI provider.

OpenAILLM ¤

OpenAILLM(
    openai_token: str = None,
    model: str = None,
    headers: dict = None,
    timeout: int = None,
    stream: bool = None,
    request_timeout: int = None,
)

Bases: BaseOpenAILLM

OpenAI LLM implementation that uses openai sdk to make predictions. Args: openai_token: OpenAI API key model: OpenAI model name headers: Headers to use for the request timeout: Timeout to use for the request stream: Stream to use for the request request_timeout: Request timeout to use for the request

Methods:

Name Description
predict

Predicts the next message using OpenAI

Attributes:

Name Type Description
streaming bool

Returns whether the LLM is streaming or not
Source code in src/declarai/operators/openai_operators/openai_llm.py 
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
170
171
def __init__(
    self,
    openai_token: str = None,
    model: str = None,
    headers: dict = None,
    timeout: int = None,
    stream: bool = None,
    request_timeout: int = None,
):
    openai_token = openai_token or OPENAI_API_KEY
    model = model or OPENAI_MODEL
    if not openai_token:
        raise OpenAIError(
            "Missing an OpenAI API key"
            "In order to work with OpenAI, you will need to provide an API key"
            "either by setting the DECLARAI_OPENAI_API_KEY or by providing"
            "the API key via the init interface."
        )
    if not model:
        raise OpenAIError(
            "Missing an OpenAI model"
            "In order to work with OpenAI, you will need to provide a model"
            "either by setting the DECLARAI_OPENAI_MODEL or by providing"
            "the model via the init interface."
        )
    super().__init__(
        openai_token, "openai", model, headers, timeout, stream, request_timeout
    )

streaming property ¤

streaming: bool

Returns whether the LLM is streaming or not Returns: bool: True if the LLM is streaming, False otherwise

predict ¤

predict(
    messages: List[Message],
    model: str = None,
    temperature: float = 0,
    max_tokens: int = 3000,
    top_p: float = 1,
    frequency_penalty: int = 0,
    presence_penalty: int = 0,
    stream: bool = None,
) -> Union[Iterator[LLMResponse], LLMResponse]

Predicts the next message using OpenAI Args: stream: if to stream the response messages: List of messages that are used as context for the prediction model: the model to use for the prediction temperature: the temperature to use for the prediction max_tokens: the maximum number of tokens to use for the prediction top_p: the top p to use for the prediction frequency_penalty: the frequency penalty to use for the prediction presence_penalty: the presence penalty to use for the prediction

Returns:

Name Type Description
LLMResponse Union[Iterator[LLMResponse], LLMResponse]

The response from the LLM
Source code in src/declarai/operators/openai_operators/openai_llm.py 
 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
def predict(
    self,
    messages: List[Message],
    model: str = None,
    temperature: float = 0,
    max_tokens: int = 3000,
    top_p: float = 1,
    frequency_penalty: int = 0,
    presence_penalty: int = 0,
    stream: bool = None,
) -> Union[Iterator[LLMResponse], LLMResponse]:
    """
    Predicts the next message using OpenAI
    Args:
        stream: if to stream the response
        messages: List of messages that are used as context for the prediction
        model: the model to use for the prediction
        temperature: the temperature to use for the prediction
        max_tokens: the maximum number of tokens to use for the prediction
        top_p: the top p to use for the prediction
        frequency_penalty: the frequency penalty to use for the prediction
        presence_penalty: the presence penalty to use for the prediction

    Returns:
        LLMResponse: The response from the LLM

    """
    if stream is None:
        stream = self.stream
    openai_messages = [{"role": m.role, "content": m.message} for m in messages]
    res = self.openai.ChatCompletion.create(
        model=model or self.model,
        messages=openai_messages,
        temperature=temperature,
        max_tokens=max_tokens,
        top_p=top_p,
        frequency_penalty=frequency_penalty,
        presence_penalty=presence_penalty,
        api_key=self.api_key,
        api_type=self.api_type,
        stream=stream,
        **self._kwargs,
    )

    if stream:
        return handle_streaming_response(res)

    else:
        return LLMResponse(
            response=res.choices[0]["message"]["content"],
            model=res.model,
            prompt_tokens=res["usage"]["prompt_tokens"],
            completion_tokens=res["usage"]["completion_tokens"],
            total_tokens=res["usage"]["total_tokens"],
            raw_response=res.to_dict_recursive(),
        )

OpenAILLMParams ¤

Bases: BaseLLMParams

OpenAI LLM Params when running execution

Attributes:

Name Type Description
temperature Optional[float]

the temperature to use for the prediction
max_tokens Optional[int]

the maximum number of tokens to use for the prediction
top_p Optional[float]

the top p to use for the prediction
frequency_penalty Optional[int]

the frequency penalty to use for the prediction
presence_penalty Optional[int]

the presence penalty to use for the prediction

handle_streaming_response ¤

handle_streaming_response(
    api_response: OpenAIObject,
) -> Iterator[LLMResponse]

Accumulate chunk deltas into a full response. Returns the full message.

Source code in src/declarai/operators/openai_operators/openai_llm.py 
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
def handle_streaming_response(api_response: OpenAIObject) -> Iterator[LLMResponse]:
    """
    Accumulate chunk deltas into a full response. Returns the full message.
    """
    response = {"role": None, "response": "", "raw_response": ""}

    chunk: OpenAIObject
    for chunk in api_response:  # noqa
        response["raw_response"] = chunk.to_dict_recursive()
        delta = chunk.choices[0]["delta"]
        response["model"] = chunk.model
        if chunk.get("usage"):
            response["prompt_tokens"] = chunk.usage.get("prompt_tokens")
            response["completion_tokens"] = chunk.usage.get("completion_tokens")
            response["total_tokens"] = chunk.usage.get("total_tokens")

        if "role" in delta:
            response["role"] = delta["role"]

        if delta.get("function_call"):
            fn_call = delta.get("function_call")
            if "function_call" not in response["data"]:
                response["data"]["function_call"] = {"name": None, "arguments": ""}
            if "name" in fn_call:
                response["data"]["function_call"]["name"] = fn_call.name
            if "arguments" in fn_call:
                response["data"]["function_call"]["arguments"] += (
                    fn_call.arguments or ""
                )

        if "content" in delta:
            response["response"] += delta.content or ""

        yield LLMResponse(**response)