take the size the caller asked for, and the words they trained (v0.4.90)

Three things the image endpoints accepted and then quietly did something else
with.

size. A size that did not parse fell through to the default, so "1024" or
"1k x 1k" generated a square nobody asked for and the response looked like a
success. A side that was not a multiple of 8 went to the pipeline as-is, where
the VAE cannot represent it. Nothing capped it either, so 4096x4096 was
accepted and came back as a CUDA OOM -- which reads like a busy GPU rather
than a size the caller chose. Now: a 400 quoting what was sent, a 400 for a
zero dimension (snapping it up would answer an uninitialised client variable
with an 8x8 image), snapping to the grid with a log line, and a 400 naming
image.max_side / image.max_pixels when it is over the ceiling. "auto" is
accepted as in the OpenAI API, and the dead `width != width` NaN guard on a
value that was already an int is gone. The archive records the size actually
generated alongside the one requested.

/v1/images/inpaint no longer defaults to 1024x1024, which squashed every
non-square photo and pulled the mask out of alignment with it -- worse than
the same default on /v1/images/edits, where it was fixed in 0.4.86. Omitted
now keeps the source's shape.

Textual inversions. easynegative, bad-hands-5 and the rest are vectors living
in CLIP's embedding table under a word of their own, and until the file is
loaded that word is an ordinary unknown token -- so a large part of the
ecosystem's negative prompting asked for nothing at all. They load from the
request (embeddings: [...]), from the model's config (textual_inversion), or
by filename from image.embeddings_dir, which is how A1111 works and what a
prompt copied off a model card assumes. Two details decide whether this works
at all: an SDXL-format embedding holds a clip_l and a clip_g tensor that must
go into the matching tokenizer/encoder pair, and a multi-vector embedding
occupies several tokens (tok, tok_1, tok_2...) that the prompt has to be
expanded into -- without it an eight-vector embedding contributes one vector
and appears inert. A name the caller gave that will not load is a 400; a bad
file in the directory is a line in the log, because one file should not break
every request.

/v1/images/outfit goes through the same encoder now, so the weight grammar and
clip_skip reach it too. Its per-frame video path encodes the prompt once for
the clip instead of once per frame.

And a 400 is no longer reported as a failed backend. On the first request for
a model that is not loaded yet, the generate call sits inside the try that
falls back from diffusers to sd.cpp, so a rejected request was caught as a
load failure and answered with a report about which backends failed. The three
edit endpoints did the same through their catch-alls. All four re-raise
HTTPException untouched.

Found by the tests while writing them: the embeddings directory was searched
by the renamed trigger word rather than the filename, so "easynegative:myneg"
never resolved, and "768X512" did not parse.

