# 早期融合示例
def early_fusion(text_features, image_features):
    # 在輸入層或淺層融合特征
    combined = torch.cat([text_features, image_features], dim=1)
    fused = fusion_transformer(combined)
    return fused

# 中期融合示例  
def mid_fusion(text_embeddings, image_embeddings):
    # 通過(guò)交叉注意力機(jī)制實(shí)現(xiàn)模態(tài)交互
    cross_attention = CrossModalAttention(
        text_embeddings, image_embeddings, image_embeddings
    )
    return cross_attention

# 晚期融合示例
def late_fusion(text_logits, image_logits):
    # 分別處理不同模態(tài)，最后融合決策
    combined_logits = text_logits * 0.6 + image_logits * 0.4
    return combined_logits

# 安裝依賴
!pip install torch==2.1.0 torchvision==0.16.0
!pip install transformers==4.37.0 datasets==2.14.0
!pip install accelerate==0.25.0 peft==0.8.0 bitsandbytes==0.41.0
!pip install pillow==10.1.0 opencv-python==4.8.1
!pip install qwen-vl-utils==0.0.3


# 驗(yàn)證安裝
import torch
import transformers
print(f"PyTorch版本: {torch.__version__}")
print(f"Transformers版本: {transformers.__version__}")

import base64
from PIL import Image
from io import BytesIO
import json
from datasets import Dataset, DatasetDict
import random
from typing import List, Dict, Any


class MultimodalDataProcessor:
    """多模態(tài)數(shù)據(jù)處理器"""


    def __init__(self, model_name: str = "Qwen/Qwen-VL-Chat"):
        from transformers import AutoProcessor
        self.processor = AutoProcessor.from_pretrained(
            model_name, 
            trust_remote_code=True
        )
        self.image_size = (224, 224)


    def process_image(self, image_path: str) -> str:
        """處理圖像并編碼為base64"""
        try:
            with Image.open(image_path) as img:
                # 統(tǒng)一圖像尺寸
                img = img.resize(self.image_size, Image.Resampling.LANCZOS)
                # 轉(zhuǎn)換為RGB（處理可能出現(xiàn)的RGBA或灰度圖）
                if img.mode != 'RGB':
                    img = img.convert('RGB')


                buffered = BytesIO()
                img.save(buffered, format="JPEG", quality=90)
                img_str = base64.b64encode(buffered.getvalue()).decode()
                return f"data:image/jpeg;base64,{img_str}"
        except Exception as e:
            print(f"圖像處理錯(cuò)誤 {image_path}: {e}")
            return ""


    def create_multimodal_example(self, text1: str, text2: str, 
                                image_path: str, label: int) -> Dict[str, Any]:
        """創(chuàng)建多模態(tài)訓(xùn)練樣本"""


        image_data = self.process_image(image_path)
        if not image_data:
            return None


        # 構(gòu)建對(duì)話格式
        if label == 1:
            answer = "相關(guān)"
            reasoning = "圖像內(nèi)容與兩個(gè)文本描述高度一致，它們?cè)谡Z(yǔ)義上緊密相關(guān)。"
        else:
            answer = "不相關(guān)" 
            reasoning = "圖像內(nèi)容與文本描述存在明顯差異，它們?cè)谡Z(yǔ)義上沒(méi)有關(guān)聯(lián)。"


        messages = [
            {
                "role": "user",
                "content": [
                    {"type": "image", "image": image_data},
                    {"type": "text", "text": f"請(qǐng)分析圖像并判斷文本相關(guān)性：\n文本1：{text1}\n文本2：{text2}"}
                ]
            },
            {
                "role": "assistant",
                "content": [
                    {"type": "text", "text": f"分析：{reasoning}\n結(jié)論：{answer}"}
                ]
            }
        ]


        return {
            "messages": messages,
            "text1": text1,
            "text2": text2, 
            "image_path": image_path,
            "label": label,
            "reasoning": reasoning
        }


    def generate_dataset(self, text_pairs: List, image_paths: List, 
                        num_samples: int = 1000) -> Dataset:
        """生成多模態(tài)數(shù)據(jù)集"""


        data = []
        for _ in range(num_samples):
            # 隨機(jī)選擇文本對(duì)和圖像
            text_pair = random.choice(text_pairs)
            image_path = random.choice(image_paths)


            example = self.create_multimodal_example(
                text_pair[0], text_pair[1], image_path, text_pair[2]
            )
            if example:
                data.append(example)


        return Dataset.from_list(data)


    def save_dataset(self, dataset: Dataset, output_path: str):
        """保存數(shù)據(jù)集"""
        dataset.to_json(output_path, orient="records", lines=True)


    def load_dataset(self, data_path: str) -> DatasetDict:
        """加載數(shù)據(jù)集"""
        dataset = Dataset.from_json(data_path)


        # 數(shù)據(jù)集分割
        splits = dataset.train_test_split(test_size=0.2, seed=42)
        train_val = splits["train"].train_test_split(test_size=0.1, seed=42)


        return DatasetDict({
            "train": train_val["train"],
            "validation": train_val["test"], 
            "test": splits["test"]
        })

