import logging import math import re from datetime import datetime from typing import Optional, Any import uuid from open_webui.utils.misc import get_last_user_message, get_messages_content from open_webui.config import DEFAULT_RAG_TEMPLATE log = logging.getLogger(__name__) def get_task_model_id( default_model_id: str, task_model: str, task_model_external: str, models ) -> str: # Set the task model task_model_id = default_model_id # Check if the user has a custom task model and use that model if models[task_model_id].get("connection_type") == "local": if task_model and task_model in models: task_model_id = task_model else: if task_model_external and task_model_external in models: task_model_id = task_model_external return task_model_id def prompt_variables_template(template: str, variables: dict[str, str]) -> str: for variable, value in variables.items(): template = template.replace(variable, value) return template def prompt_template(template: str, user: Optional[Any] = None) -> str: USER_VARIABLES = {} if user: if hasattr(user, "model_dump"): user = user.model_dump() if isinstance(user, dict): user_info = user.get("info", {}) or {} birth_date = user.get("date_of_birth") age = None if birth_date: try: # If birth_date is str, convert to datetime if isinstance(birth_date, str): birth_date = datetime.strptime(birth_date, "%Y-%m-%d") today = datetime.now() age = ( today.year - birth_date.year - ( (today.month, today.day) < (birth_date.month, birth_date.day) ) ) except Exception as e: pass USER_VARIABLES = { "name": str(user.get("name")), "email": str(user.get("email")), "location": str(user_info.get("location")), "bio": str(user.get("bio")), "gender": str(user.get("gender")), "birth_date": str(birth_date), "age": str(age), } # Get the current date current_date = datetime.now() # Format the date to YYYY-MM-DD formatted_date = current_date.strftime("%Y-%m-%d") formatted_time = current_date.strftime("%I:%M:%S %p") formatted_weekday = current_date.strftime("%A") template = template.replace("{{CURRENT_DATE}}", formatted_date) template = template.replace("{{CURRENT_TIME}}", formatted_time) template = template.replace( "{{CURRENT_DATETIME}}", f"{formatted_date} {formatted_time}" ) template = template.replace("{{CURRENT_WEEKDAY}}", formatted_weekday) template = template.replace("{{USER_NAME}}", USER_VARIABLES.get("name", "Unknown")) template = template.replace( "{{USER_EMAIL}}", USER_VARIABLES.get("email", "Unknown") ) template = template.replace("{{USER_BIO}}", USER_VARIABLES.get("bio", "Unknown")) template = template.replace( "{{USER_GENDER}}", USER_VARIABLES.get("gender", "Unknown") ) template = template.replace( "{{USER_BIRTH_DATE}}", USER_VARIABLES.get("birth_date", "Unknown") ) template = template.replace( "{{USER_AGE}}", str(USER_VARIABLES.get("age", "Unknown")) ) template = template.replace( "{{USER_LOCATION}}", USER_VARIABLES.get("location", "Unknown") ) return template def replace_prompt_variable(template: str, prompt: str) -> str: def replacement_function(match): full_match = match.group( 0 ).lower() # Normalize to lowercase for consistent handling start_length = match.group(1) end_length = match.group(2) middle_length = match.group(3) if full_match == "{{prompt}}": return prompt elif start_length is not None: return prompt[: int(start_length)] elif end_length is not None: return prompt[-int(end_length) :] elif middle_length is not None: middle_length = int(middle_length) if len(prompt) <= middle_length: return prompt start = prompt[: math.ceil(middle_length / 2)] end = prompt[-math.floor(middle_length / 2) :] return f"{start}...{end}" return "" # Updated regex pattern to make it case-insensitive with the `(?i)` flag pattern = r"(?i){{prompt}}|{{prompt:start:(\d+)}}|{{prompt:end:(\d+)}}|{{prompt:middletruncate:(\d+)}}" template = re.sub(pattern, replacement_function, template) return template def truncate_content(content: str, max_chars: int, mode: str = "middletruncate") -> str: """Truncate a string to max_chars using the specified mode. Modes: - middletruncate: keep beginning and end, join with '...' - start: keep first max_chars characters - end: keep last max_chars characters """ if not content or len(content) <= max_chars: return content if mode == "start": return content[:max_chars] elif mode == "end": return content[-max_chars:] else: # middletruncate half = max_chars // 2 return f"{content[:half]}...{content[-(max_chars - half):]}" def apply_content_filter(messages: list[dict], filter_str: str) -> list[dict]: """Apply a content filter to each message's content. filter_str is like 'middletruncate:500', 'start:200', or 'end:200'. Returns a new list with truncated content (original messages are not mutated). """ parts = filter_str.split(":") if len(parts) != 2: return messages mode = parts[0].lower() try: max_chars = int(parts[1]) except ValueError: return messages if mode not in ("middletruncate", "start", "end"): return messages result = [] for msg in messages: new_msg = dict(msg) if isinstance(new_msg.get("content"), str): new_msg["content"] = truncate_content(new_msg["content"], max_chars, mode) elif isinstance(new_msg.get("content"), list): new_content = [] for item in new_msg["content"]: if isinstance(item, dict) and item.get("type") == "text": new_item = dict(item) new_item["text"] = truncate_content( item.get("text", ""), max_chars, mode ) new_content.append(new_item) else: new_content.append(item) new_msg["content"] = new_content result.append(new_msg) return result def replace_messages_variable( template: str, messages: Optional[list[dict]] = None ) -> str: def replacement_function(match): # Groups: (1) filter for bare MESSAGES # (2) START count, (3) filter for START # (4) END count, (5) filter for END # (6) MIDDLE count,(7) filter for MIDDLE bare_filter = match.group(1) start_length = match.group(2) start_filter = match.group(3) end_length = match.group(4) end_filter = match.group(5) middle_length = match.group(6) middle_filter = match.group(7) # If messages is None, handle it as an empty list if messages is None: return "" # Select messages based on the variant if start_length is not None: selected = messages[: int(start_length)] content_filter = start_filter elif end_length is not None: selected = messages[-int(end_length) :] content_filter = end_filter elif middle_length is not None: mid = int(middle_length) if len(messages) <= mid: selected = messages else: half = mid // 2 start_msgs = messages[:half] end_msgs = messages[-half:] if mid % 2 == 0 else messages[-(half + 1) :] selected = start_msgs + end_msgs content_filter = middle_filter else: # Bare {{MESSAGES}} or {{MESSAGES|filter}} selected = messages content_filter = bare_filter # Apply content filter if present if content_filter: selected = apply_content_filter(selected, content_filter) return get_messages_content(selected) template = re.sub( r"(?:" r"\{\{MESSAGES(?:\|(\w+:\d+))?\}\}" r"|\{\{MESSAGES:START:(\d+)(?:\|(\w+:\d+))?\}\}" r"|\{\{MESSAGES:END:(\d+)(?:\|(\w+:\d+))?\}\}" r"|\{\{MESSAGES:MIDDLETRUNCATE:(\d+)(?:\|(\w+:\d+))?\}\}" r")", replacement_function, template, ) return template # {{prompt:middletruncate:8000}} def rag_template(template: str, context: str, query: str): if template.strip() == "": template = DEFAULT_RAG_TEMPLATE template = prompt_template(template) if "[context]" not in template and "{{CONTEXT}}" not in template: log.debug( "WARNING: The RAG template does not contain the '[context]' or '{{CONTEXT}}' placeholder." ) if "" in context and "" in context: log.debug( "WARNING: Potential prompt injection attack: the RAG " "context contains '' and ''. This might be " "nothing, or the user might be trying to hack something." ) query_placeholders = [] if "[query]" in context: query_placeholder = "{{QUERY" + str(uuid.uuid4()) + "}}" template = template.replace("[query]", query_placeholder) query_placeholders.append((query_placeholder, "[query]")) if "{{QUERY}}" in context: query_placeholder = "{{QUERY" + str(uuid.uuid4()) + "}}" template = template.replace("{{QUERY}}", query_placeholder) query_placeholders.append((query_placeholder, "{{QUERY}}")) template = template.replace("[context]", context) template = template.replace("{{CONTEXT}}", context) template = template.replace("[query]", query) template = template.replace("{{QUERY}}", query) for query_placeholder, original_placeholder in query_placeholders: template = template.replace(query_placeholder, original_placeholder) return template def title_generation_template( template: str, messages: list[dict], user: Optional[Any] = None ) -> str: prompt = get_last_user_message(messages) template = replace_prompt_variable(template, prompt) template = replace_messages_variable(template, messages) template = prompt_template(template, user) return template def follow_up_generation_template( template: str, messages: list[dict], user: Optional[Any] = None ) -> str: prompt = get_last_user_message(messages) template = replace_prompt_variable(template, prompt) template = replace_messages_variable(template, messages) template = prompt_template(template, user) return template def tags_generation_template( template: str, messages: list[dict], user: Optional[Any] = None ) -> str: prompt = get_last_user_message(messages) template = replace_prompt_variable(template, prompt) template = replace_messages_variable(template, messages) template = prompt_template(template, user) return template def image_prompt_generation_template( template: str, messages: list[dict], user: Optional[Any] = None ) -> str: prompt = get_last_user_message(messages) template = replace_prompt_variable(template, prompt) template = replace_messages_variable(template, messages) template = prompt_template(template, user) return template def emoji_generation_template( template: str, prompt: str, user: Optional[Any] = None ) -> str: template = replace_prompt_variable(template, prompt) template = prompt_template(template, user) return template def autocomplete_generation_template( template: str, prompt: str, messages: Optional[list[dict]] = None, type: Optional[str] = None, user: Optional[Any] = None, ) -> str: template = template.replace("{{TYPE}}", type if type else "") template = replace_prompt_variable(template, prompt) template = replace_messages_variable(template, messages) template = prompt_template(template, user) return template def query_generation_template( template: str, messages: list[dict], user: Optional[Any] = None ) -> str: prompt = get_last_user_message(messages) template = replace_prompt_variable(template, prompt) template = replace_messages_variable(template, messages) template = prompt_template(template, user) return template def moa_response_generation_template( template: str, prompt: str, responses: list[str] ) -> str: def replacement_function(match): full_match = match.group(0) start_length = match.group(1) end_length = match.group(2) middle_length = match.group(3) if full_match == "{{prompt}}": return prompt elif start_length is not None: return prompt[: int(start_length)] elif end_length is not None: return prompt[-int(end_length) :] elif middle_length is not None: middle_length = int(middle_length) if len(prompt) <= middle_length: return prompt start = prompt[: math.ceil(middle_length / 2)] end = prompt[-math.floor(middle_length / 2) :] return f"{start}...{end}" return "" template = re.sub( r"{{prompt}}|{{prompt:start:(\d+)}}|{{prompt:end:(\d+)}}|{{prompt:middletruncate:(\d+)}}", replacement_function, template, ) responses = [f'"""{response}"""' for response in responses] responses = "\n\n".join(responses) template = template.replace("{{responses}}", responses) return template def tools_function_calling_generation_template(template: str, tools_specs: str) -> str: template = template.replace("{{TOOLS}}", tools_specs) return template