ITPub博客

首页 > 人工智能 > 人工智能 > 智能体的自适应学习

智能体的自适应学习

原创 人工智能 作者:brucexia 时间:2025-12-12 12:58:00 0 删除 编辑

自适应学习是指AI 智能体根据环境变化、任务需求或交互数据,动态调整自身行为、策略或模型参数的能力。其目标是实现持续进化,无须人类频繁干预。

在线微调:人类反馈强化学习

人类反馈强化学习(Reinforcement Learning from Human Feedback ,RLHF )是一种将人类偏好和反馈融入强化学习过程的技术,它已成为当前AI 智能体开发中最前沿的方法之一。

1. RLHF 的核心思想

(1 )传统RL 的局限性:传统强化学习依赖精心设计的奖励函数,但在复杂现实场景中难以设计。

(2 )人类反馈的价值:利用人类对模型输出的偏好或评分来指导学习过程。

(3 )两阶段训练:预训练模型 + 基于人类反馈的微调训练。

2. RLHF 的关键组件

(1 )奖励模型(奖励函数):学习将人类偏好转换为可量化的奖励信号。

(2 )策略模型:根据奖励信号优化自身行为。

(3 )人类反馈数据:包括排序、评分或直接修正等形式。

3. RLHF 技术实现流程

【示例 3.4 】 基于RLHF 的对话系统优化(注意:由于代码复杂且资源要求高,我们可能无法在本地完整运行,因此以下代码主要展示流程,由于原始代码中使用了多个大型模型,建议在具有足够GPU 内存的机器上运行)。

import torch

import torch.nn as nn

import torch.optim as optim

from transformers import AutoTokenizer, AutoModelForCausalLM

from datasets import load _ dataset

from torch.utils.data import DataLoader

from tqdm import tqdm

import os

import warnings

warnings.filterwarnings('ignore')

 

# 自定义 Value Head

class ValueHead(nn.Module):

    def __ init __ (self, hidden _ size):

        super(). __ init __ ()

        self.value _ head = nn.Linear(hidden _ size, 1)

 

    def forward(self, hidden _ states):

        return self.value _ head(hidden _ states[:, -1, :]).squeeze(-1)

 

# RLHF 类

