for epoch in range(n_epoch): for j, data in enumerate(train_loader, 0): inputs, labels = data input_data = inputs.cuda() class_label = labels.cuda() my_net.zero_grad() my_net.train() class_output= my_net(input_data) err_s_label = loss_class(class_output, class_label) err = err_s_label err.backward() optimizer.step()

for batch_idx, (inputs, labels) in enumerate(self.dataloaders[phase]): if phase != 'source_train' or epoch < args.middle_epoch: inputs = inputs.to(self.device) labels = labels.to(self.device) else: source_inputs = inputs target_inputs, target_labels = iter_target.next() inputs = torch.cat((source_inputs, target_inputs), dim=0) inputs = inputs.to(self.device) labels = labels.to(self.device) if (step + 1) % len_target_loader == 0: iter_target = iter(self.dataloaders['target_train'])

在代码的开头，您使用了一个for循环来遍历self.dataloaders[phase]，并使用enumerate函数获取每个批次的inputs和labels。在else子句中，您尝试从iter_target迭代器中获取target_inputs和target_labels。但是，在第一...

transform = transforms.Compose([ transforms.Resize((224, 224)), transforms.ToTensor(), transforms.Normalize((0.5,0.5,0.5), (0.5,0.5,0.5)) ]) # 使用原始字符串处理Windows路径和空格 dataset_path = r"C:\Users\29930\Desktop\IOS图像\结构参数图" # 加载完整数据集 full_dataset = datasets.ImageFolder( root=dataset_path, transform=transform ) # 按8:2比例划分 train_size = int(0.8 * len(full_dataset)) test_size = len(full_dataset) - train_size train_dataset, test_dataset = random_split( full_dataset, [train_size, test_size] ) # Create data loaders for batching and shuffling train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=64, shuffle=True) test_loader = torch.utils.data.DataLoader(test_dataset, batch_size=64, shuffle=False) # Define the CNN Model class CNN(nn.Module): def init(self, input_channels=3, img_size=224): # 添加尺寸参数 super().init() self.conv1 = nn.Conv2d(input_channels,32,kernel_size=3,stride=1,padding=1) self.pool = nn.MaxPool2d(2,2) self.conv2 = nn.Conv2d(32,64,3,1,1) # 自动计算展平维度 with torch.no_grad(): self.flatten_size = 64 * (img_size//4) * (img_size//4) self.fc1 = nn.Linear(self.flatten_size,128) self.fc2 = nn.Linear(128,10) def forward(self, x): x = self.pool(F.relu(self.conv1(x))) x = self.pool(F.relu(self.conv2(x))) x = x.view(-1, self.flatten_size) x = F.relu(self.fc1(x)) x = self.fc2(x) return x # Instantiate the model model = CNN() print(model) # Define the loss function and optimizer criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=0.001) # Train the model num_epochs = 10 for epoch in range(num_epochs): running_loss =0.0 for i, data in enumerate(train_loader,0): inputs, labels = data optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() running_loss += loss.item() if i % 100==99: print(f'Epoch {epoch + 1}, Batch {i + 1}, Loss: {running_loss /100:.3f}') running_loss = 0.0 print('Finished Training') correct =0 total =0 with torch.no_grad(): for data in test_loader: images, labels = data outputs = model(images) _, predicted = torch.max(outputs.data,1) total += labels.size(0) correct += (predicted == labels).sum().item() print(f'Accuracy of the model on the test set: {100 * correct / total:.2f}%') all_probs = [] all_labels = [] with torch.no_grad(): for data in test_loader: images, labels = data outputs = model(images) probs = F.softmax(outputs, dim=1) all_probs.append(probs.numpy()) all_labels.append(labels.numpy()) all_probs = np.concatenate(all_probs) all_labels = np.concatenate(all_labels) # 计算多分类AUC（使用One-vs-Rest方法） n_classes = len(full_dataset.classes) fpr = dict() tpr = dict() roc_auc = dict() # 将标签转换为二值化形式 y_test = label_binarize(all_labels, classes=range(n_classes)) for i in range(n_classes): fpr[i], tpr[i], _ = roc_curve(y_test[:, i], all_probs[:, i]) roc_auc[i] = auc(fpr[i], tpr[i]) # 计算宏观平均AUC fpr["macro"], tpr["macro"], _ = roc_curve(y_test.ravel(), all_probs.ravel()) roc_auc["macro"] = auc(fpr["macro"], tpr["macro"])raise ValueError( ValueError: Found input variables with inconsistent numbers of samples: [103, 1030]

for data in test_loader: images, labels = data outputs = model(images) probs = F.softmax(outputs, dim=1) all_probs.append(probs.numpy()) all_labels.append(labels.numpy()) 这里，probs的shape应该...

