import os import gc os.environ["CUDA_VISIBLE_DEVICES"]="0,1" import torch from tqdm import tqdm from typing import Optional, Union import pandas as pd, numpy as np, torch from datasets import Dataset, load_dataset from dataclasses import dataclass from transformers import AutoTokenizer, AutoConfig, AutoModelForSequenceClassification, AutoModel from transformers import EarlyStoppingCallback from transformers.tokenization_utils_base import PreTrainedTokenizerBase, PaddingStrategy from transformers import AutoModelForMultipleChoice, TrainingArguments, Trainer import numpy as np from sklearn.metrics import recall_score, accuracy_score from transformers import DataCollatorWithPadding import logging # import mylib logger = logging.getLogger(__name__) VER = 1 MAX_LEN = 256 TOKENIZER_BINARY = "crarojasca/BinaryAugmentedCARDS" BINARY_MODEL = "Medissa/Roberta_Binary" TOKENIZER_MULTI_CLASS = "crarojasca/TaxonomyAugmentedCARDS" MULTI_CLASS_MODEL = "Medissa/Deberta_Taxonomy" ID2LABEL = { 0: '1_not_happening', 1: '2_not_human', 2: '3_not_bad', 3: '4_solutions_harmful_unnecessary', 4: '7_fossil_fuels_needed', 5: '5_science_unreliable', 6: '6_proponents_biased'} device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') from fastapi import APIRouter from datetime import datetime from datasets import load_dataset from sklearn.metrics import accuracy_score import random from .utils.evaluation import TextEvaluationRequest from .utils.emissions import tracker, clean_emissions_data, get_space_info router = APIRouter() DESCRIPTION = "Random Baseline" ROUTE = "/text" @router.post(ROUTE, tags=["Text Task"], description=DESCRIPTION) async def evaluate_text(request: TextEvaluationRequest): """ Evaluate text classification for climate disinformation detection. Current Model: Random Baseline - Makes random predictions from the label space (0-7) - Used as a baseline for comparison """ # Get space info username, space_url = get_space_info() # Define the label mapping LABEL_MAPPING = { "0_not_relevant": 0, "1_not_happening": 1, "2_not_human": 2, "3_not_bad": 3, "4_solutions_harmful_unnecessary": 4, "5_science_unreliable": 5, "6_proponents_biased": 6, "7_fossil_fuels_needed": 7 } # Load and prepare the dataset dataset = load_dataset(request.dataset_name) # Convert string labels to integers dataset = dataset.map(lambda x: {"label": LABEL_MAPPING[x["label"]]}) # Split dataset train_test = dataset["train"] test_dataset = dataset["test"] # Start tracking emissions tracker.start() tracker.start_task("inference") #-------------------------------------------------------------------------------------------- # YOUR MODEL INFERENCE CODE HERE # Update the code below to replace the random baseline by your model inference within the inference pass where the energy consumption and emissions are tracked. #-------------------------------------------------------------------------------------------- print('Start Binary') # Binary Model tokenizer = AutoTokenizer.from_pretrained(BINARY_MODEL) print('Loaded Tokenizer') model = AutoModelForSequenceClassification.from_pretrained(BINARY_MODEL) print(device) model.to(device) model.eval() print('Loaded Model') predictions = [] for i,text in tqdm(enumerate(test_dataset["quote"])): print(i) with torch.no_grad(): tokenized_text = tokenizer(text, truncation=True, padding='max_length', return_tensors = "pt") inputt = {k:v.to(device) for k,v in tokenized_text.items()} # Running Binary Model outputs = model(**inputt) binary_prediction = torch.argmax(outputs.logits, axis=1) binary_predictions = binary_prediction.to('cpu').item() prediction = "0_not_relevant" if binary_prediction==0 else 1 predictions.append(prediction) gc.collect() ## 2. Taxonomy Model print('Start Multi') tokenizer = AutoTokenizer.from_pretrained(MULTI_CLASS_MODEL) print('Loaded Tokenizer') model = AutoModelForSequenceClassification.from_pretrained(MULTI_CLASS_MODEL) model.to(device) model.eval() print('Loaded Model') for i,text in tqdm(enumerate(test_dataset["quote"])): if isinstance(predictions[i], str): continue with torch.no_grad(): tokenized_text = tokenizer(text, truncation=True, padding='max_length', return_tensors = "pt") inputt = {k:v.to(device) for k,v in tokenized_text.items()} outputs = model(**inputt) taxonomy_prediction = torch.argmax(outputs.logits, axis=1) taxonomy_prediction = taxonomy_prediction.to('cpu').item() prediction = ID2LABEL[taxonomy_prediction] predictions[i] = prediction if i%10: print(f'iteration: {i}') predictions = [LABEL_MAPPING[pred] for pred in predictions] #-------------------------------------------------------------------------------------------- # YOUR MODEL INFERENCE STOPS HERE #-------------------------------------------------------------------------------------------- # Stop tracking emissions emissions_data = tracker.stop_task() # Calculate accuracy accuracy = accuracy_score(true_labels, predictions) # Prepare results dictionary results = { "username": username, "space_url": space_url, "submission_timestamp": datetime.now().isoformat(), "model_description": DESCRIPTION, "accuracy": float(accuracy), "energy_consumed_wh": emissions_data.energy_consumed * 1000, "emissions_gco2eq": emissions_data.emissions * 1000, "emissions_data": clean_emissions_data(emissions_data), "api_route": ROUTE, "dataset_config": { "dataset_name": request.dataset_name, "test_size": request.test_size, "test_seed": request.test_seed } } return results