import torch
import torch.nn as nn
from transformers import (
    AutoModelForVision2Seq,
    AutoProcessor,
    BitsAndBytesConfig,
    TrainingArguments,
    Trainer
)
from peft import LoraConfig, get_peft_model, prepare_model_for_kbit_training


class MultimodalMatchingModel:
    """多模態(tài)匹配模型"""


    def __init__(self, model_name: str = "Qwen/Qwen-VL-Chat"):
        self.model_name = model_name
        self.setup_model()


    def setup_model(self):
        """初始化模型配置"""


        # 4-bit量化配置
        bnb_config = BitsAndBytesConfig(
            load_in_4bit=True,
            bnb_4bit_use_double_quant=True,
            bnb_4bit_quant_type="nf4",
            bnb_4bit_compute_dtype=torch.bfloat16
        )


        # 加載多模態(tài)模型
        self.model = AutoModelForVision2Seq.from_pretrained(
            self.model_name,
            quantization_config=bnb_config,
            device_map="auto",
            trust_remote_code=True,
            torch_dtype=torch.bfloat16
        )


        # 準(zhǔn)備k-bit訓(xùn)練
        self.model = prepare_model_for_kbit_training(self.model)


        # LoRA配置 - 針對(duì)多模態(tài)特性優(yōu)化
        lora_config = LoraConfig(
            r=16,  # LoRA秩
            lora_alpha=32,  # 縮放參數(shù)
            target_modules=[
                "q_proj", "k_proj", "v_proj", "o_proj",
                "gate_proj", "up_proj", "down_proj", 
                "vision_proj"  # 視覺(jué)投影層
            ],
            lora_dropout=0.05,
            bias="none",
            task_type="CAUSAL_LM",
        )


        # 應(yīng)用LoRA
        self.model = get_peft_model(self.model, lora_config)


        self._print_trainable_parameters()


    def _print_trainable_parameters(self):
        """打印可訓(xùn)練參數(shù)信息"""
        trainable_params = 0
        all_params = 0
        for _, param in self.model.named_parameters():
            all_params += param.numel()
            if param.requires_grad:
                trainable_params += param.numel()


        print(f"可訓(xùn)練參數(shù): {trainable_params:,}")
        print(f"總參數(shù): {all_params:,}") 
        print(f"訓(xùn)練參數(shù)占比: {100 * trainable_params / all_params:.2f}%")


    def train(self, dataset_dict: DatasetDict, processor, output_dir: str):
        """訓(xùn)練模型"""


        # 數(shù)據(jù)整理器
        collator = MultimodalCollator(processor)


        # 訓(xùn)練參數(shù)
        training_args = TrainingArguments(
            output_dir=output_dir,
            num_train_epochs=5,
            per_device_train_batch_size=2,  # 多模態(tài)需要較小批次
            per_device_eval_batch_size=2,
            gradient_accumulation_steps=4,
            learning_rate=1e-4,  # 多模態(tài)學(xué)習(xí)率通常較小
            warmup_ratio=0.1,
            logging_steps=50,
            eval_steps=200,
            save_steps=500,
            save_total_limit=3,
            evaluation_strategy="steps",
            load_best_model_at_end=True,
            metric_for_best_model="eval_loss",
            greater_is_better=False,
            fp16=True,
            remove_unused_columns=False,
            dataloader_pin_memory=False,
            report_to=["tensorboard"],
        )


        # 創(chuàng)建訓(xùn)練器
        trainer = Trainer(
            model=self.model,
            args=training_args,
            train_dataset=dataset_dict["train"],
            eval_dataset=dataset_dict["validation"],
            data_collator=collator,
            tokenizer=processor.tokenizer,
        )


        # 開始訓(xùn)練
        print("開始多模態(tài)訓(xùn)練...")
        train_result = trainer.train()


        # 保存模型
        trainer.save_model()
        trainer.save_state()


        metrics = train_result.metrics
        trainer.log_metrics("train", metrics)
        trainer.save_metrics("train", metrics)


        print(f"訓(xùn)練完成! 模型保存在: {output_dir}")
        return trainer


class MultimodalCollator:
    """多模態(tài)數(shù)據(jù)整理器"""


    def __init__(self, processor):
        self.processor = processor


    def __call__(self, batch):
        """處理批次數(shù)據(jù)"""
        messages_batch = [item["messages"] for item in batch]


        # 使用processor處理多模態(tài)輸入
        processed = self.processor(
            messages_batch,
            padding=True,
            return_tensors="pt",
        )


        # 添加標(biāo)簽
        labels = torch.tensor([item["label"] for item in batch])
        processed["labels"] = labels


        return processed

