utils/llm.py

import re
import json
import ast
import os
import chevron
from typing import Collection, Dict, List, Union
from pydantic import BaseModel
import copy
import functools
import inspect
import pprint
import textwrap

from tinytroupe import utils
from tinytroupe.utils import logger
from tinytroupe.utils.rendering import break_text_at_length

################################################################################
# Model input utilities
################################################################################

def compose_initial_LLM_messages_with_templates(system_template_name:str, user_template_name:str=None, 
                                                base_module_folder:str=None,
                                                rendering_configs:dict={}) -> list:
    """
    Composes the initial messages for the LLM model call, under the assumption that it always involves 
    a system (overall task description) and an optional user message (specific task description). 
    These messages are composed using the specified templates and rendering configurations.
    """

    # ../ to go to the base library folder, because that's the most natural reference point for the user
    if base_module_folder is None:
        sub_folder =  "../prompts/" 
    else:
        sub_folder = f"../{base_module_folder}/prompts/"

    base_template_folder = os.path.join(os.path.dirname(__file__), sub_folder)    

    system_prompt_template_path = os.path.join(base_template_folder, f'{system_template_name}')
    user_prompt_template_path = os.path.join(base_template_folder, f'{user_template_name}')

    messages = []

    messages.append({"role": "system", 
                         "content": chevron.render(
                             open(system_prompt_template_path, 'r', encoding='utf-8', errors='replace').read(), 
                             rendering_configs)})
    
    # optionally add a user message
    if user_template_name is not None:
        messages.append({"role": "user", 
                            "content": chevron.render(
                                    open(user_prompt_template_path, 'r', encoding='utf-8', errors='replace').read(), 
                                    rendering_configs)})
    return messages


#
# Data structures to enforce output format during LLM API call.
#

class LLMScalarWithJustificationResponse(BaseModel):
    """
    Represents a typed response from an LLM (Language Learning Model) including justification.
    Attributes:
        justification (str): The justification or explanation for the response.
        value (str, int, float, bool): The value of the response.
        confidence (float): The confidence level of the response.        
    """
    justification: str
    value: Union[str, int, float, bool]    
    confidence: float

class LLMScalarWithJustificationAndReasoningResponse(BaseModel):
    """
    Represents a typed response from an LLM (Language Learning Model) including justification and reasoning.
    Attributes:
        reasoning (str): The reasoning behind the response.
        justification (str): The justification or explanation for the response.
        value (str, int, float, bool): The value of the response.
        confidence (float): The confidence level of the response.
    """
    reasoning: str

    # we need to repeat these fields here, instead of inheriting from LLMScalarWithJustificationResponse,
    # because we need to ensure `reasoning` is always the first field in the JSON object.
    justification: str
    value: Union[str, int, float, bool]
    confidence: float



###########################################################################
# Model calling helpers
###########################################################################

