1. 下载依赖

python -m pip install --upgrade pip
# 更换 pypi 源加速库的安装
pip config set global.index-url https://pypi.tuna.tsinghua.edu.cn/simple

#若有的版本不兼容，就更新为最新版
pip install modelscope==1.18.0
pip install transformers==4.46.2
pip install sentencepiece==0.2.0
pip install git+https://github.com/huggingface/transformers accelerate
pip install qwen-vl-utils[decord]==0.0.8 #从魔塔社区找版本，这个版本目前适配
pip install datasets==2.18.0
pip install peft==0.13.2
pip install swanlab==0.3.25
pip install qwen-vl-utils==0.0.8

2. 拉取源码

git clone https://github.com/QwenLM/Qwen2.5-VL.git

3. 下载模型

# 创建模型保存目录
mkdir -p ~/llm_models
mkdir -p ~/llm_models/Qwen2.5-VL #下载到你想下载的位置

# 下载模型
modelscope download --model Qwen/Qwen2.5-VL-7B-Instruct --cache_dir ~/llm_models/Qwen2.5-VL #改为自己的模型下载的路径

4. 测试代码 test.py，不测试直接看下面
   1. 更改模型路径，使用绝对路径
   2. 更改图片路径
5. 准备数据
   1. 使用labelimg等标注工具标注目标区域，xml格式
   2. 在同一目录下，将xml格式转化成jsonl格式，转化代码是convert_dataset.py
   3. 将json格式放到Qwen2.5vl目录下
6. 模型训练
   1. 更改模型名称和路径
   2. 运行train.py(Qwen2.5vl目录下)
   3. 修改测试模型名称，checkpoint-155，将155改为训练后实际的权重名称

val_peft_model = PeftModel.from_pretrained(model, model_id="/mnt/workspace/Qwen/Qwen2.5-VL/output/Qwen2.5-VL-7B/checkpoint-155", config=val_config)

7. 注册swanlab（用训练和测试的结果可视化），终端输入数字2，复制swanlab密钥并粘贴到终端（会有提示的）
8. 打开ip地址，查看训练和测试结果
   test.py、convert_dataset.py和train.py 依次在下面

from transformers import Qwen2_5_VLForConditionalGeneration, AutoProcessor
from qwen_vl_utils import process_vision_info

# default: Load the model on the available device(s)
model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
    "Qwen/Qwen2.5-VL-7B-Instruct", torch_dtype="auto", device_map="auto"
)

# We recommend enabling flash_attention_2 for better acceleration and memory saving, especially in multi-image and video scenarios.
# model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
#     "Qwen/Qwen2.5-VL-7B-Instruct",
#     torch_dtype=torch.bfloat16,
#     attn_implementation="flash_attention_2",
#     device_map="auto",
# )

# default processor
processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-7B-Instruct")

# The default range for the number of visual tokens per image in the model is 4-16384.
# You can set min_pixels and max_pixels according to your needs, such as a token range of 256-1280, to balance performance and cost.
# min_pixels = 256*28*28
# max_pixels = 1280*28*28
# processor = AutoProcessor.from_pretrained("Qwen/Qwen2.5-VL-7B-Instruct", min_pixels=min_pixels, max_pixels=max_pixels)

messages = [
    {
        "role": "user",
        "content": [
            {
                "type": "image",
                "image": "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen-VL/assets/demo.jpeg",
            },
            {"type": "text", "text": "Describe this image."},
        ],
    }
]

# Preparation for inference
text = processor.apply_chat_template(
    messages, tokenize=False, add_generation_prompt=True
)
image_inputs, video_inputs = process_vision_info(messages)
inputs = processor(
    text=[text],
    images=image_inputs,
    videos=video_inputs,
    padding=True,
    return_tensors="pt",
)
inputs = inputs.to(model.device)

# Inference: Generation of the output
generated_ids = model.generate(**inputs, max_new_tokens=128)
generated_ids_trimmed = [
    out_ids[len(in_ids) :] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
]
output_text = processor.batch_decode(
    generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
)
print(output_text)

import xml.etree.ElementTree as ET
import json
import os
import math
from PIL import Image


def smart_resize(height, width, factor=28, min_pixels=56 * 56, max_pixels=14 * 14 * 4 * 1280):
    """Qwen2.5-VL的resize函数"""
    if height < factor or width < factor:
        raise ValueError(f"height:{height} or width:{width} must be larger than factor:{factor}")
    elif max(height, width) / min(height, width) > 200:
        raise ValueError(
            f"absolute aspect ratio must be smaller than 200, got {max(height, width) / min(height, width)}"
        )
    h_bar = round(height / factor) * factor
    w_bar = round(width / factor) * factor
    if h_bar * w_bar > max_pixels:
        beta = math.sqrt((height * width) / max_pixels)
        h_bar = math.floor(height / beta / factor) * factor
        w_bar = math.floor(width / beta / factor) * factor
    elif h_bar * w_bar < min_pixels:
        beta = math.sqrt(min_pixels / (height * width))
        h_bar = math.ceil(height * beta / factor) * factor
        w_bar = math.ceil(width * beta / factor) * factor
    return h_bar, w_bar


def convert_to_qwen25vl_format(bbox, orig_height, orig_width, factor=28, min_pixels=56 * 56,
                               max_pixels=14 * 14 * 4 * 1280):
    """转换为Qwen2.5-VL格式"""
    new_height, new_width = smart_resize(orig_height, orig_width, factor, min_pixels, max_pixels)
    scale_w = new_width / orig_width
    scale_h = new_height / orig_height

    x1, y1, x2, y2 = bbox
    x1_new = round(x1 * scale_w)
    y1_new = round(y1 * scale_h)
    x2_new = round(x2 * scale_w)
    y2_new = round(y2 * scale_h)

    x1_new = max(0, min(x1_new, new_width - 1))
    y1_new = max(0, min(y1_new, new_height - 1))
    x2_new = max(0, min(x2_new, new_width - 1))
    y2_new = max(0, min(y2_new, new_height - 1))

    return [x1_new, y1_new, x2_new, y2_new]


def find_image_file(image_dir, xml_filename):
    """根据XML文件名找到对应的图像文件"""
    base_name = os.path.splitext(xml_filename)[0]
    for ext in ['.jpg', '.jpeg', '.png', '.bmp', '.tiff', '.JPG', '.JPEG', '.PNG']:
        image_file = base_name + ext
        image_path = os.path.join(image_dir, image_file)
        if os.path.exists(image_path):
            return image_file
    return None


def parse_xml(xml_path):
    """解析LabelImg生成的XML文件"""
    tree = ET.parse(xml_path)
    root = tree.getroot()

    # 获取图像信息
    filename = root.find('filename').text
    size = root.find('size')
    width = int(size.find('width').text)
    height = int(size.find('height').text)

    # 获取所有目标
    objects = []
    for obj in root.findall('object'):
        name = obj.find('name').text
        bndbox = obj.find('bndbox')
        xmin = int(float(bndbox.find('xmin').text))
        ymin = int(float(bndbox.find('ymin').text))
        xmax = int(float(bndbox.find('xmax').text))
        ymax = int(float(bndbox.find('ymax').text))

        objects.append({
            'label': name,
            'bbox': [xmin, ymin, xmax, ymax]
        })

    return filename, width, height, objects