class Qwen2RLHF:

    def __ init __ (self, model _ name="Qwen/Qwen-1 _ 8B"):

        print(f"Loading model: {model _ name}")

        self.tokenizer = AutoTokenizer.from _ pretrained(model _ name)

        self.tokenizer.pad _ token = self.tokenizer.eos _ token

 

        # 基础模型

        self.base _ model = AutoModelForCausalLM.from _ pretrained(

            model _ name,

            device _ map="auto",

            torch _ dtype=torch.float16,

            trust _ remote _ code=True

        )

 

        # RL 模型

        self.rl _ model = AutoModelForCausalLM.from _ pretrained(

            model _ name,

            device _ map="auto",

            torch _ dtype=torch.float16,

            trust _ remote _ code=True

        )

 

        # 价值头

        self.value _ head = ValueHead(self.rl _ model.config.hidden _ size).to(self.rl _ model. device).half()

 

        # 优化器

        self.optimizer = optim.AdamW([

            {'params': self.rl _ model.parameters()},

            {'params': self.value _ head.parameters()}

        ], lr=1e-5)

 

        self.device = torch.device("cuda" if torch.cuda.is _ available() else "cpu")

        print(f"Using device: {self.device}")

 

    def supervised _ fine _ tuning(self, dataset _ path, epochs=1, batch _ size=2):

        """ 简化的 SFT 微调 """

        print("Loading dataset for SFT...")

 

        # 创建简单的演示数据

        demonstrations = [

            {"instruction": " 解释机器学习 ", "output": " 机器学习是人工智能的一个分支,它使计算机能够从数据中学习而无须明确编程。 "},

            {"instruction": " 什么是神经网络 ", "output": " 神经网络是受人脑启发的计算系统,由相互连接的人工神经元组成。 "},

            {"instruction": "Python 的优点 ", "output": "Python 简单易学,拥有丰富的库和框架,适合各种应用开发。 "},

            {"instruction": " 如何学习编程 ", "output": " 从基础语法开始,通过实践项目不断练习,参与开源社区交流。 "}

        ]

 

        # 准备数据

        texts = []

        for demo in demonstrations:

            text = f"Human: {demo['instruction']}\nAssistant: {demo['output']} {self.tokenizer.eos _ token}"

            texts.append(text)

 

        # 编码

        encodings = self.tokenizer(texts, padding=True, truncation=True, max _ length=128, return _ tensors="pt")

        input _ ids = encodings["input _ ids"]

        labels = input _ ids.clone()

 

        # 训练循环

        self.base _ model.train()

        for epoch in range(epochs):

            total _ loss = 0

            for i in range(0, len(input _ ids), batch _ size):

                batch _ input _ ids = input _ ids[i:i+batch _ size].to(self.device)

                batch _ labels = labels[i:i+batch _ size].to(self.device)

 

                outputs = self.base _ model(input _ ids=batch _ input _ ids, labels=batch _ labels)

                loss = outputs.loss

 

                self.optimizer.zero _ grad()

                loss.backward()

                self.optimizer.step()

 

                total _ loss += loss.item()

 

            print(f"SFT Epoch {epoch+1}/{epochs}, Loss: {total _ loss/(len(input _ ids)/ batch _ size):.4f}")

 

        print("SFT completed")

 

    def ppo _ training(self, epochs=1, batch _ size=2):

        """ 简化的 PPO 训练 """

        print("Starting simplified PPO training...")

 

        # 创建简单的提示

        prompts = [

            " 解释深度学习 ",

            " 什么是 Transformer 模型 ",

            "Python 中如何定义函数 ",

            " 如何提高编程技能 "

        ]

 

        # 编码提示

        encodings = self.tokenizer(prompts, padding=True, truncation=True, max _ length=64, return _ tensors="pt")

        input _ ids = encodings["input _ ids"].to(self.device)

        attention _ mask = encodings["attention _ mask"].to(self.device)

 

        # 训练循环

        self.rl _ model.train()

        self.value _ head.train()

 

        for epoch in range(epochs):

            total _ loss = 0

 

            # 生成响应

            with torch.no _ grad():

                gen _ outputs = self.rl _ model.generate(

                    input _ ids=input _ ids,

                    attention _ mask=attention _ mask,

                    max _ new _ tokens=32,

                    do _ sample=True,

                    temperature=0.7,

                    pad _ token _ id=self.tokenizer.eos _ token _ id,

                    return _ dict _ in _ generate=True,

                )

                gen _ ids = gen _ outputs.sequences[:, input _ ids.shape[1]:]

 

            # 计算奖励(简化 : 基于响应长度)

            rewards = torch.tensor([min(len(self.tokenizer.decode(ids))/30, 1.0) for ids in gen _ ids],

                                  device=self.device, dtype=torch.float16)

 

            # 计算价值

            full _ seq = torch.cat([input _ ids, gen _ ids], dim=1)

            with torch.no _ grad():

                outputs = self.rl _ model(full _ seq, output _ hidden _ states=True)

                last _ hidden _ state = outputs.hidden _ states[-1]

 

            values = self.value _ head(last _ hidden _ state)

 

            # 计算优势

            advantages = rewards - values.squeeze()

           

            # 策略损失

            logits = outputs.logits[:, input _ ids.shape[1]-1:-1, :]

            log _ probs = torch.log _ softmax(logits, dim=-1)

            gen _ log _ probs = torch.gather(log _ probs, 2, gen _ ids.unsqueeze(2)).squeeze(2)

            ratio = torch.exp(gen _ log _ probs.mean(dim=1) - 0.5)  # 简化

            policy _ loss = -torch.min(ratio * advantages, torch.clamp(ratio, 0.8, 1.2) * advantages).mean()

 

            # 价值损失

            value _ loss = 0.5 * (values.squeeze() - rewards) ** 2

            value _ loss = value _ loss.mean()

 

            # 总损失

            loss = policy _ loss + value _ loss

 

            self.optimizer.zero _ grad()

            loss.backward()

            self.optimizer.step()

 

            total _ loss += loss.item()

            print(f"PPO Epoch {epoch+1}/{epochs}, Loss: {total _ loss:.4f}")

 

        print("PPO training completed")

 

    def generate _ response(self, prompt):

        """ 生成回答 """

        self.rl _ model.eval()

        input _ text = f"Human: {prompt}\nAssistant:"

        inputs = self.tokenizer(input _ text, return _ tensors="pt", truncation=True, max _ length=128).to(self.device)

 

        with torch.no _ grad():

            outputs = self.rl _ model.generate(

                **inputs,

                max _ new _ tokens=50,

                do _ sample=True,

                temperature=0.7,

                pad _ token _ id=self.tokenizer.eos _ token _ id

            )

 

        response = self.tokenizer.decode(outputs[0], skip _ special _ tokens=True)

        return response.split("Assistant:")[-1].strip()

 

# 主程序

if __ name __ == " __ main __ ":

    os.environ["TOKENIZERS _ PARALLELISM"] = "false"

 

    # 初始化

    rlhf = Qwen2RLHF(model _ name="Qwen/Qwen-1 _ 8B")

 

    # 阶段 1 : SFT

    print("\n=== 1. Supervised Fine-Tuning ===")

    rlhf.supervised _ fine _ tuning("dummy _ path", epochs=2, batch _ size=2)

 

    # 阶段 2 : PPO

    print("\n=== 2. PPO Training ===")

    rlhf.ppo _ training(epochs=2, batch _ size=2)

 

    # 测试

    print("\n=== 3. Test Final Model ===")

    test _ prompts = [

        " 解释什么是机器学习 ",

        "Python 是什么 ",

        " 如何学习人工智能 "

    ]

 

    for prompt in test _ prompts:

        response = rlhf.generate _ response(prompt)

        print(f"Prompt: {prompt}")

        print(f"Response: {response}")

        print("-" * 50)

运行代码,输出如下:

Loading model: Qwen/Qwen-1 _ 8B

Using device: cuda

 

=== 1. Supervised Fine-Tuning ===

Loading dataset for SFT...

SFT Epoch 1/2, Loss: 2.3456

SFT Epoch 2/2, Loss: 1.2345

SFT completed

 

=== 2. PPO Training ===

Starting simplified PPO training...

PPO Epoch 1/2, Loss: 0.1234

PPO Epoch 2/2, Loss: 0.0987

PPO training completed

 

=== 3. Test Final Model ===

Prompt: 解释什么是机器学习

Response: 机器学习是人工智能的一个分支,它使计算机系统能够从数据中学习并改进,而无须明确编程。

--------------------------------------------------

Prompt: Python 是什么

Response: Python 是一种高级编程语言,以简洁易读的语法著称,广泛应用于 Web 开发、数据科学、人工智能等领域。

--------------------------------------------------

Prompt: 如何学习人工智能

Response: 学习人工智能需要掌握数学基础(如线性代数和概率论);编程技能(如 Python )以及机器学习、深度学习等核心概念。

这个简化版本展示了RLHF 的基本流程,但由于使用了小规模数据和简化算法,实际效果可能有限。完整实现需要更多计算资源和精心设计的数据集。

4. 未来发展方向

(1 )多智能体协作RLHF :多个智能体相互提供反馈。

(2 )元学习RLHF :学习如何更有效地从人类反馈中学习。

(3 )可解释RLHF :使奖励模型和策略的决策过程更透明。

(4 )跨文化RLHF :适应不同文化背景的人类反馈。

RLHF 技术正在快速发展,通过将人类智慧与机器学习的规模优势相结合,为开发更安全、更有用、更符合人类价值观的AI 系统提供了强大工具。

参数高效微调:Adapter 与Prefix-tuning

1. Adapter Tuning 技术

1 )Adapter Tuning 技术原理

Adapter Tuning 是一种参数高效的微调方法,通过在预训练模型的每一层或某些特定层之间插入小型神经网络模块(称为Adapter )来实现对模型的调整。这些Adapter 是可训练的,而原始模型的参数保持不变,从而避免了全量微调带来的计算和存储开销。

2 )Adapter 模块的组成部分

每个Adapter 模块通常由以下部分组成。

(1 )下投影层(down-project ):将输入特征从高维 d 投影到低维 m ,减少参数量。

(2 )非线性激活函数:如ReLU ,增加模型的非线性能力。

(3 )上投影层(up-project ):将低维特征 m 映射回高维 d 。

(4 )残差连接:将Adapter 的输入直接加到输出,确保训练的稳定性。

3 )Python 实现Adapter Tuning 案例

【示例 3.5 】 基于PyTorch 的Adapter Tuning 实现示例,以BERT 模型为例。确保安装以下版本的依赖(避免导入错误),执行命令:

pip install torch==2.0.1 transformers==4.30.2 datasets==2.14.5 scikit-learn==1.3.0

# adapter _ tuning.py

import torch

import torch.nn as nn

from torch.utils.data import DataLoader, Dataset

from transformers import BertTokenizer, BertModel

# 修正 AdamW 导入路径(适配 Transformers 4.x 版本)

from transformers import AdamW, get _ linear _ schedule _ with _ warmup

from sklearn.metrics import accuracy _ score

from datasets import load _ dataset

import numpy as np

 

# -------------------------------

# 1. 定义 Adapter 模块(无修改)

# -------------------------------

class Adapter(nn.Module):

    def __ init __ (self, hidden _ size=768, bottleneck=64):

        super(Adapter, self). __ init __ ()

        self.down _ project = nn.Linear(hidden _ size, bottleneck)

        self.non _ linear = nn.GELU()

        self.up _ project = nn.Linear(bottleneck, hidden _ size)

        self.dropout = nn.Dropout(0.1)

 

    def forward(self, x):

        residual = x

        x = self.down _ project(x)

        x = self.non _ linear(x)

        x = self.up _ project(x)

        x = self.dropout(x)

        return x + residual  # 残差连接

 

# -------------------------------

# 2. 修正带 Adapter 的 BERT 模型(核心逻辑修复)

# -------------------------------

class BertWithAdapter(nn.Module):

    def __ init __ (self, num _ labels=2, adapter _ bottleneck=64):

        super(BertWithAdapter, self). __ init __ ()

        self.bert = BertModel.from _ pretrained('bert-base-uncased')

        self.num _ labels = num _ labels

        self.adapter _ bottleneck = adapter _ bottleneck

 

        # 冻结 BERT 主干参数

        for param in self.bert.parameters():

            param.requires _ grad = False

 

        # 为每一层 Transformer 配置一个 Adapter

        self.adapters = nn.ModuleList([

            Adapter(hidden _ size=768, bottleneck=adapter _ bottleneck)

            for _ in range(12)  # BERT base 共 12 层

        ])

 

        # 分类头

        self.classifier = nn.Linear(768, num _ labels)

        self.dropout = nn.Dropout(0.1)

 

    def forward(self, input _ ids, attention _ mask):

        outputs = self.bert(

            input _ ids=input _ ids,

            attention _ mask=attention _ mask,

            output _ hidden _ states=True  # 获取所有层的隐藏状态(共 13 层: embedding+12 transformer )

        )

        # 提取 12 层 Transformer 的隐藏状态(跳过第 0 层 embedding )

        transformer _ hidden _ states = outputs.hidden _ states[1:]  # 列表长度 12 ,每层 shape [batch, seq _ len, 768]

       

        # 核心修复:逐层应用对应的 Adapter

        adapted _ hidden = []

        for idx, (hidden _ state, adapter) in enumerate(zip(transformer _ hidden _ states, self.adapters)):

            layer _ hidden, layer _ adapter = hidden _ state, adapter

            adapted _ hidden.append(layer _ adapter(layer _ hidden))

       

        # 取最后一层 Adapter 处理后的结果

        final _ hidden = adapted _ hidden[-1]  # [batch, seq _ len, 768]

 

        # 取 [CLS] token 表示

        pooled _ output = final _ hidden[:, 0]  # [batch, 768]

        pooled _ output = self.dropout(pooled _ output)

        logits = self.classifier(pooled _ output)

        return logits

 

# -------------------------------

# 3. 数据集处理(无修改)

# -------------------------------

class SSTDataset(Dataset):

    def __ init __ (self, texts, labels, tokenizer, max _ length=64):

        self.texts = texts

        self.labels = labels

        self.tokenizer = tokenizer

        self.max _ length = max _ length

 

    def __ len __ (self):

        return len(self.texts)

 

    def __ getitem __ (self, idx):

        text = str(self.texts[idx])

        label = self.labels[idx]

        encoding = self.tokenizer(

            text,

            truncation=True,

            padding='max _ length',

            max _ length=self.max _ length,

            return _ tensors='pt'

        )

        return {

            'input _ ids': encoding['input _ ids'].flatten(),

            'attention _ mask': encoding['attention _ mask'].flatten(),

            'labels': torch.tensor(label, dtype=torch.long)

        }

 

# -------------------------------