from tqdm import tqdm
from sklearn.metrics import accuracy_score, f1_score, classification_report
import numpy as np


class MultimodalEvaluator:
    """多模態(tài)評(píng)估器"""


    def __init__(self, model, processor):
        self.model = model
        self.processor = processor
        self.device = next(model.parameters()).device


    def predict_single(self, text1: str, text2: str, image_path: str) -> Dict[str, Any]:
        """單樣本預(yù)測(cè)"""


        # 處理圖像
        processor = MultimodalDataProcessor()
        image_data = processor.process_image(image_path)


        if not image_data:
            return {"prediction": "不相關(guān)", "confidence": 0.0}


        # 構(gòu)建推理消息
        messages = [
            {
                "role": "user", 
                "content": [
                    {"type": "image", "image": image_data},
                    {"type": "text", "text": f"分析圖像并判斷文本相關(guān)性，只回答'相關(guān)'或'不相關(guān)'：\n文本1：{text1}\n文本2：{text2}"}
                ]
            }
        ]


        try:
            # 處理輸入
            inputs = self.processor(
                messages, 
                padding=True, 
                return_tensors="pt"
            ).to(self.device)


            # 生成響應(yīng)
            with torch.no_grad():
                outputs = self.model.generate(
                    **inputs,
                    max_new_tokens=10,
                    do_sample=False,
                    temperature=0.1,
                    pad_token_id=self.processor.tokenizer.eos_token_id
                )


            # 解碼響應(yīng)
            response = self.processor.decode(outputs[0], skip_special_tokens=True)


            # 解析結(jié)果
            if "相關(guān)" in response:
                return {"prediction": "相關(guān)", "confidence": 0.9, "response": response}
            else:
                return {"prediction": "不相關(guān)", "confidence": 0.9, "response": response}


        except Exception as e:
            print(f"推理錯(cuò)誤: {e}")
            return {"prediction": "不相關(guān)", "confidence": 0.0, "error": str(e)}


    def evaluate_dataset(self, test_dataset, batch_size: int = 2) -> Dict[str, Any]:
        """數(shù)據(jù)集評(píng)估"""


        predictions = []
        ground_truth = []
        confidences = []


        for i in tqdm(range(0, len(test_dataset), batch_size)):
            batch = test_dataset[i:i+batch_size]


            for item in batch:
                result = self.predict_single(
                    item["text1"], item["text2"], item["image_path"]
                )


                pred_label = 1 if result["prediction"] == "相關(guān)" else 0
                true_label = item["label"]


                predictions.append(pred_label)
                ground_truth.append(true_label)
                confidences.append(result["confidence"])


        # 計(jì)算指標(biāo)
        accuracy = accuracy_score(ground_truth, predictions)
        f1 = f1_score(ground_truth, predictions)
        avg_confidence = np.mean(confidences)


        metrics = {
            "accuracy": accuracy,
            "f1_score": f1,
            "average_confidence": avg_confidence,
            "total_samples": len(predictions)
        }


        # 詳細(xì)報(bào)告
        report = classification_report(ground_truth, predictions, output_dict=True)


        return {
            "metrics": metrics,
            "detailed_report": report,
            "predictions": predictions,
            "confidences": confidences
        }


    def create_comprehensive_report(self, test_dataset, output_path: str):
        """創(chuàng)建綜合評(píng)估報(bào)告"""


        print("開始多模態(tài)評(píng)估...")
        results = self.evaluate_dataset(test_dataset)


        report_data = {
            **results,
            "evaluation_time": str(np.datetime64('now')),
            "model_type": "multimodal",
            "test_set_size": len(test_dataset)
        }


        # 保存報(bào)告
        import json
        with open(output_path, 'w', encoding='utf-8') as f:
            json.dump(report_data, f, indent=2, ensure_ascii=False)


        print(f"評(píng)估報(bào)告已保存: {output_path}")
        return report_data

