"""
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: (optional — user-owned. Carries style "
"template body like `masterpiece, best quality, very awa` or "
"`Photorealistic, sharp focus`. Server may inject or override this.)",
"