Spaces:
Sleeping
Sleeping
File size: 5,731 Bytes
42a7266 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
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
|