class LLMChat:
    """
    A class that represents an ongoing LLM conversation. It maintains the conversation history,
    allows adding new messages, and handles model output type coercion.
    """

    def __init__(self, system_template_name:str=None, system_prompt:str=None,
                 user_template_name:str=None, user_prompt:str=None,
                 base_module_folder=None,
                 output_type=None,
                 enable_json_output_format:bool=True,
                 enable_justification_step:bool=True,
                 enable_reasoning_step:bool=False,
                 **model_params):
        """
        Initializes an LLMChat instance with the specified system and user templates, or the system and user prompts.
        If a template is specified, the corresponding prompt must be None, and vice versa.

        Args:
            system_template_name (str): Name of the system template file.
            system_prompt (str): System prompt content.
            user_template_name (str): Name of the user template file.
            user_prompt (str): User prompt content.
            base_module_folder (str): Optional subfolder path within the library where templates are located.
            output_type (type): Expected type of the model output.
            enable_reasoning_step (bool): Flag to enable reasoning step in the conversation. This IS NOT the use of "reasoning models" (e.g., o1, o3),
              but rather the use of an additional reasoning step in the regular text completion.
            enable_justification_step (bool): Flag to enable justification step in the conversation. Must be True if reasoning step is enabled as well.
            enable_json_output_format (bool): Flag to enable JSON output format for the model response. Must be True if reasoning or justification steps are enabled.
            **model_params: Additional parameters for the LLM model call.

        """
        if (system_template_name is not None and system_prompt is not None) or \
        (user_template_name is not None and user_prompt is not None) or\
        (system_template_name is None and system_prompt is None) or \
        (user_template_name is None and user_prompt is None):
            raise ValueError("Either the template or the prompt must be specified, but not both.")

        self.base_module_folder = base_module_folder

        self.system_template_name = system_template_name
        self.user_template_name = user_template_name

        self.system_prompt = textwrap.dedent(system_prompt) if system_prompt is not None else None
        self.user_prompt = textwrap.dedent(user_prompt) if user_prompt is not None else None

        self.output_type = output_type

        self.enable_reasoning_step = enable_reasoning_step
        self.enable_justification_step = enable_justification_step
        self.enable_json_output_format = enable_json_output_format

        self.model_params = model_params
        
        # Conversation history
        self.messages = []
        self.conversation_history = []
        
        # Response tracking
        self.response_raw = None
        self.response_json = None
        self.response_reasoning = None
        self.response_value = None
        self.response_justification = None
        self.response_confidence = None
        
    def __call__(self, *args, **kwds):
        return self.call(*args, **kwds)
        
    def _render_template(self, template_name, base_module_folder=None, rendering_configs={}):
        """
        Helper method to render templates for messages.
        
        Args:
            template_name: Name of the template file
            base_module_folder: Optional subfolder path within the library
            rendering_configs: Configuration variables for template rendering
            
        Returns:
            Rendered template content
        """
        if base_module_folder is None:
            sub_folder =  "../prompts/" 
        else:
            sub_folder = f"../{base_module_folder}/prompts/"
            
        base_template_folder = os.path.join(os.path.dirname(__file__), sub_folder)
        template_path = os.path.join(base_template_folder, template_name)

        return chevron.render(open(template_path, 'r', encoding='utf-8', errors='replace').read(), rendering_configs)

    def add_user_message(self, message=None, template_name=None, base_module_folder=None, rendering_configs={}):
        """
        Add a user message to the conversation.
        
        Args:
            message: The direct message content from the user (mutually exclusive with template_name)
            template_name: Optional template file name to use for the message
            base_module_folder: Optional subfolder for template location
            rendering_configs: Configuration variables for template rendering
            
        Returns:
            self for method chaining
        """
        if message is not None and template_name is not None:
            raise ValueError("Either message or template_name must be specified, but not both.")
        
        if template_name is not None:
            content = self._render_template(template_name, base_module_folder, rendering_configs)
        else:
            content = textwrap.dedent(message)
            
        self.messages.append({"role": "user", "content": content})
        return self
        
    def add_system_message(self, message=None, template_name=None, base_module_folder=None, rendering_configs={}):
        """
        Add a system message to the conversation.
        
        Args:
            message: The direct message content from the system (mutually exclusive with template_name)
            template_name: Optional template file name to use for the message
            base_module_folder: Optional subfolder for template location
            rendering_configs: Configuration variables for template rendering
            
        Returns:
            self for method chaining
        """
        if message is not None and template_name is not None:
            raise ValueError("Either message or template_name must be specified, but not both.")
        
        if template_name is not None:
            content = self._render_template(template_name, base_module_folder, rendering_configs)
        else:
            content = textwrap.dedent(message)
            
        self.messages.append({"role": "system", "content": content})
        return self
    
    def add_assistant_message(self, message=None, template_name=None, base_module_folder=None, rendering_configs={}):
        """
        Add an assistant message to the conversation.
        
        Args:
            message: The direct message content from the assistant (mutually exclusive with template_name)
            template_name: Optional template file name to use for the message
            base_module_folder: Optional subfolder for template location
            rendering_configs: Configuration variables for template rendering
            
        Returns:
            self for method chaining
        """
        if message is not None and template_name is not None:
            raise ValueError("Either message or template_name must be specified, but not both.")
        
        if template_name is not None:
            content = self._render_template(template_name, base_module_folder, rendering_configs)
        else:
            content = textwrap.dedent(message)
            
        self.messages.append({"role": "assistant", "content": content})
        return self

    def set_model_params(self, **model_params):
        """
        Set or update the model parameters for the LLM call.
        
        Args:
            model_params: Key-value pairs of model parameters to set or update
        """
        self.model_params.update(model_params)
        return self

    def call(self, output_type="default", 
                   enable_json_output_format:bool=None,        
                   enable_justification_step:bool=None,
                   enable_reasoning_step:bool=None, 
                   **rendering_configs):
        """
        Initiates or continues the conversation with the LLM model using the current message history.

        Args:
            output_type: Optional parameter to override the output type for this specific call. If set to "default", it uses the instance's output_type.
                         If set to None, removes all output formatting and coercion.
            enable_json_output_format: Optional flag to enable JSON output format for the model response. If None, uses the instance's setting.
            enable_justification_step: Optional flag to enable justification step in the conversation. If None, uses the instance's setting.
            enable_reasoning_step: Optional flag to enable reasoning step in the conversation. If None, uses the instance's setting.
            rendering_configs: The rendering configurations (template variables) to use when composing the initial messages.

        Returns:
            The content of the model response.
        """
        from tinytroupe.openai_utils import client # import here to avoid circular import

        try:

            # Initialize the conversation if this is the first call
            if not self.messages:
                if self.system_template_name is not None and self.user_template_name is not None:
                    self.messages = utils.compose_initial_LLM_messages_with_templates(
                        self.system_template_name, 
                        self.user_template_name, 
                        base_module_folder=self.base_module_folder,
                        rendering_configs=rendering_configs
                    )
                else:
                    if self.system_prompt:
                        self.messages.append({"role": "system", "content": self.system_prompt})
                    if self.user_prompt:
                        self.messages.append({"role": "user", "content": self.user_prompt})

            # Use the provided output_type if specified, otherwise fall back to the instance's output_type
            current_output_type = output_type if output_type != "default" else self.output_type

            # Set up typing for the output
            if current_output_type is not None:
                
                # TODO obsolete?
                #
                ## Add type coercion instructions if not already added
                #if not any(msg.get("content", "").startswith("In your response, you **MUST** provide a value") 
                #          for msg in self.messages if msg.get("role") == "system"):
                
                # the user can override the response format by specifying it in the model_params, otherwise
                # we will use the default response format
                if "response_format" not in self.model_params or self.model_params["response_format"] is None:

                    if utils.first_non_none(enable_json_output_format, self.enable_json_output_format):

                        self.model_params["response_format"] = {"type": "json_object"}

                        typing_instruction = {"role": "system",
                            "content": "Your response **MUST** be a JSON object."}

                        # Special justification format can be used (will also include confidence level)
                        if utils.first_non_none(enable_justification_step, self.enable_justification_step):

                            # Add reasoning step if enabled provides further mechanism to think step-by-step
                            if not (utils.first_non_none(enable_reasoning_step, self.enable_reasoning_step)):
                                # Default structured output
                                self.model_params["response_format"] = LLMScalarWithJustificationResponse
                                
                                typing_instruction = {"role": "system",
                                    "content": "In your response, you **MUST** provide a value, along with a justification and your confidence level that the value and justification are correct (0.0 means no confidence, 1.0 means complete confidence). "+
                                    "Furtheremore, your response **MUST** be a JSON object with the following structure: {\"justification\": justification, \"value\": value, \"confidence\": confidence}. "+
                                    "Note that \"justification\" comes first in order to help you think about the value you are providing."}
                            
                            else: 
                                # Override the response format to also use a reasoning step
                                self.model_params["response_format"] = LLMScalarWithJustificationAndReasoningResponse
                                
                                typing_instruction = {"role": "system",
                                    "content": \
                                        "In your response, you **FIRST** think step-by-step on how you are going to compute the value, and you put this reasoning in the \"reasoning\" field (which must come before all others). "+
                                        "This allows you to think carefully as much as you need to deduce the best and most correct value. "+
                                        "After that, you **MUST** provide the resulting value, along with a justification (which can tap into the previous reasoning), and your confidence level that the value and justification are correct (0.0 means no confidence, 1.0 means complete confidence)."+
                                        "Furtheremore, your response **MUST** be a JSON object with the following structure: {\"reasoning\": reasoning, \"justification\": justification, \"value\": value, \"confidence\": confidence}." +
                                        " Note that \"justification\" comes after \"reasoning\" but before \"value\" to help with further formulation of the resulting \"value\"."}
                        
                    
                            # Specify the value type
                            if current_output_type == bool:
                                typing_instruction["content"] += " " + self._request_bool_llm_message()["content"]
                            elif current_output_type == int:
                                typing_instruction["content"] += " " + self._request_integer_llm_message()["content"]
                            elif current_output_type == float:
                                typing_instruction["content"] += " " + self._request_float_llm_message()["content"]
                            elif isinstance(current_output_type, list) and all(isinstance(option, str) for option in current_output_type):
                                typing_instruction["content"] += " " + self._request_enumerable_llm_message(current_output_type)["content"]
                            elif current_output_type == List[Dict[str, any]]:
                                # Override the response format
                                self.model_params["response_format"] = {"type": "json_object"}
                                typing_instruction["content"] += " " + self._request_list_of_dict_llm_message()["content"]
                            elif current_output_type == dict or current_output_type == "json":
                                # Override the response format
                                self.model_params["response_format"] = {"type": "json_object"}
                                typing_instruction["content"] += " " + self._request_dict_llm_message()["content"]
                            elif current_output_type == list:
                                # Override the response format
                                self.model_params["response_format"] = {"type": "json_object"}
                                typing_instruction["content"] += " " + self._request_list_llm_message()["content"]
                            # Check if it is actually a pydantic model
                            elif issubclass(current_output_type, BaseModel):
                                # Completely override the response format
                                self.model_params["response_format"] = current_output_type
                                typing_instruction = {"role": "system", "content": "Your response **MUST** be a JSON object."}
                            elif current_output_type == str:
                                typing_instruction["content"] += " " + self._request_str_llm_message()["content"]
                                #pass # no coercion needed, it is already a string
                            else:
                                raise ValueError(f"Unsupported output type: {current_output_type}")
                            
                            self.messages.append(typing_instruction)
                    
                    else: # output_type is None
                        self.model_params["response_format"] = None
                        typing_instruction = {"role": "system", "content": \
                                                "If you were given instructions before about the **format** of your response, please ignore them from now on. "+
                                                "The needs of the user have changed. You **must** now use regular text -- not numbers, not booleans, not JSON. "+
                                                "There are no fields, no types, no special formats. Just regular text appropriate to respond to the last user request."}
                        self.messages.append(typing_instruction)
                        #pass  # nothing here for now


            # Call the LLM model with all messages in the conversation
            model_output = client().send_message(self.messages, **self.model_params)

            if 'content' in model_output:
                self.response_raw = self.response_value = model_output['content']
                logger.debug(f"Model raw 'content'  response: {self.response_raw}")
                
                # Add the assistant's response to the conversation history
                self.add_assistant_message(self.response_raw)
                self.conversation_history.append({"messages": copy.deepcopy(self.messages)})

                # Type coercion if output type is specified
                if current_output_type is not None:
                    
                    if self.enable_json_output_format:
                        # output is supposed to be a JSON object
                        self.response_json = self.response_value = utils.extract_json(self.response_raw)
                        logger.debug(f"Model output JSON response: {self.response_json}")

                        if self.enable_justification_step and not (hasattr(current_output_type, 'model_validate') or hasattr(current_output_type, 'parse_obj')):
                            # if justification step is enabled, we expect a JSON object with reasoning (optionally), justification, value, and confidence
                            # BUT not for Pydantic models which expect direct JSON structure
                            self.response_reasoning = self.response_json.get("reasoning", None)
                            self.response_value = self.response_json.get("value", None)
                            self.response_justification = self.response_json.get("justification", None)
                            self.response_confidence = self.response_json.get("confidence", None)
                        else:
                            # For direct JSON output (like Pydantic models), use the whole JSON as the value
                            self.response_value = self.response_json
                        
                    # if output type was specified, we need to coerce the response value
                    if self.response_value is not None:
                        if current_output_type == bool:
                            self.response_value = self._coerce_to_bool(self.response_value)
                        elif current_output_type == int:
                            self.response_value = self._coerce_to_integer(self.response_value)
                        elif current_output_type == float:
                            self.response_value = self._coerce_to_float(self.response_value)
                        elif isinstance(current_output_type, list) and all(isinstance(option, str) for option in current_output_type):
                            self.response_value = self._coerce_to_enumerable(self.response_value, current_output_type)
                        elif current_output_type == List[Dict[str, any]]:
                            self.response_value = self._coerce_to_dict_or_list(self.response_value)
                        elif current_output_type == dict or current_output_type == "json":
                            self.response_value = self._coerce_to_dict_or_list(self.response_value)
                        elif current_output_type == list:
                            self.response_value = self._coerce_to_list(self.response_value)
                        elif hasattr(current_output_type, 'model_validate') or hasattr(current_output_type, 'parse_obj'):
                            # Handle Pydantic model - try modern approach first, then fallback
                            try:
                                if hasattr(current_output_type, 'model_validate'):
                                    self.response_value = current_output_type.model_validate(self.response_json)
                                else:
                                    self.response_value = current_output_type.parse_obj(self.response_json)
                            except Exception as e:
                                logger.error(f"Failed to parse Pydantic model: {e}")
                                raise
                        elif current_output_type == str:
                            pass # no coercion needed, it is already a string
                        else:
                            raise ValueError(f"Unsupported output type: {current_output_type}")
                    
                    else:
                        logger.error(f"Model output is None: {self.response_raw}")
                
                logger.debug(f"Model output coerced response value: {self.response_value}")
                logger.debug(f"Model output coerced response justification: {self.response_justification}")
                logger.debug(f"Model output coerced response confidence: {self.response_confidence}")

                return self.response_value
            else:
                logger.error(f"Model output does not contain 'content' key: {model_output}")
                return None
        
        except ValueError as ve:
            # Re-raise ValueError exceptions (like unsupported output type) instead of catching them
            if "Unsupported output type" in str(ve):
                raise
            else:
                logger.error(f"Error during LLM call: {ve}. Will return None instead of failing.")
                return None
        except Exception as e:
            logger.error(f"Error during LLM call: {e}. Will return None instead of failing.")
            return None
    
    def continue_conversation(self, user_message=None, **rendering_configs):
        """
        Continue the conversation with a new user message and get a response.
        
        Args:
            user_message: The new message from the user
            rendering_configs: Additional rendering configurations
            
        Returns:
            The content of the model response
        """
        if user_message:
            self.add_user_message(user_message)
        return self.call(**rendering_configs)
    
    def reset_conversation(self):
        """
        Reset the conversation state but keep the initial configuration.
        
        Returns:
            self for method chaining
        """
        self.messages = []
        self.response_raw = None
        self.response_json = None
        self.response_value = None
        self.response_justification = None
        self.response_confidence = None
        return self
    
    def get_conversation_history(self):
        """
        Get the full conversation history.
        
        Returns:
            List of all messages in the conversation
        """
        return self.messages
    
    # Keep all the existing coercion methods
    def _coerce_to_bool(self, llm_output):
        """
        Coerces the LLM output to a boolean value.

        This method looks for the string "True", "False", "Yes", "No", "Positive", "Negative" in the LLM output, such that
          - case is neutralized;
          - the first occurrence of the string is considered, the rest is ignored. For example,  " Yes, that is true" will be considered "Yes";
          - if no such string is found, the method raises an error. So it is important that the prompts actually requests a boolean value. 

        Args:
            llm_output (str, bool): The LLM output to coerce.

        Returns:
            The boolean value of the LLM output.
        """

        # if the LLM output is already a boolean, we return it
        if isinstance(llm_output, bool):
            return llm_output

        # let's extract the first occurrence of the string "True", "False", "Yes", "No", "Positive", "Negative" in the LLM output.
        # using a regular expression
        import re
        match = re.search(r'\b(?:True|False|Yes|No|Positive|Negative)\b', llm_output, re.IGNORECASE)
        if match:
            first_match = match.group(0).lower()
            if first_match in ["true", "yes", "positive"]:
                return True
            elif first_match in ["false", "no", "negative"]:
                return False

        raise ValueError("Cannot convert the LLM output to a boolean value.")

    def _request_str_llm_message(self):
        return {"role": "user",
                "content": "The `value` field you generate from now on has no special format, it can be any string you find appropriate to the current conversation. "+
                           "Make sure you move to `value` **all** relevant information you used in reasoning or justification, so that it is not lost. "}
    
    def _request_bool_llm_message(self):
        return {"role": "user",
                "content": "The `value` field you generate **must** be either 'True' or 'False'. This is critical for later processing. If you don't know the correct answer, just output 'False'."}


    def _coerce_to_integer(self, llm_output:str):
        """
        Coerces the LLM output to an integer value.

        This method looks for the first occurrence of an integer in the LLM output, such that
          - the first occurrence of the integer is considered, the rest is ignored. For example,  "There are 3 cats" will be considered 3;
          - if no integer is found, the method raises an error. So it is important that the prompts actually requests an integer value. 

        Args:
            llm_output (str, int): The LLM output to coerce.

        Returns:
            The integer value of the LLM output.
        """

        # if the LLM output is already an integer, we return it
        if isinstance(llm_output, int):
            return llm_output
        
        # if it's a float that represents a whole number, convert it
        if isinstance(llm_output, float):
            if llm_output.is_integer():
                return int(llm_output)
            else:
                raise ValueError("Cannot convert the LLM output to an integer value.")

        # Convert to string for regex processing
        llm_output_str = str(llm_output)

        # let's extract the first occurrence of an integer in the LLM output.
        # using a regular expression
        import re
        # Match integers that are not part of a decimal number
        # First check if the string contains a decimal point - if so, reject it for integer coercion
        if '.' in llm_output_str and any(c.isdigit() for c in llm_output_str.split('.')[1]):
            # This looks like a decimal number, not a pure integer
            raise ValueError("Cannot convert the LLM output to an integer value.")
        
        match = re.search(r'-?\b\d+\b', llm_output_str)
        if match:
            return int(match.group(0))

        raise ValueError("Cannot convert the LLM output to an integer value.")

    def _request_integer_llm_message(self):
        return {"role": "user",
                "content": "The `value` field you generate **must** be an integer number (e.g., '1'). This is critical for later processing.."}

    def _coerce_to_float(self, llm_output:str):
        """
        Coerces the LLM output to a float value.

        This method looks for the first occurrence of a float in the LLM output, such that
          - the first occurrence of the float is considered, the rest is ignored. For example,  "The price is $3.50" will be considered 3.50;
          - if no float is found, the method raises an error. So it is important that the prompts actually requests a float value. 

        Args:
            llm_output (str, float): The LLM output to coerce.

        Returns:
            The float value of the LLM output.
        """

        # if the LLM output is already a float, we return it
        if isinstance(llm_output, float):
            return llm_output
        
        # if it's an integer, convert to float
        if isinstance(llm_output, int):
            return float(llm_output)

        # let's extract the first occurrence of a number (float or int) in the LLM output.
        # using a regular expression that handles negative numbers and both int/float formats
        import re
        match = re.search(r'-?\b\d+(?:\.\d+)?\b', llm_output)
        if match:
            return float(match.group(0))

        raise ValueError("Cannot convert the LLM output to a float value.")

    def _request_float_llm_message(self):
        return {"role": "user",
                "content": "The `value` field you generate **must** be a float number (e.g., '980.16'). This is critical for later processing."}

    def _coerce_to_enumerable(self, llm_output:str, options:list):
        """
        Coerces the LLM output to one of the specified options.

        This method looks for the first occurrence of one of the specified options in the LLM output, such that
          - the first occurrence of the option is considered, the rest is ignored. For example,  "I prefer cats" will be considered "cats";
          - if no option is found, the method raises an error. So it is important that the prompts actually requests one of the specified options. 

        Args:
            llm_output (str): The LLM output to coerce.
            options (list): The list of options to consider.

        Returns:
            The option value of the LLM output.
        """

        # let's extract the first occurrence of one of the specified options in the LLM output.
        # using a regular expression
        import re
        match = re.search(r'\b(?:' + '|'.join(options) + r')\b', llm_output, re.IGNORECASE)
        if match:
            # Return the canonical option (from the options list) instead of the matched text
            matched_text = match.group(0).lower()
            for option in options:
                if option.lower() == matched_text:
                    return option
            return match.group(0)  # fallback

        raise ValueError("Cannot find any of the specified options in the LLM output.")

    def _request_enumerable_llm_message(self, options:list):
        options_list_as_string = ', '.join([f"'{o}'" for o in options])
        return {"role": "user",
                "content": f"The `value` field you generate **must** be exactly one of the following strings: {options_list_as_string}. This is critical for later processing."}

    def _coerce_to_dict_or_list(self, llm_output:str):
        """
        Coerces the LLM output to a list or dictionary, i.e., a JSON structure.

        This method looks for a JSON object in the LLM output, such that
          - the JSON object is considered;
          - if no JSON object is found, the method raises an error. So it is important that the prompts actually requests a JSON object. 

        Args:
            llm_output (str): The LLM output to coerce.

        Returns:
            The dictionary value of the LLM output.
        """

        # if the LLM output is already a dictionary or list, we return it
        if isinstance(llm_output, (dict, list)):
            return llm_output

        try:
            result = utils.extract_json(llm_output)
            # extract_json returns {} on failure, but we need dict or list
            if result == {} and not (isinstance(llm_output, str) and ('{}' in llm_output or '{' in llm_output and '}' in llm_output)):
                raise ValueError("Cannot convert the LLM output to a dict or list value.")
            # Check if result is actually dict or list
            if not isinstance(result, (dict, list)):
                raise ValueError("Cannot convert the LLM output to a dict or list value.")
            return result
        except Exception:
            raise ValueError("Cannot convert the LLM output to a dict or list value.")

    def _request_dict_llm_message(self):
            return {"role": "user",
                    "content": "The `value` field you generate **must** be a JSON structure embedded in a string. This is critical for later processing."}

    def _request_list_of_dict_llm_message(self):
            return {"role": "user",
                    "content": "The `value` field you generate **must** be a list of dictionaries, specified as a JSON structure embedded in a string. For example, `[\{...\}, \{...\}, ...]`. This is critical for later processing."}

    def _coerce_to_list(self, llm_output:str):
        """
        Coerces the LLM output to a list.

        This method looks for a list in the LLM output, such that
          - the list is considered;
          - if no list is found, the method raises an error. So it is important that the prompts actually requests a list. 

        Args:
            llm_output (str): The LLM output to coerce.

        Returns:
            The list value of the LLM output.
        """

        # if the LLM output is already a list, we return it
        if isinstance(llm_output, list):
            return llm_output

        # must make sure there's actually a list. Let's start with regex
        import re
        match = re.search(r'\[.*\]', llm_output)
        if match:
            return json.loads(match.group(0))

        raise ValueError("Cannot convert the LLM output to a list.")

    def _request_list_llm_message(self):
        return {"role": "user",
                "content": "The `value` field you generate **must** be a JSON **list** (e.g., [\"apple\", 1, 0.9]), NOT a dictionary, always embedded in a string. This is critical for later processing."}

    def __repr__(self):
        return f"LLMChat(messages={self.messages}, model_params={self.model_params})"