def xml_to_qwen_format(xml_dir, image_dir, output_file, task_description=None):
    """将XML标注转换为Qwen2.5-VL训练格式"""

    if task_description is None:
        task_description = "Detect all objects in this image and return their locations in JSON format."

    training_data = []
    xml_files = [f for f in os.listdir(xml_dir) if f.endswith('.xml')]

    print(f"找到 {len(xml_files)} 个XML文件")

    for xml_file in xml_files:
        xml_path = os.path.join(xml_dir, xml_file)

        try:
            filename, width, height, objects = parse_xml(xml_path)

            if not objects:  # 跳过没有标注的图像
                print(f"跳过无标注文件: {xml_file}")
                continue

            # 查找对应的图像文件
            image_file = find_image_file(image_dir, xml_file)
            if not image_file:
                print(f"未找到对应图像文件: {xml_file}")
                continue

            # 转换边界框格式
            converted_objects = []
            for obj in objects:
                bbox = obj['bbox']
                converted_bbox = convert_to_qwen25vl_format(bbox, height, width)

                converted_objects.append({
                    "bbox_2d": converted_bbox,
                    "label": obj['label']
                })

            # 构建训练样本
            training_sample = {
                "image": image_file,  # 只存文件名，训练时会从image_dir加载
                "conversations": [
                    {
                        "from": "human",
                        "value": f"<image>\n{task_description}"
                    },
                    {
                        "from": "gpt",
                        "value": json.dumps(converted_objects, ensure_ascii=False)
                    }
                ]
            }

            training_data.append(training_sample)
            print(f"✓ 处理完成: {image_file}, 检测到 {len(objects)} 个目标")

        except Exception as e:
            print(f"✗ 处理 {xml_file} 时出错: {e}")
            continue

    # 保存训练数据（JSONL格式）
    with open(output_file, 'w', encoding='utf-8') as f:
        for sample in training_data:
            f.write(json.dumps(sample, ensure_ascii=False) + '\n')

    print(f"\n转换完成！共生成 {len(training_data)} 个训练样本")
    print(f"数据已保存到: {output_file}")
    return len(training_data)


# 使用示例
if __name__ == "__main__":
    # 配置路径
    xml_directory = "annotations"  # XML标注文件目录
    image_directory = "images"  # 图像文件目录
    output_jsonl = "train_data_detection.jsonl"  # 输出文件

    # 转换数据
    xml_to_qwen_format(
        xml_dir=xml_directory,
        image_dir=image_directory,
        output_file=output_jsonl,
        task_description="Detect all objects in this image and return their locations in JSON format."
    )

import torch
from datasets import Dataset
from modelscope import snapshot_download, AutoTokenizer
from swanlab.integration.transformers import SwanLabCallback
from qwen_vl_utils import process_vision_info
from peft import LoraConfig, TaskType, get_peft_model, PeftModel

from transformers import BitsAndBytesConfig
from peft import prepare_model_for_kbit_training

from transformers import (
    TrainingArguments,
    Trainer,
    DataCollatorForSeq2Seq,
    Qwen2_5_VLForConditionalGeneration,
    AutoProcessor,
)
import swanlab
import json
import os


def load_jsonl_data(jsonl_path):
    """加载JSONL数据并转换为原格式"""
    data = []
    with open(jsonl_path, 'r', encoding='utf-8') as f:
        for i, line in enumerate(f):
            item = json.loads(line.strip())
            # 转换为原来的格式，添加id
            data.append({
                "id": f"detection_{i + 1}",
                "image": item["image"],
                "conversations": item["conversations"]
            })
    return data


