models/business_logic_utils/input_processing.py

from .config import SCENARIOS
from .prompt_generation import get_template
from streamlit.logger import get_logger

logger = get_logger(__name__)

def parse_app_request(app_request: dict) -> tuple:
    """Parse the APP request and convert it to model_input format, returning model_input, prompt, and conversation_id."""
    inputs = app_request.get("inputs", {})

    # Extract fields
    conversation_id = inputs.get("conversation_id", [""])[0]
    ip_address = inputs.get("ip_address", [""])[0]
    prompt = inputs.get("prompt", [""])[0]
    issue = inputs.get("issue", ["full_convo"])[0]
    language = inputs.get("language", ["en"])[0]
    temperature = float(inputs.get("temperature", ["0.7"])[0])
    max_tokens = int(inputs.get("max_tokens", ["128"])[0])
    texter_name = inputs.get("texter_name", [None])[0]

    # Build the model_input dictionary without messages
    model_input = {
        "issue": issue,
        "language": language,
        "texter_name": texter_name,  # Assuming empty unless provided elsewhere
        "messages": [],
        "max_tokens": max_tokens,
        "temperature": temperature,
    }

    # Return model_input, prompt, and conversation_id
    return model_input, prompt, conversation_id

def initialize_conversation(model_input: dict, conversation_id: str) -> dict:
    """Initialize the conversation by adding the system message."""
    messages = model_input.get("messages", [])

    # Check if the first message is already a system message
    if not messages or messages[0].get('role') != 'system':
        texter_name = model_input.get("texter_name", None)
        language = model_input.get("language", "en")

        # Retrieve the scenario configuration based on 'issue'
        scenario_key = model_input["issue"]
        scenario_config = SCENARIOS.get(scenario_key)
        if not scenario_config:
            raise ValueError(f"The scenario '{scenario_key}' is not defined in SCENARIOS.")
        # Generate the system message (prompt)
        system_message_content = get_template(
            language=language, texter_name=texter_name, **scenario_config
        )
        logger.debug(f"System message is: {system_message_content}")
        system_message = {"role": "system", "content": system_message_content}

        # Insert the system message at the beginning
        messages.insert(0, system_message)

    model_input['messages'] = messages

    return model_input

def parse_prompt(
    prompt: str, 
    user_prefix: str = "helper:", 
    assistant_prefix: str = "texter:", 
    delimitator: str = "\n"
) -> list:
    """
    Parse the prompt string into a list of messages.

    - Includes an initial empty 'user' message if not present.
    - Combines consecutive messages from the same role into a single message.
    - Handles punctuation when combining messages.
    - The prefixes for user and assistant can be customized.

    Args:
        prompt (str): The conversation history string.
        user_prefix (str): Prefix for user messages (default: "helper:").
        assistant_prefix (str): Prefix for assistant messages (default: "texter:").
        delimitator (str): The delimiter used to split the prompt into lines. Defaults to "\n".

    Returns:
        list: Parsed messages in the form of dictionaries with 'role' and 'content'.
    """

    # Check if the prompt starts with the user prefix; if not, add an initial empty user message
    if not prompt.strip().startswith(user_prefix):
        prompt = f"{user_prefix}{delimitator}" + prompt

    # Split the prompt using the specified delimiter
    lines = [line.strip() for line in prompt.strip().split(delimitator) if line.strip()]

    messages = []
    last_role = None
    last_content = ""
    last_line_empty_texter = False

    for line in lines:
        if line.startswith(user_prefix):
            content = line[len(user_prefix):].strip()
            role = 'user'
            # Include 'user' messages even if content is empty
            if last_role == role:
                # Combine with previous content
                if last_content and not last_content.endswith(('...', '.', '!', '?')):
                    last_content += '.'
                last_content += f" {content}"
            else:
                # Save previous message if exists
                if last_role is not None:
                    messages.append({'role': last_role, 'content': last_content})
                last_role = role
                last_content = content
        elif line.startswith(assistant_prefix):
            content = line[len(assistant_prefix):].strip()
            role = 'assistant'
            if content:
                if last_role == role:
                    # Combine with previous content
                    if last_content and not last_content.endswith(('...', '.', '!', '?')):
                        last_content += '.'
                    last_content += f" {content}"
                else:
                    # Save previous message if exists
                    if last_role is not None:
                        messages.append({'role': last_role, 'content': last_content})
                    last_role = role
                    last_content = content
            else:
                # Empty 'texter:' line, mark for exclusion
                last_line_empty_texter = True
        else:
            # Ignore or handle unexpected lines
            pass

    # After processing all lines, add the last message if it's not an empty assistant message
    if last_role == 'assistant' and not last_content:
        # Do not add empty assistant message
        pass
    else:
        messages.append({'role': last_role, 'content': last_content})

    return messages