Delete app.py.py

app.py.py

@@ -1,484 +0,0 @@
-import copy
-import torch
-import math
-import torch.nn as nn
-from torch.nn.parameter import Parameter
-import random
-import numpy as np
-from load_weights import load_weight
-from sklearn.model_selection import train_test_split
-from transformers import GPT2TokenizerFast
-import pandas as pd
-from torch.utils.data import Dataset, DataLoader
-from transformers import AdamW, get_linear_schedule_with_warmup
-torch.manual_seed(42)
-import nltk
-nltk.download('punkt')
-
-from transformers import GPT2Tokenizer
-from torch.utils.data import Dataset, DataLoader, random_split, RandomSampler, SequentialSampler
-import datetime
-import time
-import os
-os.environ["CUDA_LAUNCH_BLOCKING"] = "1"
-from tqdm import trange
-import gradio as gr
-import re
-
-
-
-
-def gelu(x):
-    return 0.5 * x * (1 + torch.tanh(math.sqrt(2 / math.pi) * (x + 0.044715 * torch.pow(x, 3))))
-
-class Conv1D(nn.Module):
-    def __init__(self, nf, nx):
-        super(Conv1D, self).__init__()
-        self.nf = nf
-        w = torch.empty(nx, nf)
-        nn.init.normal_(w, std=0.02)
-        self.weight = Parameter(w)
-        self.bias = Parameter(torch.zeros(nf))
-
-    def forward(self, x):
-        size_out = x.size()[:-1] + (self.nf,)
-        x = torch.addmm(self.bias, x.view(-1, x.size(-1)), self.weight)
-        x = x.view(*size_out)
-        return x
-    
-class LayerNorm(nn.Module):
-    def __init__(self, hidden_size, eps=1e-12):
-        """Construct a layernorm module in the TF style (epsilon inside the square root).
-        """
-        super(LayerNorm, self).__init__()
-        self.weight = nn.Parameter(torch.ones(hidden_size))
-        self.bias = nn.Parameter(torch.zeros(hidden_size))
-        self.variance_epsilon = eps
-
-    def forward(self, x):
-        u = x.mean(-1, keepdim=True)
-        s = (x - u).pow(2).mean(-1, keepdim=True)
-        x = (x - u) / torch.sqrt(s + self.variance_epsilon)
-        return self.weight * x + self.bias
-    
-
-
-class Attention(nn.Module):
-    def __init__(self, nx, n_ctx, config, scale=False):
-        super(Attention, self).__init__()
-        n_state = nx  # in Attention: n_state=768 (nx=n_embd)
-        # [switch nx => n_state from Block to Attention to keep identical to TF implem]
-        assert n_state % config.n_head == 0
-        self.register_buffer("bias", torch.tril(torch.ones(n_ctx, n_ctx)).view(1, 1, n_ctx, n_ctx))
-        self.n_head = config.n_head
-        self.split_size = n_state
-        self.scale = scale
-        self.c_attn = Conv1D(n_state * 3, nx)
-        self.c_proj = Conv1D(n_state, nx)
-
-    def _attn(self, q, k, v):
-        w = torch.matmul(q, k)
-        if self.scale:
-            w = w / math.sqrt(v.size(-1))
-        nd, ns = w.size(-2), w.size(-1)
-        b = self.bias[:, :, ns-nd:ns, :ns]
-        w = w * b - 1e10 * (1 - b)
-        w = nn.Softmax(dim=-1)(w)
-        return torch.matmul(w, v)
-
-    def merge_heads(self, x):
-        x = x.permute(0, 2, 1, 3).contiguous()
-        new_x_shape = x.size()[:-2] + (x.size(-2) * x.size(-1),)
-        return x.view(*new_x_shape)  # in Tensorflow implem: fct merge_states
-
-    def split_heads(self, x, k=False):
-        new_x_shape = x.size()[:-1] + (self.n_head, x.size(-1) // self.n_head)
-        x = x.view(*new_x_shape)  # in Tensorflow implem: fct split_states
-        if k:
-            return x.permute(0, 2, 3, 1)  # (batch, head, head_features, seq_length)
-        else:
-            return x.permute(0, 2, 1, 3)  # (batch, head, seq_length, head_features)
-
-    def forward(self, x, layer_past=None):
-        x = self.c_attn(x)
-        query, key, value = x.split(self.split_size, dim=2)
-        query = self.split_heads(query)
-        key = self.split_heads(key, k=True)
-        value = self.split_heads(value)
-        if layer_past is not None:
-            past_key, past_value = layer_past[0].transpose(-2, -1), layer_past[1]  # transpose back cf below
-            key = torch.cat((past_key, key), dim=-1)
-            value = torch.cat((past_value, value), dim=-2)
-        present = torch.stack((key.transpose(-2, -1), value))  # transpose to have same shapes for stacking
-        a = self._attn(query, key, value)
-        a = self.merge_heads(a)
-        a = self.c_proj(a)
-        return a, present
-    
-
-class MLP(nn.Module):
-    def __init__(self, n_state, config):  # in MLP: n_state=3072 (4 * n_embd)
-        super(MLP, self).__init__()
-        nx = config.n_embd
-        self.c_fc = Conv1D(n_state, nx)
-        self.c_proj = Conv1D(nx, n_state)
-        self.act = gelu
-
-    def forward(self, x):
-        h = self.act(self.c_fc(x))
-        h2 = self.c_proj(h)
-        return h2
-    
-
-class Block(nn.Module):
-    def __init__(self, n_ctx, config, scale=False):
-        super(Block, self).__init__()
-        nx = config.n_embd
-        self.ln_1 = LayerNorm(nx, eps=config.layer_norm_epsilon)
-        self.attn = Attention(nx, n_ctx, config, scale)
-        self.ln_2 = LayerNorm(nx, eps=config.layer_norm_epsilon)
-        self.mlp = MLP(4 * nx, config)
-
-    def forward(self, x, layer_past=None):
-        a, present = self.attn(self.ln_1(x), layer_past=layer_past)
-        x = x + a
-        m = self.mlp(self.ln_2(x))
-        x = x + m
-        return x, present
-    
-
-
-class GPT2Model(nn.Module):
-    def __init__(self, config):
-        super(GPT2Model, self).__init__()
-        self.n_layer = config.n_layer
-        self.n_embd = config.n_embd
-        self.n_vocab = config.vocab_size
-
-        self.wte = nn.Embedding(config.vocab_size, config.n_embd)
-        self.wpe = nn.Embedding(config.n_positions, config.n_embd)
-        block = Block(config.n_ctx, config, scale=True)
-        self.h = nn.ModuleList([copy.deepcopy(block) for _ in range(config.n_layer)])
-        self.ln_f = LayerNorm(config.n_embd, eps=config.layer_norm_epsilon)
-
-    def set_embeddings_weights(self, model_embeddings_weights):
-        embed_shape = model_embeddings_weights.shape
-        self.decoder = nn.Linear(embed_shape[1], embed_shape[0], bias=False)
-        self.decoder.weight = model_embeddings_weights  # Tied weights
-
-    
-
-    def forward(self, input_ids, position_ids=None, token_type_ids=None, past=None):
-
-        if (input_ids >= self.n_vocab).any():
-            raise ValueError(f"Invalid token ID found in input_ids: {input_ids}")
-
-        # print(f"input_ids: {input_ids}")  # Debugging statement
-        # print(f"Max input_id: {input_ids.max().item()}")  # Debugging statement
-        # print(f"Min input_id: {input_ids.min().item()}")  # Debugging statement
-
-        if past is None:
-            past_length = 0
-            past = [None] * len(self.h)
-        else:
-            past_length = past[0][0].size(-2)
-        if position_ids is None:
-            position_ids = torch.arange(past_length, input_ids.size(-1) + past_length, dtype=torch.long,
-                                        device=input_ids.device)
-            position_ids = position_ids.unsqueeze(0).expand_as(input_ids)
-
-        input_shape = input_ids.size()
-        input_ids = input_ids.view(-1, input_ids.size(-1))
-        position_ids = position_ids.view(-1, position_ids.size(-1))
-
-        inputs_embeds = self.wte(input_ids)
-        position_embeds = self.wpe(position_ids)
-
-        # print(f"inputs_embeds shape: {inputs_embeds.shape}")
-        # print(f"position_embeds shape: {position_embeds.shape}")
-
-
-        if token_type_ids is not None:
-            token_type_ids = token_type_ids.view(-1, token_type_ids.size(-1))
-            token_type_embeds = self.wte(token_type_ids)
-        else:
-            token_type_embeds = 0
-        hidden_states = inputs_embeds + position_embeds + token_type_embeds
-        presents = []
-        for block, layer_past in zip(self.h, past):
-            hidden_states, present = block(hidden_states, layer_past)
-            presents.append(present)
-        hidden_states = self.ln_f(hidden_states)
-        output_shape = input_shape + (hidden_states.size(-1),)
-        return hidden_states.view(*output_shape), presents
-
-class GPT2LMHead(nn.Module):
-    def __init__(self, model_embeddings_weights, config):
-        super(GPT2LMHead, self).__init__()
-        self.n_embd = config.n_embd
-        self.set_embeddings_weights(model_embeddings_weights)
-
-    def set_embeddings_weights(self, model_embeddings_weights):
-        embed_shape = model_embeddings_weights.shape
-        self.decoder = nn.Linear(embed_shape[1], embed_shape[0], bias=False)
-        self.decoder.weight = model_embeddings_weights  # Tied weights
-
-    def forward(self, hidden_state):
-        # Truncated Language modeling logits (we remove the last token)
-        # h_trunc = h[:, :-1].contiguous().view(-1, self.n_embd)
-        lm_logits = self.decoder(hidden_state)
-        return lm_logits
-    
-import torch.nn.functional as F
-
-def top_k_top_p_filtering(logits, top_k=0, top_p=0.0, filter_value=-float('Inf')):
-        """ Filter a distribution of logits using top-k and/or nucleus (top-p) filtering
-            Args:
-                logits: logits distribution shape (batch size, vocabulary size)
-                top_k > 0: keep only top k tokens with highest probability (top-k filtering).
-                top_p > 0.0: keep the top tokens with cumulative probability >= top_p (nucleus filtering).
-                filter_value: value to replace filtered logits.
-        """
-        assert logits.dim() == 2  # batch size x vocabulary size
-        top_k = min(top_k, logits.size(-1))  # Safety check
-        if top_k > 0:
-            # Remove all tokens with a probability less than the last token of the top-k
-            indices_to_remove = logits < torch.topk(logits, top_k)[0][..., -1, None]
-            logits[indices_to_remove] = filter_value
-
-        if top_p > 0.0:
-            sorted_logits, sorted_indices = torch.sort(logits, descending=True)
-            cumulative_probs = torch.cumsum(F.softmax(sorted_logits, dim=-1), dim=-1)
-
-            # Remove tokens with cumulative probability above the threshold
-            sorted_indices_to_remove = cumulative_probs > top_p
-            # Shift the indices to the right to keep also the first token above the threshold
-            sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
-            sorted_indices_to_remove[..., 0] = 0
-
-            indices_to_remove = sorted_indices[sorted_indices_to_remove]
-            logits[indices_to_remove] = filter_value
-        return logits
-
-
-class GPT2LMHeadModel(nn.Module):
-    def __init__(self, config):
-        super(GPT2LMHeadModel, self).__init__()
-        self.transformer = GPT2Model(config)
-        self.lm_head = GPT2LMHead(self.transformer.wte.weight, config)
-
-    def set_tied(self):
-        """ Make sure we are sharing the embeddings
-        """
-        self.lm_head.set_embeddings_weights(self.transformer.wte.weight)
-
-    def forward(self, input_ids, position_ids=None, token_type_ids=None, lm_labels=None, past=None):
-        hidden_states, presents = self.transformer(input_ids, position_ids, token_type_ids, past)
-        lm_logits = self.lm_head(hidden_states)
-
-        outputs = (lm_logits,presents)
-
-        if lm_labels is not None:
-            shift_logits = lm_logits[..., :-1, :].contiguous()
-            shift_labels = lm_labels[..., 1:].contiguous()
-            loss_fct = nn.CrossEntropyLoss()
-            loss = loss_fct(shift_logits.view(-1, shift_logits.size(-1)), shift_labels.view(-1))
-            outputs = (loss,) + outputs
-        return outputs
-    
-    import torch.nn.functional as F
-
-    
-    
-    def generate(
-        self, input_ids, max_length, temperature=1.0, top_k=0, top_p=0.9, repetition_penalty=1.0, device='cuda'
-    ):
-        self.eval()
-        input_ids = input_ids.to(device)
-        batch_size = input_ids.shape[0]
-        past = None
-
-        generated = input_ids
-        with torch.no_grad():
-            for _ in range(max_length):
-                outputs = self(input_ids, past=past)
-                next_token_logits = outputs[0][:, -1, :]
-                past = outputs[1]
-
-                for i in range(batch_size):
-                    for token_id in set(generated[i].tolist()):
-                        next_token_logits[i, token_id] /= repetition_penalty
-
-                next_token_logits = next_token_logits / temperature
-                filtered_logits = top_k_top_p_filtering(next_token_logits, top_k=top_k, top_p=top_p)
-                next_token = torch.multinomial(F.softmax(filtered_logits, dim=-1), num_samples=1)
-                generated = torch.cat((generated, next_token), dim=1)
-
-                if (next_token == self.config.eos_token_id).all():
-                    break
-
-                input_ids = next_token
-
-        return generated
-        
-
-class GPT2Config(object):
-    def __init__(
-            self,
-            vocab_size_or_config_json_file=50257,
-            n_positions=1024,
-            n_ctx=1024,
-            n_embd=768,
-            n_layer=12,
-            n_head=12,
-            layer_norm_epsilon=1e-5,
-            initializer_range=0.02,
-    ):
-        self.vocab_size = vocab_size_or_config_json_file
-        self.n_ctx = n_ctx
-        self.n_positions = n_positions
-        self.n_embd = n_embd
-        self.n_layer = n_layer
-        self.n_head = n_head
-        self.layer_norm_epsilon = layer_norm_epsilon
-        self.initializer_range = initializer_range
-
-
-
-device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-config = GPT2Config()
-model = GPT2LMHeadModel(config)
-state_dict = torch.load(r'C:\vision_model\gpt-2-Pytorch\test\gpt_today\weights\epoch_1.pth', map_location='cpu' if not torch.cuda.is_available() else None)
-model = load_weight(model, state_dict)
-model.to(device)
-print(model)
-model.eval()
-
-tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
-tokenizer.pad_token = tokenizer.eos_token
-
-
-
-def top_k_top_p_filtering(logits, top_k=0, top_p=0.0, filter_value=-float('Inf')):
-    """ Filter a distribution of logits using top-k and/or nucleus (top-p) filtering
-        Args:
-            logits: logits distribution shape (batch size x vocabulary size)
-            top_k > 0: keep only top k tokens with highest probability (top-k filtering).
-            top_p > 0.0: keep the top tokens with cumulative probability >= top_p (nucleus filtering).
-                Nucleus filtering is described in Holtzman et al. (http://arxiv.org/abs/1904.09751)
-    """
-    assert logits.dim() == 2, "Expected logits dimension to be 2 (batch size x vocabulary size)"
-    top_k = min(top_k, logits.size(-1))  # Safety check
-    if top_k > 0:
-        # Remove all tokens with a probability less than the last token of the top-k
-        indices_to_remove = logits < torch.topk(logits, top_k)[0][..., -1, None]
-        logits[indices_to_remove] = filter_value
-
-    if top_p > 0.0:
-        sorted_logits, sorted_indices = torch.sort(logits, descending=True)
-        cumulative_probs = torch.cumsum(nn.Softmax(dim=-1)(sorted_logits), dim=-1)
-
-        # Remove tokens with cumulative probability above the threshold
-        sorted_indices_to_remove = cumulative_probs > top_p
-        # Shift the indices to the right to keep also the first token above the threshold
-        sorted_indices_to_remove[..., 1:] = sorted_indices_to_remove[..., :-1].clone()
-        sorted_indices_to_remove[..., 0] = 0
-
-        # Ensure that the dimensions match
-        if sorted_indices_to_remove.size() != sorted_indices.size():
-            raise ValueError(f"Size mismatch: {sorted_indices_to_remove.size()} vs {sorted_indices.size()}")
-
-        indices_to_remove = sorted_indices[sorted_indices_to_remove]
-
-        # Expand dimensions to match logits tensor and use scatter_
-        for batch_idx in range(logits.size(0)):
-            logits[batch_idx, indices_to_remove[batch_idx]] = filter_value
-            
-    return logits
-
-# prompt_text = "What is the classical conceptualisation of oxidation and reduction in redox reactions?"
-# prompt = f"\n<|startoftext|>[WP] {prompt_text} \n[RESPONSE]"
-# input_ids = tokenizer.encode(prompt, return_tensors='pt').to(device)
-
-
-# max_length = 50
-# temperature = 0.7
-# top_k = 50
-# top_p = 0.95
-# repetition_penalty = 1.0
-
-# with torch.no_grad():
-#     for _ in range(max_length):
-#         outputs = model(input_ids)
-#         logits = outputs[0]
-#         next_token_logits = logits[:, -1, :] / temperature
-
-#         # Apply repetition penalty
-#         for i in range(input_ids.size(0)):
-#             for token_id in set(input_ids[i].tolist()):
-#                 next_token_logits[0, token_id] /= repetition_penalty
-
-#         # Filter logits using top-k and/or top-p filtering
-#         filtered_logits = top_k_top_p_filtering(next_token_logits, top_k=top_k, top_p=top_p)
-#         next_token = torch.multinomial(F.softmax(filtered_logits, dim=-1), num_samples=1)
-#         input_ids = torch.cat([input_ids, next_token], dim=-1).to(device)
-
-
-# import re
-# # generated_text = tokenizer.decode(input_ids[0], skip_special_tokens=True)
-# # wp_responses = re.split(r"\[WP\].*?\n|\[RESPONSE\]", generated_text)[1:]
-# print(input_ids[0])
-
-# generated_text = tokenizer.decode(input_ids[0], skip_special_tokens=True)
-# wp_responses = re.split(r"\[WP\].*?\n|\[RESPONSE\]", generated_text)[1:]
-# print(wp_responses)
-
-
-# Define the generation function
-def generate_text(prompt_text, max_length=50, temperature=0.7, top_k=50, top_p=0.95, repetition_penalty=1.0):
-    prompt = f"\n[WP] {prompt_text} \n[RESPONSE]"
-    input_ids = tokenizer.encode(prompt, return_tensors='pt').to(device)
-
-    with torch.no_grad():
-        for _ in range(max_length):
-            outputs = model(input_ids)
-            logits = outputs[0]
-            next_token_logits = logits[:, -1, :] / temperature
-
-            # Apply repetition penalty
-            for i in range(input_ids.size(0)):
-                for token_id in set(input_ids[i].tolist()):
-                    next_token_logits[0, token_id] /= repetition_penalty
-
-            # Filter logits using top-k and/or top-p filtering
-            filtered_logits = top_k_top_p_filtering(next_token_logits, top_k=top_k, top_p=top_p)
-            next_token = torch.multinomial(F.softmax(filtered_logits, dim=-1), num_samples=1)
-            input_ids = torch.cat([input_ids, next_token], dim=-1).to(device)
-
-    generated_text = tokenizer.decode(input_ids[0], skip_special_tokens=True)
-    wp_responses = re.split(r"\[WP\].*?\n|\[RESPONSE\]", generated_text)[1:]
-    return wp_responses[1]
-
-# Define the Gradio interface using Blocks
-with gr.Blocks() as demo:
-    with gr.Row():
-        gr.Markdown("<h1 style='text-align: center'>GPT-2 Text Generator</h1>")
-    with gr.Row():
-        with gr.Column():
-            prompt = gr.Textbox(lines=2, placeholder="Enter prompt here...", label="Prompt")
-            max_length = gr.Slider(minimum=10, maximum=100, step=1, value=50, label="Max Length")
-            temperature = gr.Slider(minimum=0.1, maximum=1.0, step=0.1, value=0.7, label="Temperature")
-            top_k = gr.Slider(minimum=0, maximum=100, step=1, value=50, label="Top K")
-            top_p = gr.Slider(minimum=0.0, maximum=1.0, step=0.05, value=0.95, label="Top P")
-            repetition_penalty = gr.Slider(minimum=1.0, maximum=2.0, step=0.1, value=1.0, label="Repetition Penalty")
-            generate_button = gr.Button("Generate")
-        with gr.Column():
-            output_text = gr.Textbox(lines=20, label="Generated Text")
-
-    generate_button.click(
-        fn=generate_text,
-        inputs=[prompt, max_length, temperature, top_k, top_p, repetition_penalty],
-        outputs=output_text
-    )
-
-demo.launch()