def llm(enable_json_output_format:bool=True, enable_justification_step:bool=True, enable_reasoning_step:bool=False, **model_overrides):
    """
    Decorator that turns the decorated function into an LLM-based function.
    The decorated function must either return a string (the instruction to the LLM)
    or a one-argument function that will be used to post-process the LLM response.

    If the function returns a string, the function's docstring will be used as the system prompt,
    and the returned string will be used as the user prompt. If the function returns a function,
    the parameters of the function will be used instead as the system instructions to the LLM,
    and the returned function will be used to post-process the LLM response.


    The LLM response is coerced to the function's annotated return type, if present.

    Usage example:
        @llm(model="gpt-4-0613", temperature=0.5, max_tokens=100)
        def joke():
            return "Tell me a joke."
    
    Usage example with post-processing:
        @llm()
        def unique_joke_list():
            \"\"\"Creates a list of unique jokes.\"\"\"
            return lambda x: list(set(x.split("\n")))
    
    """
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            result = func(*args, **kwargs)
            sig = inspect.signature(func)
            return_type = sig.return_annotation if sig.return_annotation != inspect.Signature.empty else str
            postprocessing_func = lambda x: x # by default, no post-processing
            
            system_prompt = "You are an AI system that executes a computation as defined below.\n\n"
            if func.__doc__ is not None:
                system_prompt += func.__doc__.strip() 
            
            #
            # Setup user prompt
            #
            if isinstance(result, str):
                user_prompt = "EXECUTE THE INSTRUCTIONS BELOW:\n\n " + result
            
            else:
                # if there's a parameter named "self" in the function signature, remove it from args
                if "self" in sig.parameters:
                    args = args[1:]
                
                # TODO obsolete?
                #
                # if we are relying on parameters, they must be named
                #if len(args) > 0:
                #    raise ValueError("Positional arguments are not allowed in LLM-based functions whose body does not return a string.")               

                user_prompt  = f"Execute your computation as best as you can using the following input parameter values.\n\n"
                user_prompt += f" ## Unnamed parameters\n{json.dumps(args, indent=4)}\n\n" 
                user_prompt += f" ## Named parameters\n{json.dumps(kwargs, indent=4)}\n\n" 
            
            #
            # Set the post-processing function if the function returns a function
            #
            if inspect.isfunction(result):
                # uses the returned function as a post-processing function
                postprocessing_func = result
            
            
            llm_req = LLMChat(system_prompt=system_prompt,
                              user_prompt=user_prompt,
                              output_type=return_type,
                              enable_json_output_format=enable_json_output_format,
                              enable_justification_step=enable_justification_step,
                              enable_reasoning_step=enable_reasoning_step,
                              **model_overrides)
            
            llm_result = postprocessing_func(llm_req.call())
            
            return llm_result
        return wrapper
    return decorator

