From 3676a9fdb02836b3ecf08bdf3bd511142ab8b826 Mon Sep 17 00:00:00 2001 From: digital-garbage <133010426+digital-garbage@users.noreply.github.com> Date: Tue, 25 Aug 2026 12:53:29 +0300 Subject: [PATCH] maint: remove the v2.4-era Lucky and Into the Void machinery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 32 methods, 1317 lines: the co-occurrence memory, token/pair banks, phrase placements, embedding canvas and prompt composer. Every entry point was called only from the v1 refine() removed in the previous commit, or from another method in the cluster — verified individually after that removal, all clean. Lucky and Void are one commit because they share the token banks and call into each other; splitting them would leave a commit referencing a method that no longer exists. Into the Void's toggle only ever existed on the unregistered v1 node, so no UI could reach it. The Lucky TOGGLE stays: the Studio's Lucky is a separate 56-line composer over phrase_memory (_v2_compose_lucky_prompt), unrelated to this code and still live. The v2 state schema declares none of these keys, so refinement-key training never touched them. Tests: 1377 passed, same as baseline. Co-Authored-By: Claude Opus 5 --- conditioning.py | 1303 +---------------------------------------------- 1 file changed, 9 insertions(+), 1294 deletions(-) diff --git a/conditioning.py b/conditioning.py index b6572e06..b4806f3c 100644 --- a/conditioning.py +++ b/conditioning.py @@ -1584,27 +1584,8 @@ class FunPackVideoRefiner: key=lambda group: (group[0], group[1], group[2].lower()), ) - def _void_empty_status(self, enabled=False): - return f"Void: {'on' if enabled else 'off'} | idle | tokens: none" - def _void_token_embedding(self, conditioning: torch.Tensor, start: int, end: int): - if not isinstance(conditioning, torch.Tensor) or conditioning.dim() <= 1 or end <= start: - return None - if conditioning.dim() == 3: - return conditioning[:, start:end, :].detach().mean(dim=(0, 1)) - return conditioning[start:end, :].detach().mean(dim=0) - def _void_bank_score(self, item): - likes = float(item.get("liked_count", 0)) - neutral = float(item.get("neutral_count", 0)) - awful = float(item.get("awful_count", 0)) - wanted = float(item.get("wanted_count", 0)) - missing = float(item.get("missing_count", 0)) - concept_missing = float(item.get("missing_concept_count", 0)) - detail_missing = float(item.get("missing_detail_count", 0)) - quality_missing = float(item.get("missing_quality_count", 0)) - missing_pressure = missing * 0.22 + concept_missing * 0.14 + detail_missing * 0.10 + quality_missing * 0.08 - return float(item.get("score", 0.0)) + likes * 0.40 + neutral * 0.08 + wanted * 0.30 + missing_pressure - awful * 0.95 def _missing_axes_for_rating_key(self, rating_key): for profile in RATING_PROFILES.values(): @@ -1647,136 +1628,11 @@ class FunPackVideoRefiner: item["last_missing_iter"] = int(iter_num) return item - def _ensure_void_token_bank(self, global_adaptive): - bank = global_adaptive.get("void_token_bank") - if not isinstance(bank, dict): - bank = {} - if bank and global_adaptive.get("void_token_bank_validated") is True: - return bank - cleaned = {} - for token, item in bank.items(): - if not isinstance(token, str) or not isinstance(item, dict): - continue - vector = item.get("embedding") - if not isinstance(vector, dict): - continue - try: - shape = vector.get("shape", []) - if len(shape) != 1 or int(shape[0]) <= 0: - continue - item["score"] = round(max(-2.0, min(8.0, float(item.get("score", 0.0)))), 4) - item["liked_count"] = max(0, int(item.get("liked_count", 0))) - item["neutral_count"] = max(0, int(item.get("neutral_count", 0))) - item["awful_count"] = max(0, int(item.get("awful_count", 0))) - item["wanted_count"] = max(0, int(item.get("wanted_count", 0))) - item["missing_count"] = max(0, int(item.get("missing_count", 0))) - item["missing_concept_count"] = max(0, int(item.get("missing_concept_count", 0))) - item["missing_detail_count"] = max(0, int(item.get("missing_detail_count", 0))) - item["missing_quality_count"] = max(0, int(item.get("missing_quality_count", 0))) - item["last_seen_iter"] = max(0, int(item.get("last_seen_iter", 0))) - item["last_missing_iter"] = max(0, int(item.get("last_missing_iter", 0))) - item["sample_count"] = max(0, int(item.get("sample_count", 0))) - cleaned[token] = item - except (TypeError, ValueError): - continue - global_adaptive["void_token_bank"] = cleaned - global_adaptive["void_token_bank_validated"] = True - return cleaned - def _void_pair_key(self, left_token, right_token): - return f"{str(left_token).strip().lower()}\t{str(right_token).strip().lower()}" - def _void_pair_score(self, item): - likes = float(item.get("liked_count", 0)) - neutral = float(item.get("neutral_count", 0)) - awful = float(item.get("awful_count", 0)) - wanted = float(item.get("wanted_count", 0)) - missing = float(item.get("missing_count", 0)) - return float(item.get("score", 0.0)) + likes * 0.35 + neutral * 0.05 + wanted * 0.22 + missing * 0.16 - awful * 0.85 - def _ensure_void_pair_bank(self, global_adaptive): - bank = global_adaptive.get("void_token_pairs") - if not isinstance(bank, dict): - bank = {} - if bank and global_adaptive.get("void_token_pairs_validated") is True: - return bank - cleaned = {} - for key, item in bank.items(): - if not isinstance(key, str) or "\t" not in key or not isinstance(item, dict): - continue - try: - left, right = key.split("\t", 1) - if not self._is_valuable_token(left) or not self._is_valuable_token(right): - continue - item["score"] = round(max(-3.0, min(6.0, float(item.get("score", 0.0)))), 4) - item["liked_count"] = max(0, int(item.get("liked_count", 0))) - item["neutral_count"] = max(0, int(item.get("neutral_count", 0))) - item["awful_count"] = max(0, int(item.get("awful_count", 0))) - item["wanted_count"] = max(0, int(item.get("wanted_count", 0))) - item["missing_count"] = max(0, int(item.get("missing_count", 0))) - item["missing_concept_count"] = max(0, int(item.get("missing_concept_count", 0))) - item["missing_detail_count"] = max(0, int(item.get("missing_detail_count", 0))) - item["missing_quality_count"] = max(0, int(item.get("missing_quality_count", 0))) - item["last_seen_iter"] = max(0, int(item.get("last_seen_iter", 0))) - item["last_missing_iter"] = max(0, int(item.get("last_missing_iter", 0))) - cleaned[key] = item - except (TypeError, ValueError): - continue - global_adaptive["void_token_pairs"] = cleaned - global_adaptive["void_token_pairs_validated"] = True - return cleaned - def _lucky_context_score(self, item): - likes = float(item.get("liked_count", 0)) - neutral = float(item.get("neutral_count", 0)) - awful = float(item.get("awful_count", 0)) - wanted = float(item.get("wanted_count", 0)) - missing = float(item.get("missing_count", 0)) - concept_missing = float(item.get("missing_concept_count", 0)) - detail_missing = float(item.get("missing_detail_count", 0)) - return float(item.get("score", 0.0)) + likes * 0.38 + neutral * 0.06 + wanted * 0.24 + missing * 0.18 + concept_missing * 0.10 + detail_missing * 0.08 - awful * 0.90 - def _ensure_lucky_context_memory(self, global_adaptive): - memory = global_adaptive.get("lucky_context_memory") - if not isinstance(memory, dict): - memory = {} - if memory and global_adaptive.get("lucky_context_memory_validated") is True: - return memory - - cleaned = {} - for anchor, neighbors in memory.items(): - anchor = str(anchor).strip().lower() - if not self._is_valuable_token(anchor) or not isinstance(neighbors, dict): - continue - - clean_neighbors = {} - for neighbor, item in neighbors.items(): - neighbor = str(neighbor).strip().lower() - if neighbor == anchor or not self._is_valuable_token(neighbor) or not isinstance(item, dict): - continue - try: - item["score"] = round(max(-4.0, min(8.0, float(item.get("score", 0.0)))), 4) - item["liked_count"] = max(0, int(item.get("liked_count", 0))) - item["neutral_count"] = max(0, int(item.get("neutral_count", 0))) - item["awful_count"] = max(0, int(item.get("awful_count", 0))) - item["wanted_count"] = max(0, int(item.get("wanted_count", 0))) - item["missing_count"] = max(0, int(item.get("missing_count", 0))) - item["missing_concept_count"] = max(0, int(item.get("missing_concept_count", 0))) - item["missing_detail_count"] = max(0, int(item.get("missing_detail_count", 0))) - item["missing_quality_count"] = max(0, int(item.get("missing_quality_count", 0))) - item["last_seen_iter"] = max(0, int(item.get("last_seen_iter", 0))) - item["last_missing_iter"] = max(0, int(item.get("last_missing_iter", 0))) - item["co_count"] = max(0, int(item.get("co_count", 0))) - clean_neighbors[neighbor] = item - except (TypeError, ValueError): - continue - - if clean_neighbors: - cleaned[anchor] = clean_neighbors - - global_adaptive["lucky_context_memory"] = cleaned - global_adaptive["lucky_context_memory_validated"] = True - return cleaned def _phrase_position_score(self, item): likes = float(item.get("liked_count", 0)) @@ -1786,52 +1642,6 @@ class FunPackVideoRefiner: missing = float(item.get("missing_count", 0)) return float(item.get("score", 0.0)) + likes * 0.45 + neutral * 0.06 + wanted * 0.26 + missing * 0.18 - awful * 0.85 - def _ensure_lucky_phrase_placements(self, global_adaptive): - memory = global_adaptive.get("lucky_phrase_placements") - if not isinstance(memory, dict): - memory = {} - if memory and global_adaptive.get("lucky_phrase_placements_validated") is True: - return memory - - cleaned = {} - for phrase, item in memory.items(): - phrase_key = str(phrase).strip().lower() - if len(phrase_key) < 3 or not isinstance(item, dict): - continue - positions = item.get("positions") - if not isinstance(positions, dict): - continue - clean_positions = {} - for pos, pos_item in positions.items(): - if not isinstance(pos_item, dict): - continue - try: - pos_key = str(max(0, min(31, int(pos)))) - pos_item["score"] = round(max(-4.0, min(8.0, float(pos_item.get("score", 0.0)))), 4) - pos_item["liked_count"] = max(0, int(pos_item.get("liked_count", 0))) - pos_item["neutral_count"] = max(0, int(pos_item.get("neutral_count", 0))) - pos_item["awful_count"] = max(0, int(pos_item.get("awful_count", 0))) - pos_item["wanted_count"] = max(0, int(pos_item.get("wanted_count", 0))) - pos_item["missing_count"] = max(0, int(pos_item.get("missing_count", 0))) - pos_item["missing_concept_count"] = max(0, int(pos_item.get("missing_concept_count", 0))) - pos_item["missing_detail_count"] = max(0, int(pos_item.get("missing_detail_count", 0))) - pos_item["missing_quality_count"] = max(0, int(pos_item.get("missing_quality_count", 0))) - pos_item["count"] = max(0, int(pos_item.get("count", 0))) - pos_item["last_seen_iter"] = max(0, int(pos_item.get("last_seen_iter", 0))) - pos_item["last_missing_iter"] = max(0, int(pos_item.get("last_missing_iter", 0))) - clean_positions[pos_key] = pos_item - except (TypeError, ValueError): - continue - if not clean_positions: - continue - cleaned[phrase_key] = { - "text": str(item.get("text", phrase_key)).strip().lower() or phrase_key, - "positions": clean_positions, - } - - global_adaptive["lucky_phrase_placements"] = cleaned - global_adaptive["lucky_phrase_placements_validated"] = True - return cleaned def _rating_memory_deltas(self, rating_key): if rating_key == "like": @@ -1854,235 +1664,12 @@ class FunPackVideoRefiner: return 0.0, 0, 1, 0 return 0.12, 0, 1, 0 - def _update_lucky_context_memory(self, global_adaptive, tokens, rating_key, iter_num): - memory = self._ensure_lucky_context_memory(global_adaptive) - ordered_tokens = [] - seen = set() - for token in tokens or []: - token = str(token).strip().lower() - if token in seen or not self._is_valuable_token(token): - continue - seen.add(token) - ordered_tokens.append(token) - if len(ordered_tokens) < 2: - return memory - score_delta, liked_delta, neutral_delta, awful_delta = self._rating_memory_deltas(rating_key) - missing_axes = self._missing_axes_for_rating_key(rating_key) - if missing_axes: - score_delta += self._axis_memory_boost(missing_axes) * 0.55 - # Keep context learning local and ordered. A full all-to-all update makes - # long Lucky prompts create thousands of JSON entries per click. - window = 6 - for index, anchor in enumerate(ordered_tokens): - neighbors = memory.setdefault(anchor, {}) - start = max(0, index - window) - end = min(len(ordered_tokens), index + window + 1) - for neighbor_index in range(start, end): - if neighbor_index == index: - continue - neighbor = ordered_tokens[neighbor_index] - distance = abs(neighbor_index - index) - if distance <= 0: - continue - local_weight = 1.0 / float(distance) - item = neighbors.get(neighbor, {}) - item["score"] = round(max(-4.0, min(8.0, float(item.get("score", 0.0)) + score_delta * local_weight)), 4) - item["liked_count"] = max(0, int(item.get("liked_count", 0)) + liked_delta) - item["neutral_count"] = max(0, int(item.get("neutral_count", 0)) + neutral_delta) - item["awful_count"] = max(0, int(item.get("awful_count", 0)) + awful_delta) - if missing_axes: - item = self._apply_missing_memory_pressure(item, missing_axes, iter_num) - item["co_count"] = max(0, int(item.get("co_count", 0)) + 1) - item["last_seen_iter"] = int(iter_num) - neighbors[neighbor] = item - return memory - def _update_lucky_phrase_placements(self, global_adaptive, prompt_text, rating_key, iter_num): - phrases = self._ordered_prompt_phrases(prompt_text) - if not phrases: - return self._ensure_lucky_phrase_placements(global_adaptive) - memory = self._ensure_lucky_phrase_placements(global_adaptive) - score_delta, liked_delta, neutral_delta, awful_delta = self._rating_memory_deltas(rating_key) - missing_axes = self._missing_axes_for_rating_key(rating_key) - if missing_axes: - score_delta += self._axis_memory_boost(missing_axes) * 0.70 - for index, phrase in enumerate(phrases[:32]): - text = str(phrase.get("text", "")).strip().lower() - if len(text) < 3: - continue - entry = memory.setdefault(text, {"text": text, "positions": {}}) - positions = entry.setdefault("positions", {}) - pos_key = str(index) - item = positions.get(pos_key, {}) - item["score"] = round(max(-4.0, min(8.0, float(item.get("score", 0.0)) + score_delta)), 4) - item["liked_count"] = max(0, int(item.get("liked_count", 0)) + liked_delta) - item["neutral_count"] = max(0, int(item.get("neutral_count", 0)) + neutral_delta) - item["awful_count"] = max(0, int(item.get("awful_count", 0)) + awful_delta) - if missing_axes: - item = self._apply_missing_memory_pressure(item, missing_axes, iter_num) - item["count"] = max(0, int(item.get("count", 0)) + 1) - item["last_seen_iter"] = int(iter_num) - positions[pos_key] = item - entry["text"] = text - memory[text] = entry - global_adaptive["lucky_phrase_placements"] = memory - return memory - - def _update_void_token_pairs(self, global_adaptive, word_groups, rating_key, iter_num): - pair_bank = self._ensure_void_pair_bank(global_adaptive) - ordered_tokens = [] - seen_positions = set() - for start, _, full_word, _ in sorted(word_groups or [], key=lambda group: (group[0], group[1])): - token = str(full_word).strip().lower() - if not self._is_valuable_token(token): - continue - marker = (int(start), token) - if marker in seen_positions: - continue - seen_positions.add(marker) - ordered_tokens.append(token) - - if len(ordered_tokens) < 2: - return pair_bank - - base_delta, liked_delta, neutral_delta, awful_delta = self._rating_memory_deltas(rating_key) - missing_axes = self._missing_axes_for_rating_key(rating_key) - score_delta = base_delta * 0.84 - if missing_axes: - score_delta += self._axis_memory_boost(missing_axes) * 0.48 - - for left, right in zip(ordered_tokens, ordered_tokens[1:]): - if left == right: - continue - key = self._void_pair_key(left, right) - item = pair_bank.get(key, {}) - item["left"] = left - item["right"] = right - item["score"] = round(max(-3.0, min(6.0, float(item.get("score", 0.0)) + score_delta)), 4) - item["liked_count"] = max(0, int(item.get("liked_count", 0)) + liked_delta) - item["neutral_count"] = max(0, int(item.get("neutral_count", 0)) + neutral_delta) - item["awful_count"] = max(0, int(item.get("awful_count", 0)) + awful_delta) - if missing_axes: - item = self._apply_missing_memory_pressure(item, missing_axes, iter_num) - item["last_seen_iter"] = int(iter_num) - pair_bank[key] = item - - self._update_lucky_context_memory(global_adaptive, ordered_tokens, rating_key, iter_num) - global_adaptive["void_token_pairs"] = pair_bank - return global_adaptive["void_token_pairs"] - - def _void_pair_is_poor(self, global_adaptive, left_token, right_token): - if not left_token or not right_token: - return False - pair_bank = global_adaptive.get("void_token_pairs") - if not isinstance(pair_bank, dict): - return False - item = pair_bank.get(self._void_pair_key(left_token, right_token)) - if not isinstance(item, dict): - return False - awful_count = int(item.get("awful_count", 0)) - positive_count = int(item.get("liked_count", 0)) + int(item.get("neutral_count", 0)) - return awful_count > 0 and self._void_pair_score(item) < -0.05 and awful_count >= max(1, positive_count) - - def _update_void_token_bank(self, global_adaptive, word_groups, cur_positive, - rating_profile, iter_num, token_mask=None): - bank = self._ensure_void_token_bank(global_adaptive) - rating_key = rating_profile.get("key", "") - if not word_groups or not isinstance(cur_positive, torch.Tensor) or rating_profile.get("skip_learning"): - return bank - - score_delta, liked_delta, neutral_delta, awful_delta = self._rating_memory_deltas(rating_key) - missing_axes = set(rating_profile.get("missing_axes", [])) - if missing_axes: - score_delta += self._axis_memory_boost(missing_axes) - - token_mask_list = None - if token_mask is not None: - token_mask_list = token_mask.detach().cpu().tolist() - - for start, end, full_word, _ in word_groups: - if not self._is_valuable_token(full_word): - continue - if token_mask_list is not None and not all(token_mask_list[start:min(len(token_mask_list), end)]): - continue - embedding = self._void_token_embedding(cur_positive, start, end) - if embedding is None: - continue - - token = full_word.strip().lower() - item = bank.get(token, {}) - count = max(0, int(item.get("sample_count", 0))) - if "embedding" in item and count > 0 and rating_key != "awful": - try: - previous = serializable_to_tensor(item["embedding"]).to(embedding.device) - if list(previous.shape) == list(embedding.shape): - if rating_key == "like": - mix = 0.62 - elif rating_key == "missing_details": - mix = 0.48 - else: - mix = min(0.30, 1.0 / float(count + 1)) - embedding = previous.lerp(embedding, mix) - except Exception: - pass - - item["embedding"] = tensor_to_serializable(embedding.float().cpu()) - item["score"] = round(max(-2.0, min(8.0, float(item.get("score", 0.0)) + score_delta)), 4) - item["liked_count"] = max(0, int(item.get("liked_count", 0)) + liked_delta) - item["neutral_count"] = max(0, int(item.get("neutral_count", 0)) + neutral_delta) - item["awful_count"] = max(0, int(item.get("awful_count", 0)) + awful_delta) - if missing_axes: - item = self._apply_missing_memory_pressure(item, missing_axes, iter_num) - item["sample_count"] = count + (0 if rating_key == "awful" else 1) - item["last_seen_iter"] = int(iter_num) - bank[token] = item - - self._update_void_token_pairs(global_adaptive, word_groups, rating_key, iter_num) - - global_adaptive["void_token_bank"] = bank - return global_adaptive["void_token_bank"] - - def _seed_void_token_bank(self, global_adaptive, word_groups, cur_positive, iter_num, token_mask=None): - return self._update_void_token_bank( - global_adaptive, - word_groups, - cur_positive, - {"key": "discover", "skip_learning": False}, - iter_num, - token_mask=token_mask, - ) - - def _lucky_groups_from_history(self, history_entry, seq_len): - if not isinstance(history_entry, dict) or seq_len <= 0: - return [] - lucky = history_entry.get("lucky") if isinstance(history_entry.get("lucky"), dict) else {} - injections = lucky.get("injections", []) - groups = [] - seen = set() - for item in injections: - if not isinstance(item, dict): - continue - token = str(item.get("token", "")).strip().lower() - if not self._is_valuable_token(token): - continue - try: - start = int(item.get("start", 0)) - end = int(item.get("end", start + 1)) - except (TypeError, ValueError): - continue - start = max(0, min(seq_len - 1, start)) - end = max(start + 1, min(seq_len, end)) - marker = (start, token) - if marker in seen: - continue - seen.add(marker) - groups.append((start, end, token, [])) - return groups def _history_modified_conditioning(self, history_entry, reference, device): if not isinstance(history_entry, dict): @@ -2198,80 +1785,6 @@ class FunPackVideoRefiner: return 0 return int(shape[1] if len(shape) == 3 else shape[0]) - def _select_lucky_memory_canvas(self, prompt_histories, template, device, dtype): - if not isinstance(prompt_histories, dict) or not isinstance(template, torch.Tensor): - return None, "" - - candidates = [] - current_seq = self._get_conditioning_seq_len(template) - - def consider(payload, label, quality=0.0, iteration=0): - if payload is None: - return - if not self._serialized_conditioning_compatible(payload, template): - return - seq_len = self._serialized_conditioning_seq_len(payload) - candidates.append({ - "payload": payload, - "label": label, - "seq_len": int(seq_len), - "quality": float(quality), - "iteration": int(iteration or 0), - }) - - for prompt_key, prompt_state in prompt_histories.items(): - if not isinstance(prompt_state, dict): - continue - short_key = str(prompt_key)[:48] or "prompt" - liked_count = int(prompt_state.get("liked_reference_count", 0) or 0) - if liked_count > 0: - consider( - prompt_state.get("liked_reference_embeds"), - f"liked '{short_key}'", - quality=2.0 + min(3.0, liked_count * 0.25), - iteration=prompt_state.get("liked_reference_last_iteration", 0), - ) - consider(prompt_state.get("reference_embeds"), f"reference '{short_key}'", quality=0.5) - consider(prompt_state.get("source_conditioning_embeds"), f"source '{short_key}'", quality=0.25) - - for entry in prompt_state.get("history", []) or []: - if not isinstance(entry, dict): - continue - profile = normalize_refiner_rating(entry.get("rating_label", entry.get("rating", 0))) - key = profile.get("key", "") - if key in {"forget", "awful"}: - continue - quality = float(profile.get("reward", 0.0)) - if key == "like": - quality += 1.5 - elif key.startswith("missing"): - quality += 0.45 - consider( - entry.get("modified_embeds"), - f"history {entry.get('iteration', 0)} '{short_key}'", - quality=quality, - iteration=entry.get("iteration", 0), - ) - - if not candidates: - return None, "" - - best = max( - candidates, - key=lambda item: ( - item["seq_len"] >= current_seq, - item["seq_len"], - item["quality"], - item["iteration"], - ) - ) - try: - canvas = serializable_to_tensor(best["payload"]).to(device=device, dtype=dtype) - except Exception: - return None, "" - if not self._conditioning_canvas_compatible(canvas, template): - return None, "" - return canvas.clone(), f"{best['label']} ({best['seq_len']} positions)" def _legacy_auto_inject_token_allowed(self, token): token = str(token or "").strip().lower() @@ -2282,731 +1795,17 @@ class FunPackVideoRefiner: return False return True - def _eligible_void_bank_items(self, global_adaptive, embedding_dim): - bank = self._ensure_void_token_bank(global_adaptive) - eligible = [] - for token, item in bank.items(): - if not self._legacy_auto_inject_token_allowed(token): - continue - if int(item.get("awful_count", 0)) >= 2 and int(item.get("liked_count", 0)) <= 0: - continue - score = self._void_bank_score(item) - if score <= 0.35: - continue - try: - vector = serializable_to_tensor(item["embedding"]).float() - except Exception: - continue - if vector.dim() != 1 or int(vector.shape[0]) != int(embedding_dim): - continue - eligible.append((token, item, score)) - return eligible - - def _eligible_lucky_bank_items(self, global_adaptive, embedding_dim): - bank = self._ensure_void_token_bank(global_adaptive) - candidates = [] - for token, item in bank.items(): - if not self._legacy_auto_inject_token_allowed(token): - continue - score = self._void_bank_score(item) - embedding_shape = self._serialized_tensor_shape(item.get("embedding")) - if embedding_shape is None or len(embedding_shape) != 1 or int(embedding_shape[0]) != int(embedding_dim): - continue - candidates.append((token, item, score)) - - return candidates - - def _lucky_prompt_anchor_tokens(self, word_groups, token_vectors, global_adaptive): - context_memory = self._ensure_lucky_context_memory(global_adaptive) - anchors = [] - seen = set() - for _, _, full_word, _ in sorted(word_groups or [], key=lambda group: (group[0], group[1])): - token = str(full_word).strip().lower() - if token in seen or not self._is_valuable_token(token): - continue - if token in token_vectors or token in context_memory: - anchors.append(token) - seen.add(token) - return anchors - - def _lucky_context_neighbors(self, global_adaptive, anchor_tokens, token_vectors, present_tokens=None): - context_memory = self._ensure_lucky_context_memory(global_adaptive) - present = set(present_tokens or []) - scored = {} - for anchor in anchor_tokens or []: - for neighbor, item in context_memory.get(anchor, {}).items(): - if neighbor in present or neighbor not in token_vectors: - continue - score = self._lucky_context_score(item) - if score <= 0.15 and int(item.get("liked_count", 0)) <= 0: - continue - if self._void_pair_is_poor(global_adaptive, anchor, neighbor): - continue - scored[neighbor] = max(scored.get(neighbor, -999.0), score) - return [ - token for token, _ in sorted(scored.items(), key=lambda pair: pair[1], reverse=True) - ] - - def _lucky_global_tokens(self, lucky_pool): - scored = sorted(lucky_pool, key=lambda item: item[2], reverse=True) - preferred = [ - token for token, item, score in scored - if score > 0.10 or int(item.get("liked_count", 0)) > 0 or int(item.get("neutral_count", 0)) > 0 - ] - return preferred or [token for token, _, _ in scored] - - def _lucky_weighted_pick(self, token_scores, candidates, previous_token=None, global_adaptive=None): - usable = [] - weights = [] - for token in candidates or []: - token = str(token).strip().lower() - if not token: - continue - if previous_token and global_adaptive is not None and self._void_pair_is_poor(global_adaptive, previous_token, token): - continue - usable.append(token) - weights.append(max(0.05, float(token_scores.get(token, 0.0)) + 1.25)) - - if not usable: - return None - - if len(usable) == 1: - return usable[0] - - weights_tensor = torch.tensor(weights, dtype=torch.float32) - index = int(torch.multinomial(weights_tensor, 1).item()) - return usable[index] - - def _lucky_compose_tokens(self, global_adaptive, word_groups, lucky_pool, token_vectors, slot_count): - token_scores = {token: score for token, _, score in lucky_pool} - token_items = {token: item for token, item, _ in lucky_pool} - present_tokens = { - str(group[2]).strip().lower() - for group in (word_groups or []) - if isinstance(group, (list, tuple)) and len(group) >= 3 and self._is_valuable_token(str(group[2]).strip()) - } - anchor_tokens = self._lucky_prompt_anchor_tokens(word_groups, token_vectors, global_adaptive) - context_tokens = self._lucky_context_neighbors( - global_adaptive, - anchor_tokens, - token_vectors, - present_tokens=present_tokens, - ) - global_tokens = self._lucky_global_tokens(lucky_pool) - - seed_sequence = [] - for token in anchor_tokens + context_tokens + global_tokens: - if token in token_vectors and token not in seed_sequence: - seed_sequence.append(token) - - if not seed_sequence: - seed_sequence = [token for token, _, _ in lucky_pool] - - max_slots = max(0, int(slot_count)) - required_candidates = [] - for index, token in enumerate(anchor_tokens): - item = token_items.get(token, {}) - missing_pressure = ( - int(item.get("wanted_count", 0)) + - int(item.get("missing_count", 0)) + - int(item.get("missing_concept_count", 0)) + - int(item.get("missing_detail_count", 0)) + - int(item.get("missing_quality_count", 0)) - ) - # Current prompt anchors are required; missing pressure decides which - # ones win when the conditioning budget is tight. - priority = float(token_scores.get(token, 0.0)) + missing_pressure * 0.55 + 0.12 / float(index + 1) - required_candidates.append((priority, index, token)) - required_candidates = sorted(required_candidates, key=lambda item: (-item[0], item[1])) - reserve_count = min(len(required_candidates), max_slots) - if reserve_count > 3: - reserve_count = min(reserve_count, max(3, min(16, int(math.ceil(max_slots * 0.40))))) - required_tokens = [ - token for _, _, token in sorted(required_candidates[:reserve_count], key=lambda item: item[1]) - ] - - chosen_tokens = list(required_tokens) - previous_token = None - context_hits = 0 - if chosen_tokens: - previous_token = chosen_tokens[-1] - for index in range(len(chosen_tokens), max_slots): - context_index = index - len(required_tokens) - if context_tokens and 0 <= context_index < len(context_tokens): - candidates = [context_tokens[context_index]] - elif seed_sequence and random.random() < 0.72: - candidates = seed_sequence - else: - candidates = [token for token, _, _ in lucky_pool] - - token = self._lucky_weighted_pick(token_scores, candidates, previous_token, global_adaptive) - if token is None: - token = self._lucky_weighted_pick(token_scores, [token for token, _, _ in lucky_pool], previous_token, global_adaptive) - if token is None: - continue - - if token in context_tokens: - context_hits += 1 - chosen_tokens.append(token) - previous_token = token - - if anchor_tokens and context_hits: - source = "anchors+context" - elif anchor_tokens: - source = "prompt anchors" - elif context_hits: - source = "context" - else: - source = "global preferences" - - return chosen_tokens, { - "source": source, - "anchor_tokens": anchor_tokens, - "required_tokens": required_tokens, - "context_tokens": context_tokens, - "context_hits": context_hits, - } - - def _compose_lucky_prompt_text(self, global_adaptive, word_groups, embedding_dim, - prompt_sequences=None, phrase_sequences=None, target_count=64): - bank = self._ensure_void_token_bank(global_adaptive) - lucky_pool = [] - for token, item in bank.items(): - if not self._legacy_auto_inject_token_allowed(token): - continue - embedding_shape = self._serialized_tensor_shape(item.get("embedding")) - if embedding_shape is None or len(embedding_shape) != 1 or int(embedding_shape[0]) != int(embedding_dim): - continue - lucky_pool.append((token, item, self._void_bank_score(item))) - - if len(lucky_pool) < 2: - return "", { - "source": "bank too small", - "anchor_tokens": [], - "context_tokens": [], - "context_hits": 0, - "token_count": 0, - } - - token_vectors = {token: True for token, _, _ in lucky_pool} - - if len(token_vectors) < 2: - return "", { - "source": "bank unreadable", - "anchor_tokens": [], - "context_tokens": [], - "context_hits": 0, - "token_count": 0, - } - - count = max(4, min(int(target_count or 0), 32)) - chosen_tokens, compose_info = self._lucky_compose_tokens( - global_adaptive, - word_groups, - lucky_pool, - token_vectors, - count, - ) - - ordered_sequences = [] - if isinstance(prompt_sequences, list): - for sequence in prompt_sequences: - if not isinstance(sequence, list): - continue - ordered = [str(token).strip().lower() for token in sequence if str(token).strip().lower() in token_vectors] - if len(ordered) >= 2: - ordered_sequences.append(ordered) - - if ordered_sequences and compose_info.get("source") == "global preferences": - sequence = random.choice(ordered_sequences) - chosen_tokens = [ - sequence[index % len(sequence)] - if random.random() < 0.55 else chosen_tokens[index] - for index in range(min(len(chosen_tokens), count)) - ] - compose_info["source"] = "saved prompt order" - - token_scores = {token: score for token, _, score in lucky_pool} - prompt_tokens = [] - seen = set() - for token in chosen_tokens: - token = str(token).strip().lower() - if token in seen or not self._is_valuable_token(token): - continue - prompt_tokens.append(token) - seen.add(token) - - if not prompt_tokens: - return "", { - **compose_info, - "token_count": 0, - } - - prompt_phrases = [] - phrase_seen = set() - chosen_set = set(prompt_tokens) - if isinstance(phrase_sequences, list): - phrase_candidates = [] - placement_memory = self._ensure_lucky_phrase_placements(global_adaptive) - for sequence_index, sequence in enumerate(phrase_sequences[-64:]): - if not isinstance(sequence, list): - continue - for phrase_index, phrase in enumerate(sequence): - if not isinstance(phrase, dict): - continue - text = str(phrase.get("text", "")).strip() - if not text: - continue - tokens = [ - str(token).strip().lower() - for token in phrase.get("tokens", []) - if self._is_valuable_token(str(token).strip()) - ] - if not tokens: - continue - overlap = chosen_set & set(tokens) - if not overlap: - continue - score = sum(max(0.05, token_scores.get(token, 0.0) + 1.25) for token in overlap) - score += 0.18 / float(phrase_index + 1) - score += 0.03 / float(max(1, len(phrase_sequences) - sequence_index)) - preferred_position, placement_score = self._lucky_phrase_preferred_position( - placement_memory, - text, - fallback_position=phrase_index, - ) - score += max(0.0, placement_score) * 0.20 - phrase_candidates.append((score, preferred_position, phrase_index, text, tokens)) - - sorted_candidates = sorted(phrase_candidates, key=lambda item: item[0], reverse=True) - selected = [] - for score, preferred_position, phrase_index, text, tokens in sorted_candidates: - key = text.lower() - if key in phrase_seen: - continue - selected.append({ - "score": score, - "position": preferred_position, - "source_position": phrase_index, - "text": text, - "tokens": tokens, - }) - phrase_seen.add(key) - if len(selected) >= 12: - break - prompt_phrases = [ - item["text"] - for item in sorted( - selected, - key=lambda item: (item["position"], -item["score"], item["source_position"], item["text"]), - ) - ] - - if not prompt_phrases: - prompt_phrases = prompt_tokens - - prompt_text = ", ".join(prompt_phrases) - return prompt_text, { - **compose_info, - "token_count": len(prompt_tokens), - "tokens": prompt_tokens, - "phrases": prompt_phrases, - } - - def _lucky_phrase_preferred_position(self, placement_memory, phrase_text, fallback_position=0): - phrase_key = str(phrase_text).strip().lower() - entry = placement_memory.get(phrase_key) if isinstance(placement_memory, dict) else None - if not isinstance(entry, dict): - return int(fallback_position), 0.0 - positions = entry.get("positions") - if not isinstance(positions, dict) or not positions: - return int(fallback_position), 0.0 - - candidates = [] - for pos_key, item in positions.items(): - if not isinstance(item, dict): - continue - try: - pos = max(0, min(31, int(pos_key))) - except (TypeError, ValueError): - continue - score = self._phrase_position_score(item) - count_bonus = min(0.30, math.log1p(float(item.get("count", 0))) * 0.08) - candidates.append((score + count_bonus, pos)) - - if not candidates: - return int(fallback_position), 0.0 - - candidates = sorted(candidates, key=lambda item: item[0], reverse=True) - top_score = candidates[0][0] - close = [item for item in candidates if item[0] >= top_score - 0.18] - if len(close) > 1 and random.random() < 0.35: - weights = torch.tensor([max(0.05, item[0] + 1.25) for item in close], dtype=torch.float32) - selected = close[int(torch.multinomial(weights, 1).item())] - else: - selected = candidates[0] - return int(selected[1]), float(selected[0]) - - def _encode_lucky_prompt_conditioning(self, clip, prompt_text, template, device, dtype): - if clip is None or not prompt_text: - return None, None, "" - try: - tokens = clip.tokenize(prompt_text) - encoded = clip.encode_from_tokens_scheduled(tokens) - except Exception as e: - print(f"[FunPackVideoRefiner] Lucky prompt CLIP/Gemma encode failed: {e}") - return None, None, f"encode failed: {e}" - - if not isinstance(encoded, list) or not encoded: - return None, None, "encode returned empty conditioning" - - item = encoded[0] - if isinstance(item, (list, tuple)) and len(item) >= 2: - cond = item[0] - meta = item[1] if isinstance(item[1], dict) else {"pooled_output": None} - else: - cond = item if isinstance(item, torch.Tensor) else None - meta = {"pooled_output": None} - - if not isinstance(cond, torch.Tensor): - return None, None, "encode returned invalid conditioning" - - cond = cond.to(device=device, dtype=dtype) - if not self._conditioning_canvas_compatible(cond, template): - return None, None, f"encoded shape {tuple(cond.shape)} incompatible" - - return cond, meta, f"encoded prompt ({self._get_conditioning_seq_len(cond)} positions)" - - def _select_void_tokens(self, global_adaptive, word_groups, cur_positive, - word_importance, token_mask=None, max_count=3): - if not isinstance(cur_positive, torch.Tensor) or cur_positive.dim() <= 1: - return [], "idle" - embedding_dim = int(cur_positive.shape[-1]) - eligible = self._eligible_void_bank_items(global_adaptive, embedding_dim) - if eligible: - pool = sorted(eligible, key=lambda item: item[2], reverse=True)[:24] - weights = torch.tensor([max(0.01, item[2]) for item in pool], dtype=torch.float32) - count = min(max_count, max(1, int(torch.randint(1, min(3, len(pool)) + 1, (1,)).item()))) - picked_indexes = torch.multinomial(weights, count, replacement=False).tolist() - picked = [] - for index in picked_indexes: - token, item, _ = pool[index] - picked.append((token, serializable_to_tensor(item["embedding"]).float())) - return picked, "bank" - - fallback = [] - token_mask_list = None - if token_mask is not None: - token_mask_list = token_mask.detach().cpu().tolist() - for start, end, full_word, _ in word_groups: - token = full_word.strip().lower() - if not self._is_valuable_token(token): - continue - if not self._legacy_auto_inject_token_allowed(token): - continue - if token_mask_list is not None and not all(token_mask_list[start:min(len(token_mask_list), end)]): - continue - score = float(word_importance.get(token, 1.0)) - if score < 1.05: - continue - embedding = self._void_token_embedding(cur_positive, start, end) - if embedding is not None: - fallback.append((token, embedding.float().cpu(), score)) - fallback = sorted(fallback, key=lambda item: item[2], reverse=True)[:max_count] - return [(token, vector) for token, vector, _ in fallback], "current" if fallback else "empty" - - def _apply_into_the_void(self, new_positive, reference, global_adaptive, word_groups, - word_importance, into_the_void=False, im_feeling_lucky=False, - token_mask=None, prompt_sequences=None, lucky_canvas=None, - lucky_canvas_label="", lucky_prompt_text="", lucky_prompt_tokens=None): - if not into_the_void and not im_feeling_lucky: - return new_positive, self._void_empty_status(False), {} - if not isinstance(new_positive, torch.Tensor) or not isinstance(reference, torch.Tensor): - return new_positive, "Void: on | unavailable | tokens: none", {} - if new_positive.dim() <= 1 or list(new_positive.shape) != list(reference.shape): - return new_positive, "Void: on | incompatible conditioning | tokens: none", {} - - device = new_positive.device - dtype = new_positive.dtype - seq_len = new_positive.shape[1] if new_positive.dim() == 3 else new_positive.shape[0] - active_len = self._get_effective_seq_len(token_mask, seq_len) - active_len = max(1, min(seq_len, active_len)) - mixed = new_positive.clone() - status_parts = [] - lucky_metadata = {} - lucky_canvas_used = False - lucky_canvas_status = "" - - if ( - im_feeling_lucky and - self._conditioning_canvas_compatible(lucky_canvas, new_positive) - ): - try: - canvas = lucky_canvas.to(device=device, dtype=dtype) - except Exception: - canvas = None - if isinstance(canvas, torch.Tensor): - mixed = canvas.clone() - reference = canvas.clone() - seq_len = mixed.shape[1] if mixed.dim() == 3 else mixed.shape[0] - if token_mask is None or int(token_mask.numel()) != int(seq_len): - token_mask = self._get_conditioning_token_mask(mixed) - active_len = self._get_effective_seq_len(token_mask, seq_len) - active_len = max(1, min(seq_len, active_len)) - lucky_canvas_used = True - lucky_canvas_status = lucky_canvas_label or f"memory ({seq_len} positions)" - - clamp_base = mixed.clone() - - if into_the_void: - selected, source = self._select_void_tokens( - global_adaptive, - word_groups, - mixed, - word_importance, - token_mask=token_mask, - ) - if selected: - positions = torch.linspace(0, active_len - 1, steps=len(selected), device=device).round().long() - strength = min(0.055, 0.09 / max(1, len(selected))) - - for index, (_, vector) in enumerate(selected): - vector = vector.to(device=device, dtype=dtype) - if vector.dim() != 1 or int(vector.shape[0]) != int(mixed.shape[-1]): - continue - pos = int(positions[index].item()) - if token_mask is not None and pos < token_mask.shape[0] and not bool(token_mask[pos].item()): - active_positions = torch.nonzero(token_mask[:active_len], as_tuple=False).flatten() - if active_positions.numel() == 0: - continue - pos = int(active_positions[min(index, active_positions.numel() - 1)].item()) - if mixed.dim() == 3: - mixed[:, pos, :] = mixed[:, pos, :].lerp(vector.view(1, -1), strength) - else: - mixed[pos, :] = mixed[pos, :].lerp(vector, strength) - - token_list = ", ".join(token for token, _ in selected) - status_parts.append(f"Void: on | {source} | tokens: {token_list}") - else: - status_parts.append(f"Void: on | {source} | tokens: none") - else: - status_parts.append("Void: off | idle | tokens: none") - - if im_feeling_lucky: - active_positions = torch.arange(active_len, device=device) - if token_mask is not None: - active_positions = torch.nonzero(token_mask[:active_len], as_tuple=False).flatten() - - prompt_tokens = [ - str(token).strip().lower() - for token in (lucky_prompt_tokens or []) - if self._is_valuable_token(str(token).strip()) - ] - if lucky_canvas_used and prompt_tokens: - positions = torch.linspace( - 0, - max(0, int(active_positions.numel()) - 1), - steps=min(len(prompt_tokens), max(1, int(active_positions.numel()))), - device=device, - ).round().long() - injections = [] - for index, token in enumerate(prompt_tokens[:int(positions.numel())]): - pos = int(active_positions[int(positions[index].item())].item()) if active_positions.numel() > 0 else index - injections.append({"token": token, "start": pos, "end": pos + 1}) - lucky_metadata = { - "enabled": True, - "source": "encoded prompt", - "tokens": prompt_tokens, - "unique_tokens": sorted(set(prompt_tokens)), - "injections": injections, - "anchor_tokens": [], - "required_tokens": prompt_tokens, - "context_tokens": [], - "context_hits": 0, - "canvas": lucky_canvas_status, - "prompt": lucky_prompt_text, - } - token_preview = ", ".join(prompt_tokens[:8]) - if len(prompt_tokens) > 8: - token_preview += ", ..." - prompt_preview = re.sub(r"\s+", " ", lucky_prompt_text).strip() - if len(prompt_preview) > 160: - prompt_preview = prompt_preview[:157].rstrip() + "..." - status_parts.append( - f"Lucky: on | encoded prompt | {len(prompt_tokens)} prompt tokens | " - f"canvas: {lucky_canvas_status} | prompt: {prompt_preview} | tokens: {token_preview}" - ) - else: - lucky_pool = self._eligible_lucky_bank_items(global_adaptive, int(mixed.shape[-1])) - if len(lucky_pool) < 2: - if lucky_canvas_used: - lucky_metadata = { - "enabled": True, - "source": "memory canvas", - "tokens": [], - "unique_tokens": [], - "injections": [], - "anchor_tokens": [], - "required_tokens": [], - "context_tokens": [], - "context_hits": 0, - "canvas": lucky_canvas_status, - "prompt": lucky_prompt_text, - } - status_parts.append( - f"Lucky: on | memory canvas {lucky_canvas_status} | bank too small ({len(lucky_pool)}/2 eligible) | tokens: none" - ) - else: - status_parts.append(f"Lucky: on | bank too small ({len(lucky_pool)}/2 eligible) | tokens: none") - elif active_positions.numel() <= 0: - if lucky_canvas_used: - lucky_metadata = { - "enabled": True, - "source": "memory canvas", - "tokens": [], - "unique_tokens": [], - "injections": [], - "anchor_tokens": [], - "required_tokens": [], - "context_tokens": [], - "context_hits": 0, - "canvas": lucky_canvas_status, - "prompt": lucky_prompt_text, - } - status_parts.append(f"Lucky: on | memory canvas {lucky_canvas_status} | no active positions | tokens: none") - else: - status_parts.append("Lucky: on | no active positions | tokens: none") - else: - token_membership = {token: True for token, _, _ in lucky_pool} - - ordered_sequences = [] - if isinstance(prompt_sequences, list): - for sequence in prompt_sequences: - if not isinstance(sequence, list): - continue - ordered = [token for token in sequence if token in token_membership] - if len(ordered) >= 2: - ordered_sequences.append(ordered) - - chosen_tokens, compose_info = self._lucky_compose_tokens( - global_adaptive, - word_groups, - lucky_pool, - token_membership, - int(active_positions.numel()), - ) - if ordered_sequences and compose_info.get("source") == "global preferences": - sequence = random.choice(ordered_sequences) - chosen_tokens = [ - sequence[index % len(sequence)] - if random.random() < 0.55 else chosen_tokens[index] - for index in range(min(len(chosen_tokens), int(active_positions.numel()))) - ] - compose_info["source"] = "saved prompt order" - - needed_tokens = set(chosen_tokens) - token_vectors = {} - for token, item, _ in lucky_pool: - if token not in needed_tokens: - continue - try: - vector = serializable_to_tensor(item["embedding"]).to(device=device, dtype=dtype) - except Exception: - continue - if vector.dim() == 1 and int(vector.shape[0]) == int(mixed.shape[-1]): - token_vectors[token] = vector - - lucky_field = mixed.clone() - picked_tokens = [] - injections = [] - for pos_tensor, token in zip(active_positions, chosen_tokens): - vector = token_vectors.get(token) - if vector is None: - continue - pos = int(pos_tensor.item()) - if lucky_field.dim() == 3: - lucky_field[:, pos, :] = vector.view(1, -1) - else: - lucky_field[pos, :] = vector - picked_tokens.append(token) - injections.append({"token": token, "start": pos, "end": pos + 1}) - - if picked_tokens: - strength = 0.030 if into_the_void else 0.040 - mixed = mixed.lerp(lucky_field, strength) - token_preview = [] - seen_tokens = set() - for token in picked_tokens: - if token in seen_tokens: - continue - token_preview.append(token) - seen_tokens.add(token) - if len(token_preview) >= 8: - break - token_list = ", ".join(token_preview) - if len(seen_tokens) < len(set(picked_tokens)): - token_list += ", ..." - anchor_preview = ", ".join(compose_info.get("anchor_tokens", [])[:4]) or "none" - required_preview = ", ".join(compose_info.get("required_tokens", [])[:6]) or "none" - context_preview = ", ".join(compose_info.get("context_tokens", [])[:6]) or "none" - lucky_metadata = { - "enabled": True, - "source": compose_info.get("source", "global preferences"), - "tokens": picked_tokens, - "unique_tokens": sorted(set(picked_tokens)), - "injections": injections, - "anchor_tokens": compose_info.get("anchor_tokens", []), - "required_tokens": compose_info.get("required_tokens", []), - "context_tokens": compose_info.get("context_tokens", []), - "context_hits": int(compose_info.get("context_hits", 0)), - "canvas": lucky_canvas_status if lucky_canvas_used else "current conditioning", - "prompt": lucky_prompt_text, - } - canvas_phrase = f" | canvas: {lucky_canvas_status}" if lucky_canvas_used else "" - prompt_preview = re.sub(r"\s+", " ", lucky_prompt_text).strip() - if len(prompt_preview) > 160: - prompt_preview = prompt_preview[:157].rstrip() + "..." - prompt_phrase = f" | prompt: {prompt_preview}" if prompt_preview else "" - status_parts.append( - f"Lucky: on | {lucky_metadata['source']} | field {len(picked_tokens)} positions from {len(set(picked_tokens))} tokens | " - f"anchors: {anchor_preview} | required: {required_preview} | context add: {context_preview}{canvas_phrase}{prompt_phrase} | tokens: {token_list}" - ) - else: - if lucky_canvas_used: - lucky_metadata = { - "enabled": True, - "source": "memory canvas", - "tokens": [], - "unique_tokens": [], - "injections": [], - "anchor_tokens": [], - "required_tokens": [], - "context_tokens": [], - "context_hits": 0, - "canvas": lucky_canvas_status, - "prompt": lucky_prompt_text, - } - status_parts.append(f"Lucky: on | memory canvas {lucky_canvas_status} | bank unreadable | tokens: none") - else: - status_parts.append("Lucky: on | bank unreadable | tokens: none") - else: - status_parts.append("Lucky: off") - - delta = mixed - clamp_base - reference_norm = reference.norm(dim=-1, keepdim=True).clamp_min(1e-8) - max_delta = reference_norm * 0.08 - delta_norm = delta.norm(dim=-1, keepdim=True).clamp_min(1e-8) - scale = torch.minimum(torch.ones_like(delta_norm), max_delta / delta_norm) - mixed = clamp_base + delta * scale - mixed = torch.clamp(mixed, min=-60.0, max=60.0) - norm_factor = clamp_base.norm(dim=-1, keepdim=True).clamp_min(1e-8) - mixed = mixed / mixed.norm(dim=-1, keepdim=True).clamp_min(1e-8) * norm_factor - return mixed, " | ".join(status_parts), lucky_metadata - # ========================================================================= - # SCHEDULER - # ========================================================================= + + + + + + + + + def _get_scheduler_factors(self, mode, rating, reward, similarity, iter_num, total_iters, word_importance, word_groups, global_adaptive, device): @@ -3401,91 +2200,7 @@ class FunPackVideoRefiner: }) return phrases - def _lucky_prompt_phrase_sequences(self, prompt_histories: dict): - if not isinstance(prompt_histories, dict): - return [] - sequences = [] - seen = set() - for prompt_key, history_entry in prompt_histories.items(): - if not isinstance(history_entry, dict): - continue - - history = history_entry.get("history", []) - good_history = [] - if isinstance(history, list): - for item in history[-16:]: - if not isinstance(item, dict): - continue - profile = normalize_refiner_rating(item.get("rating_label", item.get("rating", 0))) - if int(profile.get("level", 0)) >= 5: - good_history.append(item) - - texts = [] - for item in good_history: - for key in ("prompt_full", "analysis_prompt", "prompt"): - text = item.get(key) - if isinstance(text, str) and text.strip(): - texts.append(text) - if good_history: - texts.extend(self._history_prompt_texts(prompt_key, history_entry)) - - for text in texts: - phrases = self._ordered_prompt_phrases(text) - if not phrases: - continue - signature = tuple(phrase["text"] for phrase in phrases) - if signature in seen: - continue - seen.add(signature) - sequences.append(phrases[:32]) - - return sequences[-48:] - - def _lucky_prompt_sequences(self, prompt_histories: dict): - if not isinstance(prompt_histories, dict): - return [] - - sequences = [] - seen = set() - for prompt_key, history_entry in prompt_histories.items(): - if not isinstance(history_entry, dict): - continue - - history = history_entry.get("history", []) - good_history = [] - if isinstance(history, list): - for item in history[-24:]: - if not isinstance(item, dict): - continue - profile = normalize_refiner_rating(item.get("rating_label", item.get("rating", 0))) - if int(profile.get("level", 0)) >= 5: - good_history.append(item) - - for item in good_history: - for key in ("prompt_full", "analysis_prompt", "prompt"): - text = item.get(key) - if not isinstance(text, str) or not text.strip(): - continue - ordered = self._ordered_prompt_words(text) - if len(ordered) < 2: - continue - signature = tuple(ordered) - if signature not in seen: - seen.add(signature) - sequences.append(ordered[:64]) - - if good_history: - for text in self._history_prompt_texts(prompt_key, history_entry): - ordered = self._ordered_prompt_words(text) - if len(ordered) < 2: - continue - signature = tuple(ordered) - if signature not in seen: - seen.add(signature) - sequences.append(ordered[:64]) - - return sequences[-96:] def _history_prompt_texts(self, prompt_key: str, history_entry: dict): texts = []