def process_func(example):
    """
    将数据集进行预处理 - 适配目标检测数据
    """
    MAX_LENGTH = 1024
    input_ids, attention_mask, labels = [], [], []
    conversation = example["conversations"]
    input_content = conversation[0]["value"]
    output_content = conversation[1]["value"]

    # 适配目标检测数据格式
    if "<|vision_start|>" in input_content and "<|vision_end|>" in input_content:
        # 原来的COCO格式
        file_path = input_content.split("<|vision_start|>")[1].split("<|vision_end|>")[0]
        prompt_text = "COCO Yes:"
    else:
        # 新的目标检测格式: "<image>\nDetect all objects..."
        # 从conversations中获取图像信息，需要从外部传入
        file_path = os.path.join("images", example.get("image", ""))  # 从example中获取图像路径
        prompt_text = input_content.replace("<image>\n", "")

    messages = [
        {
            "role": "user",
            "content": [
                {
                    "type": "image",
                    "image": f"{file_path}",
                    "resized_height": 256,
                    "resized_width": 256,
                },
                {"type": "text", "text": prompt_text},
            ],
        }
    ]
    text = processor.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    image_inputs, video_inputs = process_vision_info(messages)
    inputs = processor(
        text=[text],
        images=image_inputs,
        videos=video_inputs,
        padding=True,
        return_tensors="pt",
    )
    inputs = {key: value.tolist() for key, value in inputs.items()}
    instruction = inputs

    response = tokenizer(f"{output_content}", add_special_tokens=False)

    input_ids = (
            instruction["input_ids"][0] + response["input_ids"] + [tokenizer.pad_token_id]
    )

    attention_mask = instruction["attention_mask"][0] + response["attention_mask"] + [1]
    labels = (
            [-100] * len(instruction["input_ids"][0])
            + response["input_ids"]
            + [tokenizer.pad_token_id]
    )
    if len(input_ids) > MAX_LENGTH:
        input_ids = input_ids[:MAX_LENGTH]
        attention_mask = attention_mask[:MAX_LENGTH]
        labels = labels[:MAX_LENGTH]

    input_ids = torch.tensor(input_ids)
    attention_mask = torch.tensor(attention_mask)
    labels = torch.tensor(labels)
    inputs['pixel_values'] = torch.tensor(inputs['pixel_values'])
    inputs['image_grid_thw'] = torch.tensor(inputs['image_grid_thw']).squeeze(0)
    return {"input_ids": input_ids, "attention_mask": attention_mask, "labels": labels,
            "pixel_values": inputs['pixel_values'], "image_grid_thw": inputs['image_grid_thw']}


def predict(messages, model):
    # 准备推理
    text = processor.apply_chat_template(
        messages, tokenize=False, add_generation_prompt=True
    )
    image_inputs, video_inputs = process_vision_info(messages)
    inputs = processor(
        text=[text],
        images=image_inputs,
        videos=video_inputs,
        padding=True,
        return_tensors="pt",
    )
    inputs = inputs.to("cuda")

    # 生成输出 - 增加token数适应检测结果
    generated_ids = model.generate(**inputs, max_new_tokens=256)
    generated_ids_trimmed = [
        out_ids[len(in_ids):] for in_ids, out_ids in zip(inputs.input_ids, generated_ids)
    ]
    output_text = processor.batch_decode(
        generated_ids_trimmed, skip_special_tokens=True, clean_up_tokenization_spaces=False
    )

    return output_text[0]


# 使用Transformers加载模型权重
tokenizer = AutoTokenizer.from_pretrained("/mnt/workspace/Qwen/Qwen2.5-VL-7B-Instruct/", use_fast=False,
                                          trust_remote_code=True)
processor = AutoProcessor.from_pretrained("/mnt/workspace/Qwen/Qwen2.5-VL-7B-Instruct")


# model = Qwen2_5_VLForConditionalGeneration.from_pretrained("/mnt/workspace/Qwen/Qwen2.5-VL-7B-Instruct/",
#                                                            device_map="auto", torch_dtype=torch.bfloat16,
#                                                            trust_remote_code=True, )
# model.enable_input_require_grads()

# 处理数据集：读取JSONL文件并转换
train_json_path = "train_data_detection.jsonl"  # 改为你的JSONL文件
data = load_jsonl_data(train_json_path)

# 拆分训练和测试数据
train_data = data[:-3] if len(data) > 3 else data
test_data = data[-3:] if len(data) > 3 else data

# 保存为JSON格式（保持原来的逻辑）
with open("data_vl_train.json", "w") as f:
    json.dump(train_data, f)

with open("data_vl_test.json", "w") as f:
    json.dump(test_data, f)

train_ds = Dataset.from_json("data_vl_train.json")
train_dataset = train_ds.map(process_func)

