Files
comfyui-nvidia/deployments/ai-stack/openwebui-tools/smart_image_gen.py
T
57_WolveandClaude Opus 4.7 d4e2058859 smart_image_gen v0.3: add edit_image (img2img) method
The Tool now exposes two methods the LLM picks between based on whether
the user attached an image:

  generate_image — txt2img (existing, unchanged behavior)
  edit_image     — img2img on the most recently attached image

edit_image extracts the source image from __messages__ (base64 data
URIs in image_url content blocks) or __files__ (local path or URL),
uploads to ComfyUI's /upload/image, runs an img2img workflow at the
caller-specified denoise (default 0.7), and returns the edited result.
Same per-style routing / sampler / CFG / prefix logic as generation.

Refactored the submit-and-poll loop into _submit_and_fetch shared by
both methods. Image extraction is defensive — tries messages first,
then files (path then URL), returns a clear "no image attached"
message rather than silently generating from scratch.

Image Studio system prompt rewritten to teach the LLM when to call
edit_image vs generate_image and how to pick denoise.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-04-19 12:59:13 -05:00

587 lines
25 KiB
Python

"""
title: Smart Image Generator & Editor (ComfyUI)
author: ai-stack
version: 0.3.0
description: Generate or edit images via ComfyUI with automatic SDXL
checkpoint routing. Two methods — generate_image (txt2img) and
edit_image (img2img on the user's most recently attached image). The
LLM picks (or auto-detects) the right model — photoreal, Pony
score-tag, NoobAI/Illustrious furry, etc. — and each style ships
with the creator-recommended sampler, scheduler, CFG, steps, CLIP
skip, prompt-prefix dialect, and negatives.
required_open_webui_version: 0.5.0
"""
import asyncio
import base64
import re
import time
import uuid
from typing import Awaitable, Callable, Optional
import aiohttp
from pydantic import BaseModel, Field
from typing import Literal
StyleName = Literal[
"photo", "juggernaut", "pony", "general",
"furry-nai", "furry-noob", "furry-il",
]
# ─────────────────────────────────────────────────────────────────────────────
# Per-style settings — sampler/scheduler/cfg/steps/clip_skip/prefix/negatives
# come from each model's creator page on Civitai. Three prefix dialects in
# play: photoreal (no prefix, natural language), Pony score chain (REQUIRED
# for any Pony-derived checkpoint), and Booru quality tags (NoobAI /
# Illustrious lineage). Never cross-contaminate.
# ─────────────────────────────────────────────────────────────────────────────
STYLES = {
"photo": {
"ckpt": "CyberRealisticXLPlay_V8.0_FP16.safetensors",
"sampler": "dpmpp_2m_sde",
"scheduler": "karras",
"cfg": 4.0,
"steps": 28,
"clip_skip": 1,
"prefix": "", # natural language only — no quality tags
"negative": (
"cartoon, drawing, illustration, anime, manga, painting, sketch, "
"render, 3d, cgi, watercolor, plastic skin, doll-like, oversaturated, "
"lowres, blurry, jpeg artifacts, noisy, grainy, low quality, worst quality, "
"bad anatomy, deformed, mutated, extra limbs, extra fingers, missing fingers, "
"fused fingers, malformed hands, asymmetric face, "
"watermark, signature, text, logo, label, username"
),
},
"juggernaut": {
"ckpt": "Juggernaut-XL_v9_RunDiffusionPhoto_v2.safetensors",
"sampler": "dpmpp_2m_sde",
"scheduler": "karras",
"cfg": 4.5,
"steps": 35,
"clip_skip": 1,
"prefix": "", # natural language only
"negative": (
"cartoon, drawing, illustration, anime, manga, painting, sketch, "
"render, 3d, cgi, plastic skin, washed out, oversaturated, "
"lowres, blurry, jpeg artifacts, low quality, worst quality, "
"bad anatomy, deformed, mutated, extra limbs, extra fingers, missing fingers, "
"fused fingers, malformed hands, "
"watermark, signature, text, logo, username"
),
},
"pony": {
"ckpt": "ponyDiffusionV6XL_v6StartWithThisOne.safetensors",
"sampler": "euler_ancestral",
"scheduler": "normal",
"cfg": 7.5,
"steps": 25,
"clip_skip": 2,
# REQUIRED — the full chain. Just `score_9` alone is much weaker.
"prefix": "score_9, score_8_up, score_7_up, score_6_up, score_5_up, score_4_up, ",
# Pony's creator notes negatives are usually unnecessary; conservative
# baseline only. Source-toggle tags (source_pony/furry/anime/cartoon)
# are intentionally omitted — they exclude entire content domains.
"negative": (
"score_6, score_5, score_4, "
"worst quality, low quality, lowres, blurry, jpeg artifacts, noisy, "
"bad anatomy, bad proportions, bad hands, extra digit, fewer digits, "
"fused fingers, malformed limbs, deformed, ugly, "
"censored, monochrome, "
"watermark, signature, text, logo, artist name, patreon username, twitter username"
),
},
"general": {
"ckpt": "talmendoxlSDXL_v11Beta.safetensors",
"sampler": "dpmpp_2m",
"scheduler": "karras",
"cfg": 8.0, # Talmendo wants notably higher CFG than the others
"steps": 30,
"clip_skip": 2,
"prefix": "", # creator says don't push "masterpiece" — fights the amateur aesthetic
"negative": (
"lowres, blurry, jpeg artifacts, noisy, grainy, low quality, worst quality, "
"bad anatomy, deformed, mutated, extra limbs, missing fingers, fused fingers, "
"malformed hands, ugly, "
"watermark, signature, text, logo"
),
},
"furry-nai": {
"ckpt": "reedFURRYMixSDXL_v23nai.safetensors",
"sampler": "euler_ancestral",
"scheduler": "normal",
"cfg": 5.0,
"steps": 30,
"clip_skip": 2,
"prefix": (
"masterpiece, best quality, high quality, good quality, "
"detailed eyes, highres, absurdres, furry, "
),
"negative": (
"human, realistic, photorealistic, 3d, cgi, "
"worst quality, bad_quality, normal quality, lowres, "
"anatomical nonsense, bad anatomy, interlocked fingers, extra fingers, "
"bad_feet, bad_hands, deformed anatomy, bad proportions, "
"censored, simple background, transparent, face backlighting, "
"watermark, signature, text, logo, username, jpeg artifacts"
),
},
"furry-noob": {
"ckpt": "indigoVoidFurryFusedXL_noobaiV32.safetensors",
"sampler": "euler_ancestral", # creator: other samplers won't work
"scheduler": "normal",
"cfg": 4.5,
"steps": 20,
"clip_skip": 2,
"prefix": (
"masterpiece, best quality, perfect quality, absurdres, newest, "
"very aesthetic, vibrant colors, "
),
"negative": (
"human, realistic, photorealistic, 3d, cgi, "
"shiny skin, shiny clothing, "
"worst quality, low quality, lowres, blurry, jpeg artifacts, noisy, "
"bad anatomy, bad hands, mutated hands, bad proportions, "
"extra digit, fewer digits, fused fingers, malformed limbs, deformed, ugly, "
"watermark, signature, text, logo, username, artist signature"
),
},
"furry-il": {
"ckpt": "novaFurryXL_ilV170.safetensors",
"sampler": "euler_ancestral",
"scheduler": "normal",
"cfg": 4.0,
"steps": 30,
"clip_skip": 2,
# Illustrious wants `newest` in positive and `old`/`oldest` in negative
# — these are year-bucket tags from the training set. `furry` and
# `anthro` are universally helpful here.
"prefix": (
"masterpiece, best quality, amazing quality, very aesthetic, "
"high resolution, ultra-detailed, absurdres, newest, furry, anthro, "
),
"negative": (
"human, multiple tails, modern, recent, old, oldest, "
"graphic, cartoon, painting, crayon, graphite, abstract, glitch, "
"deformed, mutated, ugly, disfigured, long body, conjoined, "
"lowres, bad anatomy, bad hands, missing fingers, extra digits, fewer digits, "
"cropped, very displeasing, worst quality, bad quality, sketch, "
"jpeg artifacts, signature, watermark, username, text, simple background, "
"bad ai-generated"
),
},
}
DEFAULT_STYLE = "general"
# First-match-wins keyword router used when the caller didn't pick a style.
# Order matters — narrower patterns above broader ones.
ROUTING_RULES = [
# Pony score chain is the single strongest signal — Pony only
(re.compile(r"\bscore_\d", re.I), "pony"),
(re.compile(r"\bpony\b", re.I), "pony"),
# NoobAI / Illustrious explicit mentions
(re.compile(r"\b(noobai|noob)\b", re.I), "furry-noob"),
(re.compile(r"\b(illustrious|ilxl)\b", re.I), "furry-il"),
# Generic furry — defaults to NovaFurry (Illustrious lineage, current sweet spot)
(re.compile(r"\b(furry|anthro|feral|kemono|fursona|species)\b", re.I), "furry-il"),
# Photo / photoreal
(re.compile(r"\b(juggernaut)\b", re.I), "juggernaut"),
(re.compile(r"\b(photo|photograph|realistic|portrait|selfie|cinematic)\b", re.I), "photo"),
# Generic anime / illustration → Pony covers anime well
(re.compile(r"\b(anime|manga|2d|illustration)\b", re.I), "pony"),
]
def _route_style(prompt: str) -> str:
for pattern, style in ROUTING_RULES:
if pattern.search(prompt):
return style
return DEFAULT_STYLE
def _seed_value(seed: int) -> int:
return seed if seed > 0 else int(time.time() * 1000) % (2**31)
def _build_txt2img(positive: str, negative: str, settings: dict,
width: int, height: int, seed: int) -> dict:
"""
SDXL txt2img workflow. CLIP skip via CLIPSetLastLayer so the same graph
handles skip 1 (-1) and skip 2 (-2).
"""
return {
"3": {"class_type": "KSampler", "inputs": {
"seed": _seed_value(seed),
"steps": settings["steps"], "cfg": settings["cfg"],
"sampler_name": settings["sampler"], "scheduler": settings["scheduler"],
"denoise": 1.0,
"model": ["4", 0], "positive": ["6", 0],
"negative": ["7", 0], "latent_image": ["5", 0],
}},
"4": {"class_type": "CheckpointLoaderSimple",
"inputs": {"ckpt_name": settings["ckpt"]}},
"5": {"class_type": "EmptyLatentImage",
"inputs": {"width": width, "height": height, "batch_size": 1}},
"6": {"class_type": "CLIPTextEncode", "inputs": {"text": positive, "clip": ["10", 0]}},
"7": {"class_type": "CLIPTextEncode", "inputs": {"text": negative, "clip": ["10", 0]}},
"8": {"class_type": "VAEDecode", "inputs": {"samples": ["3", 0], "vae": ["4", 2]}},
"9": {"class_type": "SaveImage",
"inputs": {"filename_prefix": "smartgen", "images": ["8", 0]}},
"10": {"class_type": "CLIPSetLastLayer",
"inputs": {"stop_at_clip_layer": -settings["clip_skip"],
"clip": ["4", 1]}},
}
def _build_img2img(positive: str, negative: str, settings: dict,
image_filename: str, denoise: float, seed: int) -> dict:
"""
SDXL img2img workflow. Loads `image_filename` (already uploaded to
ComfyUI's /input/), VAE-encodes it to latent, and feeds that into the
sampler at the requested denoise. Resolution is whatever the source
image is — no resize.
"""
return {
"3": {"class_type": "KSampler", "inputs": {
"seed": _seed_value(seed),
"steps": settings["steps"], "cfg": settings["cfg"],
"sampler_name": settings["sampler"], "scheduler": settings["scheduler"],
"denoise": denoise,
"model": ["4", 0], "positive": ["6", 0],
"negative": ["7", 0], "latent_image": ["11", 0],
}},
"4": {"class_type": "CheckpointLoaderSimple",
"inputs": {"ckpt_name": settings["ckpt"]}},
"6": {"class_type": "CLIPTextEncode", "inputs": {"text": positive, "clip": ["10", 0]}},
"7": {"class_type": "CLIPTextEncode", "inputs": {"text": negative, "clip": ["10", 0]}},
"8": {"class_type": "VAEDecode", "inputs": {"samples": ["3", 0], "vae": ["4", 2]}},
"9": {"class_type": "SaveImage",
"inputs": {"filename_prefix": "smartedit", "images": ["8", 0]}},
"10": {"class_type": "CLIPSetLastLayer",
"inputs": {"stop_at_clip_layer": -settings["clip_skip"],
"clip": ["4", 1]}},
"11": {"class_type": "VAEEncode", "inputs": {"pixels": ["12", 0], "vae": ["4", 2]}},
"12": {"class_type": "LoadImage", "inputs": {"image": image_filename}},
}
async def _extract_attached_image(
files: Optional[list],
messages: Optional[list],
session: aiohttp.ClientSession,
) -> Optional[bytes]:
"""
Find the most recent image the user attached to the chat. Tries three
sources in order: (1) base64 data URIs in `image_url` content blocks
of the recent messages (works for vision-capable models), (2) a local
filesystem path on the file dict (open-webui stores uploads under
/app/backend/data/uploads/), (3) the file's url field, fetched over
HTTP. Returns raw image bytes, or None if nothing matched.
"""
# Messages: standard OpenAI image_url content blocks.
for msg in reversed(messages or []):
content = msg.get("content") if isinstance(msg, dict) else None
if isinstance(content, list):
for block in content:
if not isinstance(block, dict) or block.get("type") != "image_url":
continue
url = (block.get("image_url") or {}).get("url", "")
if url.startswith("data:image"):
try:
return base64.b64decode(url.split(",", 1)[1])
except Exception:
pass
# Files: try local path, then URL.
for f in files or []:
if not isinstance(f, dict):
continue
ftype = (f.get("type") or "").lower()
fname = (f.get("name") or f.get("filename") or "").lower()
is_image = "image" in ftype or fname.endswith((".png", ".jpg", ".jpeg", ".webp"))
if not is_image:
continue
for path_key in ("path", "filepath", "file_path"):
path = f.get(path_key)
if path:
try:
with open(path, "rb") as fh:
return fh.read()
except OSError:
pass
url = f.get("url")
if url:
full = url if url.startswith("http") else f"http://localhost:8080{url}"
try:
async with session.get(full) as resp:
if resp.status == 200:
return await resp.read()
except aiohttp.ClientError:
pass
return None
async def _upload_to_comfyui(
session: aiohttp.ClientSession, base: str, raw: bytes
) -> Optional[str]:
"""POST raw bytes to ComfyUI /upload/image and return the saved name."""
name = f"smartedit_{uuid.uuid4().hex[:12]}.png"
form = aiohttp.FormData()
form.add_field("image", raw, filename=name, content_type="image/png")
form.add_field("overwrite", "true")
async with session.post(f"{base}/upload/image", data=form) as resp:
if resp.status != 200:
return None
return (await resp.json()).get("name", name)
async def _submit_and_fetch(
session: aiohttp.ClientSession,
base: str,
workflow: dict,
timeout_seconds: int,
emit: Callable[[str, bool], Awaitable[None]],
settings: dict,
) -> tuple[Optional[bytes], Optional[str]]:
"""Submit a workflow, poll history, fetch the first output image. Returns
(image_bytes, error_message)."""
client_id = str(uuid.uuid4())
async with session.post(
f"{base}/prompt", json={"prompt": workflow, "client_id": client_id}
) as resp:
if resp.status != 200:
return None, f"ComfyUI rejected the prompt: {resp.status} {await resp.text()}"
prompt_id = (await resp.json()).get("prompt_id")
if not prompt_id:
return None, "ComfyUI didn't return a prompt_id."
await emit(
f"Sampling — {settings['sampler']}/{settings['scheduler']}, "
f"CFG {settings['cfg']}, {settings['steps']} steps", False
)
deadline = time.time() + timeout_seconds
output_images: list = []
while time.time() < deadline:
await asyncio.sleep(1.5)
async with session.get(f"{base}/history/{prompt_id}") as resp:
if resp.status != 200:
continue
history = await resp.json()
if prompt_id in history:
for node_out in history[prompt_id].get("outputs", {}).values():
output_images.extend(node_out.get("images", []))
if output_images:
break
if not output_images:
return None, f"Timed out after {timeout_seconds}s waiting for image."
img = output_images[0]
params = {
"filename": img["filename"],
"subfolder": img.get("subfolder", ""),
"type": img.get("type", "output"),
}
async with session.get(f"{base}/view", params=params) as resp:
if resp.status != 200:
return None, f"Failed to fetch image: {resp.status}"
return await resp.read(), None
class Tools:
class Valves(BaseModel):
COMFYUI_BASE_URL: str = Field(
default="http://comfyui:8188",
description="ComfyUI server URL reachable from the open-webui container.",
)
TIMEOUT_SECONDS: int = Field(
default=240,
description="Maximum wait for a single generation to complete.",
)
def __init__(self):
self.valves = self.Valves()
async def generate_image(
self,
prompt: str,
style: Optional[StyleName] = None,
negative_prompt: Optional[str] = None,
width: int = 1024,
height: int = 1024,
seed: int = 0,
__event_emitter__: Optional[Callable[[dict], Awaitable[None]]] = None,
) -> str:
"""
Create a NEW image from scratch and show it to the user. Use this
whenever the user asks you to draw, generate, create, make, paint,
render, or imagine any visual content — photographs, portraits,
characters, scenes, illustrations, anime, drawings — and they have
NOT attached an existing image. If they did attach an image and
want it modified, use edit_image instead.
Pick `style` to match what the user wants:
- "photo" — photorealistic photographs, portraits, cinematic shots.
- "juggernaut" — alternate photoreal style (sharper, more saturated).
- "pony" — anime / illustration / cartoon (Pony Diffusion).
- "general" — fallback for anything that doesn't fit the others.
- "furry-nai" — anthropomorphic characters (NAI-trained mix).
- "furry-noob" — anthropomorphic characters (NoobAI base).
- "furry-il" — anthropomorphic characters (Illustrious base, default
for any "furry" / "anthro" request unless specified otherwise).
Each style auto-prepends the right quality tags and picks the right
sampler / CFG / steps / CLIP skip. Do NOT add tags like
"masterpiece" or "score_9" to `prompt` yourself; the tool handles
that.
:param prompt: Plain description of the image (subject, scene,
style notes, lighting, etc.). No quality tags.
:param style: One of the values above. Omit to auto-detect.
:param negative_prompt: Extra terms to exclude. Usually unneeded.
:param width: Pixels (default 1024 — SDXL native). For portraits
use 832 with height 1216; for landscapes 1216 with height 832.
:param height: Pixels (default 1024).
:param seed: 0 to randomize, otherwise a specific seed for repeats.
:return: Markdown image of the result.
"""
chosen = style or _route_style(prompt)
settings = STYLES.get(chosen)
if not settings:
return f"Unknown style '{chosen}'. Available: {', '.join(STYLES.keys())}"
async def emit(msg: str, done: bool = False):
if __event_emitter__:
await __event_emitter__({
"type": "status",
"data": {"description": msg, "done": done},
})
await emit(f"Routing to {chosen} ({settings['ckpt']})")
positive = f"{settings['prefix']}{prompt}"
negative = settings["negative"]
if negative_prompt:
negative = f"{negative}, {negative_prompt}"
workflow = _build_txt2img(positive, negative, settings, width, height, seed)
base = self.valves.COMFYUI_BASE_URL.rstrip("/")
async with aiohttp.ClientSession() as session:
raw, err = await _submit_and_fetch(
session, base, workflow, self.valves.TIMEOUT_SECONDS, emit, settings,
)
if err:
return err
b64 = base64.b64encode(raw).decode("ascii")
await emit(f"Done — {chosen}", done=True)
return f"![{chosen}](data:image/png;base64,{b64})"
async def edit_image(
self,
edit_instruction: str,
style: Optional[StyleName] = None,
denoise: float = 0.7,
negative_prompt: Optional[str] = None,
seed: int = 0,
__files__: Optional[list] = None,
__messages__: Optional[list] = None,
__event_emitter__: Optional[Callable[[dict], Awaitable[None]]] = None,
) -> str:
"""
Edit, modify, transform, or restyle an image the user has ATTACHED
to the chat. Use this whenever the user uploads or attaches an
image and asks to change it — "make this a sunset", "add a hat",
"turn this into anime", "remove the background", etc. If no image
is attached, use generate_image instead.
The tool finds the most recent attached image automatically, picks
the right SDXL checkpoint via `style` (same options as
generate_image), and applies the edit at the requested `denoise`.
Denoise tuning:
- 0.3 to 0.5 → subtle (style transfer, color tweaks, preserve composition)
- 0.6 to 0.8 → moderate (add/remove objects, lighting changes — default)
- 0.85 to 1.0 → heavy (basically reimagining; loses original details)
Pick `style` based on the desired output, NOT the input image.
Same options as generate_image: photo, juggernaut, pony, general,
furry-nai, furry-noob, furry-il.
:param edit_instruction: What to change. The tool auto-prepends the
style's quality tags so don't include those.
:param style: One of the StyleName values. Omit to auto-detect from
edit_instruction.
:param denoise: 0.0 = no change, 1.0 = ignore source. Default 0.7.
:param negative_prompt: Extra terms to exclude. Usually unneeded.
:param seed: 0 to randomize, otherwise a specific seed.
:return: Markdown image of the result, or an error if no image is attached.
"""
chosen = style or _route_style(edit_instruction)
settings = STYLES.get(chosen)
if not settings:
return f"Unknown style '{chosen}'. Available: {', '.join(STYLES.keys())}"
denoise = max(0.0, min(1.0, denoise))
async def emit(msg: str, done: bool = False):
if __event_emitter__:
await __event_emitter__({
"type": "status",
"data": {"description": msg, "done": done},
})
base = self.valves.COMFYUI_BASE_URL.rstrip("/")
async with aiohttp.ClientSession() as session:
await emit("Looking for attached image…")
raw_in = await _extract_attached_image(__files__, __messages__, session)
if raw_in is None:
return (
"No image found in the chat. Ask the user to attach the "
"image they want edited (paperclip / drag-drop), or call "
"generate_image instead if they want a new image."
)
await emit("Uploading source to ComfyUI…")
uploaded_name = await _upload_to_comfyui(session, base, raw_in)
if not uploaded_name:
return "Failed to upload source image to ComfyUI."
await emit(
f"Routing to {chosen} ({settings['ckpt']}), denoise {denoise:.2f}"
)
positive = f"{settings['prefix']}{edit_instruction}"
negative = settings["negative"]
if negative_prompt:
negative = f"{negative}, {negative_prompt}"
workflow = _build_img2img(
positive=positive,
negative=negative,
settings=settings,
image_filename=uploaded_name,
denoise=denoise,
seed=seed,
)
raw_out, err = await _submit_and_fetch(
session, base, workflow, self.valves.TIMEOUT_SECONDS, emit, settings,
)
if err:
return err
b64 = base64.b64encode(raw_out).decode("ascii")
await emit(f"Done — {chosen} (denoise {denoise:.2f})", done=True)
return f"![edit:{chosen}](data:image/png;base64,{b64})"