AutoModelForCausalLM            # 自動加載因果語言模型（如GPT系列）
AutoModelForSequenceClassification  # 自動加載序列分類模型
GPT2LMHeadModel                 # GPT2語言模型（帶語言建模頭）
GPT2Config                      # GPT2模型配置類

Trainer                         # 主要的訓練器類
TrainingArguments               # 訓練參數配置
DataCollatorForLanguageModeling # 語言模型數據整理器
EarlyStoppingCallback           # 早停回調函數
TrainerCallback                 # 訓練回調函數基類

PreTrainedTokenizerFast         # 快速分詞器
BitsAndBytesConfig              # 量化配置（用于4bit/8bit量化）

from transformers import GPT2LMHeadModel, Trainer, TrainingArguments

model = GPT2LMHeadModel.from_pretrained("gpt2")
training_args = TrainingArguments(
　　output_dir="./results",
　　per_device_train_batch_size=4,
　　num_train_epochs=3,
)

trainer = Trainer(
　　model=model,
　　args=training_args,
　　train_dataset=train_dataset,
)

from transformers import BitsAndBytesConfig

bnb_config = BitsAndBytesConfig(
　　load_in_4bit=True,
　　bnb_4bit_quant_type="nf4",
)
model = AutoModelForCausalLM.from_pretrained(
　　"model_name",
　　quantization_config=bnb_config,
)

PeftModel                       # 參數高效微調模型

from peft import PeftModel

# 加載LoRA等PEFT適配器
model = PeftModel.from_pretrained(model, "./lora-adapter")

GRPOTrainer                     # GRPO訓練器（基于PPO的改進）
GRPOConfig                      # GRPO訓練配置

from trl import GRPOTrainer, GRPOConfig

grpo_config = GRPOConfig(
　　model_name="gpt2",
　　learning_rate=1.41e-5,
)

trainer = GRPOTrainer(
　　model=model,
　　args=grpo_config,
　　train_dataset=dataset,
)

import torch
from transformers import (
    AutoTokenizer, AutoModelForSequenceClassification,
    Trainer, TrainingArguments, DataCollatorWithPadding,
    pipeline, EarlyStoppingCallback
)
from datasets import load_dataset, DatasetDict
from peft import LoraConfig, get_peft_model, TaskType
import evaluate
import numpy as np
from sklearn.metrics import accuracy_score, f1_score, classification_report
import pandas as pd

#1.數據加載

def load_and_preprocess_data():
    """加載和預處理情感分析數據集"""
    
    # 加載 IMDb 電影評論數據集
    dataset = load_dataset("imdb")
    
    # 查看數據集結構
    print("數據集結構:", dataset)
    print("訓練集樣本數:", len(dataset['train']))
    print("測試集樣本數:", len(dataset['test']))
    
    # 查看樣例數據
    print("\n訓練集樣例:")
    print("文本:", dataset['train'][0]['text'][:200])
    print("標簽:", dataset['train'][0]['label'])
    
    return dataset

# 加載數據
dataset = load_and_preprocess_data()

#2.分詞器和模型初始化

def initialize_tokenizer_and_model():
    """初始化分詞器和模型"""
    
    # 模型名稱
    model_name = "distilbert-base-uncased"
    
    # 加載分詞器
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    
    # 加載模型 (2個分類: 正面/負面)
    model = AutoModelForSequenceClassification.from_pretrained(
        model_name,
        num_labels=2,
        id2label={0: "負面", 1: "正面"},
        label2id={"負面": 0, "正面": 1}
    )
    
    return tokenizer, model

tokenizer, model = initialize_tokenizer_and_model()

 #3.數據預處理

def tokenize_function(examples):
    """分詞處理函數"""
    return tokenizer(
        examples["text"],
        truncation=True,
        padding=False,
        max_length=256,
        return_tensors=None
    )