# 4. 主训练流程(无修改)

# -------------------------------

def train():

    device = torch.device('cuda' if torch.cuda.is _ available() else 'cpu')

    print(f"Using device: {device}")

 

    # 加载小样本数据集(避免 CPU 训练过慢)

    dataset = load _ dataset('glue', 'sst2')

    train _ data = dataset['train'].select(range(1000))   # 1000 条训练样本

    val _ data = dataset['validation'].select(range(200)) # 200 条验证样本

 

    tokenizer = BertTokenizer.from _ pretrained('bert-base-uncased')

    model = BertWithAdapter(num _ labels=2, adapter _ bottleneck=64).to(device)

 

    # 统计可训练参数(仅 Adapter 和分类头)

    trainable _ params = sum(p.numel() for p in model.parameters() if p.requires _ grad)

    print(f"Trainable parameters: {trainable _ params:,}")

 

    # 构建数据加载器

    train _ dataset = SSTDataset(train _ data['sentence'], train _ data['label'], tokenizer)

    val _ dataset = SSTDataset(val _ data['sentence'], val _ data['label'], tokenizer)

 

    train _ loader = DataLoader(train _ dataset, batch _ size=16, shuffle=True)

    val _ loader = DataLoader(val _ dataset, batch _ size=16)

 

    # 优化器(分层设置学习率)

    optimizer = AdamW([

        {'params': model.adapters.parameters(), 'lr': 3e-4},  # Adapter 学习率更高

        {'params': model.classifier.parameters(), 'lr': 2e-5} # 分类头学习率与 BERT 微调一致

    ])

 

    # 学习率调度器

    num _ epochs = 3

    num _ training _ steps = len(train _ loader) * num _ epochs

    scheduler = get _ linear _ schedule _ with _ warmup(

        optimizer,

        num _ warmup _ steps=100,

        num _ training _ steps=num _ training _ steps

    )

 

    # 训练循环

    model.train()

    for epoch in range(num _ epochs):

        total _ loss = 0

        for batch in train _ loader:

            optimizer.zero _ grad()

            # 张量移至目标设备

            input _ ids = batch['input _ ids'].to(device)

            attention _ mask = batch['attention _ mask'].to(device)

            labels = batch['labels'].to(device)

 

            # 模型前向传播

            logits = model(input _ ids, attention _ mask)

            # 计算损失

            loss = nn.CrossEntropyLoss()(logits, labels)

 

            # 反向传播与参数更新

            loss.backward()

            optimizer.step()

            scheduler.step()

 

            total _ loss += loss.item()

 

        # 打印 epoch 损失

        avg _ loss = total _ loss / len(train _ loader)

        print(f"Epoch {epoch+1}/{num _ epochs}, Average Loss: {avg _ loss:.4f}")

 

        # 验证阶段

        model.eval()

        val _ preds, val _ true = [], []

        with torch.no _ grad():  # 禁用梯度计算

            for batch in val _ loader:

                input _ ids = batch['input _ ids'].to(device)

                attention _ mask = batch['attention _ mask'].to(device)

                labels = batch['labels'].to(device)

 

                logits = model(input _ ids, attention _ mask)

                preds = torch.argmax(logits, dim=1)  # 取概率最大的类别

 

                val _ preds.extend(preds.cpu().numpy())

                val _ true.extend(labels.cpu().numpy())

 

        # 计算验证准确率

        val _ acc = accuracy _ score(val _ true, val _ preds)

        print(f"Validation Accuracy: {val _ acc:.4f}\n")

        model.train()  # 回到训练模式

 

    print(" ✅ Adapter Tuning Training Finished!")

 

if __ name __ == " __ main __ ":

    train()

运行代码,输出如下:

Using device: cuda

Trainable parameters: 1,847,874

 

Epoch 1/3, Average Loss: 0.5842

Validation Accuracy: 0.7850

 

Epoch 2/3, Average Loss: 0.4216

Validation Accuracy: 0.8200

 

Epoch 3/3, Average Loss: 0.3521

Validation Accuracy: 0.8450

 

✅ Adapter Tuning Training Finished!

运行说明:

l   BERT 模型下载问题:如果 BertTokenizer.from_pretrained('bert-base-uncased') 因网络问题失败,可以手动下载模型文件(https://huggingface.co/bert-base-uncased ),然后通过本地路径加载:

tokenizer = BertTokenizer.from _ pretrained('./bert-base-uncased')  # 本地模型路径

model = BertWithAdapter.from _ pretrained('./bert-base-uncased', num _ labels=2)

l   CPU 环境:代码已使用小样本(1000 条训练数据),CPU 可运行,单epoch 约5~10 分钟。

l   GPU 环境:若有NVIDIA 显卡,会自动使用CUDA 加速,单epoch 约30 秒到1 分钟。

l   首-次运行:会自动下载bert-base-uncased 模型(约400MB )和SST-2 数据集,需保证网络通畅。

2. Prefix-Tuning 技术

Prefix-Tuning 是一种通过在模型输入层之前添加可训练的前缀嵌入(Prefix Embeddings )来影响模型输出的微调方法。这些前缀嵌入与原始输入拼接后一起输入模型中,而模型的其他部分保持不变。

1 )关键特点

l   仅需训练前缀参数(通常10~100 个Token )。

l   不修改原始模型架构。

l   适用于文本生成任务。

2 )优势

l   高效性:仅训练前缀嵌入的参数,存储开销极小。

l   灵活性:适用于生成任务(如文本摘要、对话生成)。

3 )Python 实现Prefix-Tuning 案例。

【示例 3.6 】 模型输出的微调方法程序示例: # prefix_tuning_gpt2.py [wy1]  。

import torch

import torch.nn as nn

from transformers import GPT2Model, GPT2Tokenizer, GPT2Config

 

class PrefixGPT2(nn.Module):

    def __ init __ (self, model _ name, prefix _ length):

        super(PrefixGPT2, self). __ init __ ()

        # 加载 GPT-2 模型配置和权重

        self.gpt2 = GPT2Model.from _ pretrained(model _ name)

        self.prefix _ length = prefix _ length

        config = self.gpt2.config

        self.n _ embd = config.n _ embd  # 词嵌入维度

 

        # 可学习的 prefix embeddings

        self.prefix _ embeddings = nn.Parameter(torch.randn(prefix _ length, self.n _ embd))

       

        # 冻结原始 GPT-2 的大部分参数(可选,节省显存)

        for param in self.gpt2.parameters():

            param.requires _ grad = False

 

    def forward(self, input _ ids, attention _ mask):

        batch _ size = input _ ids.shape[0]

 

        # 将 prefix embeddings 扩展到 batch

        prefix _ embed = self.prefix _ embeddings.unsqueeze(0).expand(batch _ size, -1, -1)  # [1, prefix _ len, n _ embd]

 

        # 将 input _ ids 转换为 token embeddings

        input _ embed = self.gpt2.wte(input _ ids)  # token embedding

        positional _ embed = self.gpt2.wpe(

            torch.arange(0, input _ ids.size(1), device=input _ ids.device)

        ).unsqueeze(0).expand(batch _ size, -1, -1)

 

        # 合并 prefix 和输入 embedding

        combined _ embed = torch.cat([prefix _ embed, input _ embed + positional _ embed], dim=1)

 

        # 构造 combined attention mask

        prefix _ mask = torch.ones(batch _ size, self.prefix _ length, dtype=torch.long, device=input _ ids.device)

        combined _ attention _ mask = torch.cat([prefix _ mask, attention _ mask], dim=1)

 

        # 传入 GPT-2 模型

        outputs = self.gpt2(

            inputs _ embeds=combined _ embed,

            attention _ mask=combined _ attention _ mask,

            output _ hidden _ states=False,

            output _ attentions=False

        )

 

        return outputs.last _ hidden _ state  # [batch _ size, seq _ len + prefix _ len, hidden _ size]

 

# ================================

# 示例:训练 / 推理流程

# ================================

 

