feat(v5): prompt markup for H3 -- (word:1.5), [phrase@t0-t1], [a|b]
Prompt Markup strips it before the encoder; Apply Prompt Markup puts weights and timed windows on attention and blends on the conditioning. Where the words sit is the H3 module's (text_tokenizer, prompt_rows, video_time_rows). Wired into both H3 presets. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
digital-garbage
co-authored by
Claude Opus 5.5
Claude Opus 5.5
parent
3cd0c1e00a
commit
9d028ccc92
@@ -54,6 +54,8 @@ FROZEN = {
|
||||
"FunPackLoadModifiers": ["model", "settings"],
|
||||
"FunPackLatentUpscalerLoader": ["upscaler_name"],
|
||||
"FunPackCutOpening": ["images", "frames", "fps", "audio"],
|
||||
"FunPackPromptMarkup": ["text"],
|
||||
"FunPackApplyPromptMarkup": ["model", "clip", "positive", "latent", "markup"],
|
||||
"FunPackLatentResample": ["latent", "upscaler", "operation", "scale", "positive", "negative"],
|
||||
"FunPackModifierSettings": ["settings"],
|
||||
"FunPackDecode": ["samples", "vae", "model", "audio_vae"],
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
"""Prompt markup -- (word:1.5), [phrase@2-4], [a|b]. A node-only module.
|
||||
|
||||
Two nodes, because the markup has to be split around the encoder: it must come
|
||||
OUT before the encoder sees the text (Qwen reads "(word:1.5)" as punctuation in
|
||||
your sentence), and it can only be applied AFTER, once there is a conditioning
|
||||
to find the words in and a model to bias.
|
||||
|
||||
text -> Prompt Markup -> clean -> (any H3 encode node) -> positive
|
||||
-> markup ---------------------> Apply Prompt Markup
|
||||
(+ model, clip, latent)
|
||||
|
||||
Where the words sit, and how seconds map to rows, is the MODEL's business: model
|
||||
modules announce `text_tokenizer`, `prompt_rows` and `video_time_rows`. With
|
||||
none that recognise the encoder, the markup is stripped and said to be unused --
|
||||
never silently ignored.
|
||||
|
||||
The weight is an attention BIAS, and SLA runs any biased call dense, so a run
|
||||
with weights or timed phrases runs dense attention.
|
||||
"""
|
||||
|
||||
from .nodes import FunPackApplyPromptMarkup, FunPackPromptMarkup
|
||||
|
||||
ID = "conditioning_prompt_markup"
|
||||
TITLE = "Prompt markup"
|
||||
STAGE = "conditioning"
|
||||
CATEGORY = "conditioning"
|
||||
STATUS = "proven"
|
||||
|
||||
NODES = [FunPackPromptMarkup, FunPackApplyPromptMarkup]
|
||||
@@ -0,0 +1,135 @@
|
||||
"""Prompt markup: the text part. Pure -- strings in, spans out, tensors built.
|
||||
|
||||
Three markups, stripped in ONE left-to-right pass so every span indexes the
|
||||
same clean text (stripping them one after another left the first list's
|
||||
offsets pointing into a string that no longer existed):
|
||||
|
||||
* `(phrase:1.5)` / `[phrase:1.5]` -- weight. A logit bias of log(w) on that
|
||||
phrase's key positions: softmax(logits + log w) multiplies that key's share
|
||||
of the attention by w, which is what a weight is supposed to mean.
|
||||
* `[phrase@2-3.5]` / `[phrase:1.5@2-3.5]` -- a timed phrase: only the video
|
||||
rows inside the window see it (weighted); outside, it is masked from video.
|
||||
Text and audio rows keep the untimed bias. Confirmed working on H3 in v4.
|
||||
* `[phraseA|phraseB]` / `[A|B:0.8]` -- blend: (1-s)*A + s*B conditioning,
|
||||
row by row. Confirmed running end-to-end in v4; visual quality unjudged.
|
||||
|
||||
Weighting only exists because the text is IN the attention sequence (H3's
|
||||
single packed stream). Scaling embeddings would not work: Qwen's rows are
|
||||
contextual hidden states, with no per-word magnitude to turn up.
|
||||
"""
|
||||
|
||||
import math
|
||||
import re
|
||||
|
||||
_NUM = r"-?(?:\d+\.\d*|\.\d+|\d+)"
|
||||
_WEIGHTED = re.compile(r"(?<!\\)\(([^():]*?):\s*(" + _NUM + r")\s*\)")
|
||||
_TIMED = re.compile(
|
||||
r"(?<!\\)\[([^\[\]]*?)(?::\s*(" + _NUM + r"))?\s*@\s*(\d+(?:\.\d+)?)\s*-\s*(\d+(?:\.\d+)?)\s*\]")
|
||||
_BRACKET_WEIGHTED = re.compile(r"(?<!\\)\[([^\[\]@]*?):\s*(" + _NUM + r")\s*\]")
|
||||
_BLENDED = re.compile(r"(?<!\\)\[([^\[\]|]+)\|([^\[\]|]+?)(?::\s*(" + _NUM + r"))?\]")
|
||||
_MARKUP = re.compile("|".join(p.pattern for p in (_TIMED, _BLENDED, _WEIGHTED, _BRACKET_WEIGHTED)))
|
||||
|
||||
BLEND_STRENGTH_DEFAULT = 0.5
|
||||
# Below this a weight means "remove": log(0) is -inf, which is NaN the moment a
|
||||
# query attends to nothing else.
|
||||
MIN_WEIGHT = 1e-3
|
||||
MASKED_BIAS = -30.0
|
||||
# Past this a phrase wins every query regardless of content: the prompt reads as
|
||||
# collapsing onto one word.
|
||||
MAX_ABS_BIAS = 6.0
|
||||
# Overlapping spans (a phrase and one of its own words) add, bounded at this
|
||||
# multiple of the strongest single span -- a lone (word:2.0) still means 2.0.
|
||||
OVERLAP_HEADROOM = 1.5
|
||||
|
||||
|
||||
def parse(text):
|
||||
"""-> {"clean", "weighted": [(c0, c1, w)], "timed": [(c0, c1, w, t0, t1)],
|
||||
"blended": [(c0, c1, alt, strength)]}, char spans into `clean`."""
|
||||
out = {"clean": text or "", "weighted": [], "timed": [], "blended": []}
|
||||
if not text or ("(" not in text and "[" not in text):
|
||||
return out
|
||||
pieces, pos = [], 0
|
||||
for m in _MARKUP.finditer(text):
|
||||
pieces.append(text[pos:m.start()])
|
||||
start = sum(len(p) for p in pieces)
|
||||
if m.group(1) is not None:
|
||||
phrase, t0, t1 = m.group(1), float(m.group(3)), float(m.group(4))
|
||||
if t1 > t0 and phrase.strip():
|
||||
out["timed"].append((start, start + len(phrase),
|
||||
float(m.group(2)) if m.group(2) else 1.0, t0, t1))
|
||||
elif m.group(5) is not None:
|
||||
phrase, alt = m.group(5), m.group(6)
|
||||
strength = float(m.group(7)) if m.group(7) is not None else BLEND_STRENGTH_DEFAULT
|
||||
if phrase.strip() and alt.strip():
|
||||
out["blended"].append((start, start + len(phrase), alt.strip(), strength))
|
||||
else:
|
||||
phrase = m.group(8) if m.group(8) is not None else m.group(10)
|
||||
weight = m.group(9) if m.group(9) is not None else m.group(11)
|
||||
out["weighted"].append((start, start + len(phrase), float(weight)))
|
||||
pieces.append(phrase)
|
||||
pos = m.end()
|
||||
pieces.append(text[pos:])
|
||||
out["clean"] = "".join(pieces)
|
||||
return out
|
||||
|
||||
|
||||
def token_spans(offsets, char_spans):
|
||||
"""Char spans -> (tok0, tok1, *rest) using a tokenizer's offset mapping."""
|
||||
out = []
|
||||
for c0, c1, *rest in char_spans:
|
||||
toks = [i for i, (a, b) in enumerate(offsets) if a < c1 and b > c0]
|
||||
if toks:
|
||||
out.append((toks[0], toks[-1] + 1, *rest))
|
||||
return out
|
||||
|
||||
|
||||
def bias_value(weight):
|
||||
if weight <= MIN_WEIGHT:
|
||||
return MASKED_BIAS
|
||||
return max(-MAX_ABS_BIAS, min(MAX_ABS_BIAS, math.log(weight)))
|
||||
|
||||
|
||||
def key_bias(spans, base, seq_len, device, dtype):
|
||||
"""[1, 1, 1, seq_len] additive bias for (tok0, tok1, w) spans, prompt at `base`."""
|
||||
import torch
|
||||
bias, strongest = None, 0.0
|
||||
for t0, t1, weight in spans:
|
||||
b = bias_value(weight)
|
||||
if b == 0.0 or t1 <= t0:
|
||||
continue
|
||||
if bias is None:
|
||||
bias = torch.zeros(1, 1, 1, seq_len, device=device, dtype=dtype)
|
||||
bias[..., base + t0:base + t1] += b
|
||||
strongest = max(strongest, abs(b))
|
||||
if bias is not None:
|
||||
limit = min(MAX_ABS_BIAS, strongest * OVERLAP_HEADROOM)
|
||||
bias.clamp_(min=-limit, max=limit)
|
||||
return bias
|
||||
|
||||
|
||||
def plan(weighted, timed, base, seq_len, video_start, device, dtype):
|
||||
"""The bias to add, as one tensor, or [(q0, q1, bias)] query chunks when a
|
||||
phrase is timed. Softmax is per query row, so attending the video rows in
|
||||
chunks, each with its own key bias, is exact -- it only adds kernel launches.
|
||||
|
||||
`timed` = [(tok0, tok1, w, r0, r1)] with r0/r1 absolute rows of the sequence.
|
||||
"""
|
||||
import torch
|
||||
base_bias = key_bias(weighted, base, seq_len, device, dtype)
|
||||
if not timed:
|
||||
return base_bias
|
||||
edges = sorted({0, video_start, seq_len, *(r for *_x, r0, r1 in timed for r in (r0, r1))})
|
||||
chunks = []
|
||||
for a, b in zip(edges, edges[1:]):
|
||||
if b <= a:
|
||||
continue
|
||||
bias = (base_bias.clone() if base_bias is not None
|
||||
else torch.zeros(1, 1, 1, seq_len, device=device, dtype=dtype))
|
||||
if a >= video_start:
|
||||
for t0, t1, w, r0, r1 in timed:
|
||||
if r0 <= a and b <= r1:
|
||||
bias[..., base + t0:base + t1] += bias_value(w)
|
||||
else:
|
||||
bias[..., base + t0:base + t1] = MASKED_BIAS
|
||||
chunks.append((a, b, bias))
|
||||
return chunks
|
||||
@@ -0,0 +1,189 @@
|
||||
"""The two markup nodes. See the module docstring for the shape."""
|
||||
|
||||
import torch
|
||||
import torch.nn.functional as F
|
||||
from comfy_api.latest import io
|
||||
|
||||
from ..._core import dit_hooks, log, patching, registry as registry_mod
|
||||
from . import markup as mk
|
||||
|
||||
MARKUP = io.Custom("FUNPACK_MARKUP")
|
||||
KEY = "funpack_markup.prompt"
|
||||
|
||||
|
||||
def _ask(capability, *args):
|
||||
"""The first model module with an answer, or None."""
|
||||
for _spec, provider in registry_mod.current().providers(capability):
|
||||
got = provider(*args)
|
||||
if got is not None:
|
||||
return got
|
||||
return None
|
||||
|
||||
|
||||
def _say(message):
|
||||
log.once(f"prompt_markup:{message}", log.ALERT, "FunPack Prompt Markup", message)
|
||||
|
||||
|
||||
class FunPackPromptMarkup(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls) -> io.Schema:
|
||||
return io.Schema(
|
||||
node_id="FunPackPromptMarkup",
|
||||
display_name="FunPack Prompt Markup",
|
||||
category="FunPack/Conditioning",
|
||||
description="Take (word:1.5), [phrase@2-4] and [a|b] out of a prompt, so the "
|
||||
"encoder sees clean text and the markup can be applied properly.",
|
||||
inputs=[io.String.Input("text", multiline=True, default="")],
|
||||
outputs=[io.String.Output(display_name="clean"),
|
||||
MARKUP.Output(display_name="markup"),
|
||||
io.String.Output(display_name="status")],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, text: str) -> io.NodeOutput:
|
||||
parsed = mk.parse(text or "")
|
||||
n = {k: len(parsed[k]) for k in ("weighted", "timed", "blended")}
|
||||
status = (", ".join(f"{v} {k}" for k, v in n.items() if v) or "no markup")
|
||||
return io.NodeOutput(parsed["clean"], parsed, status)
|
||||
|
||||
|
||||
def _blend(cond, rows, clip, tokenizer, clean, offsets, blended):
|
||||
"""(1-s)*A + s*B over the prompt rows, each span measured against the
|
||||
ORIGINAL A so several blends add instead of compounding."""
|
||||
start, end = rows
|
||||
a_rows = cond[:, start:end, :]
|
||||
delta, done = torch.zeros_like(a_rows), 0
|
||||
for c0, c1, alt, strength in blended:
|
||||
if not mk.token_spans(offsets, [(c0, c1)]):
|
||||
continue
|
||||
alt_text = clean[:c0] + alt + clean[c1:]
|
||||
encoded = clip.encode_from_tokens_scheduled(clip.tokenize(alt_text))
|
||||
b_cond, b_meta = encoded[0][0], encoded[0][1]
|
||||
n_b = len(tokenizer(alt_text, add_special_tokens=False,
|
||||
return_offsets_mapping=True)["offset_mapping"])
|
||||
b_rows = _ask("prompt_rows", b_meta, int(b_cond.shape[1]), n_b)
|
||||
if b_rows is None:
|
||||
continue
|
||||
b = b_cond[:, b_rows[0]:b_rows[1], :].to(a_rows)
|
||||
if b.shape[1] != a_rows.shape[1]:
|
||||
# Linear along the token axis: most rows already correspond (only
|
||||
# the swapped phrase differs); a smoothing, not a word alignment.
|
||||
b = F.interpolate(b.transpose(1, 2), size=a_rows.shape[1], mode="linear",
|
||||
align_corners=True).transpose(1, 2)
|
||||
delta += float(strength) * (b - a_rows)
|
||||
done += 1
|
||||
if not done:
|
||||
return cond, 0
|
||||
out = cond.clone()
|
||||
out[:, start:end, :] += delta
|
||||
return out, done
|
||||
|
||||
|
||||
class FunPackApplyPromptMarkup(io.ComfyNode):
|
||||
@classmethod
|
||||
def define_schema(cls) -> io.Schema:
|
||||
return io.Schema(
|
||||
node_id="FunPackApplyPromptMarkup",
|
||||
display_name="FunPack Apply Prompt Markup",
|
||||
category="FunPack/Conditioning",
|
||||
description="Apply the markup FunPack Prompt Markup took out: weights and "
|
||||
"timed phrases on the model, blends on the conditioning.",
|
||||
inputs=[io.Model.Input("model"), io.Clip.Input("clip"),
|
||||
io.Conditioning.Input("positive"), io.Latent.Input("latent"),
|
||||
MARKUP.Input("markup")],
|
||||
outputs=[io.Model.Output(display_name="model"),
|
||||
io.Conditioning.Output(display_name="positive"),
|
||||
io.String.Output(display_name="status")],
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def execute(cls, model, clip, positive, latent, markup) -> io.NodeOutput:
|
||||
patched = model.clone()
|
||||
patching.strip(patched, "funpack_markup.")
|
||||
weighted, timed, blended = (markup or {}).get("weighted"), (markup or {}).get("timed"), \
|
||||
(markup or {}).get("blended")
|
||||
if not (weighted or timed or blended):
|
||||
return io.NodeOutput(patched, positive, "no markup")
|
||||
|
||||
def refuse(why):
|
||||
_say(f"markup NOT applied: {why}. The prompt still went in, without it.")
|
||||
return io.NodeOutput(patched, positive, f"not applied: {why}")
|
||||
|
||||
tokenizer = _ask("text_tokenizer", clip)
|
||||
if tokenizer is None:
|
||||
return refuse("this text encoder is not one a model module can place words in")
|
||||
clean = markup["clean"]
|
||||
offsets = list(tokenizer(clean, add_special_tokens=False,
|
||||
return_offsets_mapping=True)["offset_mapping"])
|
||||
cond, meta = positive[0][0], positive[0][1]
|
||||
cond_len = int(cond.shape[1])
|
||||
rows = _ask("prompt_rows", meta, cond_len, len(offsets))
|
||||
if rows is None:
|
||||
return refuse("the encoded prompt does not match this text -- was the clean "
|
||||
"text from Prompt Markup what got encoded?")
|
||||
base = rows[0]
|
||||
said = []
|
||||
|
||||
if blended:
|
||||
new_cond, n = _blend(cond, rows, clip, tokenizer, clean, offsets, blended)
|
||||
if n:
|
||||
positive = [[new_cond, meta], *positive[1:]]
|
||||
said.append(f"{n} blend(s)")
|
||||
|
||||
w_spans = mk.token_spans(offsets, weighted or [])
|
||||
t_spans = []
|
||||
for tok0, tok1, w, t0, t1 in mk.token_spans(offsets, timed or []):
|
||||
window = _ask("video_time_rows", latent["samples"], t0, t1)
|
||||
if window is None:
|
||||
_say(f"a timed phrase's window {t0:g}-{t1:g}s misses this clip; ignored")
|
||||
continue
|
||||
t_spans.append((tok0, tok1, w, window))
|
||||
if w_spans or t_spans:
|
||||
cls._install(patched, w_spans, t_spans, base, cond_len)
|
||||
if w_spans:
|
||||
said.append(f"{len(w_spans)} weight(s)")
|
||||
if t_spans:
|
||||
said.append(f"{len(t_spans)} timed phrase(s)")
|
||||
return io.NodeOutput(patched, positive, ", ".join(said) or "nothing placeable")
|
||||
|
||||
@staticmethod
|
||||
def _install(patcher, w_spans, t_spans, base, cond_len):
|
||||
cache = {}
|
||||
|
||||
def build(seq_len, device, dtype):
|
||||
timed, video_start = [], seq_len
|
||||
for tok0, tok1, w, (n_video, r0, r1) in t_spans:
|
||||
video_start = seq_len - n_video
|
||||
if video_start < cond_len:
|
||||
_say("this sequence is too short to hold the video the timed "
|
||||
"phrases were placed on; they are off for this call")
|
||||
return mk.plan(w_spans, [], base, seq_len, seq_len, device, dtype)
|
||||
timed.append((tok0, tok1, w, video_start + r0, video_start + r1))
|
||||
return mk.plan(w_spans, timed, base, seq_len, video_start, device, dtype)
|
||||
|
||||
def override(func, q, k, v, heads, *args, mask=None, **kwargs):
|
||||
def run(m, qq=q):
|
||||
return func(qq, k, v, heads, *args, mask=m, **kwargs)
|
||||
|
||||
# Only the packed self-attention: the prompt plus the streams after it.
|
||||
# A sequence exactly as long as the prompt is the text-only token
|
||||
# refiner, which must never see a video window's mask.
|
||||
if k.ndim != 4 or q.shape[2] != k.shape[2] or k.shape[2] <= cond_len:
|
||||
return run(mask)
|
||||
try:
|
||||
key = (int(k.shape[2]), k.device, k.dtype)
|
||||
if key not in cache:
|
||||
cache[key] = build(*key)
|
||||
plan = cache[key]
|
||||
if plan is None:
|
||||
return run(mask)
|
||||
if not isinstance(plan, list):
|
||||
return run(plan if mask is None else mask + plan)
|
||||
outs = [run(b if mask is None else mask + b, q[:, :, a:z].contiguous())
|
||||
for a, z, b in plan]
|
||||
return torch.cat(outs, dim=2 if outs[0].ndim == 4 else 1)
|
||||
except Exception as exc: # noqa: BLE001 -- markup never costs the step
|
||||
_say(f"markup failed inside attention and was dropped ({exc})")
|
||||
return run(mask)
|
||||
|
||||
dit_hooks.add_attention_override(patcher, KEY, override)
|
||||
@@ -0,0 +1,137 @@
|
||||
"""Prompt markup, with H3's REAL tokenizer (offline) and the tiny real H3 model.
|
||||
|
||||
The encoder is a stand-in -- one fixed vector per token id -- because Qwen3-VL-32B
|
||||
is not a unit-test dependency; placement depends on the tokenizer and the tags,
|
||||
which are real here.
|
||||
"""
|
||||
|
||||
import math
|
||||
|
||||
import pytest
|
||||
import torch
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _needs_comfy(comfyui):
|
||||
"""Imports comfy."""
|
||||
|
||||
|
||||
class FakeH3Clip:
|
||||
def __init__(self):
|
||||
from comfy.text_encoders.minimax import MiniMaxH3Tokenizer
|
||||
self.tokenizer = MiniMaxH3Tokenizer(embedding_directory=None, tokenizer_data={})
|
||||
|
||||
def tokenize(self, text):
|
||||
return self.tokenizer.tokenize_with_weights(text)
|
||||
|
||||
def encode_from_tokens_scheduled(self, tokens):
|
||||
ids = [t[0] for t in tokens["qwen3vl_32b"][0]]
|
||||
rows = torch.stack([torch.randn(48, generator=torch.Generator().manual_seed(int(i)))
|
||||
for i in ids]).unsqueeze(0)
|
||||
return [[rows, {"minimax_token_tags": torch.ones(len(ids), dtype=torch.long)}]]
|
||||
|
||||
|
||||
def _encode(clip, text):
|
||||
return clip.encode_from_tokens_scheduled(clip.tokenize(text))
|
||||
|
||||
|
||||
def _apply(tiny, raw):
|
||||
from modules.conditioning.prompt_markup.nodes import (FunPackApplyPromptMarkup,
|
||||
FunPackPromptMarkup)
|
||||
clean, markup, _ = FunPackPromptMarkup.execute(raw).result
|
||||
clip = FakeH3Clip()
|
||||
positive = _encode(clip, clean)
|
||||
tiny.context = positive[0][0]
|
||||
model, pos, status = FunPackApplyPromptMarkup.execute(
|
||||
tiny.patcher, clip, positive, {"samples": tiny.video}, markup).result
|
||||
return model, pos, status, positive
|
||||
|
||||
|
||||
def _spy_masks(tiny, model):
|
||||
"""Every (q rows, mask) the backend receives, via an override UNDER ours."""
|
||||
from core import dit_hooks
|
||||
seen = []
|
||||
|
||||
def spy(func, q, k, v, *a, mask=None, **kw):
|
||||
seen.append((q.shape[2], mask))
|
||||
return func(q, k, v, *a, mask=mask, **kw)
|
||||
|
||||
under = tiny.patcher.clone()
|
||||
dit_hooks.add_attention_override(under, "spy", spy)
|
||||
ours = model.model_options["transformer_options"]["optimized_attention_override"]
|
||||
# Put ours on top of the spy.
|
||||
under.model_options["transformer_options"]["optimized_attention_override"] = (
|
||||
lambda func, *a, **kw: ours(lambda *x, **y: under_spy(func, *x, **y), *a, **kw))
|
||||
under_spy = lambda func, *a, **kw: spy(func, *a, **kw) # noqa: E731
|
||||
tiny.run(under)
|
||||
return seen
|
||||
|
||||
|
||||
def test_the_node_strips_markup_and_reports_it():
|
||||
from modules.conditioning.prompt_markup.nodes import FunPackPromptMarkup
|
||||
clean, markup, status = FunPackPromptMarkup.execute("a (red:1.5) car [sits|runs]").result
|
||||
assert clean == "a red car sits"
|
||||
assert status == "1 weighted, 1 blended"
|
||||
|
||||
|
||||
def test_a_weight_biases_exactly_that_words_keys(tiny_h3):
|
||||
model, _p, status, _pos = _apply(tiny_h3, "a (red:1.5) car drives")
|
||||
assert status == "1 weight(s)"
|
||||
masks = [m for n, m in _spy_masks(tiny_h3, model) if m is not None]
|
||||
assert masks, "the bias never reached attention"
|
||||
row = masks[0].reshape(-1)
|
||||
# tokens: "a" " red" " car" " drives" -> "red" is token 1, prompt at row 0
|
||||
assert row[1].item() == pytest.approx(math.log(1.5), rel=1e-4)
|
||||
assert torch.count_nonzero(row).item() == 1
|
||||
|
||||
|
||||
def test_it_changes_the_output_and_weight_one_does_not(tiny_h3):
|
||||
base_model, _p, _s, pos = _apply(tiny_h3, "a red car drives")
|
||||
base = tiny_h3.run(base_model)
|
||||
weighted, *_ = _apply(tiny_h3, "a (red:3) car drives")
|
||||
assert not torch.allclose(tiny_h3.run(weighted)[0], base[0])
|
||||
|
||||
|
||||
def test_a_timed_phrase_is_masked_from_video_outside_its_window(tiny_h3):
|
||||
# Latent frame 0 = pixel frame 0, frame 1 = pixel frames 1-4 at 24 fps;
|
||||
# 0.05-0.2 s touches only latent frame 1 = video rows 4..8 of 8.
|
||||
model, _p, status, _pos = _apply(tiny_h3, "a red [car@0.05-0.2] drives")
|
||||
assert status == "1 timed phrase(s)"
|
||||
calls = _spy_masks(tiny_h3, model)
|
||||
# The text-only token refiner runs first on exactly the prompt: untouched.
|
||||
n, m = calls[0]
|
||||
assert n == int(tiny_h3.context.shape[1]) and m is None, "the refiner saw the window"
|
||||
per_block = [(n, m) for n, m in calls if m is not None][:3] # block 0's chunks
|
||||
assert [n for n, _m in per_block] == [10, 4, 4] # text+audio | video out | in
|
||||
assert sum(n for n, _m in per_block) == tiny_h3_seq(tiny_h3)
|
||||
key = [m.reshape(-1)[2].item() for _n, m in per_block] # "car" is token 2
|
||||
assert key[0] == 0.0, "text/audio rows lost the phrase"
|
||||
assert key[1] == pytest.approx(-30.0), "video outside the window still sees it"
|
||||
assert key[2] == 0.0, "video inside the window lost it"
|
||||
|
||||
|
||||
def tiny_h3_seq(tiny):
|
||||
video_rows = 2 * 2 * 2
|
||||
return int(tiny.context.shape[1]) + video_rows + 2 * 3
|
||||
|
||||
|
||||
def test_a_blend_moves_only_the_prompt_rows_and_zero_strength_moves_nothing(tiny_h3):
|
||||
_m, pos, status, original = _apply(tiny_h3, "a red car [drives|flies]")
|
||||
assert status == "1 blend(s)"
|
||||
assert not torch.allclose(pos[0][0], original[0][0])
|
||||
_m, pos0, _s, original0 = _apply(tiny_h3, "a red car [drives|flies:0]")
|
||||
assert torch.allclose(pos0[0][0], original0[0][0])
|
||||
|
||||
|
||||
def test_an_encoder_no_model_module_knows_is_refused_out_loud(tiny_h3):
|
||||
from modules.conditioning.prompt_markup.nodes import (FunPackApplyPromptMarkup,
|
||||
FunPackPromptMarkup)
|
||||
_clean, markup, _ = FunPackPromptMarkup.execute("a (red:2) car").result
|
||||
|
||||
class OtherClip:
|
||||
tokenizer = object()
|
||||
|
||||
positive = [[torch.zeros(1, 4, 48), {}]]
|
||||
model, pos, status = FunPackApplyPromptMarkup.execute(
|
||||
tiny_h3.patcher, OtherClip(), positive, {"samples": tiny_h3.video}, markup).result
|
||||
assert status.startswith("not applied") and pos is positive
|
||||
@@ -31,6 +31,7 @@ from .pipeline import presets as _pipeline_presets
|
||||
from .anchor import anchor_pin as _anchor_pin
|
||||
from .anchor import rescale_pins as _rescale_pins
|
||||
from .latent_upscaler import load_upscaler as _load_upscaler
|
||||
from . import prompt_layout as _prompt_layout
|
||||
from .rows import target_rows as _target_rows
|
||||
|
||||
has_block = _traits.has_block
|
||||
@@ -187,4 +188,7 @@ TRAITS = traits
|
||||
PROVIDES = {"empty_latent": empty_latent, "decode": decode, "detect": detect,
|
||||
"pipeline_presets": _pipeline_presets, "probe_traits": probe_traits,
|
||||
"target_rows": _target_rows, "anchor_pin": _anchor_pin,
|
||||
"latent_upscaler": _load_upscaler, "rescale_conditioning": _rescale_pins}
|
||||
"latent_upscaler": _load_upscaler, "rescale_conditioning": _rescale_pins,
|
||||
"text_tokenizer": _prompt_layout.text_tokenizer,
|
||||
"prompt_rows": _prompt_layout.prompt_rows,
|
||||
"video_time_rows": _prompt_layout.video_time_rows}
|
||||
|
||||
@@ -79,12 +79,14 @@ def h3_reference_to_video():
|
||||
"invert_crop": "disabled",
|
||||
"bboxes": ""}},
|
||||
|
||||
{"id": "markup", "group": "Preparation", "node": "FunPackPromptMarkup",
|
||||
"roles": [{"at": "generation.prompt", "input": "text", "label": "Prompt"}],
|
||||
"inputs": {"text": ""}},
|
||||
{"id": "r2v", "group": "Preparation", "node": "MiniMaxH3ReferenceToVideo",
|
||||
"roles": [{"at": "generation.prompt", "input": "prompt", "label": "Prompt"},
|
||||
{"at": "project.video", "input": "length", "label": "Length"}],
|
||||
"roles": [{"at": "project.video", "input": "length", "label": "Length"}],
|
||||
"inputs": {
|
||||
"clip": ["clip", 0], "vae": ["vae", 0], "audio_vae": ["audio_vae", 0],
|
||||
"prompt": "", "width": ["image_transform", 4], "height": ["image_transform", 5],
|
||||
"prompt": ["markup", 0], "width": ["image_transform", 4], "height": ["image_transform", 5],
|
||||
"length": 124, "ref_image_size": "match"}},
|
||||
{"id": "negative", "group": "Preparation", "node": "CLIPTextEncode",
|
||||
# H3 samples at CFG 1, where a negative prompt does nothing -- wired
|
||||
@@ -95,13 +97,17 @@ def h3_reference_to_video():
|
||||
{"id": "modifiers", "group": "Preparation", "node": "FunPackLoadModifiers", "inputs": {
|
||||
"model": ["shift", 0], "settings": ["settings", 0]}},
|
||||
|
||||
{"id": "markup_apply", "group": "Preparation", "node": "FunPackApplyPromptMarkup",
|
||||
"inputs": {"model": ["modifiers", 0], "clip": ["clip", 0], "positive": ["r2v", 0],
|
||||
"latent": ["r2v", 1], "markup": ["markup", 1]}},
|
||||
|
||||
{"id": "sampler", "group": "Sampling", "node": "FunPackSampler",
|
||||
"roles": [{"at": "generation.sampling", "input": "steps", "label": "Steps"},
|
||||
{"at": "generation.sampling", "input": "sampler_name", "label": "Sampler"},
|
||||
{"at": "generation.sampling", "input": "scheduler", "label": "Scheduler"}],
|
||||
"inputs": {
|
||||
"model": ["modifiers", 0], "positive": ["r2v", 0], "negative": ["negative", 0],
|
||||
"latent": ["r2v", 1], "settings": ["settings", 0],
|
||||
"model": ["markup_apply", 0], "positive": ["markup_apply", 1],
|
||||
"negative": ["negative", 0], "latent": ["r2v", 1], "settings": ["settings", 0],
|
||||
"seed": 0, "steps": 20, "cfg": 1.0,
|
||||
"sampler_name": "euler", "scheduler": "normal", "denoise": 1.0}},
|
||||
|
||||
@@ -158,12 +164,12 @@ def h3_reference_to_video_second_pass():
|
||||
{"id": "resample", "group": "Second pass", "node": "FunPackLatentResample",
|
||||
"inputs": {"latent": ["sampler", 0], "upscaler": ["upscaler", 0],
|
||||
"operation": "sharpen", "scale": 2.0,
|
||||
"positive": ["r2v", 0], "negative": ["negative", 0]}},
|
||||
"positive": ["markup_apply", 1], "negative": ["negative", 0]}},
|
||||
{"id": "pass2_sigmas", "group": "Second pass", "node": "ManualSigmas",
|
||||
"inputs": {"sigmas": "0.4, 0.3, 0.2, 0.1, 0.0"}},
|
||||
{"id": "pass2", "group": "Second pass", "node": "FunPackSampler",
|
||||
"inputs": {
|
||||
"model": ["modifiers", 0], "positive": ["resample", 1],
|
||||
"model": ["markup_apply", 0], "positive": ["resample", 1],
|
||||
"negative": ["resample", 2], "latent": ["resample", 0],
|
||||
"settings": ["settings", 0], "sigmas": ["pass2_sigmas", 0],
|
||||
"seed": 0, "steps": 4, "cfg": 1.0,
|
||||
|
||||
@@ -0,0 +1,86 @@
|
||||
"""Where H3 puts the prompt, and how its time axis maps to rows.
|
||||
|
||||
Three capabilities, so prompt markup never learns H3's layout:
|
||||
|
||||
* `text_tokenizer(clip)` -- the raw HF tokenizer behind an H3 CLIP, for exact
|
||||
character -> token offsets. Tokenizing is free; re-encoding with Qwen3-VL-32B
|
||||
to find a phrase would cost a 32B forward pass per phrase.
|
||||
* `prompt_rows(meta, cond_len, prompt_tokens)` -- (start, end) of the prompt
|
||||
inside a conditioning. The prompt is the LAST thing H3's tokenizer appends
|
||||
(reference labels and vision blocks come first), and `minimax_token_tags`
|
||||
marks text 1, vision 0, so it is the tail of the final run of 1s. Measured
|
||||
from the END so labels in front of it (an audio reference's "<Audio n>: ",
|
||||
which has no vision block) do not shift it.
|
||||
* `video_time_rows(samples, t0, t1)` -- a window in seconds as rows of the
|
||||
target video, which is the last segment of the packed sequence.
|
||||
|
||||
The conditioning block leads the packed sequence, so a row inside it is also a
|
||||
row of the sequence the attention runs over.
|
||||
"""
|
||||
|
||||
FPS = 24
|
||||
# Latent frame k spans FRAME_PER_TOKEN[k % 5] pixel frames (H3's 5k+2 grid).
|
||||
FRAME_PER_TOKEN = (1, 4, 4, 4, 4)
|
||||
|
||||
|
||||
def text_tokenizer(clip):
|
||||
try:
|
||||
tok = getattr(getattr(getattr(clip, "tokenizer", None), "qwen3vl_32b", None),
|
||||
"tokenizer", None)
|
||||
if tok is None:
|
||||
return None
|
||||
tok("probe", add_special_tokens=False, return_offsets_mapping=True)
|
||||
return tok
|
||||
except Exception: # noqa: BLE001 -- not H3's, or no offsets
|
||||
return None
|
||||
|
||||
|
||||
def _tags(meta):
|
||||
tags = (meta or {}).get("minimax_token_tags")
|
||||
if tags is None:
|
||||
return None
|
||||
try:
|
||||
return [int(t) for t in tags.reshape(-1).tolist()]
|
||||
except AttributeError:
|
||||
return [int(t) for t in tags] if isinstance(tags, (list, tuple)) else None
|
||||
|
||||
|
||||
def prompt_rows(meta, cond_len, prompt_tokens):
|
||||
"""(start, end) of the prompt in the conditioning, or None if it cannot be
|
||||
proven. Without tags the prompt is the final `prompt_tokens` rows."""
|
||||
if prompt_tokens <= 0 or cond_len <= 0:
|
||||
return None
|
||||
tags = _tags(meta)
|
||||
if tags is None:
|
||||
start = cond_len - prompt_tokens
|
||||
return (start, cond_len) if start >= 0 else None
|
||||
end = min(len(tags), cond_len)
|
||||
while end > 0 and tags[end - 1] != 1:
|
||||
end -= 1
|
||||
start = end
|
||||
while start > 0 and tags[start - 1] == 1:
|
||||
start -= 1
|
||||
if end - start < prompt_tokens:
|
||||
return None # a different string was encoded
|
||||
return end - prompt_tokens, end
|
||||
|
||||
|
||||
def video_time_rows(samples, t0, t1):
|
||||
"""(video_rows, row_start, row_end) for a window in seconds, rows counted
|
||||
from the start of the target video segment. None if it misses the clip."""
|
||||
video = samples.unbind()[0] if getattr(samples, "is_nested", False) else samples
|
||||
if getattr(video, "ndim", 0) != 5:
|
||||
return None
|
||||
latent_t = int(video.shape[2])
|
||||
frame_rows = ((int(video.shape[3]) + 1) // 2) * ((int(video.shape[4]) + 1) // 2)
|
||||
cum = [0]
|
||||
for k in range(latent_t):
|
||||
cum.append(cum[-1] + FRAME_PER_TOKEN[k % 5])
|
||||
f0, f1 = int(round(t0 * FPS)), int(round(t1 * FPS))
|
||||
if f1 <= f0 or f0 >= cum[-1]:
|
||||
return None
|
||||
k0 = max(k for k in range(latent_t) if cum[k] <= f0)
|
||||
k1 = min(latent_t, max(k for k in range(latent_t) if cum[k] < f1) + 1)
|
||||
if k1 <= k0:
|
||||
return None
|
||||
return latent_t * frame_rows, k0 * frame_rows, k1 * frame_rows
|
||||
Reference in New Issue
Block a user