64 new tests; full suite 4475 passed with the same 15 pre-existing failures as
0.4.89 (diff empty).
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01S8LAP3ueNzQJtwzhgVbq66
parent 777406dc
......@@ -838,9 +838,11 @@ persisted per `config_id`, so the configs don't overwrite each other.
| `POST /v1/images/from3d` | 3D model → rendered 2D image |
| `GET /v1/images/progress` | Current generation progress (step/total) |
Generations, edits and inpaint take `negative_prompt`, `clip_skip` and the
`(tag:1.3)` / `[tag]` / `BREAK` prompt grammar, and encode prompts of any
length (CLIP's 77-token limit does not apply) — see
Generations, edits and inpaint take `negative_prompt`, `clip_skip`,
`embeddings` (textual inversions, so `easynegative` and friends work in a
prompt) and the `(tag:1.3)` / `[tag]` / `BREAK` prompt grammar, and encode
prompts of any length — CLIP's 77-token limit does not apply. `size` is
validated rather than silently replaced. See
[docs/image-prompts.md](docs/image-prompts.md).
### Video
......
......@@ -16,7 +16,7 @@
# Canonical product version for CoderAI — single source of truth. Both the API
# metadata and the admin web UI read from here.
__version__ = "0.4.89"
__version__ = "0.4.90"
__tagline__ = "Complete orchestration, distribution and escalation of remotizable advanced inference."
# Configure the CUDA caching allocator BEFORE torch is imported anywhere.
......
......@@ -2931,7 +2931,7 @@ async def api_model_configure(request: Request, username: str = Depends(require_
# the A1111 weight grammar is parsed. Per model because the
# booru-tag fine-tunes need clip_skip 2 whatever the caller
# sends, and an OpenAI-API client has no field to send it in.
"clip_skip", "prompt_weighting",
"clip_skip", "prompt_weighting", "textual_inversion",
"component_quantization", "output_crf", "force_vram_update",
"balanced_gpu_percent", "acceleration",
"cache_type_k", "cache_type_v", "kv_offload", "n_batch", "n_ubatch", "n_seq_max",
......
......@@ -1553,6 +1553,12 @@ window.__DEFAULT_WHISPER_SERVER_PATH__ = {{ default_whisper_server_path|tojson }
</div>
<span class="form-hint">On, the A1111 grammar is parsed: <code>(tag:1.3)</code> emphasises, <code>[tag]</code> attenuates, bare <code>(tag)</code> is 1.1, nesting multiplies, <code>\(</code> is a literal bracket, and <code>BREAK</code> starts a new 75-token window. Off, brackets reach the model as text. Either way the prompt is encoded in as many windows as it needs, so nothing is cut at 77 tokens.</span>
<div class="form-row" style="margin-top:.75rem">
<label class="form-label">Textual inversions <span class="muted">(comma-separated; always loaded for this model)</span></label>
<input type="text" id="cfg-textual-inversion" class="form-input" placeholder="easynegative, bad-hands-5, my-emb.safetensors:myneg">
<span class="form-hint">Learned prompt tokens, so their trigger words work in any prompt sent to this model. A bare name is looked up in the server's embeddings directory (<code>image.embeddings_dir</code> — everything in it is loaded anyway); a path or a HuggingFace repo id also works, and <code>ref:token</code> renames the trigger word. A request's own <code>embeddings</code> list is loaded on top of these.</span>
</div>
<!-- video output quality (CRF) — only relevant for video models -->
<div class="form-row" style="margin-top:.75rem;max-width:340px">
<label class="form-label">Video output quality — CRF
......@@ -4305,6 +4311,8 @@ function openCfgModal(idx, cfgIdx){
document.getElementById('cfg-vae-tiling').checked = !!s.vae_tiling;
document.getElementById('cfg-clip-skip').value = (s.clip_skip != null ? s.clip_skip : '');
document.getElementById('cfg-prompt-weighting').checked = (s.prompt_weighting !== false);
document.getElementById('cfg-textual-inversion').value =
(Array.isArray(s.textual_inversion) ? s.textual_inversion.join(', ') : (s.textual_inversion || ''));
document.getElementById('cfg-output-crf').value = (s.output_crf != null ? s.output_crf : '');
document.getElementById('cfg-balanced-gpu-pct').value = (s.balanced_gpu_percent != null ? s.balanced_gpu_percent : '');
_populateCompSelects();
......@@ -5410,6 +5418,8 @@ async function saveModelConfig(){
clip_skip: (document.getElementById('cfg-clip-skip').value.trim() === ''
? null : parseInt(document.getElementById('cfg-clip-skip').value, 10)),
prompt_weighting: document.getElementById('cfg-prompt-weighting').checked,
textual_inversion: document.getElementById('cfg-textual-inversion').value
.split(',').map(v => v.trim()).filter(v => v !== ''),
};
try{
const r = await fetch(ROOT_PATH + '/admin/api/model-configure',{
......
......@@ -23,6 +23,7 @@ import base64
import io
import logging
import os
import re as _re
import time
import uuid
from typing import Optional
......@@ -887,6 +888,74 @@ def _apply_loras(pipeline, loras):
_log.warning("Could not apply LoRA weights: %s", e)
#: Diffusers' VAE downsamples by 8, so a side that is not a multiple of 8
#: cannot be expressed as a latent: the pipeline either raises or renders a
#: ragged edge, depending on the model.
_SIZE_MULTIPLE = 8
_SIZE_RE = _re.compile(r"^\s*(\d+)\s*[x\u00d7*]\s*(\d+)\s*$", _re.I)
def _size_limits(args=None):
"""(max_side, max_pixels) this install allows, from config."""
a = args or global_args
side = getattr(a, 'image_max_side', None) or 4096
pixels = getattr(a, 'image_max_pixels', None) or 4096 * 1024
return int(side), int(pixels)
def _parse_size(size, default_w: int, default_h: int, args=None):
"""``"WIDTHxHEIGHT"`` as a usable (width, height), or HTTP 400.
Three things used to go wrong here, all of them quietly. A size that did
not parse fell through to the default, so ``"1024"`` or ``"1k x 1k"``
generated a square nobody asked for and the response looked successful. A
side that was not a multiple of 8 went to the pipeline as-is, where the
VAE cannot represent it. And there was no ceiling at all, so ``4096x4096``
was accepted and the request came back as a CUDA OOM -- which reads like a
busy GPU rather than a size the caller chose.
So: a malformed size is a 400 that says what was wrong, an oversized one
is a 400 that names the limit, and a side off the grid is snapped to the
nearest multiple of 8 and logged. ``"auto"`` means the default, as it does
in the OpenAI image API.
"""
if not size or str(size).strip().lower() == "auto":
return default_w, default_h
m = _SIZE_RE.match(str(size))
if not m:
raise HTTPException(status_code=400, detail=(
f"size {size!r} is not understood: expected \"WIDTHxHEIGHT\" in "
f"pixels (e.g. \"1024x1024\"), or \"auto\""))
width, height = int(m.group(1)), int(m.group(2))
if width == 0 or height == 0:
# Snapping this up to the 8-pixel minimum would answer an uninitialised
# client variable with an 8x8 image, which is not what it asked for.
raise HTTPException(status_code=400, detail=(
f"size {size!r} has a zero dimension"))
max_side, max_pixels = _size_limits(args)
if width > max_side or height > max_side:
raise HTTPException(status_code=400, detail=(
f"size {width}x{height} exceeds the maximum side of {max_side} px "
f"(image.max_side)"))
if width * height > max_pixels:
raise HTTPException(status_code=400, detail=(
f"size {width}x{height} is {width * height / 1e6:.1f} MP, above "
f"this install's limit of {max_pixels / 1e6:.1f} MP "
f"(image.max_pixels). Raise it if the card can take it."))
snapped_w = max(_SIZE_MULTIPLE,
int(round(width / _SIZE_MULTIPLE)) * _SIZE_MULTIPLE)
snapped_h = max(_SIZE_MULTIPLE,
int(round(height / _SIZE_MULTIPLE)) * _SIZE_MULTIPLE)
if (snapped_w, snapped_h) != (width, height):
_log.info("size %dx%d is not on the 8-pixel grid the VAE needs — "
"generating %dx%d", width, height, snapped_w, snapped_h)
return snapped_w, snapped_h
def _resolve_clip_skip(request, model_config, args=None):
"""``clip_skip`` for this request: asked for, else the model's, else global.
......@@ -918,6 +987,30 @@ def _resolve_prompt_weighting(request, model_config, args=None) -> bool:
return True
def _apply_embeddings(pipeline, request, model_config=None, args=None):
"""Load this request's textual inversions onto the pipeline, if any.
Must happen before the prompt is tokenized: loading registers the trigger
word, and until it is registered the word tokenizes as ordinary text, so
the prompt quietly asks for something else. A name the caller gave
explicitly and that cannot be loaded is a 400 -- rendering without it
would look like a success.
"""
try:
from codai.models import textual_inversion as ti
except Exception:
return []
refs = list(getattr(request, 'embeddings', None) or [])
refs += list((model_config or {}).get('textual_inversion') or [])
directory = getattr(args or global_args, 'image_embeddings_dir', None)
if not refs and not directory:
return []
try:
return ti.apply(pipeline, refs, directory, strict=bool(refs))
except (FileNotFoundError, RuntimeError) as exc:
raise HTTPException(status_code=400, detail=str(exc))
def _weighted_embed_kwargs(pipeline, prompt, neg_prompt, clip_skip, weighting):
"""Encode prompt + negative ourselves, or None to use the pipeline's path.
......@@ -972,21 +1065,9 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request=
if getattr(request, 'loras', None):
_apply_loras(pipeline, request.loras)
# Determine size
width, height = 1024, 1024
if request.size:
parts = request.size.split("x")
if len(parts) == 2:
try:
width = int(parts[0])
height = int(parts[1])
except ValueError:
pass
if width != width or width == float('inf'):
width = 512
if height != height or height == float('inf'):
height = 512
# Determine size. A malformed or oversized size is a 400 from here, not a
# silent fall-back to a square the caller did not ask for.
width, height = _parse_size(request.size, 1024, 1024, global_args)
# Enable memory optimizations
try:
......@@ -1042,13 +1123,17 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request=
neg_prompt = getattr(request, 'negative_prompt', None) or ""
do_cfg = cfg_scale > 1.0
# Learned prompt tokens first: they change what the prompt tokenizes TO.
_ti_tokens = _apply_embeddings(pipeline, request, model_config, global_args)
# clip_skip and the weight grammar both change what the embeddings ARE, so
# they belong in the cache key: without them the first request's layer
# choice would be served to every later one.
_clip_skip = _resolve_clip_skip(request, model_config, global_args)
_weighting = _resolve_prompt_weighting(request, model_config, global_args)
model_id = (f"{_model_name}#{id(pipeline)}"
f"|cs={_clip_skip}|w={int(_weighting)}")
f"|cs={_clip_skip}|w={int(_weighting)}"
f"|ti={','.join(_ti_tokens)}")
# Some pipelines (Z-Image) return prompt embeddings as *per-sample lists*
# rather than stacked tensors, and the pipeline consumes/extends those
......@@ -1319,7 +1404,8 @@ async def _generate_with_diffusers(pipeline, request, global_args, http_request=
getattr(request, 'model', None) or model_id,
request.prompt,
{
"size": request.size,
"size": f"{width}x{height}",
"requested_size": request.size,
"n": request.n,
"steps": num_steps,
"guidance_scale": cfg_scale,
......@@ -1345,16 +1431,9 @@ async def _generate_with_sdcpp(sd_model, request, global_args, http_request=None
"""Generate images using stable-diffusion-cpp-python."""
import time
# Parse size
width, height = 512, 512
if request.size:
parts = request.size.split("x")
if len(parts) == 2:
try:
width = int(parts[0])
height = int(parts[1])
except ValueError:
pass
# Parse size (same validation as the diffusers path: sd.cpp's VAE is also
# an 8x downsample, and it answers an off-grid size with a C-level abort).
width, height = _parse_size(request.size, 512, 512, global_args)
# Acceleration/distillation defaults (Lightning / Turbo / LCM): sd.cpp can't
# fuse a diffusers LoRA, but honour the preset's low step-count / guidance and
......@@ -1468,7 +1547,8 @@ async def _generate_with_sdcpp(sd_model, request, global_args, http_request=None
getattr(request, 'model', None),
request.prompt,
{
"size": request.size,
"size": f"{width}x{height}",
"requested_size": request.size,
"n": request.n,
"steps": steps,
"seed": seed,
......@@ -1757,6 +1837,12 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
return await _generate_with_diffusers(pipeline, request, global_args,
http_request, model_config=_diff_cfg)
except HTTPException:
# The model loaded and the request itself was rejected (a bad
# size, an embedding that does not exist). Falling through to
# the next backend would answer a 400 with a confused report
# about which backends failed to load.
raise
except ImportError as e:
diffusers_error = str(e)
print(f"diffusers not available: {diffusers_error}")
......@@ -1798,6 +1884,8 @@ async def create_image_generation(request: ImageGenerationRequest, http_request:
sdcpp_error = f"Model '{model_name}' is not a local file, cannot use sd.cpp"
print(sdcpp_error)
except HTTPException:
raise # the request was rejected, not the backend
except ImportError as e:
sdcpp_error = str(e)
print(f"stable-diffusion-cpp-python not available: {sdcpp_error}")
......@@ -1898,6 +1986,7 @@ def _prompt_call_kwargs(pipeline, request, model_config=None):
Either way the negative prompt is passed -- the edit endpoints used to have
nowhere to put one, so an edit could only ever be steered positively.
"""
_apply_embeddings(pipeline, request, model_config, global_args)
neg = getattr(request, 'negative_prompt', None) or ""
kw = _weighted_embed_kwargs(
pipeline, request.prompt, neg,
......@@ -1991,14 +2080,12 @@ async def create_image_edit(request: ImageEditRequest, http_request: Request = N
source_img = _decode_b64_image(request.image)
width, height = source_img.size
if request.size:
parts = request.size.split("x")
if len(parts) == 2:
try:
width, height = int(parts[0]), int(parts[1])
source_img = source_img.resize((width, height))
except ValueError:
pass
if request.size and str(request.size).strip().lower() != "auto":
# Validated, so a typo is a 400 instead of an edit at some other
# size; snapped, because the VAE needs the 8-pixel grid.
width, height = _parse_size(request.size, width, height, global_args)
if (width, height) != source_img.size:
source_img = source_img.resize((width, height))
else:
# No size asked for: keep the photo's own shape at ~1 MP.
width, height = _default_edit_size(*source_img.size)
......@@ -2025,6 +2112,8 @@ async def create_image_edit(request: ImageEditRequest, http_request: Request = N
**_prompt_call_kwargs(pipe, request, _edit_cfg),
)
)
except HTTPException:
raise # a 400 from here is the caller's, not ours
except Exception as e:
raise HTTPException(status_code=500, detail=f"Image editing failed: {e}")
......@@ -2047,11 +2136,14 @@ class ImageInpaintRequest(BaseModel):
prompt: str
image: str # base64 source image
mask: str # base64 mask (white = inpaint region)
#: Omitted means "keep the source's shape" (see _default_edit_size), as on
#: /v1/images/edits. It used to default to 1024x1024, which squashed every
#: non-square photo -- and dragged the mask out of alignment with it.
size: Optional[str] = None
negative_prompt: Optional[str] = None
clip_skip: Optional[int] = None
prompt_weighting: Optional[bool] = None
n: int = 1
size: Optional[str] = "1024x1024"
response_format: Optional[str] = "url"
steps: Optional[int] = None
guidance_scale: Optional[float] = None
......@@ -2131,15 +2223,18 @@ async def create_image_inpaint(request: ImageInpaintRequest, http_request: Reque
import torch
source_img = _decode_b64_image(request.image)
mask_img = _decode_b64_image(request.mask).convert("L") # greyscale mask
if request.size:
parts = request.size.split("x")
if len(parts) == 2:
try:
w, h = int(parts[0]), int(parts[1])
source_img = source_img.resize((w, h))
mask_img = mask_img.resize((w, h))
except ValueError:
pass
if request.size and str(request.size).strip().lower() != "auto":
w, h = _parse_size(request.size, *source_img.size, args=global_args)
if (w, h) != source_img.size:
source_img = source_img.resize((w, h))
mask_img = mask_img.resize((w, h))
else:
# No size asked for: keep the photo's own shape on the grid, the
# same rule the edit endpoint follows.
w, h = _default_edit_size(*source_img.size)
if (w, h) != source_img.size:
source_img = source_img.resize((w, h))
mask_img = mask_img.resize((w, h))
seed = request.seed or getattr(global_args, 'image_seed', None)
generator = torch.Generator(device=pipe.device).manual_seed(seed) if seed else None
quality = request.quality or "standard"
......@@ -2160,6 +2255,8 @@ async def create_image_inpaint(request: ImageInpaintRequest, http_request: Reque
**_prompt_call_kwargs(pipe, request, _inp_cfg),
)
)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Inpainting failed: {e}")
images = [save_image_response(img, request.response_format, http_request) for img in result.images]
......@@ -2770,6 +2867,8 @@ class ImageOutfitRequest(BaseModel):
video: Optional[str] = None # base64 source video (video mode)
prompt: str # description of the new outfit
negative_prompt: Optional[str] = None
clip_skip: Optional[int] = None # None = the model's configured default
prompt_weighting: Optional[bool] = None # None = on
mask: Optional[str] = None # optional manual mask (base64); auto-generated if absent
steps: Optional[int] = 30
guidance_scale: Optional[float] = 7.5
......@@ -2851,15 +2950,13 @@ async def create_image_outfit(request: ImageOutfitRequest, http_request: Request
def _run():
kwargs = dict(
prompt=request.prompt,
image=img,
mask_image=mask_img,
num_inference_steps=request.steps or 30,
guidance_scale=request.guidance_scale or 7.5,
strength=request.strength or 0.99,
**_prompt_call_kwargs(pipeline, request),
)
if request.negative_prompt:
kwargs['negative_prompt'] = request.negative_prompt
if generator:
kwargs['generator'] = generator
if hasattr(pipeline, 'safety_checker'):
......@@ -2868,6 +2965,8 @@ async def create_image_outfit(request: ImageOutfitRequest, http_request: Request
try:
result_img = await asyncio.get_event_loop().run_in_executor(None, _run)
except HTTPException:
raise
except Exception as e:
raise HTTPException(status_code=500, detail=f"Outfit change failed: {e}")
......@@ -2918,6 +3017,9 @@ async def _outfit_video(request: ImageOutfitRequest, http_request):
generator = torch.Generator().manual_seed(request.seed) if request.seed is not None else None
def _process_frames():
# Encoded once for the whole clip: the prompt does not change
# between frames, and a text-encoder pass per frame is pure waste.
prompt_kwargs = _prompt_call_kwargs(pipeline, request)
for fname in sorted(os.listdir(frames_dir)):
fpath = os.path.join(frames_dir, fname)
img = PILImage.open(fpath).convert("RGB")
......@@ -2929,15 +3031,13 @@ async def _outfit_video(request: ImageOutfitRequest, http_request):
mask_arr = _generate_clothing_mask(img_arr)
mask_img = PILImage.fromarray(mask_arr)
kwargs = dict(
prompt=request.prompt,
image=img,
mask_image=mask_img,
num_inference_steps=request.steps or 30,
guidance_scale=request.guidance_scale or 7.5,
strength=request.strength or 0.99,
**prompt_kwargs,
)
if request.negative_prompt:
kwargs['negative_prompt'] = request.negative_prompt
if generator:
kwargs['generator'] = generator
if hasattr(pipeline, 'safety_checker'):
......
......@@ -310,6 +310,13 @@ class ImageConfig:
# Parse the A1111 prompt grammar -- (tag:1.3), [tag], BREAK. Off keeps
# brackets literal; the prompt is windowed past 77 tokens either way.
prompt_weighting: bool = True
# Request-size ceiling. A size above it is a 400 naming the limit, which
# is a better answer than the CUDA OOM an absurd size used to produce.
max_side: int = 4096
max_pixels: int = 4096 * 1024 # 4.2 MP -- 2048x2048, 2560x1440, …
# A flat directory of textual-inversion embeddings, loaded by filename so
# a prompt copied off a model card finds its trigger word. Empty = none.
embeddings_dir: str = ""
@dataclass
......@@ -2137,7 +2144,10 @@ class ConfigManager:
"vae_tiling": self.config.image.vae_tiling,
"clip_on_cpu": self.config.image.clip_on_cpu,
"clip_skip": self.config.image.clip_skip,
"prompt_weighting": self.config.image.prompt_weighting
"prompt_weighting": self.config.image.prompt_weighting,
"max_side": self.config.image.max_side,
"max_pixels": self.config.image.max_pixels,
"embeddings_dir": self.config.image.embeddings_dir
},
"archive": {
"enabled": self.config.archive.enabled,
......
......@@ -1227,6 +1227,9 @@ def main():
global_args.clip_on_cpu = config.image.clip_on_cpu
global_args.image_clip_skip = config.image.clip_skip
global_args.image_prompt_weighting = config.image.prompt_weighting
global_args.image_max_side = config.image.max_side
global_args.image_max_pixels = config.image.max_pixels
global_args.image_embeddings_dir = config.image.embeddings_dir or None
global_args.system_prompt = config.system_prompt
global_args.tools_closer_prompt = config.tools_closer_prompt
global_args.grammar_guided_gen = config.grammar_guided
......
......@@ -256,6 +256,30 @@ def _is_sdxl(pipeline) -> bool:
return getattr(pipeline, "text_encoder_2", None) is not None
def _expand_learned_tokens(pipeline, text: str, tokenizer) -> str:
"""``text`` with multi-vector textual inversions written out in full.
A textual inversion trained on several vectors is registered as ``tok``,
``tok_1``, ``tok_2``..., and a prompt only ever says ``tok``. Diffusers
expands it in ``maybe_convert_prompt`` on its way into the tokenizer, so
anything encoding a prompt itself has to do the same -- otherwise an
embedding like ``easynegative`` (eight vectors) contributes exactly one of
them and the prompt looks like it did nothing.
It replaces only registered trigger words, so the weight grammar passes
through untouched and this is safe to run before parsing.
"""
if not text:
return text
convert = getattr(pipeline, "maybe_convert_prompt", None)
if convert is None:
return text
try:
return convert(text, tokenizer)
except Exception:
return text
def _hidden(out, text_encoder, clip_skip: Optional[int], sdxl: bool):
"""The hidden state a diffusers pipeline would have used.
......@@ -331,8 +355,13 @@ def encode_weighted_prompt(pipeline, prompt: str, negative_prompt: str = "",
tokenized = []
n_windows = 1
for tokenizer, _ in pairs:
pos = tokenize_weighted(tokenizer, prompt or "", plain=plain)
neg = tokenize_weighted(tokenizer, negative_prompt or "", plain=plain)
pos = tokenize_weighted(
tokenizer, _expand_learned_tokens(pipeline, prompt, tokenizer),
plain=plain)
neg = tokenize_weighted(
tokenizer,
_expand_learned_tokens(pipeline, negative_prompt, tokenizer),
plain=plain)
tokenized.append((tokenizer, pos, neg))
n_windows = max(n_windows, len(pos[0]), len(neg[0]))
......
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Learned prompt tokens: ``easynegative`` and the rest of the embeddings.
A textual inversion is a handful of vectors trained to sit in CLIP's embedding
table under a word of their own. Writing that word in a prompt is how half the
Stable-Diffusion ecosystem's negative prompts are built -- ``easynegative``,
``bad-hands-5``, ``FastNegativeV2`` -- and without the embedding loaded the
word is just an unknown token, so the prompt silently asks for nothing.
Three ways to get one loaded, in the order a request resolves them:
* ``embeddings: ["easynegative"]`` on the request, for an ad-hoc one.
* ``textual_inversion`` on the model's config, for the ones a model always
wants.
* every file in ``image.embeddings_dir``, discovered by filename -- which is
how A1111 works, and what a prompt copied from a model card assumes.
A reference is a bare name (matched against the directory), a path to a
``.safetensors`` / ``.pt`` / ``.bin`` file, or a HuggingFace repo id. Appending
``:token`` renames the trigger word: ``my-emb.safetensors:myneg``.
Two details decide whether this actually works:
* **SDXL keeps two text encoders**, so an SDXL-format embedding holds a
``clip_l`` and a ``clip_g`` tensor that have to go into the matching
tokenizer/encoder pair. Loading only the first gives the token an embedding
in one encoder and noise in the other.
* **A multi-vector embedding occupies several tokens**, registered as
``tok``, ``tok_1``, ``tok_2``... Diffusers expands the prompt through
``maybe_convert_prompt`` before tokenizing; anything encoding a prompt
itself has to do the same or only the first vector is ever used. That is
handled in :mod:`codai.models.prompt_weighting`.
"""
import logging
import os
from typing import Dict, Iterable, List, Optional, Tuple
_log = logging.getLogger(__name__)
#: What an embedding file looks like on disk.
SUFFIXES = (".safetensors", ".pt", ".bin", ".ckpt")
#: Marks the tokens already loaded into a pipeline, so a second request does
#: not try to register the same trigger word again (diffusers raises on that).
_LOADED_ATTR = "_coderai_textual_inversions"
def _split_ref(ref: str) -> Tuple[str, Optional[str]]:
"""``"file.safetensors:token"`` -> (``"file.safetensors"``, ``"token"``).
A Windows drive letter or an ``hf://`` style prefix would be mangled by a
naive split, so only a colon in the last path segment counts.
"""
ref = str(ref).strip()
head, sep, tail = ref.rpartition(":")
if sep and tail and "/" not in tail and os.sep not in tail and len(head) > 1:
return head, tail
return ref, None
def _token_for(ref: str, explicit: Optional[str]) -> str:
"""The trigger word: the one asked for, else the filename, as A1111 does."""
if explicit:
return explicit
base = os.path.basename(ref.rstrip("/"))
for suffix in SUFFIXES:
if base.endswith(suffix):
return base[: -len(suffix)]
return base
def discover(directory: Optional[str]) -> Dict[str, str]:
"""``{token: path}`` for every embedding file in ``directory``.
Not recursive and not sorted by anything meaningful: a flat directory of
trigger words is the convention these files arrive in.
"""
out: Dict[str, str] = {}
if not directory:
return out
directory = os.path.expanduser(directory)
try:
names = os.listdir(directory)
except OSError as exc:
_log.warning("embeddings_dir %s cannot be read: %s", directory, exc)
return out
for name in sorted(names):
if not name.endswith(SUFFIXES):
continue
path = os.path.join(directory, name)
if os.path.isfile(path):
out[_token_for(name, None)] = path
return out
def resolve(ref: str, directory: Optional[str] = None) -> Tuple[str, str]:
"""``(source, token)`` for one reference, or raise ``FileNotFoundError``.
``source`` is a path that exists, or a HuggingFace repo id left for
diffusers to fetch. A bare name is looked up in ``directory`` first,
because that is what a prompt copied off a model card means by it.
"""
raw, explicit = _split_ref(ref)
token = _token_for(raw, explicit)
if os.path.isfile(os.path.expanduser(raw)):
return os.path.expanduser(raw), token
if directory:
# Looked up by the name of the FILE, not by the trigger word: with
# "easynegative:myneg" the word is myneg and the file is still
# easynegative.safetensors.
name = _token_for(raw, None)
found = discover(directory)
if name in found:
return found[name], token
# a name given with its extension, or a file sitting beside the others
candidate = os.path.join(os.path.expanduser(directory),
os.path.basename(raw))
if os.path.isfile(candidate):
return candidate, token
if "/" in raw and not raw.startswith((".", "/", "~")):
return raw, token # a repo id; diffusers resolves and caches it
raise FileNotFoundError(
f"textual inversion {ref!r} not found: no such file, nothing by that "
f"name in the embeddings directory"
+ (f" ({directory})" if directory else " (image.embeddings_dir is not set)")
+ ", and it is not a HuggingFace repo id")
def loaded_tokens(pipeline) -> set:
"""Trigger words already registered on this pipeline."""
return set(getattr(pipeline, _LOADED_ATTR, None) or ())
def _load_one(pipeline, source: str, token: str) -> None:
"""Register ``token`` on the pipeline, both encoders included on SDXL."""
sdxl = getattr(pipeline, "text_encoder_2", None) is not None
state = None
if sdxl and os.path.isfile(source) and source.endswith(".safetensors"):
try:
from safetensors.torch import load_file
state = load_file(source)
except Exception as exc:
_log.debug("cannot inspect %s as safetensors (%s)", source, exc)
if state is not None and "clip_g" in state and "clip_l" in state:
# The SDXL layout: one tensor per text encoder, and they are not
# interchangeable -- 768 wide for clip_l, 1280 for clip_g.
pipeline.load_textual_inversion(
state["clip_l"], token=token,
text_encoder=pipeline.text_encoder, tokenizer=pipeline.tokenizer)
pipeline.load_textual_inversion(
state["clip_g"], token=token,
text_encoder=pipeline.text_encoder_2, tokenizer=pipeline.tokenizer_2)
else:
pipeline.load_textual_inversion(source, token=token)
def apply(pipeline, refs: Iterable[str] = (), directory: Optional[str] = None,
strict: bool = False) -> List[str]:
"""Load ``refs`` (plus everything in ``directory``) onto ``pipeline``.
Returns the trigger words now usable in a prompt. Already-loaded ones are
skipped, so this is cheap to call on every request -- which it is, because
a pipeline outlives the request that loaded it and a later request must
find the same tokens there.
``strict`` raises on a reference that cannot be resolved or loaded: that is
for the ones a caller named explicitly, where quietly rendering without the
embedding is the failure mode worth avoiding. Directory sweeps are never
strict -- one unreadable file should not break every request.
"""
if not hasattr(pipeline, "load_textual_inversion"):
if strict and list(refs):
raise RuntimeError("this pipeline cannot load textual inversions")
return []
already = loaded_tokens(pipeline)
# (source, token, asked for by name) -- the last flag decides whether a
# failure is this request's problem or just a note in the log.
wanted: List[Tuple[str, str, bool]] = []
for ref in refs or ():
if not str(ref).strip():
continue
source, token = resolve(str(ref), directory) # may raise, by design
wanted.append((source, token, True))
named = {token for _, token, _ in wanted}
for token, path in discover(directory).items():
if token not in named:
wanted.append((path, token, False))
added = []
for source, token, explicit in wanted:
if token in already:
continue
try:
_load_one(pipeline, source, token)
except Exception as exc:
if strict and explicit:
raise RuntimeError(
f"textual inversion {token!r} could not be loaded: {exc}") from exc
_log.warning("textual inversion %r (%s) not loaded: %s",
token, source, exc)
continue
already.add(token)
added.append(token)
setattr(pipeline, _LOADED_ATTR, already)
if added:
_log.info("Textual inversions loaded: %s", ", ".join(sorted(added)))
return sorted(already)
......@@ -57,6 +57,12 @@ class ImageGenerationRequest(BaseModel):
"Take the text embedding this many CLIP layers from the end instead of the last. "
"The booru-tag SDXL fine-tune families (Pony, Illustrious, NoobAI) are trained at 2. "
"Omitted uses the model's configured default, then the pipeline's own."))
embeddings: Optional[List[str]] = Field(None, description=(
"Textual-inversion embeddings to load before generating, so their trigger "
"words work in the prompt (e.g. [\"easynegative\"]). A bare name is looked "
"up in image.embeddings_dir; a path or a HuggingFace repo id also works, and "
"'ref:token' renames the trigger word. Everything in embeddings_dir, plus the "
"model's own textual_inversion list, is loaded whether this is set or not."))
prompt_weighting: Optional[bool] = Field(None, description=(
"Parse the A1111 prompt grammar: '(tag:1.3)' to emphasise, '[tag]' to attenuate, "
"'(tag)' for 1.1, nesting multiplies, '\\\\(' escapes, and BREAK starts a new "
......
......@@ -28,7 +28,7 @@ reference; these are the deep dives.
| [ocr.md](ocr.md) | The `/v1/ocr` document-transcription subsystem |
| [chunking.md](chunking.md) | `POST /v1/chunk`: splitting a document for a retrieval index, with the structure decided per request by a template |
| [expressive-tts.md](expressive-tts.md) | Expressive text-to-speech |
| [image-prompts.md](image-prompts.md) | Image prompts: the `(tag:1.3)` / `[tag]` / `BREAK` grammar, prompts past CLIP's 77 tokens, `clip_skip`, negative prompts |
| [image-prompts.md](image-prompts.md) | Image prompts: the `(tag:1.3)` / `[tag]` / `BREAK` grammar, prompts past CLIP's 77 tokens, `clip_skip`, negative prompts, textual-inversion embeddings, size validation |
| [zimage-lora-training.md](zimage-lora-training.md) | LoRA training for image models |
| [dtype-auto-selection.md](dtype-auto-selection.md) | How model dtype is chosen automatically |
......
......@@ -49,6 +49,7 @@ take:
| `negative_prompt` | What to avoid. The counterpart of the quality tags (`score_4, score_5, score_6, worst quality`) and as load-bearing as the prompt itself. |
| `clip_skip` | Read the embedding this many CLIP layers from the end. Omitted uses the model's configured value, then the pipeline's own default. |
| `prompt_weighting` | `false` keeps brackets literal, for a caller sending prose. The windowing stays on either way, so turning it off never costs tokens. |
| `embeddings` | Textual inversions to load, so their trigger words work in the prompt — `["easynegative"]`. |
```json
{
......@@ -80,6 +81,62 @@ through. Until v0.4.89 the negative prompt was not: the call hardcoded an empty
string, so for every GGUF image model whatever the caller asked to avoid was
discarded between the API and the engine.
## Size
`size` is `"WIDTHxHEIGHT"` in pixels, or `"auto"` for the model's default.
A size that does not parse is a `400` saying so, rather than a silent
fall-back: `"1024"` used to generate a square nobody asked for, and the
response looked like a success. A side that is not a multiple of 8 is snapped
to the nearest one and logged, because the VAE downsamples by 8 and cannot
represent an odd side. A size above the install's ceiling is a `400` naming the
limit — `4096x4096` used to be accepted and come back as a CUDA OOM, which
reads like a busy GPU rather than a size the caller chose.
The ceiling is `image.max_side` (default 4096) and `image.max_pixels` (default
4.2 MP, which covers 2048×2048 and 2560×1440). Raise both if the card can take
it.
On `/v1/images/edits` and `/v1/images/inpaint`, omitting `size` keeps the
source photo's own aspect at about 1 MP on a 64-pixel grid. Both used to
default to a square, which squashed every non-square photo — and on inpaint it
pulled the mask out of alignment with the image as well.
## Learned prompt tokens (textual inversion)
A textual inversion is a few vectors trained to sit in CLIP's embedding table
under a word of their own. Writing that word is how much of the ecosystem's
negative prompting is built — `easynegative`, `bad-hands-5`, `FastNegativeV2`
— and until the file is loaded the word is an ordinary unknown token, so the
prompt quietly asks for nothing.
Three ways to get one loaded, all of which can be combined:
* `embeddings: ["easynegative"]` on the request, for an ad-hoc one.
* `textual_inversion` on the model's config (admin → Models → Prompt
Handling), for the ones a model always wants.
* every file in `image.embeddings_dir`, discovered by filename — which is how
A1111 works, and what a prompt copied off a model card assumes.
A reference is a bare name (looked up in the directory), a path to a
`.safetensors` / `.pt` / `.bin`, or a HuggingFace repo id. `ref:token` renames
the trigger word: `my-emb.safetensors:myneg`.
A name the caller gave that cannot be loaded is a `400` — rendering without it
would look like a success. A file in the directory that fails to load is a line
in the log, because one bad file should not break every request.
Two details that decide whether this works at all, both handled:
* **SDXL keeps two text encoders**, so an SDXL-format embedding holds a
`clip_l` and a `clip_g` tensor that must go into the matching
tokenizer/encoder pair. Loading only the first gives the trigger word an
embedding in one encoder and nothing in the other.
* **A multi-vector embedding occupies several tokens**, registered as `tok`,
`tok_1`, `tok_2`… A prompt only ever writes `tok`, so it is expanded before
tokenizing. Without that, an eight-vector embedding contributes one vector
and appears to do nothing.
## What is not covered
Flux, SD3 and Z-Image carry a T5 encoder with a much longer limit and
......@@ -88,10 +145,6 @@ path — the weight grammar is a CLIP-era convention and a T5 reads it as text,
which is what those models expect. Negative prompts work there as the pipeline
supports them.
Textual-inversion embeddings (`easynegative` and friends) are not loaded by
name yet; a LoRA is the supported way to add learned tokens
(see [zimage-lora-training.md](zimage-lora-training.md)).
## Implementation
`codai/models/prompt_weighting.py` holds the parser, the windowing and the
......@@ -99,5 +152,12 @@ encoder, and documents how `clip_skip` maps to a hidden-state index — SDXL
takes the penultimate layer by default and counts from there, SD 1.x takes the
final output and needs the layer norm applied by hand. Getting that mapping
wrong shifts every image, so it mirrors diffusers exactly.
`codai/models/textual_inversion.py` resolves and loads the embeddings, and
tracks what a pipeline already carries — a pipeline outlives the request that
loaded it, and diffusers raises on registering the same trigger word twice.
`tests/test_prompt_weighting_and_clip_skip.py` pins the grammar table, the
window arithmetic, both layer conventions and the three defects above.
window arithmetic and both layer conventions;
`tests/test_image_size_and_embeddings.py` pins the size rules and the
embedding resolution, including the SDXL two-encoder split and the
multi-vector expansion.
"""Request size, and the learned prompt tokens a prompt needs loaded.
Two sets of silent failures, both of which answered 200 with an image:
* **size.** A size that did not parse fell through to the default, so
``"1024"`` generated a square nobody asked for. A side that was not a
multiple of 8 went to the pipeline as-is, where the VAE cannot represent it.
And nothing capped it, so ``4096x4096`` came back as a CUDA OOM -- which
reads like a busy GPU rather than a size the caller chose.
* **textual inversions.** ``easynegative`` and the rest are vectors living in
CLIP's embedding table under a word of their own. Until the file is loaded
the word is an ordinary unknown token, so half the ecosystem's negative
prompts asked for nothing at all. A multi-vector embedding needs its prompt
expanded too, or only the first of its vectors is ever used.
"""
import os
import types
import pytest
from fastapi import HTTPException
from codai.api import images as images_mod
from codai.models import textual_inversion as ti
@pytest.fixture
def no_limits():
"""An install with the shipped defaults: 4096 px a side, 4.2 MP."""
return types.SimpleNamespace()
# --------------------------------------------------------------------------- #
# size
# --------------------------------------------------------------------------- #
@pytest.mark.parametrize("size,expected", [
("1024x1024", (1024, 1024)),
("832x1216", (832, 1216)), # the usual SDXL portrait
(" 768 x 512 ", (768, 512)),
("768X512", (768, 512)),
("768×512", (768, 512)), # the unicode multiplication sign
])
def test_a_good_size_is_taken_as_written(size, expected, no_limits):
assert images_mod._parse_size(size, 1024, 1024, no_limits) == expected
@pytest.mark.parametrize("size", [None, "", "auto", "AUTO"])
def test_nothing_asked_for_means_the_default(size, no_limits):
assert images_mod._parse_size(size, 640, 480, no_limits) == (640, 480)
@pytest.mark.parametrize("size", ["1024", "1k x 1k", "1024x", "x1024", "big",
"1024*1024*1024", "-512x512", "0x0x0"])
def test_a_size_that_does_not_parse_is_a_400(size, no_limits):
"""It used to silently generate at the default instead."""
with pytest.raises(HTTPException) as exc:
images_mod._parse_size(size, 1024, 1024, no_limits)
assert exc.value.status_code == 400
assert "WIDTHxHEIGHT" in exc.value.detail
def test_the_error_says_what_was_sent(no_limits):
with pytest.raises(HTTPException) as exc:
images_mod._parse_size("1k x 1k", 1024, 1024, no_limits)
assert "'1k x 1k'" in exc.value.detail
@pytest.mark.parametrize("size,expected", [
("1000x1000", (1000, 1000)), # already a multiple of 8
("1001x1001", (1000, 1000)), # nearest
("1023x1023", (1024, 1024)),
("3x3", (8, 8)), # never below one latent pixel
])
def test_a_side_off_the_grid_is_snapped(size, expected, no_limits):
"""The VAE downsamples by 8: an odd side is not representable."""
assert images_mod._parse_size(size, 1024, 1024, no_limits) == expected
@pytest.mark.parametrize("size", ["0x0", "0x512", "512x0"])
def test_a_zero_dimension_is_a_400(size, no_limits):
"""Snapping it up would answer an uninitialised client variable with an
8x8 image."""
with pytest.raises(HTTPException) as exc:
images_mod._parse_size(size, 1024, 1024, no_limits)
assert exc.value.status_code == 400
assert "zero dimension" in exc.value.detail
def test_an_oversized_side_is_a_400_naming_the_limit(no_limits):
with pytest.raises(HTTPException) as exc:
images_mod._parse_size("8192x512", 1024, 1024, no_limits)
assert exc.value.status_code == 400
assert "maximum side" in exc.value.detail and "4096" in exc.value.detail
def test_too_many_pixels_is_a_400_not_an_oom(no_limits):
"""4096x4096 used to be accepted and come back as a CUDA OOM."""
with pytest.raises(HTTPException) as exc:
images_mod._parse_size("4096x4096", 1024, 1024, no_limits)
assert exc.value.status_code == 400
assert "MP" in exc.value.detail and "image.max_pixels" in exc.value.detail
@pytest.mark.parametrize("size", ["2048x2048", "2560x1440", "1920x1080",
"1024x1024", "832x1216"])
def test_every_ordinary_size_is_under_the_default_ceiling(size, no_limits):
images_mod._parse_size(size, 1024, 1024, no_limits) # no raise
def test_the_ceiling_is_configurable_upwards():
"""An operator with the VRAM for it can raise both."""
big = types.SimpleNamespace(image_max_side=8192, image_max_pixels=8192 * 8192)
assert images_mod._parse_size("8192x8192", 1024, 1024, big) == (8192, 8192)
def test_the_dead_nan_guard_is_gone():
"""`if width != width` on a value that is already int can never fire."""
src = open("codai/api/images.py").read()
assert "if width != width or width == float('inf')" not in src
def test_inpaint_no_longer_defaults_to_a_square():
"""A 1024x1024 default squashed every non-square photo -- and pulled the
mask out of alignment with it, which is worse than on a plain edit."""
assert images_mod.ImageInpaintRequest.model_fields["size"].default is None
# --------------------------------------------------------------------------- #
# textual inversions: resolving
# --------------------------------------------------------------------------- #
@pytest.fixture
def embeddings_dir(tmp_path):
d = tmp_path / "embeddings"
d.mkdir()
for name in ("easynegative.safetensors", "bad-hands-5.pt", "notes.txt"):
(d / name).write_bytes(b"\x00")
return str(d)
def test_a_bare_name_is_looked_up_in_the_directory(embeddings_dir):
"""What a prompt copied off a model card means by `easynegative`."""
source, token = ti.resolve("easynegative", embeddings_dir)
assert source == os.path.join(embeddings_dir, "easynegative.safetensors")
assert token == "easynegative"
def test_a_name_with_its_extension_also_resolves(embeddings_dir):
source, token = ti.resolve("bad-hands-5.pt", embeddings_dir)
assert source.endswith("bad-hands-5.pt")
assert token == "bad-hands-5"
def test_a_path_is_taken_as_given(embeddings_dir, tmp_path):
path = tmp_path / "elsewhere.safetensors"
path.write_bytes(b"\x00")
source, token = ti.resolve(str(path), None)
assert source == str(path)
assert token == "elsewhere"
def test_a_trigger_word_can_be_renamed(embeddings_dir):
source, token = ti.resolve("easynegative:myneg", embeddings_dir)
assert source.endswith("easynegative.safetensors")
assert token == "myneg"
def test_a_repo_id_is_left_for_diffusers_to_fetch():
source, token = ti.resolve("sd-concepts-library/cat-toy", None)
assert source == "sd-concepts-library/cat-toy"
assert token == "cat-toy"
def test_a_missing_embedding_says_where_it_looked(embeddings_dir):
with pytest.raises(FileNotFoundError) as exc:
ti.resolve("nosuchthing", embeddings_dir)
assert "nosuchthing" in str(exc.value)
assert embeddings_dir in str(exc.value)
def test_a_missing_embedding_with_no_directory_says_so():
with pytest.raises(FileNotFoundError) as exc:
ti.resolve("nosuchthing", None)
assert "image.embeddings_dir is not set" in str(exc.value)
def test_discovery_takes_the_embeddings_and_leaves_the_rest(embeddings_dir):
found = ti.discover(embeddings_dir)
assert set(found) == {"easynegative", "bad-hands-5"} # not notes.txt
def test_discovery_of_a_directory_that_is_not_there_is_not_an_error():
assert ti.discover("/nonexistent/embeddings") == {}
assert ti.discover(None) == {}
# --------------------------------------------------------------------------- #
# textual inversions: loading
# --------------------------------------------------------------------------- #
class _FakePipeline:
"""Records what was loaded, and into which encoder pair."""
def __init__(self, sdxl=False, fail=()):
self.calls = []
self.fail = set(fail)
self.tokenizer = "tok"
self.text_encoder = "enc"
if sdxl:
self.tokenizer_2 = "tok2"
self.text_encoder_2 = "enc2"
def load_textual_inversion(self, source, token=None, text_encoder=None,
tokenizer=None):
if token in self.fail:
raise RuntimeError("bad file")
self.calls.append((source, token, tokenizer))
def test_loading_registers_the_trigger_word(embeddings_dir):
pipe = _FakePipeline()
tokens = ti.apply(pipe, ["easynegative"], embeddings_dir)
assert "easynegative" in tokens
assert pipe.calls[0][1] == "easynegative"
def test_everything_in_the_directory_is_loaded_unasked(embeddings_dir):
"""A1111's behaviour, and what a copied prompt assumes."""
pipe = _FakePipeline()
tokens = ti.apply(pipe, [], embeddings_dir)
assert set(tokens) == {"easynegative", "bad-hands-5"}
def test_a_second_request_does_not_load_them_again(embeddings_dir):
"""The pipeline outlives the request; diffusers raises on a re-register."""
pipe = _FakePipeline()
ti.apply(pipe, ["easynegative"], embeddings_dir)
before = len(pipe.calls)
tokens = ti.apply(pipe, ["easynegative"], embeddings_dir)
assert len(pipe.calls) == before
assert "easynegative" in tokens
def test_a_name_the_caller_gave_that_cannot_load_is_an_error(embeddings_dir):
"""Rendering without it would look like a success."""
pipe = _FakePipeline(fail=("easynegative",))
with pytest.raises(RuntimeError):
ti.apply(pipe, ["easynegative"], embeddings_dir, strict=True)
def test_one_bad_file_in_the_directory_does_not_break_the_request(embeddings_dir):
"""A sweep is never strict: the other embeddings still load."""
pipe = _FakePipeline(fail=("bad-hands-5",))
tokens = ti.apply(pipe, [], embeddings_dir, strict=True)
assert tokens == ["easynegative"]
def test_an_sdxl_embedding_goes_into_both_encoders(tmp_path, monkeypatch):
"""clip_l is 768 wide and clip_g is 1280: loading only one gives the token
an embedding in one encoder and nothing in the other."""
path = tmp_path / "sdxlneg.safetensors"
path.write_bytes(b"\x00")
monkeypatch.setattr(
"safetensors.torch.load_file",
lambda p: {"clip_l": "L", "clip_g": "G"})
pipe = _FakePipeline(sdxl=True)
ti.apply(pipe, [str(path)], None)
assert [(c[0], c[2]) for c in pipe.calls] == [("L", "tok"), ("G", "tok2")]
assert {c[1] for c in pipe.calls} == {"sdxlneg"}
def test_a_single_tensor_embedding_takes_the_plain_path(tmp_path, monkeypatch):
path = tmp_path / "sd15neg.pt"
path.write_bytes(b"\x00")
pipe = _FakePipeline(sdxl=True)
ti.apply(pipe, [str(path)], None)
assert pipe.calls == [(str(path), "sd15neg", None)]
def test_a_pipeline_that_cannot_load_them_is_left_alone():
assert ti.apply(types.SimpleNamespace(), [], None) == []
with pytest.raises(RuntimeError):
ti.apply(types.SimpleNamespace(), ["easynegative"], None, strict=True)
# --------------------------------------------------------------------------- #
# textual inversions: the prompt side
# --------------------------------------------------------------------------- #
def test_a_multi_vector_embedding_is_written_out_before_tokenizing():
"""`easynegative` is eight vectors registered as easynegative,
easynegative_1 ... If the prompt is not expanded, seven are unused."""
from codai.models.prompt_weighting import _expand_learned_tokens
def convert(text, tokenizer):
return text.replace("easynegative",
"easynegative easynegative_1 easynegative_2")
pipe = types.SimpleNamespace(maybe_convert_prompt=convert)
out = _expand_learned_tokens(pipe, "blurry, easynegative", "tok")
assert out == "blurry, easynegative easynegative_1 easynegative_2"
def test_expansion_leaves_the_weight_grammar_alone():
from codai.models.prompt_weighting import _expand_learned_tokens
pipe = types.SimpleNamespace(
maybe_convert_prompt=lambda text, tok: text.replace("emb", "emb emb_1"))
assert _expand_learned_tokens(pipe, "(emb:1.3) BREAK x", "tok") == \
"(emb emb_1:1.3) BREAK x"
def test_a_pipeline_without_expansion_is_no_obstacle():
from codai.models.prompt_weighting import _expand_learned_tokens
assert _expand_learned_tokens(types.SimpleNamespace(), "a cat", "tok") == "a cat"
def test_expansion_failure_falls_back_to_the_prompt_as_written():
from codai.models.prompt_weighting import _expand_learned_tokens
def boom(text, tok):
raise RuntimeError("no")
pipe = types.SimpleNamespace(maybe_convert_prompt=boom)
assert _expand_learned_tokens(pipe, "a cat", "tok") == "a cat"
# --------------------------------------------------------------------------- #
# the API surface
# --------------------------------------------------------------------------- #
def test_the_request_takes_an_embeddings_list():
from codai.pydantic.imagerequest import ImageGenerationRequest
r = ImageGenerationRequest(model="m", prompt="p",
embeddings=["easynegative", "x.pt:y"])
assert r.embeddings == ["easynegative", "x.pt:y"]
assert ImageGenerationRequest(model="m", prompt="p").embeddings is None
def test_a_missing_embedding_a_caller_named_is_a_400(monkeypatch, embeddings_dir):
pipe = _FakePipeline()
req = types.SimpleNamespace(embeddings=["nosuchthing"])
args = types.SimpleNamespace(image_embeddings_dir=embeddings_dir)
with pytest.raises(HTTPException) as exc:
images_mod._apply_embeddings(pipe, req, {}, args)
assert exc.value.status_code == 400
assert "nosuchthing" in exc.value.detail
def test_nothing_configured_and_nothing_asked_for_does_nothing():
pipe = _FakePipeline()
req = types.SimpleNamespace(embeddings=None)
assert images_mod._apply_embeddings(
pipe, req, {}, types.SimpleNamespace()) == []
assert pipe.calls == []
def test_a_model_can_carry_its_own_embeddings(embeddings_dir):
"""So every request to it gets the trigger words, whatever the client is."""
pipe = _FakePipeline()
req = types.SimpleNamespace(embeddings=None)
tokens = images_mod._apply_embeddings(
pipe, req, {"textual_inversion": ["easynegative"]},
types.SimpleNamespace(image_embeddings_dir=embeddings_dir))
assert "easynegative" in tokens
assert any(c[1] == "easynegative" for c in pipe.calls)
def test_the_loaded_token_set_is_part_of_the_embedding_cache_key():
"""The same prompt text tokenizes differently once a trigger word exists,
so a cache entry from before the load must not be reused."""
src = open("codai/api/images.py").read()
assert "|ti={','.join(_ti_tokens)}" in src
def test_both_are_configurable_per_model_and_server_wide():
assert '"textual_inversion"' in open("codai/admin/routes.py").read()
assert "cfg-textual-inversion" in open(
"codai/admin/templates/models.html").read()
from codai.config import ImageConfig
cfg = ImageConfig()
assert cfg.embeddings_dir == ""
assert cfg.max_side == 4096 and cfg.max_pixels == 4096 * 1024
def test_a_400_is_not_relabelled_as_a_server_fault():
"""The edit endpoints caught every Exception and answered 500, so a 400
raised while preparing the prompt -- an embedding that does not exist --
arrived as "Image editing failed: 400: ...", a server fault it is not."""
src = open("codai/api/images.py").read()
for message in ("Image editing failed", "Inpainting failed",
"Outfit change failed"):
idx = src.index(message)
assert "except HTTPException:" in src[max(0, idx - 250):idx], message
def test_a_rejected_request_is_not_mistaken_for_a_failed_backend():
"""/v1/images/generations tries diffusers, then sd.cpp, and reports which
backends failed if both do. A 400 from the request itself (a bad size, a
missing embedding) raised on that path used to be caught as a backend
failure, so the caller got a report about loading instead of the reason."""
src = open("codai/api/images.py").read()
head = src.index("Step 4: Model not loaded")
tail = src.index("Step 5: Both backends failed")
section = src[head:tail]
# one pass-through per backend attempt, each above its catch-all
assert section.count("except HTTPException:") == 2
for idx in (section.index("diffusers_error = str(e)\n print(f\"diffusers error"),
section.index("sdcpp_error = str(e)\n print(f\"sd.cpp error")):
assert "except HTTPException:" in section[max(0, idx - 700):idx]
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment