krinlove/open-webui

1from pathlib import Path2import hashlib3import json4import re5from datetime import timedelta6from typing import Optional, List, Tuple7import uuid8import time9 10 11def get_last_user_message(messages: List[dict]) -> str:12    for message in reversed(messages):13        if message["role"] == "user":14            if isinstance(message["content"], list):15                for item in message["content"]:16                    if item["type"] == "text":17                        return item["text"]18            return message["content"]19    return None20 21 22def get_last_assistant_message(messages: List[dict]) -> str:23    for message in reversed(messages):24        if message["role"] == "assistant":25            if isinstance(message["content"], list):26                for item in message["content"]:27                    if item["type"] == "text":28                        return item["text"]29            return message["content"]30    return None31 32 33def get_system_message(messages: List[dict]) -> dict:34    for message in messages:35        if message["role"] == "system":36            return message37    return None38 39 40def remove_system_message(messages: List[dict]) -> List[dict]:41    return [message for message in messages if message["role"] != "system"]42 43 44def pop_system_message(messages: List[dict]) -> Tuple[dict, List[dict]]:45    return get_system_message(messages), remove_system_message(messages)46 47 48def add_or_update_system_message(content: str, messages: List[dict]):49    """50    Adds a new system message at the beginning of the messages list51    or updates the existing system message at the beginning.52 53    :param msg: The message to be added or appended.54    :param messages: The list of message dictionaries.55    :return: The updated list of message dictionaries.56    """57 58    if messages and messages[0].get("role") == "system":59        messages[0]["content"] += f"{content}\n{messages[0]['content']}"60    else:61        # Insert at the beginning62        messages.insert(0, {"role": "system", "content": content})63 64    return messages65 66 67def stream_message_template(model: str, message: str):68    return {69        "id": f"{model}-{str(uuid.uuid4())}",70        "object": "chat.completion.chunk",71        "created": int(time.time()),72        "model": model,73        "choices": [74            {75                "index": 0,76                "delta": {"content": message},77                "logprobs": None,78                "finish_reason": None,79            }80        ],81    }82 83 84def get_gravatar_url(email):85    # Trim leading and trailing whitespace from86    # an email address and force all characters87    # to lower case88    address = str(email).strip().lower()89 90    # Create a SHA256 hash of the final string91    hash_object = hashlib.sha256(address.encode())92    hash_hex = hash_object.hexdigest()93 94    # Grab the actual image URL95    return f"https://www.gravatar.com/avatar/{hash_hex}?d=mp"96 97 98def calculate_sha256(file):99    sha256 = hashlib.sha256()100    # Read the file in chunks to efficiently handle large files101    for chunk in iter(lambda: file.read(8192), b""):102        sha256.update(chunk)103    return sha256.hexdigest()104 105 106def calculate_sha256_string(string):107    # Create a new SHA-256 hash object108    sha256_hash = hashlib.sha256()109    # Update the hash object with the bytes of the input string110    sha256_hash.update(string.encode("utf-8"))111    # Get the hexadecimal representation of the hash112    hashed_string = sha256_hash.hexdigest()113    return hashed_string114 115 116def validate_email_format(email: str) -> bool:117    if email.endswith("@localhost"):118        return True119 120    return bool(re.match(r"[^@]+@[^@]+\.[^@]+", email))121 122 123def sanitize_filename(file_name):124    # Convert to lowercase125    lower_case_file_name = file_name.lower()126 127    # Remove special characters using regular expression128    sanitized_file_name = re.sub(r"[^\w\s]", "", lower_case_file_name)129 130    # Replace spaces with dashes131    final_file_name = re.sub(r"\s+", "-", sanitized_file_name)132 133    return final_file_name134 135 136def extract_folders_after_data_docs(path):137    # Convert the path to a Path object if it's not already138    path = Path(path)139 140    # Extract parts of the path141    parts = path.parts142 143    # Find the index of '/data/docs' in the path144    try:145        index_data_docs = parts.index("data") + 1146        index_docs = parts.index("docs", index_data_docs) + 1147    except ValueError:148        return []149 150    # Exclude the filename and accumulate folder names151    tags = []152 153    folders = parts[index_docs:-1]154    for idx, part in enumerate(folders):155        tags.append("/".join(folders[: idx + 1]))156 157    return tags158 159 160def parse_duration(duration: str) -> Optional[timedelta]:161    if duration == "-1" or duration == "0":162        return None163 164    # Regular expression to find number and unit pairs165    pattern = r"(-?\d+(\.\d+)?)(ms|s|m|h|d|w)"166    matches = re.findall(pattern, duration)167 168    if not matches:169        raise ValueError("Invalid duration string")170 171    total_duration = timedelta()172 173    for number, _, unit in matches:174        number = float(number)175        if unit == "ms":176            total_duration += timedelta(milliseconds=number)177        elif unit == "s":178            total_duration += timedelta(seconds=number)179        elif unit == "m":180            total_duration += timedelta(minutes=number)181        elif unit == "h":182            total_duration += timedelta(hours=number)183        elif unit == "d":184            total_duration += timedelta(days=number)185        elif unit == "w":186            total_duration += timedelta(weeks=number)187 188    return total_duration189 190 191def parse_ollama_modelfile(model_text):192    parameters_meta = {193        "mirostat": int,194        "mirostat_eta": float,195        "mirostat_tau": float,196        "num_ctx": int,197        "repeat_last_n": int,198        "repeat_penalty": float,199        "temperature": float,200        "seed": int,201        "tfs_z": float,202        "num_predict": int,203        "top_k": int,204        "top_p": float,205        "num_keep": int,206        "typical_p": float,207        "presence_penalty": float,208        "frequency_penalty": float,209        "penalize_newline": bool,210        "numa": bool,211        "num_batch": int,212        "num_gpu": int,213        "main_gpu": int,214        "low_vram": bool,215        "f16_kv": bool,216        "vocab_only": bool,217        "use_mmap": bool,218        "use_mlock": bool,219        "num_thread": int,220    }221 222    data = {"base_model_id": None, "params": {}}223 224    # Parse base model225    base_model_match = re.search(226        r"^FROM\s+(\w+)", model_text, re.MULTILINE | re.IGNORECASE227    )228    if base_model_match:229        data["base_model_id"] = base_model_match.group(1)230 231    # Parse template232    template_match = re.search(233        r'TEMPLATE\s+"""(.+?)"""', model_text, re.DOTALL | re.IGNORECASE234    )235    if template_match:236        data["params"] = {"template": template_match.group(1).strip()}237 238    # Parse stops239    stops = re.findall(r'PARAMETER stop "(.*?)"', model_text, re.IGNORECASE)240    if stops:241        data["params"]["stop"] = stops242 243    # Parse other parameters from the provided list244    for param, param_type in parameters_meta.items():245        param_match = re.search(rf"PARAMETER {param} (.+)", model_text, re.IGNORECASE)246        if param_match:247            value = param_match.group(1)248 249            try:250                if param_type == int:251                    value = int(value)252                elif param_type == float:253                    value = float(value)254                elif param_type == bool:255                    value = value.lower() == "true"256            except Exception as e:257                print(e)258                continue259 260            data["params"][param] = value261 262    # Parse adapter263    adapter_match = re.search(r"ADAPTER (.+)", model_text, re.IGNORECASE)264    if adapter_match:265        data["params"]["adapter"] = adapter_match.group(1)266 267    # Parse system description268    system_desc_match = re.search(269        r'SYSTEM\s+"""(.+?)"""', model_text, re.DOTALL | re.IGNORECASE270    )271    system_desc_match_single = re.search(272        r"SYSTEM\s+([^\n]+)", model_text, re.IGNORECASE273    )274 275    if system_desc_match:276        data["params"]["system"] = system_desc_match.group(1).strip()277    elif system_desc_match_single:278        data["params"]["system"] = system_desc_match_single.group(1).strip()279 280    # Parse messages281    messages = []282    message_matches = re.findall(r"MESSAGE (\w+) (.+)", model_text, re.IGNORECASE)283    for role, content in message_matches:284        messages.append({"role": role, "content": content})285 286    if messages:287        data["params"]["messages"] = messages288 289    return data290