Utils

Utility functions for the OpenAI Lambda functions.

smarter.apps.provider.services.text_completion.utils.ensure_system_role_present(messages, default_system_role='You are a helpful llmclient. When given the opportunity to utilize function calling, you should always do so. This will allow you to provide the best possible responses to the user. If you are unable to provide a response, you should prompt the user for more information. If you are still unable to provide a response, you should inform the user that you are unable to help them at this time.')[source]

Ensure that a system role is present in the messages list.

Return type:

list

smarter.apps.provider.services.text_completion.utils.exception_response_factory(exception, request_meta_data=None)[source]

Generate a standardized error response dictionary that includes the Python exception type and stack trace.

exception: a descendant of Python Exception class

Return type:

list | dict[str, Any]

smarter.apps.provider.services.text_completion.utils.get_content_for_role(messages, role)[source]

Get the text content from the messages list for a given role.

Return type:

str

smarter.apps.provider.services.text_completion.utils.get_message_history(messages)[source]

Get the text content from the messages list for a given role.

Return type:

list

smarter.apps.provider.services.text_completion.utils.get_messages_for_role(messages, role)[source]

Get the text content from the messages list for a given role.

Return type:

list

smarter.apps.provider.services.text_completion.utils.get_request_body(data)[source]

Returns the request body as a dictionary.

Parameters:

event – The event object containing the request body.

Return type:

dict[str, Any]

Returns:

A dictionary representing the request body.

smarter.apps.provider.services.text_completion.utils.http_response_factory(status, body, debug_mode=False)[source]

Generate a standardized JSON return dictionary for all possible response scenarios.

Parameters:
  • status (int) – an HTTP response code. see https://developer.mozilla.org/en-US/docs/Web/HTTP/Status

  • body – a JSON dict of http response for status 200, an error dict otherwise. Both are serialized into the “body”, so that an error’s message reaches the client.

  • debug_mode (bool) – if True, log the response to the CloudWatch log for this Lambda.

Return type:

list | dict[str, Any]

See https://docs.aws.amazon.com/lambda/latest/dg/python-handler.html

smarter.apps.provider.services.text_completion.utils.pair_tool_messages(messages)[source]

Return the messages with every tool call paired with its tool reply, as OpenAI requires.

The OpenAI chat completions API refuses, with a 400, a tool message that does not answer a tool_calls entry of the assistant message immediately before it, and an assistant tool_calls entry that no tool message answers. A message thread can break this rule when it comes from a client, such as the chat window, that keeps the tool replies but not the assistant’s tool_calls, which happens when the prompt history has not been persisted yet. So:

  • a tool message is kept only if it answers a tool call of the preceding assistant message.

  • a tool call that no tool message answers is removed from its assistant message.

The messages themselves are not modified: an assistant message that changes is copied.

Parameters:

messages (list[dict[str, Any]]) – The message thread.

Return type:

list[dict[str, Any]]

Returns:

The message thread, in which every tool call is paired with its reply.

smarter.apps.provider.services.text_completion.utils.parse_request(request_body)[source]

Parse the request body and return the endpoint, model, messages, and input_text.

smarter.apps.provider.services.text_completion.utils.should_log(level)[source]

Check if logging should be done based on the waffle switch.