################################################################################	
# Model output utilities
################################################################################
def extract_json(text: str) -> dict:
    """
    Extracts a JSON object from a string, ignoring: any text before the first 
    opening curly brace; and any Markdown opening (```json) or closing(```) tags.
    """
    try:
        logger.debug(f"Extracting JSON from text: {text}")

        # if it already is a dictionary or list, return it
        if isinstance(text, dict) or isinstance(text, list):

            # validate that all the internal contents are indeed JSON-like
            try:
                json.dumps(text)
            except Exception as e:
                logger.error(f"Error occurred while validating JSON: {e}. Input text: {text}.")
                return {}

            logger.debug(f"Text is already a dictionary. Returning it.")
            return text

        filtered_text = ""

        # remove any text before the first opening curly or square braces, using regex. Leave the braces.
        filtered_text = re.sub(r'^.*?({|\[)', r'\1', text, flags=re.DOTALL)

        # remove any trailing text after the LAST closing curly or square braces, using regex. Leave the braces.
        filtered_text  =  re.sub(r'(}|\])(?!.*(\]|\})).*$', r'\1', filtered_text, flags=re.DOTALL)
        
        # remove invalid escape sequences, which show up sometimes
        filtered_text = re.sub("\\'", "'", filtered_text) # replace \' with just '
        filtered_text = re.sub("\\,", ",", filtered_text)

        # parse the final JSON in a robust manner, to account for potentially messy LLM outputs
        try:
            # First try standard JSON parsing
            # use strict=False to correctly parse new lines, tabs, etc.
            parsed = json.loads(filtered_text, strict=False)
        except json.JSONDecodeError:
            # If JSON parsing fails, try ast.literal_eval which accepts single quotes
            try:
                parsed = ast.literal_eval(filtered_text)
                logger.debug("Used ast.literal_eval as fallback for single-quoted JSON-like text")
            except:
                # If both fail, try converting single quotes to double quotes and parse again
                # Replace single-quoted keys and values with double quotes, without using look-behind
                # This will match single-quoted strings that are keys or values in JSON-like structures
                # It may not be perfect for all edge cases, but works for most LLM outputs
                converted_text = re.sub(r"'([^']*)'", r'"\1"', filtered_text)
                parsed = json.loads(converted_text, strict=False)
                logger.debug("Converted single quotes to double quotes before parsing")
        
        # return the parsed JSON object
        return parsed
    
    except Exception as e:
        logger.error(f"Error occurred while extracting JSON: {e}. Input text: {text}. Filtered text: {filtered_text}")
        return {}