检查一下：import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader, TensorDataset from sklearn.metrics import roc_auc_score # 定义神经网络模型 class Net(nn.Module): def init(self): super(Net, self).init() self.fc1 = nn.Linear(10, 64) self.fc2 = nn.Linear(64, 32) self.fc3 = nn.Linear(32, 1) self.sigmoid = nn.Sigmoid() def forward(self, x): x = self.fc1(x) x = nn.functional.relu(x) x = self.fc2(x) x = nn.functional.relu(x) x = self.fc3(x) x = self.sigmoid(x) return x # 加载数据集 data = torch.load('data.pt') x_train, y_train, x_test, y_test = data train_dataset = TensorDataset(x_train, y_train) train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True) test_dataset = TensorDataset(x_test, y_test) test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False) # 定义损失函数和优化器 criterion = nn.BCELoss() optimizer = optim.Adam(net.parameters(), lr=0.01) # 训练模型 net = Net() for epoch in range(10): running_loss = 0.0 for i, data in enumerate(train_loader): inputs, labels = data optimizer.zero_grad() outputs = net(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() running_loss += loss.item() # 在测试集上计算AUC y_pred = [] y_true = [] with torch.no_grad(): for data in test_loader: inputs, labels = data outputs = net(inputs) y_pred += outputs.tolist() y_true += labels.tolist() auc = roc_auc_score(y_true, y_pred) print('Epoch %d, loss: %.3f, test AUC: %.3f' % (epoch + 1, running_loss / len(train_loader), auc))

