pip install transformers torch datasets

from transformers import (
    AutoTokenizer,                    # 自动加载分词器
    AutoModel,                        # 基础 BERT 模型（提取特征）
    AutoModelForMaskedLM,             # 填空任务模型
    AutoModelForNextSentencePrediction,  # 下一句预测模型
)

model_name = "google-bert/bert-base-chinese"  # 模型名称
cache_dir = r"D:\Desktop\LLM_start\3_Bert_Model_train\model"  # 本地缓存路径

# 从本地加载模型
tokenizer = AutoTokenizer.from_pretrained(
    model_name,
    cache_dir=cache_dir,
    local_files_only=True  # 仅从本地加载，不联网下载
)
model = AutoModel.from_pretrained(
    model_name,
    cache_dir=cache_dir,
    local_files_only=True
)

import torch
from transformers import BertTokenizer, BertForSequenceClassification, pipeline

model_name = "google-bert/bert-base-chinese"
cache_dir = r"D:\Desktop\LLM_start\3_Bert_Model_train\model"

tokenizer = BertTokenizer.from_pretrained(model_name, cache_dir=cache_dir, local_files_only=True)
model = BertForSequenceClassification.from_pretrained(model_name, cache_dir=cache_dir, local_files_only=True)

# 创建分类 pipeline
classifier = pipeline("text-classification", model=model, tokenizer=tokenizer, device="cuda")

# 推理
result = classifier("我喜欢学习人工智能。", top_k=2)
print(result)

from transformers import AutoTokenizer, AutoModelForMaskedLM, pipeline
import torch

model_name = "google-bert/bert-base-chinese"
cache_dir = r"D:\Desktop\LLM_start\3_Bert_Model_train\model"

tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir, local_files_only=True)
model = AutoModelForMaskedLM.from_pretrained(model_name, cache_dir=cache_dir, local_files_only=True)

# 创建填空 pipeline
fill_mask = pipeline("fill-mask", model=model, tokenizer=tokenizer, 
                     device="cuda" if torch.cuda.is_available() else "cpu")

# 使用 [MASK] 遮盖一个词
text = f"我喜欢学{tokenizer.mask_token}语。"

# 预测被遮盖的词
results = fill_mask(text, top_k=5)

print("输入:", text)
for item in results:
    print(f"score={item['score']:.4f}, token={item['token_str']}, sentence={item['sequence']}")

import torch
from transformers import AutoTokenizer, AutoModelForNextSentencePrediction

model_name = "google-bert/bert-base-chinese"
cache_dir = r"D:\Desktop\LLM_start\3_Bert_Model_train\model"

tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=cache_dir, local_files_only=True)
model = AutoModelForNextSentencePrediction.from_pretrained(model_name, cache_dir=cache_dir, local_files_only=True)
model.eval()

# 输入两个句子
sentence_a = "我喜欢学习人工智能。"
sentence_b = "它可以帮助我们理解语言。"

# 编码（会将两个句子拼接）
inputs = tokenizer(sentence_a, sentence_b, return_tensors="pt")

with torch.no_grad():
    outputs = model(**inputs)

# 输出是二分类 logits
probs = torch.softmax(outputs.logits, dim=-1)[0]
labels = ["是下一句", "不是下一句"]
prediction = labels[int(torch.argmax(probs))]

print("句子 A:", sentence_a)
print("句子 B:", sentence_b)
print(f"{labels[0]}: {probs[0].item():.4f}")
print(f"{labels[1]}: {probs[1].item():.4f}")
print("预测结果:", prediction)

from datasets import load_dataset, load_from_disk

# 首次下载（取消注释执行）
# ds = load_dataset(
#     "lansinuote/ChnSentiCorp",
#     cache_dir=r"D:\Desktop\LLM_start\3_Bert_Model_train\Dataset"
# )

# 保存到本地磁盘
# ds.save_to_disk(r"D:\Desktop\LLM_start\3_Bert_Model_train\Dataset")

# 后续从本地加载
ds_local = load_from_disk(r"D:\Desktop\LLM_start\3_Bert_Model_train\Dataset")
print(ds_local)

# 查看数据
test_ds = ds_local["test"]
print(test_ds)

for data in test_ds:
    print(data)