def extract_code_block(text: str) -> str:
    """
    Extracts a code block from a string, ignoring any text before the first 
    opening triple backticks and any text after the closing triple backticks.
    """
    try:
        # remove any text before the first opening triple backticks, using regex. Leave the backticks.
        text = re.sub(r'^.*?(```)', r'\1', text, flags=re.DOTALL)

        # remove any trailing text after the LAST closing triple backticks, using regex. Leave the backticks.
        text  =  re.sub(r'(```)(?!.*```).*$', r'\1', text, flags=re.DOTALL)
        
        return text
    
    except Exception:
        return ""

################################################################################
# Model control utilities
################################################################################    

def repeat_on_error(retries:int, exceptions:list):
    """
    Decorator that repeats the specified function call if an exception among those specified occurs, 
    up to the specified number of retries. If that number of retries is exceeded, the
    exception is raised. If no exception occurs, the function returns normally.

    Args:
        retries (int): The number of retries to attempt.
        exceptions (list): The list of exception classes to catch.
    """
    def decorator(func):
        def wrapper(*args, **kwargs):
            for i in range(retries):
                try:
                    return func(*args, **kwargs)
                except tuple(exceptions) as e:
                    logger.debug(f"Exception occurred: {e}")
                    if i == retries - 1:
                        raise e
                    else:
                        logger.debug(f"Retrying ({i+1}/{retries})...")
                        continue
        return wrapper
    return decorator

 