# 示例數(shù)據(jù)生成
def prepare_training_data():
    """準(zhǔn)備訓(xùn)練數(shù)據(jù)"""


    # 文本對(duì)示例（實(shí)際應(yīng)用中替換為真實(shí)數(shù)據(jù)）
    text_pairs = [
        ("一只可愛(ài)的貓咪", "毛茸茸的寵物在玩耍", 1),
        ("美麗的日落景象", "黃昏時(shí)分的天空色彩", 1),
        ("現(xiàn)代城市建筑", "高樓林立的都市風(fēng)光", 1),
        ("一只可愛(ài)的貓咪", "汽車在高速公路上行駛", 0),
        ("美麗的日落景象", "計(jì)算機(jī)編程代碼界面", 0),
        ("現(xiàn)代城市建筑", "海底珊瑚礁生態(tài)系統(tǒng)", 0),
    ]


    # 圖像路徑（實(shí)際應(yīng)用中替換為真實(shí)路徑）
    image_paths = [
        "path/to/cat_images/",
        "path/to/sunset_images/", 
        "path/to/city_images/",
        "path_to_various_images/"
    ]


    processor = MultimodalDataProcessor()
    dataset = processor.generate_dataset(text_pairs, image_paths, 1000)
    processor.save_dataset(dataset, "multimodal_training_data.json")


    return processor.load_dataset("multimodal_training_data.json")

def train_multimodal_model():
    """完整訓(xùn)練流程"""


    print("=== 多模態(tài)短文本匹配訓(xùn)練 ===")


    # 1. 數(shù)據(jù)準(zhǔn)備
    print("步驟1: 準(zhǔn)備數(shù)據(jù)...")
    dataset_dict = prepare_training_data()


    # 2. 初始化處理器和模型
    print("步驟2: 初始化模型...")
    processor = MultimodalDataProcessor()
    model = MultimodalMatchingModel()


    # 3. 開始訓(xùn)練
    print("步驟3: 開始訓(xùn)練...")
    trainer = model.train(
        dataset_dict, 
        processor.processor, 
        "./multimodal_match_output"
    )


    # 4. 評(píng)估模型
    print("步驟4: 評(píng)估模型...")
    evaluator = MultimodalEvaluator(model.model, processor.processor)
    report = evaluator.create_comprehensive_report(
        dataset_dict["test"], 
        "multimodal_evaluation.json"
    )


    # 輸出結(jié)果
    metrics = report["metrics"]
    print(f"\n=== 訓(xùn)練結(jié)果 ===")
    print(f"準(zhǔn)確率: {metrics['accuracy']:.4f}")
    print(f"F1分?jǐn)?shù): {metrics['f1_score']:.4f}")
    print(f"平均置信度: {metrics['average_confidence']:.4f}")
    print(f"測(cè)試樣本數(shù): {metrics['total_samples']}")


    return model, processor, report

class MultimodalMatchingSystem:
    """多模態(tài)匹配系統(tǒng)"""


    def __init__(self, model_path: str, processor_path: str = None):
        from transformers import AutoProcessor, AutoModelForVision2Seq


        self.processor = AutoProcessor.from_pretrained(
            processor_path or model_path, 
            trust_remote_code=True
        )
        self.model = AutoModelForVision2Seq.from_pretrained(
            model_path,
            device_map="auto",
            trust_remote_code=True,
            torch_dtype=torch.bfloat16
        )
        self.data_processor = MultimodalDataProcessor()


    def predict(self, text1: str, text2: str, image_path: str) -> Dict[str, Any]:
        """預(yù)測(cè)接口"""


        result = self.predict_single(text1, text2, image_path)


        return {
            "text1": text1,
            "text2": text2, 
            "image_path": image_path,
            "prediction": result["prediction"],
            "confidence": result["confidence"],
            "related": result["prediction"] == "相關(guān)"
        }


    def batch_predict(self, inputs: List[Dict]) -> List[Dict]:
        """批量預(yù)測(cè)"""
        results = []
        for input_item in inputs:
            result = self.predict(
                input_item["text1"],
                input_item["text2"], 
                input_item["image_path"]
            )
            results.append(result)
        return results


# 使用示例
def demo_system():
    """系統(tǒng)演示"""
    system = MultimodalMatchingSystem("./multimodal_match_output")


    test_cases = [
        {
            "text1": "一只白色貓咪",
            "text2": "可愛(ài)的寵物在沙發(fā)上",
            "image_path": "path/to/cat_image.jpg"
        },
        {
            "text1": "城市夜景", 
            "text2": "海灘日落景色",
            "image_path": "path/to/city_image.jpg"
        }
    ]


    results = system.batch_predict(test_cases)
    for result in results:
        print(f"文本1: {result['text1']}")
        print(f"文本2: {result['text2']}")
        print(f"預(yù)測(cè): {result['prediction']} (置信度: {result['confidence']:.2f})")
        print(f"是否相關(guān): {result['related']}\n")

# 梯度檢查點(diǎn)
training_args = TrainingArguments(
    gradient_checkpointing=True,
    gradient_checkpointing_kwargs={"use_reentrant": False}
)


# 混合精度訓(xùn)練
training_args = TrainingArguments(
    fp16=True,  # 或 bf16=True
    tf32=True   # 如果硬件支持
)


# 梯度累積
training_args = TrainingArguments(
    per_device_train_batch_size=1,
    gradient_accumulation_steps=8
)