nvlabs--sana/diffusion/post_training/stat_tracking.py

import numpy as np
import torch


class PerPromptStatTracker:
    def __init__(self, global_std=False):
        self.global_std = global_std
        self.stats = {}
        self.history_prompts = set()

    def update(self, prompts, rewards, exp=False):
        prompts = np.array(prompts)
        rewards = np.array(rewards, dtype=np.float64)
        unique = np.unique(prompts)
        advantages = np.empty_like(rewards) * 0.0

        for prompt in unique:
            prompt_rewards = rewards[prompts == prompt]
            if prompt not in self.stats:
                self.stats[prompt] = []
            self.stats[prompt].extend(prompt_rewards)
            self.history_prompts.add(hash(prompt))
        for prompt in unique:
            self.stats[prompt] = np.stack(self.stats[prompt])
            prompt_rewards = rewards[prompts == prompt]
            mean = np.mean(self.stats[prompt], axis=0, keepdims=True)
            if self.global_std:
                std = np.std(rewards, axis=0, keepdims=True) + 1e-4
            else:
                std = np.std(self.stats[prompt], axis=0, keepdims=True) + 1e-4
            advantages[prompts == prompt] = (prompt_rewards - mean) / std
        return advantages

    def get_stats(self):
        avg_group_size = sum(len(v) for v in self.stats.values()) / len(self.stats) if self.stats else 0
        history_prompts = len(self.history_prompts)
        return avg_group_size, history_prompts

    def clear(self):
        self.stats = {}

    def get_mean_of_top_rewards(self, top_percentage):
        if not self.stats:
            return 0.0

        assert 0 <= top_percentage <= 100

        per_prompt_top_means = []
        for prompt_rewards in self.stats.values():
            if isinstance(prompt_rewards, list):
                rewards = np.array(prompt_rewards)
            else:
                rewards = prompt_rewards

            if rewards.size == 0:
                continue

            if top_percentage == 100:
                per_prompt_top_means.append(np.mean(rewards))
                continue

            lower_bound_percentile = 100 - top_percentage
            threshold = np.percentile(rewards, lower_bound_percentile)

            top_rewards = rewards[rewards >= threshold]

            if top_rewards.size > 0:
                per_prompt_top_means.append(np.mean(top_rewards))

        if not per_prompt_top_means:
            return 0.0

        return np.mean(per_prompt_top_means)


def filter_top_bottom_k_gpu(data, unique_num, num_in_group, k=2):
    prompts = data["prompt_ids"]  # [N, L]
    advs = data["advantages"][:, 0]  # [N]

    # 1. Pure GPU grouping.
    # unique identifies unique rows in [N, 300].
    # inverse_indices has shape [N] and values 0..num_groups-1, indicating each row's unique group.
    unique_prompts, inverse_indices = torch.unique(prompts, dim=0, return_inverse=True)

    num_groups = unique_prompts.shape[0]
    keep_indices = []

    print(f"unique_num: {unique_num}, num_groups: {num_groups}")
    assert unique_num == num_groups, f"unique_num: {unique_num}, num_groups: {num_groups}"

    # 2. Iterate over each group; the group count is small, so a Python loop is acceptable.
    for group_idx in range(num_groups):
        # Create a mask for the current group.
        group_mask = inverse_indices == group_idx

        # Get the global indices for the current group.
        # nonzero returns [m, 1]; squeeze converts it to [m].
        global_indices = torch.nonzero(group_mask).squeeze(1)

        current_count = global_indices.numel()

        assert current_count == num_in_group, f"current_count: {current_count}, num_in_group: {num_in_group}"

        # Get the corresponding advantages.
        group_advs = advs[global_indices]

        # Check whether there are enough samples.
        if group_advs.shape[0] < 2 * k:
            # If there are not enough samples, the policy could skip or error; this implementation errors.
            assert False, f"group_advs.shape[0]: {group_advs.shape[0]}, k: {k}"

        # 3. Sort and select Top/Bottom.
        # argsort returns indices relative to the group.
        sorted_relative_indices = torch.argsort(group_advs)

        # Map back to global indices.
        sorted_global_indices = global_indices[sorted_relative_indices]

        # Select the smallest k entries (Bottom).
        keep_indices.append(sorted_global_indices[:k])
        # Select the largest k entries (Top).
        keep_indices.append(sorted_global_indices[-k:])

    # 4. Concatenate indices and slice.
    if len(keep_indices) > 0:
        final_indices = torch.cat(keep_indices)
        # Optional: preserve the original batch order.
        final_indices, _ = torch.sort(final_indices)
    else:
        # Edge case: no group satisfies the condition.
        final_indices = torch.tensor([], device=prompts.device, dtype=torch.long)

    # 5. Rebuild the dictionary.
    new_dict = {}
    n_total = prompts.shape[0]
    for key, val in data.items():
        if torch.is_tensor(val) and val.shape[0] == n_total:
            new_dict[key] = val[final_indices]
        else:
            new_dict[key] = val

    return new_dict


def main():
    tracker = PerPromptStatTracker()
    prompts = ["a", "b", "a", "c", "b", "a"]
    rewards = [1, 2, 3, 4, 5, 6]
    advantages = tracker.update(prompts, rewards)
    print("Advantages:", advantages)
    avg_group_size, history_prompts = tracker.get_stats()
    print("Average Group Size:", avg_group_size)
    print("History Prompts:", history_prompts)
    tracker.clear()
    print("Stats after clear:", tracker.stats)


if __name__ == "__main__":
    main()