第 17 章 视觉模型实战与部署¶

import torch
from transformers import (
    Qwen2VLForConditionalGeneration,
    AutoProcessor,
    TrainingArguments,
    Trainer,
)
from peft import LoraConfig, get_peft_model, TaskType
from datasets import load_dataset
from PIL import Image

# ==================== Step1: 加载基础模型 ====================
model_name = "Qwen/Qwen2-VL-7B-Instruct"

model = Qwen2VLForConditionalGeneration.from_pretrained(
    model_name,
    torch_dtype=torch.bfloat16,
    attn_implementation="flash_attention_2",  # 使用Flash Attention 2加速
)
processor = AutoProcessor.from_pretrained(model_name)

# ==================== Step2: 配置LoRA ====================
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    r=64,                          # LoRA秩，越大表达能力越强但显存越大
    lora_alpha=16,                 # 缩放系数
    target_modules=[               # 对LLM的注意力层添加LoRA
        "q_proj", "k_proj", "v_proj", "o_proj",
        "gate_proj", "up_proj", "down_proj",
    ],
    lora_dropout=0.05,
    modules_to_save=["visual"],    # 视觉编码器的投影层也参与训练
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# 输出示例: trainable params: 83,886,080 || all params: 8,112,345,088 || trainable%: 1.034%

# ==================== Step3: 数据加载与处理 ====================
def preprocess_function(examples):
    """预处理VLM微调数据"""
    messages_list = []
    for conv in examples["conversations"]:
        messages = []
        for turn in conv:
            if turn["from"] == "human":
                content = []
                if "<image>" in turn["value"]:
                    content.append({"type": "image", "image": examples["image"]})
                    text = turn["value"].replace("<image>\n", "").replace("<image>", "")
                else:
                    text = turn["value"]
                content.append({"type": "text", "text": text})
                messages.append({"role": "user", "content": content})
            else:
                messages.append({"role": "assistant", "content": turn["value"]})
        messages_list.append(messages)

    # 使用processor处理多模态输入
    texts = [processor.apply_chat_template(m, tokenize=False) for m in messages_list]
    batch = processor(text=texts, padding=True, truncation=True,
                      max_length=2048, return_tensors="pt")
    batch["labels"] = batch["input_ids"].clone()
    return batch

dataset = load_dataset("json", data_files="train_data.json", split="train")
train_dataset = dataset.map(preprocess_function, batched=True, batch_size=4)

# ==================== Step4: 训练配置 ====================
training_args = TrainingArguments(
    output_dir="./qwen2vl-lora-medical",
    num_train_epochs=3,
    per_device_train_batch_size=2,
    gradient_accumulation_steps=8,  # 等效batch_size = 2 * 8 = 16
    learning_rate=1e-4,
    warmup_ratio=0.03,
    lr_scheduler_type="cosine",
    bf16=True,
    logging_steps=10,
    save_strategy="steps",
    save_steps=200,
    dataloader_num_workers=4,
    gradient_checkpointing=True,    # 显存优化：用时间换空间
    report_to="wandb",              # 实验追踪
    remove_unused_columns=False,    # 多模态字段不能被 Trainer 自动丢弃
)

# ==================== Step5: 开始训练 ====================
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
)
trainer.train()

# ==================== Step6: 保存LoRA权重 ====================
model.save_pretrained("./qwen2vl-lora-medical/final")
processor.save_pretrained("./qwen2vl-lora-medical/final")

from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
from torch.distributed.fsdp import MixedPrecision, ShardingStrategy

# FSDP配置
fsdp_config = {
    "sharding_strategy": ShardingStrategy.FULL_SHARD,  # 类似ZeRO-3
    "mixed_precision": MixedPrecision(
        param_dtype=torch.bfloat16,
        reduce_dtype=torch.bfloat16,
        buffer_dtype=torch.bfloat16,
    ),
    "activation_checkpointing": True,
    "cpu_offload": False,
}

# 在HuggingFace Trainer中使用FSDP
training_args = TrainingArguments(
    fsdp="full_shard auto_wrap",
    fsdp_config=fsdp_config,
    # ... 其他参数
)

# BitsAndBytes 4-bit 量化示例：Qwen2-VL 推理侧更稳妥的入门方案
import torch
from transformers import Qwen2VLForConditionalGeneration, AutoProcessor, BitsAndBytesConfig

model_path = "Qwen/Qwen2-VL-7B-Instruct"
quant_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_use_double_quant=True,
    bnb_4bit_compute_dtype=torch.bfloat16,
)

model = Qwen2VLForConditionalGeneration.from_pretrained(
    model_path,
    quantization_config=quant_config,
    device_map="auto",
)
processor = AutoProcessor.from_pretrained(model_path)
print('已按 4-bit 方式加载 Qwen2-VL，可继续进行推理或 LoRA 适配训练。')

import tensorrt as trt
import torch
import onnx

# Step1: PyTorch模型导出为ONNX
dummy_input = torch.randn(1, 3, 640, 640).cuda()
torch.onnx.export(
    model,
    dummy_input,
    "yolo_world.onnx",
    opset_version=17,
    input_names=["images"],
    output_names=["output"],
    dynamic_axes={"images": {0: "batch"}, "output": {0: "batch"}},
)

