Upload app.py

app.py

@@ -1,292 +1,45 @@
-import torch
-import torch.nn as nn
-import tiktoken
-import gradio as gr
-
-# ============== Model Classes ==============
-class PolarisAIMultiHeadAttention(nn.Module):
-    def __init__(self, d_in, d_out, context_length, dropout, num_heads, qkv_bias=False):
-        super().__init__()
-        assert d_out % num_heads == 0, "d_out must be divisible by num_heads"
-        self.d_out = d_out
-        self.num_heads = num_heads
-        self.head_dim = d_out // num_heads
-        self.W_query = nn.Linear(d_in, d_out, bias=qkv_bias)
-        self.W_key = nn.Linear(d_in, d_out, bias=qkv_bias)
-        self.W_value = nn.Linear(d_in, d_out, bias=qkv_bias)
-        self.W_output = nn.Linear(d_out, d_out, bias=qkv_bias)
-        self.dropout = nn.Dropout(dropout)
-        self.register_buffer('mask', torch.triu(torch.ones(context_length, context_length), diagonal=1))
-
-    def split_heads(self, x):
-        seq_len, d_out = x.shape
-        x = x.view(seq_len, self.num_heads, self.head_dim)
-        return x.transpose(0, 1)
-
-    def combine_heads(self, x):
-        num_heads, seq_len, head_dim = x.shape
-        x = x.transpose(0, 1)
-        return x.contiguous().view(seq_len, num_heads * head_dim)
-
-    def forward(self, x):
-        num_tokens, d_in = x.shape
-        allqueries = self.W_query(x)
-        allkeys = self.W_key(x)
-        allvalues = self.W_value(x)
-        queries_heads = self.split_heads(allqueries)
-        keys_heads = self.split_heads(allkeys)
-        values_heads = self.split_heads(allvalues)
-        attention_scores = queries_heads @ keys_heads.transpose(-2, -1)
-        masked = attention_scores.masked_fill(
-            self.mask.bool()[:num_tokens, :num_tokens], -torch.inf
-        )
-        attention_weights = torch.softmax(masked / self.head_dim**0.5, dim=-1)
-        dropout_attention_weights = self.dropout(attention_weights)
-        context_heads = dropout_attention_weights @ values_heads
-        context_combined = self.combine_heads(context_heads)
-        return self.W_output(context_combined)
-
-
-class PolarisAILayerNorm(nn.Module):
-    def __init__(self, emb_dim):
-        super().__init__()
-        self.eps = 1e-5
-        self.scale = nn.Parameter(torch.ones(emb_dim))
-        self.shift = nn.Parameter(torch.zeros(emb_dim))
-
-    def forward(self, x):
-        mean = x.mean(dim=-1, keepdim=True)
-        var = x.var(dim=-1, keepdim=True, unbiased=False)
-        norm_x = (x - mean) / torch.sqrt(var + self.eps)
-        return self.scale * norm_x + self.shift
-
-
-class PolarisAIGELUActivation(nn.Module):
-    def __init__(self):
-        super().__init__()
-
-    def forward(self, x):
-        return 0.5 * x * (1 + torch.tanh(
-            torch.sqrt(torch.tensor(2.0 / torch.pi)) *
-            (x + 0.044715 * torch.pow(x, 3))
-        ))
-
-
-class PolarisAIFeedForwardNetwork(nn.Module):
-    def __init__(self, cfg):
-        super().__init__()
-        self.layers = nn.Sequential(
-            nn.Linear(cfg["emb_dim"], 4 * cfg["emb_dim"]),
-            PolarisAIGELUActivation(),
-            nn.Linear(4 * cfg["emb_dim"], cfg["emb_dim"]),
-        )
-
-    def forward(self, x):
-        return self.layers(x)
-
-
-class PolarisAITransformerBlock(nn.Module):
-    def __init__(self, cfg):
-        super().__init__()
-        self.att = PolarisAIMultiHeadAttention(
-            d_in=cfg["emb_dim"], d_out=cfg["emb_dim"],
-            context_length=cfg["context_length"], num_heads=cfg["n_heads"],
-            dropout=cfg["drop_rate"], qkv_bias=cfg["qkv_bias"])
-        self.ff = PolarisAIFeedForwardNetwork(cfg)
-        self.norm1 = PolarisAILayerNorm(cfg["emb_dim"])
-        self.norm2 = PolarisAILayerNorm(cfg["emb_dim"])
-        self.drop_shortcut = nn.Dropout(cfg["drop_rate"])
-
-    def forward(self, x):
-        shortcut = x
-        x = self.norm1(x)
-        x = self.att(x)
-        x = self.drop_shortcut(x)
-        x = x + shortcut
-        shortcut = x
-        x = self.norm2(x)
-        x = self.ff(x)
-        x = self.drop_shortcut(x)
-        return x + shortcut
-
-
-class PolarisAIPlatformModel(nn.Module):
-    def __init__(self, cfg):
-        super().__init__()
-        self.tok_emb = nn.Embedding(cfg["vocab_size"], cfg["emb_dim"])
-        self.pos_emb = nn.Embedding(cfg["context_length"], cfg["emb_dim"])
-        self.drop_emb = nn.Dropout(cfg["drop_rate"])
-        self.trf_blocks = nn.Sequential(
-            *[PolarisAITransformerBlock(cfg) for _ in range(cfg["n_layers"])])
-        self.final_norm = PolarisAILayerNorm(cfg["emb_dim"])
-        self.out_head = nn.Linear(cfg["emb_dim"], cfg["vocab_size"], bias=False)
-        self.cfg = cfg
-
-    def forward(self, in_idx):
-        seq_len = in_idx.shape[0]
-        tok_embeds = self.tok_emb(in_idx)
-        pos_embeds = self.pos_emb(torch.arange(seq_len, device=in_idx.device))
-        x = tok_embeds + pos_embeds
-        x = self.drop_emb(x)
-        x = self.trf_blocks(x)
-        x = self.final_norm(x)
-        return self.out_head(x)
-
-
-# ============== Generation Functions ==============
-def generate_text_simple(model, idx, max_new_tokens, context_size):
-    for _ in range(max_new_tokens):
-        idx_cond = idx[-context_size:]
-        with torch.no_grad():
-            logits = model(idx_cond)
-        logits = logits[-1, :]
-        probas = torch.softmax(logits, dim=-1)
-        idx_next = torch.argmax(probas).unsqueeze(0)
-        idx = torch.cat((idx, idx_next), dim=0)
-    return idx
-
-
-def generate_text_with_temperature(model, idx, max_new_tokens, context_size, temperature=1.0, top_k=None):
-    for _ in range(max_new_tokens):
-        idx_cond = idx[-context_size:]
-        with torch.no_grad():
-            logits = model(idx_cond)
-        logits = logits[-1, :]
-        if temperature > 0:
-            logits = logits / temperature
-            if top_k is not None and top_k > 0:
-                top_k = min(top_k, logits.size(-1))
-                values, indices = torch.topk(logits, top_k)
-                logits = torch.full_like(logits, float('-inf'))
-                logits.scatter_(-1, indices, values)
-            probas = torch.softmax(logits, dim=-1)
-            idx_next = torch.multinomial(probas, num_samples=1)
-        else:
-            idx_next = torch.argmax(logits).unsqueeze(0)
-        idx = torch.cat((idx, idx_next), dim=0)
-    return idx
-
-
-# ============== Initialize Tokenizer ==============
-tokenizer = tiktoken.get_encoding("gpt2")
-
-
-# ============== Gradio Function ==============
-def generate_text_gradio(
-    input_text, 
-    max_new_tokens, 
-    temperature, 
-    top_k, 
-    seed, 
-    decoding_strategy,
-    vocab_size,
-    context_length,
-    emb_dim,
-    n_heads,
-    n_layers,
-    drop_rate,
-    qkv_bias
-):
-    if not input_text.strip():
-        return "Please enter some text to generate from.", ""
-    
-    # Validate emb_dim is divisible by n_heads
-    if emb_dim % n_heads != 0:
-        return f"Error: Embedding dimension ({emb_dim}) must be divisible by number of heads ({n_heads}).", ""
-    
-    # Build config from UI inputs
-    config = {
-        "vocab_size": int(vocab_size),
-        "context_length": int(context_length),
-        "emb_dim": int(emb_dim),
-        "n_heads": int(n_heads),
-        "n_layers": int(n_layers),
-        "drop_rate": float(drop_rate),
-        "qkv_bias": bool(qkv_bias)
-    }
-    
-    # Initialize model with user config
-    torch.manual_seed(int(seed))
-    model = PolarisAIPlatformModel(config)
-    model.eval()
-    
-    # Calculate model info
-    total_params = sum(p.numel() for p in model.parameters())
-    model_size_mb = total_params * 4 / (1024 * 1024)
-    model_info = f"Parameters: {total_params:,} | Size: {model_size_mb:.2f} MB"
-    
-    # Encode input
-    input_ids = torch.tensor(tokenizer.encode(input_text))
-    
-    # Generate
-    if decoding_strategy == "Greedy":
-        output_ids = generate_text_simple(model, input_ids, int(max_new_tokens), config["context_length"])
-    else:
-        output_ids = generate_text_with_temperature(
-            model, input_ids, int(max_new_tokens), 
-            config["context_length"], temperature, 
-            int(top_k) if top_k > 0 else None
-        )
-    
-    return tokenizer.decode(output_ids.tolist()), model_info
-
-
-# ============== Gradio Interface ==============
-with gr.Blocks(title="PolarisAI Platform",theme=gr.themes.Default(primary_hue='sky')) as PolarisAIPlatform:
-    
-    with gr.Row():
-        # Left Column - Input/Output
-        with gr.Column(scale=2):
-            input_text = gr.Textbox(
-                label="Input Text",
-                placeholder="Enter text here...",
-                lines=3,
-                value=""
-            )
-            generate_btn = gr.Button("Generate Text", variant="primary", size="lg")
-            output_text = gr.Textbox(label="Generated Output", lines=8, interactive=False)
-            model_info_text = gr.Textbox(label="Model Info", interactive=False)
-        
-        # Right Column - Parameters
-        with gr.Column(scale=1):
-            # Generation Parameters
-            decoding_strategy = gr.Radio(
-                ["Greedy", "Temperature Sampling"],
-                value="Greedy",
-                label="Decoding Strategy"
-            )
-            max_new_tokens = gr.Slider(1, 100, value=10, step=1, label="Max New Tokens")
-            temperature = gr.Slider(0.0, 2.0, value=1.0, step=0.1, label="Temperature")
-            top_k = gr.Slider(0, 100, value=0, step=1, label="Top-K (0=disabled)")
-            seed = gr.Number(value=123, label="Random Seed", precision=0)
-            
-            # Model Configuration Parameters
-            vocab_size = gr.Number(value=50257, label="Vocab Size", precision=0)
-            context_length = gr.Number(value=1024, label="Context Length", precision=0)
-            emb_dim = gr.Number(value=768, label="Embedding Dimension", precision=0)
-            n_heads = gr.Number(value=12, label="Number of Heads", precision=0)
-            n_layers = gr.Number(value=12, label="Number of Layers", precision=0)
-            drop_rate = gr.Slider(0.0, 0.5, value=0.1, step=0.01, label="Dropout Rate")
-            qkv_bias = gr.Checkbox(value=False, label="QKV Bias")
-    
-    # Connect button
-    generate_btn.click(
-        generate_text_gradio,
-        inputs=[
-            input_text, max_new_tokens, temperature, top_k, seed, decoding_strategy,
-            vocab_size, context_length, emb_dim, n_heads, n_layers, drop_rate, qkv_bias
-        ],
-        outputs=[output_text, model_info_text]
-    )
-    
-    # Submit on Enter
-    input_text.submit(
-        generate_text_gradio,
-        inputs=[
-            input_text, max_new_tokens, temperature, top_k, seed, decoding_strategy,
-            vocab_size, context_length, emb_dim, n_heads, n_layers, drop_rate, qkv_bias
-        ],
-        outputs=[output_text, model_info_text]
-    )
-
-PolarisAIPlatform.launch()
+_H='custom'
+_G='primary'
+_F='e.g., business, technology, sports, entertainment'
+_E='Custom Labels (for custom classification)'
+_D='Classification Type:'
+_C='sentiment'
+_B='Spam'
+_A='Sentiment'
+import os,gradio as gr
+from openai import OpenAI
+API_KEY=os.environ['API_KEY']
+client=OpenAI(base_url='https://openrouter.ai/api/v1',api_key=API_KEY)
+def classify_text(text,classification_type=_C,custom_labels=''):
+	"\n    Classify text using OpenRouter's GPT-OSS-20B model\n    ";E='content';D='role';B=classification_type;A=text
+	if not A.strip():return'Please enter some text to classify.'
+	if B==_A:C=f"Classify the sentiment of the following text as Positive, Negative, or Neutral. Only respond with one word: Positive, Negative, or Neutral.\n\nText: {A}"
+	elif B==_B:C=f"Classify whether the following text is Spam or Not Spam. Only respond with: Spam or Not Spam.\n\nText: {A}"
+	try:F=client.chat.completions.create(model='openai/gpt-oss-20b',messages=[{D:'system',E:'You are a text classification assistant. Provide concise, accurate classifications.'},{D:'user',E:C}],max_tokens=50,temperature=.1,extra_headers={'Authorization':f"Bearer {API_KEY}",'HTTP-Referer':'https://your-app-url.com','X-Title':''});G=F.choices[0].message.content.strip();return f"Classification Result: {G}"
+	except Exception as H:return f"Error: {str(H)}"
+def batch_classify(file,classification_type=_C,custom_labels=''):
+	'\n    Classify multiple texts from uploaded file\n    '
+	if file is None:return'Please upload a text file.'
+	try:
+		with open(file.name,'r',encoding='utf-8')as C:D=C.readlines()
+		B=[]
+		for(E,A)in enumerate(D[:10],1):
+			A=A.strip()
+			if A:F=classify_text(A,classification_type,custom_labels);B.append(f"{E}. **Text:** {A}\n   **Result:** {F}\n")
+		return'\n'.join(B)if B else'No text found in file.'
+	except Exception as G:return f"Error processing file: {str(G)}"
+with gr.Blocks(title='',theme=gr.themes.Default(primary_hue='sky'))as demo:
+	with gr.Tabs():
+		with gr.Tab('Single Text'):
+			with gr.Row():
+				with gr.Column(scale=2):text_input=gr.Textbox(label='',placeholder='Enter text to classify...',lines=4);classification_type=gr.Radio(choices=[_A,_B],value=_A,label=_D);custom_labels=gr.Textbox(label=_E,placeholder=_F,visible=False);classify_btn=gr.Button('Classify Text',variant=_G)
+				with gr.Column(scale=2):single_output=gr.Markdown(value='')
+			def toggle_custom_labels(choice):return gr.update(visible=choice==_H)
+			classification_type.change(toggle_custom_labels,inputs=[classification_type],outputs=[custom_labels]);classify_btn.click(classify_text,inputs=[text_input,classification_type,custom_labels],outputs=[single_output])
+		with gr.Tab('Batch Classification'):
+			with gr.Row():
+				with gr.Column(scale=2):gr.Markdown('Upload a text or csv file:');file_input=gr.File(label='Upload File',file_types=['.txt','.csv']);batch_classification_type=gr.Radio(choices=[_A,_B],value=_A,label=_D);batch_custom_labels=gr.Textbox(label=_E,placeholder=_F,visible=False);batch_classify_btn=gr.Button('🔍 Classify Batch',variant=_G)
+				with gr.Column(scale=2):batch_output=gr.Markdown(value='')
+			def toggle_batch_custom_labels(choice):return gr.update(visible=choice==_H)
+			batch_classification_type.change(toggle_batch_custom_labels,inputs=[batch_classification_type],outputs=[batch_custom_labels]);batch_classify_btn.click(batch_classify,inputs=[file_input,batch_classification_type,batch_custom_labels],outputs=[batch_output])
+if __name__=='__main__':demo.launch(server_name='0.0.0.0',server_port=7860,share=True,show_error=True)