bnb_config = BitsAndBytesConfig(
    load_in_4bit=True,
    bnb_4bit_quant_type="nf4",
    bnb_4bit_compute_dtype=torch.bfloat16,
)

model = Qwen2_5_VLForConditionalGeneration.from_pretrained(
    "/mnt/workspace/Qwen/Qwen2.5-VL-7B-Instruct/",
    quantization_config=bnb_config,  # 4bit
    device_map="auto",
    torch_dtype=torch.bfloat16,
    trust_remote_code=True,
)

model = prepare_model_for_kbit_training(model)  # 去掉gradient_checkpointing参数

config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],  # 去掉gate_proj等
    inference_mode=False,
    r=64,
    lora_alpha=16,
    lora_dropout=0.05,
    bias="none",
)

peft_model = get_peft_model(model, config)

args = TrainingArguments(
    output_dir="./output/Qwen2.5-VL-Detection",  # 改个名字
    per_device_train_batch_size=2,  # 减小batch size
    gradient_accumulation_steps=8,
    logging_steps=10,
    logging_first_step=5,
    num_train_epochs=10,  # 增加epoch
    save_steps=100,
    learning_rate=5e-5,  # 降低学习率
    save_on_each_node=True,
    gradient_checkpointing=True,
    report_to="none",
)

swanlab_callback = SwanLabCallback(
    project="Qwen2.5-VL-detection",
    experiment_name="qwen2.5-vl-object-detection",
    config={
        "model": "https://modelscope.cn/models/Qwen/Qwen2.5-VL-7B-Instruct",
        "dataset": "custom_detection_dataset",
        "prompt": "Object Detection",
        "train_data_number": len(train_data),
        "lora_rank": 64,
        "lora_alpha": 16,
        "lora_dropout": 0.05,
    },
)

trainer = Trainer(
    model=peft_model,
    args=args,
    train_dataset=train_dataset,
    data_collator=DataCollatorForSeq2Seq(tokenizer=tokenizer, padding=True),
    callbacks=[swanlab_callback],
)

# 开启模型训练
trainer.train()

# ====================测试模式===================
val_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    # target_modules=["q_proj", "k_proj", "v_proj", "o_proj", "gate_proj", "up_proj", "down_proj"],
    target_modules=["q_proj", "k_proj", "v_proj", "o_proj"],
    inference_mode=True,
    r=8,
    lora_alpha=16,
    lora_dropout=0.05,
    bias="none",
)

# 这里需要改为你的实际checkpoint路径
val_peft_model = PeftModel.from_pretrained(model,
                                           model_id="./output/Qwen2.5-VL-Detection/checkpoint-184",  # 需要根据实际情况修改
                                           config=val_config)

# 读取测试数据
with open("data_vl_test.json", "r") as f:
    test_dataset = json.load(f)

test_image_list = []
for item in test_dataset:
    # 适配新的数据格式
    conversations = item["conversations"]

    # 获取图像路径
    if "<|vision_start|>" in conversations[0]["value"]:
        # 原格式
        origin_image_path = conversations[0]["value"].split("<|vision_start|>")[1].split("<|vision_end|>")[0]
        prompt_text = "COCO Yes:"
    else:
        # 新格式，需要从item中获取图像信息
        origin_image_path = os.path.join("images", item.get("image", ""))
        prompt_text = "Detect all objects in this image and return their locations in JSON format."

    messages = [{
        "role": "user",
        "content": [
            {
                "type": "image",
                "image": origin_image_path
            },
            {
                "type": "text",
                "text": prompt_text
            }
        ]}]

    response = predict(messages, val_peft_model)
    messages.append({"role": "assistant", "content": f"{response}"})
    print(f"图片: {origin_image_path}")
    print(f"预测: {response}")
    print(f"真实: {conversations[1]['value']}")
    print("-" * 50)

    test_image_list.append(swanlab.Image(origin_image_path, caption=response))

swanlab.log({"Prediction": test_image_list})
swanlab.finish()