{
    'text': '这个酒店服务态度很好，房间干净整洁',  # 文本内容
    'label': 1                                    # 标签（0=负面, 1=正面）
}

from torch.utils.data import Dataset
from datasets import load_from_disk

class mydataset(Dataset):
    """
    自定义 PyTorch Dataset 类，用于加载情感分析数据集
    
    Args:
        split: 数据划分，可选 'train', 'test', 'validation'
    """
    
    def __init__(self, split):
        # 从本地加载整个数据集
        self.dataset = load_from_disk(r"D:\Desktop\LLM_start\3_Bert_Model_train\Dataset")
        
        # 根据 split 选择对应划分
        if split == "train":
            self.dataset = self.dataset["train"]
        elif split == "test":
            self.dataset = self.dataset["test"]
        elif split == "validation":
            self.dataset = self.dataset["validation"]
        else:
            raise ValueError("Invalid split name. Use 'train', 'test', or 'validation'.")

    def __len__(self):
        """返回数据集大小"""
        return len(self.dataset)

    def __getitem__(self, item):
        """获取单个样本"""
        data = self.dataset[item]["text"]    # 获取文本
        label = self.dataset[item]["label"]  # 获取标签
        return data, label
    
if __name__ == "__main__":
    test_dataset = mydataset("test")
    
    for data in test_dataset:
        print(data)

from transformers import BertModel
import torch

# 加载预训练 BERT 模型
pretrained_model = BertModel.from_pretrained(
    "google-bert/bert-base-chinese",
    cache_dir=r"D:\Desktop\LLM_start\3_Bert_Model_train\model",
    local_files_only=True
)

class Model(torch.nn.Module):
    """
    自定义 BERT 文本分类模型
    
    架构：BERT (冻结) + 全连接层 (可训练)
    """
    
    def __init__(self):
        super().__init__()
        self.bert = pretrained_model           # 预训练 BERT
        self.fc = torch.nn.Linear(768, 2)      # 分类头：768 → 2

    def forward(self, input_ids, attention_mask, token_type_ids):
        """
        前向传播
        
        Args:
            input_ids: token ID 序列
            attention_mask: 注意力掩码（区分真实token和padding）
            token_type_ids: token 类型 ID（区分句子A和句子B）
        
        Returns:
            logits: 分类概率 [batch_size, 2]
        """
        # BERT 前向传播（冻结参数，不参与训练）
        with torch.no_grad():
            outputs = self.bert(
                input_ids=input_ids,
                attention_mask=attention_mask,
                token_type_ids=token_type_ids
            )
        
        # 取 [CLS] token 的输出作为句子表示
        # last_hidden_state: [batch_size, seq_len, 768]
        # [:, 0, :] 取第0个位置（[CLS]）的向量
        cls_output = outputs.last_hidden_state[:, 0, :]  # [batch_size, 768]
        
        # 全连接层分类
        outputs = self.fc(cls_output)  # [batch_size, 2]
        
        # softmax 转为概率
        logits = torch.softmax(outputs, dim=-1)
        
        return logits

with torch.no_grad():
    outputs = self.bert(...)

from ex3_1_New_class_model import Model      # 导入自定义模型
from transformers import BertTokenizer        # 分词器
from ex2_2_Make_dataset import mydataset     # 导入数据集类
import torch
from torch.utils.data import DataLoader      # 数据加载器
from torch.optim import AdamW                # 优化器

# ==================== 配置 ====================
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model_name = "google-bert/bert-base-chinese"
cache_dir = r"D:\Desktop\LLM_start\3_Bert_Model_train\model"

# 加载分词器
token = BertTokenizer.from_pretrained(
    model_name,
    cache_dir=cache_dir,
    local_files_only=True
)

# ==================== 数据预处理函数 ====================
def collate_fn(data):
    """
    DataLoader 的批处理函数
    
    **collate_fn（整理函数）**：
    - collate_fn 是 DataLoader 的自定义批处理函数
    - 当 DataLoader 从 Dataset 中取出一批数据时，需要将这些数据整理成统一的 tensor 格式
    - 默认的 collate_fn 只能处理简单的数据格式，复杂场景需要自定义
    - 在本项目中，collate_fn 负责将文本列表批量编码为 token 序列，并转换为 tensor
    
    Args:
        data: list of (text, label) tuples
    
    Returns:
        input_ids, attention_mask, token_type_ids, labels (均为 tensor)
    """
    # 分离文本和标签
    sentes = [i[0] for i in data]    # 文本列表
    labels = [i[1] for i in data]    # 标签列表
    
    # 批量编码
    encoding = token.batch_encode_plus(
        batch_text_or_text_pairs=sentes,  # 待编码文本
        truncation=True,                  # 超长截断。**Truncation（截断）**：当文本长度超过 max_length 时，自动截断超出部分。BERT 最大支持 512 个 token，超出会导致错误，截断可确保输入长度符合模型要求
        padding="max_length",             # 填充到固定长度
        max_length=360,                   # 最大长度
        return_tensors="pt",              # 返回 PyTorch tensor
        return_length=True                # 返回实际长度
    )
    
    # 提取各部分
    input_ids = encoding["input_ids"]           # token ID
    attention_mask = encoding["attention_mask"] # 注意力掩码
    token_type_ids = encoding["token_type_ids"] # token 类型
    
    # 标签转为 tensor
    labels = torch.tensor(labels, dtype=torch.long)
    
    return input_ids, attention_mask, token_type_ids, labels


# ==================== 数据加载 ====================
train_dataset = mydataset("train")
train_dataloader = DataLoader(
    train_dataset,
    batch_size=64,        # 每批 64 个样本
    shuffle=True,         # 打乱顺序。**Shuffle（打乱）**：在每个 epoch 开始时随机打乱数据顺序，避免模型按固定顺序学习样本，有助于模型更泛化，防止因样本顺序产生的偏差
    drop_last=True,       # 丢弃不完整的最后一批
    collate_fn=collate_fn  # 批处理函数
)


# ==================== 训练主循环 ====================
if __name__ == "__main__":
    
    epochs = 10                          # 训练轮数
    print(device)
    
    model = Model().to(device)           # 初始化模型并移到 GPU
    
    # 只优化全连接层参数（BERT 已冻结）
    optimizer = AdamW(model.fc.parameters(), lr=5e-4)
    
    # 交叉熵损失函数
    loss_fn = torch.nn.CrossEntropyLoss()

    for epoch in range(epochs):
        model.train()
        model.bert.eval()  # BERT 保持 eval 模式
        
        total_loss = 0.0
        total_correct = 0
        total_num = 0
        
        for step, batch in enumerate(train_dataloader, start=1):
            # 获取 batch 数据
            input_ids, attention_mask, token_type_ids, labels = batch
            
            # 移到 GPU
            input_ids = input_ids.to(device)
            attention_mask = attention_mask.to(device)
            token_type_ids = token_type_ids.to(device)
            labels = labels.to(device)
            
            # ===== 前向传播 =====
            outputs = model(input_ids, attention_mask, token_type_ids)
            
            # ===== 计算损失 =====
            loss = loss_fn(outputs, labels)
            
            # ===== 反向传播 =====
            optimizer.zero_grad()  # 清零梯度
            loss.backward()        # 计算梯度
            optimizer.step()       # 更新参数
            
            # ===== 统计 =====
            batch_size = labels.size(0)
            total_loss += loss.item()
            preds = torch.argmax(outputs, dim=1)
            total_correct += (preds == labels).sum().item()
            total_num += batch_size
            
            # 每 64 步打印进度
            if step % 64 == 0 or step == len(train_dataloader):
                avg_loss = total_loss / step
                train_acc = total_correct / total_num
                
                print(
                    f"Epoch [{epoch + 1}/{epochs}] "
                    f"Step [{step}/{len(train_dataloader)}] "
                    f"Loss: {avg_loss:.4f} "
                    f"Train Acc: {train_acc:.4f}"
                )
        
        # 每轮结束统计
        avg_loss = total_loss / len(train_dataloader)
        train_acc = total_correct / total_num
        
        print(
            f"========== Epoch {epoch + 1}/{epochs} 完成 ==========\n"
            f"Avg Loss: {avg_loss:.4f}\n"
            f"Train Acc: {train_acc:.4f}\n"
        )
    
    # ===== 保存模型 =====
    torch.save(
        model.state_dict(),
        r"D:\Desktop\LLM_start\3_Bert_Model_train\bert_fc_sentiment.pth"
    )
    
    print("模型参数已保存")