def try_function(func, postcond_func=None, retries=5, exceptions=[Exception]):

    @repeat_on_error(retries=retries, exceptions=exceptions)
    def aux_apply_func():
        logger.debug(f"Trying function {func.__name__}...")
        result = func()
        logger.debug(f"Result of function {func.__name__}: {result}")
        
        if postcond_func is not None:
            if not postcond_func(result):
                # must raise an exception if the postcondition is not met.
                raise ValueError(f"Postcondition not met for function {func.__name__}!")

        return result
    
    return aux_apply_func()
   
################################################################################
# Prompt engineering
################################################################################
def add_rai_template_variables_if_enabled(template_variables: dict) -> dict:
    """
    Adds the RAI template variables to the specified dictionary, if the RAI disclaimers are enabled.
    These can be configured in the config.ini file. If enabled, the variables will then load the RAI disclaimers from the 
    appropriate files in the prompts directory. Otherwise, the variables will be set to None.

    Args:
        template_variables (dict): The dictionary of template variables to add the RAI variables to.

    Returns:
        dict: The updated dictionary of template variables.
    """

    from tinytroupe import config # avoids circular import
    rai_harmful_content_prevention = config["Simulation"].getboolean(
        "RAI_HARMFUL_CONTENT_PREVENTION", True 
    )
    rai_copyright_infringement_prevention = config["Simulation"].getboolean(
        "RAI_COPYRIGHT_INFRINGEMENT_PREVENTION", True
    )

    # Harmful content
    with open(os.path.join(os.path.dirname(__file__), "prompts/rai_harmful_content_prevention.md"), "r", encoding="utf-8", errors="replace") as f:
        rai_harmful_content_prevention_content = f.read()

    template_variables['rai_harmful_content_prevention'] = rai_harmful_content_prevention_content if rai_harmful_content_prevention else None

    # Copyright infringement
    with open(os.path.join(os.path.dirname(__file__), "prompts/rai_copyright_infringement_prevention.md"), "r", encoding="utf-8", errors="replace") as f:
        rai_copyright_infringement_prevention_content = f.read()

    template_variables['rai_copyright_infringement_prevention'] = rai_copyright_infringement_prevention_content if rai_copyright_infringement_prevention else None

    return template_variables


