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 = []