for i, data in enumerate(train_loader): inputs, labels = data optimizer.zero_grad() outputs = net(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() running_loss += ...

torch import torch.nn as nn import torch.optim as optim from torch.utils.data import DataLoader from torch.nn.parallel import DataParallel from torch.optim.lr_scheduler import CosineAnnealingLR def train_model(model, train_loader, val_loader, num_epochs=50, device_ids=[0,1,2,3]): device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu") model = DataParallel(model, device_ids=device_ids).to(device) criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=0.001) scheduler = CosineAnnealingLR(optimizer, T_max=num_epochs) best_acc = 0.0 for epoch in range(num_epochs): model.train() running_loss = 0.0 for inputs, labels in train_loader: inputs, labels = inputs.to(device), labels.to(device) optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() running_loss += loss.item() scheduler.step() # Validation model.eval() correct = 0 total = 0 with torch.no_grad(): for inputs, labels in val_loader: inputs, labels = inputs.to(device), labels.to(device) outputs = model(inputs) _, predicted = torch.max(outputs.data, 1) total += labels.size(0) correct += (predicted == labels).sum().item() val_acc = 100 * correct / total print(f'Epoch {epoch+1}/{num_epochs}, Loss: {running_loss/len(train_loader):.4f}, Val Acc: {val_acc:.2f}%') if val_acc > best_acc: best_acc = val_acc torch.save(model.module.state_dict(), 'best_model.pth') print(f'Training complete. Best Val Acc: {best_acc:.2f}%')输出结果

for i, (inputs, labels) in enumerate(train_loader): inputs, labels = inputs.to(device), labels.to(device) optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss....

def train(model, train_loader, criterion, optimizer): model.train() train_loss = 0.0 train_acc = 0.0 for i, (inputs, labels) in enumerate(train_loader): optimizer.zero_grad() outputs = model(inputs.unsqueeze(1).float()) loss = criterion(outputs, labels.long()) loss.backward() optimizer.step() train_loss += loss.item() * inputs.size(0) _, preds = torch.max(outputs, 1) train_acc += torch.sum(preds == labels.data) train_loss = train_loss / len(train_loader.dataset) train_acc = train_acc.double() / len(train_loader.dataset) return train_loss, train_acc def test(model, verify_loader, criterion): model.eval() test_loss = 0.0 test_acc = 0.0 with torch.no_grad(): for i, (inputs, labels) in enumerate(test_loader): outputs = model(inputs.unsqueeze(1).float()) loss = criterion(outputs, labels.long()) test_loss += loss.item() * inputs.size(0) _, preds = torch.max(outputs, 1) test_acc += torch.sum(preds == labels.data) test_loss = test_loss / len(test_loader.dataset) test_acc = test_acc.double() / len(test_loader.dataset) return test_loss, test_acc # Instantiate the model model = CNN() # Define the loss function and optimizer criterion = nn.CrossEntropyLoss() optimizer = optim.Adam(model.parameters(), lr=0.001) # Instantiate the data loaders train_dataset = MyDataset1('1MATRICE') train_loader = DataLoader(train_dataset, batch_size=5, shuffle=True) test_dataset = MyDataset2('2MATRICE') test_loader = DataLoader(test_dataset, batch_size=5, shuffle=False) train_losses, train_accs, test_losses, test_accs = [], [], [], [] for epoch in range(500): train_loss, train_acc = train(model, train_loader, criterion, optimizer) test_loss, test_acc = test(model, test_loader, criterion) train_losses.append(train_loss) train_accs.append(train_acc) test_losses.append(test_loss) test_accs.append(test_acc) print('Epoch: {} Train Loss: {:.4f} Train Acc: {:.4f} Test Loss: {:.4f} Test Acc: {:.4f}'.format( epoch, train_loss, train_acc, test_loss, test_acc))

代码中定义了两个函数：train和test，分别用于训练模型和测试模型。在训练过程中，首先将模型设置为训练模式，然后遍历训练数据集，对每个batch的数据进行前向传播、反向传播和优化器更新。在每个batch的训练结束...

for epoch in range(6): # # for epoch in range(Epoch + 1): time1 = time.time() model.train() los = [] bar = tqdm(enumerate(data_train_loader), colour='yellow', total=len(data_train_loader)) bar.set_description(f'Epoch: {epoch}') for step, data in bar: inputs, labels = data inputs = inputs.to(device) labels = labels.to(device) outputs = model(inputs) # 计算损失函数 loss = loss_func(outputs, labels) # 清空上一轮的梯度 optimizer.zero_grad() # 反向传播 loss.backward() # 参数更新 optimizer.step() los.append(loss.cpu().detach().numpy()) bar.set_postfix({'loss': loss.cpu().detach().numpy()}) print('epoch:', epoch, 'loss', np.array(los).mean(), 'time:', time.time() - time1, 's')

5. 获取一个批次(batch)的数据(inputs和labels)； 6. 将数据放到GPU上进行计算(如果GPU可用)； 7. 将数据输入模型，得到输出(outputs)； 8. 计算损失函数(loss)； 9. 清空上一轮的梯度； 10. 反向传播求梯度； 11. ...

运行以下Python代码：import torchimport torch.nn as nnimport torch.optim as optimfrom torchvision import datasets, transformsfrom torch.utils.data import DataLoaderfrom torch.autograd import Variableclass Generator(nn.Module): def init(self, input_dim, output_dim, num_filters): super(Generator, self).init() self.input_dim = input_dim self.output_dim = output_dim self.num_filters = num_filters self.net = nn.Sequential( nn.Linear(input_dim, num_filters), nn.ReLU(), nn.Linear(num_filters, num_filters2), nn.ReLU(), nn.Linear(num_filters2, num_filters4), nn.ReLU(), nn.Linear(num_filters4, output_dim), nn.Tanh() ) def forward(self, x): x = self.net(x) return xclass Discriminator(nn.Module): def init(self, input_dim, num_filters): super(Discriminator, self).init() self.input_dim = input_dim self.num_filters = num_filters self.net = nn.Sequential( nn.Linear(input_dim, num_filters4), nn.LeakyReLU(0.2), nn.Linear(num_filters4, num_filters2), nn.LeakyReLU(0.2), nn.Linear(num_filters2, num_filters), nn.LeakyReLU(0.2), nn.Linear(num_filters, 1), nn.Sigmoid() ) def forward(self, x): x = self.net(x) return xclass ConditionalGAN(object): def init(self, input_dim, output_dim, num_filters, learning_rate): self.generator = Generator(input_dim, output_dim, num_filters) self.discriminator = Discriminator(input_dim+1, num_filters) self.optimizer_G = optim.Adam(self.generator.parameters(), lr=learning_rate) self.optimizer_D = optim.Adam(self.discriminator.parameters(), lr=learning_rate) def train(self, data_loader, num_epochs): for epoch in range(num_epochs): for i, (inputs, labels) in enumerate(data_loader): # Train discriminator with real data real_inputs = Variable(inputs) real_labels = Variable(labels) real_labels = real_labels.view(real_labels.size(0), 1) real_inputs = torch.cat((real_inputs, real_labels), 1) real_outputs = self.discriminator(real_inputs) real_loss = nn.BCELoss()(real_outputs, torch.ones(real_outputs.size())) # Train discriminator with fake data noise = Variable(torch.randn(inputs.size(0), self.generator.input_dim)) fake_labels = Variable(torch.LongTensor(inputs.size(0)).random_(0, 10)) fake_labels = fake_labels.view(fake_labels.size(0), 1) fake_inputs = self.generator(torch.cat((noise, fake_labels.float()), 1)) fake_inputs = torch.cat((fake_inputs, fake_labels), 1) fake_outputs = self.discriminator(fake_inputs) fake_loss = nn.BCELoss()(fake_outputs, torch.zeros(fake_outputs.size())) # Backpropagate and update weights for discriminator discriminator_loss = real_loss + fake_loss self.discriminator.zero_grad() discriminator_loss.backward() self.optimizer_D.step() # Train generator noise = Variable(torch.randn(inputs.size(0), self.generator.input_dim)) fake_labels = Variable(torch.LongTensor(inputs.size(0)).random_(0,

for i, (inputs, labels) in enumerate(data_loader): ... 其中，ConditionalGAN 类接受输入维度 input_dim、输出维度 output_dim、特征数 num_filters 和学习率 learning_rate。train 方法则接受数据加载器 ...

import numpy as np import torch from torch.utils.data import Dataset, DataLoader class DiabetesDataset(Dataset): def init(self, filepath): xy = np.loadtxt(filepath, delimiter=',', dtype=np.float32) self.len = xy.shape[0] self.x_data = torch.from_numpy(xy[:, :-1]) self.y_data = torch.from_numpy(xy[:, [-1]]) def getitem(self, index): return self.x_data[index], self.y_data[index] def len(self): return self.len dataset = DiabetesDataset('diabetes.csv.gz') train_loader = DataLoader(dataset=dataset,batch_size=32,shuffle=True,num_workers=2) class Model(torch.nn.Module): def init(self): super(Model, self).init() self.linear1 = torch.nn.Linear(8, 6) self.linear2 = torch.nn.Linear(6, 4) self.linear3 = torch.nn.Linear(4, 1) self.sigmoid = torch.nn.Sigmoid() def forward(self, x): x = self.sigmoid(self.linear1(x)) x = self.sigmoid(self.linear2(x)) x = self.sigmoid(self.linear3(x)) return x model = Model() criterion = torch.nn.BCELoss(size_average=True) optimizer = torch.optim.SGD(model.parameters(), lr=0.01) if name == 'main': for epoch in range(100): for i, data in enumerate(train_loader, 0): inputs, labels = data y_pred = model(inputs) loss = criterion(y_pred, labels) print(epoch, i, loss.item()) optimizer.zero_grad() loss.backward() optimizer.step() x_test = torch.Tensor([-0.0588235,0.839196,0.0491803,0,0,-0.305514,-0.492741,-0.633333]) y_test = model(x_test) print("y=",y_test) 对上述代码进行逐条注释，不要引用其他人的代码

for batch_idx, (data, target) in enumerate(loader): """遍历所有批次数据[^2][^4]""" # batch_idx: 批次索引 # data: 当前批次特征张量，形状为[batch_size, 特征维度] # target: 当前批次标签张量，形状为...

for epoch in range(10): running_loss = 0.0 for i, (inputs, labels) in enumerate(train_loader): optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() running_loss += loss.item() print('Epoch %d loss: %.3f' % (epoch + 1, running_loss / len(train_loader)))分开写

for i, (inputs, labels) in enumerate(train_loader): # 梯度清零 optimizer.zero_grad() # 前向传播 outputs = model(inputs) # 计算损失 loss = criterion(outputs, labels) # 反向传播 loss....

import torch from torchvision import transforms from torchvision import datasets from torch.utils.data import DataLoader import torch.nn.functional as F import torch.optim as optim batch_size = 64 transform = transforms.Compose([transforms.ToTensor(),transforms.Normalize((0.1307, ), (0.3081, ))]) train_dataset = datasets.MNIST(root='../dataset/mnist/',train=True,download=True,transform=transform) train_loader = DataLoader(train_dataset,shuffle=True,batch_size=batch_size) test_dataset = datasets.MNIST(root='../dataset/mnist/',train=False,download=True,transform=transform) test_loader = DataLoader(test_dataset,shuffle=False,batch_size=batch_size) class Net(torch.nn.Module): def init(self): super(Net, self).init() self.l1 = torch.nn.Linear(784, 512) self.l2 = torch.nn.Linear(512, 256) self.l3 = torch.nn.Linear(256, 128) self.l4 = torch.nn.Linear(128, 64) self.l5 = torch.nn.Linear(64, 10) def forward(self, x): x = x.view(-1, 784) x = F.relu(self.l1(x)) x = F.relu(self.l2(x)) x = F.relu(self.l3(x)) x = F.relu(self.l4(x)) return self.l5(x) model = Net() criterion = torch.nn.CrossEntropyLoss() optimizer = optim.SGD(model.parameters(), lr=0.01, momentum=0.5) def train(epoch): running_loss = 0.0 for batch_idx, data in enumerate(train_loader, 0): inputs, target = data optimizer.zero_grad() # forward + backward + update outputs = model(inputs) loss = criterion(outputs, target) loss.backward() optimizer.step() running_loss += loss.item() if batch_idx % 300 == 299: print('[%d, %5d] loss: %.3f' % (epoch + 1, batch_idx + 1, running_loss / 300)) running_loss = 0.0 def test(): correct = 0 total = 0 with torch.no_grad(): for data in test_loader: images, labels = data outputs = model(images) _, predicted = torch.max(outputs.data, dim=1) total += labels.size(0) correct += (predicted == labels).sum().item() print(f'测试集准确率: {100 * correct / total:.2f}%') # 修正缩进 if name == 'main': for epoch in range(10): train(epoch) test()

for batch_idx, (inputs, targets) in enumerate(train_loader): inputs, targets = inputs.to(device), targets.to(device) optimizer.zero_grad() outputs = model(inputs) loss = criterion(outputs, ...

for epoch in range(10): running_loss = 0.0 for i, data in tqdm(enumerate(fashion_trainloader), total=len(fashion_trainloader)): inputs, labels = data optimizer.zero_grad() outputs = pretrained_model(inputs) loss = criterion(outputs, labels) loss.backward() optimizer.step() running_loss += loss.item() print('Epoch %d, loss: %.3f' % (epoch + 1, running_loss / len(fashion_trainloader)))

- 然后，在内层的 for 循环中，我们使用 optimizer.zero_grad() 来清空梯度，然后将输入数据 inputs 喂给预训练的模型 pretrained_model，得到输出 outputs。 - 接着，我们计算输出和标签之间的损失，使用 criterion...

接下来我会给出一段代码，请理解这段代码的逻辑，并且仿照其将其改变为医疗图像分析代码如下： from future import print_function, division import numpy as np import torch import torch.nn as nn import torch.nn.functional as F import torch.optim as optim from torch.autograd import Variable from torch.utils.data import Dataset from torchvision import transforms, datasets, models from Dataloader import DogCatDataSet # 配置参数 random_state = 1 torch.manual_seed(random_state) # 设置随机数种子，确保结果可重复 torch.cuda.manual_seed(random_state) torch.cuda.manual_seed_all(random_state) np.random.seed(random_state) # random.seed(random_state) epochs = 30 # 训练次数 batch_size = 16 # 批处理大小 num_workers = 4 # 多线程的数目 use_gpu = torch.cuda.is_available() # 对加载的图像作归一化处理，并裁剪为[224x224x3]大小的图像 data_transform = transforms.Compose([ transforms.Resize(256), transforms.CenterCrop(224), transforms.ToTensor(), ]) train_dataset = DogCatDataSet(img_dir="/mnt/d/深度学习1/train", transform=data_transform) train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=4) test_dataset = DogCatDataSet(img_dir="/mnt/d/深度学习1/test", transform=data_transform) test_loader = torch.utils.data.DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=4) # 加载resnet18 模型， net = models.resnet18(pretrained=False) num_ftrs = net.fc.in_features net.fc = nn.Linear(num_ftrs, 2) if use_gpu: net = net.cuda() print(net) # 定义loss和optimizer cirterion = nn.CrossEntropyLoss() optimizer = optim.SGD(net.parameters(), lr=0.0001, momentum=0.9) # 开始训练 net.train() for epoch in range(epochs): running_loss = 0.0 train_correct = 0 train_total = 0 for i, data in enumerate(train_loader, 0): inputs, train_labels = data if use_gpu: inputs, labels = Variable(inputs.cuda()), Variable(train_labels.cuda()) else: inputs, labels = Variable(inputs), Variable(train_labels) # inputs, labels = Variable(inputs), Variable(train_labels) optimi

for inputs, targets in dataloader: outputs = model(inputs.to(device)) probs.extend(torch.softmax(outputs, dim=1)[:,1].cpu().numpy()) labels.extend(targets.numpy()) return roc_auc_score(labels, ...

import torch import json import os import argparse import numpy as np import re from torch.utils.data import Dataset, DataLoader from tqdm import tqdm from PIL import Image from peft import LoraConfig, get_peft_model from transformers import ( AutoModelForCausalLM, AutoProcessor, TrainingArguments, BitsAndBytesConfig, GenerationConfig, AutoTokenizer, AutoImageProcessor, get_cosine_schedule_with_warmup ) import torch.optim as optim from sklearn.metrics import f1_score, accuracy_score, precision_score, recall_score, classification_report import warnings warnings.filterwarnings("ignore", message="Could not find a config file") # 标签映射定义 TASK1_LABEL_MAP = {"无害": 0, "有害": 1} TASK2_LABEL_MAP = { 0: "无害", 1: "针对性有害", 2: "一般性有害", 3: "性暗示", 4: "沮丧文化" } class HarmfulMemeDataset(Dataset): def init(self, annotation_path, tokenizer, image_processor, task_name, image_root=None): print(f"Loading annotations from: {annotation_path}") self.image_root = image_root with open(annotation_path, 'r', encoding='utf-8') as f: self.data = json.load(f) self.tokenizer = tokenizer self.task_name = task_name self.image_processor = image_processor self.label_token_ids = self._get_label_tokens() def _get_label_tokens(self): """获取标签对应的token ID""" if self.task_name == "task_1": labels = ["无害", "有害"] else: labels = [str(i) for i in range(5)] # 0-4 token_ids = {} tokenizer = self.processor.tokenizer for label in labels: tokens = tokenizer.encode(label) token_ids[label] = tokens # 保存完整的token序列 if len(tokens) > 1: print(f"警告: 标签 '{label}' 被分词为多个token: {tokens}") return token_ids def len(self): return len(self.data) def getitem(self, idx): sample = self.data[idx] if self.image_root: image_path = os.path.join(self.image_root, sample["path"]) else: image_path = sample["path"] if not os.path.exists(image_path): raise FileNotFoundError(f"图片不存在：{image_path}") image = Image.open(image_path).convert("RGB") if self.task_name == "task_1": user_content = f"任务：有害模因检测（是否有害）。\n输入文本：'{sample['text']}'\n请判断此图文组合是否有害。" raw_label = sample['label']# 原始标签是字符串（"无害"或"有害"） # 根据TASK1_LABEL_MAP进行映射 if isinstance(raw_label, int): # 如果是整数，转换为对应的字符串标签 label_map = {v: k for k, v in TASK1_LABEL_MAP.items()} # 反转映射 label = label_map.get(raw_label, "无害") # 默认值为"无害" else: # 如果已经是字符串，直接使用 label = raw_label label_token = self.label_token_ids[label] assistant_content = f"结论：{label}。\n理由：{sample['explanation']}" else: user_content = f"任务：有害模因类型分类。\n输入文本：'{sample['text']}'\n请判断此图文组合的有害类型（0-4）。" raw_label = str(sample['type'])# 将整数标签转换为字符串 label = str(raw_label) label_token = self.label_token_ids[label] assistant_content = f"结论：{label}。\n理由：{sample['explanation']}" messages = [ {"role": "user", "content": [{"type": "image"}, {"type": "text", "text": user_content}]}, {"role": "assistant", "content": [{"type": "text", "text": assistant_content}]} ] prompt = self.processor.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, chat_format="chatml" ) # 单独处理图像 image = self.image_processor( images=image, return_tensors="pt" )["pixel_values"].squeeze(0) # 单独处理文本 encoding = self.tokenizer( text=prompt, return_tensors="pt", padding=False, truncation=False ) prompt_tokens = encoding["input_ids"][0].tolist() # 找到结论标签的位置 conclusion_start = self.processor.tokenizer.encode("结论：") # 在prompt中查找"结论："的位置 start_idx = -1 for i in range(len(prompt_tokens) - len(conclusion_start) + 1): if prompt_tokens[i:i+len(conclusion_start)] == conclusion_start: start_idx = i + len(conclusion_start) break inputs = self.processor( text=prompt, images=image, return_tensors="pt", padding="max_length", truncation=True, max_length=512 ) inputs = {k: v.squeeze(0) for k, v in inputs.items()} # 创建标签张量，只标记结论位置 labels = torch.full_like(inputs["input_ids"], fill_value=-100, dtype=torch.long) if start_idx != -1 and start_idx < len(labels): # 标记整个标签token序列 label_tokens = self.label_token_ids[label] for i, token_id in enumerate(label_tokens): if start_idx + i < len(labels): labels[start_idx + i] = token_id inputs["labels"] = labels return inputs def parse_generated_text(self,text): """解析生成的文本，提取结论标签""" conclusion_match = re.search(r"结论[:：]\s(\S+)", text) if not conclusion_match: return None conclusion = conclusion_match.group(1).strip().rstrip('。.') # 处理多token标签 if conclusion in ["无害", "有害"]: # 任务1标签 return conclusion elif conclusion.isdigit() and 0 <= int(conclusion) <= 4: # 任务2标签 return conclusion # 尝试分词匹配 tokenizer = AutoProcessor.from_pretrained(args.model_id).tokenizer conclusion_tokens = tokenizer.encode(conclusion, add_special_tokens=False) # 与已知标签的token序列匹配 for label, tokens in self.label_token_ids.items(): if conclusion_tokens == tokens: return label return None def compute_metrics(task_name, preds, labels): """计算评估指标""" mask = labels != -100 preds = preds[mask] labels = labels[mask] if task_name == "task_1": # 二分类任务 return { "accuracy": accuracy_score(labels, preds), "f1": f1_score(labels, preds, average="binary"), "precision": precision_score(labels, preds, average="binary"), "recall": recall_score(labels, preds, average="binary") } else: # 多分类任务 report = classification_report(labels, preds, output_dict=True, zero_division=0) return { "accuracy": accuracy_score(labels, preds), "f1_macro": f1_score(labels, preds, average="macro"), "precision_macro": precision_score(labels, preds, average="macro"), "recall_macro": recall_score(labels, preds, average="macro"), "class_report": report } def main(args): os.environ["TOKENIZERS_PARALLELISM"] = "false" # 1. 加载模型和预处理器 print("Loading model and processor...") quantization_config = BitsAndBytesConfig( load_in_4bit=True, bnb_4bit_quant_type="nf4", bnb_4bit_compute_dtype=torch.bfloat16 ) model = AutoModelForCausalLM.from_pretrained( args.model_id, quantization_config=quantization_config, trust_remote_code=True, device_map="auto", bf16=True ) model.generation_config = GenerationConfig.from_pretrained( args.model_id, trust_remote_code=True, chat_format="chatml", max_new_tokens=100, pad_token_id=model.generation_config.eos_token_id ) # 分别初始化文本和图像处理器 tokenizer = AutoTokenizer.from_pretrained( args.model_id, trust_remote_code=True, pad_token='<|endoftext|>' # 显式设置pad_token ) image_processor = AutoImageProcessor.from_pretrained( args.model_id, trust_remote_code=True ) tokenizer.chat_template = """{% for message in messages %} <|im_start|>{{ message['role'] }} {{ message['content'] }} <|im_end|> {% endfor %} {% if add_generation_prompt %} <|im_start|>assistant {% endif %}""" # 设置pad token # 确保pad_token正确设置 if tokenizer.pad_token is None: tokenizer.pad_token = tokenizer.eos_token tokenizer.pad_token_id = tokenizer.eos_token_id # 2. LoRA配置 print("Configuring LoRA...") lora_config = LoraConfig( r=args.lora_rank, lora_alpha=args.lora_alpha, lora_dropout=args.lora_dropout, bias="none", task_type="CAUSAL_LM", target_modules=[ "c_attn", "c_proj", "w1", "w2", "w3", "visual.proj", "visual.image_encoder" ] ) peft_model = get_peft_model(model, lora_config) peft_model.print_trainable_parameters() # 3. 初始化优化器和调度器 optimizer = optim.AdamW( peft_model.parameters(), lr=args.learning_rate, weight_decay=args.weight_decay ) # 4. 训练参数配置 training_args = TrainingArguments( output_dir=os.path.join(args.output_dir, args.task), num_train_epochs=args.epochs, per_device_train_batch_size=args.batch_size, per_device_eval_batch_size=args.eval_batch_size, gradient_accumulation_steps=args.grad_accum_steps, learning_rate=args.learning_rate, weight_decay=args.weight_decay, lr_scheduler_type="cosine", logging_strategy="steps", logging_steps=10, save_strategy="epoch", eval_strategy="epoch", eval_accumulation_steps=1, metric_for_best_model="f1" if args.task == "task_1" else "f1_macro", greater_is_better=True, load_best_model_at_end=True, bf16=True, report_to="none", remove_unused_columns=False, disable_tqdm=False, skip_memory_metrics=True, dataloader_pin_memory=False, ) # 5. 加载数据集 print(f"Loading datasets for {args.task}...") train_dataset = HarmfulMemeDataset( annotation_path=args.train_annotation_path, tokenizer=tokenizer, image_processor=image_processor, task_name=args.task, image_root=args.image_root ) test_dataset = HarmfulMemeDataset( annotation_path=args.test_annotation_path, tokenizer=tokenizer, image_processor=image_processor, task_name=args.task, image_root=args.image_root ) # 创建数据加载器 train_loader = DataLoader( train_dataset, batch_size=args.batch_size, shuffle=True, num_workers=args.num_workers, pin_memory=True ) eval_loader = DataLoader( test_dataset, batch_size=args.eval_batch_size, shuffle=False, num_workers=args.num_workers, pin_memory=True ) # 计算总步数，初始化学习率调度器 total_train_steps = len(train_loader) // args.grad_accum_steps args.epochs scheduler = get_cosine_schedule_with_warmup( optimizer, num_warmup_steps=args.warmup_steps, num_training_steps=total_train_steps ) # 6. 训练循环 print(f"Starting {args.task} training...") best_metric = -1 for epoch in range(args.epochs): print(f"\n===== Epoch {epoch + 1}/{args.epochs} =====") # 训练阶段 peft_model.train() total_train_loss = 0.0 train_pbar = tqdm(train_loader, desc=f"Training Epoch {epoch + 1}", unit="batch") for step, batch in enumerate(train_pbar): batch = {k: v.to(peft_model.device) for k, v in batch.items()} # 前向传播 outputs = peft_model(**batch) loss = outputs.loss total_train_loss += loss.item() # 梯度累积 loss = loss / args.grad_accum_steps loss.backward() # 参数更新 if (step + 1) % args.grad_accum_steps == 0: optimizer.step() scheduler.step() optimizer.zero_grad() # 更新进度条 train_pbar.set_postfix({"loss": f"{loss.item() * args.grad_accum_steps:.4f}"}) avg_train_loss = total_train_loss / len(train_loader) print(f"Epoch {epoch + 1} 平均训练损失: {avg_train_loss:.4f}") # 评估阶段 peft_model.eval() all_preds = [] all_labels = [] all_generated_texts = [] eval_pbar = tqdm(eval_loader, desc=f"Evaluating Epoch {epoch + 1}", unit="batch") with torch.no_grad(): for batch in eval_pbar: # 获取真实标签 labels = batch["labels"].cpu().numpy() mask = labels != -100 valid_labels = labels[mask].reshape(-1) # 生成文本 inputs = {k: v.to(peft_model.device) for k, v in batch.items() if k != "labels"} pad_token_id = tokenizer.pad_token_id or tokenizer.eos_token_id generated_ids = peft_model.generate( **inputs, generation_config=model.generation_config, pad_token_id=pad_token_id # 使用修正后的值 ) # 解码生成的文本 generated_texts = tokenizer.batch_decode( generated_ids, skip_special_tokens=True, clean_up_tokenization_spaces=True ) # 解析生成的文本获取预测标签 batch_preds = [] for text in generated_texts: # 提取assistant的响应部分 if "<|im_start|>assistant" in text: response = text.split("<|im_start|>assistant")[-1].strip() else: response = text # 解析结论 conclusion = parse_generated_text(response) if conclusion is None: # 无法解析结论，使用默认值 pred_label = 0 if args.task == "task_1" else "0" else: pred_label = conclusion # 转换为数字标签 if args.task == "task_1": # 二分类任务 if "无害" in pred_label: pred_value = 0 elif "有害" in pred_label: pred_value = 1 else: # 无法解析，使用默认值 pred_value = 0 else: # 多分类任务 if pred_label in ["0", "1", "2", "3", "4"]: pred_value = int(pred_label) else: # 无法解析，使用默认值 pred_value = 0 batch_preds.append(pred_value) all_preds.extend(batch_preds) all_labels.extend(valid_labels.tolist()) all_generated_texts.extend(generated_texts) # 计算评估指标 metrics = compute_metrics(args.task, np.array(all_preds), np.array(all_labels)) # 打印评估结果 print("\n评估指标:") print("=" * 50) if args.task == "task_1": print(f"Accuracy: {metrics['accuracy']:.4f}") print(f"F1 Score: {metrics['f1']:.4f}") print(f"Precision: {metrics['precision']:.4f}") print(f"Recall: {metrics['recall']:.4f}") else: print(f"Accuracy: {metrics['accuracy']:.4f}") print(f"Macro F1: {metrics['f1_macro']:.4f}") print(f"Macro Precision: {metrics['precision_macro']:.4f}") print(f"Macro Recall: {metrics['recall_macro']:.4f}") print("\n分类报告:") print(classification_report(all_labels, all_preds, target_names=list(TASK2_LABEL_MAP.values()), zero_division=0)) print("=" * 50) # 保存最佳模型 current_metric = metrics["f1"] if args.task == "task_1" else metrics["f1_macro"] if current_metric > best_metric: best_metric = current_metric save_path = os.path.join(training_args.output_dir, f"best_model_epoch{epoch+1}") print(f"保存最佳模型（指标 {current_metric:.4f}）到 {save_path}") peft_model.save_pretrained(save_path) # 保存生成的文本示例 sample_output_path = os.path.join(save_path, "sample_outputs.txt") with open(sample_output_path, "w", encoding="utf-8") as f: for i, text in enumerate(all_generated_texts[:10]): f.write(f"样本 {i+1}:\n") f.write(text) f.write("\n" + "-"80 + "\n") print(f"训练完成！最佳指标: {best_metric:.4f}") if name == "main": parser = argparse.ArgumentParser(description="训练有害模因检测模型") parser.add_argument("--model_id", default="/xzwu/Qwen-VL-Chat", help="预训练模型路径") parser.add_argument("--output_dir", default="/xzwu/explain-m3-adapter", help="输出目录") parser.add_argument("--epochs", type=int, default=5, help="训练轮数") parser.add_argument("--batch_size", type=int, default=4, help="训练批次大小") parser.add_argument("--eval_batch_size", type=int, default=4, help="评估批次大小") parser.add_argument("--grad_accum_steps", type=int, default=2, help="梯度累积步数") parser.add_argument("--learning_rate", type=float, default=1e-5, help="学习率") parser.add_argument("--weight_decay", type=float, default=0.01, help="权重衰减") parser.add_argument("--warmup_steps", type=int, default=100, help="预热步数") parser.add_argument("--lora_rank", type=int, default=8, help="LoRA秩") parser.add_argument("--lora_alpha", type=int, default=16, help="LoRA alpha") parser.add_argument("--lora_dropout", type=float, default=0.1, help="LoRA dropout") parser.add_argument("--num_workers", type=int, default=4, help="数据加载工作线程数") parser.add_argument("--task", choices=["task_1", "task_2"], default="task_1", help="任务类型") parser.add_argument("--train_annotation_path", default="/xzwu/data/data/train_data_explanation.json", help="训练标注路径") parser.add_argument("--test_annotation_path", default="/xzwu/data/data/test_data_explanation.json", help="测试标注路径") parser.add_argument("--image_root", default="/xzwu/data/meme", help="图片根目录") args = parser.parse_args() # 打印配置 print("=" 50) print("训练配置:") for arg in vars(args): print(f"{arg}: {getattr(args, arg)}") print("=" * 50) main(args)运行以上代码报错：Traceback (most recent call last): File "/xzwu/explain-m3/explain-m3-project/train2.py", line 530, in <module> main(args) File "/xzwu/explain-m3/explain-m3-project/train2.py", line 314, in main train_dataset = HarmfulMemeDataset( File "/xzwu/explain-m3/explain-m3-project/train2.py", line 46, in init self.label_token_ids = self._get_label_tokens() File "/xzwu/explain-m3/explain-m3-project/train2.py", line 56, in _get_label_tokens tokenizer = self.processor.tokenizer AttributeError: 'HarmfulMemeDataset' object has no attribute 'processor'

input_ids = encoding["input_ids"][0] # 取第一个序列 return { "pixel_values": pixel_values, "input_ids": input_ids, "labels": input_ids.clone() # 对于CausalLM，标签就是输入ID } 这样修改后...

【路径规划】基于ADMM求解时间窗车辆路径问题VRPTW附Matlab代码.rar

【路径规划】基于ADMM求解时间窗车辆路径问题VRPTW附Matlab代码

基于GTD的项目管理系统。.zip

基于GTD的项目管理系统。

请解释这段代码for epoch in range(num_epochs): for i, (inputs, labels) in enumerate(train_loader): loss = train_step(inputs, labels) # 自定义训练函数 losses.append(loss.item())

相关推荐

请解释这段代码for epoch in range(num_epochs): for i, (inputs, labels) in enumerate(train_loader): loss = train_step(inputs, labels) # 自定义训练函数 losses.append(loss.item())

相关推荐

Pytorch-Classification_MNIST:用Pytorch对MNIST数据集进行分类

pytorch_mnist-python源码.zip

Mnist-Torch_torch_Mnist-Torch_

【路径规划】基于ADMM求解时间窗车辆路径问题VRPTW附Matlab代码.rar

基于GTD的项目管理系统。.zip

大家在看

KGM转MP3或者FLAC_kgma_kgma格式_FLAC_kgma转换器_kgm转换成flac_亲测完美转换！保证可用。

SPP Workshop.pdf

STM32F4U盘升级程序实例.zip

Easyquery焓熵表焓熵图查询软件V3.0，水和水蒸气焓熵图表查询软件

ST7789V_320x240TFT屏驱动应用可行.zip

最新推荐

获取本机IP地址的程序源码分析

【权威指南】：Win11笔记本上RTX4080驱动的安装与调试秘籍

windows环境举例

QQ自动发送/回复系统源代码开放

【7步打造Win11深度学习利器】：Tensorflow-GPU与RTX4080终极优化指南

ue画笔画画蓝图版

VB.NET图表曲线组件实现多种图表绘制

【MultiExtractor_Pro实战演练】：8个复杂场景的高效数据提取解决方案

k-means聚类分析算法推导

创意智力游戏：Pintu拼图挑战