利用 DeepSeek-R1 模型和计算机视觉微调模型实现马铃薯叶片疾病识别和分类

# 自定义 Mixup 函数
def mixup_data(x, y, alpha=0.2):
    if alpha > 0:
        lam = np.random.beta(alpha, alpha)
    else:
        lam = 1
    batch_size = x.size()[0]
    index = torch.randperm(batch_size)
    mixed_x = lam * x + (1 - lam) * x[index, :]
    y_a, y_b = y, y[index]
    return mixed_x, y_a, y_b, lam
# 自定义 Mixup 损失函数
def mixup_criterion(criterion, pred, y_a, y_b, lam):
    return lam * criterion(pred, y_a) + (1 - lam) * criterion(pred, y_b)
# 随机灰度化变换
class RandomGrayscale:
    def __init__(self, p=0.2):
        self.p = p
    def __call__(self, img):
        if torch.rand(1) < self.p:
            return transforms.functional.rgb_to_grayscale(img, num_output_channels=3)
        return img
# 训练数据预处理和增强
train_transform = transforms.Compose([
    transforms.Resize((224, 224)),
    RandomGrayscale(p=0.2),
    transforms.RandomRotation(30),
    transforms.RandomAffine(degrees=0, translate=(0.3, 0.3), shear=0.3),
    transforms.RandomHorizontalFlip(),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])
# 验证数据预处理
val_transform = transforms.Compose([
    transforms.Resize((224, 224)),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])

from ultralytics import YOLO
data_path = '*'
model = YOLO('yolo11n-cls.pt') 
model.train(task='classify', mode='train', data=data_path, epochs=300) 

    device = torch.device("cuda" if torch.cuda.is_available() else "cpu") 
    train_dataset = datasets.ImageFolder(root=train_dir, transform=transform)
    val_dataset = datasets.ImageFolder(root=val_dir, transform=transform)    
    train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True, num_workers=4)
    val_loader = DataLoader(val_dataset, batch_size=batch_size, shuffle=False, num_workers=4)    
    num_classes = len(train_dataset.classes)    
    model = resnet50(num_classes=num_classes).to(device)    
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=0.0001)
    scheduler = optim.lr_scheduler.StepLR(optimizer, step_size=5, gamma=0.1)
    best_acc = 0.0
    early_stopping_patience = 5
    early_stopping_counter = 0

BATCH_SIZE = 32
train_loader = DataLoader(train_dataset, batch_size=BATCH_SIZE, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=BATCH_SIZE, shuffle=False)
model = mobilenet_v3_small(pretrained=True)
num_ftrs = model.classifier[3].in_features
model.classifier[3] = nn.Linear(num_ftrs, len(train_dataset.classes))
criterion = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.0001, momentum=0.9)
scheduler = ReduceLROnPlateau(optimizer, mode='min', factor=0.5, patience=5, verbose=1)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device)
train_losses = []
val_losses = []
train_accuracies = []
val_accuracies = []
best_val_loss = float('inf')
best_model_weights = copy.deepcopy(model.state_dict())
early_stopping_counter = 0

EPOCHS = 300
BATCH_SIZE = 32
model = Sequential([
    Conv2D(32, (3, 3), activation='relu', input_shape=(224, 224, 3)),
    MaxPooling2D(2, 2),
    Conv2D(64, (3, 3), activation='relu'),
    MaxPooling2D(2, 2),
    Conv2D(128, (3, 3), activation='relu'),
    Conv2D(128, (3, 3), activation='relu'),
    MaxPooling2D(2, 2),
    Flatten(),
    Dense(256, activation='relu'),  
    Dropout(0.5),
    Dense(128, activation='relu'), 
    Dropout(0.5),
    Dense(3, activation='softmax')
])
model.compile(loss='categorical_crossentropy',
            optimizer=RMSprop(learning_rate=0.0001),
            metrics=['accuracy'])
history = model.fit(
    train_generator,
    steps_per_epoch=len(train_generator.filenames) 
    epochs=EPOCHS,
    validation_data=validation_generator,
    validation_steps=len(validation_generator.filenames) 
    callbacks=callbacks,
    verbose=1
)
model.save_weights(MODEL_WEIGHTS_FILE)
model.save(MODEL_FILE)        

for class_folder in os.listdir(test_folder):
    class_folder_path = os.path.join(test_folder, class_folder)
        if os.path.isdir(class_folder_path):
        ground_truth_class = class_folder  
        print(f"Processing class folder (Ground Truth): {ground_truth_class}")
        image_files = [os.path.join(class_folder_path, img) for img in os.listdir(class_folder_path) if img.endswith(('.jpg', '.jpeg', '.png'))]
        for img_path in image_files:
            results = model.predict(img_path, conf=0.50)
            for result in results:
                predicted_class_index = result.probs.top1              
                predicted_class = result.names[predicted_class_index] 
                confidence = result.probs.top1conf                    
                print(f"Image: {img_path}")
                print(f"Ground Truth: {ground_truth_class}, Predicted class: {predicted_class}, Confidence: {confidence:.2f}")

def deepseek_r1_response(history, msg):
    messages = []
    for chat in history:
        messages.append({"role": "user", "content": chat[0]})
        messages.append({"role": "assistant", "content": chat[1]})

    messages.append({"role": "user", "content": msg})

    openai_api_key = "123"
    openai_api_base = "http://localhost:8000/v1"
    client = OpenAI(
        api_key=openai_api_key,
        base_url=openai_api_base,
        http_client=httpx.Client(timeout=1800),
    )
    completion = client.chat.completions.create(
        model="deepseek-ai/DeepSeek-R1-Distill-Qwen-14B",
        messages=messages,
        temperature=0.7,
        top_p=0.85,
        extra_body={
            "repetition_penalty": 1.05,
        },
        stream=True,
    )

    history.append([msg, ""])
    for chunk in completion:
        for choice in chunk.choices:
            content = choice.delta.content
            if content:
                history[-1][1] = history[-1][1] + content
                time.sleep(0.05)
                yield history