nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Pair category repair (#550) and saved instructions + dialog picker (#431) (#590)

Browse Source 
* feat(maintenance): repair old Pair prompts' empty category and tags (#550)

Squashed from PR #579. Closes #550.

* feat(tagging): saved instructions with a default per engine (#431)

Squashed from PR #587.

* feat(tagging): pick saved instructions in the tagging dialogs (#431)

Squashed from PR #588. Closes #431.
This commit is contained in:
Vito
2026-11-01 03:01:07 +00:00
committed by GitHub
parent 9640ca86f8
commit 3295ff8b20
44 changed files with 3905 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
py/api/autotag_routes.py
+23 -3
View File
@@ -78,6 +78,12 @@ MODEL_DOWNLOAD_BODY_MAX = 1024
_download_titles = {}
def _instruction_engine(model_type: str) -> str:
"""The language-model engine whose default instructions ``model_type``
reads: itself, or GGUF for a WD14 engine (which takes none)."""
return model_type if model_type in ("gguf", "hf") else "gguf"
def _holder_label(holder: Optional[str]) -> str:
if not holder:
return ANOTHER_OPERATION
@@ -294,7 +300,15 @@ class AutotagRoutesMixin:
"success": True,
"models": public_models(status, loaded_type),
# The saved tagging defaults (#328, System › Auto-tagging).
"default_prompt": AutotagConfig.TEMPLATE,
# The saved engine's default instructions (GGUF for a
# WD14 engine); kept one release for cached pages (#431).
"default_prompt": AutotagConfig.default_instructions_text(
_instruction_engine(AutotagConfig.MODEL_TYPE)
),
# #431: Built-in first, with every text, and each
# language-model engine's default.
"instructions": AutotagConfig.all_instructions(),
"default_instructions": dict(AutotagConfig.DEFAULT_INSTRUCTIONS),
"model_loaded": loaded,
"loaded_model_type": loaded_type,
"wd14_general_threshold": AutotagConfig.GENERAL_THRESHOLD,
@@ -549,7 +563,10 @@ class AutotagRoutesMixin:
else:
prompt = params.get("prompt")
template = prompt if isinstance(prompt, str) and prompt.strip() else None
options = {"instructions": template or AutotagConfig.TEMPLATE}
options = {
"instructions": template
or AutotagConfig.default_instructions_text(model_type)
}
try:
_model, options = validate_options(
model_type,
@@ -1092,7 +1109,10 @@ class AutotagRoutesMixin:
else:
prompt = data.get("prompt")
template = prompt if isinstance(prompt, str) and prompt.strip() else None
options = {"instructions": template or AutotagConfig.TEMPLATE}
options = {
"instructions": template
or AutotagConfig.default_instructions_text(model_type)
}
use_gpu = data.get("use_gpu", True)
options["use_gpu"] = use_gpu if isinstance(use_gpu, bool) else True
return validate_options(
py/api/jobs.py
+30 -17
View File
@@ -204,6 +204,8 @@ class Job:
self.total = 0
self.counters = dict.fromkeys(groups, 0)
self.outcomes = {group: [] for group in groups}
# Groups whose report lists every row, beyond MAX_OUTCOME_ROWS (#550).
self.complete = {}
self.log = []
self._cancel_requested = False
self.last_event = None
@@ -297,31 +299,36 @@ class Job:
)
self.log_line(log_line)
def record_counted(self, group, count, rows=()):
def record_counted(self, group, count, rows=(), *, complete=False):
"""Count ``count`` outcomes in ``group`` at once, listing ``rows``.
For outcomes that need no log line each (a thumbnail job's already
fresh previews): the counter is exact, ``rows`` (``record``'s row
shape) are listed up to MAX_OUTCOME_ROWS, and nothing is logged.
``complete`` also keeps every row for the report (#550: a record of
each change that must not be cut short; the caller bounds it).
"""
if group not in self.counters:
raise ValueError(f"Unknown outcome group {group!r} for {self.kind}")
self.counters[group] += max(0, int(count))
listed = self.outcomes[group]
kept = self.complete.setdefault(group, []) if complete else None
for row in rows:
if len(listed) >= MAX_OUTCOME_ROWS:
if kept is None and len(listed) >= MAX_OUTCOME_ROWS:
break
index = row.get("root_index")
if isinstance(index, bool) or not isinstance(index, int) or index < 0:
index = None
listed.append(
{
"path": _relative_path(row.get("path")),
"root": str(row.get("root") or ""),
"root_index": index,
"detail": str(row.get("detail") or ""),
}
)
shaped = {
"path": _relative_path(row.get("path")),
"root": str(row.get("root") or ""),
"root_index": index,
"detail": str(row.get("detail") or ""),
}
if kept is not None:
kept.append(shaped)
if len(listed) < MAX_OUTCOME_ROWS:
listed.append(dict(shaped))
def finish(self, state="finished"):
"""End the job in ``state`` and release every subscriber.
@@ -406,8 +413,9 @@ class Job:
"""The downloadable report: the summary, every outcome group and the log.
Each group carries its exact count, its listed rows (at most
MAX_OUTCOME_ROWS) and whether the list is cut short. Rows and log lines
already hold root-relative paths only.
MAX_OUTCOME_ROWS, or every row of a ``complete`` group) and whether
the list is cut short. Rows and log lines already hold root-relative
paths only.
"""
summary = self.summary()
summary.pop("listed", None)
@@ -417,16 +425,21 @@ class Job:
"generated_at": self._clock(),
"job": summary,
"outcomes": {
group: {
"count": self.counters.get(group, len(rows)),
"rows": [dict(row) for row in rows],
"truncated": self.counters.get(group, len(rows)) > len(rows),
}
group: self._report_group(group, rows)
for group, rows in self.outcomes.items()
},
"log": list(self.log),
}
def _report_group(self, group, listed):
rows = self.complete.get(group, listed)
count = self.counters.get(group, len(rows))
return {
"count": count,
"rows": [dict(row) for row in rows],
"truncated": count > len(rows),
}
def report_filename(self):
"""The download name: prompt-manager-KIND-YYYYMMDD-HHMM.json (local time)."""
stamp = time.strftime("%Y%m%d-%H%M", time.localtime(self._clock()))
py/api/maintenance.py
+60 -6
View File
@@ -1,8 +1,9 @@
"""Maintenance routes: reviewed, read-only plans (#249 System S6).
``POST /prompt_manager/maintenance/preview {operation, options:{}}`` reviews
one operation (``duplicate_prompts``, ``optimize``, ``unused_components`` or
``missing_links``; ``database/maintenance.py``; ``duplicate_files``,
one operation (``duplicate_prompts``, ``optimize``, ``unused_components``,
``missing_links`` or ``pair_metadata``, #550; ``database/maintenance.py``;
``duplicate_files``,
``py/api/duplicate_files.py``; ``rotated_logs``, the rotated log copies,
``py/api/log_cleanup.py``, which has its own apply job) off the event loop and
answers a ``Plan``: ``{success, token, operation, expires_at, scope:{count,
@@ -97,7 +98,11 @@ MAINTENANCE_TITLES = {
"unused_components": "Unused prompt components",
"missing_links": "Missing-file links",
"workflow_storage": "Store each workflow once",
# #550, beyond the concept: fills old Pair prompts' empty category and tags.
"pair_metadata": "Pair category and tags",
}
# Operations whose job keeps every changed row for its report (#550).
COMPLETE_RECORD = frozenset({"pair_metadata"})
_OPERATION_ERROR = "operation must be one of: " + ", ".join(PREVIEW_OPERATIONS)
MAINTENANCE_GROUPS = ("changed", "protected", "errors")
MAX_TOKEN_LENGTH = 128
@@ -281,9 +286,27 @@ def short_space_message(short):
)
def _changed_line(operation, count, marks=0):
def _pair_metadata_line(result):
count = result["changed"]
noun = (
"prompt from its saved pairs"
if count == 1
else "prompts from their saved pairs"
)
categories = result.get("categories", 0)
tag_sets = result.get("tag_sets", 0)
return (
f"Filled {count:,} {noun} "
f"({categories:,} {'category' if categories == 1 else 'categories'}, "
f"{tag_sets:,} {'tag set' if tag_sets == 1 else 'tag sets'})"
)
def _changed_line(operation, count, marks=0, result=None):
if operation == "optimize":
return "Optimized the database"
if operation == "pair_metadata":
return _pair_metadata_line(result or {"changed": count})
if operation == "duplicate_prompts":
noun = "duplicate prompt" if count == 1 else "duplicate prompts"
return f"Merged {count:,} {noun}"
@@ -668,9 +691,28 @@ class MaintenanceRoutesMixin:
rows = await self._run_in_executor(
self._maintenance_rows, plan.operation, result["rows"]
)
job.record_counted("changed", result["changed"], rows)
job.record_counted("protected", payload["protected"])
line = _changed_line(plan.operation, result["changed"], result.get("marks", 0))
job.record_counted(
"changed",
result["changed"],
rows,
complete=plan.operation in COMPLETE_RECORD,
)
if "kept" in result:
# #550: each kept record with its reason, like the changed ones.
kept = [
{
"path": "",
"detail": f"Prompt {item['prompt_id']}: kept, "
+ "; ".join(item["reasons"]),
}
for item in result["kept"]
]
job.record_counted("protected", len(kept), kept, complete=True)
else:
job.record_counted("protected", payload["protected"])
line = _changed_line(
plan.operation, result["changed"], result.get("marks", 0), result
)
job.log_line(line)
job.update(progress=100, status_line=f"{line} · {backup_note(backup)}")
return None
@@ -691,6 +733,8 @@ class MaintenanceRoutesMixin:
}
)
return rows
if operation == "pair_metadata":
return [{"path": "", "detail": _filled_detail(item)} for item in described]
if operation == "duplicate_prompts":
return [
{
@@ -708,6 +752,16 @@ class MaintenanceRoutesMixin:
]
def _filled_detail(item):
"""``Prompt 12: category “portrait”; tags a, b``: one filled prompt (#550)."""
parts = []
if "category" in item:
parts.append(f"category “{item['category']}”")
if item.get("tags"):
parts.append("tags " + ", ".join(item["tags"]))
return f"Prompt {item['prompt_id']}: " + "; ".join(parts)
def backup_note(backup):
"""The status line's backup part: the file name, or that there is none."""
return f"Safety backup {backup}" if backup else NO_BACKUP
py/api/settings_routes.py
+100 -14
View File
@@ -54,13 +54,24 @@ _SETTINGS_FILE_KEYS = {
"AUTOTAG_EXISTING": ("autotag", "defaults", "existing"),
"AUTOTAG_GENERAL_THRESHOLD": ("autotag", "defaults", "general_threshold"),
"AUTOTAG_CHARACTER_THRESHOLD": ("autotag", "defaults", "character_threshold"),
"AUTOTAG_TEMPLATE": ("autotag", "defaults", "template"),
"AUTOTAG_SUGGESTION_LIMIT": ("autotag", "defaults", "suggestion_limit"),
"AUTOTAG_SKIPPED_TAGS": ("autotag", "defaults", "skipped_tags"),
# #431: the saved instructions and each engine's default, always written
# together (server-side pairing).
"AUTOTAG_INSTRUCTIONS": ("autotag", "defaults", "instructions"),
"AUTOTAG_DEFAULT_INSTRUCTIONS": ("autotag", "defaults", "default_instructions"),
}
# The old single template: written only as the GGUF default's mirror for a
# downgrade (#431 B1.1), deleted when that default is Built-in.
_TEMPLATE_MIRROR_KEY = ("autotag", "defaults", "template")
TEMPLATE_REPLACED_ERROR = (
"LLM tag template was replaced by Saved instructions. Reload the page."
)
# Whole-list settings (#431): a save without ``expected_revision`` would merge
# a stale copy over a list changed elsewhere, so it is refused.
_REVISION_REQUIRED = frozenset({"autotag_skipped_tags"})
_REVISION_REQUIRED = frozenset(
{"autotag_skipped_tags", "autotag_instructions", "autotag_default_instructions"}
)
REVISION_REQUIRED_ERROR = "Reload the page and try again."
# The thumbnail settings a running Convert outdated job fixes (#317).
_THUMBNAIL_SETTINGS = ("THUMBNAIL_FORMAT", "THUMBNAIL_QUALITY", "THUMBNAIL_EDGE")
@@ -142,13 +153,20 @@ def _settings_mutation(updates, new_roots):
"""The config commit for a settings save: only the submitted keys change.
Every other section and key (integrations and the CivitAI key, database,
unknown sections) is kept as it is on disk.
unknown sections) is kept as it is on disk. A save of the instructions
also writes the old ``template`` key as the GGUF default's mirror.
"""
def mutate(config):
for name, value in updates.items():
*sections, key = _SETTINGS_FILE_KEYS[name]
_set_nested(config, sections, key, value)
_set_nested(config, sections, key, _jsonable(value))
if "AUTOTAG_INSTRUCTIONS" in updates:
*sections, key = _TEMPLATE_MIRROR_KEY
mirror = _mirror_text(
updates["AUTOTAG_INSTRUCTIONS"], updates["AUTOTAG_DEFAULT_INSTRUCTIONS"]
)
_set_nested(config, sections, key, _DELETE if mirror is None else mirror)
if new_roots is not None:
gallery = _section(config, "gallery")
monitoring = _section(gallery, "monitoring")
@@ -160,10 +178,37 @@ def _settings_mutation(updates, new_roots):
return mutate
# ``_set_nested``'s value that removes the key instead of setting it.
_DELETE = object()
def _jsonable(value):
"""A runtime value as config.json holds it (tuples become lists)."""
if isinstance(value, (list, tuple)):
return [_jsonable(item) for item in value]
if isinstance(value, dict):
return {key: _jsonable(item) for key, item in value.items()}
return value
def _mirror_text(entries, defaults):
"""The GGUF default's text, or None when it is Built-in (#431 B1.1)."""
from ..tagging_defaults import BUILTIN_INSTRUCTIONS_ID
for item in entries:
if item["id"] == defaults["gguf"] != BUILTIN_INSTRUCTIONS_ID:
return item["text"]
return None
def _set_nested(config, sections, key, value):
"""``config[s1][s2]...[key] = value``, each section a fresh dict."""
"""``config[s1][s2]...[key] = value``, each section a fresh dict;
``_DELETE`` removes the key."""
if not sections:
config[key] = value
if value is _DELETE:
config.pop(key, None)
else:
config[key] = value
return
merged = _section(config, sections[0])
_set_nested(merged, sections[1:], key, value)
@@ -182,7 +227,6 @@ def _validate_autotag_settings(data):
MODEL_TYPES,
validate_saved_threshold,
validate_skipped_tags,
validate_template,
)
updates = {}
@@ -211,12 +255,7 @@ def _validate_autotag_settings(data):
raise ValueError(f"{error}.") from None
updates[key.upper()] = value
if "autotag_template" in data:
try:
updates["AUTOTAG_TEMPLATE"] = validate_template(
data["autotag_template"], "LLM tag template"
)
except ValueError as error:
raise ValueError(f"{error}.") from None
raise ValueError(TEMPLATE_REPLACED_ERROR)
if "autotag_suggestion_limit" in data:
value = data["autotag_suggestion_limit"]
if type(value) is not int or not 1 <= value <= MAX_SUGGESTION_LIMIT:
@@ -229,9 +268,37 @@ def _validate_autotag_settings(data):
updates["AUTOTAG_SKIPPED_TAGS"] = tuple(
validate_skipped_tags(data["autotag_skipped_tags"])
)
updates.update(_validate_instruction_settings(data))
return updates
def _validate_instruction_settings(data):
"""Server-side pairing (#431): a save that carries either instruction key
gets the other from the running configuration (which holds the migrated
list), the defaults are checked against the merged list, and both are
written. Empty when the save carries neither."""
from ..config import AutotagConfig
from ..tagging_defaults import (
validate_default_instructions,
validate_instructions,
)
if not {"autotag_instructions", "autotag_default_instructions"} & set(data):
return {}
if "autotag_instructions" in data:
entries = validate_instructions(data["autotag_instructions"])
else:
entries = [dict(item) for item in AutotagConfig.INSTRUCTIONS]
defaults = data.get(
"autotag_default_instructions", dict(AutotagConfig.DEFAULT_INSTRUCTIONS)
)
defaults = validate_default_instructions(defaults, entries)
return {
"AUTOTAG_INSTRUCTIONS": tuple(entries),
"AUTOTAG_DEFAULT_INSTRUCTIONS": defaults,
}
async def _read_json(request):
"""Parse a JSON object body through the package's capped reader."""
from . import _read_json_body
@@ -259,6 +326,7 @@ class SettingsRoutesMixin:
"""Get current settings."""
try:
from ..config import AutotagConfig, PromptManagerConfig, GalleryConfig
from ..tagging_defaults import DEFAULT_TAG_TEMPLATE
root_paths = [
self._public_path(d) for d in GalleryConfig.MONITORING_DIRECTORIES
@@ -291,7 +359,19 @@ class SettingsRoutesMixin:
"autotag_character_threshold": (
AutotagConfig.CHARACTER_THRESHOLD
),
"autotag_template": AutotagConfig.TEMPLATE,
# Read-only for one release (#431): a page cached
# from before the upgrade still renders.
"autotag_template": AutotagConfig.default_instructions_text(
"gguf"
),
"autotag_instructions": [
dict(item) for item in AutotagConfig.INSTRUCTIONS
],
"autotag_default_instructions": dict(
AutotagConfig.DEFAULT_INSTRUCTIONS
),
# Built-in's text, which System shows read-only.
"autotag_builtin_instructions": DEFAULT_TAG_TEMPLATE,
"autotag_suggestion_limit": AutotagConfig.SUGGESTION_LIMIT,
"autotag_skipped_tags": list(AutotagConfig.SKIPPED_TAGS),
"cpu_count": PromptManagerConfig.max_worker_threads(),
@@ -526,6 +606,12 @@ class SettingsRoutesMixin:
if name in _AUTOTAG_SETTINGS:
setattr(AutotagConfig, name[len(_AUTOTAG_PREFIX) :], value)
continue
if "AUTOTAG_INSTRUCTIONS" in updates:
# What config.json now holds as the old key (the mirror).
AutotagConfig._LEGACY_TEMPLATE = AutotagConfig.mirror_template()
for name, value in updates.items():
if name in _AUTOTAG_SETTINGS:
continue
owner = GalleryConfig if name in _GALLERY_SETTINGS else PromptManagerConfig
setattr(owner, name, value)
restart_fields = []
py/config.py
+167 -10
View File
@@ -30,6 +30,7 @@ routes = server_instance.routes
# Extension info
extension_uri = None # Will be set in __init__.py
import hashlib # noqa: E402
import os # noqa: E402
from typing import Any, Dict, List, Optional, Tuple # noqa: E402
@@ -590,6 +591,12 @@ try:
validate_skipped_tags,
validate_template,
validate_threshold,
BUILTIN_INSTRUCTIONS_ID,
INSTRUCTION_ENGINES,
MAX_SAVED_INSTRUCTIONS,
builtin_instructions,
validate_default_instructions,
validate_instructions,
)
except ImportError:
from py.tagging_defaults import ( # noqa: F401 - re-exported
@@ -604,6 +611,12 @@ except ImportError:
validate_skipped_tags,
validate_template,
validate_threshold,
BUILTIN_INSTRUCTIONS_ID,
INSTRUCTION_ENGINES,
MAX_SAVED_INSTRUCTIONS,
builtin_instructions,
validate_default_instructions,
validate_instructions,
)
@@ -623,13 +636,24 @@ def _valid_skipped_tags(value: Any) -> Tuple[str, ...]:
return tuple(validate_skipped_tags(value))
def _reason(error: ValueError) -> str:
"""A settings sentence as a load warning's reason ("entry 2 is empty")."""
text = str(error).rstrip(".")
for prefix in ("Skipped tags", "Saved instructions"):
if text.startswith(prefix):
return text.removeprefix(prefix).lstrip(": ")
return text
def _text_id(text: str) -> str:
"""A migrated entry's id: deterministic, so every load gives the same."""
return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]
def _fallback_warning(key: str, error: ValueError, default: Any) -> str:
"""The one load warning of a saved tagging default that fails its check."""
if key == "template":
return f"autotag.defaults.template {error}; using the built-in one"
if key == "skipped_tags":
reason = str(error).removeprefix("Skipped tags").lstrip(": ").rstrip(".")
return f"autotag.defaults.skipped_tags {reason}; using no skipped tags"
return f"autotag.defaults.skipped_tags {_reason(error)}; using no skipped tags"
return f"autotag.defaults.{key} {error}; using {default!r}"
@@ -653,9 +677,18 @@ class AutotagConfig:
EXISTING = "skip"
GENERAL_THRESHOLD = 0.35
CHARACTER_THRESHOLD = 0.85
TEMPLATE = DEFAULT_TAG_TEMPLATE
SUGGESTION_LIMIT = DEFAULT_SUGGESTION_LIMIT
SKIPPED_TAGS: Tuple[str, ...] = ()
# #431: the saved instructions (Built-in is never stored) and the default
# of each language-model engine, by id.
INSTRUCTIONS: Tuple[Dict[str, str], ...] = ()
DEFAULT_INSTRUCTIONS: Dict[str, str] = {
"gguf": BUILTIN_INSTRUCTIONS_ID,
"hf": BUILTIN_INSTRUCTIONS_ID,
}
# The old single template as read at load (None when absent or invalid);
# only the migration reads it. Saves keep it as the GGUF default's mirror.
_LEGACY_TEMPLATE: Optional[str] = None
# config.json key -> (attribute, validator, default)
_KEYS = {
@@ -671,7 +704,6 @@ class AutotagConfig:
lambda v: validate_saved_threshold("value", v),
0.85,
),
"template": ("TEMPLATE", validate_template, DEFAULT_TAG_TEMPLATE),
"suggestion_limit": (
"SUGGESTION_LIMIT",
_valid_suggestion_limit,
@@ -691,18 +723,142 @@ class AutotagConfig:
type(value) is int and 1 <= value <= MAX_SUGGESTION_LIMIT
) # bool, float and str are rejected
@classmethod
def all_instructions(cls) -> List[Dict[str, str]]:
"""Built-in first, then the saved entries (copies)."""
return [builtin_instructions()] + [dict(item) for item in cls.INSTRUCTIONS]
@classmethod
def default_instructions_text(cls, engine: str) -> str:
"""The text of ``engine``'s (``gguf`` or ``hf``) default instructions."""
if engine not in INSTRUCTION_ENGINES:
raise ValueError(f"{engine} takes no instructions")
wanted = cls.DEFAULT_INSTRUCTIONS.get(engine, BUILTIN_INSTRUCTIONS_ID)
for item in cls.INSTRUCTIONS:
if item["id"] == wanted:
return item["text"]
return DEFAULT_TAG_TEMPLATE
@classmethod
def mirror_template(cls) -> Optional[str]:
"""What an older build reads as its one template (#431 B1.1): the
GGUF default's text, or None (no key) when that default is Built-in."""
if cls.DEFAULT_INSTRUCTIONS.get("gguf") == BUILTIN_INSTRUCTIONS_ID:
return None
return cls.default_instructions_text("gguf")
@classmethod
def get_config(cls) -> Dict[str, Any]:
"""The saved defaults; the template only when it is not the built-in
one, so a later built-in template still reaches who never set one."""
"""The saved defaults; ``template`` only as the GGUF default's mirror."""
defaults = {
key: getattr(cls, attr) for key, (attr, _v, _d) in cls._KEYS.items()
}
if defaults["template"] == DEFAULT_TAG_TEMPLATE:
del defaults["template"]
defaults["skipped_tags"] = list(defaults["skipped_tags"])
defaults["instructions"] = [dict(item) for item in cls.INSTRUCTIONS]
defaults["default_instructions"] = dict(cls.DEFAULT_INSTRUCTIONS)
mirror = cls.mirror_template()
if mirror is not None:
defaults["template"] = mirror
return {"defaults": defaults}
@classmethod
def _load_instructions(cls, defaults: Dict[str, Any]) -> None:
"""Each instruction key alone, then the cross-key default check, then
the in-memory migration of the old template (#431). Nothing is
written here, so the revision never changes at load."""
cls._LEGACY_TEMPLATE = None
if "template" in defaults:
try:
cls._LEGACY_TEMPLATE = validate_template(defaults["template"])
except ValueError as error:
config_logger.warning(
f"autotag.defaults.template {error}; using the built-in one"
)
entries: List[Dict[str, str]] = []
damaged = False
if "instructions" in defaults:
try:
entries = validate_instructions(defaults["instructions"])
except ValueError as error:
damaged = True
config_logger.warning(
f"autotag.defaults.instructions {_reason(error)}; "
"using Built-in only"
)
chosen = {engine: BUILTIN_INSTRUCTIONS_ID for engine in INSTRUCTION_ENGINES}
stored = defaults.get("default_instructions")
if damaged:
# Both defaults fall back with the list (its one warning says so);
# the old template is migrated as on a first upgrade.
pass
elif "default_instructions" in defaults and (
not isinstance(stored, dict) or set(stored) != set(INSTRUCTION_ENGINES)
):
config_logger.warning(
"autotag.defaults.default_instructions must name gguf and hf; "
"using Built-in"
)
elif isinstance(stored, dict):
ids = {item["id"] for item in entries}
for engine in INSTRUCTION_ENGINES:
value = stored[engine]
if isinstance(value, str) and (
value == BUILTIN_INSTRUCTIONS_ID or value in ids
):
chosen[engine] = value
else:
config_logger.warning(
f"autotag.defaults.default_instructions.{engine} names no "
"saved instructions; using Built-in"
)
cls._migrate_template(
entries,
chosen,
first=damaged or "instructions" not in defaults,
keep_defaults=not damaged and "default_instructions" in defaults,
)
cls.INSTRUCTIONS = tuple(entries)
cls.DEFAULT_INSTRUCTIONS = chosen
@classmethod
def _migrate_template(cls, entries, chosen, *, first, keep_defaults) -> None:
"""Bring a template saved by an older build in as a named entry."""
legacy = cls._LEGACY_TEMPLATE
if legacy is None or legacy == DEFAULT_TAG_TEMPLATE:
return
if first:
# The first upgrade: it becomes "Custom template", the default for
# both engines unless defaults are stored already (20A).
entries.append(
{"id": _text_id(legacy), "name": "Custom template", "text": legacy}
)
if not keep_defaults:
for engine in INSTRUCTION_ENGINES:
chosen[engine] = entries[-1]["id"]
return
if any(item["text"] == legacy for item in entries):
return # the mirror of a saved default
# Edited on an older build after an upgrade (B1.2): import it.
if len(entries) >= MAX_SAVED_INSTRUCTIONS:
config_logger.warning(
"The tagging template edited on an older version was not imported: "
f"the saved instructions are full ({MAX_SAVED_INSTRUCTIONS})"
)
return
names = {item["name"].casefold() for item in entries}
number = 2
while f"custom template ({number})" in names:
number += 1
name = f"Custom template ({number})"
ids = {item["id"] for item in entries}
new_id, n = _text_id(legacy), 2
while new_id in ids:
new_id, n = _text_id(f"{legacy}\n{n}"), n + 1
entries.append({"id": new_id, "name": name, "text": legacy})
config_logger.info(
"Imported the tagging template edited on an older version as " f"'{name}'"
)
@classmethod
def update_config(cls, new_config: Any) -> None:
if not isinstance(new_config, dict):
@@ -720,6 +876,7 @@ class AutotagConfig:
except ValueError as error:
config_logger.warning(_fallback_warning(key, error, default))
setattr(cls, attr, default)
cls._load_instructions(defaults)
class PromptManagerConfig:
py/tagging_defaults.py
+94
View File
@@ -226,3 +226,97 @@ def wd14_label_names(path) -> Tuple[str, ...]:
with _label_cache_lock:
_label_cache[key] = (stamp, names)
return names
# Saved instructions (#431): named language-model instructions with a default
# per engine. Built-in is never stored; readers put it first.
BUILTIN_INSTRUCTIONS_ID = "builtin"
BUILTIN_INSTRUCTIONS_NAME = "Built-in"
MAX_SAVED_INSTRUCTIONS = 20
MAX_INSTRUCTION_NAME_CHARS = 60
INSTRUCTION_ENGINES = ("gguf", "hf")
ENGINE_LABELS = {"gguf": "GGUF JoyCaption", "hf": "HuggingFace JoyCaption"}
_INSTRUCTION_ID = re.compile(r"[0-9a-f]{12}")
_CONTROL = re.compile(r"[\x00-\x1f\x7f-\x9f

]")
def builtin_instructions() -> Dict[str, str]:
"""The read-only Built-in entry, always the shipped template."""
return {
"id": BUILTIN_INSTRUCTIONS_ID,
"name": BUILTIN_INSTRUCTIONS_NAME,
"text": DEFAULT_TAG_TEMPLATE,
}
def _instruction_entry(raw: Any, number: int) -> Dict[str, str]:
prefix = f"Saved instructions: entry {number}"
if (
not isinstance(raw, dict)
or set(raw) != {"id", "name", "text"}
or not isinstance(raw["id"], str)
or _INSTRUCTION_ID.fullmatch(raw["id"]) is None
):
raise ValueError(f"{prefix} has an invalid id.")
name = raw["name"].strip() if isinstance(raw["name"], str) else ""
if not name:
raise ValueError(f"{prefix} needs a name.")
if len(name) > MAX_INSTRUCTION_NAME_CHARS:
raise ValueError(
f"{prefix} name must be at most {MAX_INSTRUCTION_NAME_CHARS} characters."
)
if _CONTROL.search(name):
raise ValueError(f"{prefix} name must not contain a line break.")
if name.casefold() == BUILTIN_INSTRUCTIONS_NAME.casefold():
raise ValueError("Saved instructions: the name “Built-in” is reserved.")
try:
text = validate_template(raw["text"], f"{prefix} text")
except ValueError as error:
raise ValueError(f"{error}.") from None
return {"id": raw["id"], "name": name, "text": text}
def validate_instructions(value: Any) -> List[Dict[str, str]]:
"""The saved entries ``[{id, name, text}]`` in order (names trimmed), or
ValueError with the sentence System shows (Built-in is not counted)."""
if not isinstance(value, list):
raise ValueError("Saved instructions must be a list.")
if len(value) > MAX_SAVED_INSTRUCTIONS:
raise ValueError(
f"Saved instructions: at most {MAX_SAVED_INSTRUCTIONS} entries besides "
"Built-in."
)
entries: List[Dict[str, str]] = []
ids: Dict[str, int] = {}
names: Dict[str, int] = {}
for number, raw in enumerate(value, 1):
item = _instruction_entry(raw, number)
if item["id"] in ids:
raise ValueError(
f"Saved instructions: entry {number} repeats the id of entry "
f"{ids[item['id']]}."
)
key = item["name"].casefold()
if key in names:
raise ValueError(
f"Saved instructions: entry {number} name “{item['name']}” is "
"already used."
)
ids[item["id"]] = number
names[key] = number
entries.append(item)
return entries
def validate_default_instructions(value: Any, entries) -> Dict[str, str]:
"""``{gguf: id, hf: id}``, each Built-in or an id of ``entries``."""
if not isinstance(value, dict) or set(value) != set(INSTRUCTION_ENGINES):
raise ValueError("Default instructions must name GGUF and HuggingFace.")
ids = {BUILTIN_INSTRUCTIONS_ID} | {item["id"] for item in entries}
for engine in INSTRUCTION_ENGINES:
if not isinstance(value[engine], str) or value[engine] not in ids:
raise ValueError(
f"Default for {ENGINE_LABELS[engine]} must be one of the saved "
"instructions."
)
return {engine: value[engine] for engine in INSTRUCTION_ENGINES}