def preprocess_data(dataset, tokenizer):
    """數據預處理流程"""
    
    # 應用分詞
    tokenized_datasets = dataset.map(
        tokenize_function,
        batched=True,
        remove_columns=["text"]  # 移除原始文本列
    )
    
    # 重命名標簽列以符合 Trainer 要求
    tokenized_datasets = tokenized_datasets.rename_column("label", "labels")
    
    # 分割訓練集和驗證集
    if "validation" not in tokenized_datasets:
        train_test_split = tokenized_datasets["train"].train_test_split(
            test_size=0.1, seed=42
        )
        tokenized_datasets = DatasetDict({
            "train": train_test_split["train"],
            "validation": train_test_split["test"],
            "test": tokenized_datasets["test"]
        })
    
    print("預處理后的數據集:", tokenized_datasets)
    return tokenized_datasets

# 預處理數據
tokenized_datasets = preprocess_data(dataset, tokenizer)

#4.評估指標設置

def compute_metrics(eval_pred):
    """計算評估指標"""
    predictions, labels = eval_pred
    predictions = np.argmax(predictions, axis=1)
    
    accuracy = accuracy_score(labels, predictions)
    f1 = f1_score(labels, predictions, average="weighted")
    
    return {
        "accuracy": accuracy,
        "f1": f1,
    }

# 加載準確率評估指標
accuracy_metric = evaluate.load("accuracy")

#5.基礎模型訓練

def setup_training_args():
    """設置訓練參數"""
    
    training_args = TrainingArguments(
        output_dir="./sentiment-analysis-results",
        overwrite_output_dir=True,
        
        # 訓練參數
        num_train_epochs=3,
        per_device_train_batch_size=16,
        per_device_eval_batch_size=16,
        learning_rate=2e-5,
        weight_decay=0.01,
        
        # 評估參數
        evaluation_strategy="epoch",
        save_strategy="epoch",
        load_best_model_at_end=True,
        metric_for_best_model="accuracy",
        
        # 日志參數
        logging_dir="./logs",
        logging_steps=500,
        report_to=None,  # 禁用 wandb 等外部記錄器
        
        # 優化參數
        warmup_steps=500,
        fp16=torch.cuda.is_available(),  # 自動啟用混合精度
    )
    
    return training_args

def train_model(model, tokenized_datasets, training_args):
    """訓練模型"""
    
    # 數據整理器
    data_collator = DataCollatorWithPadding(
        tokenizer=tokenizer,
        padding=True
    )
    
    # 創建訓練器
    trainer = Trainer(
        model=model,
        args=training_args,
        train_dataset=tokenized_datasets["train"],
        eval_dataset=tokenized_datasets["validation"],
        data_collator=data_collator,
        compute_metrics=compute_metrics,
        callbacks=[EarlyStoppingCallback(early_stopping_patience=2)],
    )
    
    # 開始訓練
    print("開始訓練模型...")
    train_result = trainer.train()
    
    # 保存模型
    trainer.save_model()
    tokenizer.save_pretrained(training_args.output_dir)
    
    # 評估模型
    eval_results = trainer.evaluate(tokenized_datasets["test"])
    print("測試集評估結果:", eval_results)
    
    return trainer, train_result

# 設置訓練參數并開始訓練
training_args = setup_training_args()
trainer, train_result = train_model(model, tokenized_datasets, training_args)

#6.使用peft進行模型微調

def setup_peft_training():
    """設置 PEFT (LoRA) 微調"""
    
    # LoRA 配置
    lora_config = LoraConfig(
        task_type=TaskType.SEQ_CLS,
        inference_mode=False,
        r=16,  # 秩
        lora_alpha=32,
        lora_dropout=0.1,
        target_modules=["q_lin", "k_lin", "v_lin", "out_lin"]  # DistilBERT 的注意力模塊
    )
    
    # 應用 PEFT
    peft_model = get_peft_model(model, lora_config)
    
    # 打印可訓練參數
    peft_model.print_trainable_parameters()
    
    return peft_model, lora_config

