Upload pipeline.py with huggingface_hub

pipeline.py

@@ -0,0 +1,106 @@
+from typing import Dict, List, Union
+import os
+import torch
+import torch.nn as nn
+import numpy as np
+from torch.nn import TransformerEncoder, TransformerEncoderLayer
+
+class PreTrainedPipeline():
+  
+    def __init__(self, path="./", device=(torch.device("cuda") if torch.cuda.is_available() else torch.device("cpu"))):
+        # IMPLEMENT_THIS
+        # Preload all the elements you are going to need at inference.
+        # For instance your model, processors, tokenizer that might be needed.
+        # This function is only called once, so do all the heavy processing I/O here"""
+        self.device = device
+        self.mean = np.load("train_mean.npy")
+        self.std = np.load("train_std.npy")
+        self.selected_keys = (
+            "sex", "has_photo", "has_mobile", "followers_count", "relation", "life_main", "people_main"
+        )
+        BATCH_SIZE = 1024
+        MAX_LR = 1e-4
+        EPOCHS = 50
+        D_MODEL = 512
+        NHEAD = 16
+        NUM_LAYERS = 32
+        DROPOUT = 0.2
+        self.model = self.TabularTransformer(
+            input_dim=7,
+            d_model=D_MODEL,
+            nhead=NHEAD,
+            num_layers=NUM_LAYERS,
+            dropout=DROPOUT
+        ).to(self.device)
+        self.model.load_state_dict(torch.load(os.path.join(path, "pytorch_model.bin"), weights_only=True))
+        self.model.eval()
+    def _convert_inputs(self, inputs):
+        data = inputs.get("data", {})
+        # Фильтрация ключей: оставляем только выбранные и существующие в данных
+        valid_keys = [key for key in self.selected_keys if key in data]
+        # Определение минимальной длины среди выбранных колонок
+        lengths = [len(data[key]) for key in valid_keys if key in data]
+        if not lengths:
+            return []
+        num_rows = min(lengths)
+        # Сборка списка строк (только с выбранными ключами)
+        return [
+            {key: data[key][i] for key in valid_keys} 
+            for i in range(num_rows)
+        ]
+    def _normalize(self, flist):
+      return (flist - self.mean) / (self.std + 1e-8)
+    def __call__(
+        self, inputs: Dict[str, Dict[str, List[Union[str, float]]]]
+    ) -> List[Union[str, float, bool]]:
+        """
+        Args:
+            inputs (:obj:`dict`):
+                a dictionary containing a key 'data' mapping to a dict in which
+                the values represent each column.
+        Return:
+            A :obj:`list` of floats or strings: The classification output for each row.
+        """
+        # IMPLEMENT_THIS
+        ppinputs = self._convert_inputs(inputs)
+        modout = list()
+        for i in ppinputs:
+          newstr = [i["sex"], i["has_photo"], i["has_mobile"], i["followers_count"], 
+              i["relation"], i["life_main"], i["people_main"]]
+          modell = torch.tensor(self._normalize(newstr), dtype=torch.float32).unsqueeze(0).to(self.device)
+          modout.append(self.model(modell).squeeze())
+        print([i.item() > 0 for i in torch.stack(modout)])
+          
+    class TabularTransformer(nn.Module):
+      def __init__(self, input_dim, d_model, nhead, num_layers, dropout):
+          super().__init__()
+          self.input_proj = nn.Linear(input_dim, d_model)
+
+          encoder_layers = TransformerEncoderLayer(
+              d_model=d_model,
+              nhead=nhead,
+              dim_feedforward=d_model*4,
+              dropout=dropout,
+              activation='gelu',
+              batch_first=True
+          )
+          self.transformer = TransformerEncoder(encoder_layers, num_layers)
+
+          self.head = nn.Sequential(
+              nn.LayerNorm(d_model),
+              nn.Linear(d_model, 1)
+          )
+
+          self._init_weights()
+
+      def _init_weights(self):
+          for module in self.modules():
+              if isinstance(module, nn.Linear):
+                  nn.init.xavier_normal_(module.weight)
+                  if module.bias is not None:
+                      nn.init.zeros_(module.bias)
+
+      def forward(self, x):
+          x = self.input_proj(x)
+          x = self.transformer(x.unsqueeze(1)).squeeze(1)
+          return self.head(x)