optimizer = AdamW(model.fc.parameters(), lr=5e-4)

encoding = token.batch_encode_plus(
    batch_text_or_text_pairs=sentes,  # 输入文本列表
    truncation=True,                  # 超过 max_length 截断
    padding="max_length",             # 填充到 max_length
    max_length=360,                   # 最大序列长度
    return_tensors="pt",              # 返回 PyTorch tensor
)

from ex3_1_New_class_model import Model
from ex2_2_Make_dataset import mydataset
from transformers import BertTokenizer
import torch
from torch.utils.data import DataLoader

# ==================== 配置 ====================
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
batch_size = 64
max_length = 360
model_path = r"D:\Desktop\LLM_start\3_Bert_Model_train\bert_fc_sentiment.pth"
model_name = "google-bert/bert-base-chinese"
cache_dir = r"D:\Desktop\LLM_start\3_Bert_Model_train\model"

token = BertTokenizer.from_pretrained(
    model_name,
    cache_dir=cache_dir,
    local_files_only=True
)

# ==================== collate_fn ====================
def collate_fn(data):
    sentes = [i[0] for i in data]
    labels = [i[1] for i in data]
    
    encoding = token.batch_encode_plus(
        batch_text_or_text_pairs=sentes,
        truncation=True,
        padding="max_length",
        max_length=max_length,
        return_tensors="pt"
    )
    
    input_ids = encoding["input_ids"]
    attention_mask = encoding["attention_mask"]
    token_type_ids = encoding["token_type_ids"]
    labels = torch.tensor(labels, dtype=torch.long)
    
    return input_ids, attention_mask, token_type_ids, labels


# ==================== 评估函数 ====================
def evaluate(model, dataloader, loss_fn):
    """
    模型评估函数
    
    Args:
        model: 待评估模型
        dataloader: 测试数据加载器
        loss_fn: 损失函数
    
    Returns:
        avg_loss: 平均损失
        acc: 准确率
        stat: 统计信息字典
    """
    model.eval()
    
    total_loss = 0.0
    total_correct = 0
    total_num = 0
    
    # 分类统计
    total_positive = 0    # 真实正面样本数
    total_negative = 0    # 真实负面样本数
    pred_positive = 0     # 预测正面样本数
    pred_negative = 0     # 预测负面样本数
    
    with torch.no_grad():  # 评估时不计算梯度
        for batch in dataloader:
            input_ids, attention_mask, token_type_ids, labels = batch
            
            # 移到 GPU
            input_ids = input_ids.to(device)
            attention_mask = attention_mask.to(device)
            token_type_ids = token_type_ids.to(device)
            labels = labels.to(device)
            
            # 前向传播
            outputs = model(input_ids, attention_mask, token_type_ids)
            
            # 计算损失
            loss = loss_fn(outputs, labels)
            
            # 预测结果
            preds = torch.argmax(outputs, dim=1)
            
            batch_num = labels.size(0)
            
            # 累计统计
            total_loss += loss.item() * batch_num
            total_correct += (preds == labels).sum().item()
            total_num += batch_num
            
            # 分类统计
            total_positive += (labels == 1).sum().item()
            total_negative += (labels == 0).sum().item()
            pred_positive += (preds == 1).sum().item()
            pred_negative += (preds == 0).sum().item()
    
    avg_loss = total_loss / total_num
    acc = total_correct / total_num
    
    return avg_loss, acc, {
        "total_num": total_num,
        "total_positive": total_positive,
        "total_negative": total_negative,
        "pred_positive": pred_positive,
        "pred_negative": pred_negative,
    }


# ==================== 主程序 ====================
if __name__ == "__main__":
    print("当前设备:", device)
    
    # 1. 加载测试集
    test_dataset = mydataset("test")
    test_dataloader = DataLoader(
        test_dataset,
        batch_size=batch_size,
        shuffle=False,
        drop_last=False,
        collate_fn=collate_fn
    )
    
    print("测试集样本数:", len(test_dataset))
    print("测试集 batch 数:", len(test_dataloader))
    
    # 2. 加载训练好的模型
    model = Model().to(device)
    state_dict = torch.load(model_path, map_location=device)
    model.load_state_dict(state_dict)
    print("模型参数加载成功:", model_path)
    
    # 3. 评估模型
    loss_fn = torch.nn.CrossEntropyLoss()
    
    test_loss, test_acc, stat = evaluate(
        model=model,
        dataloader=test_dataloader,
        loss_fn=loss_fn
    )
    
    # 4. 打印结果
    print("========== 测试结果 ==========")
    print(f"Test Loss: {test_loss:.4f}")
    print(f"Test Acc : {test_acc:.4f}")
    print(f"测试样本总数: {stat['total_num']}")
    print(f"真实正面样本数: {stat['total_positive']}")
    print(f"真实负面样本数: {stat['total_negative']}")
    print(f"预测正面样本数: {stat['pred_positive']}")
    print(f"预测负面样本数: {stat['pred_negative']}")

from ex3_1_New_class_model import Model
from transformers import BertTokenizer
import torch

# ==================== 配置 ====================
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model_name = "google-bert/bert-base-chinese"
cache_dir = r"D:\Desktop\LLM_start\3_Bert_Model_train\model"
model_path = r"D:\Desktop\LLM_start\3_Bert_Model_train\bert_fc_sentiment.pth"
max_length = 360
names = ["负面", "正面"]

# ==================== 加载 tokenizer ====================
tokenizer = BertTokenizer.from_pretrained(
    model_name,
    cache_dir=cache_dir,
    local_files_only=True
)

# ==================== 单句编码函数 ====================
def encode_sentence(sentence):
    """
    将单个句子编码为模型输入格式
    
    Args:
        sentence: 输入文本
    
    Returns:
        input_ids, attention_mask, token_type_ids (均为 tensor)
    """
    encoding = tokenizer.batch_encode_plus(
        batch_text_or_text_pairs=[sentence],
        truncation=True,
        padding="max_length",
        max_length=max_length,
        return_tensors="pt"
    )
    
    input_ids = encoding["input_ids"].to(device)
    attention_mask = encoding["attention_mask"].to(device)
    token_type_ids = encoding["token_type_ids"].to(device)
    
    return input_ids, attention_mask, token_type_ids


# ==================== 加载训练好的模型 ====================
def load_trained_model():
    """加载训练好的模型"""
    model = Model().to(device)
    
    # 加载权重
    state_dict = torch.load(model_path, map_location=device)
    model.load_state_dict(state_dict)
    
    # 设置为评估模式
    model.eval()
    
    return model


# ==================== 预测函数 ====================
def predict(model, sentence):
    """
    对单个句子进行情感预测
    
    Args:
        model: 分类模型
        sentence: 输入文本
    
    Returns:
        pred: 预测类别 (0 或 1)
        label_name: 类别名称 ("负面" 或 "正面")
        negative_prob: 负面概率
        positive_prob: 正面概率
    """
    # 编码输入
    input_ids, attention_mask, token_type_ids = encode_sentence(sentence)
    
    with torch.no_grad():
        outputs = model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            token_type_ids=token_type_ids
        )
        
        # 获取概率
        probs = outputs  # 模型输出已经是 softmax 结果
        pred = torch.argmax(probs, dim=1).item()
    
    label_name = names[pred]
    negative_prob = probs[0][0].item()
    positive_prob = probs[0][1].item()
    
    return pred, label_name, negative_prob, positive_prob


# ==================== 主程序：交互式输入 ====================
if __name__ == "__main__":
    print("当前设备:", device)
    print("正在加载模型...")
    
    model = load_trained_model()
    
    print("模型加载完成")
    print("请输入测试文本，输入 q 退出")
    print("-" * 50)
    
    while True:
        sentence = input("请输入测试文本：").strip()
        
        if sentence.lower() == "q":
            print("测试结束")
            break
        
        if sentence == "":
            print("输入为空，请重新输入")
            continue
        
        # 进行预测
        pred, label_name, negative_prob, positive_prob = predict(model, sentence)
        
        # 输出结果
        print("模型判定:", label_name)
        print("预测类别:", pred)
        print(f"负面概率: {negative_prob:.4f}")
        print(f"正面概率: {positive_prob:.4f}")
        print("-" * 50)