LoRA微调技术

def lora_forward(x, W_0, B, A, scaling):
    # 原始输出
    h_0 = x @ W_0.T
    
    # LoRA输出
    h_lora = x @ (B @ A).T * scaling
    
    # 最终输出
    return h_0 + h_lora

def lora_backward(grad_output, x, B, A, scaling):
    # 计算LoRA参数的梯度
    grad_A = (grad_output * scaling).T @ x @ B.T
    grad_B = (grad_output * scaling).T @ x @ A.T
    
    return grad_A, grad_B

# 注意力层
target_modules = ["q", "v"]  # Query和Value投影层
# 或
target_modules = ["q", "k", "v", "o"]  # 所有注意力层

# 前馈网络
target_modules = ["w1", "w2"]  # 前馈网络的两个线性层

# 全连接层
target_modules = ["dense"]  # 分类头等全连接层

# 基于任务复杂度的选择
if task_complexity == "high":
    r = 64
elif task_complexity == "medium":
    r = 16
else:  # low
    r = 8

# 基于模型大小的选择
if model_size == "large":
    r = 32
elif model_size == "base":
    r = 16
else:  # small
    r = 8

def initialize_lora_matrices(r, d, k):
    # A矩阵使用随机初始化
    A = torch.randn(r, k) * 0.01
    
    # B矩阵使用零初始化
    B = torch.zeros(d, r)
    
    return A, B

# 缩放因子用于控制LoRA的影响程度
scaling = alpha / r

# 其中alpha是超参数，通常设置为r的1-2倍
alpha = 32  # 当r=16时

# 安装PEFT库
pip install peft

# 导入必要模块
from peft import LoraConfig, get_peft_model, TaskType
from transformers import AutoModelForSeq2SeqLM, AutoTokenizer

# 创建LoRA配置
lora_config = LoraConfig(
    task_type=TaskType.SEQ_2_SEQ_LM,  # 任务类型
    r=16,                             # 秩
    lora_alpha=32,                    # 缩放参数
    lora_dropout=0.1,                 # Dropout率
    target_modules=["q", "v"],        # 目标模块
    bias="none",                      # 偏置处理
    inference_mode=False,             # 推理模式
)

# 加载预训练模型
model = AutoModelForSeq2SeqLM.from_pretrained("google/flan-t5-base")

# 应用LoRA
model = get_peft_model(model, lora_config)

# 打印可训练参数
model.print_trainable_parameters()

# 训练参数
training_args = TrainingArguments(
    output_dir="./lora_output",
    per_device_train_batch_size=4,
    per_device_eval_batch_size=4,
    gradient_accumulation_steps=4,
    num_train_epochs=3,
    learning_rate=2e-4,
    warmup_steps=100,
    fp16=True,  # 混合精度训练
    logging_steps=10,
    eval_steps=100,
    save_steps=500,
    eval_strategy="steps",
    save_strategy="steps",
    load_best_model_at_end=True,
    metric_for_best_model="eval_loss",
    greater_is_better=False,
)

# 训练数据格式
{
    "instruction": "黑金风格的电商首页，顶部搜索，中间两列商品卡，底部导航",
    "output_json_minified": '{"page":{"name":"home_page","theme":"obsidian-gold",...}}'
}

class UIDataset(Dataset):
    def __init__(self, data_path: str, tokenizer, max_length: int = 512):
        self.tokenizer = tokenizer
        self.max_length = max_length
        self.data = self._load_data(data_path)
    
    def _load_data(self, data_path: str) -> List[Dict[str, str]]:
        data = []
        with open(data_path, 'r', encoding='utf-8') as f:
            for line in f:
                sample = json.loads(line.strip())
                data.append(sample)
        return data
    
    def __getitem__(self, idx):
        sample = self.data[idx]
        
        # 编码输入和输出
        input_text = sample["instruction"]
        output_text = sample["output_json_minified"]
        
        # Tokenize输入
        input_encoding = self.tokenizer(
            input_text,
            max_length=self.max_length,
            padding="max_length",
            truncation=True,
            return_tensors="pt"
        )
        
        # Tokenize输出
        output_encoding = self.tokenizer(
            output_text,
            max_length=self.max_length,
            padding="max_length",
            truncation=True,
            return_tensors="pt"
        )
        
        return {
            "input_ids": input_encoding["input_ids"].squeeze(),
            "attention_mask": input_encoding["attention_mask"].squeeze(),
            "labels": output_encoding["input_ids"].squeeze()
        }

# 数据整理器
data_collator = DataCollatorForSeq2Seq(
    tokenizer=tokenizer,
    model=model,
    padding=True
)

# 创建训练器
trainer = Trainer(
    model=model,
    args=training_args,
    train_dataset=train_dataset,
    eval_dataset=val_dataset,
    data_collator=data_collator,
    tokenizer=tokenizer,
)

# 开始训练
trainer.train()

# 训练过程中的监控指标
def compute_metrics(eval_pred):
    predictions, labels = eval_pred
    predictions = np.argmax(predictions, axis=-1)
    
    # 计算准确率
    accuracy = (predictions == labels).mean()
    
    return {
        "accuracy": accuracy,
        "perplexity": torch.exp(torch.tensor(eval_loss)).item()
    }

# 保存LoRA适配器
model.save_pretrained("./lora_output")
tokenizer.save_pretrained("./lora_output")

# 保存的权重文件
# adapter_config.json    - LoRA配置
# adapter_model.safetensors - LoRA权重
# tokenizer.json        - 分词器配置

# 加载基础模型
base_model = AutoModelForSeq2SeqLM.from_pretrained("google/flan-t5-base")

# 加载LoRA权重
model = PeftModel.from_pretrained(base_model, "./lora_output")

# 启用混合精度训练
training_args = TrainingArguments(
    fp16=True,  # 使用FP16
    dataloader_pin_memory=True,
    dataloader_num_workers=4,
)

# 使用梯度累积减少内存使用
training_args = TrainingArguments(
    per_device_train_batch_size=2,      # 减小批次大小
    gradient_accumulation_steps=8,      # 增加累积步数
    per_device_eval_batch_size=4,
)

# 启用梯度检查点
model.gradient_checkpointing_enable()

# 使用余弦退火学习率
from transformers import get_cosine_schedule_with_warmup

scheduler = get_cosine_schedule_with_warmup(
    optimizer,
    num_warmup_steps=100,
    num_training_steps=1000
)

# 添加权重衰减防止过拟合
training_args = TrainingArguments(
    weight_decay=0.01,
    adam_epsilon=1e-8,
)

# 将LoRA权重合并到基础模型中
merged_model = model.merge_and_unload()

# 保存合并后的模型
merged_model.save_pretrained("./merged_model")

# 使用量化减少推理内存
from transformers import BitsAndBytesConfig

quantization_config = BitsAndBytesConfig(
    load_in_8bit=True,
    llm_int8_threshold=6.0,
)

model = AutoModelForSeq2SeqLM.from_pretrained(
    "google/flan-t5-base",
    quantization_config=quantization_config
)

# 对比不同方法的性能
methods = {
    "full_finetuning": "全参数微调",
    "lora_r8": "LoRA r=8",
    "lora_r16": "LoRA r=16",
    "lora_r32": "LoRA r=32",
    "lora_r64": "LoRA r=64",
}

def evaluate_model(model, test_dataset):
    """评估模型性能"""
    results = {}
    
    # 生成测试
    predictions = []
    for sample in test_dataset:
        input_text = sample["instruction"]
        output = model.generate(
            input_text,
            max_length=512,
            temperature=0.7,
            do_sample=True
        )
        predictions.append(output)
    
    # 计算指标
    results["bleu_score"] = calculate_bleu(predictions, test_dataset)
    results["rouge_score"] = calculate_rouge(predictions, test_dataset)
    results["exact_match"] = calculate_exact_match(predictions, test_dataset)
    
    return results

# 超参数网格
param_grid = {
    "r": [8, 16, 32, 64],
    "lora_alpha": [16, 32, 64],
    "lora_dropout": [0.1, 0.2, 0.3],
    "learning_rate": [1e-4, 2e-4, 5e-4],
}

# 网格搜索
best_score = 0
best_params = None