################################################################################
# Truncation
################################################################################

def truncate_actions_or_stimuli(list_of_actions_or_stimuli: Collection[dict], max_content_length: int) -> Collection[str]:
    """
    Truncates the content of actions or stimuli at the specified maximum length. Does not modify the original list.

    Args:
        list_of_actions_or_stimuli (Collection[dict]): The list of actions or stimuli to truncate.
        max_content_length (int): The maximum length of the content.

    Returns:
        Collection[str]: The truncated list of actions or stimuli. It is a new list, not a reference to the original list, 
        to avoid unexpected side effects.
    """
    cloned_list = copy.deepcopy(list_of_actions_or_stimuli)
    
    for element in cloned_list:
        # the external wrapper of the LLM message: {'role': ..., 'content': ...}
        if "content" in element and "role" in element and element["role"] != "system":
            msg_content = element["content"] 

            # now the actual action or stimulus content

            # has action, stimuli or stimulus as key?
            if isinstance(msg_content, dict):
                if "action" in msg_content:
                    # is content there?
                    if "content" in msg_content["action"]:
                        msg_content["action"]["content"] = break_text_at_length(msg_content["action"]["content"], max_content_length)
                elif "stimulus" in msg_content:
                    # is content there?
                    if "content" in msg_content["stimulus"]:
                        msg_content["stimulus"]["content"] = break_text_at_length(msg_content["stimulus"]["content"], max_content_length)
                elif "stimuli" in msg_content:
                    # for each element in the list
                    for stimulus in msg_content["stimuli"]:
                        # is content there?
                        if "content" in stimulus:
                            stimulus["content"] = break_text_at_length(stimulus["content"], max_content_length)

        # if no condition was met, we just ignore it. It is not an action or a stimulus.
    
    return cloned_list