""" AI provider layer — Claude + OpenAI-compatible (Ollama, llama.cpp, LM Studio...). Config lives in {user_dir}/PromptChain/ai_config.json and is read server-side only. API keys are never returned to the client — the config GET returns a sanitized shape with has_key flags in place of the raw key. Streaming: generate spawns an asyncio task keyed by request_id. Tokens are sent to the client via the existing send_ws broadcast channel on the `promptchain_ai_stream` event. Cancel looks up the task and cancels it. """ from __future__ import annotations import asyncio import json import logging import os import re import shutil import subprocess import sys import time import uuid from pathlib import Path from typing import Callable import aiohttp from aiohttp import web import folder_paths import server from .api_utils import atomic_write_json, error_response, parse_json from . import model_settings from .shared import send_ws from .tags import get_store as get_tag_store logger = logging.getLogger("promptchain.ai") # Dedicated debug channel — dumps full system prompt, user message, body # buffer, STYLE parsing, splice result. Enabled by default so we can trace # why a given model went off-script. Mute via logging config to silence. dbg = logging.getLogger("promptchain.ai.debug") if dbg.level == logging.NOTSET: dbg.setLevel(logging.INFO) routes = server.PromptServer.instance.routes def _safe_for_log(text: str) -> str: """ComfyUI's default log stream on Windows is cp1252 — characters like `→`, emoji, and curly quotes raise UnicodeEncodeError and the whole log record gets dropped. Round-trip via cp1252 with `errors='replace'` so unencodable code points become `?` instead of losing the dump entirely. No-op when the underlying stream is utf-8.""" if not text: return text or "" try: return text.encode("cp1252", "replace").decode("cp1252") except Exception: return text.encode("ascii", "replace").decode("ascii") def _trunc(text: str, limit: int = 6000) -> str: if not text: return "" safe = _safe_for_log(text) if len(safe) <= limit: return safe return safe[:limit] + f"\n...[+{len(safe) - limit} chars truncated]" def _dump(request_id: str, label: str, body: str): safe_label = _safe_for_log(label) dbg.info("ai[%s] %s (%d chars):\n%s\n--- /%s ---", request_id, safe_label, len(body or ""), _trunc(body or ""), safe_label) # Model-agnostic safety limits for the aiohttp-side call. Providers # enforce their own limits on top. _CONNECT_TIMEOUT = 10 _READ_TIMEOUT = 300 # 5 min — vision + reasoning can stretch # Generous cap for reasoning models (Qwen3, DeepSeek-R1, etc.) that burn # tokens on ... before any visible output. With our # typical ~1K user_chars of directives + bios, reasoning alone can run # 3-5K tokens. max_tokens is a runaway cap, not a target — models stop # when done, so erring high costs nothing. _MAX_TOKENS = 16384 # The local model onboarding offers to pull and the AI assistant's # prompt-engineering is tuned around. Single source of truth — the rest of # this file references the tag in comments only. RECOMMENDED_LOCAL_MODEL = "qwen3-vl:8b-instruct" _DEFAULT_OLLAMA_ROOT = "http://localhost:11434" # ── cloud service registry ──────────────────────────────────────── # Labels/help links are on the frontend; the backend only needs the # routing info (base_url and whether it speaks Anthropic vs OpenAI shape). _CLOUD_SERVICES: dict[str, dict] = { "claude": {"base_url": None, "shape": "anthropic"}, "openai": {"base_url": "https://api.openai.com/v1", "shape": "openai"}, "grok": {"base_url": "https://api.x.ai/v1", "shape": "openai"}, "gemini": {"base_url": "https://generativelanguage.googleapis.com/v1beta/openai", "shape": "openai"}, "openrouter": {"base_url": "https://openrouter.ai/api/v1", "shape": "openai"}, "deepseek": {"base_url": "https://api.deepseek.com/v1", "shape": "openai"}, "groq": {"base_url": "https://api.groq.com/openai/v1", "shape": "openai"}, "mistral": {"base_url": "https://api.mistral.ai/v1", "shape": "openai"}, "other": {"base_url": None, "shape": "openai"}, } def _cloud_base_url(cloud: dict) -> str: """Return the effective base URL for a cloud service (registry entry or user-supplied 'other').""" service = cloud.get("service") or "claude" spec = _CLOUD_SERVICES.get(service) if not spec: return "" if service == "other": return (cloud.get("base_url") or "").strip().rstrip("/") return spec["base_url"] or "" # ── config storage ──────────────────────────────────────────────── def _config_path() -> Path: return Path(folder_paths.get_user_directory()) / "PromptChain" / "ai_config.json" def _load_config() -> dict: path = _config_path() if not path.exists(): return {} try: with open(path, "r", encoding="utf-8") as f: return _migrate_config(json.load(f)) except Exception as e: logger.warning("ai_config parse failed: %s", e) return {} def _migrate_config(cfg: dict) -> dict: """v1 shape was `{provider: 'claude', claude: {api_key, model}}`. v2 folds claude into the cloud-services object. Migrate lazily on load so existing users don't re-enter their key.""" if not isinstance(cfg, dict): return {} if cfg.get("provider") == "claude": claude = cfg.pop("claude", None) or {} cfg["provider"] = "cloud" cfg["cloud"] = { "service": "claude", "api_key": claude.get("api_key"), "model": claude.get("model") or "claude-haiku-4-5", } return cfg def _save_config(config: dict): path = _config_path() path.parent.mkdir(parents=True, exist_ok=True) atomic_write_json(path, config) def _sanitize(config: dict) -> dict: """Strip secrets before sending to the browser.""" cloud = config.get("cloud") or {} local = config.get("local") or {} return { "provider": config.get("provider") or None, "cloud": { "service": cloud.get("service") or "claude", "model": cloud.get("model") or "", "base_url": cloud.get("base_url") or "", "has_key": bool(cloud.get("api_key")), }, "local": { "base_url": local.get("base_url") or "", "model": local.get("model") or "", "auto_start": _local_auto_start(local), }, } def _local_auto_start(local: dict) -> bool: """Default ON — matches the shipped behavior; missing key in legacy configs reads as enabled.""" v = local.get("auto_start") return True if v is None else bool(v) # ── config endpoints ────────────────────────────────────────────── @routes.get("/promptchain/ai/config") async def _api_get_config(request): return web.json_response(_sanitize(_load_config())) @routes.post("/promptchain/ai/config") async def _api_set_config(request): body, err = await parse_json(request) if err: return err existing = _load_config() provider = body.get("provider") if provider not in (None, "", "cloud", "local"): return error_response("invalid provider") merged = dict(existing) merged["provider"] = provider or None # Records that the choice came from the user (even when it's None), so the # first-run Ollama auto-default never overrides a deliberate selection. merged["user_set"] = True # Cloud: preserve api_key when a new one isn't supplied (same pattern # as the CivitAI key — blank means "keep", empty-string-with-the-key # key present means "clear"). if "cloud" in body: c_in = body["cloud"] or {} c_cur = dict(existing.get("cloud") or {}) if "service" in c_in: svc = (c_in.get("service") or "").strip().lower() c_cur["service"] = svc if svc in _CLOUD_SERVICES else "claude" if "model" in c_in: c_cur["model"] = (c_in.get("model") or "").strip() if "base_url" in c_in: c_cur["base_url"] = (c_in.get("base_url") or "").strip().rstrip("/") if "api_key" in c_in: key = (c_in.get("api_key") or "").strip() if key: c_cur["api_key"] = key else: c_cur.pop("api_key", None) merged["cloud"] = c_cur if "local" in body: l_in = body["local"] or {} l_cur = dict(existing.get("local") or {}) if "base_url" in l_in: l_cur["base_url"] = (l_in.get("base_url") or "").strip().rstrip("/") if "model" in l_in: l_cur["model"] = (l_in.get("model") or "").strip() if "auto_start" in l_in: l_cur["auto_start"] = bool(l_in.get("auto_start")) merged["local"] = l_cur _save_config(merged) return web.json_response(_sanitize(merged)) # ── connection tests / model detection ──────────────────────────── @routes.post("/promptchain/ai/test") async def _api_test(request): """Validate a provider config. Body values take precedence over stored.""" body, err = await parse_json(request) if err: return err stored = _load_config() provider = body.get("provider") or stored.get("provider") if provider == "cloud": cloud = {**(stored.get("cloud") or {}), **(body.get("cloud") or {})} service = cloud.get("service") or "claude" api_key = (cloud.get("api_key") or "").strip() if not api_key: return web.json_response({"ok": False, "error": "API key missing"}) if service == "claude": model = cloud.get("model") or "claude-haiku-4-5" return web.json_response(await _test_claude(api_key, model)) base_url = _cloud_base_url(cloud) if not base_url: return web.json_response({"ok": False, "error": "Base URL missing"}) return web.json_response(await _test_openai_compat(base_url, api_key)) if provider == "local": local = {**(stored.get("local") or {}), **(body.get("local") or {})} base_url = (local.get("base_url") or "").strip().rstrip("/") if not base_url: return web.json_response({"ok": False, "error": "Base URL missing"}) return web.json_response(await _test_local(base_url)) return web.json_response({"ok": False, "error": "No provider selected"}) def _ollama_exe() -> str | None: """Locate the ollama binary. shutil.which resolves PATH as it was when ComfyUI booted — a freshly-installed Ollama won't be on that PATH until restart, so we also probe the per-OS default install location.""" exe = shutil.which("ollama") if exe: return exe candidates = [] if sys.platform == "win32": local = os.environ.get("LOCALAPPDATA") if local: candidates.append(Path(local) / "Programs" / "Ollama" / "ollama.exe") elif sys.platform == "darwin": candidates += [Path("/usr/local/bin/ollama"), Path("/opt/homebrew/bin/ollama"), Path("/Applications/Ollama.app/Contents/Resources/ollama")] else: candidates += [Path("/usr/local/bin/ollama"), Path("/usr/bin/ollama")] for c in candidates: if c.exists(): return str(c) return None def _winget_exe() -> str | None: """winget ships with App Installer on Win10 1809+/Win11 but isn't guaranteed (LTSC, stripped images, older builds). None == fall back to the manual download link.""" if sys.platform != "win32": return None return shutil.which("winget") def _spawn_ollama(exe: str) -> None: """Launch `ollama serve` fully detached so it outlives the ComfyUI process and doesn't inherit its console.""" if sys.platform == "win32": DETACHED = 0x00000008 # DETACHED_PROCESS NEW_GROUP = 0x00000200 # CREATE_NEW_PROCESS_GROUP subprocess.Popen( [exe, "serve"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=DETACHED | NEW_GROUP, close_fds=True, ) else: subprocess.Popen( [exe, "serve"], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True, close_fds=True, ) async def _ensure_ollama_running(ollama_root: str, exe: str | None) -> bool: """Return True if Ollama answers on its root URL, spawning it once if a binary is available and it isn't already up.""" if (await _test_local(f"{ollama_root}/v1")).get("ok") or await _is_ollama(ollama_root): return True if not exe: return False try: _spawn_ollama(exe) except Exception: return False for _ in range(12): await asyncio.sleep(0.5) if await _is_ollama(ollama_root): return True return False @routes.post("/promptchain/ai/wake-local") async def _api_wake_local(request): """Try to start Ollama when the configured local provider isn't answering. Windows installs put Ollama in PATH and auto-start on login by default, so the typical hit-rate case is 'user killed it via Task Manager' — `ollama serve` brings it back in <1s. No-op for non-local provider configs. Returns: - {ok: True, already_running: True} — was already up - {ok: True, started: True} — we spawned it; port responded - {ok: False, error: "..."} — not in PATH, or didn't come up """ cfg = _load_config() if cfg.get("provider") != "local": return web.json_response({"ok": False, "error": "Local provider not configured"}) local = cfg.get("local") or {} base_url = (local.get("base_url") or "").strip().rstrip("/") if not base_url: return web.json_response({"ok": False, "error": "No local base_url configured"}) if not _local_auto_start(local): return web.json_response({"ok": False, "error": "Auto-start disabled in settings."}) # Already up? Skip the spawn — avoids creating zombie processes when # the panel re-opens during a working session. if (await _test_local(base_url)).get("ok"): return web.json_response({"ok": True, "already_running": True}) exe = _ollama_exe() if not exe: return web.json_response({ "ok": False, "error": "Ollama not found in PATH. Install from https://ollama.com/download.", }) try: _spawn_ollama(exe) except Exception as e: return web.json_response({"ok": False, "error": f"Failed to spawn ollama: {e}"}) # Poll for up to ~6s — cold first-launch on Windows is usually <2s, # but the desktop app stack sometimes takes longer on first boot. for _ in range(12): await asyncio.sleep(0.5) if (await _test_local(base_url)).get("ok"): return web.json_response({"ok": True, "started": True}) return web.json_response({ "ok": False, "error": "Spawned ollama but it didn't answer on the configured port.", }) def _configured_ollama_root() -> str: """Ollama root the setup flow targets: the user's saved local base_url if they have one, else the default port. Independent of which provider is currently active so onboarding can probe before anything's saved.""" local = (_load_config().get("local") or {}) base_url = (local.get("base_url") or "").strip().rstrip("/") return _ollama_root(base_url) if base_url else _DEFAULT_OLLAMA_ROOT async def _list_ollama_model_names(ollama_root: str) -> list[str]: try: timeout = aiohttp.ClientTimeout(total=8) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(f"{ollama_root}/api/tags") as resp: if resp.status != 200: return [] data = await resp.json() return [m.get("name") or m.get("model") for m in (data.get("models") or []) if (m.get("name") or m.get("model"))] except Exception: return [] def _model_in_list(model: str, names: list[str]) -> bool: """Ollama appends `:latest` when a name carries no tag. A tagged request (qwen3-vl:8b-instruct) must match exactly; an untagged one matches any installed tag of the same base model.""" if ":" not in model: return any(n.split(":")[0] == model for n in names) return model in names or f"{model}:latest" in names # Well-known local LLM server ports, by product convention. We can only # *detect* these (and use them over the OpenAI-compat path) — unlike Ollama # we can't start, install, or pull models for them, so they're surfaced as # "already running? use it" rather than a guided setup. _LOCAL_PROBE_TARGETS = [ (11434, "Ollama"), (1234, "LM Studio"), (8080, "llama.cpp"), (5001, "KoboldCpp"), (1337, "Jan"), (8000, "vLLM / LocalAI"), ] async def _probe_local_servers() -> list[dict]: """Fan out a fast GET to each well-known port and report whichever answer an Ollama /api/tags or OpenAI-compat /v1/models. Refused ports fail instantly; the short timeout only bounds a port that accepts but hangs.""" async def probe(port: int, label: str) -> dict | None: root = f"http://localhost:{port}" try: timeout = aiohttp.ClientTimeout(total=1.0) async with aiohttp.ClientSession(timeout=timeout) as session: try: async with session.get(f"{root}/api/tags") as r: if r.status == 200: return {"port": port, "kind": "ollama", "label": "Ollama", "base_url": f"{root}/v1"} except Exception: pass async with session.get(f"{root}/v1/models") as r: if r.status == 200: return {"port": port, "kind": "openai", "label": label, "base_url": f"{root}/v1"} except Exception: return None return None found = await asyncio.gather(*[probe(p, l) for p, l in _LOCAL_PROBE_TARGETS]) return [f for f in found if f] @routes.get("/promptchain/ai/setup-status") async def _api_setup_status(request): """One-shot snapshot for the onboarding/panel AI-setup UI. Triggered on open (event-driven) — never polled.""" ollama_root = _configured_ollama_root() running, detected = await asyncio.gather( _is_ollama(ollama_root), _probe_local_servers(), ) names = await _list_ollama_model_names(ollama_root) if running else [] return web.json_response({ "platform": sys.platform, "recommended_model": RECOMMENDED_LOCAL_MODEL, "ollama_installed": _ollama_exe() is not None, "ollama_running": running, "model_present": _model_in_list(RECOMMENDED_LOCAL_MODEL, names), "winget_available": _winget_exe() is not None, "installed_models": names, # Non-Ollama OpenAI-compatible servers we found listening locally. "detected_servers": [d for d in detected if d["kind"] != "ollama"], }) @routes.post("/promptchain/ai/auto-configure") async def _api_auto_configure(request): """First-run default: if the user hasn't configured a provider yet and Ollama is already serving the recommended model, select Local so the assistant works with zero setup. Skips once a provider is set or the user has made any explicit choice (`user_set`) so None is never overridden.""" cfg = _load_config() if cfg.get("provider") or cfg.get("user_set"): return web.json_response({"configured": False, "reason": "already chosen"}) ollama_root = _configured_ollama_root() if not await _is_ollama(ollama_root): return web.json_response({"configured": False, "reason": "ollama offline"}) names = await _list_ollama_model_names(ollama_root) if not _model_in_list(RECOMMENDED_LOCAL_MODEL, names): return web.json_response({"configured": False, "reason": "recommended model missing"}) _save_config({ "provider": "local", "local": {"base_url": f"{ollama_root}/v1", "model": RECOMMENDED_LOCAL_MODEL}, }) return web.json_response({"configured": True, "provider": "local", "model": RECOMMENDED_LOCAL_MODEL}) async def _open_sse(request) -> web.StreamResponse: resp = web.StreamResponse(headers={ "Content-Type": "text/event-stream", "Cache-Control": "no-cache", "X-Accel-Buffering": "no", }) await resp.prepare(request) return resp async def _sse_send(resp: web.StreamResponse, obj: dict) -> None: await resp.write(b"data: " + json.dumps(obj).encode("utf-8") + b"\n\n") @routes.post("/promptchain/ai/pull-model") async def _api_pull_model(request): """Proxy Ollama's /api/pull progress stream to the browser as SSE so the setup UI can render a live progress bar. Spawns Ollama first if a binary exists but the server is down.""" body, err = await parse_json(request) if err: return err model = (body.get("model") or RECOMMENDED_LOCAL_MODEL).strip() ollama_root = _configured_ollama_root() if not await _ensure_ollama_running(ollama_root, _ollama_exe()): return web.json_response( {"error": "Ollama isn't running and couldn't be started. Install it first."}, status=409) resp = await _open_sse(request) try: # No total timeout — a multi-GB pull legitimately runs for minutes. timeout = aiohttp.ClientTimeout(total=None, sock_read=120) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(f"{ollama_root}/api/pull", json={"name": model, "stream": True}) as upstream: if upstream.status != 200: detail = (await upstream.text())[:200] await _sse_send(resp, {"error": f"HTTP {upstream.status}: {detail}"}) return resp async for raw in upstream.content: line = raw.strip() if not line: continue try: evt = json.loads(line) except json.JSONDecodeError: continue await _sse_send(resp, evt) await _sse_send(resp, {"done": True}) except ConnectionResetError: pass # client navigated away mid-pull; Ollama keeps the download except Exception as e: logger.warning("pull-model stream failed", exc_info=True) try: await _sse_send(resp, {"error": str(e)}) except Exception: pass return resp @routes.post("/promptchain/ai/install-ollama") async def _api_install_ollama(request): """Install Ollama via winget (Windows only) and stream its output as SSE. Every other platform falls back to the manual download link in the UI — we don't attempt unattended installs there (macOS is a .app drag, Linux needs sudo for the systemd service).""" winget = _winget_exe() if not winget: return web.json_response( {"error": "Automatic install needs winget (Windows). " "Download Ollama from https://ollama.com/download."}, status=400) resp = await _open_sse(request) try: proc = await asyncio.create_subprocess_exec( winget, "install", "--id", "Ollama.Ollama", "-e", "--silent", "--accept-source-agreements", "--accept-package-agreements", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, stdin=asyncio.subprocess.DEVNULL, ) assert proc.stdout is not None async for raw in proc.stdout: text = raw.decode("utf-8", "replace").rstrip() if text: await _sse_send(resp, {"line": text}) code = await proc.wait() # winget exits 0 on success. A fresh install isn't on the running # ComfyUI's PATH, but _ollama_exe() probes the install dir directly # so setup-status will see it without a restart. await _sse_send(resp, {"done": True, "code": code, "ok": code == 0, "installed": _ollama_exe() is not None}) except ConnectionResetError: pass except Exception as e: logger.warning("install-ollama failed", exc_info=True) try: await _sse_send(resp, {"error": str(e)}) except Exception: pass return resp async def _test_claude(api_key: str, model: str) -> dict: try: timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, json={ "model": model, "max_tokens": 8, "messages": [{"role": "user", "content": "ping"}], }, ) as resp: if resp.status == 200: return {"ok": True} data = await resp.json() msg = (data.get("error") or {}).get("message") or f"HTTP {resp.status}" return {"ok": False, "error": msg} except Exception as e: return {"ok": False, "error": f"Connection failed: {e}"} async def _test_local(base_url: str) -> dict: # Prefer /v1/models (standard OpenAI shape). Fall back to Ollama's # /api/tags when /v1/models is missing — that's the telltale sign of # raw Ollama running without the /v1 shim. try: timeout = aiohttp.ClientTimeout(total=10) async with aiohttp.ClientSession(timeout=timeout) as session: try: async with session.get(f"{base_url}/models") as resp: if resp.status == 200: return {"ok": True} except Exception: pass ollama = _ollama_root(base_url) async with session.get(f"{ollama}/api/tags") as resp: if resp.status == 200: return {"ok": True} return {"ok": False, "error": f"HTTP {resp.status}"} except Exception as e: return {"ok": False, "error": f"Connection failed: {e}"} async def _test_openai_compat(base_url: str, api_key: str) -> dict: try: timeout = aiohttp.ClientTimeout(total=15) async with aiohttp.ClientSession(timeout=timeout) as session: headers = {"Authorization": f"Bearer {api_key}"} async with session.get(f"{base_url}/models", headers=headers) as resp: if resp.status == 200: return {"ok": True} text = await resp.text() return {"ok": False, "error": f"HTTP {resp.status}: {text[:160]}"} except Exception as e: return {"ok": False, "error": f"Connection failed: {e}"} def _ollama_root(base_url: str) -> str: """Derive Ollama root (http://host:11434) from a config base_url that typically ends in /v1. If the user's URL doesn't end in /v1 we return it unchanged — Ollama's native endpoints are siblings of /v1 so this only helps when the user configured the OpenAI-compat path.""" if base_url.endswith("/v1"): return base_url[:-3] return base_url @routes.post("/promptchain/ai/detect-models") async def _api_detect_models(request): """Try to enumerate models at a base URL. Works for Ollama (native /api/tags) and any server exposing /v1/models (OpenAI-compat).""" body, err = await parse_json(request) if err: return err base_url = (body.get("base_url") or "").strip().rstrip("/") if not base_url: return web.json_response({"models": [], "source": "none"}) # Optional bearer token for cloud providers (OpenAI, Grok, ...). # Local servers don't expect auth; absent key = unauthenticated GET. api_key = (body.get("api_key") or "").strip() auth_headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} timeout = aiohttp.ClientTimeout(total=10) try: async with aiohttp.ClientSession(timeout=timeout) as session: # Ollama native first — richer (capability flags). ollama_root = _ollama_root(base_url) try: async with session.get(f"{ollama_root}/api/tags") as resp: if resp.status == 200: data = await resp.json() models = [] for m in (data.get("models") or []): name = m.get("name") or m.get("model") if name: models.append({"name": name}) if models: # Second pass to flag vision-capable models. models = await _flag_ollama_vision(session, ollama_root, models) return web.json_response({"models": models, "source": "ollama"}) except Exception: pass # Generic OpenAI-compat fallback. try: async with session.get(f"{base_url}/models", headers=auth_headers) as resp: if resp.status == 200: data = await resp.json() items = data.get("data") or data.get("models") or [] models = [] for m in items: if isinstance(m, dict): name = m.get("id") or m.get("name") else: name = str(m) if name: models.append({"name": name}) return web.json_response({"models": models, "source": "openai"}) except Exception: pass except Exception as e: return web.json_response({"models": [], "error": str(e)}) return web.json_response({"models": []}) async def _flag_ollama_vision(session, ollama_root: str, models: list[dict]) -> list[dict]: """Probe each model's capabilities via /api/show. Best-effort — on any failure we leave vision unset (UI treats that as unknown/permissive).""" async def probe(m): try: async with session.post(f"{ollama_root}/api/show", json={"name": m["name"]}) as resp: if resp.status == 200: data = await resp.json() caps = data.get("capabilities") or [] if "vision" in caps: m["vision"] = True except Exception: pass return m return await asyncio.gather(*[probe(m) for m in models]) # ── active request tracking ─────────────────────────────────────── _active_requests: dict[str, asyncio.Task] = {} # Per-request reasoning-char counter set by streaming endpoints when the # upstream surfaces a `reasoning`/`reasoning_content` delta. Lets callers # distinguish a genuinely empty response from the "thought hard then # emitted nothing" failure mode common to abliterated/thinking-mode # variants — those return empty body with non-empty reasoning, and # benefit from a retry with an output-forcing nudge. _request_reasoning_chars: dict[str, int] = {} def _cleanup_request(request_id: str): _active_requests.pop(request_id, None) _request_reasoning_chars.pop(request_id, None) @routes.post("/promptchain/ai/cancel") async def _api_cancel(request): body, err = await parse_json(request) if err: return err request_id = body.get("request_id") or "" task = _active_requests.get(request_id) if task and not task.done(): task.cancel() return web.json_response({"status": "ok"}) # ── grounding block assembly ────────────────────────────────────── def _build_grounding(model_hash: str) -> dict: cfg = model_settings.load(model_hash) if model_hash else None if not cfg: return {} keys = ( "display_name", "model_name", "version", "architecture", "family", "author", "description", "tags", "url", "trigger", "quality_position", "negative", "notes", "tag_sources", "tag_format", "prompt_style", "default_prompt_id", ) return {k: cfg.get(k) for k in keys if cfg.get(k) is not None} def _list_arch_prompts(arch: str | None) -> list[dict]: if not arch: return [] from . import prompts as _prompts try: return _prompts.list_prompts(architecture=arch) except Exception: logger.debug("failed to load prompts for arch %s", arch, exc_info=True) return [] def _style_full_name(p: dict) -> str: return f"{p.get('category', 'Misc')} > {p.get('name', 'Unnamed')}" def _find_prompt_by_full_name(prompts_list: list[dict], full_name: str) -> dict | None: target = full_name.strip().lower() for p in prompts_list: if _style_full_name(p).lower() == target: return p return None def _find_prompt_by_id(prompts_list: list[dict], prompt_id: str) -> dict | None: for p in prompts_list: if p.get("id") == prompt_id: return p return None # ── style template assembly ──────────────────────────────────────────── # Templates in `data/prompts/*.json` come in three shapes (per the B0 # probe of all 587 templates): # 1. Flux/Illustrious: "// Your Tags\n{cursor}\n\n// SectionHeader\n # " with optional trailing "\n\nNegative Prompt:\n" # 2. Pony quality: "score_9, score_8_up, score_7_up, {cursor}" — cursor # inline among modifiers. # 3. Pony style: bare modifier line, no cursor, no wrapper. # All three flatten to (positive_modifiers, negative_modifiers) for # server-side `// Style:` section injection. _TEMPLATE_YOUR_TAGS_BLOCK = re.compile( r"^\s*//\s*Your\s+Tags\s*\n\s*\{cursor\}\s*\n\s*\n", re.IGNORECASE, ) _TEMPLATE_LEADING_HEADER = re.compile(r"^\s*//\s*[^\n]+\n") _TEMPLATE_NEG_MARKER = re.compile( r"\n\s*Negative\s+Prompt\s*:\s*\n", re.IGNORECASE, ) def _parse_style_template_text(text: str) -> tuple[list[str], list[str]]: """Pull positive modifiers and embedded negatives out of a prompt template's `text` field. Both lists are bare comma-split tokens suitable for direct insertion into output sections. Returns ([], []) when the template body is empty after wrapper- stripping (template was just `{cursor}` — nothing to inject).""" if not text: return [], [] body = text body = _TEMPLATE_YOUR_TAGS_BLOCK.sub("", body, count=1) body = body.replace("{cursor}", "").strip() if not body: return [], [] neg_split = _TEMPLATE_NEG_MARKER.split(body, maxsplit=1) positive_part = neg_split[0].strip() negative_part = neg_split[1].strip() if len(neg_split) > 1 else "" positive_part = _TEMPLATE_LEADING_HEADER.sub("", positive_part, count=1) positive_part = positive_part.strip() pos_tokens = [ t.strip() for t in _split_prompt_tokens(positive_part) if t.strip() ] neg_tokens = [ t.strip() for t in _split_prompt_tokens(negative_part) if t.strip() ] return pos_tokens, neg_tokens def _build_style_section(template: dict) -> dict | None: """Construct a `// Style: ` section dict from a template. Returns None if the template has no usable positive modifiers (after stripping wrapper + cursor). `body_text` holds the joined positive prose (mirrors what natlang assembly emits) so prose-mode output keeps the template body verbatim rather than re-joining comma-fragments — matters when the template body has multi-clause prose where ", ".join would alter spacing.""" name = (template.get("name") or "").strip() or "Style" pos_tokens, _ = _parse_style_template_text(template.get("text") or "") if not pos_tokens: return None return { "header": f"// Style: {name}", "tokens": pos_tokens, "body_text": ", ".join(pos_tokens), "is_style_injected": True, } def _replace_or_append_style_section(sections: list[dict], style_section: dict) -> list[dict]: """Replace any existing `// Style:` section in `sections` with the new one, or append it as the last positive section if none exists. Negative Prompt sections always stay last in output_text assembly, so appending here puts Style after Setting/Scene but before Neg.""" out: list[dict] = [] inserted = False for s in sections: if (not s.get("is_negative") and _section_key_from_header(s.get("header") or "") == "style"): if not inserted: out.append(style_section) inserted = True continue out.append(s) if inserted: return out # Insert before the first negative section (so Style stays in the # positive block) or at the end if no negs exist. final: list[dict] = [] placed = False for s in out: if not placed and s.get("is_negative"): final.append(style_section) placed = True final.append(s) if not placed: final.append(style_section) return final def _merge_template_negatives(sections: list[dict], template: dict) -> list[dict]: """Pull the embedded `Negative Prompt:` line out of a style template and merge into the output's Negative Prompt section. Deduped by canonical (lowercase, weighted-stripped) form so user-typed negs aren't overwritten and re-applying the same style doesn't duplicate. No-op if the template carries no negatives or every negative is already present. If there's no Negative Prompt section in `sections`, one is created at the end.""" _, neg_tokens = _parse_style_template_text(template.get("text") or "") if not neg_tokens: return sections target: dict | None = None for s in sections: if s.get("is_negative"): target = s break if target is None: target = { "header": "Negative Prompt:", "tokens": [], "is_negative": True, } sections = sections + [target] existing: set[str] = set() for t in target.get("tokens") or []: canon = _canonicalize_token(t).lower() if canon: existing.add(canon) m = _WEIGHTED_TOKEN_RE.match(t) if m: inner = _canonicalize_token(m.group(1)).lower() if inner: existing.add(inner) for t in neg_tokens: canon = _canonicalize_token(t).lower() if canon and canon in existing: continue target["tokens"].append(t) if canon: existing.add(canon) return sections _STYLE_HEADER_RE = re.compile(r"^//\s*style\b\s*:?", re.IGNORECASE) _GENERIC_HEADER_RE = re.compile(r"^//\s*\w[\w\s\-]*\s*:") def _swap_style_template_in_prompt(prompt: str, template: dict) -> str: """Replace the `// Style: ` section (header + body) and the `Negative Prompt:` section content with the given template's full contents. Used as a post-pass after rails/hybrid finishes a style apply — the rails dispatcher works at sentence-level and doesn't promote a style intent to a full template swap, so leftover scene- shaped sentences and stale negatives persist without this pass. If the prompt has no `// Style:` section, one is inserted before the Negative Prompt section (or at the end). If the prompt has no Negative Prompt section, one is appended. No-op when the template carries no positive tokens — treats that as "not really a usable template" rather than wiping the section. """ if not template or not prompt: return prompt pos_tokens, neg_tokens = _parse_style_template_text( template.get("text") or "" ) if not pos_tokens and not neg_tokens: return prompt name = (template.get("name") or "").strip() new_style_header = f"// Style: {name}" if name else "// Style:" new_style_body = ", ".join(pos_tokens) if pos_tokens else "" new_neg_body = ", ".join(neg_tokens) if neg_tokens else "" lines = prompt.splitlines() out: list[str] = [] i = 0 style_replaced = False neg_replaced = False def _is_section_boundary(s: str) -> bool: return bool(_GENERIC_HEADER_RE.match(s.strip())) or s.strip() == "Negative Prompt:" while i < len(lines): line = lines[i] stripped = line.strip() if _STYLE_HEADER_RE.match(stripped): # Swap entire style section: header + body until next # section boundary or EOF. Ensure a blank line follows # so it visually separates from the next section. out.append(new_style_header) if new_style_body: out.append(new_style_body) i += 1 while i < len(lines) and not _is_section_boundary(lines[i]): i += 1 # If a next section follows, leave a blank line gap. if i < len(lines): out.append("") style_replaced = True continue if stripped == "Negative Prompt:": out.append("Negative Prompt:") if new_neg_body: out.append(new_neg_body) i += 1 while i < len(lines) and not _is_section_boundary(lines[i]): i += 1 if i < len(lines): out.append("") neg_replaced = True continue out.append(line) i += 1 if not style_replaced and (new_style_body or new_style_header): # No style section existed. Insert before Negative Prompt (if # we already emitted it via the negative branch) or at end. if neg_replaced: for j in range(len(out) - 1, -1, -1): if out[j].strip() == "Negative Prompt:": if out and out[max(0, j - 1)].strip(): out.insert(j, "") out.insert(j, new_style_header) if new_style_body: out.insert(j + 1, new_style_body) break else: if out and out[-1].strip(): out.append("") out.append(new_style_header) if new_style_body: out.append(new_style_body) if not neg_replaced and new_neg_body: if out and out[-1].strip(): out.append("") out.append("Negative Prompt:") out.append(new_neg_body) return "\n".join(out) _SECTION_HEADER_RE_MC = re.compile(r"^//\s*(\w[\w\-]*)\s*[:](.*)$", re.IGNORECASE) def _parse_existing_prompt_state(node_prompt: str) -> dict: """Light parse of an existing `// Section:` prompt. Used by the multi-char edit compose path to preserve outfit / style / negs from the existing prompt when rebuilding around a new character. Returns: { 'character_sections': [(header_line, body_text), ...], 'outfit_body': str, # combined Outfit section body 'pose_body': str, # combined Pose section body 'scene_body': str, 'style_body': str, 'quality_body': str, 'negative_body': str, } """ out = { "character_sections": [], "outfit_body": "", "pose_body": "", "scene_body": "", "style_header_line": "", "style_body": "", "quality_body": "", "negative_body": "", } if not (node_prompt or "").strip(): return out lines = node_prompt.splitlines() i = 0 while i < len(lines): line = lines[i] stripped = line.strip() m = _SECTION_HEADER_RE_MC.match(stripped) if m: concept = m.group(1).lower() body_lines: list[str] = [] i += 1 while i < len(lines): nxt = lines[i].strip() if _SECTION_HEADER_RE_MC.match(nxt) or nxt == "Negative Prompt:": break if nxt: body_lines.append(lines[i]) i += 1 body = "\n".join(body_lines).strip() if concept == "character": out["character_sections"].append((stripped, body)) elif concept == "outfit": out["outfit_body"] = ( (out["outfit_body"] + "\n" + body).strip() if out["outfit_body"] else body ) elif concept == "pose": out["pose_body"] = ( (out["pose_body"] + "\n" + body).strip() if out["pose_body"] else body ) elif concept == "scene": out["scene_body"] = ( (out["scene_body"] + "\n" + body).strip() if out["scene_body"] else body ) elif concept == "style": out["style_body"] = ( (out["style_body"] + "\n" + body).strip() if out["style_body"] else body ) if not out["style_header_line"]: out["style_header_line"] = stripped elif concept == "quality": out["quality_body"] = ( (out["quality_body"] + "\n" + body).strip() if out["quality_body"] else body ) continue if stripped == "Negative Prompt:": i += 1 neg_lines: list[str] = [] while i < len(lines): nxt = lines[i].strip() if _SECTION_HEADER_RE_MC.match(nxt) or nxt == "Negative Prompt:": break if nxt: neg_lines.append(lines[i]) i += 1 out["negative_body"] = "\n".join(neg_lines).strip() continue i += 1 return out async def _maybe_compose_multichar_edit( request: web.Request, node_prompt: str, user_request: str, character_queries: list[str] | None, model_hash: str, bios: list[dict], ) -> str | None: """Edit-mode multi-character compose path. Triggers when the chat agent passed 2+ entries in character_queries but the existing node_prompt has 0-1 `// Character:` sections — i.e., the user is ADDING a character. The agent's request text is unreliable in this transition (treats new chars as scene content). Bypass the hybrid path entirely: 1. Build a planner-shaped dict from character_queries (cast), existing prompt state (preserved outfit/style/negs), and the user request decompose (interaction verb + scene_text). 2. compose_from_plan → structured `// Section:` body 3. compose_scene_paragraph → inline-bundle cinematic prose 4. Re-attach the preserved negative block verbatim. Returns the final prompt as cinematic prose, or None if conditions don't match — caller falls through to normal hybrid path. Single- character paths are untouched. """ cleaned_queries = [ q.strip() for q in (character_queries or []) if q and q.strip() ] if len(cleaned_queries) < 2: return None existing_char_count = sum( 1 for line in (node_prompt or "").splitlines() if line.strip().lower().startswith("// character:") ) if existing_char_count >= 2: # Already multi-char; let hybrid + polish post-pass handle. return None logger.info( "multichar-edit: trigger queries=%r existing_chars=%d", cleaned_queries, existing_char_count, ) existing_state = _parse_existing_prompt_state(node_prompt) # Resolve each character via the tag-builder DB so we get series + # base_natlang. Bios passed in by the chat agent's preflight may # already have this, but re-resolve for safety so a missing field # doesn't drop a character. import os as _os import sqlite3 db_path = _os.path.join( _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))), "data", "tag-builder", "tag-builder.db", ) if not _os.path.exists(db_path): logger.warning("multichar-edit: tag-builder.db not at %r — fallback", db_path) return None resolved_bios: list[dict] = [] try: conn = sqlite3.connect(db_path) conn.row_factory = sqlite3.Row try: for q in cleaned_queries: # Normalize '-'<->'_' on BOTH sides: the chat agent emits # underscore-canonical Danbooru tags (chun_li), but a few # characters are stored hyphenated (chun-li). Without this # the lookup misses, <2 resolve, and the whole multi-char # compose silently falls back to hybrid. No two distinct # character tags differ only by separator, so this is safe. row = conn.execute( "SELECT tag, display, series, base_natlang " "FROM characters WHERE " "LOWER(REPLACE(tag, '-', '_')) = " "LOWER(REPLACE(?, '-', '_')) LIMIT 1", (q,), ).fetchone() if row: resolved_bios.append(dict(row)) else: logger.info( "multichar-edit: tag=%r not in characters table — skipping", q, ) finally: conn.close() except Exception: logger.warning("multichar-edit: bio lookup failed", exc_info=True) return None if len(resolved_bios) < 2: logger.info( "multichar-edit: only %d/%d characters resolved — fallback", len(resolved_bios), len(cleaned_queries), ) return None # Run decompose on the user request to extract interaction + scene. # Pose intents map to the interaction verb. Scene intents map to # scene_text. Other intents are intentionally ignored — the # composer derives outfits from existing state + DB defaults. try: _pc_root = _os.path.dirname(_os.path.dirname(_os.path.abspath(__file__))) if _pc_root not in sys.path: sys.path.insert(0, _pc_root) from scripts.natlang_decompose_probe import decompose except Exception: logger.warning("multichar-edit: decompose import failed", exc_info=True) return None try: decomposed, _decompose_raw = await decompose(user_request) except Exception: logger.warning("multichar-edit: decompose call failed", exc_info=True) return None intents = (decomposed or {}).get("intents") or [] interaction_text = "" scene_text = "" for it in intents: c = (it.get("concept") or "").lower() t = (it.get("text") or "").strip() if not t: continue if c in ("pose", "action") and not interaction_text: interaction_text = t elif c == "scene" and not scene_text: scene_text = t # Heuristic fallback when decompose doesn't isolate the interaction: # the user's literal request usually contains the verb. Common # interaction verbs surface here even when decompose is noisy. if not interaction_text: lc = (user_request or "").lower() for verb in ( "fighting", "sparring", "dueling", "racing", "dancing", "kissing", "hugging", "embracing", "chasing", "wrestling", "boxing", "punching", "kicking", "blocking", ): if verb in lc: interaction_text = verb break # Decide outfit_text per character. Preserve existing prompt's # outfit body for the EXISTING character (matched by display name # in any of the existing `// Character:` headers). New characters # get outfit_text="" + outfit_source="canon" (use their default). existing_displays_lc: set[str] = set() for header, _body in existing_state["character_sections"]: # Header shape: "// Character: ()" m = re.match(r"//\s*character\s*:\s*(.+?)(?:\s*\([^)]+\))?\s*$", header, re.IGNORECASE) if m: existing_displays_lc.add(m.group(1).strip().lower()) per_character: list[dict] = [] for bio in resolved_bios: display = (bio.get("display") or "").strip() tag = (bio.get("tag") or "").strip() if display.lower() in existing_displays_lc and existing_state["outfit_body"]: # Existing character — carry forward the outfit verbatim outfit_text = existing_state["outfit_body"] outfit_source = "literal" else: # New character — use her default canon outfit outfit_text = "" outfit_source = "canon" per_character.append({ "tag": tag, "outfit_text": outfit_text, "outfit_source": outfit_source, "pose_text": "", }) cast = [ {"tag": b.get("tag"), "display": b.get("display")} for b in resolved_bios ] plan_dict = { "cast": cast, "per_character": per_character, "interaction": interaction_text, "scene_text": scene_text, "style_text": "", # style is preserved separately, see below "lighting_text": "", } logger.info( "multichar-edit: plan cast=%s interaction=%r scene=%r", [c["display"] for c in cast], interaction_text or "(none)", scene_text or "(none)", ) # compose_from_plan builds the structured `// Section:` body. We # then INJECT the preserved style + negs back in (compose_from_plan # doesn't preserve existing prompt state — it builds from the plan # only) before polishing. try: from scripts.natlang_compose_from_plan import compose_from_plan structured = compose_from_plan( plan_dict, resolved_bios, db_path, default_negative_block="", # we re-attach existing negs below ) except Exception: logger.warning("multichar-edit: compose_from_plan failed", exc_info=True) return None if not (structured or "").strip(): logger.warning("multichar-edit: compose_from_plan returned empty") return None # Polish to inline-bundle prose. Skip the style step in the # composer — we re-attach the preserved `// Style: ` section # AFTER the prose so the style-template swap pipeline (which # operates on section headers) continues to work on multi-char # output. Style as a global modifier doesn't need to be inside the # subject prose for the encoder; only the per-character bundles + # scene need that binding. try: from scripts.natlang_scene_composer import compose_scene_paragraph paragraph, _raw = await compose_scene_paragraph( structured, include_style_in_prose=False, ) except Exception: logger.warning("multichar-edit: scene composer failed", exc_info=True) return None if not (paragraph or "").strip(): logger.warning("multichar-edit: scene composer returned empty") return None # Sanity check: every named character must survive. char_display_list = [b.get("display") for b in resolved_bios] dropped = [ d for d in char_display_list if d and d.lower() not in paragraph.lower() ] if dropped: logger.warning( "multichar-edit: composer dropped %s — fallback to hybrid", dropped, ) return None # Assemble final: prose + // Style: + Negative Prompt. pieces: list[str] = [paragraph.rstrip()] style_header = existing_state.get("style_header_line") or "" style_body = existing_state.get("style_body") or "" if style_header and style_body: pieces.append(f"{style_header}\n{style_body.strip()}") elif style_body: pieces.append(f"// Style:\n{style_body.strip()}") neg = existing_state["negative_body"].strip() if neg: pieces.append(f"Negative Prompt:\n{neg}") final = "\n\n".join(pieces) logger.info( "multichar-edit: composed %d -> %d chars (cast=%s, style_split=%s)", len(node_prompt or ""), len(final), char_display_list, bool(style_header or style_body), ) return final async def _polish_multichar_to_prose(final_prompt: str, pipeline_name: str = "hybrid-v1") -> str: """When `final_prompt` has 2+ `// Character:` sections, fold the sectioned body into ONE flowing cinematic paragraph via the scene composer. Single-character prompts are returned verbatim (sectioned form is KB-rich and structure-preserving is preferred). Modern natlang T2I encoders (T5-XXL on Flux, Qwen3-4B on Z-Image, Qwen2.5-VL on Qwen-Image) don't honor `// Section:` boundaries — a multi-character prompt with shared `// Outfit:` and `// Pose:` sections leaves the encoder with no way to bind features per subject. Inline-bundle prose with spatial anchors (`On the left, A wears X, doing Y. On the right, B wears Z…`) is what these models train on and render reliably. Sanity check: every named character must survive the rewrite — if the composer drops one (LLM nondeterminism), keep the structured form rather than ship a broken paragraph. Negative Prompt block is preserved verbatim and re-attached. No-op when the composer isn't importable, the prompt is empty, or only one character is in scope. Errors fall back to input. """ if not final_prompt or not final_prompt.strip(): return final_prompt char_section_count = sum( 1 for line in final_prompt.splitlines() if line.strip().lower().startswith("// character:") ) if char_section_count < 2: return final_prompt try: from scripts.natlang_scene_composer import compose_scene_paragraph except Exception: logger.warning( "ai-patch (%s): scene composer unavailable for multi-char polish", pipeline_name, ) return final_prompt if "\n\nNegative Prompt:" in final_prompt: positive, _, negative = final_prompt.partition("\n\nNegative Prompt:") elif final_prompt.startswith("Negative Prompt:"): positive = "" negative = final_prompt[len("Negative Prompt:"):] else: positive = final_prompt negative = "" char_names: list[str] = [] for line in positive.splitlines(): if line.strip().lower().startswith("// character:"): name = line.split(":", 1)[1].strip() name = re.sub(r"\s*\([^)]+\)\s*$", "", name).strip() if name: char_names.append(name) try: paragraph, _raw = await compose_scene_paragraph(positive) except Exception: logger.warning( "ai-patch (%s): scene composer raised", pipeline_name, exc_info=True, ) return final_prompt if not (paragraph or "").strip(): logger.warning( "ai-patch (%s): scene composer returned empty — keeping structured", pipeline_name, ) return final_prompt dropped = [ name for name in char_names if name and name.lower() not in paragraph.lower() ] if dropped: logger.warning( "ai-patch (%s): multi-char polish dropped %s — keeping structured", pipeline_name, dropped, ) return final_prompt logger.info( "ai-patch (%s): multi-char polish %d -> %d chars (chars=%s)", pipeline_name, len(positive), len(paragraph), char_names, ) if negative.strip(): return paragraph.rstrip() + "\n\nNegative Prompt:" + negative return paragraph def _find_style_intent(intents: list[dict]) -> dict | None: """Find the last `concept=style` intent in the trace that resolved against a known template (resolved_source=='style'). Returns the intent dict (with `resolved_match_name`) or None. Last-wins semantics: if a turn applied two styles, the final state is the second swap.""" found = None for it in intents or []: if ((it.get("concept") or "").lower() == "style" and (it.get("resolved_source") or "").lower() == "style" and (it.get("op") or "replace").lower() in ("replace", "add") and it.get("resolved_match_name")): found = it return found _OUTER_WEIGHT_RE = re.compile(r"^\((.+):(\d+\.?\d*)\)$") def _format_output(text: str, tag_format: str) -> str: """Post-process a single tag token. Model emits Danbooru-canonical underscored tags; convert to spaces when the target model expects spaces, AND escape any literal parens that come from canonical-tag franchise suffixes (`mythra_(xenoblade)` → `mythra \\(xenoblade\\)`). The outermost `(content:weight)` weight-wrapper is preserved as-is — its parens are A1111 syntax, not literal characters. Only the parens *inside* the tag name get escaped, because SD's parser otherwise reads them as nested weight markers and the whole token structure breaks.""" if tag_format != "spaces" or not text: return text text = text.replace("_", " ") m = _OUTER_WEIGHT_RE.match(text) if m: # Weighted form: only escape parens INSIDE the content. content = _escape_unescaped_parens(m.group(1)) return f"({content}:{m.group(2)})" # Non-weighted token: escape all unescaped parens. return _escape_unescaped_parens(text) def _escape_unescaped_parens(s: str) -> str: """Add `\\` before `(` / `)` chars not already preceded by `\\`. Idempotent: applying twice gives the same result.""" s = re.sub(r"(? str: images = images or [] if provider == "cloud": cloud = config.get("cloud") or {} service = cloud.get("service") or "claude" api_key = (cloud.get("api_key") or "").strip() model = (cloud.get("model") or "").strip() if not api_key or not model: return "" if service == "claude": return await _claude_complete(request_id, api_key, model, system, user_request, images) base_url = _cloud_base_url(cloud) if not base_url: return "" return await _openai_compat_complete( request_id, base_url, model, system, user_request, images, api_key=api_key, ) if provider == "local": local = config.get("local") or {} base_url = (local.get("base_url") or "").strip().rstrip("/") model = (local.get("model") or "").strip() if not base_url or not model: return "" return await _openai_compat_complete( request_id, base_url, model, system, user_request, images, ) return "" async def _claude_complete(request_id, api_key, model, system, user, images) -> str: content = [] for img in images: data = img.get("data") media = img.get("media_type") or "image/jpeg" if not data: continue content.append({ "type": "image", "source": {"type": "base64", "media_type": media, "data": data}, }) content.append({"type": "text", "text": user}) payload = { "model": model, "max_tokens": _MAX_TOKENS, "system": system, "messages": [{"role": "user", "content": content}], } timeout = aiohttp.ClientTimeout(connect=_CONNECT_TIMEOUT, sock_read=_READ_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, json=payload, ) as resp: if resp.status != 200: err = (await resp.text())[:200] logger.warning("claude_complete[%s] HTTP %d: %s", request_id, resp.status, err) return "" data = await resp.json() blocks = data.get("content") or [] return "".join(b.get("text", "") for b in blocks if b.get("type") == "text") async def _openai_compat_complete( request_id, base_url, model, system, user, images, api_key: str | None = None, ) -> str: content = [] for img in images: data = img.get("data") media = img.get("media_type") or "image/jpeg" if not data: continue content.append({ "type": "image_url", "image_url": {"url": f"data:{media};base64,{data}"}, }) content.append({"type": "text", "text": user}) messages = [ {"role": "system", "content": system}, {"role": "user", "content": content if images else user}, ] payload = { "model": model, "messages": messages, "stream": False, "max_tokens": _MAX_TOKENS, } headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} timeout = aiohttp.ClientTimeout(connect=_CONNECT_TIMEOUT, sock_read=_READ_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post( f"{base_url}/chat/completions", json=payload, headers=headers, ) as resp: if resp.status != 200: err = (await resp.text())[:200] logger.warning("openai_complete[%s] HTTP %d: %s", request_id, resp.status, err) return "" data = await resp.json() choices = data.get("choices") or [] if not choices: return "" msg = choices[0].get("message") or {} return msg.get("content") or "" async def _run_generation(request_id: str, provider: str, config: dict, system: str, user_request: str, images: list[dict]) -> str: """Streams from the configured provider with thinking events for the live counter, then returns the collected body for the caller to parse + validate. Visible deltas are not emitted — `/ai/patch` (the only caller) reads the full body post-stream as JSON.""" try: if provider == "cloud": cloud = config.get("cloud") or {} service = cloud.get("service") or "claude" api_key = (cloud.get("api_key") or "").strip() model = (cloud.get("model") or "").strip() if not api_key: _emit(request_id, "error", error="Cloud API key not configured") return "" if not model: _emit(request_id, "error", error="Cloud model not configured") return "" if service == "claude": return await _stream_claude( request_id, api_key, model, system, user_request, images, ) base_url = _cloud_base_url(cloud) if not base_url: _emit(request_id, "error", error="Cloud base URL missing") return "" return await _stream_openai_compat( request_id, base_url, model, system, user_request, images, api_key=api_key, ) if provider == "local": return await _stream_local( request_id, config, system, user_request, images, ) _emit(request_id, "error", error=f"unknown provider {provider}") return "" except asyncio.CancelledError: _emit(request_id, "cancelled") raise except Exception as e: logger.exception("generation failed") _emit(request_id, "error", error=str(e)) return "" def _emit(request_id: str, event: str, *, content: str = "", error: str = "", tokens: int | None = None): payload = { "request_id": request_id, "event": event, "content": content, "error": error, "t": time.time(), } if tokens is not None: payload["tokens"] = tokens send_ws("promptchain_ai_stream", payload) # ── Claude streaming ────────────────────────────────────────────── async def _stream_claude(request_id: str, api_key: str, model: str, system: str, user_request: str, images: list[dict]) -> str: """Streams from Claude with thinking events for the live counter, then returns the collected body for the caller to parse + validate. Visible deltas are suppressed — the patch endpoint reads the full body post- stream and emit-as-JSON would be noise.""" content = [] for img in images: data = img.get("data") media = img.get("media_type") or "image/jpeg" if not data: continue content.append({ "type": "image", "source": {"type": "base64", "media_type": media, "data": data}, }) content.append({"type": "text", "text": user_request}) payload = { "model": model, "max_tokens": _MAX_TOKENS, "system": system, "messages": [{"role": "user", "content": content}], "stream": True, } logger.info("stream_claude[%s] model=%s images=%d user_chars=%d", request_id, model, len(images), len(user_request)) chars_streamed = 0 deltas_received = 0 saw_stop = False parse_failures = 0 body_buffer: list[str] = [] timeout = aiohttp.ClientTimeout(connect=_CONNECT_TIMEOUT, sock_read=_READ_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, json=payload, ) as resp: logger.info("stream_claude[%s] HTTP %d", request_id, resp.status) if resp.status != 200: data = await resp.text() logger.warning("stream_claude[%s] error body: %s", request_id, data[:500]) _emit(request_id, "error", error=f"HTTP {resp.status}: {data[:200]}") return "" async for raw in resp.content: line = raw.decode("utf-8", errors="replace").strip() if not line or not line.startswith("data:"): continue data_str = line[5:].strip() if not data_str: continue try: evt = json.loads(data_str) except Exception as e: parse_failures += 1 logger.debug("stream_claude[%s] json parse fail: %s | line=%r", request_id, e, data_str[:200]) continue evt_type = evt.get("type") if evt_type == "content_block_delta": delta = evt.get("delta") or {} if delta.get("type") == "text_delta": text = delta.get("text", "") deltas_received += 1 chars_streamed += len(text) body_buffer.append(text) elif evt_type == "message_stop": saw_stop = True return "".join(body_buffer) elif evt_type == "error": err = (evt.get("error") or {}).get("message") or "stream error" logger.warning("stream_claude[%s] stream error event: %s", request_id, err) _emit(request_id, "error", error=err) return "" _log_stream_exit("stream_claude", request_id, deltas_received, chars_streamed, saw_stop, parse_failures) return "".join(body_buffer) def _log_stream_exit(fn: str, request_id: str, deltas: int, chars: int, saw_stop: bool, parse_failures: int): # Stream loop exited without an explicit stop event — log whether we # got any content at all so "empty response" cases are visible in the # server log rather than silently resolving as a zero-delta success. level = logging.WARNING if chars == 0 else logging.INFO logger.log( level, "%s[%s] stream loop exited without stop: deltas=%d chars=%d parse_failures=%d", fn, request_id, deltas, chars, parse_failures, ) # ── Local (OpenAI-compat + Ollama native) streaming ─────────────── async def _stream_local(request_id: str, config: dict, system: str, user_request: str, images: list[dict]) -> str: local = config.get("local") or {} base_url = (local.get("base_url") or "").strip().rstrip("/") model = (local.get("model") or "").strip() if not base_url: _emit(request_id, "error", error="Local base URL not configured") return "" if not model: _emit(request_id, "error", error="Local model not configured") return "" # If the user has images AND we're talking to Ollama, prefer the native # /api/chat path since it takes images at the message level (cleaner # than the OpenAI-compat data: URL shape). Otherwise use /v1/chat/completions # universally — works for llama.cpp, LM Studio, Ollama's /v1, KoboldCPP. ollama_root = _ollama_root(base_url) use_ollama_native = bool(images) and await _is_ollama(ollama_root) if use_ollama_native: return await _stream_ollama_native( request_id, ollama_root, model, system, user_request, images, ) # Probe Ollama once even on the no-images branch so we can pass # think=false through the OpenAI-compat path (Ollama tolerates it, # cloud providers reject unknown fields). is_ollama = await _is_ollama(ollama_root) return await _stream_openai_compat( request_id, base_url, model, system, user_request, images, is_ollama=is_ollama, ) async def _is_ollama(ollama_root: str) -> bool: try: timeout = aiohttp.ClientTimeout(total=3) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(f"{ollama_root}/api/tags") as resp: return resp.status == 200 except Exception: return False async def _is_ollama_model_loaded(ollama_root: str, model: str) -> bool: """True if `model` is currently loaded into Ollama's VRAM (`/api/ps`). Used to gate the 'Loading model' status indicator on cold first calls. Returns False on any probe failure — the request continues normally without the indicator rather than blocking.""" try: timeout = aiohttp.ClientTimeout(total=3) target = (model or "").strip().lower() if not target: return False async with aiohttp.ClientSession(timeout=timeout) as session: async with session.get(f"{ollama_root}/api/ps") as resp: if resp.status != 200: return False data = await resp.json() for m in (data.get("models") or []): for key in ("name", "model"): v = (m.get(key) or "").strip().lower() if v == target: return True return False except Exception: return False async def _warmup_ollama_model(ollama_root: str, model: str) -> None: """Force Ollama to load `model` into VRAM by hitting /api/generate with no prompt — Ollama loads the model and returns once it's ready. This makes the 'Loading model' status linger accurately during the actual VRAM populate, instead of being clobbered by the next stage's status the moment we emit it. Quietly no-ops on failure; the real chat call would have triggered the load anyway, we just lose the indicator.""" try: timeout = aiohttp.ClientTimeout(total=300) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post( f"{ollama_root}/api/generate", json={"model": model, "keep_alive": "5m"}, ) as resp: await resp.read() except Exception: logger.warning("ollama warmup failed", exc_info=True) _THINK_BLOCK_RE = re.compile(r"]*>([\s\S]*?)", re.IGNORECASE) _THINK_OPEN_RE = re.compile(r"]*>", re.IGNORECASE) async def _stream_ollama_native(request_id: str, ollama_root: str, model: str, system: str, user_request: str, images: list[dict]) -> str: messages = [{"role": "system", "content": system}] msg = {"role": "user", "content": user_request} if images: msg["images"] = [img.get("data") for img in images if img.get("data")] messages.append(msg) # Suppress reasoning blocks: qwen3-thinking and similar burn 5-20k tokens # of ... per turn. We're feeding the model curated bundle # data, not asking it to reason — pure waste. Ollama 0.5+ honors `think`. # num_ctx: Ollama defaults to 2048 unless the Modelfile sets higher. Our # system prompt + bio + bundles + thinking can exceed that easily, which # silently truncates input mid-prompt and confuses the model — likely # cause of the recurring deadlock pattern. 32K gives qwen3-vl plenty of # room (model itself supports 256K). # repeat_penalty=1.3: 8B Qwen occasionally degenerates into a # comma-list repetition loop when given thin context (empty bios, # short user message). Default 1.1 isn't strong enough to break the # KV-cache reinforcement once it starts. 1.3 stops the runaway # without hurting legitimate output. Seen in the wild: the # `mythra_(xenoblade_chronicles_2)` matcher-miss → 175s of # `mythra_serene, mythra_graceful, mythra_radiant, ...` cycle. payload = { "model": model, "messages": messages, "stream": True, "think": False, "options": {"num_ctx": 32768, "repeat_penalty": 1.3}, } logger.info("stream_ollama[%s] url=%s model=%s images=%d user_chars=%d", request_id, f"{ollama_root}/api/chat", model, len(images), len(user_request)) chars_streamed = 0 deltas_received = 0 saw_done = False parse_failures = 0 body_buffer: list[str] = [] thinking_buffer: list[str] = [] # Ollama streams approximately one detokenized output token per chunk, # so chunk count is a close proxy for token count. eval_count on the # final done event is the ground truth if we ever need to reconcile. reasoning_tokens = 0 saw_think_block = False first_event_logged = False timeout = aiohttp.ClientTimeout(connect=_CONNECT_TIMEOUT, sock_read=_READ_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(f"{ollama_root}/api/chat", json=payload) as resp: logger.info("stream_ollama[%s] HTTP %d", request_id, resp.status) if resp.status != 200: body = await resp.text() logger.warning("stream_ollama[%s] error body: %s", request_id, body[:500]) _emit(request_id, "error", error=f"HTTP {resp.status}: {body[:200]}") return "" async for raw in resp.content: line = raw.decode("utf-8", errors="replace").strip() if not line: continue try: evt = json.loads(line) except Exception as e: parse_failures += 1 logger.debug("stream_ollama[%s] json parse fail: %s | line=%r", request_id, e, line[:200]) continue if not first_event_logged: first_event_logged = True dbg.info("ai[%s] ollama first event keys=%s message_keys=%s raw=%s", request_id, list(evt.keys()), list((evt.get("message") or {}).keys()), _trunc(json.dumps(evt, ensure_ascii=True), 1500)) msg = evt.get("message") or {} # Newer Ollama splits reasoning into a dedicated `thinking` # field. Surface it via a separate WS event so the client # can show "Reasoning…" without polluting the prompt body. thinking_chunk = msg.get("thinking") or "" if thinking_chunk: thinking_buffer.append(thinking_chunk) reasoning_tokens += 1 _request_reasoning_chars[request_id] = ( _request_reasoning_chars.get(request_id, 0) + len(thinking_chunk) ) _emit(request_id, "thinking", content=thinking_chunk, tokens=reasoning_tokens) chunk = msg.get("content", "") if chunk: if not saw_think_block and _THINK_OPEN_RE.search(chunk): saw_think_block = True dbg.info("ai[%s] inline block detected in content " "(chunk=%r) — body_buffer will contain reasoning", request_id, chunk[:200]) deltas_received += 1 chars_streamed += len(chunk) body_buffer.append(chunk) if evt.get("done"): saw_done = True # Ollama puts timing/eval counts on the done event. dbg.info( "ai[%s] ollama done metadata: total=%sms load=%sms " "prompt_eval=%s tokens (%sms) eval=%s tokens (%sms) " "thinking_chars=%d body_chars=%d inline_think=%s", request_id, _ms(evt.get("total_duration")), _ms(evt.get("load_duration")), evt.get("prompt_eval_count"), _ms(evt.get("prompt_eval_duration")), evt.get("eval_count"), _ms(evt.get("eval_duration")), sum(len(t) for t in thinking_buffer), chars_streamed, saw_think_block, ) if thinking_buffer: _dump(request_id, "thinking_buffer", "".join(thinking_buffer)) return "".join(body_buffer) _log_stream_exit("stream_ollama", request_id, deltas_received, chars_streamed, saw_done, parse_failures) if thinking_buffer: _dump(request_id, "thinking_buffer", "".join(thinking_buffer)) return "".join(body_buffer) def _ms(ns) -> str: """Ollama reports durations in nanoseconds. Render as integer ms for logs.""" if not isinstance(ns, (int, float)): return "?" return f"{int(ns / 1_000_000)}" async def _stream_openai_compat(request_id: str, base_url: str, model: str, system: str, user_request: str, images: list[dict], api_key: str | None = None, is_ollama: bool = False) -> str: content = [] for img in images: data = img.get("data") media = img.get("media_type") or "image/jpeg" if not data: continue content.append({ "type": "image_url", "image_url": {"url": f"data:{media};base64,{data}"}, }) content.append({"type": "text", "text": user_request}) messages = [ {"role": "system", "content": system}, {"role": "user", "content": content if images else user_request}, ] payload = { "model": model, "messages": messages, "stream": True, "max_tokens": _MAX_TOKENS, } if is_ollama: # Ollama-served reasoning models burn 5-20k tokens on # blocks per turn. We're feeding curated bundle data, not asking # the model to derive anything — disable reasoning. Field is # Ollama-specific; OpenAI/Anthropic reject unknown payload keys. payload["think"] = False # Match the native /api/chat path — bump num_ctx so our system # prompt + bio + bundles + thinking aren't truncated to 2048. # qwen3-vl supports 256K natively; 32K is comfortable headroom. # repeat_penalty=1.3: breaks comma-list degenerate loops on # thin-context calls (see _stream_ollama_native for the # observed-in-wild cause). payload["options"] = {"num_ctx": 32768, "repeat_penalty": 1.3} headers = {"Authorization": f"Bearer {api_key}"} if api_key else {} logger.info("stream_openai_compat[%s] url=%s model=%s images=%d user_chars=%d", request_id, f"{base_url}/chat/completions", model, len(images), len(user_request)) chars_streamed = 0 deltas_received = 0 saw_done = False parse_failures = 0 body_buffer: list[str] = [] thinking_buffer: list[str] = [] # Most OpenAI-compat servers stream ~1 token per delta chunk; counting # reasoning_content deltas is a close (not exact) proxy for token count. reasoning_tokens = 0 saw_think_block = False first_event_logged = False timeout = aiohttp.ClientTimeout(connect=_CONNECT_TIMEOUT, sock_read=_READ_TIMEOUT) async with aiohttp.ClientSession(timeout=timeout) as session: async with session.post(f"{base_url}/chat/completions", json=payload, headers=headers) as resp: logger.info("stream_openai_compat[%s] HTTP %d", request_id, resp.status) if resp.status != 200: body = await resp.text() logger.warning("stream_openai_compat[%s] error body: %s", request_id, body[:500]) _emit(request_id, "error", error=f"HTTP {resp.status}: {body[:200]}") return "" async for raw in resp.content: line = raw.decode("utf-8", errors="replace").strip() if not line or not line.startswith("data:"): continue data_str = line[5:].strip() if data_str == "[DONE]": saw_done = True if thinking_buffer: _dump(request_id, "thinking_buffer", "".join(thinking_buffer)) return "".join(body_buffer) if not data_str: continue try: evt = json.loads(data_str) except Exception as e: parse_failures += 1 logger.debug("stream_openai_compat[%s] json parse fail: %s | line=%r", request_id, e, data_str[:200]) continue if not first_event_logged: first_event_logged = True delta_obj = (evt.get("choices") or [{}])[0].get("delta") or {} dbg.info("ai[%s] openai-compat first event delta_keys=%s raw=%s", request_id, list(delta_obj.keys()), _trunc(json.dumps(evt, ensure_ascii=True), 1500)) choices = evt.get("choices") or [] if not choices: continue delta_obj = choices[0].get("delta") or {} # DeepSeek-R1, Qwen-Reasoning, vLLM-with-think etc. expose # reasoning via `reasoning_content`. Capture and surface as # a separate event so the client can show "Reasoning…". reasoning = delta_obj.get("reasoning_content") or delta_obj.get("reasoning") or "" if reasoning: thinking_buffer.append(reasoning) reasoning_tokens += 1 _request_reasoning_chars[request_id] = ( _request_reasoning_chars.get(request_id, 0) + len(reasoning) ) _emit(request_id, "thinking", content=reasoning, tokens=reasoning_tokens) delta = delta_obj.get("content") or "" if delta: if not saw_think_block and _THINK_OPEN_RE.search(delta): saw_think_block = True dbg.info("ai[%s] inline block detected in content " "(chunk=%r) — body_buffer will contain reasoning", request_id, delta[:200]) deltas_received += 1 chars_streamed += len(delta) body_buffer.append(delta) finish = choices[0].get("finish_reason") if finish: logger.info("stream_openai_compat[%s] finish_reason=%s deltas=%d chars=%d", request_id, finish, deltas_received, chars_streamed) _log_stream_exit("stream_openai_compat", request_id, deltas_received, chars_streamed, saw_done, parse_failures) if thinking_buffer: _dump(request_id, "thinking_buffer", "".join(thinking_buffer)) return "".join(body_buffer) # ── patch planner ───────────────────────────────────────────────── # Single endpoint backing the AI Assistant panel. The model streams # internally for the live token counter but returns one round-trip # response shaped as sectioned tag output (or natlang prose); the # server validates and post-processes before reply. def _patch_system_prompt(bios: list[dict] | None = None, has_node_prompt: bool = False, prompt_style: str = "tags") -> str: bios = bios or [] has_bio = bool(bios) if prompt_style == "natural": return _patch_system_prompt_natlang_impl( has_bio=has_bio, has_node_prompt=has_node_prompt, ) has_default_outfit_slots = any( (b.get("default_outfit") or {}).get("slots") for b in bios ) has_override_outfit_slots = any( (b.get("user_requested_outfit") or {}).get("slots") for b in bios ) has_any_outfit_slots = has_default_outfit_slots or has_override_outfit_slots multi_char = sum(1 for b in bios if b and b.get("tag")) >= 2 return _patch_system_prompt_impl( has_bio=has_bio, has_default_outfit_slots=has_default_outfit_slots, has_override_outfit_slots=has_override_outfit_slots, has_any_outfit_slots=has_any_outfit_slots, has_node_prompt=has_node_prompt, multi_char=multi_char, ) def _patch_system_prompt_natlang_impl(has_bio: bool, has_node_prompt: bool = False) -> str: """Natlang-mode system prompt for checkpoints trained on natural- language captions (z-image, Qwen-Image, etc.). Mirrors the tag-mode sectioned output structure — same `// Section:` headers, same order — but section bodies are prose paragraphs instead of comma-separated tag tokens. Negatives stay tag-shaped (template authoring convention, downstream conditioning expectation).""" parts = [ "You are writing a Stable Diffusion prompt for the user. " "The target checkpoint is trained on natural-language captions, " "not Danbooru tags — section bodies are descriptive PROSE, not " "comma-separated tag lists.", "", ] if has_node_prompt: parts.extend([ "PATCH MODE — a non-empty node_prompt is provided in the user " "message. You are PATCHING it, not rebuilding it.", "- Reproduce every section that exists in node_prompt VERBATIM. " "Section headers, ordering, and prose bodies survive unchanged " "unless the user's request directly modifies them.", "- The ONLY edits you may make are direct consequences of the " "user's request (an explicit add, remove, or swap).", "- Sentences the user did not mention MUST appear unchanged in " "their original section. Sections the user did not mention MUST " "appear unchanged in your output.", "- Do NOT \"clean up\", \"streamline\", or \"focus\" the existing " "prompt. Minimum-change is the rule.", "- AUTO-DECOMPOSE: if node_prompt has NO `// Section:` headers " "(it's flat prose pasted from somewhere else), distribute its " "content across the canonical sections (Character / Outfit / " "Pose, Action & Prop / Expression / Setting / Scene) in your " "output. The content survives verbatim where possible — you're " "just splitting it into the right buckets, not rewording.", "", ]) parts.extend([ "Output ONLY the prompt body — no preamble, no explanations, no " "markdown fences, no commentary. Output sectioned plain text in this " "exact format:", "", "// Section Header", "", "", "// Next Section Header", "", "", "Each section gets:", "- A `// Section: ` header line — use the exact header supplied " "in the user message bio block, including any ` from Character: ` " "or `(signature)` suffix.", "- A blank line between sections; one prose body per section " "(multi-sentence, descriptive).", "- Body is PROSE — flowing sentences, not comma-separated tokens. " "Do NOT emit Danbooru-canonical tag forms (`cammy_white`, `1girl`, " "`presenting_foot`) in section bodies — write the natural-language " "equivalent.", "", "SECTION STRUCTURE for fresh prompts (only emit a section if the " "user or bio supplies content for it; do NOT add empty/speculative " "sections):", " // Character: ()", " ", " // Outfit: from Character: ", " ", " // Pose: (or `// Pose: (signature) from Character: ` " "when a bio matched a named pose — use that exact header verbatim " "from the user message bio block)", " ", " // Expression", " ", " // Setting / Scene", " ", "", "ORDER inside the prompt body (top to bottom): character → outfit " "→ pose → expression → setting/scene. These five are the only " "positive sections that exist. The server may inject an additional " "`// Style: ` section after setting; do not emit one yourself. " "Do NOT invent any other section header.", "", "BIO SOURCE PRIORITY:", " 1. If a character bio block is provided in the user message, use " "its `base_natlang`, `outfit_natlang`, and `pose_natlang` text as the " "source of truth. Reproduce the prose verbatim where possible; weave " "in user modifications minimally.", " 2. For everything else, write a short descriptive prose phrase. " "Do not invent details that contradict the bio.", "", "NEGATIVE PROMPT:", "If a Negative Prompt is needed, place it at the very end on its own " "line, starting with `Negative Prompt:` followed by a comma-separated " "list of canonical-tag tokens (NOT prose). Negatives describe what to " "suppress in latent space and are tag-shaped regardless of positive " "shape.", "", "RULES:", "- Never invent character or outfit details you aren't sure about. " "If unsure, omit.", "- Make the minimum change the user requested. Only emit a section " "if the user or bio supplies content for it; never speculatively add " "a section the user didn't mention.", "- Do not include placeholder tags from examples in this prompt as " "content.", "- 'wearing only X' / 'just X' / 'only X' is a STRIP instruction, NOT " "a confirmation. The user wants to REMOVE every other outfit item " "and keep ONLY X. Rewrite the // Outfit prose to describe ONLY the " "named item(s); the rest is implicitly nude.", " * The new Outfit prose must NOT reference any removed items, " "even as location markers or anchors. Wrong: 'red socks visible " "above the open tops of the boots' (boots are removed). Right: " "'Wearing only red socks; otherwise nude.'", " * The new Outfit prose must NOT enumerate what was removed " "(\"the leotard, beret, etc. are all removed\"). Just describe the " "current state.", " * CASCADE: if Outfit was stripped, audit the // Pose, Action & " "Prop and // Expression and // Setting / Scene sections. Any " "sentence that references a now-removed outfit item (boots, " "leotard, beret, gloves, harness, etc.) MUST be rewritten to drop " "that reference. Example: pose said 'feet visible through the open " "tops of her combat boots'; outfit now has no boots; pose should " "be rewritten to 'feet visible' (or whatever the action is) " "without the boot anchor. The pose stays — only the removed-item " "anchors get stripped.", "", ]) if has_bio: parts.append( "- A character bio is attached below. Treat its natlang fields " "(`base_natlang`, outfit `natlang`, pose `natlang`) as " "authoritative — use the prose verbatim where the user's request " "doesn't require modification." ) parts.extend(["", "/no_think"]) return "\n".join(parts) def _patch_system_prompt_impl(has_bio: bool, has_default_outfit_slots: bool, has_override_outfit_slots: bool, has_any_outfit_slots: bool, has_node_prompt: bool = False, multi_char: bool = False) -> str: """Sectioned plain-text output. No JSON, no tools. The patch UX (remove/add/keep chips) is derived client-side by diffing the new output against existing editor content.""" parts = [ "You are writing a Stable Diffusion prompt for the user.", "", ] if has_node_prompt: # The model's biggest failure mode in patch flow is dropping # untouched sections — it reads "only emit a section if the user # supplies content" (a build-mode rule) and inverts it into "drop # any section the user didn't mention this turn." Lead with an # explicit patch-mode rule so untouched sections survive. parts.extend([ "PATCH MODE — a non-empty node_prompt is provided in the user " "message. You are PATCHING it, not rebuilding it.", "- Reproduce every section that exists in node_prompt VERBATIM. " "Sections, headers, ordering, and tags survive unchanged unless the " "user's request directly modifies them.", "- The ONLY edits you may make are direct consequences of the " "user's request (an explicit add, remove, or swap).", "- Tags the user did not mention MUST appear unchanged in their " "original section. Sections the user did not mention MUST appear " "unchanged in your output.", "- Do NOT \"clean up\", \"streamline\", or \"focus\" the existing " "prompt. Minimum-change is the rule.", "- This rule overrides the section-template guidance below — " "the templates are for FRESH prompts only.", "- AUTO-DECOMPOSE: if node_prompt has NO `// Section:` headers " "(it's a flat tag list pasted from somewhere else), distribute " "its tokens across the canonical sections (Character / Outfit / " "Pose, Action & Prop / Expression / Setting / Scene) in your " "output. Tokens survive verbatim — you're just routing each " "token to the right section based on what it describes.", "", ]) else: # BUILD MODE symmetric to the PATCH MODE block above. qwen3-vl:8b # routinely speculates // Pose, // Setting / Scene, and // Quality # on bare requests like "cammy white" or "set up cammy_white", # even though the inline "only if supplied" rule says not to. # Lead with an explicit no-speculation rule so the inline rule # isn't the only line of defense. parts.extend([ "BUILD MODE — node_prompt is empty. You are composing a FRESH " "prompt from scratch using the user's request and the character " "bio (if provided).", "- Emit ONLY sections the user explicitly asked for, plus // " "Character and // Outfit when a character bio is attached.", "- DO NOT speculate. If the user said \"cammy white\", emit // " "Character and // Outfit (from bio) — nothing else. Do not invent " "// Pose, // Expression, // Setting / Scene, or // Quality " "content the user did not request.", "- // Style is server-managed — do not emit a // Style section " "unless the user explicitly named a style. The server injects " "the model's default style template automatically.", "- // Quality is user-owned and OPTIONAL. Emit it only when the " "user explicitly named quality tokens (e.g. `add masterpiece, " "absurdres`). Generic SD prompting tokens like `highres`, " "`detailed`, `sharp focus`, `cinematic lighting` are NOT defaults " "you should add on your own — they belong only when the user " "asked.", "", ]) parts.extend([ "Output ONLY the prompt body — no preamble, no explanations, no markdown " "fences, no commentary. Output sectioned plain text in this exact format:", "", "// Section Header", "tag1, tag2, tag3, ...", "", "// Next Section Header", "tag1, tag2, ...", "", "Each section gets:", "- A `// Section: ` header line", "- A blank line is fine between sections; one comma-separated tag line per section.", "- Use Danbooru-canonical underscored tag forms (lowercase, words joined by `_`). " "The output post-processor converts to spaces if the target model wants spaces.", "- Weighted form `(tag:1.1)` is a single literal token — preserve verbatim.", "", "SECTION STRUCTURE for fresh prompts (only emit a section if the user or bio supplies " "content for it; do NOT add empty/speculative sections):", " // Character: ", " ", " // Outfit: ", " ", " // Pose, Action & Prop (comma-separated canonical danbooru tags — covers body " "position, limb arrangement, gestures, where the head/eyes are directed, presented " "body parts, and props the character interacts with. Break the user's natural-language " "pose into its component tags; do NOT fuse them into one phrase.)", " ", " // Expression (facial affect ONLY — what the face shows emotionally. Gaze " "direction does NOT belong here; it goes in Pose, Action & Prop.)", " ", " // Setting / Scene", " ", " // Style: