Llama模型家族之Stanford NLP ReFT源代码探索（三）reft

 from .interventions import (
    LoreftIntervention,
    NoreftIntervention
)
from .trainer import (
    ReftTrainerForCausalLM,
    ReftTrainerForSequenceClassification
)


REFT_TYPE_TO_INTERVENTION_MAPPING = {
    "LOREFT": ReftTrainerForCausalLM,
    "NOREFT": ReftTrainerForSequenceClassification
}


MODEL_TYPE_TO_REFT_TRAINER_MAPPING = {
    "SEQ_CLS": PeftModelForSequenceClassification,
    "CAUSAL_LM": PeftModelForCausalLM,
}

这段代码创建了两个映射字典，将特定的干预类型和模型类型映射到相应的训练器类。

导入干预层类：

from .interventions import LoreftIntervention, NoreftIntervention

interventions 模块导入了两种干预层类：LoreftIntervention 和 NoreftIntervention。

导入训练器类：
```
from .trainer import ReftTrainerForCausalLM, ReftTrainerForSequenceClassification
```
从 trainer 模块导入了两种训练器类：ReftTrainerForCausalLM 和 ReftTrainerForSequenceClassification。这些类用于训练特定类型的模型，分别是因果语言模型和序列分类模型。
REFT类型到干预层的映射：
```
REFT_TYPE_TO_INTERVENTION_MAPPING = {
    "LOREFT": ReftTrainerForCausalLM,
    "NOREFT": ReftTrainerForSequenceClassification
}
```
这是一个字典，它将 REFT 类型的字符串标识符映射到对应的训练器类。"LOREFT" 对应于因果语言模型的训练器，而 "NOREFT" 对应于序列分类模型的训练器。根据干预类型快速找到对应的训练器类。
模型类型到REFT训练器的映射：
```
MODEL_TYPE_TO_REFT_TRAINER_MAPPING = {
    "SEQ_CLS": PeftModelForSequenceClassification,
    "CAUSAL_LM": PeftModelForCausalLM,
}
```
这是另一个字典，将模型类型的字符串标识符映射到对应的 REFT 训练器类。"SEQ_CLS" 映射到序列分类模型的训练器，"CAUSAL_LM" 映射到因果语言模型的训练器。

pyreft-main\pyreft\reft_model.py

 import pyvene as pv


def count_parameters(model):
    """Count parameters of a model that require gradients"""
    return sum(p.numel() for p in model.parameters() if p.requires_grad)


class ReftModel(pv.IntervenableModel):
    """
    Base model for Reft methods.
    """
    def __init__(self, config, model, **kwargs):
        super().__init__(config, model, **kwargs)

    @staticmethod
    def _convert_to_reft_model(intervenable_model):
        reft_model = ReftModel(intervenable_model.config, intervenable_model.model)
        # Copy any other necessary attributes
        for attr in vars(intervenable_model):
            setattr(reft_model, attr, getattr(intervenable_model, attr))
        return reft_model

    @staticmethod
    def load(*args, **kwargs):
        model = pv.IntervenableModel.load(*args, **kwargs)
        return ReftModel._convert_to_reft_model(model)

    def print_trainable_parameters(self):
        """
        Print trainable parameters.
        """
        _linked_key_set = set([])
        trainable_intervention_parameters = 0
        for k, v in self.interventions.items():
            if isinstance(v[0], pv.TrainableIntervention):
                if k in self._intervention_reverse_link:
                    if not self._intervention_reverse_link[k] in _linked_key_set:
                        _linked_key_set.add(self._intervention_reverse_link[k])
                        trainable_intervention_parameters += count_parameters(v[0])
                else:
                    trainable_intervention_parameters += count_parameters(v[0])

        trainable_model_parameters = sum(
            p.numel() for p in self.model.parameters() if p.requires_grad)

        all_model_parameters = sum(
            p.numel() for p in self.model.parameters())

        total_trainable_parameters = trainable_intervention_parameters + trainable_model_parameters
        
        print(
            f"trainable intervention params: {trainable_intervention_parameters:,d} || trainable model params: {trainable_model_parameters:,d}\n"
            f"model params: {all_model_parameters:,d} || trainable%: {100 * total_trainable_parameters / all_model_parameters}"
        )

定义了一个 ReftModel 的类，继承自 pyvene 模块中的 IntervenableModel 类，提供了一些与模型参数相关的功能。

导入 pyvene 模块：
```
import pyvene as pv
```
这行代码导入了 pyvene 模块，并且给它指定了一个别名 pv，以便于在代码中引用。
定义 count_parameters 函数：
```
def count_parameters(model):
    """Count parameters of a model that require gradients"""
    return sum(p.numel() for p in model.parameters() if p.requires_grad)
```
这个函数用于计算模型中需要梯度的参数总数。通过迭代模型的所有参数，并使用 numel() 函数计算每个参数的元素数量，然后对所有需要梯度的参数求和。
定义 ReftModel 类：
```
class ReftModel(pv.IntervenableModel):
    """
    Base model for Reft methods.
    """
```
ReftModel 类继承自 pv.IntervenableModel，是一个基类，用于 Reft模型。
ReftModel 类的构造函数：
```
def __init__(self, config, model, **kwargs):
    super().__init__(config, model, **kwargs)
```
构造函数接受配置 config、模型 model 和其他关键字参数 **kwargs，然后调用基类的构造函数。
_convert_to_reft_model 静态方法：
```
@staticmethod
def _convert_to_reft_model(intervenable_model):
    ...
```
这是一个静态方法，用于将 IntervenableModel 对象转换为 ReftModel 对象。返回一个新的 ReftModel 实例。
load 静态方法：
```
@staticmethod
def load(*args, **kwargs):
    model = pv.IntervenableModel.load(*args, **kwargs)
    return ReftModel._convert_to_reft_model(model)
```
load 静态方法用于加载一个 IntervenableModel 对象，并将其转换为 ReftModel 对象。它首先调用基类的 load 方法加载模型，然后使用 _convert_to_reft_model 方法进行转换。
print_trainable_parameters 方法：
```
def print_trainable_parameters(self):
    ...
```
这个方法用于打印模型中可训练参数的数量。首先计算所有可训练的干预层参数的总数，然后计算模型参数中需要梯度的参数数量，计算总的可训练参数数量，并打印出来。
- _linked_key_set 用于跟踪已经计算过的干预层，以避免重复计算。
- trainable_intervention_parameters 用于累计可训练干预层的参数数量。
- trainable_model_parameters 和 all_model_parameters 分别用于计算模型中需要梯度的参数总数和所有参数的总数。
- total_trainable_parameters 是所有可训练参数的总和。

打印格式：

print(
    f"trainable intervention params: {trainable_intervention_parameters:,d} || trainable model params: {trainable_model_parameters:,d}\n"
    f"model params: {all_model_parameters:,d} || trainable%: {100 * total_trainable_parameters / all_model_parameters}"
)

这行代码使用格式化字符串打印出可训练干预层参数、模型参数、总参数以及可训练参数的百分比。

pyreft-main\pyreft\reft_trainer.py

 import pyvene as pv
import torch.nn as nn
from torch.utils.data import DataLoader
from transformers import (
    Trainer,
    TrainingArguments,
    DataCollator,
    DataCollatorForSeq2Seq,
    AutoTokenizer
)
from transformers.trainer_utils import (
    EvalPrediction,
    has_length,
    denumpify_detensorize
)
from datasets import Dataset
from dataclasses import dataclass
from typing import Dict, Optional, Sequence

from tqdm import tqdm
import os
import torch
import re
import evaluate
import numpy as np
from torch.nn import BCEWithLogitsLoss, CrossEntropyLoss, MSELoss
from transformers.utils import logging

logger = logging.get_logger(__name__)

@dataclass
class ReftDataCollator(object):
    """Collate examples for ReFT."""

    data_collator: DataCollator

    def __call__(self, instances: Sequence[Dict]) -> Dict[str, torch.Tensor]:
        batch_inputs = self.data_collator(instances)
        max_seq_length = batch_inputs["input_ids"].shape[-1]
        batch_inputs["intervention_locations"] = batch_inputs["intervention_locations"][..., :max_seq_length]
        return batch_inputs


def make_data_collator(tokenizer, model) -> ReftDataCollator:
    data_collator_fn = DataCollatorForSeq2Seq(
        tokenizer=tokenizer,
        model=model,
        label_pad_token_id=-100,
        padding="longest",
        max_length=2048,
    )
    return ReftDataCollator(data_collator=data_collator_fn)


def make_dataloader(dataset: Dataset, batch_size: int, collate_fn: DataCollatorForSeq2Seq, shuffle: bool) -> DataLoader:
    return DataLoader(dataset, shuffle=shuffle, batch_size=batch_size, collate_fn=collate_fn)


class ReftTrainer(Trainer):
    def save_model(self, output_dir, _internal_call=False):
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
        self.model.save_intervention(
            save_directory=f"{output_dir}/intervenable_model", 
            include_model=True
        )

    def _load_best_model(self):
        logger.warning(f"Loading best model from {self.state.best_model_checkpoint} (score: {self.state.best_metric}).")
        self.model.load_intervention(
            f"{self.state.best_model_checkpoint}/intervenable_model", 
            include_model=True
        )

    def compute_loss(
        self,
        intervenable: pv.IntervenableModel,
        inputs,
        return_outputs=False
    ):
        # run intervened forward pass
        _, cf_outputs = intervenable(
            {
                "input_ids": inputs["input_ids"],
                "attention_mask": inputs["attention_mask"]
            },
            unit_locations={"sources->base": (
                None,
                inputs["intervention_locations"].permute(1, 0, 2).tolist()
            )},
            labels=inputs["labels"],
            subspaces=inputs["subspaces"].permute(1, 0, 2).tolist() if "subspaces" in inputs else None
        )
        # return
        return (cf_outputs.loss, cf_outputs) if return_outputs else cf_outputs.loss


class ReftTrainerForCausalLM(ReftTrainer):
    def get_train_dataloader(self) -> DataLoader:
        return make_dataloader(self.train_dataset, self._train_batch_size, self.data_collator, shuffle=True)


class ReftTrainerForSequenceClassification(ReftTrainer):
    def evaluate(
        self, ignore_keys,
    ):

        # ensure everything is in eval mode
        self.model.model.eval()
        for k,v in  self.model.interventions.items():
            _ = v[0].eval()
        
        batch_size = self.args.eval_batch_size
        data_collator = self.data_collator
        eval_dataset = self.eval_dataset
        intervenable = self.model
        
        dataloader = make_dataloader(
            eval_dataset, batch_size, data_collator, shuffle=False)

        logger.info(f"***** Running In-Training Evaluation *****")
        if has_length(dataloader):
            logger.info(f"  Num examples = {self.num_examples(dataloader)}")
        else:
            logger.info("  Num examples: Unknown")
        logger.info(f"  Batch size = {batch_size}")

        eval_iterator = tqdm(dataloader, position=0, leave=True)
        all_preds = []
        all_labels = []
        with torch.no_grad():
            for step, inputs in enumerate(eval_iterator):
                for k, v in inputs.items():
                    if v is not None and isinstance(v, torch.Tensor):
                        inputs[k] = v.to(self.model.get_device())
                
                # [layers, batch_size, positions]
                intervention_locations = inputs["intervention_locations"].permute(1, 0, 2).tolist()
                _, cf_outputs = intervenable(
                    {"input_ids": inputs["input_ids"], "attention_mask": inputs["attention_mask"]},
                    unit_locations={"sources->base": (None, intervention_locations)})
            
                all_preds += [cf_outputs.logits]
                all_labels += [inputs["labels"]]
        all_preds = torch.cat(all_preds, dim=0).cpu().to(torch.float32)
        all_labels = torch.cat(all_labels, dim=0).cpu().to(torch.float32)
        metrics = self.compute_metrics(EvalPrediction(predictions=all_preds, label_ids=all_labels))
        metrics = denumpify_detensorize(metrics)
        
        metric_key_prefix = "eval"
        for key in list(metrics.keys()):
            if not key.startswith(f"{metric_key_prefix}_"):
                metrics[f"{metric_key_prefix}_{key}"] = metrics.pop(key)
        
        self.log(metrics)
        self.control = self.callback_handler.on_evaluate(self.args, self.state, self.control, metrics)
        self._memory_tracker.stop_and_update_metrics(metrics)
        
        return metrics

定义了一个用于训练和评估ReFT模型的框架，包括数据整理、模型训练和评估等组件。

导入必要的库：
- pyvene 作为 pv 被导入，用于处理干预模型。
- torch.nn 作为 nn 导入，用于访问PyTorch的神经网络模块。
- DataLoader 从 torch.utils.data 导入，用于创建数据加载器。
- transformers 库导入，包括 Trainer、TrainingArguments 等组件。
- datasets 库导入，用于处理数据集。
- tqdm 用于显示进度条。
- os、torch、re 等Python标准库被导入。
创建日志记录器：
```
logger = logging.get_logger(__name__)
```
使用 transformers.utils 中的 logging 创建一个日志记录器。
定义 ReftDataCollator 类：
- 使用 @dataclass 装饰器定义了一个数据整理类，它继承自 object。
- 该类包含一个方法 __call__，它接收一系列字典，并将它们整理成一个批次的输入数据。
定义 make_data_collator 函数：
- 这个函数创建并返回一个 ReftDataCollator 实例，它使用 DataCollatorForSeq2Seq 来整理数据。
定义 make_dataloader 函数：
- 这个函数接收一个数据集、批量大小、数据整理函数和是否打乱数据的标志，并返回一个 DataLoader 实例。
定义 ReftTrainer 类：
- 继承自 transformers 的 Trainer 类。
- 重写了 save_model 方法，用于保存模型的干预信息。
- 重写了 _load_best_model 方法，用于加载最佳模型的干预信息。
- 定义了 compute_loss 方法，用于计算模型的损失。
定义 ReftTrainerForCausalLM 类：
- 继承自 ReftTrainer 类。
- 重写了 get_train_dataloader 方法，用于获取因果语言模型的训练数据加载器。
定义 ReftTrainerForSequenceClassification 类：
- 继承自 ReftTrainer 类。
- 重写了 evaluate 方法，用于评估序列分类模型的性能。
评估函数 evaluate：
- 将模型设置为评估模式。
- 创建评估数据加载器。
- 使用 tqdm 显示评估进度。
- 收集所有预测结果和标签。
- 计算并记录评估指标。
计算指标：
- 使用 self.compute_metrics 计算评估指标。
- 使用 denumpify_detensorize 将指标转换为适当的格式。
日志记录和内存跟踪：
- 使用日志记录器记录评估信息。
- 使用内存跟踪器更新指标。

以上代码提供了一个完整的训练和评估流程，包括数据整理、模型保存和加载、损失计算、评估指标计算等功能。代码的设计允许灵活性和可扩展性，可以适应不同的ReFT模型和任务。

pyreft-main\pyreft\utils.py

import enum
from .reft_model import ReftModel


class ReftType(str, enum.Enum):
    """
    Enum class for the different types of adapters in REFT.

    Supported REFT types:
    - LOREFT
    """

    LOREFT = "LOREFT"
    NLOREFT = "NOREFT"
    # Add yours here!


class TaskType(str, enum.Enum):
    """
    Enum class for the different types of tasks supported by REFT.

    Overview of the supported task types:
    - SEQ_CLS: Text classification.
    - CAUSAL_LM: Causal language modeling.
    """

    SEQ_CLS = "SEQ_CLS"
    CAUSAL_LM = "CAUSAL_LM"


def get_reft_model(model, reft_config, set_device=True, disable_model_grads=True):
    """
    Create an instance of ReFT model.
    """
    reft_model = ReftModel(reft_config, model)
    if set_device:
        reft_model.set_device(model.device)
    if disable_model_grads:
        reft_model.disable_model_gradients()    
    return reft_model

大模型技术分享

在这里插入图片描述

《企业级生成式人工智能LLM大模型技术、算法及案例实战》线上高级研修讲座

模块一：Generative AI 原理本质、技术内核及工程实践周期详解
模块二：工业级 Prompting 技术内幕及端到端的基于LLM 的会议助理实战
模块三：三大 Llama 2 模型详解及实战构建安全可靠的智能对话系统
模块四：生产环境下 GenAI/LLMs 的五大核心问题及构建健壮的应用实战
模块五：大模型应用开发技术：Agentic-based 应用技术及案例实战
模块六：LLM 大模型微调及模型 Quantization 技术及案例实战
模块七：大模型高效微调 PEFT 算法、技术、流程及代码实战进阶
模块八：LLM 模型对齐技术、流程及进行文本Toxicity 分析实战
模块九：构建安全的 GenAI/LLMs 核心技术Red Teaming 解密实战
模块十：构建可信赖的企业私有安全大模型Responsible AI 实战

Llama3关键技术深度解析与构建Responsible AI、算法及开发落地实战

1、Llama开源模型家族大模型技术、工具和多模态详解：学员将深入了解Meta Llama 3的创新之处，比如其在语言模型技术上的突破，并学习到如何在Llama 3中构建trust and safety AI。他们将详细了解Llama 3的五大技术分支及工具，以及如何在AWS上实战Llama指令微调的案例。
2、解密Llama 3 Foundation Model模型结构特色技术及代码实现：深入了解Llama 3中的各种技术，比如Tiktokenizer、KV Cache、Grouped Multi-Query Attention等。通过项目二逐行剖析Llama 3的源码，加深对技术的理解。
3、解密Llama 3 Foundation Model模型结构核心技术及代码实现：SwiGLU Activation Function、FeedForward Block、Encoder Block等。通过项目三学习Llama 3的推理及Inferencing代码，加强对技术的实践理解。
4、基于LangGraph on Llama 3构建Responsible AI实战体验：通过项目四在Llama 3上实战基于LangGraph的Responsible AI项目。他们将了解到LangGraph的三大核心组件、运行机制和流程步骤，从而加强对Responsible AI的实践能力。
5、Llama模型家族构建技术构建安全可信赖企业级AI应用内幕详解：深入了解构建安全可靠的企业级AI应用所需的关键技术，比如Code Llama、Llama Guard等。项目五实战构建安全可靠的对话智能项目升级版，加强对安全性的实践理解。
6、Llama模型家族Fine-tuning技术与算法实战：学员将学习Fine-tuning技术与算法，比如Supervised Fine-Tuning(SFT)、Reward Model技术、PPO算法、DPO算法等。项目六动手实现PPO及DPO算法，加强对算法的理解和应用能力。
7、Llama模型家族基于AI反馈的强化学习技术解密：深入学习Llama模型家族基于AI反馈的强化学习技术，比如RLAIF和RLHF。项目七实战基于RLAIF的Constitutional AI。
8、Llama 3中的DPO原理、算法、组件及具体实现及算法进阶：学习Llama 3中结合使用PPO和DPO算法，剖析DPO的原理和工作机制，详细解析DPO中的关键算法组件，并通过综合项目八从零开始动手实现和测试DPO算法，同时课程将解密DPO进阶技术Iterative DPO及IPO算法。
9、Llama模型家族Safety设计与实现：在这个模块中，学员将学习Llama模型家族的Safety设计与实现，比如Safety in Pretraining、Safety Fine-Tuning等。构建安全可靠的GenAI/LLMs项目开发。
10、Llama 3构建可信赖的企业私有安全大模型Responsible AI系统：构建可信赖的企业私有安全大模型Responsible AI系统，掌握Llama 3的Constitutional AI、Red Teaming。

解码Sora架构、技术及应用

一、为何Sora通往AGI道路的里程碑？
1，探索从大规模语言模型(LLM)到大规模视觉模型(LVM)的关键转变，揭示其在实现通用人工智能(AGI)中的作用。
2，展示Visual Data和Text Data结合的成功案例，解析Sora在此过程中扮演的关键角色。
3，详细介绍Sora如何依据文本指令生成具有三维一致性(3D consistency)的视频内容。 4，解析Sora如何根据图像或视频生成高保真内容的技术路径。
5，探讨Sora在不同应用场景中的实践价值及其面临的挑战和局限性。

二、解码Sora架构原理
1，DiT (Diffusion Transformer)架构详解
2，DiT是如何帮助Sora实现Consistent、Realistic、Imaginative视频内容的？
3，探讨为何选用Transformer作为Diffusion的核心网络，而非技术如U-Net。
4，DiT的Patchification原理及流程，揭示其在处理视频和图像数据中的重要性。
5，Conditional Diffusion过程详解，及其在内容生成过程中的作用。
三、解码Sora关键技术解密
1，Sora如何利用Transformer和Diffusion技术理解物体间的互动，及其对模拟复杂互动场景的重要性。
2，为何说Space-time patches是Sora技术的核心，及其对视频生成能力的提升作用。
3，Spacetime latent patches详解，探讨其在视频压缩和生成中的关键角色。
4，Sora Simulator如何利用Space-time patches构建digital和physical世界，及其对模拟真实世界变化的能力。
5，Sora如何实现faithfully按照用户输入文本而生成内容，探讨背后的技术与创新。
6，Sora为何依据abstract concept而不是依据具体的pixels进行内容生成，及其对模型生成质量与多样性的影响。

Llama模型家族之Stanford NLP ReFT源代码探索（三）reft_model.py代码解析

LlaMA 3 系列博客