ARB Security Solutions, LLC.

PyTorch Training Loss Threshold Breach Posting Alerts To Microsoft Teams

This script trains a model using the CIFAR10 dataset. During each epoch, it calculates the average training loss and checks if it exceeds a predefined threshold. If the threshold is exceeded, an alert is sent to a Microsoft Teams channel using a Microsoft Teams card created with the pymsteams library.

import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from pymsteams import connectorcard

# Microsoft Teams webhook URL
teams_webhook_url = 'https://your_teams_webhook_url'

# Define the model architecture
class CustomModel(nn.Module):
    def __init__(self):
        super(CustomModel, self).__init__()
        # Your model architecture goes here

    def forward(self, x):
        # Your forward pass logic goes here
        return x

# Main script
if __name__ == '__main__':
    # Data augmentation and normalization
    transform = transforms.Compose([
        transforms.RandomHorizontalFlip(),
        transforms.RandomCrop(32, padding=4),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.5, 0.5, 0.5], std=[0.5, 0.5, 0.5])
    ])

    # Load CIFAR10 dataset
    train_dataset = datasets.CIFAR10(root='./data', train=True, download=True, transform=transform)
    test_dataset = datasets.CIFAR10(root='./data', train=False, download=True, transform=transform)

    # Data loaders
    train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=128, shuffle=True, num_workers=2)
    test_loader = torch.utils.data.DataLoader(test_dataset, batch_size=128, shuffle=False, num_workers=2)

    # Model initialization
    model = CustomModel()

    # Loss function and optimizer
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.SGD(model.parameters(), lr=0.001, momentum=0.9)

    # Training loop
    num_epochs = 10
    loss_threshold = 1.0
    for epoch in range(num_epochs):
        # Training
        model.train()
        train_loss = 0.0
        train_correct = 0
        train_total = 0
        for inputs, labels in train_loader:
            optimizer.zero_grad()

            # Forward pass
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()
            optimizer.step()

            # Update training metrics
            train_loss += loss.item()
            _, predicted = torch.max(outputs.data, 1)
            train_total += labels.size(0)
            train_correct += (predicted == labels).sum().item()

        # Calculate average training loss
        train_loss /= len(train_loader)

        # Check if the loss exceeds the threshold
        if train_loss > loss_threshold:
            # Send alert to Microsoft Teams channel
            message = f"Training loss exceeded the threshold: {train_loss:.4f}"

            # Create a Microsoft Teams card
            card = connectorcard.ConnectorCard(webhookurl=teams_webhook_url)
            card.title('Training Alert')
            card.text(message)
            card.send()

            print(message)

        print(f"Epoch [{epoch+1}/{num_epochs}], Train Loss: {train_loss:.4f}, Train Accuracy: {train_accuracy:.2f}%")

Comments are closed.