def fine_tune_with_peft(peft_model, tokenized_datasets):
    """使用 PEFT 進行微調"""
    
    # 微調訓練參數 (更小的學習率，更少的輪次)
    peft_training_args = TrainingArguments(
        output_dir="./peft-sentiment-analysis",
        overwrite_output_dir=True,
        
        num_train_epochs=2,
        per_device_train_batch_size=16,
        per_device_eval_batch_size=16,
        learning_rate=1e-4,  # 更小的學習率
        
        evaluation_strategy="epoch",
        save_strategy="epoch",
        load_best_model_at_end=True,
        metric_for_best_model="accuracy",
        
        logging_steps=100,
        report_to=None,
    )
    
    # 數據整理器
    data_collator = DataCollatorWithPadding(tokenizer=tokenizer)
    
    # PEFT 訓練器
    peft_trainer = Trainer(
        model=peft_model,
        args=peft_training_args,
        train_dataset=tokenized_datasets["train"],
        eval_dataset=tokenized_datasets["validation"],
        data_collator=data_collator,
        compute_metrics=compute_metrics,
    )
    
    # 開始微調
    print("開始 PEFT 微調...")
    peft_trainer.train()
    
    # 保存 PEFT 適配器
    peft_trainer.save_model()
    
    return peft_trainer

# 設置并運行 PEFT 微調
peft_model, lora_config = setup_peft_training()
peft_trainer = fine_tune_with_peft(peft_model, tokenized_datasets)

#7.模型推理和預測

class SentimentAnalyzer:
    """情感分析推理類"""
    
    def __init__(self, model_path, use_peft=False):
        self.tokenizer = AutoTokenizer.from_pretrained(model_path)
        
        if use_peft:
            # 加載基礎模型
            base_model = AutoModelForSequenceClassification.from_pretrained(
                "distilbert-base-uncased",
                num_labels=2
            )
            # 加載 PEFT 模型
            self.model = PeftModel.from_pretrained(base_model, model_path)
        else:
            self.model = AutoModelForSequenceClassification.from_pretrained(model_path)
        
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        self.model.to(self.device)
        self.model.eval()
        
        # 創建 pipeline
        self.pipeline = pipeline(
            "text-classification",
            model=self.model,
            tokenizer=self.tokenizer,
            device=0 if torch.cuda.is_available() else -1
        )
    
    def predict_single(self, text):
        """預測單個文本"""
        result = self.pipeline(text)
        return result[0]
    
    def predict_batch(self, texts):
        """批量預測"""
        results = self.pipeline(texts, batch_size=8)
        return results
    
    def detailed_prediction(self, text):
        """詳細預測，返回原始 logits"""
        inputs = self.tokenizer(
            text,
            return_tensors="pt",
            truncation=True,
            padding=True,
            max_length=256
        ).to(self.device)
        
        with torch.no_grad():
            outputs = self.model(**inputs)
            predictions = torch.nn.functional.softmax(outputs.logits, dim=-1)
            
        confidence = predictions.max().item()
        predicted_class = predictions.argmax().item()
        predicted_label = self.model.config.id2label[predicted_class]
        
        return {
            "text": text,
            "predicted_label": predicted_label,
            "confidence": confidence,
            "probabilities": {
                self.model.config.id2label[i]: prob.item() 
                for i, prob in enumerate(predictions[0])
            }
        }

#8.測試推理
def test_inference():
    """測試模型推理"""
    
    # 使用訓練好的模型
    analyzer = SentimentAnalyzer("./sentiment-analysis-results")
    
    # 測試文本
    test_texts = [
        "This movie is absolutely fantastic! The acting was brilliant.",
        "Terrible film, wasted my time and money.",
        "It's an okay movie, nothing special but not bad either.",
        "The plot was confusing and the characters were poorly developed.",
        "I loved every minute of this cinematic masterpiece!"
    ]
    
    print("=== 情感分析預測結果 ===")
    for text in test_texts:
        result = analyzer.detailed_prediction(text)
        print(f"\n文本: {text}")
        print(f"預測: {result['predicted_label']}")
        print(f"置信度: {result['confidence']:.4f}")
        print(f"概率分布: {result['probabilities']}")
    
    # 批量預測
    print("\n=== 批量預測 ===")
    batch_results = analyzer.predict_batch(test_texts)
    for text, result in zip(test_texts, batch_results):
        print(f"文本: {text[:50]}... -> {result['label']} (得分: {result['score']:.4f})")

# 運行推理測試
test_inference()

#9.模型評估與可視化

def comprehensive_evaluation(analyzer, test_dataset):
    """綜合評估模型性能"""
    
    # 獲取測試集預測
    test_texts = test_dataset["text"][:100]  # 取前100個樣本測試
    true_labels = test_dataset["label"][:100]
    
    predictions = analyzer.predict_batch(test_texts)
    predicted_labels = [1 if pred['label'] == "正面" else 0 for pred in predictions]
    
    # 計算評估指標
    accuracy = accuracy_score(true_labels, predicted_labels)
    f1 = f1_score(true_labels, predicted_labels, average="weighted")
    
    print("=== 綜合評估結果 ===")
    print(f"準確率: {accuracy:.4f}")
    print(f"F1分數: {f1:.4f}")
    print("\n詳細分類報告:")
    print(classification_report(true_labels, predicted_labels, 
                              target_names=["負面", "正面"]))
    
    # 創建結果 DataFrame
    results_df = pd.DataFrame({
        "text": test_texts,
        "true_label": ["負面" if label == 0 else "正面" for label in true_labels],
        "predicted_label": [pred['label'] for pred in predictions],
        "confidence": [pred['score'] for pred in predictions],
        "correct": [true_labels[i] == predicted_labels[i] for i in range(len(true_labels))]
    })
    
    return results_df

# 運行綜合評估
results_df = comprehensive_evaluation(analyzer, dataset["test"])

# 顯示錯誤預測樣本
print("\n=== 錯誤預測樣本 ===")
wrong_predictions = results_df[~results_df["correct"]]
print(wrong_predictions[["text", "true_label", "predicted_label", "confidence"]].head(10))

#10.模型保存和部署

def save_and_export_model(model, tokenizer, export_path="./final-sentiment-model"):
    """保存和導出模型"""
    
    # 保存模型和分詞器
    model.save_pretrained(export_path)
    tokenizer.save_pretrained(export_path)
    
    print(f"模型已保存到: {export_path}")
    
    # 保存模型配置
    model_config = {
        "model_name": "distilbert-sentiment-analysis",
        "num_labels": 2,
        "id2label": model.config.id2label,
        "label2id": model.config.label2id,
        "max_length": 256
    }
    
    import json
    with open(f"{export_path}/config.json", "w") as f:
        json.dump(model_config, f, indent=2)
    
    return export_path

# 保存最終模型
final_model_path = save_and_export_model(model, tokenizer)

print("?? 完整流程完成！")
print("? 數據加載和預處理")
print("? 模型訓練")
print("? PEFT 微調")
print("? 模型推理")
print("? 性能評估")
print("? 模型導出")

#11.快速使用

# 快速使用訓練好的模型
def quick_sentiment_analysis(text):
    """快速情感分析函數"""
    classifier = pipeline(
        "text-classification",
        model="./final-sentiment-model",
        tokenizer="./final-sentiment-model"
    )
    return classifier(text)

# 示例使用
sample_text = "I really enjoyed this movie, the acting was superb!"
result = quick_sentiment_analysis(sample_text)
print(f"文本: {sample_text}")
print(f"情感: {result[0]['label']}, 置信度: {result[0]['score']:.4f}")