for params in itertools.product(*param_grid.values()):
    config = dict(zip(param_grid.keys(), params))
    
    # 训练模型
    score = train_and_evaluate(config)
    
    if score > best_score:
        best_score = score
        best_params = config

from skopt import gp_minimize
from skopt.space import Real, Integer

# 定义搜索空间
space = [
    Integer(8, 64, name='r'),
    Real(16, 64, name='lora_alpha'),
    Real(0.1, 0.3, name='lora_dropout'),
    Real(1e-4, 5e-4, name='learning_rate'),
]

# 目标函数
def objective(params):
    r, lora_alpha, lora_dropout, learning_rate = params
    config = {
        'r': int(r),
        'lora_alpha': lora_alpha,
        'lora_dropout': lora_dropout,
        'learning_rate': learning_rate,
    }
    
    score = train_and_evaluate(config)
    return -score  # 最小化负分数

# 贝叶斯优化
result = gp_minimize(objective, space, n_calls=50)

# UI生成任务的LoRA配置
lora_config = LoraConfig(
    task_type=TaskType.SEQ_2_SEQ_LM,
    r=16,                    # 适中的秩
    lora_alpha=32,           # 缩放参数
    lora_dropout=0.1,        # 防止过拟合
    target_modules=["q", "v"], # 注意力层
    bias="none",
    inference_mode=False,
)

# 数据增强策略
def augment_data(sample):
    """数据增强"""
    # 同义词替换
    instruction = synonym_replacement(sample["instruction"])
    
    # 句式变换
    instruction = sentence_transformation(instruction)
    
    return {
        "instruction": instruction,
        "output_json_minified": sample["output_json_minified"]
    }

# 渐进式训练
def progressive_training():
    """渐进式训练策略"""
    # 第一阶段：基础理解
    train_stage1(epochs=1, learning_rate=1e-4)
    
    # 第二阶段：结构生成
    train_stage2(epochs=2, learning_rate=2e-4)
    
    # 第三阶段：精细调优
    train_stage3(epochs=1, learning_rate=1e-4)

# 模型优化
def optimize_model(model):
    """模型优化"""
    # 合并LoRA权重
    model = model.merge_and_unload()
    
    # 模型量化
    model = quantize_model(model)
    
    # 模型剪枝
    model = prune_model(model)
    
    return model

# 推理优化
def optimize_inference(model, tokenizer):
    """推理优化"""
    # 编译模型
    model = torch.compile(model)
    
    # 设置推理模式
    model.eval()
    
    # 预热模型
    warmup_inference(model, tokenizer)
    
    return model

# 解决方案
lora_config = LoraConfig(
    lora_dropout=0.2,        # 增加dropout
    r=8,                     # 减小秩
)

training_args = TrainingArguments(
    weight_decay=0.01,       # 添加权重衰减
    early_stopping_patience=3, # 早停
)

# 解决方案
lora_config = LoraConfig(
    r=32,                    # 增加秩
    lora_alpha=64,           # 增加缩放参数
    target_modules=["q", "k", "v", "o"], # 增加目标模块
)

# 解决方案
training_args = TrainingArguments(
    fp16=True,               # 混合精度
    dataloader_num_workers=4, # 多进程加载
    gradient_accumulation_steps=4, # 梯度累积
)

# 解决方案
training_args = TrainingArguments(
    per_device_train_batch_size=1, # 减小批次大小
    gradient_accumulation_steps=8,  # 增加累积步数
    gradient_checkpointing=True,    # 梯度检查点
)

# 解决方案
def improve_generation(model, input_text):
    """改善生成质量"""
    # 调整生成参数
    output = model.generate(
        input_text,
        max_length=512,
        temperature=0.7,
        top_p=0.9,
        top_k=50,
        do_sample=True,
        num_beams=4,  # 使用束搜索
        early_stopping=True,
    )
    return output

# 解决方案
def optimize_inference_speed(model):
    """优化推理速度"""
    # 模型合并
    model = model.merge_and_unload()
    
    # 模型量化
    model = torch.quantization.quantize_dynamic(
        model, {torch.nn.Linear}, dtype=torch.qint8
    )
    
    # 模型编译
    model = torch.compile(model)
    
    return model