# Implementation loosely based on https://github.com/tensorflow/tensor2tensor/blob/bafdc1b67730430d38d6ab802cbd51f9d053ba2e/tensor2tensor/utils/beam_search.py#L554 import requests import time from datetime import datetime, timedelta from typing import Optional, Literal import torch import torch.nn as nn from transformers import LlamaTokenizer from superposed.llama.utils import * from superposed.ngrams.ngram_models import NGram INF = 1. * 1e7 # Test by scaling # beams & verify work class Superpose(nn.Module): def __init__(self, initial_tokens, tokenizer, vocab_size, smoothing=Optional[Literal["geom", "all"]], alpha = None, verbose = False, i_weights = None, i_length = None, ngrams = None, sample_beams = False, sample_tokens = False, get_time = False, penalty = 200): # default no effect """ Initialize a beam search class. Args: initial_tokens (torch.Tensor): Initial tokens n_prompts (int): Number of prompts tokenizer (Tokenizer): Llama tokenizer vocab_size (int): Total vocab size smoothing (str): Smoothing method ("geom" for default, "all" for only ngram, None for no ngram) ngram_length (int): N gram length to consider alpha (float): Alpha parameter debug (bool): Whether to print information """ super().__init__() # primary parameters self.n_prompts, self.n_drafts, _ = initial_tokens.shape self.tokenizer = tokenizer self.vocab_size = vocab_size self.alive_seq = initial_tokens self.fin_seq = initial_tokens self.smoothing = smoothing self.alive_log_probs = torch.zeros(self.n_prompts, self.n_drafts) self.fin_log_probs = torch.full((self.n_prompts, self.n_drafts), float("-inf")) self.alpha = alpha self.verbose = verbose self.penalty = penalty # devices self.cpu = torch.device('cpu') self.gpu = torch.device('cuda') # Interpolation length and weights self.interpolation_weights = i_weights self.i_length = i_length # N-grams self.bigram = ngrams[0] if len(ngrams) >= 1 else None self.trigram = ngrams[1] if len(ngrams) >= 2 else None self.fourgram = ngrams[2] if len(ngrams) >= 3 else None self.fivegram = ngrams[3] if len(ngrams) >= 4 else None self.sixgram = ngrams[4] if len(ngrams) >= 5 else None self.sevengram = ngrams[5] if len(ngrams) >= 6 else None # Timing self.get_time = get_time self.lookup_time = None def forward(self, probs, still_prompt, is_first, cur_pos, n_token_sample): """ Apply beam decoding to update generations. Args: probs (torch.Tensor): Next token probability distribution still_prompt (torch.Tensor): Flags of prompts that should not generate yet (n_prompts, ) is_first (torch.Tensor): Flags of prompts that are on their first generation (n_prompts, ) cur_pos (int): Current generation position n_token_sample (int): Number of tokens from model distribution to use Return: if standard beam search: attention_change_ids (torch.Tensor): New indices in kv cache (n_prompts, n_drafts) if mixed: token_weights (torch.Tensor): Mixing weights (n_prompts, vocab_size) """ # Adjust input probabilities probs = self.get_top_k(probs, 32000, n_token_sample) reshaped_probs = probs.reshape(self.n_prompts, 1, -1) reshaped_probs = reshaped_probs.repeat(1, self.n_drafts, 1) # Ngram smoothing if self.smoothing is not None: if self.smoothing == "geom": ngram_probs = self.ngram_probs(self.alive_seq, cur_pos, probs=probs) # Make mask and normalize prob_mask = reshaped_probs != 0 ngram_probs *= prob_mask # Calculate logprobs and interpolate distributions llm_log_probs = torch.log(reshaped_probs) ngram_log_probs = torch.log(ngram_probs) log_probs = (1 - self.alpha) * llm_log_probs + self.alpha * ngram_log_probs # Apply penalty to drafts where no interpolation occurred is_all_inf = (log_probs != float("-inf")).sum(dim=-1, keepdims=True) == 0 log_probs = torch.where(is_all_inf, (1 - self.alpha) * llm_log_probs - self.penalty, log_probs) elif self.smoothing == "all": ngram_probs = self.ngram_probs(self.alive_seq, cur_pos, probs=None) log_probs = torch.log(ngram_probs) else: log_probs = torch.log(reshaped_probs) curr_log_probs = self.alive_log_probs.unsqueeze(dim=2) + log_probs # [n_prompts, n_drafts, vocab_size] # Warning if nan if (torch.any(torch.isnan(curr_log_probs)).item()): raise RuntimeWarning("nan in sequence log probs", file=self.output_file) # Potential Sequences flat_curr_log_probs = curr_log_probs.reshape(-1, self.vocab_size*self.n_drafts) topk_log_probs, topk_idx = torch.topk(flat_curr_log_probs, 2 * self.n_drafts, dim=-1) topk_beam_id = topk_idx // self.vocab_size # [n_prompts, 2 * n_drafts] topk_idx = topk_idx % self.vocab_size # [n_prompts, 2 * n_drafts] # First timestep uses top-k next tokens is_first_idx = is_first.nonzero(as_tuple=True)[0] if len(is_first_idx) != 0: first_time_log_probs = log_probs[is_first_idx][:, 0, :].squeeze(dim=1) first_time_log_probs, first_time_topk_idx = torch.topk(first_time_log_probs, 2 * self.n_drafts, dim=1) topk_idx[is_first_idx] = first_time_topk_idx topk_log_probs[is_first_idx] = self.alive_log_probs[is_first_idx, 0].unsqueeze(dim=1) + first_time_log_probs # New sequences topk_seq = torch.take_along_dim(self.alive_seq, topk_beam_id.unsqueeze(2), dim=1) # [n_prompts, 2 * n_drafts, vocab_size] topk_seq[:, :, cur_pos] = topk_idx topk_finished = topk_idx == self.tokenizer.eos_id # Only update sequences for those that have begun generating new_alive_seq, new_alive_log_probs = self.grow_alive(topk_seq, topk_log_probs, topk_finished) new_fin_seq, new_fin_log_probs = self.grow_fin(topk_seq, topk_log_probs, topk_finished) still_prompt_probs = still_prompt.reshape(-1, 1) still_prompt_seqs = still_prompt.reshape(-1, 1, 1) self.alive_seq = torch.where(still_prompt_seqs, self.alive_seq, new_alive_seq) self.alive_log_probs = torch.where(still_prompt_probs, self.alive_log_probs, new_alive_log_probs) self.fin_seq = torch.where(still_prompt_seqs, self.fin_seq, new_fin_seq) self.fin_log_probs = torch.where(still_prompt_probs, self.fin_log_probs, new_fin_log_probs) # Create superposition matrix and return it topk_idx = self.alive_seq[:, :, cur_pos].reshape(self.n_prompts, -1) token_weights = self.superposition_matrix(topk_idx) return token_weights def grow_alive(self, topk_seq, topk_log_probs, topk_finished): """ Extend running generations. Args: topk_seq (torch.Tensor): Top k sequences (n_prompts, 2 * n_drafts, vocab_size) topk_log_probs (torch.Tensor): Log probabilities (n_prompts, 2 * n_drafts) topk_finished (torch.Tensor): Whether a sequence is finished (n_prompts, 2 * n_drafts) Returns: new_alive_seq, new_alive_log_probs """ topk_log_probs = topk_log_probs + topk_finished * -INF new_alive_log_probs, new_alive_idx = torch.topk(topk_log_probs, self.n_drafts, dim=1) new_alive_seq = torch.take_along_dim(topk_seq, new_alive_idx.unsqueeze(2), dim=1) return new_alive_seq, new_alive_log_probs def grow_fin(self, topk_seq, topk_log_probs, topk_finished): """ Update stopped generations. Args: topk_seq (torch.Tensor): Top k sequences (n_prompts, 2 * n_drafts, vocab_size) topk_log_probs (torch.Tensor): Log probabilities (n_prompts, 2 * n_drafts) topk_finished (torch.Tensor): Whether a sequence is finished (n_prompts, 2 * n_drafts) Returns: new_fin_seq, new_fin_log_probs """ topk_log_probs = topk_log_probs + ~topk_finished * -INF new_fin_seq = torch.cat([self.fin_seq, topk_seq], dim=1) new_fin_log_probs = torch.cat([self.fin_log_probs, topk_log_probs], dim=1) new_fin_log_probs, new_fin_idx = torch.topk(new_fin_log_probs, self.n_drafts, dim=1) new_fin_seq = torch.take_along_dim(new_fin_seq, new_fin_idx.unsqueeze(2), dim=1) return new_fin_seq, new_fin_log_probs def get_top_k(self, probs, m, k): """ Zero out all but top-k tokens in a probability distribution. Args: probs (torch.Tensor): Probability distribution tensor. m (float): Number of tokens to consider (only relevant when sampling). k (int): Number of tokens to sample/keep. Returns: torch.Tensor: New probability distribution based on renormalized probabilities. """ n_prompts, _ = probs.shape probs_sort, probs_idx = torch.sort(probs, dim=-1, descending=True) top_k_mask = torch.arange(probs.shape[-1]) top_k_mask = top_k_mask.expand(probs.shape[0], -1) top_k_mask = top_k_mask >= m # Set to 1 past k elements probs_sort[top_k_mask] = 0.0 # Zero wherever mask = 1 probs_sort.div_(probs_sort.sum(dim=-1, keepdim=True)) next_token = torch.gather(probs_idx, -1, torch.topk(probs_sort, k, dim=-1)[1]) # Set all other probs to 0 new_probs_map = torch.zeros(probs.shape).bool() new_probs_map[torch.repeat_interleave(torch.arange(n_prompts), k), torch.flatten(next_token)] = True new_probs = torch.where(new_probs_map, probs, 0) # Renormalize new_probs.div_(new_probs.sum(dim=-1, keepdim=True)) return new_probs def superposition_matrix(self, tokens): """ Create superposition matrix based on provided tokens. Args: tokens (torch.Tensor): Tokens to mix (n_prompts, n_drafts) Returns: SUperposition matrix """ # Create superposition matrix mixing_matrix = torch.zeros(self.n_prompts, self.vocab_size) # Convert draft log probs to probabilities weightings = log_prob_to_prob(self.alive_log_probs) # Update probabilities in superposition matrix with draft probabilities for p_idx in range(self.n_prompts): for d_idx in range(self.n_drafts): tok_idx = tokens[p_idx][d_idx] mixing_matrix[p_idx][tok_idx] += weightings[p_idx][d_idx] # Renormalize mixing_matrix.div_(mixing_matrix.sum(dim=-1, keepdims=True)) return mixing_matrix def ngram_probs(self, alive_seq, cur_pos, probs): """ Calculate and return next token distribution using ngram models. Args: alive_seq (torch.Tensor): Current drafts (n_prompts, n_drafts, seqlen) cur_pos (int): Current timestep probs (torch.Tensor): Current next probability distribution from model (n_prompts, vocab_size). As described in the paper, only tokens w/nonzero probability in `prob` are considered for the ngram distribution. However, passing in `None` as `probs` will consider all tokens. Returns: Next token distribution for each draft (n_prompts, n_drafts, vocab_size) """ if self.get_time: # Start timer start_time = datetime.now() # Create distribution matrix next_token_probs = torch.zeros(self.n_prompts, self.n_drafts, 32000) if probs is not None: # Loop over all prefixes for p_idx in range(len(alive_seq)): # List of possible tokens for the prefix nz = torch.nonzero(probs[p_idx, :], as_tuple=True)[0].tolist() # Generate next token distribution for draft_idx in range(self.n_drafts): i_mask = torch.sum(torch.tensor(self.i_length) <= cur_pos) new_i_weights = self.interpolation_weights[:i_mask] new_i_length = self.i_length[:i_mask] # For each next token for nt in nz: # Calculate probability using ngram interpolation for i, weight in zip(new_i_length, new_i_weights): if cur_pos - i >= 0: key = tuple(alive_seq[p_idx, draft_idx, cur_pos-i:cur_pos].tolist()) if i == 1: prob = self.bigram.prob(key, nt) elif i == 2: prob = self.trigram.prob(key, nt) elif i == 3: prob = self.fourgram.prob(key, nt) elif i == 4: prob = self.fivegram.prob(key, nt) elif i == 5: prob = self.sixgram.prob(key, nt) elif i == 6: prob = self.sevengram.prob(key, nt) if prob >= 0: next_token_probs[p_idx, draft_idx, nt] += weight * prob else: for p_idx in range(len(alive_seq)): for draft_idx in range(self.n_drafts): i_mask = torch.sum(torch.tensor(self.i_length) <= cur_pos) new_i_weights = self.interpolation_weights[:i_mask] new_i_length = self.i_length[:i_mask] for i, weight in zip(new_i_length, new_i_weights): if cur_pos - i >= 0: key = tuple(alive_seq[p_idx, draft_idx, cur_pos-i:cur_pos].tolist()) if i == 1: ntd = self.bigram.ntd(key) elif i == 2: ntd = self.trigram.ntd(key) elif i == 3: ntd = self.fourgram.ntd(key) elif i == 4: ntd = self.fivegram.ntd(key) elif i == 5: ntd = self.sixgram.ntd(key) elif i == 6: ntd = self.sevengram.ntd(key) if ntd is not None: next_token_probs[p_idx, draft_idx, :] += weight * ntd if self.get_time: total_time = datetime.now() - start_time self.lookup_time = total_time if self.lookup_time is None else self.lookup_time + total_time return next_token_probs def return_results(self, prompt_len=None): """ Return generations and perplexities Args: prompt_len (int): Length of prompt in tokens. If is None, then ppl is not calculated. Returns: (self.alive_seq, alive_ppl), (self.fin_seq, fin_ppl) OR (self.alive_seq, alive_ppl), (self.fin_seq, fin_ppl), self.lookup_time """ # PPL alive_ppl = 0 fin_ppl = 0 if prompt_len is not None: alive_ppl = torch.exp(self.alive_log_probs / (-1 * (self.alive_seq.size(dim=-1)-prompt_len))) # Fin ppl fin_seq_lengths = (self.fin_seq != self.tokenizer.pad_id).sum(dim=-1) fin_ppl = torch.exp(self.fin_log_probs / (-1 * (fin_seq_lengths - prompt_len))) fin_ppl += ((fin_ppl == 0) * float("inf")) # print time if not self.get_time: return (self.alive_seq.to(torch.long), alive_ppl), (self.fin_seq.to(torch.long), fin_ppl) else: return (self.alive_seq.to(torch.long), alive_ppl), (self.fin_seq.to(torch.long), fin_ppl), self.lookup_time