Spaces:
Sleeping
Sleeping
File size: 5,074 Bytes
e6d07cd 4d6e8c2 ece5856 2c8310a 1f08781 4d6e8c2 1a885c6 4d6e8c2 c339ecd 1c33274 7eb6153 2c8310a 4d6e8c2 2c8310a e6d07cd 2c8310a e6d07cd 85c5204 e6d07cd 2c8310a 6f0e9af 2c8310a 6f0e9af c339ecd 2c8310a c339ecd 6f0e9af c339ecd 2c8310a 6f0e9af c339ecd c3f000b 2c8310a c3f000b 2c8310a c3f000b b206095 c339ecd b206095 6f0e9af 1f08781 2c8310a 1f08781 c3f000b b206095 1f08781 2c8310a e6d07cd 2c8310a 6f0e9af 2c8310a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 |
from fastapi import APIRouter
from datetime import datetime
from datasets import load_dataset
from sklearn.metrics import accuracy_score
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from torch.utils.data import DataLoader
from transformers import DataCollatorWithPadding
from .utils.evaluation import TextEvaluationRequest
from .utils.emissions import tracker, clean_emissions_data, get_space_info
router = APIRouter()
DESCRIPTION = "Climate Guard Toxic Agent is a ModernBERT for Climate Disinformation Detection"
ROUTE = "/text"
@router.post(ROUTE, tags=["Text Task"],
description=DESCRIPTION)
async def evaluate_text(request: TextEvaluationRequest):
"""
Evaluate text classification for climate disinformation detection using ModernBERT.
"""
# 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"]]})
# Get test dataset
test_dataset = dataset["test"]
# Start tracking emissions
tracker.start()
tracker.start_task("inference")
#--------------------------------------------------------------------------------------------
# MODEL INFERENCE CODE
#--------------------------------------------------------------------------------------------
try:
# Set device
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# Model and tokenizer paths
path_model = 'Tonic/climate-guard-toxic-agent'
path_tokenizer = "answerdotai/ModernBERT-base"
# Initialize tokenizer
tokenizer = AutoTokenizer.from_pretrained(path_tokenizer)
# Initialize model
model = AutoModelForSequenceClassification.from_pretrained(path_model).half().to(device)
# Set model to evaluation mode
model.eval()
# Preprocess function
def preprocess_function(examples):
return tokenizer(
examples["quote"],
truncation=True,
return_tensors=None
)
# Tokenize dataset
tokenized_test = test_dataset.map(
preprocess_function,
batched=True,
remove_columns=test_dataset.column_names
)
# Set format for pytorch
tokenized_test.set_format("torch")
# Create DataLoader
data_collator = DataCollatorWithPadding(tokenizer=tokenizer)
test_loader = DataLoader(
tokenized_test,
batch_size=16,
collate_fn=data_collator,
shuffle=False
)
# Get predictions
predictions = []
with torch.no_grad():
for batch in test_loader:
# Move batch to device
input_ids = batch['input_ids'].to(device)
attention_mask = batch['attention_mask'].to(device)
# Get model outputs
outputs = model(input_ids=input_ids, attention_mask=attention_mask)
preds = torch.argmax(outputs.logits, dim=-1)
# Add batch predictions to list
predictions.extend(preds.cpu().numpy().tolist())
# Clean up GPU memory
if torch.cuda.is_available():
torch.cuda.empty_cache()
except Exception as e:
print(f"Error during model inference: {str(e)}")
raise
#--------------------------------------------------------------------------------------------
# MODEL INFERENCE ENDS HERE
#--------------------------------------------------------------------------------------------
# Stop tracking emissions
emissions_data = tracker.stop_task()
# Calculate accuracy
accuracy = accuracy_score(test_dataset["label"], 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 |