# Step2: ONNX → TensorRT Engine
def build_trt_engine(onnx_path, engine_path, fp16=True):
    """构建TensorRT推理引擎"""
    logger = trt.Logger(trt.Logger.WARNING)
    builder = trt.Builder(logger)
    network = builder.create_network(
        1 << int(trt.NetworkDefinitionCreationFlag.EXPLICIT_BATCH)
    )
    parser = trt.OnnxParser(network, logger)

    with open(onnx_path, "rb") as f:  # with自动管理文件关闭
        parser.parse(f.read())

    config = builder.create_builder_config()
    config.set_memory_pool_limit(trt.MemoryPoolType.WORKSPACE, 4 << 30)  # 4GB

    if fp16:
        config.set_flag(trt.BuilderFlag.FP16)

    # 构建引擎
    engine_bytes = builder.build_serialized_network(network, config)
    with open(engine_path, "wb") as f:
        f.write(engine_bytes)
    print(f"TensorRT engine已保存: {engine_path}")

build_trt_engine("yolo_world.onnx", "yolo_world.engine", fp16=True)

import onnxruntime as ort
import numpy as np
from PIL import Image
from torchvision import transforms

# 创建推理会话
providers = [
    ("CUDAExecutionProvider", {"device_id": 0}),
    "CPUExecutionProvider",
]
session = ort.InferenceSession("vision_model.onnx", providers=providers)

# 推理
transform = transforms.Compose([
    transforms.Resize((640, 640)),
    transforms.ToTensor(),
    transforms.Normalize([0.485, 0.456, 0.406], [0.229, 0.224, 0.225]),
])

image = Image.open("test.jpg")
input_tensor = transform(image).unsqueeze(0).numpy()  # unsqueeze增加一个维度

outputs = session.run(None, {"images": input_tensor})
print(f"推理完成，输出shape: {outputs[0].shape}")

# 基准测试
import time
times = []
for _ in range(100):
    start = time.perf_counter()
    session.run(None, {"images": input_tensor})
    times.append(time.perf_counter() - start)

print(f"平均推理时间: {np.mean(times)*1000:.1f}ms, FPS: {1/np.mean(times):.1f}")

from vllm import LLM, SamplingParams

# 初始化vLLM多模态引擎
llm = LLM(
    model="Qwen/Qwen2-VL-7B-Instruct",
    tensor_parallel_size=1,       # GPU数量
    max_model_len=4096,
    gpu_memory_utilization=0.9,
    trust_remote_code=True,
    dtype="bfloat16",
    limit_mm_per_prompt={"image": 5},  # 每个prompt最多5张图
)

sampling_params = SamplingParams(
    temperature=0.7,
    top_p=0.9,
    max_tokens=1024,
)

# 多模态推理
from PIL import Image

image = Image.open("test.jpg")
prompt = "<|im_start|>user\n<image>\n请描述这张图片<|im_end|>\n<|im_start|>assistant\n"

outputs = llm.generate(
    [{
        "prompt": prompt,
        "multi_modal_data": {"image": image},
    }],
    sampling_params=sampling_params,
)

print(outputs[0].outputs[0].text)

# NCNN模型转换与部署示例
# Step1: PyTorch → ONNX → NCNN
# 终端执行:
# onnx2ncnn model.onnx model.param model.bin

# Step2: Python调用NCNN推理
import ncnn
import numpy as np
import cv2

net = ncnn.Net()
net.opt.use_vulkan_compute = True  # 启用GPU加速
net.load_param("model.param")
net.load_model("model.bin")

# 预处理
img = cv2.imread("test.jpg")
mat_in = ncnn.Mat.from_pixels_resize(
    img, ncnn.Mat.PixelType.PIXEL_BGR2RGB,
    img.shape[1], img.shape[0], 640, 640
)
mat_in.substract_mean_normalize([123.675, 116.28, 103.53], [1/58.395, 1/57.12, 1/57.375])

# 推理
ex = net.create_extractor()
ex.input("input", mat_in)
ret, mat_out = ex.extract("output")
output = np.array(mat_out)  # np.array创建NumPy数组
print(f"端侧推理输出shape: {output.shape}")

import wandb
from PIL import Image

# W&B视觉实验追踪
wandb.init(project="cv-model-training", name="qwen2vl-lora-medical-v3")

# 记录训练配置
wandb.config.update({
    "model": "Qwen2-VL-7B",
    "lora_r": 64,
    "learning_rate": 1e-4,
    "dataset_size": 10000,
    "resolution": "dynamic",
})

# 训练循环中记录指标
for epoch in range(num_epochs):
    # ... 训练代码 ...
    wandb.log({
        "train/loss": train_loss,
        "train/lr": current_lr,
        "eval/accuracy": eval_acc,
        "eval/bleu": eval_bleu,
    })

    # 记录预测样例（视觉模型特有）
    sample_images = []
    for img_path, pred, gt in eval_samples[:10]:  # 切片操作，取前n个元素
        sample_images.append(wandb.Image(
            img_path,
            caption=f"Pred: {pred[:100]}\nGT: {gt[:100]}"
        ))
    wandb.log({"eval/predictions": sample_images})

wandb.finish()

import bentoml
import torch
from PIL import Image

@bentoml.service(
    resources={"gpu": 1, "memory": "16Gi"},
    traffic={"timeout": 60},
)
class VisionModelService:
    def __init__(self):
        self.model = self._load_model()
        self.processor = self._load_processor()

    @bentoml.api
    async def predict(self, image: Image.Image, question: str) -> str:  # async定义异步函数
        inputs = self.processor(
            images=image, text=question, return_tensors="pt"
        ).to("cuda")  # 移至GPU/CPU
        with torch.no_grad():  # 禁用梯度计算，节省内存
            output = self.model.generate(**inputs, max_new_tokens=256)
        return self.processor.decode(output[0], skip_special_tokens=True)