def generic_query()

in src/jobs/util/language_model.py [0:0]

• 27 lines of code
• 9 McCabe index (conditional complexity)

    def generic_query(self, messages: List[MessageSequenceItem], retry_count=0, max_tokens=None, json_only=False) -> str:
        """
        Do a basic openai query
        :param text: Prompt
        :return: Response, or empty string if there is a failure
        """
        retry_allowance = retry_count
        response_format = {"type": "json_object"} if json_only else None
        while retry_allowance >= 0:
            retry_allowance -= 1
            try:
                messages = [dict(m) for m in messages]
                url_response = self.client.chat.completions.create(model=self._engine,
                temperature=self._temperature,
                messages=messages,
                max_tokens=max_tokens,
                response_format=response_format)
                response = url_response.choices[0].message.content
                stats_log(
                    {"message": messages[0]["content"][0:50], "response": response})
                if self._logger is not None:
                    self._logger.debug(f"OpenAI Response {response}")
                return response
            except Exception as ex:
                if self._logger is not None:
                    self._logger.error(f"Error calling openai {ex}")
                if retry_count == 0:
                    raise(ex)
                if self._retry_delay_sec > 0:
                    time.sleep(self._retry_delay_sec)
        self._logger.error("Exceeded retry count. Returning empty string result")
        return ""