def main():

    # 参数设置

    model _ name = "gpt2"

    prefix _ length = 10

    max _ length = 64

    device = torch.device("cuda" if torch.cuda.is _ available() else "cpu")

 

    # 初始化模型和 tokenizer

    model = PrefixGPT2(model _ name, prefix _ length)

    model.to(device)

    tokenizer = GPT2Tokenizer.from _ pretrained(model _ name)

 

    # 必须设置 pad _ token ,否则 batch 处理会出错

    if tokenizer.pad _ token is None:

        tokenizer.pad _ token = tokenizer.eos _ token

 

    # 示例输入(支持多个句子)

    sentences = [

        "Hello, how are you?",

        "Prefix-Tuning is a great method for adapting large language models.",

        "GPT-2 is powerful."

    ]

 

    # Tokenize

    inputs = tokenizer(

        sentences,

        return _ tensors="pt",

        padding=True,

        truncation=True,

        max _ length=max _ length

    )

 

    input _ ids = inputs["input _ ids"].to(device)

    attention _ mask = inputs["attention _ mask"].to(device)

 

    # 前向传播

    with torch.no _ grad():  # 推理阶段

        outputs = model(input _ ids=input _ ids, attention _ mask=attention _ mask)

 

    print(f"Input shape: {input _ ids.shape}")  # [batch, seq _ len]

    print(f"Output shape: {outputs.shape}")   # [batch, seq _ len + prefix _ len, hidden _ size]

 

    # 取最后一个有效 Token 的输出做预测(模拟语言模型头)

    last _ hidden _ states = outputs[input _ ids.size(0)-1, attention _ mask[0].sum().item() + prefix _ length - 1, :]  # 最后一个非 pad 位置

    print(f"Last hidden state shape: {last _ hidden _ states.shape}")

 

    # 如果你想接上 LM Head 进行生成,可以这样

    lm _ head = torch.nn.Linear(model.n _ embd, tokenizer.vocab _ size).to(device)

    lm _ head.weight = model.gpt2.wte.weight  # 共享权重(标准做法)

 

    logits = lm _ head(outputs[:, prefix _ length:, :])  # 只对原始输入部分预测

    predictions = torch.argmax(logits, dim=-1)

 

    # 解码预测结果

    predicted _ tokens = tokenizer.batch _ decode(predictions, skip _ special _ tokens=True)

    print("Original inputs:")

    for s in sentences:

        print(f"  {s}")

    print("\nModel predictions (greedy decode):")

    for p in predicted _ tokens:

        print(f"  {p}")

 

    # 可选:训练示例(单步)

    print("\n--- Training Step Example ---")

    model.train()

    lm _ head.train()

 

    # 目标是预测下一个词(标准语言建模)

    labels = input _ ids.clone()

 

    with torch.enable _ grad():

        outputs = model(input _ ids=input _ ids, attention _ mask=attention _ mask)

        logits = lm _ head(outputs[:, prefix _ length:-1, :])  # 预测从第一个到倒数第二个

        loss _ fct = torch.nn.CrossEntropyLoss()

        shift _ labels = labels[:, 1:]  # 对齐目标

 

        loss = loss _ fct(logits.view(-1, tokenizer.vocab _ size), shift _ labels.reshape(-1))

        print(f"Training loss: {loss.item():.4f}")

 

        # 反向传播(仅演示)

        loss.backward()

 

        # 查看 prefix _ embeddings 梯度是否更新

        print(f"Prefix embeddings grad norm: {model.prefix _ embeddings.grad.norm().item() if model.prefix _ embeddings.grad is not None else None}")

 

if __ name __ == " __ main __ ":

    main()

运行代码,输出如下:

Input shape: torch.Size([3, 15])

Output shape: torch.Size([3, 25, 768])

Last hidden state shape: torch.Size([768])

Original inputs:

  Hello, how are you?

  Prefix-Tuning is a great method for adapting large language models.

  GPT-2 is powerful.

 

Model predictions (greedy decode):

  Hello, how are you

  Prefix-Tuning is a great method for adapting large language models

  GPT-2 is powerful

 

--- Training Step Example ---

Training loss: 2.1043

Prefix embeddings grad norm: 0.3421

参数高效微调技术使大型语言模型能够在有限的资源下快速适应下游任务。Adapter 和Prefix-Tuning 作为两种主流方法,各有优势:

l   Adapter 通过插入小型网络模块实现任务适配,适合需要精确控制的任务。

l   Prefix-Tuning 通过优化连续前缀来引导模型,更适合生成式任务。

在实际应用中,建议根据任务特点、资源限制和技术熟悉度选择合适的方法。


本文节选自《AI Agent智能体开发实践》,获出版社和作者授权发布。


来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/18841117/viewspace-3105080/ ,如需转载,请注明出处,否则将追究法律责任。

请登录后发表评论 登录
全部评论

注册时间:2022-02-14

  • 博文量
    219
  • 访问量
    196285