import spaces import gradio as gr from pathlib import Path import re import torch import gc import os import urllib import json import inspect import ctypes import struct import sys import contextlib import logging import html import threading import time from typing import Any, Optional from dataclasses import dataclass from gradio import MessageDict from huggingface_hub import hf_hub_download, HfApi llama_cpp_module = None llama_cpp_logger_module = None Llama = None def ensure_llama_cpp_runtime(): global llama_cpp_module, llama_cpp_logger_module, Llama if llama_cpp_module is not None and Llama is not None: return llama_cpp_module, Llama import llama_cpp as imported_llama_cpp_module from llama_cpp import Llama as ImportedLlama try: import llama_cpp._logger as imported_llama_cpp_logger_module except Exception: imported_llama_cpp_logger_module = None llama_cpp_module = imported_llama_cpp_module llama_cpp_logger_module = imported_llama_cpp_logger_module Llama = ImportedLlama return llama_cpp_module, Llama LlamaCppAgent = None MessagesFormatterType = None LlamaCppPythonProvider = None BasicChatHistory = None Roles = None MessagesFormatter = None def ensure_llama_cpp_agent_runtime(): global LlamaCppAgent, MessagesFormatterType, LlamaCppPythonProvider, BasicChatHistory, Roles, MessagesFormatter if ( LlamaCppAgent is not None and MessagesFormatterType is not None and LlamaCppPythonProvider is not None and BasicChatHistory is not None and Roles is not None and MessagesFormatter is not None ): return LlamaCppAgent, MessagesFormatterType, LlamaCppPythonProvider, BasicChatHistory, Roles, MessagesFormatter from llama_cpp_agent import LlamaCppAgent as ImportedLlamaCppAgent, MessagesFormatterType as ImportedMessagesFormatterType from llama_cpp_agent.providers import LlamaCppPythonProvider as ImportedLlamaCppPythonProvider from llama_cpp_agent.chat_history import BasicChatHistory as ImportedBasicChatHistory from llama_cpp_agent.chat_history.messages import Roles as ImportedRoles from llama_cpp_agent.messages_formatter import MessagesFormatter as ImportedMessagesFormatter LlamaCppAgent = ImportedLlamaCppAgent MessagesFormatterType = ImportedMessagesFormatterType LlamaCppPythonProvider = ImportedLlamaCppPythonProvider BasicChatHistory = ImportedBasicChatHistory Roles = ImportedRoles MessagesFormatter = ImportedMessagesFormatter return LlamaCppAgent, MessagesFormatterType, LlamaCppPythonProvider, BasicChatHistory, Roles, MessagesFormatter from ja_to_danbooru.ja_to_danbooru import jatags_to_danbooru_tags import wrapt_timeout_decorator from formatter import mistral_v1_formatter, mistral_v2_formatter, mistral_v3_tekken_formatter from llmenv import llm_models, llm_models_dir, llm_loras, llm_loras_dir, llm_formats, llm_languages, dolphin_system_prompt, LLM_FORMAT_AUTO_GGUF_DEFAULT, GRADIO_DEBUG_ENV_NAME, LLMDOLPHIN_STATE_NAMESPACE import shutil llm_models_list = [] llm_loras_list = [] default_llm_model_filename = list(llm_models.keys())[0] default_llm_lora_filename = list(llm_loras.keys())[0] device = "cuda" if torch.cuda.is_available() else "cpu" HF_TOKEN = os.getenv("HF_TOKEN", False) # Runtime and storage defaults STORAGE_LIMIT_GB = 30.0 STORAGE_RESERVED_GB = 2.0 MAX_HISTORY_MESSAGES = 24 MAX_HISTORY_CHARS = 24000 DEFAULT_MAX_TOKENS = 1024 DEFAULT_TEMPERATURE = 0.7 DEFAULT_TOP_P = 0.95 DEFAULT_TOP_K = 40 DEFAULT_REPEAT_PENALTY = 1.1 DEFAULT_LORA_SCALE = 1.0 # State defaults DEFAULT_STATE = { "dolphin_sysprompt_mode": "Default", "dolphin_output_language": llm_languages[0], } # UI progress labels PROGRESS_DESC_CHECKING_REPO = "Checking repo..." PROGRESS_DESC_DOWNLOADING_MODEL = "Downloading model..." PROGRESS_DESC_UPDATING_MODEL_LIST = "Updating model list..." PROGRESS_DESC_WAITING_FOR_GPU = "Waiting for GPU..." PROGRESS_DESC_LOADING_RUNTIME = "Loading runtime..." PROGRESS_DESC_GENERATING = "Generating..." PROGRESS_DESC_FORMATTING_OUTPUT = "Formatting output..." PROGRESS_DESC_DONE = "Done." PROGRESS_DESC_LOADING_MODEL = "Loading model..." PROGRESS_DESC_MODEL_LOADED = "Model loaded." PROGRESS_DESC_LOADING_LORA = "Loading lora..." PROGRESS_DESC_LORA_LOADED = "Lora loaded." PROGRESS_DESC_PROCESSING = "Processing..." PROGRESS_DESC_TRANSLATING = "Translating..." # Hugging Face API defaults HF_API_RETRY_COUNT = 2 HF_API_RETRY_BACKOFF_SECONDS = 1.0 HF_API_TIMEOUT_SECONDS = 20.0 HF_HINT_TIMEOUT_SECONDS = 4.0 HF_MODEL_CARD_EXPAND_FIELDS = ["cardData", "baseModels", "tags"] STATE_SESSION_HASH_KEY = "request_session_hash" STRICT_HF_REPO_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,95}/[A-Za-z0-9][A-Za-z0-9._-]{0,95}$") HF_GGUF_FILE_URL_PATTERN = re.compile(r"^https://huggingface\.co/(?:datasets/|spaces/)?[A-Za-z0-9][A-Za-z0-9._-]{0,95}/[A-Za-z0-9][A-Za-z0-9._-]{0,95}/(?:blob|resolve)/[^\s?#]+/.+\.gguf(?:\?download=true)?$") STATE_SELECTED_MODEL_KEY = "selected_model" STATE_SELECTED_LORA_KEY = "selected_lora" # Registry, activity, and warning throttling MODEL_REGISTRY_LOCK = threading.RLock() ACTIVE_FILE_LOCK = threading.RLock() ACTIVE_GGUF_FILES = {} WARNING_EVENT_LOCK = threading.RLock() RECENT_WARNING_EVENTS = {} WARNING_EVENT_WINDOW_SECONDS = 120.0 FALLBACK_EVENT_WINDOW_SECONDS = 120.0 logger = logging.getLogger("llmdolphin") if not logging.getLogger().handlers: logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s") SAFE_LLAMA_LOG_CALLBACK = None LLMDOLPHIN_RUNTIME_LOCK = threading.RLock() LLMDOLPHIN_RUNTIME_INITIALIZED = False LLMDOLPHIN_DEFAULT_MODEL_WARMED = False RUNTIME_CAPABILITIES_LOCK = threading.RLock() RUNTIME_CAPABILITIES = {} GRADIO_DEBUG_ENABLED = os.getenv(GRADIO_DEBUG_ENV_NAME, "").lower() in ["1", "true", "yes", "on"] GGUF_METADATA_CACHE_LOCK = threading.RLock() GGUF_METADATA_CACHE = {} HF_MODEL_CARD_CACHE_LOCK = threading.RLock() HF_MODEL_CARD_CACHE = {} HF_REMOTE_TEXT_CACHE_LOCK = threading.RLock() HF_REMOTE_TEXT_CACHE = {} HF_MODEL_REPO_ANALYSIS_CACHE_LOCK = threading.RLock() HF_MODEL_REPO_ANALYSIS_CACHE = {} GGUF_MAGIC = b"GGUF" GGUF_VALUE_TYPE_UINT8 = 0 GGUF_VALUE_TYPE_INT8 = 1 GGUF_VALUE_TYPE_UINT16 = 2 GGUF_VALUE_TYPE_INT16 = 3 GGUF_VALUE_TYPE_UINT32 = 4 GGUF_VALUE_TYPE_INT32 = 5 GGUF_VALUE_TYPE_FLOAT32 = 6 GGUF_VALUE_TYPE_BOOL = 7 GGUF_VALUE_TYPE_STRING = 8 GGUF_VALUE_TYPE_ARRAY = 9 GGUF_VALUE_TYPE_UINT64 = 10 GGUF_VALUE_TYPE_INT64 = 11 GGUF_VALUE_TYPE_FLOAT64 = 12 FORMAT_PREFLIGHT_SAFE = "safe" FORMAT_PREFLIGHT_WARN = "warn" FORMAT_PREFLIGHT_HIGH_RISK = "high-risk" # Runtime initialization helpers def cleanup_zerogpu_offload_dir(): offload_root = Path("/data-nvme/zerogpu-offload") if not offload_root.exists(): return removed_items = 0 for child_path in offload_root.iterdir(): try: if child_path.is_dir(): shutil.rmtree(child_path) else: child_path.unlink() removed_items += 1 except Exception as error: log_event(logging.WARNING, "zerogpu_offload_cleanup_failed", path=child_path, error_type=type(error).__name__, error=error) log_event(logging.INFO, "zerogpu_offload_cleanup_completed", removed_items=removed_items) def probe_runtime_capabilities(): ensure_llama_cpp_runtime() capabilities = { "llama_init_parameters": set(), "native_chat_completion_parameters": set(), "supports_flash_attn": False, "supports_flash_attn_type": False, "supports_verbose": False, "supports_native_top_k": True, "supports_native_repeat_penalty": True, } try: init_parameters = inspect.signature(Llama.__init__).parameters init_parameter_names = set(init_parameters.keys()) capabilities["llama_init_parameters"] = init_parameter_names capabilities["supports_flash_attn"] = "flash_attn" in init_parameter_names capabilities["supports_flash_attn_type"] = "flash_attn_type" in init_parameter_names capabilities["supports_verbose"] = "verbose" in init_parameter_names except Exception as error: log_event(logging.WARNING, "llama_init_signature_probe_failed", error_type=type(error).__name__, error=error) try: completion_parameters = inspect.signature(Llama.create_chat_completion).parameters completion_parameter_names = set(completion_parameters.keys()) capabilities["native_chat_completion_parameters"] = completion_parameter_names capabilities["supports_native_top_k"] = "top_k" in completion_parameter_names capabilities["supports_native_repeat_penalty"] = "repeat_penalty" in completion_parameter_names except Exception as error: log_event(logging.WARNING, "llama_completion_signature_probe_failed", error_type=type(error).__name__, error=error) return capabilities def get_runtime_capabilities(force_refresh: bool = False): global RUNTIME_CAPABILITIES with RUNTIME_CAPABILITIES_LOCK: if RUNTIME_CAPABILITIES and not force_refresh: return RUNTIME_CAPABILITIES.copy() RUNTIME_CAPABILITIES = probe_runtime_capabilities() return RUNTIME_CAPABILITIES.copy() def warmup_default_llm_model(): try: download_llm_model(default_llm_model_filename) return True except Exception as error: log_event(logging.WARNING, "default_model_warmup_failed", model=default_llm_model_filename, error_type=type(error).__name__, error=error) return False def initialize_llmdolphin_runtime(preload_default_model: bool = True): global LLMDOLPHIN_RUNTIME_INITIALIZED, LLMDOLPHIN_DEFAULT_MODEL_WARMED should_warmup = False with LLMDOLPHIN_RUNTIME_LOCK: if not LLMDOLPHIN_RUNTIME_INITIALIZED: cleanup_zerogpu_offload_dir() LLMDOLPHIN_RUNTIME_INITIALIZED = True if preload_default_model and not LLMDOLPHIN_DEFAULT_MODEL_WARMED: LLMDOLPHIN_DEFAULT_MODEL_WARMED = True should_warmup = True if should_warmup: if not warmup_default_llm_model(): with LLMDOLPHIN_RUNTIME_LOCK: LLMDOLPHIN_DEFAULT_MODEL_WARMED = False LLAMA_NOISY_LOG_PREFIXES = ( "llama_model_loader: - kv", "llama_model_loader: - type", "load: control token:", "print_info:", ) def _is_noisy_llama_log_line(text: str) -> bool: normalized_text = str(text or "").lstrip() return normalized_text.startswith(LLAMA_NOISY_LOG_PREFIXES) def install_safe_llama_log_callback(): global SAFE_LLAMA_LOG_CALLBACK if SAFE_LLAMA_LOG_CALLBACK is not None: return ensure_llama_cpp_runtime() callback_factory = getattr(llama_cpp_module, "llama_log_callback", None) log_set = getattr(llama_cpp_module, "llama_log_set", None) if callback_factory is None or log_set is None: return ggml_level_map = getattr(llama_cpp_logger_module, "GGML_LOG_LEVEL_TO_LOGGING_LEVEL", {}) if llama_cpp_logger_module else {} @callback_factory def _safe_llama_log_callback(level: int, text: bytes, user_data: ctypes.c_void_p): log_level = ggml_level_map.get(level, logging.INFO) if level == 5 and llama_cpp_logger_module is not None: log_level = getattr(llama_cpp_logger_module, "_last_log_level", log_level) should_emit = True if llama_cpp_logger_module is not None: try: logger_level = getattr(llama_cpp_logger_module.logger, "level", logging.INFO) should_emit = logger_level <= ggml_level_map.get(level, log_level) except Exception: should_emit = True if should_emit: try: decoded = text.decode("utf-8", errors="replace") if isinstance(text, (bytes, bytearray)) else str(text) except Exception: decoded = str(text) if _is_noisy_llama_log_line(decoded): should_emit = False if should_emit: try: print(decoded, end="", flush=True, file=sys.stderr) except Exception: pass if llama_cpp_logger_module is not None: try: setattr(llama_cpp_logger_module, "_last_log_level", log_level) except Exception: pass try: log_set(_safe_llama_log_callback, ctypes.c_void_p(0)) SAFE_LLAMA_LOG_CALLBACK = _safe_llama_log_callback log_event(logging.INFO, "llama_log_callback_installed") except Exception as error: log_event(logging.WARNING, "llama_log_callback_install_failed", error_type=type(error).__name__, error=error) # GGUF metadata helpers GGUF_VALUE_STRUCT_FORMATS = { GGUF_VALUE_TYPE_UINT8: " bytes: data = file_obj.read(size) if len(data) != size: raise EOFError(f"Unexpected EOF while reading GGUF metadata: expected {size} bytes, got {len(data)}") return data def read_gguf_uint32(file_obj) -> int: return struct.unpack(" int: return struct.unpack(" str: length = read_gguf_uint64(file_obj) if length == 0: return "" return read_exact(file_obj, length).decode("utf-8", errors="replace") def read_gguf_scalar_value(file_obj, value_type: int): struct_format = GGUF_VALUE_STRUCT_FORMATS.get(value_type) if struct_format is None: raise ValueError(f"Unsupported GGUF scalar value type: {value_type}") return struct.unpack(struct_format, read_exact(file_obj, struct.calcsize(struct_format)))[0] def skip_gguf_value(file_obj, value_type: int): if value_type == GGUF_VALUE_TYPE_STRING: _ = read_gguf_string(file_obj) return if value_type == GGUF_VALUE_TYPE_ARRAY: element_type = read_gguf_uint32(file_obj) length = read_gguf_uint64(file_obj) for _ in range(length): skip_gguf_value(file_obj, element_type) return _ = read_gguf_scalar_value(file_obj, value_type) def read_gguf_value(file_obj, value_type: int): if value_type == GGUF_VALUE_TYPE_STRING: return read_gguf_string(file_obj) if value_type == GGUF_VALUE_TYPE_ARRAY: element_type = read_gguf_uint32(file_obj) length = read_gguf_uint64(file_obj) return [read_gguf_value(file_obj, element_type) for _ in range(length)] return read_gguf_scalar_value(file_obj, value_type) def read_targeted_gguf_metadata(model_path: Path, target_keys: set[str]): metadata = {} with model_path.open("rb") as file_obj: magic = read_exact(file_obj, 4) if magic != GGUF_MAGIC: raise ValueError(f"Invalid GGUF magic: {magic!r}") version = read_gguf_uint32(file_obj) if version < 2 or version > 3: raise ValueError(f"Unsupported GGUF version: {version}") _tensor_count = read_gguf_uint64(file_obj) metadata_kv_count = read_gguf_uint64(file_obj) metadata["__gguf_version__"] = version for _ in range(metadata_kv_count): key = read_gguf_string(file_obj) value_type = read_gguf_uint32(file_obj) if key in target_keys: metadata[key] = read_gguf_value(file_obj, value_type) else: skip_gguf_value(file_obj, value_type) return metadata def get_cached_gguf_metadata(model_path: Path, target_keys: set[str]): if not model_path.exists(): return {}, False cache_key = (str(model_path), model_path.stat().st_size, model_path.stat().st_mtime_ns, tuple(sorted(target_keys))) with GGUF_METADATA_CACHE_LOCK: cached = GGUF_METADATA_CACHE.get(cache_key) if cached is not None: return cached.copy(), True metadata = read_targeted_gguf_metadata(model_path, target_keys) with GGUF_METADATA_CACHE_LOCK: GGUF_METADATA_CACHE.clear() GGUF_METADATA_CACHE[cache_key] = metadata.copy() return metadata, False def get_local_gguf_preflight_metadata(filename: str): target_keys = {"general.architecture", "general.name", "tokenizer.chat_template"} model_path = get_model_file_path(filename) if not model_path.exists(): return {"__metadata_status__": "missing", "__metadata_cached__": False} try: metadata, cached = get_cached_gguf_metadata(model_path, target_keys) metadata["__metadata_status__"] = "ok" metadata["__metadata_cached__"] = cached return metadata except Exception as error: log_rate_limited_event( logging.WARNING, "gguf_metadata_read_failed", dedup_key=("gguf_metadata_read_failed", str(model_path), type(error).__name__), model=filename, path=model_path, error_type=type(error).__name__, error=error, ) return {"__metadata_status__": "error", "__metadata_error__": f"{type(error).__name__}: {error}", "__metadata_cached__": False} def normalize_repo_id(candidate: Any): if candidate is None: return "" repo_id = str(candidate or "").strip() if not repo_id: return "" matched_links = re.findall(r'^https?://huggingface\.co/(?!datasets/|spaces/)([^/#?\s]+/[^/#?\s]+)', repo_id) if matched_links: repo_id = matched_links[0] repo_id = repo_id.strip().strip("`[]()<>.,;:") if repo_id.startswith("datasets/") or repo_id.startswith("spaces/"): return "" if repo_id.count("/") != 1: return "" if any(character.isspace() for character in repo_id): return "" return repo_id def normalize_repo_id_list(values: Any): if values is None: return [] if isinstance(values, (list, tuple, set)): items = list(values) else: items = [values] normalized_items = [] for item in items: normalized_item = normalize_repo_id(item) if normalized_item: normalized_items.append(normalized_item) return list_uniq(normalized_items) def coerce_card_data_dict(card_data: Any): if card_data is None: return {} if isinstance(card_data, dict): return card_data.copy() if hasattr(card_data, "to_dict"): try: converted = card_data.to_dict() if isinstance(converted, dict): return converted.copy() except Exception as error: log_fallback_event("card_data_to_dict_failed", error_type=type(error).__name__) if hasattr(card_data, "__dict__"): return {key: value for key, value in vars(card_data).items() if not key.startswith("_")} return {} def get_cached_remote_model_card(repo_id: str): normalized_repo_id = normalize_repo_id(repo_id) if not normalized_repo_id: return {"repo_id": "", "base_models": [], "card_data": {}, "tags": [], "model_id": "", "error": ""} with HF_MODEL_CARD_CACHE_LOCK: cached = HF_MODEL_CARD_CACHE.get(normalized_repo_id) if cached is not None: return cached.copy() result = {"repo_id": normalized_repo_id, "base_models": [], "card_data": {}, "tags": [], "model_id": normalized_repo_id, "error": ""} info = None api = HfApi(token=HF_TOKEN) last_error = None expand_candidates = (HF_MODEL_CARD_EXPAND_FIELDS, ["cardData", "baseModels"], None) for expand in expand_candidates: try: kwargs = {"repo_id": normalized_repo_id, "timeout": HF_HINT_TIMEOUT_SECONDS, "token": HF_TOKEN} if expand is not None: kwargs["expand"] = expand info = api.model_info(**kwargs) break except Exception as error: last_error = error if info is None: if last_error is not None: result["error"] = f"{type(last_error).__name__}: {last_error}" log_fallback_event("remote_model_card_unavailable", repo_id=normalized_repo_id, error_type=type(last_error).__name__, soft_fail=True) with HF_MODEL_CARD_CACHE_LOCK: HF_MODEL_CARD_CACHE[normalized_repo_id] = result.copy() return result card_data = coerce_card_data_dict(getattr(info, "card_data", None) or getattr(info, "cardData", None)) base_models = normalize_repo_id_list(getattr(info, "base_models", None) or getattr(info, "baseModels", None)) base_models.extend(normalize_repo_id_list(card_data.get("base_model") or card_data.get("baseModel"))) tokenizer_block = card_data.get("tokenizer") if isinstance(tokenizer_block, dict): base_models.extend(normalize_repo_id_list(tokenizer_block.get("source"))) tags = getattr(info, "tags", None) if not isinstance(tags, list): tags = card_data.get("tags") if isinstance(card_data.get("tags"), list) else [] result = { "repo_id": normalized_repo_id, "base_models": list_uniq(base_models), "card_data": card_data, "tags": [str(tag) for tag in tags if tag], "model_id": str(getattr(info, "id", normalized_repo_id) or normalized_repo_id), "error": "", } with HF_MODEL_CARD_CACHE_LOCK: HF_MODEL_CARD_CACHE[normalized_repo_id] = result.copy() return result def get_cached_remote_text_file(repo_id: str, filename: str): normalized_repo_id = normalize_repo_id(repo_id) normalized_filename = str(filename or "").strip() if not normalized_repo_id or not normalized_filename: return "" cache_key = (normalized_repo_id, normalized_filename) with HF_REMOTE_TEXT_CACHE_LOCK: cached = HF_REMOTE_TEXT_CACHE.get(cache_key) if cached is not None: return cached try: path = hf_hub_download(repo_id=normalized_repo_id, filename=normalized_filename, repo_type="model", token=HF_TOKEN) text = Path(path).read_text(encoding="utf-8", errors="replace") except Exception as error: text = "" log_fallback_event("remote_text_file_unavailable", repo_id=normalized_repo_id, filename=normalized_filename, error_type=type(error).__name__, soft_fail=True) with HF_REMOTE_TEXT_CACHE_LOCK: HF_REMOTE_TEXT_CACHE[cache_key] = text return text def get_cached_remote_tokenizer_chat_template(repo_id: str): tokenizer_config_text = get_cached_remote_text_file(repo_id, "tokenizer_config.json") if not tokenizer_config_text: return "" try: tokenizer_config = json.loads(tokenizer_config_text) except Exception as error: log_fallback_event("tokenizer_config_parse_failed", repo_id=repo_id, error_type=type(error).__name__) return "" chat_template = tokenizer_config.get("chat_template") return str(chat_template or "") def extract_repo_id_candidates_from_text(text: str): if not text: return [] candidates = [] line_patterns = [ r'(?im)^\s*(?:original\s+model|base\s+model|source\s+model|parent\s+model)\s*:\s*`?([A-Za-z0-9][\w.-]*/[\w.-]+)`?', r'(?im)^\s*base_model\s*:\s*([A-Za-z0-9][\w.-]*/[\w.-]+)\s*$', r'(?ims)^\s*tokenizer\s*:\s*$\r?\n+\s*source\s*:\s*([A-Za-z0-9][\w.-]*/[\w.-]+)\s*$', ] for pattern in line_patterns: candidates.extend(re.findall(pattern, text)) candidates.extend(re.findall(r'https?://huggingface\.co/(?!datasets/|spaces/)([^/#?\s]+/[^/#?\s]+)', text)) normalized_candidates = [] for candidate in candidates: normalized_candidate = normalize_repo_id(candidate) if normalized_candidate: normalized_candidates.append(normalized_candidate) return list_uniq(normalized_candidates) def infer_original_repo(repo_id: str, include_readme: bool = False): normalized_repo_id = normalize_repo_id(repo_id) if not normalized_repo_id: return {"repo_id": "", "source": "", "confidence": ""} model_card = get_cached_remote_model_card(normalized_repo_id) preferred_candidates = [] for candidate in model_card.get("base_models", []): normalized_candidate = normalize_repo_id(candidate) if normalized_candidate and normalized_candidate != normalized_repo_id: preferred_candidates.append((normalized_candidate, "base_models", "high")) for normalized_candidate, source, confidence in preferred_candidates: if "gguf" not in normalized_candidate.lower(): return {"repo_id": normalized_candidate, "source": source, "confidence": confidence} if preferred_candidates: normalized_candidate, source, confidence = preferred_candidates[0] return {"repo_id": normalized_candidate, "source": source, "confidence": confidence} if include_readme: readme_text = get_cached_remote_text_file(normalized_repo_id, "README.md") for candidate in extract_repo_id_candidates_from_text(readme_text): if candidate != normalized_repo_id and "gguf" not in candidate.lower(): return {"repo_id": candidate, "source": "gguf_readme", "confidence": "medium"} return {"repo_id": "", "source": "", "confidence": ""} def make_format_candidate(format_value, source: str, confidence: str): if format_value is None: return None format_name = get_key_from_value(llm_formats, format_value) if not format_name: return None return {"format_value": format_value, "format_name": format_name, "source": str(source or ""), "confidence": str(confidence or "")} def map_format_hint_to_value(hint_text: str, family_hint_text: str = ""): normalized_hint = re.sub(r'[_\-]+', ' ', str(hint_text or '').lower()) normalized_family_hint = re.sub(r'[_\-]+', ' ', str(family_hint_text or '').lower()) combined = f"{normalized_hint} {normalized_family_hint}".strip() if not combined: return None if "qwen" in combined: return llm_formats.get("OPEN CHAT") if re.search(r'llama\s*3(?:\.[0-9]+)?', combined): return llm_formats.get("LLAMA 3") if "phi 3" in combined: return llm_formats.get("PHI 3") if "chatml" in combined: return llm_formats.get("CHATML") if "mistral tokenizer v3" in combined or "tekken" in combined: return llm_formats.get("Mistral Tokenizer V3 - Tekken") if "mistral tokenizer v2" in combined: return llm_formats.get("Mistral Tokenizer V2") if "mistral tokenizer v1" in combined: return llm_formats.get("Mistral Tokenizer V1") if "mistral nemo" in combined or "mistral" in combined: return llm_formats.get("MISTRAL") if "llama 2" in combined or "llama2" in combined: return llm_formats.get("LLAMA 2") if "open chat" in combined or "openchat" in combined: return llm_formats.get("OPEN CHAT") if "vicuna" in combined: return llm_formats.get("VICUNA") if "neural chat" in combined or "neuralchat" in combined: return llm_formats.get("NEURAL CHAT") if "solar" in combined: return llm_formats.get("SOLAR") if "synthia" in combined: return llm_formats.get("SYNTHIA") if "alpaca" in combined: return llm_formats.get("ALPACA") if "gemma 2" in combined or "gemma2" in combined: return llm_formats.get("Gemma 2") if "gemma" in combined and (" instruct" in combined or combined.endswith(" instruct") or " it" in combined or combined.endswith(" it")): return llm_formats.get("ALPACA") return None def infer_format_candidate_from_explicit_text(text: str, source: str, family_hint_text: str = ""): if not text: return None explicit_patterns = [ r'(?im)^\s*(?:prompt|message|chat)\s*format\s*[:=-]\s*([^\r\n]+)', r'(?im)^\s*template\s*[:=-]\s*([^\r\n]+)', ] for pattern in explicit_patterns: matches = re.findall(pattern, text) for matched in matches: format_value = map_format_hint_to_value(matched, family_hint_text=family_hint_text) if format_value is not None: return make_format_candidate(format_value, source, "high") return None def infer_format_candidate_from_chat_template_text(chat_template_text: str, source: str, family_hint_text: str = ""): normalized_template = str(chat_template_text or "").lower() if not normalized_template: return None normalized_family_hint = str(family_hint_text or "").lower() if "<|start_header_id|>" in normalized_template or "<|eot_id|>" in normalized_template: return make_format_candidate(llm_formats.get("LLAMA 3"), source, "high") if "" in normalized_template and "" in normalized_template: return make_format_candidate(llm_formats.get("Gemma 2"), source, "medium") if "<|im_start|>" in normalized_template or "<|im_end|>" in normalized_template: if "qwen" in normalized_family_hint: return make_format_candidate(llm_formats.get("OPEN CHAT"), source, "high") return make_format_candidate(llm_formats.get("CHATML"), source, "high") if "[inst]" in normalized_template and "[/inst]" in normalized_template: if "mistral" in normalized_family_hint: return make_format_candidate(llm_formats.get("MISTRAL"), source, "medium") return make_format_candidate(llm_formats.get("LLAMA 2"), source, "medium") if "<|user|>" in normalized_template and "<|assistant|>" in normalized_template: return make_format_candidate(llm_formats.get("OPEN CHAT"), source, "medium") return None def infer_format_candidate_from_family_hints(hint_text: str, source: str, confidence: str = "medium"): format_value = map_format_hint_to_value(hint_text) return make_format_candidate(format_value, source, confidence) def get_cached_model_repo_analysis(filename: str, metadata: Optional[dict] = None, include_format_hints: bool = False): normalized_metadata = metadata if isinstance(metadata, dict) else {} repo_id = normalize_repo_id(get_registered_model_repo_id(filename)) model_name_hint = str(normalized_metadata.get("general.name") or filename or "").strip() cache_key = (repo_id, str(filename or ""), model_name_hint, bool(include_format_hints)) with HF_MODEL_REPO_ANALYSIS_CACHE_LOCK: cached = HF_MODEL_REPO_ANALYSIS_CACHE.get(cache_key) if cached is not None: return cached.copy() analysis = { "registered_repo_id": repo_id, "original_repo_id": "", "original_repo_source": "", "original_repo_confidence": "", "inferred_format_value": None, "inferred_format_name": "", "inferred_format_source": "", "inferred_format_confidence": "", } if repo_id: original_repo = infer_original_repo(repo_id, include_readme=include_format_hints) analysis["original_repo_id"] = original_repo.get("repo_id", "") analysis["original_repo_source"] = original_repo.get("source", "") analysis["original_repo_confidence"] = original_repo.get("confidence", "") if include_format_hints: family_hint_parts = [filename, model_name_hint, repo_id, analysis["original_repo_id"]] family_hint_text = " ".join(str(part or "") for part in family_hint_parts if part) format_candidate = None if repo_id: gguf_model_card = get_cached_remote_model_card(repo_id) gguf_card_data = gguf_model_card.get("card_data", {}) gguf_hint_parts = [ family_hint_text, gguf_model_card.get("model_id", ""), gguf_card_data.get("model_name", ""), " ".join(gguf_model_card.get("tags", [])), ] gguf_family_hint_text = " ".join(str(part or "") for part in gguf_hint_parts if part) gguf_readme = get_cached_remote_text_file(repo_id, "README.md") format_candidate = infer_format_candidate_from_explicit_text(gguf_readme, "gguf_readme", family_hint_text=gguf_family_hint_text) if format_candidate is None: format_candidate = infer_format_candidate_from_family_hints(gguf_family_hint_text, "gguf_repo", confidence="low") original_repo_id = analysis["original_repo_id"] if original_repo_id: original_model_card = get_cached_remote_model_card(original_repo_id) original_card_data = original_model_card.get("card_data", {}) tokenizer_block = original_card_data.get("tokenizer") tokenizer_source = "" if isinstance(tokenizer_block, dict): tokenizer_source = normalize_repo_id(tokenizer_block.get("source")) original_hint_parts = [ family_hint_text, original_model_card.get("model_id", ""), original_card_data.get("model_name", ""), tokenizer_source, " ".join(original_model_card.get("tags", [])), ] original_family_hint_text = " ".join(str(part or "") for part in original_hint_parts if part) original_readme = get_cached_remote_text_file(original_repo_id, "README.md") original_tokenizer_template = get_cached_remote_tokenizer_chat_template(original_repo_id) original_candidates = [ infer_format_candidate_from_explicit_text(original_readme, "original_readme", family_hint_text=original_family_hint_text), infer_format_candidate_from_chat_template_text(original_tokenizer_template, "original_tokenizer", family_hint_text=original_family_hint_text), infer_format_candidate_from_family_hints(original_family_hint_text, "original_repo", confidence="medium"), ] for original_candidate in original_candidates: if original_candidate is not None: format_candidate = original_candidate break if format_candidate is None: local_hint_text = " ".join(part for part in [filename, model_name_hint] if part) format_candidate = infer_format_candidate_from_family_hints(local_hint_text, "model_name", confidence="low") if format_candidate is not None: analysis["inferred_format_value"] = format_candidate["format_value"] analysis["inferred_format_name"] = format_candidate["format_name"] analysis["inferred_format_source"] = format_candidate["source"] analysis["inferred_format_confidence"] = format_candidate["confidence"] with HF_MODEL_REPO_ANALYSIS_CACHE_LOCK: HF_MODEL_REPO_ANALYSIS_CACHE[cache_key] = analysis.copy() return analysis def should_accept_inferred_runtime_format(confidence: str): return str(confidence or "").lower() in {"high", "medium"} def resolve_runtime_inferred_format(filename: str, metadata: Optional[dict] = None): repo_analysis = get_cached_model_repo_analysis(filename, metadata=metadata, include_format_hints=True) inferred_format_value = repo_analysis.get("inferred_format_value") inferred_format_confidence = repo_analysis.get("inferred_format_confidence", "") if inferred_format_value is None or not should_accept_inferred_runtime_format(inferred_format_confidence): return None return repo_analysis def resolve_template_source(filename: str, state: Optional[dict] = None, forced_chat_template: Optional[str] = None): metadata = get_local_gguf_preflight_metadata(filename) metadata_template = metadata.get("tokenizer.chat_template") metadata_status = metadata.get("__metadata_status__") normalized_state = ensure_state_dict(state) override_llm_format = get_optional_state(normalized_state, "override_llm_format") registered_format_value = get_registered_model_format_value(filename) if forced_chat_template is not None and not is_native_gguf_default_format(forced_chat_template): return "forced", forced_chat_template, metadata if forced_chat_template is not None and is_native_gguf_default_format(forced_chat_template): if metadata_template: return "gguf_metadata", metadata_template, metadata if metadata_status == "missing": return "metadata_missing", None, metadata if metadata_status == "error": return "metadata_error", None, metadata inferred_runtime = resolve_runtime_inferred_format(filename, metadata=metadata) if inferred_runtime is not None: return "inferred_format", inferred_runtime["inferred_format_value"], metadata return "fallback_llama2", None, metadata if override_llm_format and not is_native_gguf_default_format(override_llm_format): return "override", override_llm_format, metadata if registered_format_value is not None and not is_native_gguf_default_format(registered_format_value): return "registered", registered_format_value, metadata if metadata_template: return "gguf_metadata", metadata_template, metadata if metadata_status == "missing": return "metadata_missing", None, metadata if metadata_status == "error": return "metadata_error", None, metadata inferred_runtime = resolve_runtime_inferred_format(filename, metadata=metadata) if inferred_runtime is not None: return "inferred_format", inferred_runtime["inferred_format_value"], metadata return "fallback_llama2", None, metadata def evaluate_format_preflight(filename: str, state: Optional[dict] = None, forced_chat_template: Optional[str] = None): template_source, resolved_template, metadata = resolve_template_source(filename, state=state, forced_chat_template=forced_chat_template) metadata_status = metadata.get("__metadata_status__", "missing") reason_codes = [] if template_source in ("forced", "override", "registered"): risk_level = FORMAT_PREFLIGHT_SAFE reason_codes.append("explicit_template") elif template_source == "gguf_metadata": risk_level = FORMAT_PREFLIGHT_SAFE reason_codes.append("gguf_chat_template") elif template_source == "fallback_llama2": risk_level = FORMAT_PREFLIGHT_HIGH_RISK reason_codes.append("llama2_fallback") else: risk_level = FORMAT_PREFLIGHT_WARN reason_codes.append(template_source) if not metadata.get("tokenizer.chat_template"): reason_codes.append("missing_gguf_chat_template") repo_analysis = get_cached_model_repo_analysis(filename, metadata=metadata, include_format_hints=False) if risk_level != FORMAT_PREFLIGHT_SAFE: repo_analysis = get_cached_model_repo_analysis(filename, metadata=metadata, include_format_hints=True) if repo_analysis.get("inferred_format_name"): reason_codes.append(f"inferred_format:{repo_analysis['inferred_format_source']}") if repo_analysis.get("original_repo_id"): reason_codes.append(f"original_repo:{repo_analysis['original_repo_source']}") return { "risk_level": risk_level, "template_source": template_source, "resolved_template": resolved_template, "has_chat_template": bool(metadata.get("tokenizer.chat_template")), "architecture": metadata.get("general.architecture") or "", "model_name": metadata.get("general.name") or "", "metadata_status": metadata_status, "metadata_error": metadata.get("__metadata_error__", ""), "reason_codes": reason_codes, "original_repo_id": repo_analysis.get("original_repo_id", ""), "original_repo_source": repo_analysis.get("original_repo_source", ""), "original_repo_confidence": repo_analysis.get("original_repo_confidence", ""), "inferred_format_name": repo_analysis.get("inferred_format_name", ""), "inferred_format_source": repo_analysis.get("inferred_format_source", ""), "inferred_format_confidence": repo_analysis.get("inferred_format_confidence", ""), } def format_preflight_summary(preflight: dict): architecture = preflight.get("architecture") or "unknown" risk_level = preflight.get("risk_level") or FORMAT_PREFLIGHT_WARN template_source = preflight.get("template_source") or "unknown" has_chat_template = "yes" if preflight.get("has_chat_template") else "no" parts = [ f"format preflight: **{risk_level}**", f"template source: `{template_source}`", f"gguf chat_template: `{has_chat_template}`", f"architecture: `{architecture}`", ] metadata_error = preflight.get("metadata_error") if metadata_error: parts.append(f"metadata error: `{metadata_error}`") inferred_format_name = preflight.get("inferred_format_name") if inferred_format_name: inferred_format_confidence = preflight.get("inferred_format_confidence") or "low" inferred_format_source = preflight.get("inferred_format_source") or "best-effort" parts.append(f"inferred fallback: `{inferred_format_name}` ({inferred_format_confidence}, {inferred_format_source})") return " \n".join(parts) def maybe_warn_format_preflight(filename: str, state: Optional[dict] = None, forced_chat_template: Optional[str] = None, *, context: str = "request"): preflight = evaluate_format_preflight(filename, state=state, forced_chat_template=forced_chat_template) risk_level = preflight["risk_level"] if risk_level == FORMAT_PREFLIGHT_SAFE: return preflight message = f"Format preflight ({context}): {risk_level}. template_source={preflight['template_source']} gguf_chat_template={'yes' if preflight['has_chat_template'] else 'no'}" log_rate_limited_event( logging.WARNING, "format_preflight_warning", dedup_key=("format_preflight_warning", context, filename, preflight["risk_level"], preflight["template_source"]), model=filename, context=context, risk_level=risk_level, template_source=preflight["template_source"], metadata_status=preflight["metadata_status"], ) try: gr.Warning(message) except Exception: pass return preflight # text2tag status state helpers def build_text2tag_status_state(last_used_model: str = "", last_used_format: str = "", phase: str = "Idle", queue_reason: str = ""): """Build the normalized status payload stored in gr.State for the text2tag UI.""" return { "last_used_model": str(last_used_model or ""), "last_used_format": str(last_used_format or ""), "phase": str(phase or "Idle"), "queue_reason": str(queue_reason or ""), } def ensure_text2tag_status_state(status_state: Optional[dict]): """Coerce arbitrary state values into the canonical text2tag status shape.""" if not isinstance(status_state, dict): return build_text2tag_status_state() return build_text2tag_status_state( status_state.get("last_used_model", ""), status_state.get("last_used_format", ""), status_state.get("phase", "Idle"), status_state.get("queue_reason", ""), ) # text2tag status formatting helpers def _shorten_text2tag_status_middle(value: str, *, max_length: int = 40): normalized_value = str(value or "").strip() if not normalized_value or len(normalized_value) <= max_length: return normalized_value if max_length <= 7: return normalized_value[:max_length - 1] + "…" left_length = (max_length - 1) // 2 right_length = max_length - 1 - left_length return normalized_value[:left_length] + "…" + normalized_value[-right_length:] def _format_text2tag_status_value(value: str, *, fallback: str = "—", max_length: int = 80, middle: bool = False): normalized_value = str(value or "").strip() if not normalized_value: normalized_value = fallback elif len(normalized_value) > max_length: if middle: normalized_value = _shorten_text2tag_status_middle(normalized_value, max_length=max_length) else: normalized_value = normalized_value[:max_length - 1] + "…" return html.escape(normalized_value) def _build_text2tag_status_item(label: str, value: str, *, full_value: str = ""): safe_label = html.escape(str(label or "")) safe_value = str(value or "") safe_full_value = str(full_value or safe_value) title_attr = f' title="{html.escape(safe_full_value)}"' if safe_full_value else "" return f'{safe_label}: {safe_value}' def _should_prioritize_text2tag_status_phase(phase: str): normalized_phase = str(phase or "").strip().lower() return normalized_phase.startswith(("generating", "parsing", "formatting", "downloading ", "updating model list", "auto-sending")) def _get_text2tag_status_css_class(phase: str): normalized_phase = str(phase or "").strip().lower() if _should_prioritize_text2tag_status_phase(phase): return "busy" if normalized_phase in {"ready", "cleared"}: return "ready" if normalized_phase.startswith(("model add failed", "stop requested", "auto-send skipped")): return "warning" return "idle" def _get_text2tag_queue_reason(phase: str): normalized_phase = str(phase or "").strip().lower() if normalized_phase.startswith("downloading "): return "shared GPU · download" if normalized_phase.startswith("updating model list"): return "shared GPU · model setup" if normalized_phase.startswith("generating"): return "shared GPU · running" if normalized_phase.startswith("parsing"): return "shared GPU · parsing" if normalized_phase.startswith("formatting"): return "shared GPU · formatting" if normalized_phase.startswith("auto-sending"): return "shared GPU · auto-send" return "" def _build_text2tag_model_phase(prefix: str, model_name: str, *, connector: str = " ", max_model_length: int = 34): normalized_prefix = str(prefix or "").strip() normalized_model_name = str(model_name or "").strip() if not normalized_model_name: return normalized_prefix or "Ready" shortened_model_name = _shorten_text2tag_status_middle(normalized_model_name, max_length=max_model_length) return f"{normalized_prefix}{connector}{shortened_model_name}".strip() def build_text2tag_status_bar_html(model_name: str, format_name: str, status_state: Optional[dict]): """Render the text2tag status bar from the normalized status payload.""" normalized_status_state = ensure_text2tag_status_state(status_state) raw_model_name = str(model_name or "").strip() raw_format_name = str(format_name or "").strip() raw_last_used_model = str(normalized_status_state.get("last_used_model", "") or "").strip() raw_phase = str(normalized_status_state.get("phase", "Idle") or "Idle").strip() raw_queue_reason = str(normalized_status_state.get("queue_reason", "") or "").strip() or _get_text2tag_queue_reason(raw_phase) phase = _format_text2tag_status_value(raw_phase, fallback="Idle", max_length=64, middle=True) planned_model = _format_text2tag_status_value(raw_model_name, fallback="No model selected", max_length=56, middle=True) planned_format = _format_text2tag_status_value(raw_format_name, fallback="Unknown", max_length=28, middle=True) last_used_model = _format_text2tag_status_value(raw_last_used_model, max_length=36, middle=True) queue_reason = _format_text2tag_status_value(raw_queue_reason, max_length=32, middle=True) status_items = [ _build_text2tag_status_item("Model", planned_model, full_value=raw_model_name), _build_text2tag_status_item("Status", phase, full_value=raw_phase), ] prioritize_phase = _should_prioritize_text2tag_status_phase(raw_phase) show_format = bool(raw_format_name) and raw_format_name != LLM_FORMAT_AUTO_GGUF_DEFAULT and not prioritize_phase show_last_used = bool(raw_last_used_model) and raw_last_used_model != raw_model_name and not prioritize_phase if show_format: status_items.insert(1, _build_text2tag_status_item("Fmt", planned_format, full_value=raw_format_name)) if show_last_used: status_items.insert(2 if show_format else 1, _build_text2tag_status_item("Last", last_used_model, full_value=raw_last_used_model)) if raw_queue_reason: status_items.append(_build_text2tag_status_item("Queue", queue_reason, full_value=raw_queue_reason)) status_css_class = _get_text2tag_status_css_class(raw_phase) return ( f'
' '' + ' | '.join(status_items) + '' '' '
' ) def build_text2tag_status_bar_update(model_name: str, format_name: str, status_state: Optional[dict]): """Wrap the rendered text2tag status bar in a gr.update payload.""" return gr.update(value=build_text2tag_status_bar_html(model_name, format_name, status_state)) # text2tag status transition helpers def set_text2tag_status_phase(model_name: str, format_name: str, status_state: Optional[dict], phase: str): """Apply a phase transition while keeping the normalized status shape.""" normalized_status_state = ensure_text2tag_status_state(status_state) normalized_status_state["phase"] = str(phase or "Idle") normalized_status_state["queue_reason"] = _get_text2tag_queue_reason(phase) return build_text2tag_status_bar_update(model_name, format_name, normalized_status_state), normalized_status_state def mark_text2tag_status_planned(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, "Ready") def mark_text2tag_status_generating(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, "Generating") def mark_text2tag_status_parsing(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, "Parsing") def mark_text2tag_status_formatting(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, "Formatting") def mark_text2tag_status_cleared(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, "Cleared") def mark_text2tag_status_downloading_model(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, _build_text2tag_model_phase("Downloading", model_name)) def mark_text2tag_status_downloading_model_from_paste(paste_ready: bool, model_name: str, format_name: str, status_state: Optional[dict]): if not paste_ready: return gr.skip(), ensure_text2tag_status_state(status_state) return mark_text2tag_status_downloading_model(model_name, format_name, status_state) def mark_text2tag_status_after_model_download(download_succeeded: bool, auto_send_enabled: bool, model_name: str, format_name: str, status_state: Optional[dict]): if not download_succeeded: return set_text2tag_status_phase(model_name, format_name, status_state, "Model add failed") if auto_send_enabled: return set_text2tag_status_phase(model_name, format_name, status_state, _build_text2tag_model_phase("Auto-sending", model_name, connector=" with ")) return set_text2tag_status_phase(model_name, format_name, status_state, "Ready") def mark_text2tag_status_updating_model_list(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, _build_text2tag_model_phase("Updating model list", model_name, connector=" for ")) def mark_text2tag_status_updating_model_list_from_paste(paste_ready: bool, model_name: str, format_name: str, status_state: Optional[dict]): if not paste_ready: return gr.skip(), ensure_text2tag_status_state(status_state) return mark_text2tag_status_updating_model_list(model_name, format_name, status_state) def mark_text2tag_status_after_model_download_from_paste(paste_ready: bool, download_succeeded: bool, model_name: str, format_name: str, status_state: Optional[dict]): if not paste_ready: return gr.skip(), ensure_text2tag_status_state(status_state) return mark_text2tag_status_after_model_download(download_succeeded, True, model_name, format_name, status_state) def mark_text2tag_status_after_send(model_name: str, format_name: str, status_state: Optional[dict]): """Record the last successful send and return the refreshed status payload.""" normalized_status_state = ensure_text2tag_status_state(status_state) normalized_status_state["last_used_model"] = str(model_name or "") normalized_status_state["last_used_format"] = str(format_name or "") normalized_status_state["phase"] = "Ready" normalized_status_state["queue_reason"] = "" return build_text2tag_status_bar_update(model_name, format_name, normalized_status_state), normalized_status_state def mark_text2tag_status_after_auto_send(history: list[MessageDict], download_succeeded: bool, auto_send_enabled: bool, model_name: str, format_name: str, status_state: Optional[dict]): """Resolve the post-auto-send status without changing the external state shape.""" if history: return mark_text2tag_status_after_send(model_name, format_name, status_state) if not download_succeeded: return set_text2tag_status_phase(model_name, format_name, status_state, "Model add failed") if auto_send_enabled: return set_text2tag_status_phase(model_name, format_name, status_state, "Auto-send skipped") return set_text2tag_status_phase(model_name, format_name, status_state, "Ready") def mark_text2tag_status_after_auto_send_from_paste(history: list[MessageDict], paste_ready: bool, download_succeeded: bool, model_name: str, format_name: str, status_state: Optional[dict]): if not paste_ready: return gr.skip(), ensure_text2tag_status_state(status_state) return mark_text2tag_status_after_auto_send(history, download_succeeded, True, model_name, format_name, status_state) def mark_text2tag_status_stop_requested(model_name: str, format_name: str, status_state: Optional[dict]): return set_text2tag_status_phase(model_name, format_name, status_state, "Stop requested") # Generic list helpers def to_list(text: str): return [item.strip() for item in text.split(",") if text != ""] def list_uniq(items: list): return sorted(set(items), key=items.index) # Text parsing helpers def to_list_ja(s: str): s = re.sub(r'[、。]', ',', s) return [x.strip() for x in s.split(",") if not s == ""] def is_japanese(s: str): import unicodedata for ch in s: name = unicodedata.name(ch, "") if "CJK UNIFIED" in name or "HIRAGANA" in name or "KATAKANA" in name: return True return False # State helpers def ensure_state_dict(state: Optional[dict]) -> dict: return state if isinstance(state, dict) else {} def get_llmdolphin_state_namespace(state: Optional[dict], create: bool = False): normalized_state = ensure_state_dict(state) namespace_state = normalized_state.get(LLMDOLPHIN_STATE_NAMESPACE) if isinstance(namespace_state, dict): return normalized_state, namespace_state if create: namespace_state = {} normalized_state[LLMDOLPHIN_STATE_NAMESPACE] = namespace_state return normalized_state, namespace_state return normalized_state, None def get_optional_state(state: Optional[dict], key: str): normalized_state, namespace_state = get_llmdolphin_state_namespace(state) if isinstance(namespace_state, dict) and key in namespace_state: return namespace_state.get(key) return normalized_state.get(key) def get_state_or_default(state: Optional[dict], key: str, warn_default: bool = True, warn_missing: bool = True): normalized_state = ensure_state_dict(state) optional_value = get_optional_state(normalized_state, key) if optional_value is not None or key in normalized_state: return optional_value if key in DEFAULT_STATE: if warn_default: log_rate_limited_event(logging.WARNING, "state_default_used", dedup_key=("state_default_used", key), key=key) return DEFAULT_STATE[key] if warn_missing: log_rate_limited_event(logging.WARNING, "state_missing", dedup_key=("state_missing", key), key=key) return None def require_state(state: Optional[dict], key: str): normalized_state, namespace_state = get_llmdolphin_state_namespace(state) if isinstance(namespace_state, dict) and key in namespace_state: return namespace_state[key] if key in normalized_state: return normalized_state[key] raise KeyError(f"Missing required state key: {key}") def get_state(state: Optional[dict], key: str): return get_state_or_default(state, key) def set_state(state: Optional[dict], key: str, value: Any): return write_state_value(state, key, value) def apply_selector_state_value(state: Optional[dict], key: str, value: Any): return write_state_value(state, key, value) def clear_selector_format_override(state: Optional[dict]): return clear_state_value(state, "override_llm_format") def read_state_value(state: Optional[dict], key: str, *, default: bool = True, warn_default: bool = True, warn_missing: bool = True): if default: return get_state_or_default(state, key, warn_default=warn_default, warn_missing=warn_missing) return get_optional_state(state, key) def write_state_value(state: Optional[dict], key: str, value: Any): normalized_state, namespace_state = get_llmdolphin_state_namespace(state, create=True) namespace_state[key] = value return normalized_state def clear_state_value(state: Optional[dict], key: str): return write_state_value(state, key, None) def get_request_session_hash(request: Optional[gr.Request] = None) -> str: raw_value = getattr(request, "session_hash", None) if request is not None else None return str(raw_value or "").strip() def get_session_hash_from_state(state: Optional[dict]) -> str: return str(get_optional_state(state, STATE_SESSION_HASH_KEY) or "").strip() def attach_request_session_hash(state: Optional[dict], request: Optional[gr.Request] = None): normalized_state = ensure_state_dict(state) session_hash = get_request_session_hash(request) if session_hash: normalized_state = write_state_value(normalized_state, STATE_SESSION_HASH_KEY, session_hash) return normalized_state, session_hash # Logging helpers def _format_log_value(value: Any): if value is None: return "-" if isinstance(value, Path): value = str(value) value = str(value).replace("\n", "\\n").replace("\r", "\\r") return value[:500] def log_event(level: int, event: str, **fields): suffix = " ".join(f"{key}={_format_log_value(value)}" for key, value in fields.items() if value is not None) logger.log(level, f"event={event}" + (f" {suffix}" if suffix else "")) def trace_event(event: str, **fields): if GRADIO_DEBUG_ENABLED: log_event(logging.INFO, f"trace_{event}", **fields) def log_rate_limited_event(level: int, event: str, *, dedup_key: tuple | None = None, window_seconds: float = WARNING_EVENT_WINDOW_SECONDS, **fields): if dedup_key is None: dedup_key = (event, tuple(sorted((key, _format_log_value(value)) for key, value in fields.items() if value is not None))) now = time.monotonic() with WARNING_EVENT_LOCK: previous = RECENT_WARNING_EVENTS.get(dedup_key) if previous is not None and (now - previous) < window_seconds: return False RECENT_WARNING_EVENTS[dedup_key] = now log_event(level, event, **fields) return True def raise_public_callback_error(event: str, error: Exception, **fields): log_event(logging.ERROR, event, error_type=type(error).__name__, error=error, **fields) raise gr.Error(f"Error: {error}") def resolve_system_message(system_message: Optional[str], state: Optional[dict]) -> str: return system_message if system_message is not None else get_dolphin_sysprompt(state) def log_request_started(function_name: str, model: str, lora: str, original_messages: int, kept_messages: int, state: Optional[dict] = None): log_event(logging.INFO, "dolphin_request_started", fn=function_name, model=model, lora=lora, session_hash=get_session_hash_from_state(state), original_history_messages=original_messages, kept_history_messages=kept_messages) def log_request_completed(function_name: str, started_at: float, model: str, lora: str, state: Optional[dict] = None): log_event(logging.INFO, "dolphin_request_completed", fn=function_name, model=model, lora=lora, session_hash=get_session_hash_from_state(state), duration_ms=int((time.monotonic() - started_at) * 1000)) def handle_request_exception(function_name: str, error: Exception, started_at: float, model: str, lora: str, state: Optional[dict] = None, raise_ui_error: bool = True): log_event(logging.ERROR, "dolphin_request_failed", fn=function_name, model=model, lora=lora, session_hash=get_session_hash_from_state(state), duration_ms=int((time.monotonic() - started_at) * 1000), error_type=type(error).__name__, error=error) if raise_ui_error: raise gr.Error(f"Error: {error}") def handle_selector_exception(event: str, error: Exception, **fields): raise_public_callback_error(event, error, **fields) def log_fallback_event(event: str, **fields): log_rate_limited_event(logging.WARNING, event, window_seconds=FALLBACK_EVENT_WINDOW_SECONDS, **fields) def handle_enhancer_passthrough(function_name: str, error: Exception, started_at: float, model: str, lora: str, history: list[MessageDict], message: str, state: Optional[dict] = None): handle_request_exception(function_name, error, started_at, model, lora, state=state, raise_ui_error=False) log_fallback_event("enhancer_passthrough_used", fn=function_name, model=model, lora=lora, session_hash=get_session_hash_from_state(state), error_type=type(error).__name__) history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": message}) return history, gr.update(), gr.update() @wrapt_timeout_decorator.timeout(dec_timeout=HF_API_TIMEOUT_SECONDS) # Progress helpers def update_progress(progress, value: float, desc: str): progress(value, desc=desc) def update_selector_progress(progress, *, loading_desc: str, loaded_desc: str): update_progress(progress, 0, loading_desc) return lambda: update_progress(progress, 1, loaded_desc) def update_request_start_progress(progress): update_progress(progress, 0, PROGRESS_DESC_PROCESSING) def update_request_translate_progress(progress): update_progress(progress, 0, PROGRESS_DESC_TRANSLATING) def update_request_stream_progress(progress): update_progress(progress, 0.5, PROGRESS_DESC_PROCESSING) # Hugging Face API wrappers def _run_hf_call(func, *args, **kwargs): return func(*args, **kwargs) @wrapt_timeout_decorator.timeout(dec_timeout=HF_API_TIMEOUT_SECONDS) def _run_hf_call_materialized(func, *args, **kwargs): return list(func(*args, **kwargs)) def format_hf_error(action: str, error: Exception): message = (str(error) or type(error).__name__).strip() lower_message = message.lower() if "timed out" in lower_message or "timeout" in lower_message: reason = "timed out" elif any(token in lower_message for token in ["401", "403", "unauthorized", "forbidden"]): reason = "access denied" elif any(token in lower_message for token in ["404", "not found", "repository not found"]): reason = "not found" else: reason = message return f"{action} failed: {reason}" def call_hf_api(action: str, func, *args, materialize: bool = False, **kwargs): last_error = None for attempt in range(HF_API_RETRY_COUNT + 1): try: runner = _run_hf_call_materialized if materialize else _run_hf_call return runner(func, *args, **kwargs) except Exception as error: last_error = error is_last_attempt = attempt >= HF_API_RETRY_COUNT log_event(logging.ERROR if is_last_attempt else logging.WARNING, "hf_api_call_failed", action=action, attempt=attempt + 1, error_type=type(error).__name__, error=error) if is_last_attempt: break time.sleep(HF_API_RETRY_BACKOFF_SECONDS * (attempt + 1)) raise RuntimeError(format_hf_error(action, last_error if last_error else RuntimeError("unknown error"))) # Hugging Face URL and download helpers def split_hf_url(url: str): matched_groups = re.findall(r'^(?:https?://huggingface.co/)(?:(datasets|spaces)/)?(.+?/.+?)/\w+?/.+?/(?:(.+)/)?(.+?.\w+)(?:\?download=true)?$', url) if not matched_groups: return "", "", None, None match_parts = list(matched_groups[0]) if len(match_parts) < 4: return "", "", None, None repo_id = match_parts[1] if match_parts[0] == "datasets": repo_type = "dataset" elif match_parts[0] == "spaces": repo_type = "space" else: repo_type = "model" subfolder = urllib.parse.unquote(match_parts[2]) if match_parts[2] else None filename = urllib.parse.unquote(match_parts[3]) return repo_id, filename, subfolder, repo_type def hf_url_exists(url: str): repo_id, filename, _subfolder, repo_type = split_hf_url(url) if not repo_id or not filename or not repo_type: return False api = HfApi(token=HF_TOKEN) return call_hf_api("hf.file_exists", api.file_exists, repo_id=repo_id, filename=filename, repo_type=repo_type, token=HF_TOKEN) def get_repo_type(repo_id: str): try: api = HfApi(token=HF_TOKEN) if call_hf_api("hf.repo_exists.dataset", api.repo_exists, repo_id=repo_id, repo_type="dataset", token=HF_TOKEN): return "dataset" if call_hf_api("hf.repo_exists.space", api.repo_exists, repo_id=repo_id, repo_type="space", token=HF_TOKEN): return "space" if call_hf_api("hf.repo_exists.model", api.repo_exists, repo_id=repo_id, token=HF_TOKEN): return "model" return None except Exception as error: log_event(logging.ERROR, "repo_type_lookup_failed", repo_id=repo_id, error_type=type(error).__name__, error=error) raise RuntimeError(format_hf_error(f"Repo lookup for {repo_id}", error)) def get_hf_blob_url(repo_id: str, repo_type: str, path: str): if repo_type == "model": return f"https://huggingface.co/{repo_id}/blob/main/{path}" elif repo_type == "dataset": return f"https://huggingface.co/datasets/{repo_id}/blob/main/{path}" elif repo_type == "space": return f"https://huggingface.co/spaces/{repo_id}/blob/main/{path}" @dataclass(frozen=True) class RegistrationTarget: raw_query: str = "" resolved_query: str = "" repo_id: str = "" filename: str = "" subfolder: Optional[str] = None repo_type: Optional[str] = None error_message: str = "" @property def is_resolved(self) -> bool: return bool(self.resolved_query) @property def is_model_repo(self) -> bool: return self.repo_type == "model" @property def is_gguf_file(self) -> bool: return str(self.filename or "").lower().endswith(".gguf") @property def is_valid_model_gguf(self) -> bool: return bool(self.repo_id and self.filename and self.is_model_repo and self.is_gguf_file) def to_legacy_dict(self): return { "filename": self.filename, "repo_id": self.repo_id, "resolved_query": self.resolved_query, "repo_type": self.repo_type, "subfolder": self.subfolder, "error_message": self.error_message, } def get_gguf_url(query_or_repo: str): def find_gguf(gguf_sizes: dict, preferred_keys: dict): matched_paths = [] for preferred_key, max_size in preferred_keys.items(): if max_size != 0: candidate_paths = [path for path, item_size in gguf_sizes.items() if preferred_key.lower() in path.lower() and item_size < max_size] else: candidate_paths = [path for path in gguf_sizes.keys() if preferred_key.lower() in path.lower()] if len(candidate_paths) > 0: matched_paths.append(candidate_paths[0]) if len(matched_paths) > 0: return matched_paths[0] return list(gguf_sizes.keys())[0] try: if query_or_repo.lower().endswith(".gguf"): return query_or_repo repo_type = get_repo_type(query_or_repo) if repo_type is None: return query_or_repo repo_id = query_or_repo api = HfApi(token=HF_TOKEN) tree = call_hf_api("hf.list_repo_tree", api.list_repo_tree, repo_id=repo_id, repo_type=repo_type, recursive=True, token=HF_TOKEN, materialize=True) gguf_dict = {item.path: item.size for item in tree if item.path.endswith(".gguf")} if len(gguf_dict) == 0: return query_or_repo return get_hf_blob_url(repo_id, repo_type, find_gguf(gguf_dict, {"Q5_K_M": 6000000000, "Q4_K_M": 0, "Q4": 0})) except Exception as error: log_fallback_event("gguf_url_lookup_failed", query=query_or_repo, error_type=type(error).__name__, error=error) return query_or_repo def resolve_hf_registration_target(query: str): raw_query = str(query or "").strip() if not raw_query: return RegistrationTarget(error_message="Enter a Hugging Face repo ID or file URL.") resolved_query = get_gguf_url(raw_query) if not resolved_query: return RegistrationTarget(raw_query=raw_query, error_message="Could not resolve a Hugging Face target.") if not hf_url_exists(resolved_query): return RegistrationTarget(raw_query=raw_query, resolved_query=resolved_query, error_message="Target file was not found on Hugging Face.") repo_id, filename, subfolder, repo_type = split_hf_url(resolved_query) if not repo_id or not filename or not repo_type: return RegistrationTarget(raw_query=raw_query, resolved_query=resolved_query, error_message="Resolved target is not a supported Hugging Face file URL.") return RegistrationTarget( raw_query=raw_query, resolved_query=resolved_query, repo_id=repo_id, filename=filename, subfolder=subfolder, repo_type=repo_type, ) def resolve_hf_blob_registration_target(query: str): return resolve_hf_registration_target(query).to_legacy_dict() def _build_registration_validation(query: str, *, kind_label: str): target = resolve_hf_registration_target(query) if not target.raw_query: return gr.validate(False, f"Enter a {kind_label} repo ID or URL."), target if target.error_message: return gr.validate(False, target.error_message), target if not target.is_model_repo: return gr.validate(False, "Only Hugging Face model repos are supported here."), target if not target.is_gguf_file: return gr.validate(False, "Only .gguf files are supported here."), target return gr.validate(True, ""), target def validate_model_registration_inputs(query: str, format_name: str, *extra_inputs): query_validation, _target = _build_registration_validation(query, kind_label="model") format_validation = gr.validate(bool(str(format_name or "").strip()), "Select a message format.") extra_validations = tuple(gr.validate(True, "") for _ in extra_inputs) return (query_validation, format_validation) + extra_validations def validate_lora_registration_input(query: str): query_validation, _target = _build_registration_validation(query, kind_label="LoRA") return query_validation def _coerce_clipboard_text(value: Any): return str(value or "").strip() def _has_clipboard_unsafe_chars(value: str): return any(ord(char) < 32 for char in value) or any(char.isspace() for char in value) def _is_strict_hf_repo_id(value: str): if not value or "://" in value or "\\" in value or _has_clipboard_unsafe_chars(value): return False if value.count("/") != 1 or not STRICT_HF_REPO_ID_PATTERN.match(value): return False for part in value.split("/"): if part.startswith((".", "-")) or part.endswith((".", "-")) or ".." in part or "--" in part: return False return True def _is_supported_hf_file_url(value: str): if not value or _has_clipboard_unsafe_chars(value): return False return bool(HF_GGUF_FILE_URL_PATTERN.match(value)) def _notify_clipboard_rejected(message: str = "Clipboard content was cleared because it is not a supported Hugging Face input."): try: gr.Info(message) except Exception: pass def gate_pasted_repo_id_for_autosend(clipboard_text: Any): text = _coerce_clipboard_text(clipboard_text) context = "prompt_translator_model_autosend" log_event(logging.INFO, "paste_gate_started", context=context, text_length=len(text)) if not text: log_event(logging.INFO, "paste_gate_completed", context=context, result="empty", accepted=False) return gr.skip(), False if _is_strict_hf_repo_id(text): log_event(logging.INFO, "paste_gate_completed", context=context, result="accepted_repo_id", accepted=True, text_length=len(text)) return text, True _notify_clipboard_rejected("Clipboard content was cleared because it is not an author/repo Hugging Face model ID.") log_event(logging.WARNING, "paste_gate_invalid", context=context, text_length=len(text)) return "", False def gate_pasted_registration_input(clipboard_text: Any): text = _coerce_clipboard_text(clipboard_text) context = "registration_input" log_event(logging.INFO, "paste_gate_started", context=context, text_length=len(text)) if not text: log_event(logging.INFO, "paste_gate_completed", context=context, result="empty", accepted=False) return gr.skip() if _is_strict_hf_repo_id(text): log_event(logging.INFO, "paste_gate_completed", context=context, result="accepted_repo_id", accepted=True, text_length=len(text)) return text if _is_supported_hf_file_url(text): log_event(logging.INFO, "paste_gate_completed", context=context, result="accepted_file_url", accepted=True, text_length=len(text)) return text _notify_clipboard_rejected() log_event(logging.WARNING, "paste_gate_invalid", context=context, text_length=len(text)) return "" def download_hf_file(directory, url, progress=gr.Progress(track_tqdm=True)): repo_id, filename, subfolder, repo_type = split_hf_url(url) if not repo_id or not filename or not repo_type: log_event(logging.ERROR, "hf_download_invalid_url", url=url) return None required_bytes = get_hf_file_size(repo_id, filename, subfolder, repo_type) try: ensure_storage_headroom(directory, required_bytes=required_bytes) log_event(logging.INFO, "hf_download_started", url=url, directory=directory, expected_bytes=required_bytes) kwargs = dict(repo_id=repo_id, filename=filename, repo_type=repo_type, local_dir=directory, token=HF_TOKEN) if subfolder is not None: kwargs["subfolder"] = subfolder path = call_hf_download("hf_hub_download", **kwargs) log_event(logging.INFO, "hf_download_completed", url=url, directory=directory, path=path) return path except Exception as error: log_event(logging.ERROR, "hf_download_failed", url=url, directory=directory, error_type=type(error).__name__, error=error) return None def call_hf_download(action: str, **kwargs): last_error = None for attempt in range(HF_API_RETRY_COUNT + 1): try: return hf_hub_download(**kwargs) except Exception as error: last_error = error is_last_attempt = attempt >= HF_API_RETRY_COUNT log_event(logging.ERROR if is_last_attempt else logging.WARNING, "hf_download_call_failed", action=action, attempt=attempt + 1, error_type=type(error).__name__, error=error) if is_last_attempt: break time.sleep(HF_API_RETRY_BACKOFF_SECONDS * (attempt + 1)) raise RuntimeError(format_hf_error(action, last_error if last_error else RuntimeError("unknown error"))) def get_storage_roots(path: str = ""): roots = [] for item in [path, llm_models_dir, llm_loras_dir]: if not item: continue normalized = str(Path(item)) if normalized not in roots: roots.append(normalized) return roots def get_total_storage_bytes(path: str = ""): total = 0 for root in get_storage_roots(path): if Path(root).exists(): total += get_dir_size(root) return total def is_cleanup_candidate(file_path: Path): if not file_path.is_file(): return False if file_path.name in [default_llm_model_filename, default_llm_lora_filename]: return False lower_name = file_path.name.lower() return lower_name.endswith(".gguf") or lower_name.endswith(".gguf.part") or lower_name.endswith(".gguf.partial") or lower_name.endswith(".part") or lower_name.endswith(".partial") or lower_name.endswith(".tmp") def is_file_in_use(file_path: Path): resolved = str(file_path.resolve()) with ACTIVE_FILE_LOCK: return ACTIVE_GGUF_FILES.get(resolved, 0) > 0 @contextlib.contextmanager def track_active_files(*paths): resolved_paths = [str(Path(path).resolve()) for path in paths if path] with ACTIVE_FILE_LOCK: for item in resolved_paths: ACTIVE_GGUF_FILES[item] = ACTIVE_GGUF_FILES.get(item, 0) + 1 try: yield finally: with ACTIVE_FILE_LOCK: for item in resolved_paths: if item not in ACTIVE_GGUF_FILES: continue ACTIVE_GGUF_FILES[item] -= 1 if ACTIVE_GGUF_FILES[item] <= 0: ACTIVE_GGUF_FILES.pop(item, None) def iter_cleanup_candidates(path: str): files = [] for root in get_storage_roots(path): root_path = Path(root) if not root_path.exists(): continue for candidate in root_path.rglob("*"): try: if is_cleanup_candidate(candidate) and not is_file_in_use(candidate): files.append(candidate) except Exception as error: log_event(logging.WARNING, "cleanup_candidate_skip", path=candidate, error_type=type(error).__name__, error=error) files.sort(key=lambda item: item.stat().st_atime if item.exists() else 0, reverse=False) return files # Legacy storage inspection helpers def get_dir_size(path: str): total = 0 with os.scandir(path) as it: for entry in it: if entry.is_file(): total += entry.stat().st_size elif entry.is_dir(): total += get_dir_size(entry.path) return total def get_dir_size_gb(path: str): try: size_gb = get_dir_size(path) / (1024 ** 3) log_event(logging.INFO, "dir_size", path=path, size_gb=f"{size_gb:.2f}") except Exception as error: size_gb = 999 log_event(logging.ERROR, "dir_size_failed", path=path, error_type=type(error).__name__, error=error) finally: return size_gb def build_storage_log_fields(cleanup_plan: dict, **fields): return { "path": cleanup_plan["path"], "current_gb": f"{cleanup_plan['current_bytes'] / (1024 ** 3):.2f}", "required_gb": f"{cleanup_plan['required_bytes'] / (1024 ** 3):.2f}", "projected_gb": f"{cleanup_plan['projected_bytes'] / (1024 ** 3):.2f}", "limit_gb": f"{cleanup_plan['limit_gb']:.2f}", "bytes_to_free_gb": f"{cleanup_plan['bytes_to_free'] / (1024 ** 3):.2f}", **fields, } def log_storage_cleanup_plan(cleanup_plan: dict, reason: str): log_event(logging.INFO, "storage_cleanup_plan", reason=reason, **build_storage_log_fields(cleanup_plan)) def log_storage_cleanup_action(cleanup_plan: dict, action: str, **fields): log_event(logging.INFO, "storage_cleanup_action", action=action, **build_storage_log_fields(cleanup_plan, **fields)) def log_storage_cleanup_result(cleanup_plan: dict, cleanup_result: dict, reason: str): log_event( logging.INFO, "storage_cleanup_result", reason=reason, cleaned_gb=f"{cleanup_result['cleaned_bytes'] / (1024 ** 3):.2f}", deleted_files=cleanup_result["deleted_files"], remaining_gb=f"{cleanup_result['remaining_bytes'] / (1024 ** 3):.2f}", **build_storage_log_fields(cleanup_plan), ) def build_storage_cleanup_plan(path: str, current_bytes: int, limit_gb: float, required_bytes: int = 0): limit_bytes = int(max(limit_gb, 0.0) * (1024 ** 3)) projected_bytes = current_bytes + max(required_bytes, 0) bytes_to_free = max(projected_bytes - limit_bytes, 0) return { "path": path, "current_bytes": current_bytes, "required_bytes": max(required_bytes, 0), "limit_gb": limit_gb, "limit_bytes": limit_bytes, "projected_bytes": projected_bytes, "bytes_to_free": bytes_to_free, } def should_run_storage_cleanup(cleanup_plan: dict): return cleanup_plan["limit_bytes"] > 0 and cleanup_plan["bytes_to_free"] > 0 def execute_storage_cleanup(cleanup_plan: dict, *, reason: str = "cleanup"): remaining_bytes = cleanup_plan["bytes_to_free"] cleaned_bytes = 0 deleted_files = 0 try: if remaining_bytes <= 0: cleanup_result = {"cleaned_bytes": 0, "deleted_files": 0, "remaining_bytes": 0} log_storage_cleanup_result(cleanup_plan, cleanup_result, reason) return cleanup_result log_storage_cleanup_action(cleanup_plan, "start", reason=reason) for file_path in iter_cleanup_candidates(cleanup_plan["path"]): if remaining_bytes <= 0: break try: size = file_path.stat().st_size file_path.unlink() remaining_bytes -= size cleaned_bytes += size deleted_files += 1 log_storage_cleanup_action(cleanup_plan, "delete_file", reason=reason, file_path=file_path, size_bytes=size) except Exception as error: log_event(logging.WARNING, "storage_file_delete_failed", path=file_path, reason=reason, error_type=type(error).__name__, error=error) cleanup_result = { "cleaned_bytes": cleaned_bytes, "deleted_files": deleted_files, "remaining_bytes": max(remaining_bytes, 0), } log_storage_cleanup_result(cleanup_plan, cleanup_result, reason) return cleanup_result except Exception as error: log_event(logging.ERROR, "storage_cleanup_failed", path=cleanup_plan["path"], reason=reason, error_type=type(error).__name__, error=error) raise def clean_dir(path: str, size_gb: float, limit_gb: float): current_bytes = int(max(size_gb, 0.0) * (1024 ** 3)) cleanup_plan = build_storage_cleanup_plan(path, current_bytes=current_bytes, limit_gb=limit_gb) if should_run_storage_cleanup(cleanup_plan): log_storage_cleanup_plan(cleanup_plan, reason="clean_dir") execute_storage_cleanup(cleanup_plan, reason="clean_dir") def update_storage(path: str, limit_gb: float=STORAGE_LIMIT_GB): current_bytes = get_total_storage_bytes(path) cleanup_plan = build_storage_cleanup_plan(path, current_bytes=current_bytes, limit_gb=limit_gb) if should_run_storage_cleanup(cleanup_plan): log_storage_cleanup_plan(cleanup_plan, reason="post_download") execute_storage_cleanup(cleanup_plan, reason="post_download") def get_hf_file_size(repo_id: str, filename: str, subfolder: str = None, repo_type: str = "model"): relative_path = f"{subfolder}/{filename}" if subfolder else filename try: api = HfApi(token=HF_TOKEN) tree = call_hf_api("hf.list_repo_tree", api.list_repo_tree, repo_id=repo_id, repo_type=repo_type, recursive=True, token=HF_TOKEN, materialize=True) for item in tree: if item.path == relative_path: return int(item.size or 0) except Exception as error: log_event(logging.WARNING, "hf_file_size_unavailable", repo_id=repo_id, filename=relative_path, repo_type=repo_type, error_type=type(error).__name__, error=error) return 0 def ensure_storage_headroom(path: str, required_bytes: int = 0, limit_gb: float = STORAGE_LIMIT_GB): safe_limit_gb = max(limit_gb - STORAGE_RESERVED_GB, 0.0) current_bytes = get_total_storage_bytes(path) cleanup_plan = build_storage_cleanup_plan(path, current_bytes=current_bytes, limit_gb=safe_limit_gb, required_bytes=required_bytes) if not should_run_storage_cleanup(cleanup_plan): return log_storage_cleanup_plan(cleanup_plan, reason="pre_download_headroom") execute_storage_cleanup(cleanup_plan, reason="pre_download_headroom") refreshed_bytes = get_total_storage_bytes(path) refreshed_plan = build_storage_cleanup_plan(path, current_bytes=refreshed_bytes, limit_gb=safe_limit_gb, required_bytes=required_bytes) if should_run_storage_cleanup(refreshed_plan): log_storage_cleanup_result( refreshed_plan, {"cleaned_bytes": 0, "deleted_files": 0, "remaining_bytes": refreshed_plan["bytes_to_free"]}, "pre_download_headroom_failed", ) raise RuntimeError( f"Insufficient storage: projected={refreshed_plan['projected_bytes'] / (1024 ** 3):.2f} GB safe_limit={safe_limit_gb:.2f} GB" ) def normalize_message_content(content: Any): if isinstance(content, str): return content if content is None: return "" if isinstance(content, list): parts = [] for item in content: if isinstance(item, str): parts.append(item) elif isinstance(item, dict): text = item.get("text") if isinstance(text, str): parts.append(text) normalized = "".join(parts) if normalized: trace_event("message_content_normalized", original_type=type(content).__name__, normalized_type="str") return normalized if isinstance(content, dict): text = content.get("text") if isinstance(text, str): trace_event("message_content_normalized", original_type=type(content).__name__, normalized_type="str") return text trace_event("message_content_normalized_fallback", original_type=type(content).__name__, normalized_type="str") return str(content) def truncate_history_for_context(history: list[MessageDict], max_messages: int = MAX_HISTORY_MESSAGES, max_chars: int = MAX_HISTORY_CHARS): if not history: return [] trimmed = [] total_chars = 0 for message in reversed(history): content = normalize_message_content(message.get("content", "")) content_len = len(content) if trimmed and (len(trimmed) >= max_messages or total_chars + content_len > max_chars): break trimmed.append(message) total_chars += content_len trimmed.reverse() if len(trimmed) != len(history): log_event(logging.INFO, "history_truncated", original_messages=len(history), kept_messages=len(trimmed), kept_chars=total_chars) return trimmed def cleanup_runtime(stream=None, agent=None, provider=None, llm=None): if stream is not None and hasattr(stream, "close"): try: stream.close() except Exception as error: log_event(logging.WARNING, "stream_close_failed", error_type=type(error).__name__, error=error) del stream, agent, provider, llm gc.collect() if torch.cuda.is_available(): try: torch.cuda.empty_cache() except Exception as error: log_event(logging.WARNING, "cuda_empty_cache_failed", error_type=type(error).__name__, error=error) try: torch.cuda.ipc_collect() except Exception: pass def build_chat_history(history: list[MessageDict]): ensure_llama_cpp_agent_runtime() messages = BasicChatHistory() truncated_history = truncate_history_for_context(history) for message_item in truncated_history: normalized_content = normalize_message_content(message_item.get("content", "")) if message_item["role"] == "user": user = {'role': Roles.user, 'content': normalized_content} messages.add_message(user) elif message_item["role"] == "assistant": assistant = {'role': Roles.assistant, 'content': normalized_content} messages.add_message(assistant) return messages, len(truncated_history), len(history) def log_llama_load_phase(event: str, *, function_name: str, request_id: str, model: str, lora: str, model_path: Optional[Path] = None, state: Optional[dict] = None, backend: str = "", phase: str = "", duration_ms: Optional[int] = None, level: int = logging.INFO, error: Optional[Exception] = None): fields = { "fn": function_name, "request_id": request_id, "model": model, "lora": lora, "session_hash": get_session_hash_from_state(state), } if model_path is not None: fields["path"] = str(model_path) if backend: fields["backend"] = backend if phase: fields["phase"] = phase if duration_ms is not None: fields["duration_ms"] = duration_ms if error is not None: fields["error_type"] = type(error).__name__ fields["error"] = error log_event(level, event, **fields) def iter_stream_with_first_output_log(stream, *, function_name: str, request_id: str, model: str, lora: str, model_path: Optional[Path] = None, state: Optional[dict] = None, backend: str = "", phase_started_at: Optional[float] = None): first_output_logged = False for item in stream: if not first_output_logged: first_output_logged = True duration_ms = None if phase_started_at is not None: duration_ms = int((time.monotonic() - phase_started_at) * 1000) log_llama_load_phase( "llama_first_output_started", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=state, backend=backend, duration_ms=duration_ms, ) yield item def build_llama_runtime(model_path: Path, lora: str = "", lora_scale: float = 1.0): ensure_llama_cpp_runtime() install_safe_llama_log_callback() kwargs = get_llama_runtime_kwargs(lora=lora, lora_scale=lora_scale) return Llama( model_path=str(model_path), n_gpu_layers=81, n_batch=1024, n_ctx=8192, **kwargs, ) def is_native_gguf_default_format(chat_template) -> bool: return chat_template == LLM_FORMAT_AUTO_GGUF_DEFAULT def resolve_agent_chat_template(filename: str, chat_template): if not is_native_gguf_default_format(chat_template): return chat_template registered_format_value = get_registered_model_format_value(filename) if registered_format_value is not None and not is_native_gguf_default_format(registered_format_value): return registered_format_value fallback_format_value = get_registered_model_format_value(default_llm_model_filename) if fallback_format_value is not None and not is_native_gguf_default_format(fallback_format_value): return fallback_format_value ensure_llama_cpp_agent_runtime() return MessagesFormatterType.CHATML def build_native_chat_messages(history: list[MessageDict], system_message: str, message: str): native_messages = [] if system_message: native_messages.append({"role": "system", "content": system_message}) truncated_history = truncate_history_for_context(history) native_messages.extend( {"role": message_item["role"], "content": normalize_message_content(message_item.get("content", ""))} for message_item in truncated_history ) native_messages.append({"role": "user", "content": message}) return native_messages def resolve_llama_agent_value(value): if hasattr(value, "resolve") and callable(value.resolve): return value.resolve() return value def build_llama_agent(model_path: Path, system_message: str, chat_template, lora: str = "", lora_scale: float = 1.0): ensure_llama_cpp_agent_runtime() resolved_chat_template = resolve_llama_agent_value(chat_template) llm = build_llama_runtime(model_path, lora=lora, lora_scale=lora_scale) provider = LlamaCppPythonProvider(llm) agent = LlamaCppAgent( provider, system_prompt=f"{system_message}", predefined_messages_formatter_type=resolved_chat_template if not isinstance(resolved_chat_template, MessagesFormatter) else None, custom_messages_formatter=resolved_chat_template if isinstance(resolved_chat_template, MessagesFormatter) else None, debug_output=False ) return llm, provider, agent def set_if_supported(target: Any, key: str, value: Any): if hasattr(target, key): setattr(target, key, value) return True return False def build_sampling_settings(provider, max_tokens: int, temperature: float, top_p: float, top_k: int, repeat_penalty: float): settings = provider.get_provider_default_settings() set_if_supported(settings, "temperature", temperature) set_if_supported(settings, "top_k", top_k) set_if_supported(settings, "top_p", top_p) set_if_supported(settings, "max_tokens", max_tokens) set_if_supported(settings, "repeat_penalty", repeat_penalty) set_if_supported(settings, "stream", True) return settings def build_native_chat_completion_kwargs(max_tokens: int, temperature: float, top_p: float, top_k: int, repeat_penalty: float): runtime_capabilities = get_runtime_capabilities() kwargs = { "max_tokens": max_tokens, "temperature": temperature, "top_p": top_p, "stream": True, } if runtime_capabilities.get("supports_native_top_k", True): kwargs["top_k"] = top_k if runtime_capabilities.get("supports_native_repeat_penalty", True): kwargs["repeat_penalty"] = repeat_penalty return kwargs def is_system_role_not_supported_error(error: Exception) -> bool: return "system role not supported" in str(error or "").lower() def create_native_chat_completion_stream(llm, history: list[MessageDict], system_message: str, message: str, max_tokens: int, temperature: float, top_p: float, top_k: int, repeat_penalty: float): messages = build_native_chat_messages(history, system_message, message) kwargs = build_native_chat_completion_kwargs(max_tokens, temperature, top_p, top_k, repeat_penalty) try: return llm.create_chat_completion(messages=messages, **kwargs) except ValueError as error: if is_system_role_not_supported_error(error) and any(message_item.get("role") == "system" for message_item in messages): log_event(logging.WARNING, "native_system_role_retry_without_system", error_type=type(error).__name__) fallback_messages = [message_item for message_item in messages if message_item.get("role") != "system"] return llm.create_chat_completion(messages=fallback_messages, **kwargs) raise @wrapt_timeout_decorator.timeout(dec_timeout=3.5) # Model and LoRA registry helpers def refresh_llm_model_choices(): global llm_models_list registered_model_names = list(llm_models.keys()) local_model_paths = Path(llm_models_dir).glob('*.gguf') llm_models_list = list_uniq(registered_model_names + [path.name for path in local_model_paths]) return llm_models_list def update_llm_model_list(): return refresh_llm_model_choices() def register_model_entry(filename: str, repo_id: str, format_value): global llm_models with MODEL_REGISTRY_LOCK: llm_models = (llm_models | {filename: [repo_id, format_value]}).copy() return refresh_llm_model_choices() # Registry read helpers. Keep direct global registry reads centralized. def get_model_file_path(filename: str) -> Path: return Path(llm_models_dir) / filename def get_lora_file_path(filename: str) -> Path: return Path(llm_loras_dir) / filename def get_registered_model_entry(filename: str): with MODEL_REGISTRY_LOCK: return llm_models.get(filename) def get_registered_model_repo_id(filename: str): model_entry = get_registered_model_entry(filename) return model_entry[0] if model_entry else None def get_registered_model_format_value(filename: str): model_entry = get_registered_model_entry(filename) return model_entry[1] if model_entry else None def get_registered_lora_url(filename: str): with MODEL_REGISTRY_LOCK: return llm_loras.get(filename) def download_llm_model(filename: str): if get_registered_model_entry(filename) is None: model_path = get_model_file_path(filename) return filename if model_path.exists() else default_llm_model_filename model_path = get_model_file_path(filename) if model_path.exists(): update_llm_model_list() return filename try: repo_id = get_registered_model_repo_id(filename) required_bytes = get_hf_file_size(repo_id, filename, repo_type="model") ensure_storage_headroom(llm_models_dir, required_bytes=required_bytes) log_event(logging.INFO, "model_download_started", model=filename, repo_id=repo_id, expected_bytes=required_bytes) call_hf_download("hf_hub_download", repo_id=repo_id, filename=filename, local_dir=llm_models_dir, token=HF_TOKEN) log_event(logging.INFO, "model_download_completed", model=filename, repo_id=repo_id) except Exception as error: log_event(logging.ERROR, "model_download_failed", model=filename, error_type=type(error).__name__, error=error) return default_llm_model_filename update_llm_model_list() return filename def refresh_llm_lora_choices(): global llm_loras_list registered_lora_names = list(llm_loras.keys()).copy() local_lora_paths = Path(llm_loras_dir).glob('*.gguf') llm_loras_list = list_uniq([""] + registered_lora_names + [path.name for path in local_lora_paths]) return llm_loras_list def update_llm_lora_list(): return refresh_llm_lora_choices() def register_lora_entry(filename: str, url: str): global llm_loras with MODEL_REGISTRY_LOCK: llm_loras = (llm_loras | {filename: url}).copy() return refresh_llm_lora_choices() def download_llm_lora(filename: str): lora_url = get_registered_lora_url(filename) if not lora_url: return "" try: path = download_hf_file(llm_loras_dir, lora_url) if not path: return "" log_event(logging.INFO, "lora_download_completed", lora=filename, path=path) except Exception as error: log_event(logging.ERROR, "lora_download_failed", lora=filename, error_type=type(error).__name__, error=error) return "" update_llm_lora_list() return filename def resolve_requested_chat_template(filename: str, state: Optional[dict], forced_chat_template: Optional[str] = None): if forced_chat_template is not None: if not is_native_gguf_default_format(forced_chat_template): return forced_chat_template template_source, resolved_template, _ = resolve_template_source( filename, state=state, forced_chat_template=forced_chat_template, ) if template_source == "inferred_format" and resolved_template is not None: return resolved_template return LLM_FORMAT_AUTO_GGUF_DEFAULT override_llm_format = get_optional_state(state, "override_llm_format") if override_llm_format: if is_native_gguf_default_format(override_llm_format): return resolve_requested_chat_template( filename, clear_selector_format_override(state), forced_chat_template=override_llm_format, ) return override_llm_format registered_format_value = get_registered_model_format_value(filename) if registered_format_value is not None: if is_native_gguf_default_format(registered_format_value): return resolve_requested_chat_template( filename, None, forced_chat_template=registered_format_value, ) return registered_format_value return get_registered_model_format_value(default_llm_model_filename) def get_dolphin_chat_template(filename: str, state: Optional[dict]): return resolve_requested_chat_template(filename, state) def get_llama_runtime_kwargs(lora: str = "", lora_scale: float = 1.0): kwargs = {} if lora: kwargs["lora_path"] = str(get_lora_file_path(lora)) kwargs["lora_scale"] = lora_scale return kwargs runtime_capabilities = get_runtime_capabilities() if runtime_capabilities.get("supports_flash_attn"): kwargs["flash_attn"] = True elif runtime_capabilities.get("supports_flash_attn_type"): flash_attn_type = getattr(llama_cpp_module, "llama_flash_attn_type", None) flash_attn_enabled = getattr(flash_attn_type, "LLAMA_FLASH_ATTN_TYPE_ENABLED", None) if flash_attn_enabled is not None: kwargs["flash_attn_type"] = flash_attn_enabled if runtime_capabilities.get("supports_verbose"): kwargs["verbose"] = False return kwargs def get_dolphin_model_info_payload(filename: str): model_entry = get_registered_model_entry(filename) registered_repo_id = normalize_repo_id(model_entry[0]) if model_entry else "" preflight = evaluate_format_preflight(filename) return { "model": str(filename or ""), "registered_repo_id": registered_repo_id, "original_repo_id": preflight.get("original_repo_id") or "", "original_repo_source": preflight.get("original_repo_source") or "", "original_repo_confidence": preflight.get("original_repo_confidence") or "", "preflight_risk": preflight.get("risk_level") or "", "template_source": preflight.get("template_source") or "", "inferred_format_name": preflight.get("inferred_format_name") or "", "inferred_format_source": preflight.get("inferred_format_source") or "", "inferred_format_confidence": preflight.get("inferred_format_confidence") or "", } def _build_text2tag_info_item(label: str, value_html: str, *, full_value: str = ""): safe_label = html.escape(str(label or "")) title_attr = f' title="{html.escape(str(full_value or ""))}"' if full_value else "" return f'{safe_label}: {value_html}' def _format_text2tag_info_repo_link(repo_id: str, *, fallback: str = "—"): normalized_repo_id = normalize_repo_id(repo_id) if not normalized_repo_id: return html.escape(fallback) safe_repo_id = html.escape(normalized_repo_id) safe_href = html.escape(f"https://huggingface.co/{normalized_repo_id}", quote=True) return f'{safe_repo_id}' def _format_text2tag_info_provenance(source: str, confidence: str): normalized_source = str(source or "").strip() normalized_confidence = str(confidence or "").strip() if normalized_source and normalized_confidence: return f"{normalized_source}/{normalized_confidence}" return normalized_source or normalized_confidence or "" def get_dolphin_model_info(filename: str): payload = get_dolphin_model_info_payload(filename) preflight_summary = f"{payload['preflight_risk'] or 'warn'} / {payload['template_source'] or 'unknown'}" fallback_parts = [] if payload.get("inferred_format_name"): fallback_parts.append(payload["inferred_format_name"]) inferred_provenance = _format_text2tag_info_provenance(payload.get("inferred_format_source"), payload.get("inferred_format_confidence")) if inferred_provenance: fallback_parts.append(inferred_provenance) fallback_value = " · ".join(part for part in fallback_parts if part) or "—" provenance_parts = [] original_provenance = _format_text2tag_info_provenance(payload.get("original_repo_source"), payload.get("original_repo_confidence")) if original_provenance: provenance_parts.append(f"orig={original_provenance}") if inferred_provenance: provenance_parts.append(f"fmt={inferred_provenance}") provenance_value = " | ".join(provenance_parts) or "—" items = [ _build_text2tag_info_item("Repo", _format_text2tag_info_repo_link(payload.get("registered_repo_id"), fallback="None"), full_value=payload.get("registered_repo_id", "")), _build_text2tag_info_item("Original", _format_text2tag_info_repo_link(payload.get("original_repo_id")), full_value=payload.get("original_repo_id", "")), _build_text2tag_info_item("Preflight", f"{html.escape(preflight_summary)}", full_value=preflight_summary), _build_text2tag_info_item("Fallback", f"{html.escape(fallback_value)}", full_value=fallback_value), _build_text2tag_info_item("Prov", f"{html.escape(provenance_value)}", full_value=provenance_value), ] return ( '
' '' + '|'.join(items) + '' '
' ) def build_selector_dropdown_update(selected_value: str, choices: list[str]): return gr.update(value=selected_value, choices=choices) def build_selector_value_update(selected_value: str): return gr.update(value=selected_value) def build_selector_state_result(component_update, normalized_state: dict): return component_update, normalized_state def build_selector_state_value_result(state: Optional[dict], key: str, value: Any, component_update): normalized_state = apply_selector_state_value(state, key, value) return build_selector_state_result(component_update, normalized_state) def prepare_model_selector_state(state: Optional[dict]): return clear_selector_format_override(state) def build_model_selector_payload(selected_model: str, normalized_state: dict): selected_format = get_dolphin_model_format(selected_model) model_info_markdown = get_dolphin_model_info(selected_model) normalized_state = write_state_value(normalized_state, STATE_SELECTED_MODEL_KEY, selected_model) return { "selected_model": selected_model, "selected_format": selected_format, "model_info": model_info_markdown, "model_update": build_selector_dropdown_update(selected_model, get_dolphin_models()), "format_update": build_selector_value_update(selected_format), "info_update": build_selector_value_update(model_info_markdown), "state": normalized_state, } def format_model_selector_payload(payload: dict): return ( payload["model_update"], payload["format_update"], payload["info_update"], payload["state"], ) def format_model_download_status_payload(payload: dict, download_succeeded: bool): return ( payload["model_update"], payload["format_update"], payload["info_update"], payload["state"], payload["selected_model"], download_succeeded, ) def build_lora_selector_updates(selected_lora: str, normalized_state: dict): normalized_state = write_state_value(normalized_state, STATE_SELECTED_LORA_KEY, selected_lora) return build_selector_state_result( build_selector_dropdown_update(selected_lora, get_dolphin_loras()), normalized_state, ) def build_format_selector_updates(format_name: str, normalized_state: dict): return build_selector_state_result(build_selector_value_update(format_name), normalized_state) def build_sysprompt_selector_updates(normalized_state: dict): return build_selector_state_result(build_selector_value_update(get_dolphin_sysprompt(normalized_state)), normalized_state) def resolve_model_registration_query(query: str, format_name: str): format_value = llm_formats[format_name] target = resolve_hf_registration_target(query) if not target.is_valid_model_gguf: return None, None, None, target return target.filename, target.repo_id, format_value, target def resolve_lora_registration_query(query: str): target = resolve_hf_registration_target(query) if not target.is_valid_model_gguf: return None, None, target return target.filename, target.resolved_query, target def build_event_progress_kwargs(*components): progress_targets = [component for component in components if component is not None] if not progress_targets: return {"show_progress": "minimal"} if len(progress_targets) == 1: return {"show_progress": "minimal", "show_progress_on": progress_targets[0]} return {"show_progress": "minimal", "show_progress_on": progress_targets} def build_playground_add_notice_update(selected_model: str, selected_format: str, add_succeeded: bool): if not add_succeeded or not selected_model: return gr.update(value="", visible=False) safe_model = html.escape(str(selected_model or "")) safe_format = html.escape(str(selected_format or "")) notice_html = ( '
' '' f'Added: {safe_model}' '|' f'Fmt: {safe_format or "—"}' '' '
' ) return gr.update(value=notice_html, visible=True) def extract_history_message_text(message: Any): if isinstance(message, str): return message if isinstance(message, list): for item in message: if isinstance(item, dict) and item.get("type") == "text": return str(item.get("text") or "") return "" if isinstance(message, dict): return str(message.get("text") or message.get("content") or "") return str(message or "") # Playground history helpers used by app.py wiring. def extract_history_prompt_from_index(history: list[MessageDict], index: Any) -> str: """Return the original user prompt for a retry/undo target index.""" history_entries = list(history or []) try: history_entry = history_entries[int(index)] except Exception: return "" if isinstance(history_entry, dict): return extract_history_message_text(history_entry.get("content", "")) return "" def resolve_event_index(event_data: Any, *, attribute_name: str = "index") -> int: """Safely read an integer event index from Gradio event payloads.""" try: return int(getattr(event_data, attribute_name, 0) or 0) except Exception: return 0 def handle_playground_undo(history: list[MessageDict], undo_data: gr.UndoData): """Trim chat history to the undo point and restore the removed prompt text.""" history_entries = list(history or []) index = resolve_event_index(undo_data) if index < 0 or index >= len(history_entries): return history_entries, "" return history_entries[:index], extract_history_prompt_from_index(history_entries, index) def retry_playground( history: list[MessageDict], model: str, system_message: Optional[str], max_tokens: int, temperature: float, top_p: float, top_k: int, repeat_penalty: float, lora: str, lora_scale: float, state: Optional[dict], retry_data: gr.RetryData, ): """Replay the selected playground prompt against the truncated history.""" history_entries = list(history or []) index = resolve_event_index(retry_data) if index < 0 or index >= len(history_entries): yield history_entries return original_prompt = extract_history_prompt_from_index(history_entries, index) retry_history = history_entries[:index] yield from respond_playground( original_prompt, retry_history, model, system_message, max_tokens, temperature, top_p, top_k, repeat_penalty, lora, lora_scale, state, ) # text2tag model-change guard helpers. def build_pending_clear_events(selected_model: str, current_model: str, selected_format: str, current_format: str) -> int: """Count how many downstream clear events must be ignored after a model/format change.""" pending_clear_count = 0 if selected_model and selected_model != current_model: pending_clear_count += 1 if selected_format and selected_format != current_format: pending_clear_count += 1 return pending_clear_count def coerce_pending_clear_events(pending_clear_events) -> int: """Normalize clear-guard counters to a non-negative integer.""" try: pending_clear_count = int(pending_clear_events or 0) except Exception as error: log_fallback_event( "pending_clear_events_coerce_failed", pending_clear_events_type=type(pending_clear_events).__name__, error_type=type(error).__name__, ) return 0 return max(0, pending_clear_count) def cleanup_text2tag_session_resources(): """Run lightweight, session-local cleanup without touching shared runtime state.""" with ACTIVE_FILE_LOCK: active_files = len(ACTIVE_GGUF_FILES) log_event(logging.INFO, "session_cleanup_started", scope="session-local", active_files=active_files) gc.collect() with ACTIVE_FILE_LOCK: active_files = len(ACTIVE_GGUF_FILES) log_event(logging.INFO, "session_cleanup_completed", scope="session-local", active_files=active_files) def prepare_request_context(function_name: str, history: list[MessageDict], model: str, lora: str, system_message: Optional[str], state: Optional[dict], *, require_model_file: bool = False, forced_chat_template: Optional[str] = None): normalized_state = ensure_state_dict(state) resolved_system_message = resolve_system_message(system_message, normalized_state) model_path = get_model_file_path(model) lora_path = get_lora_file_path(lora) if lora else None if require_model_file and not model_path.exists(): raise gr.Error(f"Model file not found: {str(model_path)}") maybe_warn_format_preflight(model, normalized_state, forced_chat_template=forced_chat_template, context=function_name) chat_template = resolve_requested_chat_template(model, normalized_state, forced_chat_template=forced_chat_template) messages, kept_messages, original_messages = build_chat_history(history) log_request_started(function_name, model, lora, original_messages, kept_messages, state=normalized_state) return normalized_state, resolved_system_message, model_path, lora_path, chat_template, messages def append_streaming_assistant_reply(history: list[MessageDict], message: str): history.append({"role": "user", "content": message}) history.append({"role": "assistant", "content": ""}) def stream_native_chat_history(history: list[MessageDict], message: str, stream): append_streaming_assistant_reply(history, message) for chunk in stream: try: delta = chunk["choices"][0].get("delta", {}) output = delta.get("content") except Exception: output = None if not output: continue history[-1]["content"] += output yield history def stream_agent_chat_history(history: list[MessageDict], message: str, stream): append_streaming_assistant_reply(history, message) for output in stream: if not output: continue history[-1]["content"] += output yield history def resolve_generation_backend(filename: str, chat_template, force_agent_backend: bool = False): if force_agent_backend: return "agent", resolve_agent_chat_template(filename, chat_template) if is_native_gguf_default_format(chat_template): return "native", chat_template return "agent", chat_template def generate_response_history( function_name: str, message: str, history: list[MessageDict], model: str, system_message: Optional[str], max_tokens: int, temperature: float, top_p: float, top_k: int, repeat_penalty: float, lora: str, lora_scale: float, state: Optional[dict], *, require_model_file: bool = False, forced_chat_template: Optional[str] = None, force_agent_backend: bool = False, show_start_progress: bool = False, show_translate_progress: bool = False, show_stream_progress: bool = False, progress=gr.Progress(track_tqdm=True), ): llm = provider = agent = stream = None model_path = None backend_name = "" normalized_state = state phase = "init" phase_started_at = time.monotonic() request_id = f"{function_name}-{int(time.time() * 1000)}-{threading.get_ident()}" try: if show_start_progress: update_request_start_progress(progress) phase = "prepare_context" phase_started_at = time.monotonic() normalized_state, resolved_system_message, model_path, lora_path, chat_template, messages = prepare_request_context( function_name, history, model, lora, system_message, state, require_model_file=require_model_file, forced_chat_template=forced_chat_template, ) phase = "resolve_backend" phase_started_at = time.monotonic() backend_name, resolved_chat_template = resolve_generation_backend(model, chat_template, force_agent_backend=force_agent_backend) trace_event( "request_context", request_id=request_id, fn=function_name, backend=backend_name, model=model, lora=lora, chat_template=resolved_chat_template, history_messages=len(history), ) with track_active_files(model_path, lora_path): log_llama_load_phase( "model_file_ready", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, ) if backend_name == "native": runtime_started_at = time.monotonic() phase = "native_runtime_build" phase_started_at = runtime_started_at log_llama_load_phase( "llama_runtime_build_started", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, phase=phase, ) llm = build_llama_runtime(model_path, lora=lora, lora_scale=lora_scale) log_llama_load_phase( "llama_runtime_build_completed", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, duration_ms=int((time.monotonic() - runtime_started_at) * 1000), ) stream_started_at = time.monotonic() phase = "native_stream_create" phase_started_at = stream_started_at stream = create_native_chat_completion_stream( llm, history, resolved_system_message, message, max_tokens, temperature, top_p, top_k, repeat_penalty, ) phase = "native_generation" phase_started_at = time.monotonic() log_llama_load_phase( "llama_generation_started", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, phase=phase, duration_ms=int((time.monotonic() - stream_started_at) * 1000), ) stream = iter_stream_with_first_output_log( stream, function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, phase_started_at=stream_started_at, ) if show_stream_progress: update_request_stream_progress(progress) yield from stream_native_chat_history(history, message, stream) else: runtime_started_at = time.monotonic() phase = "agent_runtime_build" phase_started_at = runtime_started_at log_llama_load_phase( "llama_runtime_build_started", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, phase=phase, ) llm, provider, agent = build_llama_agent(model_path, resolved_system_message, resolved_chat_template, lora=lora, lora_scale=lora_scale) log_llama_load_phase( "llama_runtime_build_completed", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, duration_ms=int((time.monotonic() - runtime_started_at) * 1000), ) settings = build_sampling_settings(provider, max_tokens, temperature, top_p, top_k, repeat_penalty) if show_translate_progress: update_request_translate_progress(progress) agent_started_at = time.monotonic() phase = "agent_stream_create" phase_started_at = agent_started_at stream = agent.get_chat_response( message, llm_sampling_settings=settings, chat_history=messages, returns_streaming_generator=True, print_output=False ) log_llama_load_phase( "llama_agent_build_completed", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, duration_ms=int((time.monotonic() - agent_started_at) * 1000), ) phase = "agent_generation" phase_started_at = time.monotonic() log_llama_load_phase( "llama_generation_started", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, phase=phase, duration_ms=int((time.monotonic() - agent_started_at) * 1000), ) stream = iter_stream_with_first_output_log( stream, function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, phase_started_at=agent_started_at, ) if show_stream_progress: update_request_stream_progress(progress) yield from stream_agent_chat_history(history, message, stream) except Exception as error: log_llama_load_phase( "llama_request_phase_failed", function_name=function_name, request_id=request_id, model=model, lora=lora, model_path=model_path, state=normalized_state, backend=backend_name, phase=phase, duration_ms=int((time.monotonic() - phase_started_at) * 1000), level=logging.ERROR, error=error, ) raise finally: cleanup_runtime(stream=stream, agent=agent, provider=provider, llm=llm) # Selector path. Returns dropdown update, format update, info update, and state. def select_dolphin_model(filename: str, state: Optional[dict], request: Optional[gr.Request] = None, progress=gr.Progress(track_tqdm=True)): normalized_state = prepare_model_selector_state(state) normalized_state, session_hash = attach_request_session_hash(normalized_state, request) requested_model = str(filename or "").strip() current_selected_model = str(read_state_value(normalized_state, STATE_SELECTED_MODEL_KEY, default=False, warn_default=False, warn_missing=False) or "").strip() if requested_model and requested_model == current_selected_model: trace_event("select_model_skipped", filename=requested_model, session_hash=session_hash, reason="unchanged") return format_model_selector_payload(build_model_selector_payload(requested_model, normalized_state)) trace_event("select_model_started", filename=filename, session_hash=session_hash) finish_progress = update_selector_progress(progress, loading_desc=PROGRESS_DESC_LOADING_MODEL, loaded_desc=PROGRESS_DESC_MODEL_LOADED) selected_model = download_llm_model(filename) finish_progress() update_storage(llm_models_dir) payload = build_model_selector_payload(selected_model, normalized_state) trace_event("select_model_completed", filename=filename, selected_model=payload["selected_model"], selected_format=payload["selected_format"], session_hash=session_hash) return format_model_selector_payload(payload) # Selector path. Returns dropdown update and state. def select_dolphin_lora(filename: str, state: Optional[dict], request: Optional[gr.Request] = None, progress=gr.Progress(track_tqdm=True)): normalized_state = ensure_state_dict(state) normalized_state, session_hash = attach_request_session_hash(normalized_state, request) requested_lora = str(filename or "").strip() current_selected_lora = str(read_state_value(normalized_state, STATE_SELECTED_LORA_KEY, default=False, warn_default=False, warn_missing=False) or "").strip() if requested_lora == current_selected_lora: trace_event("select_lora_skipped", filename=requested_lora, session_hash=session_hash, reason="unchanged") return build_lora_selector_updates(requested_lora, normalized_state) finish_progress = update_selector_progress(progress, loading_desc=PROGRESS_DESC_LOADING_LORA, loaded_desc=PROGRESS_DESC_LORA_LOADED) selected_lora = download_llm_lora(filename) finish_progress() update_storage(llm_loras_dir) trace_event("select_lora_completed", filename=filename, selected_lora=selected_lora, session_hash=session_hash) return build_lora_selector_updates(selected_lora, normalized_state) # Selector path. Returns format update and state. def select_dolphin_format(format_name: str, state: Optional[dict], request: Optional[gr.Request] = None): normalized_state, session_hash = attach_request_session_hash(state, request) resolved_format_value = llm_formats[format_name] current_format_value = read_state_value(normalized_state, "override_llm_format", default=False, warn_default=False, warn_missing=False) if current_format_value == resolved_format_value: trace_event("select_format_skipped", format_name=format_name, session_hash=session_hash, reason="unchanged") return build_selector_state_result(build_selector_value_update(format_name), normalized_state) return build_selector_state_value_result( normalized_state, "override_llm_format", resolved_format_value, build_selector_value_update(format_name), ) def get_dolphin_models(): return update_llm_model_list() def get_dolphin_loras(): return update_llm_lora_list() def get_llm_formats(): return list(llm_formats.keys()) def get_key_from_value(mapping, target_value): matching_keys = [key for key, value in mapping.items() if value == target_value] if matching_keys: return matching_keys[0] return None def get_dolphin_model_format(filename: str): format_value = get_registered_model_format_value(filename) if format_value is None: format_value = get_registered_model_format_value(default_llm_model_filename) format_name = get_key_from_value(llm_formats, format_value) return format_name # Add path. Returns model dropdown update, selected model name, and success flag. def add_dolphin_models_with_status(query: str, format_name: str): try: filename, repo_id, format_value, _target = resolve_model_registration_query(query, format_name) if not filename or not repo_id: return gr.update(), "", False choices = register_model_entry(filename, repo_id, format_value) selected_model = choices[-1] return build_selector_dropdown_update(selected_model, choices), selected_model, True except Exception as error: handle_selector_exception("add_model_failed", error, query=query) # Add path that registers, downloads, and selects the model before returning success. def add_dolphin_models_with_download_status(query: str, format_name: str, state: Optional[dict], progress=gr.Progress(track_tqdm=True)): try: update_progress(progress, 0.0, PROGRESS_DESC_CHECKING_REPO) filename, repo_id, format_value, _target = resolve_model_registration_query(query, format_name) normalized_state = prepare_model_selector_state(state) if not filename or not repo_id: return gr.update(), gr.update(), gr.update(), normalized_state, "", False register_model_entry(filename, repo_id, format_value) trace_event("add_model_download_started", query=query, filename=filename, repo_id=repo_id) update_progress(progress, 0.20, PROGRESS_DESC_DOWNLOADING_MODEL) selected_model = download_llm_model(filename) update_progress(progress, 0.85, PROGRESS_DESC_UPDATING_MODEL_LIST) update_storage(llm_models_dir) payload = build_model_selector_payload(selected_model, normalized_state) trace_event("add_model_download_completed", query=query, selected_model=payload["selected_model"], selected_format=payload["selected_format"]) update_progress(progress, 1.0, PROGRESS_DESC_MODEL_LOADED) return format_model_download_status_payload(payload, True) except Exception as error: handle_selector_exception("add_model_download_failed", error, query=query) def add_prompt_translator_model_with_download_status_and_clear_guard(query: str, format_name: str, state: Optional[dict], current_model: str, current_format: str, request: Optional[gr.Request] = None, progress=gr.Progress(track_tqdm=True)): normalized_state, session_hash = attach_request_session_hash(state, request) model_update, format_update, model_info_update, state_update, selected_model, download_succeeded = add_dolphin_models_with_download_status( query, format_name, normalized_state, progress=progress, ) selected_format = get_dolphin_model_format(selected_model) if selected_model else "" pending_clear_events = build_pending_clear_events(selected_model, current_model, selected_format, current_format) trace_event( "prompt_translator_model_transition", selected_model=selected_model, selected_format=selected_format, current_model=current_model, current_format=current_format, pending_clear_events=pending_clear_events, download_succeeded=download_succeeded, session_hash=session_hash, ) return model_update, format_update, model_info_update, state_update, selected_model, download_succeeded, pending_clear_events def add_prompt_translator_model_from_paste_with_download_status_and_clear_guard(paste_ready: bool, query: str, format_name: str, state: Optional[dict], current_model: str, current_format: str, request: Optional[gr.Request] = None, progress=gr.Progress(track_tqdm=True)): if not paste_ready: normalized_state, _session_hash = attach_request_session_hash(state, request) return gr.skip(), gr.skip(), gr.skip(), normalized_state, "", False, 0 return add_prompt_translator_model_with_download_status_and_clear_guard( query, format_name, state, current_model, current_format, request=request, progress=progress, ) def consume_chat_clear_guard(pending_clear_events): """Suppress the next clear events triggered indirectly by model/format updates.""" pending = coerce_pending_clear_events(pending_clear_events) if pending > 0: trace_event("consume_chat_clear_guard_blocked", pending_clear_events=pending, pending_after=pending - 1) return gr.skip(), pending - 1 trace_event("consume_chat_clear_guard_clear", pending_clear_events=0) return None, 0 def add_playground_model_with_status_and_clear_guard(query: str, format_name: str, current_model: str, current_format: str, request: Optional[gr.Request] = None): session_hash = get_request_session_hash(request) model_update, selected_model, add_succeeded = add_dolphin_models_with_status(query, format_name) selected_format = get_dolphin_model_format(selected_model) if selected_model else "" pending_clear_events = build_pending_clear_events(selected_model, current_model, selected_format, current_format) trace_event( "playground_model_transition", selected_model=selected_model, selected_format=selected_format, current_model=current_model, current_format=current_format, pending_clear_events=pending_clear_events, add_succeeded=add_succeeded, session_hash=session_hash, ) return model_update, selected_model, selected_format, add_succeeded, pending_clear_events # Prompt Translator follow-up helpers. def maybe_auto_send_prompt_translator( download_succeeded, auto_send_enabled, auto_send_prompt, history, selected_model, system_message, max_tokens, temperature, top_p, top_k, repeat_penalty, lora, lora_scale, state, tag_type, dummy_np, dummy_np_pony, recom_animagine, recom_pony, request: Optional[gr.Request] = None, progress=gr.Progress(), ): """Optionally auto-send a prompt after a successful model download in text2tag.""" if not download_succeeded or not auto_send_enabled: yield gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip() return normalized_state, session_hash = attach_request_session_hash(state, request) latest_history = [] trace_event("prompt_translator_auto_send_start", selected_model=selected_model, had_history=bool(history), session_hash=session_hash) update_progress(progress, 0.0, PROGRESS_DESC_WAITING_FOR_GPU) yield latest_history, gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip() preflight = evaluate_format_preflight(selected_model, state=normalized_state, forced_chat_template=LLM_FORMAT_AUTO_GGUF_DEFAULT) if preflight["risk_level"] == FORMAT_PREFLIGHT_HIGH_RISK: warning_message = f"Format preflight (auto-send): {preflight['risk_level']}. template_source={preflight['template_source']} gguf_chat_template={'yes' if preflight['has_chat_template'] else 'no'}" log_rate_limited_event( logging.WARNING, "format_preflight_auto_send_blocked", dedup_key=("format_preflight_auto_send_blocked", selected_model, preflight["template_source"]), model=selected_model, risk_level=preflight["risk_level"], template_source=preflight["template_source"], metadata_status=preflight["metadata_status"], ) try: gr.Warning(warning_message) except Exception: pass trace_event("prompt_translator_auto_send_blocked", selected_model=selected_model, template_source=preflight["template_source"], risk_level=preflight["risk_level"]) yield latest_history, "", gr.update(interactive=False), gr.update(interactive=False), "", dummy_np, dummy_np_pony return message = auto_send_prompt if auto_send_prompt is not None else "" update_progress(progress, 0.15, PROGRESS_DESC_LOADING_RUNTIME) response_stream = dolphin_respond( message, latest_history, model=selected_model, system_message=system_message, max_tokens=max_tokens, temperature=temperature, top_p=top_p, top_k=top_k, repeat_penalty=repeat_penalty, lora=lora, lora_scale=lora_scale, state=normalized_state, forced_chat_template=LLM_FORMAT_AUTO_GGUF_DEFAULT, ) first_output_seen = False for latest_history in response_stream: if not first_output_seen: update_progress(progress, 0.50, PROGRESS_DESC_GENERATING) first_output_seen = True yield latest_history, gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip() from tagger.tagger import convert_danbooru_to_e621_prompt, insert_recom_prompt update_progress(progress, 0.85, PROGRESS_DESC_FORMATTING_OUTPUT) parsed_output, copy_btn_update, copy_btn_pony_update = dolphin_parse(latest_history, normalized_state) pony_output = convert_danbooru_to_e621_prompt(parsed_output, tag_type) parsed_output, dummy_np_value = insert_recom_prompt(parsed_output, dummy_np, recom_animagine) pony_output, dummy_np_pony_value = insert_recom_prompt(pony_output, dummy_np_pony, recom_pony) trace_event("prompt_translator_auto_send_completed", selected_model=selected_model, output_length=len(parsed_output or ""), session_hash=session_hash) update_progress(progress, 1.0, PROGRESS_DESC_DONE) yield latest_history, parsed_output, copy_btn_update, copy_btn_pony_update, pony_output, dummy_np_value, dummy_np_pony_value def add_dolphin_models(query: str, format_name: str): component_update, _, _ = add_dolphin_models_with_status(query, format_name) return component_update # Add path. Returns LoRA dropdown update only. def add_dolphin_loras(query: str): try: filename, resolved_query, _target = resolve_lora_registration_query(query) if not filename or not resolved_query: return gr.update() choices = register_lora_entry(filename, resolved_query) return build_selector_dropdown_update(choices[-1], choices) except Exception as error: handle_selector_exception("add_lora_failed", error, query=query) # Prompt and language helpers def get_dolphin_sysprompt(state: Optional[dict] = None): normalized_state = ensure_state_dict(state) dolphin_sysprompt_mode = read_state_value(normalized_state, "dolphin_sysprompt_mode") dolphin_output_language = read_state_value(normalized_state, "dolphin_output_language") prompt = re.sub('', dolphin_output_language if dolphin_output_language else llm_languages[0], dolphin_system_prompt.get(dolphin_sysprompt_mode, dolphin_system_prompt[list(dolphin_system_prompt.keys())[0]])) return prompt def get_dolphin_sysprompt_mode(): return list(dolphin_system_prompt.keys()) def select_dolphin_sysprompt(key: str, state: Optional[dict]): sysprompt_mode = key if key in dolphin_system_prompt else "Default" normalized_state = apply_selector_state_value(state, "dolphin_sysprompt_mode", sysprompt_mode) return build_sysprompt_selector_updates(normalized_state) def get_dolphin_languages(): return llm_languages def select_dolphin_language(lang: str, state: Optional[dict]): normalized_state = apply_selector_state_value(state, "dolphin_output_language", lang) return build_sysprompt_selector_updates(normalized_state) @wrapt_timeout_decorator.timeout(dec_timeout=5.0) def get_raw_prompt(msg: str): matched_groups = re.findall(r'/GENBEGIN/(.+?)/GENEND/', msg, re.DOTALL) return re.sub(r'[*/:_"#\n]', ' ', ", ".join(matched_groups)).lower() if matched_groups else "" # Parser helpers def extract_latest_history_content(history: list[MessageDict]) -> str: latest_message = history[-1] if history else {} if isinstance(latest_message, dict): return normalize_message_content(latest_message.get("content", "")) return normalize_message_content(latest_message) def extract_generated_prompt_text(message_content: str) -> str: return get_raw_prompt(message_content) def build_prompt_tokens_from_raw(raw_prompt: str, sysprompt_mode: str, suffix_tokens: list[str]) -> list[str]: if sysprompt_mode == "Japanese to Danbooru Dictionary" and is_japanese(raw_prompt): base_tokens = jatags_to_danbooru_tags(to_list_ja(raw_prompt)) else: base_tokens = to_list(raw_prompt) return list_uniq(base_tokens + suffix_tokens) def build_prompt_text_from_history(history: list[MessageDict], sysprompt_mode: str, suffix_tokens: list[str]) -> str: latest_message_content = extract_latest_history_content(history) raw_prompt = extract_generated_prompt_text(latest_message_content) prompt_tokens = build_prompt_tokens_from_raw(raw_prompt, sysprompt_mode, suffix_tokens) return ", ".join(prompt_tokens) # https://llama-cpp-python.readthedocs.io/en/latest/api-reference/ @torch.inference_mode() @spaces.GPU(duration=30) # text2tag standard chat path. Returns updated chat history only. def dolphin_respond( message: str, history: list[MessageDict], model: str = default_llm_model_filename, system_message: Optional[str] = None, max_tokens: int = DEFAULT_MAX_TOKENS, temperature: float = DEFAULT_TEMPERATURE, top_p: float = DEFAULT_TOP_P, top_k: int = DEFAULT_TOP_K, repeat_penalty: float = DEFAULT_REPEAT_PENALTY, lora: str = "", lora_scale: float = DEFAULT_LORA_SCALE, state: Optional[dict] = None, forced_chat_template: Optional[str] = None, request: Optional[gr.Request] = None, progress=gr.Progress(track_tqdm=True), ): started_at = time.monotonic() normalized_state, _session_hash = attach_request_session_hash(state, request) try: yield from generate_response_history( "dolphin_respond", message, history, model, system_message, max_tokens, temperature, top_p, top_k, repeat_penalty, lora, lora_scale, normalized_state, require_model_file=True, forced_chat_template=forced_chat_template, show_start_progress=True, show_stream_progress=True, progress=progress, ) log_request_completed("dolphin_respond", started_at, model, lora, state=normalized_state) except Exception as error: handle_request_exception("dolphin_respond", error, started_at, model, lora, state=normalized_state) # text2tag parser path. Returns prompt text plus UI updates. def dolphin_parse( history: list[MessageDict], state: Optional[dict], ): try: normalized_state = ensure_state_dict(state) dolphin_sysprompt_mode = read_state_value(normalized_state, "dolphin_sysprompt_mode") if dolphin_sysprompt_mode == "Chat with LLM" or not history or len(history) < 1: return "", gr.update(), gr.update() parsed_prompt_text = build_prompt_text_from_history( history, dolphin_sysprompt_mode, ["nsfw", "explicit"], ) return parsed_prompt_text, gr.update(interactive=True), gr.update(interactive=True) except Exception as error: log_fallback_event("dolphin_parse_failed", error_type=type(error).__name__, error=error) return "", gr.update(), gr.update() @torch.inference_mode() @spaces.GPU(duration=30) # votepurchase enhancer path. On failure it must return the original input back through history. def dolphin_respond_auto( message: str, history: list[MessageDict], model: str = default_llm_model_filename, system_message: Optional[str] = None, max_tokens: int = DEFAULT_MAX_TOKENS, temperature: float = DEFAULT_TEMPERATURE, top_p: float = DEFAULT_TOP_P, top_k: int = DEFAULT_TOP_K, repeat_penalty: float = DEFAULT_REPEAT_PENALTY, lora: str = "", lora_scale: float = DEFAULT_LORA_SCALE, state: Optional[dict] = None, request: Optional[gr.Request] = None, progress=gr.Progress(track_tqdm=True), ): started_at = time.monotonic() normalized_state, _session_hash = attach_request_session_hash(state, request) try: for updated_history in generate_response_history( "dolphin_respond_auto", message, history, model, system_message, max_tokens, temperature, top_p, top_k, repeat_penalty, lora, lora_scale, normalized_state, force_agent_backend=True, show_start_progress=True, show_translate_progress=True, show_stream_progress=True, progress=progress, ): yield updated_history, gr.update(), gr.update() log_request_completed("dolphin_respond_auto", started_at, model, lora, state=normalized_state) except Exception as error: passthrough_history, textbox_update, state_update = handle_enhancer_passthrough("dolphin_respond_auto", error, started_at, model, lora, history, message, state=normalized_state) yield passthrough_history, textbox_update, state_update # votepurchase enhancer parser. On failure it should keep downstream passthrough behavior stable. def dolphin_parse_simple( message: str, history: list[MessageDict], state: Optional[dict], ): try: normalized_state = ensure_state_dict(state) #if not is_japanese(message): return message dolphin_sysprompt_mode = read_state_value(normalized_state, "dolphin_sysprompt_mode") if dolphin_sysprompt_mode == "Chat with LLM" or not history or len(history) < 1: return message return build_prompt_text_from_history( history, dolphin_sysprompt_mode, ["nsfw", "explicit", "rating_explicit"], ) except Exception as error: log_fallback_event("dolphin_parse_simple_failed", error_type=type(error).__name__, error=error) return "" # https://huggingface.co/spaces/CaioXapelaum/GGUF-Playground import cv2 cv2.setNumThreads(1) @torch.inference_mode() @spaces.GPU(duration=30) # Playground chat path. Returns updated chat history only. def respond_playground( message: str, history: list[MessageDict], model: str = default_llm_model_filename, system_message: Optional[str] = None, max_tokens: int = DEFAULT_MAX_TOKENS, temperature: float = DEFAULT_TEMPERATURE, top_p: float = DEFAULT_TOP_P, top_k: int = DEFAULT_TOP_K, repeat_penalty: float = DEFAULT_REPEAT_PENALTY, lora: str = "", lora_scale: float = DEFAULT_LORA_SCALE, state: Optional[dict] = None, request: Optional[gr.Request] = None, progress=gr.Progress(track_tqdm=True), ): started_at = time.monotonic() normalized_state, _session_hash = attach_request_session_hash(state, request) try: yield from generate_response_history( "respond_playground", message, history, model, system_message, max_tokens, temperature, top_p, top_k, repeat_penalty, lora, lora_scale, normalized_state, require_model_file=True, progress=progress, ) log_request_completed("respond_playground", started_at, model, lora, state=normalized_state) except Exception as error: handle_request_exception("respond_playground", error, started_at, model, lora, state=normalized_state)