models/business_logic_utils/response_processing.py

import re
import random
from .config import AI_PHRASES
from .response_generation import generate_sim

def parse_model_response(response: dict, name: str = "") -> str:
    """
    Parse the LLM response to extract the assistant's message and apply initial post-processing.

    Args:
        response (dict): The raw response from the LLM.
        name (str, optional): Name to strip from the beginning of the text. Defaults to "".

    Returns:
        str: The cleaned and parsed assistant's message.
    """
    assistant_message = response["choices"][0]["message"]["content"]
    cleaned_text = postprocess_text(
        assistant_message, 
        name=name, 
        human_prefix="user:",
        assistant_prefix="assistant:"
    )
    return cleaned_text

def postprocess_text(
    text: str,
    name: str = "",
    human_prefix: str = "user:",
    assistant_prefix: str = "assistant:",
    strip_name: bool = True
) -> str:
    """Eliminates whispers, reactions, ellipses, and quotation marks from generated text by LLMs.

    Args:
        text (str): The text to process.
        name (str, optional): Name to strip from the beginning of the text. Defaults to "".
        human_prefix (str, optional): The user prefix to remove. Defaults to "user:".
        assistant_prefix (str, optional): The assistant prefix to remove. Defaults to "assistant:".
        strip_name (bool, optional): Whether to remove the name at the beginning of the text. Defaults to True.

    Returns:
        str: Cleaned text.
    """
    if text:
        # Replace ellipses with a single period
        text = re.sub(r'\.\.\.', '.', text)

        # Remove unnecessary role prefixes
        text = text.replace(human_prefix, "").replace(assistant_prefix, "")

        # Remove whispers or other marked reactions
        whispers = re.compile(r"(\([\w\s]+\))")  # remove things like "(whispers)"
        reactions = re.compile(r"(\*[\w\s]+\*)")  # remove things like "*stutters*"
        text = whispers.sub("", text)
        text = reactions.sub("", text)

        # Remove all quotation marks (both single and double)
        text = text.replace('"', '').replace("'", "")

        # Normalize spaces
        text = re.sub(r"\s+", " ", text).strip()

    return text

def apply_guardrails(model_input: dict, response: str, endpoint_url: str, endpoint_bearer_token: str) -> str:
    """Apply the 'I am an AI' guardrail to model responses"""
    attempt = 0
    max_attempts = 2

    while attempt < max_attempts and contains_ai_phrase(response):
        # Regenerate the response without modifying the conversation history
        completion = generate_sim(model_input, endpoint_url, endpoint_bearer_token)
        response = parse_model_response(completion)
        attempt += 1

    if contains_ai_phrase(response):
        # Use only the last user message for regeneration
        memory = model_input['messages']
        last_user_message = next((msg for msg in reversed(memory) if msg['role'] == 'user'), None)
        if last_user_message:
            # Create a new conversation with system message and last user message
            model_input_copy = {
                **model_input,
                'messages': [memory[0], last_user_message]  # memory[0] is the system message
            }
            completion = generate_sim(model_input_copy, endpoint_url, endpoint_bearer_token)
            response = parse_model_response(completion)

    return response


def contains_ai_phrase(text: str) -> bool:
    """Check if the text contains any 'I am an AI' phrases."""
    text_lower = text.lower()
    return any(phrase.lower() in text_lower for phrase in AI_PHRASES)

def truncate_response(text: str, punctuation_marks: tuple = ('.', '!', '?', '…')) -> str:
    """
    Truncate the text at the last occurrence of a specified punctuation mark.

    Args:
        text (str): The text to truncate.
        punctuation_marks (tuple, optional): A tuple of punctuation marks to use for truncation. Defaults to ('.', '!', '?', '…').

    Returns:
        str: The truncated text.
    """
    # Find the last position of any punctuation mark from the provided set
    last_punct_position = max(text.rfind(p) for p in punctuation_marks)

    # Check if any punctuation mark is found
    if last_punct_position == -1:
        # No punctuation found, return the original text
        return text.strip()

    # Return the truncated text up to and including the last punctuation mark
    return text[:last_punct_position + 1].strip()

def split_texter_response(text: str) -> str:
    """
    Splits the texter's response into multiple messages,
    introducing '\ntexter:' prefixes after punctuation.

    The number of messages is randomly chosen based on specified probabilities:
    - 1 message: 30% chance
    - 2 messages: 25% chance
    - 3 messages: 45% chance

    The first message does not include the '\ntexter:' prefix.
    """
    # Use regex to split text into sentences, keeping the punctuation
    sentences = re.findall(r'[^.!?]+[.!?]*', text)
    # Remove empty strings from sentences
    sentences = [s.strip() for s in sentences if s.strip()]

    # Decide number of messages based on specified probabilities
    num_messages = random.choices([1, 2, 3], weights=[0.3, 0.25, 0.45], k=1)[0]

    # If not enough sentences to make the splits, adjust num_messages
    if len(sentences) < num_messages:
        num_messages = len(sentences)

    # If num_messages is 1, return the original text
    if num_messages == 1:
        return text.strip()

    # Calculate split points
    # We need to divide the sentences into num_messages parts
    avg = len(sentences) / num_messages
    split_indices = [int(round(avg * i)) for i in range(1, num_messages)]

    # Build the new text
    new_text = ''
    start = 0
    for i, end in enumerate(split_indices + [len(sentences)]):
        segment_sentences = sentences[start:end]
        segment_text = ' '.join(segment_sentences).strip()
        if i == 0:
            # First segment, do not add '\ntexter:'
            new_text += segment_text
        else:
            # Subsequent segments, add '\ntexter:'
            new_text += f"\ntexter: {segment_text}"
        start = end
    return new_text.strip()

def process_model_response(completion: dict, model_input: dict, endpoint_url: str, endpoint_bearer_token: str) -> str:
    """
    Process the raw model response, including parsing, applying guardrails,
    truncation, and splitting the response into multiple messages if necessary.

    Args:
        completion (dict): Raw response from the LLM.
        model_input (dict): The model input containing the conversation history.
        endpoint_url (str): The URL of the endpoint.
        endpoint_bearer_token (str): The authentication token for endpoint.

    Returns:
        str: Final processed response ready for the APP.
    """
    # Step 1: Parse the raw response to extract the assistant's message
    assistant_message = parse_model_response(completion)

    # Step 2: Apply guardrails (handle possible AI responses)
    guardrail_message = apply_guardrails(model_input, assistant_message, endpoint_url, endpoint_bearer_token)

    # Step 3: Apply response truncation
    truncated_message = truncate_response(guardrail_message)

    # Step 4: Split the response into multiple messages if needed
    final_response = split_texter_response(truncated_message)

    return final_response