LLM从入门到实战:2万字完整学习指南

大家好,我是你的技术博主,最近整理了 29 份 LLM 学习笔记,把它们浓缩成了这份从基础到进阶到实战的完整学习指南,不管你是刚入门的新手,还是想要动手做项目的开发者,这份指南都能帮你系统掌握大模型的核心知识,跟着代码一步步动手实践,真正把大模型技术落地到你的项目里。


第一章:LLM 基础入门,搞懂大模型的核心逻辑

1.1 什么是大语言模型?

大语言模型(Large Language Model,LLM)是基于海量文本数据训练的深度学习模型,它能够理解和生成人类语言,本质上是通过学习语言的统计规律和语义表示,从而实现对自然语言的理解、生成、翻译等多种任务。

简单来说,LLM 就像一个读遍了互联网上几乎所有文本的 “超级大脑”,它通过预测下一个词的概率,来学习语言的规律,从而能够和人类进行流畅的对话,完成各种复杂的语言任务。

1.2 LLM 的发展三阶段

LLM 的发展可以分为三个清晰的阶段:

  1. 统计语言模型阶段:最早的语言模型是基于统计的,比如 N-gram 模型,它通过统计前 n 个词出现的概率,来预测下一个词的概率,但是这种方法的缺点很明显,它无法处理长距离的依赖,而且随着 n 的增大,参数会爆炸,泛化能力也很差。

  2. 神经网络语言模型阶段:后来,随着深度学习的发展,神经网络语言模型出现了,比如 RNN、LSTM 这些模型,它们能够处理长序列的文本,学习到更好的语义表示,但是这些模型的能力还是有限,无法处理海量的文本数据。

  3. Transformer 预训练语言模型阶段:2017 年 Transformer 架构的提出,彻底改变了 NLP 的格局,基于 Transformer 的预训练模型,比如 BERT、GPT,通过在海量文本上进行预训练,然后在下游任务上微调,取得了惊人的效果,而随着模型参数的不断增大,大语言模型应运而生,涌现出了很多之前小模型没有的能力,比如上下文学习、思维链等。

1.3 语言模型的核心评估指标

想要评估一个语言模型的好坏,我们需要用到几个核心的评估指标,下面我们来逐个讲解,并且附上代码实现,让你不仅懂原理,还能自己动手计算。

1.3.1 PPL:困惑度,衡量模型的预测能力

PPL(Perplexity,困惑度)是语言模型最常用的评估指标,它衡量的是模型对文本的预测能力,PPL 越小,说明模型对文本的预测越准确,模型的性能越好。

PPL 的计算公式是:
PPL=2−1N∑i=1Nlog⁡2p(wi∣w1...wi−1)PPL = 2^{-\frac{1}{N}\sum_{i=1}^N \log_2 p(w_i|w_1...w_{i-1})}PPL=2N1i=1Nlog2p(wiw1...wi1)

简单来说,就是对每个词的预测概率取负对数,然后求平均,再取 2 的指数,这样就得到了困惑度。

代码实现:

import torch
import numpy as np
from transformers import GPT2LMHeadModel, GPT2Tokenizer

def calculate_ppl(text, model, tokenizer, device='cuda'):
    # 对文本进行tokenize
    inputs = tokenizer(text, return_tensors='pt')
    inputs = inputs.to(device)
    # 获取模型输出
    with torch.no_grad():
        outputs = model(**inputs, labels=inputs.input_ids)
    # 计算loss,PPL就是exp(loss)
    loss = outputs.loss
    ppl = torch.exp(loss).item()
    return ppl

# 测试
tokenizer = GPT2Tokenizer.from_pretrained('gpt2')
model = GPT2LMHeadModel.from_pretrained('gpt2').to('cuda')
text = "Hello, my name is John, I am a machine learning engineer."
ppl = calculate_ppl(text, model, tokenizer)
print(f"PPL: {ppl}")
1.3.2 BLEU:衡量生成文本的准确率

BLEU(Bilingual Evaluation Understudy)是用来评估机器翻译的指标,它的核心思想是 n-gram 的精确率,也就是看生成的文本和参考文本之间的 n-gram 的重合度,BLEU 越高,说明生成的文本和参考文本越接近。

代码实现:

from nltk.translate.bleu_score import sentence_bleu

def calculate_bleu(reference, hypothesis):
    # reference是参考文本,是一个列表的列表
    # hypothesis是生成的文本,是一个列表
    reference = [reference.split()]
    hypothesis = hypothesis.split()
    # 计算BLEU
    bleu = sentence_bleu(reference, hypothesis)
    return bleu

# 测试
reference = "This is a test sentence for BLEU calculation"
hypothesis = "This is a test for BLEU"
bleu = calculate_bleu(reference, hypothesis)
print(f"BLEU: {bleu}")
1.3.3 ROUGE:衡量生成文本的召回率

ROUGE(Recall-Oriented Understudy for Gisting Evaluation)是用来评估文本摘要的指标,它和 BLEU 相反,它衡量的是召回率,也就是看参考文本中有多少 n-gram 出现在了生成的文本中,ROUGE 越高,说明生成的文本覆盖了参考文本的越多内容。

代码实现:

from rouge import Rouge

def calculate_rouge(reference, hypothesis):
    rouge = Rouge()
    scores = rouge.get_scores(hypothesis, reference)
    return scores[0]

# 测试
reference = "This is a test sentence for ROUGE calculation"
hypothesis = "This is a test for ROUGE"
rouge_scores = calculate_rouge(reference, hypothesis)
print(f"ROUGE-1: {rouge_scores['rouge-1']}")
print(f"ROUGE-2: {rouge_scores['rouge-2']}")
print(f"ROUGE-L: {rouge_scores['rouge-l']}")

第二章:主流 LLM 架构与模型详解,搞懂现在的大模型都是啥

2.1 LLM 的三类核心架构

根据 Transformer 的使用方式,LLM 可以分为三类核心架构,每一类都有自己的特点和适用场景:

2.1.1 自编码模型:Encoder-only,代表 BERT

自编码模型(Auto-Encoding Models)只使用了 Transformer 的 Encoder 部分,它的核心预训练任务是 Masked Language Model(MLM),也就是随机 mask 掉输入中的一些 token,然后让模型预测这些 mask 掉的 token。

这类模型的特点是:

  • 擅长理解类的任务,比如文本分类、命名实体识别、语义理解等

  • 双向的注意力机制,能够同时看到上下文的信息

  • 代表模型:BERT、RoBERTa、ALBERT、Electra 等

2.1.2 自回归模型:Decoder-only,代表 GPT

自回归模型(Auto-Regressive Models)只使用了 Transformer 的 Decoder 部分,它的核心预训练任务是 Autoregressive Language Model,也就是从左到右依次预测下一个 token,每个 token 只能看到它左边的 token。

这类模型的特点是:

  • 擅长生成类的任务,比如文本生成、对话、翻译等

  • 单向的注意力机制,只能看到左边的信息,适合生成任务

  • 能够很好的支持上下文学习,也就是 In-Context Learning

  • 代表模型:GPT 系列、LLaMA 系列、ChatGLM 系列等

2.1.3 序列到序列模型:Encoder+Decoder,代表 T5

序列到序列模型(Seq2Seq Models)同时使用了 Transformer 的 Encoder 和 Decoder 部分,它的核心思想是把输入编码成一个向量,然后 Decoder 从这个向量中解码出输出。

这类模型的特点是:

  • 擅长转换类的任务,比如翻译、摘要、问答等

  • Encoder 是双向的,Decoder 是单向的,兼顾了理解和生成

  • T5 提出了 Text-to-Text 的统一框架,把所有的 NLP 任务都转换成文本到文本的任务,比如把翻译任务转换成 “translate English to French: ...”,把摘要任务转换成 “summarize: ...”

  • 代表模型:T5、BART、Pegasus 等

2.2 GPT 系列模型的发展历程

GPT 系列是 Decoder-only 模型的代表,它的发展历程可以说是大模型发展的一个缩影:

2.2.1 GPT-1:开启预训练 + 微调的时代

GPT-1 在 2018 年发布,参数规模是 1.17 亿,它的核心思想是 “预训练 + 微调”,首先在海量的文本上进行语言模型的预训练,然后在下游任务上进行微调,这在当时取得了很好的效果,证明了预训练模型的能力。

2.2.2 GPT-2:Zero-shot 的萌芽

GPT-2 在 2019 年发布,参数规模是 15 亿,它提出了 Zero-shot 的思想,也就是不需要在下游任务上进行微调,直接让模型完成任务,比如给模型输入 “translate English to French: Hello, how are you?”,模型就能直接输出翻译的结果,这在当时是一个很大的突破,证明了大模型的泛化能力。

2.2.3 GPT-3:Few-shot 的突破

GPT-3 在 2020 年发布,参数规模是 1750 亿,这是当时最大的语言模型,它提出了 Few-shot 的思想,也就是只需要给模型几个例子,模型就能学会完成任务,不需要微调,这就是上下文学习的能力,GPT-3 证明了当模型参数足够大的时候,会涌现出很多小模型没有的能力。

2.2.4 ChatGPT:对话能力的突破

ChatGPT 在 2022 年发布,它是基于 GPT-3.5 的模型,通过 RLHF(基于人类反馈的强化学习)进行了微调,让模型能够更好的和人类进行对话,遵循人类的指令,这直接引爆了大模型的热潮,让大模型走进了普通人的视野。

2.3 主流开源大模型盘点

除了闭源的 GPT 系列,现在有很多优秀的开源大模型,它们在各个领域都有很好的表现,下面我们来盘点一下:

2.3.1 LLaMA 系列:Meta 的开源大模型

LLaMA 是 Meta 在 2023 年发布的开源大模型,它有不同的参数规模:7B、13B、33B、65B,后来又发布了 LLaMA-2,支持商用,LLaMA 的特点是性能很强,而且很小的参数就能达到很好的效果,比如 7B 的 LLaMA 就能超过很多更大的模型,现在基于 LLaMA 的衍生模型非常多,比如 Alpaca、Vicuna、Llama 2 等。

2.3.2 ChatGLM 系列:清华的中文大模型

ChatGLM 是清华和智谱 AI 发布的中文大模型,它基于 GLM 架构,专门针对中文进行了优化,ChatGLM-6B 是一个 6B 的模型,能够在消费级的显卡上运行,非常适合个人开发者,现在已经更新到了 ChatGLM3,支持多轮对话、工具调用等功能。

2.3.3 其他主流开源模型
  • Baichuan:百川智能发布的开源大模型,有 7B、13B 等参数,中文性能很强,支持商用。

  • Qwen(通义千问):阿里发布的开源大模型,有 7B、14B、72B 等参数,支持多模态,性能很强。

  • Yi:零一万物发布的开源大模型,有 6B、34B 等参数,支持中英双语,性能很强。

  • DeepSeek:深度求索发布的开源大模型,有 7B、67B 等参数,代码能力很强。

2.3.4 不同参数模型的硬件配置参考
模型参数 最低显存要求(FP16) 最低显存要求(INT4) 适用场景
7B 13GB 4GB 个人开发、小项目
13B 26GB 8GB 中小企业、中等项目
34B 68GB 20GB 企业、大项目
70B 140GB 40GB 大型企业、高性能场景

第三章:Prompt 工程:从理论到实战,让大模型听你的话

3.1 Prompt 工程的 5 大核心原则

Prompt 工程是和大模型交互的艺术,好的 Prompt 能够让大模型输出你想要的结果,差的 Prompt 可能会让大模型胡说八道,下面我们来讲解 Prompt 工程的 5 大核心原则:

3.1.1 清晰的指令:告诉模型你想要什么

首先,你要给模型清晰的指令,不要模糊,不要让模型猜你想要什么,具体来说,你可以:

  1. 详细描述你的需求:不要只说 “帮我写个文案”,要说 “帮我写一个针对年轻人的奶茶店的宣传文案,100 字左右,要活泼有趣”。

  2. 给模型设定角色:你可以让模型扮演一个角色,比如 “你是一个资深的 Python 工程师,帮我解释一下这个代码的作用”,这样模型就会用对应的角色的语气和知识来回答你。

  3. 使用分隔符:用分隔符把你的输入和指令分开,比如用 \\\`、--- 等,这样模型就能清楚的区分哪些是指令,哪些是输入的文本。

  4. 把任务拆分成步骤:如果任务很复杂,你可以把它拆分成多个步骤,比如 “第一步,先总结这篇文章的主要内容,第二步,提取文章中的关键观点,第三步,给这篇文章写一个 100 字的评论”。

  5. 给模型举例子:Few-shot 的方式,给模型几个例子,让模型知道你想要什么样的输出,比如:

例子1:
输入:这家店的奶茶真好喝
输出:好评
例子2:
输入:这家店的服务太差了
输出:差评
现在,输入:这家店的奶茶味道一般,但是服务很好
输出:
  1. 指定输出的长度:告诉模型你想要的输出长度,比如 “输出不要超过 100 字”,这样模型就不会输出太长的内容。
3.1.2 文本参考:降低模型的幻觉

大模型很容易产生幻觉,也就是编造一些不存在的信息,为了降低幻觉,你可以给模型提供参考的文本,让模型基于参考的文本回答问题,比如:

参考文本:
{这里放你的参考文本}
请你基于上面的参考文本,回答下面的问题,不要编造信息,如果参考文本里没有答案,就说“不知道”。
问题:{你的问题}
3.1.3 复杂任务拆分为子任务:让模型一步步来

如果任务很复杂,不要让模型一步完成,把它拆分成多个子任务,比如:

  • 意图识别:首先识别用户的意图,然后根据不同的意图做不同的处理。

  • 长文本处理:如果文本很长,先把文本拆分成多个段落,然后逐个处理,最后汇总结果。

  • 长对话处理:长对话的话,先总结之前的对话内容,然后再处理新的问题。

3.1.4 给模型思考时间:让模型慢慢想

大模型有时候会犯一些低级的错误,这是因为它想的太快了,你可以让模型一步步思考,也就是思维链(Chain of Thought,CoT),比如:

请你一步步思考,然后回答这个问题,不要直接给出答案。
问题:小明有10个苹果,他给了小红3个,然后又买了5个,请问小明现在有多少个苹果?

这样模型就会一步步的计算,而不是直接给出答案,这样就能提高准确率。

3.1.5 借助外部工具:让模型能力更强

大模型不是万能的,它的知识是过时的,而且不会做数学计算,这时候你可以让模型借助外部工具,比如:

  • 搜索引擎:让模型搜索最新的信息。

  • 代码执行:让模型写代码,然后执行代码来解决数学问题。

  • 数据库:让模型查询数据库里的信息。

3.2 实战:基于 Prompt 的金融行业文本处理

讲完了理论,我们来做一个实战项目,基于 Prompt 和 ChatGLM-6B,来完成金融行业的三个任务:文本分类、信息抽取、文本匹配,这个项目不需要微调模型,只需要设计 Prompt,就能在小样本的情况下完成任务,非常适合没有标注数据的场景。

3.2.1 项目背景

在金融领域,我们有很多的文本数据,比如新闻、公告、研报等,我们需要对这些文本进行处理,但是标注数据很贵,而且很难获取,这时候我们就可以用 Zero-shot 和 Few-shot 的方式,基于 Prompt 来完成任务,不需要标注数据,只需要设计好 Prompt 就行。

我们用的模型是 ChatGLM-6B,这是一个开源的中文大模型,能够在消费级的显卡上运行,非常适合个人开发者。

3.2.2 任务 1:金融文本分类

首先,我们来做金融文本分类,我们有四个类别:新闻报道、公司公告、财务公告、分析师报告,我们想要把输入的金融文本分类到这四个类别里。

Prompt 设计

我们用 Few-shot 的方式,给模型几个例子,让模型知道我们想要什么样的输出:

现在你是一个文本分类器,你需要按照要求将我给你的句子分类到:['新闻报道', '公司公告', '财务公告', '分析师报告']类别中。

例子1:
User: "今日,股市经历了一轮震荡,受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整,以适应市场的不确定性。"
Bot: 新闻报道

例子2:
User: "本公司年度财务报告显示,去年公司实现了稳步增长的盈利,同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。"
Bot: 财务报告

例子3:
User: "本公司高兴地宣布成功完成最新一轮并购交易,收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域,提高市场竞争力"
Bot: 公司公告

例子4:
User: "最新的行业分析报告指出,科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素,投资者应关注这些趋势"
Bot: 分析师报告
代码实现

完整的代码如下,你可以直接运行:

# —*-coding:utf-8-*-
"""
利用 LLM 进行文本分类任务。
"""
from rich import print
from rich.console import Console
from transformers import AutoTokenizer, AutoModel

# 提供所有类别以及每个类别下的样例
class_examples = {
    '新闻报道': '今日,股市经历了一轮震荡,受到宏观经济数据和全球贸易紧张局势的影响。投资者密切关注美联储可能的政策调整,以适应市场的不确定性。',
    '财务报告': '本公司年度财务报告显示,去年公司实现了稳步增长的盈利,同时资产负债表呈现强劲的状况。经济环境的稳定和管理层的有效战略执行为公司的健康发展奠定了基础。',
    '公司公告': '本公司高兴地宣布成功完成最新一轮并购交易,收购了一家在人工智能领域领先的公司。这一战略举措将有助于扩大我们的业务领域,提高市场竞争力',
    '分析师报告': '最新的行业分析报告指出,科技公司的创新将成为未来增长的主要推动力。云计算、人工智能和数字化转型被认为是引领行业发展的关键因素,投资者应关注这些趋势'}

def init_prompts():
    """
 初始化前置prompt,便于模型做 incontext learning。
 """
    class_list = list(class_examples.keys())
    pre_history = [
        (
            f'现在你是一个文本分类器,你需要按照要求将我给你的句子分类到:{class_list}类别中。',
            f'好的。'
        )
    ]

    for _type, exmpale in class_examples.items():
        pre_history.append((f'“{exmpale}”是 {class_list} 里的什么类别?', _type))

    return {'class_list': class_list, 'pre_history': pre_history}

def inference(
        sentences: list,
        custom_settings: dict,
        model,
        tokenizer,
        device
):
    """
 推理函数。

 Args:
 sentences (List[str]): 待推理的句子。
 custom_settings (dict): 初始设定,包含人为给定的 few-shot example。
 """
    for sentence in sentences:
        with console.status("[bold bright_green] Model Inference..."):
            sentence_with_prompt = f"“{sentence}”是 {custom_settings['class_list']} 里的什么类别?"
            response, history = model.chat(tokenizer, sentence_with_prompt, history=custom_settings['pre_history'])
        print(f'>>> [bold bright_red]sentence: {sentence}')
        print(f'>>> [bold bright_green]inference answer: {response}')

if __name__ == '__main__':
    console = Console()
    device = 'cuda:0' if torch.cuda.is_available() else 'cpu'
    tokenizer = AutoTokenizer.from_pretrained("THUDM/chatglm-6b-int4",trust_remote_code=True)
    model = AutoModel.from_pretrained("THUDM/chatglm-6b-int4",trust_remote_code=True).float()
    model.to(device)

    sentences = [
        "今日,央行发布公告宣布降低利率,以刺激经济增长。这一降息举措将影响贷款利率,并在未来几个季度内对金融市场产生影响。",
        "ABC公司今日发布公告称,已成功完成对XYZ公司股权的收购交易。本次交易是ABC公司在扩大业务范围、加强市场竞争力方面的重要举措。据悉,此次收购将进一步巩固ABC公司在行业中的地位,并为未来业务发展提供更广阔的发展空间。详情请见公司官方网站公告栏",
        "公司资产负债表显示,公司偿债能力强劲,现金流充足,为未来投资和扩张提供了坚实的财务基础。",
        "最新的分析报告指出,可再生能源行业预计将在未来几年经历持续增长,投资者应该关注这一领域的投资机会",
        ]

    custom_settings = init_prompts()
    print(custom_settings)

    inference(
        sentences,
        custom_settings,
        model,
        tokenizer,
        device
    )

运行之后,你就能得到分类的结果,是不是很简单?

3.2.3 任务 2:金融文本信息抽取

接下来,我们来做金融文本的信息抽取,我们想要从股票的新闻里抽取日期、股票名称、开盘价、收盘价、成交量这些信息。

Prompt 设计

同样,我们用 Few-shot 的方式,给模型一个例子,让模型知道我们想要什么样的输出,而且我们要求模型输出 JSON 格式,这样我们就能直接解析结果了:

现在你需要帮助我完成信息抽取任务,当我给你一个句子时,你需要帮我抽取出句子中实体信息,并按照JSON的格式输出,上述句子中没有的信息用['原文中未提及']来表示,多个值之间用','分隔。

例子:
User: '2023-01-10,股市震荡。股票古哥-D[EOOE]美股今日开盘价100美元,一度飙升至105美元,随后回落至98美元,最终以102美元收盘,成交量达到520000。'。提取上述句子中“金融”('日期', '股票名称', '开盘价', '收盘价', '成交量')类型的实体,并按照JSON格式输出,上述句子中没有的信息用['原文中未提及']来表示,多个值之间用','分隔。
Bot: {'日期': ['2023-01-10'],'股票名称': ['古哥-D[EOOE]美股'],'开盘价': ['100美元'],  '收盘价': ['102美元'],'成交量': ['520000']}
代码实现

完整的代码如下:

import re
import json

from rich import print
from transformers import AutoTokenizer, AutoModel

# 定义不同实体下的具备属性
schema = {
    '金融': ['日期', '股票名称', '开盘价', '收盘价', '成交量'],
}

IE_PATTERN = "{}\n\n提取上述句子中{}的实体,并按照JSON格式输出,上述句子中不存在的信息用['原文中未提及']来表示,多个值之间用','分隔。"

# 提供一些例子供模型参考
ie_examples = {
        '金融': [
                    {
                        'content': '2023-01-10,股市震荡。股票古哥-D[EOOE]美股今日开盘价100美元,一度飙升至105美元,随后回落至98美元,最终以102美元收盘,成交量达到520000。',
                        'answers': {
                                        '日期': ['2023-01-10'],
                                        '股票名称': ['古哥-D[EOOE]美股'],
                                        '开盘价': ['100美元'],
                                        '收盘价': ['102美元'],
                                        '成交量': ['520000'],
                            }
                    }
        ]
}

def init_prompts():
    """
 初始化前置prompt,便于模型做 incontext learning。
 """
    ie_pre_history = [
        (
            "现在你需要帮助我完成信息抽取任务,当我给你一个句子时,你需要帮我抽取出句子中实体信息,并按照JSON的格式输出,上述句子中没有的信息用['原文中未提及']来表示,多个值之间用','分隔。",
            '好的,请输入您的句子。'
        )
    ]

    for _type, example_list in ie_examples.items():
        print(f'信息抽取样本的原始句子是--》{example_list}')

        for example in example_list:
            sentence = example['content']
            properties_str = ', '.join(schema[_type])
            schema_str_list = f'“{_type}”({properties_str})'

            sentence_with_prompt = IE_PATTERN.format(sentence, schema_str_list)

            ie_pre_history.append((
                f'{sentence_with_prompt}',
                f"{json.dumps(example['answers'], ensure_ascii=False)}"
            ))
            print(f'ie_pre_history-->{ie_pre_history}')

    return {'ie_pre_history': ie_pre_history}

def clean_response(response: str):
    """
 后处理模型输出。

 Args:
 response (str): _description_
 """
    if '```json' in response:
        res = re.findall(r'```json(.*?)```', response)
        if len(res) and res[0]:
            response = res[0]
        response.replace('、', ',')
    try:
        return json.loads(response)
    except:
        return response

def inference(
        sentences: list,
        custom_settings: dict,
        model,
        tokenizer,
        device
    ):
    """
 推理函数。

 Args:
 sentences (List[str]): 待抽取的句子。
 custom_settings (dict): 初始设定,包含人为给定的 few-shot example。
 """
    for sentence in sentences:
        cls_res = "金融"
        if cls_res not in schema:
            print(f'The type model inferenced {cls_res} which is not in schema dict, exited.')
            exit()
        properties_str = ', '.join(schema[cls_res])
        schema_str_list = f'“{cls_res}”({properties_str})'
        sentence_with_ie_prompt = IE_PATTERN.format(sentence, schema_str_list)
        ie_res, _ = model.chat(tokenizer, sentence_with_ie_prompt, history=custom_settings['ie_pre_history'])
        ie_res = clean_response(ie_res)
        print(f'>>> [bold bright_red]sentence: {sentence}')
        print(f'>>> [bold bright_green]inference answer: ')
        print(ie_res)

if __name__ == '__main__':
        #device = 'cuda:0'
    device = 'cpu'
    tokenizer = AutoTokenizer.from_pretrained("./ChatGLM-6B/THUDM/chatglm-6b",
                                              trust_remote_code=True)
    model = AutoModel.from_pretrained("./ChatGLM-6B/THUDM/chatglm-6b",
                                      trust_remote_code=True).float()
    model.to(device)

    sentences = [
        '2023-02-15,寓意吉祥的节日,股票佰笃[BD]美股开盘价10美元,虽然经历了波动,但最终以13美元收盘,成交量微幅增加至460,000,投资者情绪较为平稳。',
        '2023-04-05,市场迎来轻松氛围,股票盘古(0021)开盘价23元,尽管经历了波动,但最终以26美元收盘,成交量缩小至310,000,投资者保持观望态度。',
    ]

    custom_settings = init_prompts()

    inference(
        sentences,
        custom_settings,
        model,
        tokenizer,
        device
    )

运行之后,你就能得到抽取的结果,而且是 JSON 格式的,你可以直接用在你的项目里。

3.2.4 任务 3:金融文本匹配

最后,我们来做金融文本的匹配,也就是判断两个文本是不是同一个意思,这个任务我们用 Few-shot 的方式,给模型几个例子,让模型学会判断。

这个任务的代码和前面的很类似,核心还是 Prompt 的设计,给模型几个匹配和不匹配的例子,模型就能学会判断了。


第四章:大模型微调:参数高效的进阶之路,让大模型适配你的任务

当 Prompt 工程无法满足你的需求的时候,你就需要微调大模型了,但是大模型的参数太大了,全量微调的话,成本太高了,比如 GPT-3 有 1750 亿参数,全量微调的话,需要几百 GB 的显存,普通人根本玩不起,这时候参数高效微调(Parameter-Efficient Fine-Tuning,PEFT)就出现了,它只需要微调少量的参数,就能达到和全量微调差不多的效果,成本很低,普通人也能玩得起。

4.1 NLP 的四大范式

在讲微调之前,我们先回顾一下 NLP 的四大范式,这能帮你更好的理解微调的发展:

  1. 第一范式:传统机器学习:比如 TF-IDF + 朴素贝叶斯,这是最早的范式,需要大量的特征工程,效果也一般。

  2. 第二范式:深度学习:比如 Word2Vec+LSTM,不需要太多的特征工程,效果比传统机器学习好很多。

  3. 第三范式:预训练 + 微调:比如 BERT+Fine-tuning,预训练模型在海量文本上预训练,然后在下游任务上微调,效果很好,但是需要微调所有的参数,而且需要大量的标注数据。

  4. 第四范式:预训练 + Prompt + 预测:比如 Prompt-Tuning,不需要微调所有的参数,只需要微调少量的 Prompt 参数,就能达到很好的效果,而且只需要很少的标注数据。

4.2 传统微调的痛点

传统的全量微调有很多痛点:

  1. 成本太高:大模型的参数太大了,全量微调需要很大的显存,比如 7B 的模型,全量微调需要 100 多 GB 的显存,普通人根本玩不起。

  2. 过拟合:当标注数据很少的时候,全量微调很容易过拟合,因为模型的参数太多了,少量的数据根本不够训练。

  3. 存储成本高:每个下游任务都需要存储一个完整的模型副本,比如你有 100 个下游任务,你就需要存储 100 个完整的模型,这存储成本太高了。

  4. 语义偏差:预训练的目标和下游任务的目标不一样,导致预训练和微调之间有 Gap,影响效果。

4.3 Prompt-Tuning:让下游任务适配预训练模型

为了解决传统微调的痛点,Prompt-Tuning 出现了,它的核心思想是:不是让预训练模型去适配下游任务,而是让下游任务去适配预训练模型

简单来说,传统的微调是,预训练模型已经学好了语言的知识,然后我们把它改一改,让它适配我们的下游任务,而 Prompt-Tuning 是,预训练模型已经学好了语言的知识,我们把我们的下游任务改一改,改成预训练模型熟悉的任务,比如 MLM 任务,这样我们就不需要改预训练模型的参数了,只需要改一改我们的输入,就能让预训练模型完成我们的任务。

举个例子,比如情感分类任务,传统的微调是,我们给 BERT 加一个分类头,然后微调 BERT 的所有参数,让它输出分类的结果,而 Prompt-Tuning 是,我们把输入的文本改成:

{文本}。这是一个[MASK]评。

然后让 BERT 预测 [MASK] 这个位置的 token,如果预测的是 “好”,就是好评,如果是 “差”,就是差评,这样就把情感分类任务,转换成了 BERT 熟悉的 MLM 任务,这样我们就不需要改 BERT 的参数了,只需要设计这个 Prompt 就行。

4.3.1 PET:硬模板的 Prompt-Tuning

PET(Pattern-Exploiting Training)是最早的 Prompt-Tuning 方法之一,它用的是人工设计的硬模板,也就是我们上面说的,人工设计一个模板,把下游任务转换成 MLM 任务。

PET 的核心是两个组件:

  1. Pattern(模板):人工设计的模板,用来把输入的文本转换成 MLM 的输入,比如\{文本\}。这是一个\[MASK\]评。

  2. Verbalizer(标签词映射):把分类的标签,映射成对应的词,比如好评映射成 “好”,差评映射成 “差”,这样模型预测 [MASK] 的词,我们就能得到分类的结果。

PET 的优点是,不需要引入新的参数,只需要人工设计模板,就能在小样本的情况下达到很好的效果,但是它的缺点是,人工设计的模板不稳定,不同的模板效果差很多,而且模板不能优化,是固定的。

4.3.2 P-Tuning:软模板的 Prompt-Tuning

为了解决 PET 的缺点,P-Tuning(Pattern-Tuning)出现了,它用的是可学习的软模板,也就是不再人工设计模板了,而是用可学习的向量作为模板,这些向量是可以训练的,模型会自己学习到最好的模板。

简单来说,P-Tuning 在输入的文本前面,加了几个可学习的伪 token,这些伪 token 不是真实的词,而是可学习的向量,训练的时候,我们只训练这些伪 token 的参数,预训练模型的参数是固定的,这样就只需要微调少量的参数,就能达到很好的效果,而且这些伪 token 是可以全局优化的,解决了人工模板的不稳定的问题。

4.4 PEFT:参数高效微调的三大方法

除了 Prompt-Tuning,还有很多其他的参数高效微调的方法,现在主流的有三种:

4.4.1 Prefix-Tuning:给输入加前缀

Prefix-Tuning 的核心思想是,在输入的前面,加一些可学习的前缀 token,这些前缀 token 是可学习的向量,训练的时候,我们只训练这些前缀的参数,预训练模型的参数是固定的。

Prefix-Tuning 的优点是,它可以用在生成任务上,比如摘要、对话等,而且效果很好,但是它的缺点是,前缀会占用输入的长度,导致我们能处理的文本长度变短了。

4.4.2 Adapter-Tuning:给模型加适配器

Adapter-Tuning 的核心思想是,在预训练模型的每一层,加一些小的适配器层,这些适配器层是 bottleneck 结构的,参数很少,训练的时候,我们只训练这些适配器层的参数,预训练模型的参数是固定的。

Adapter-Tuning 的优点是,它不会占用输入的长度,而且每个下游任务只需要存储这些适配器的参数,很小,但是它的缺点是,推理的时候会增加延迟,因为要过这些适配器层。

4.4.3 LoRA:低秩适应,现在最火的 PEFT 方法

LoRA(Low-Rank Adaptation)是现在最火的 PEFT 方法,它的核心思想是,模型权重的更新,其实是一个低秩的矩阵,所以我们不需要更新整个权重矩阵,我们只需要用两个小的矩阵,来近似这个权重的更新。

简单来说,原来的权重矩阵是 W,我们在训练的时候,不是更新 W,而是给 W 加一个 ΔW,而 ΔW 我们用两个小的矩阵 A 和 B 来近似,ΔW = A * B,A 和 B 的参数很少,比如原来的 W 是 dd 的,A 是 dr 的,B 是 rd 的,r 是一个很小的数,比如 8、16,这样 A 和 B 的参数就只有 2dr,比原来的 dd 小很多。

训练的时候,我们只训练 A 和 B 的参数,W 是固定的,推理的时候,我们把 A 和 B 加到 W 上,这样就和原来的模型一样了,不会增加任何的推理延迟!

LoRA 的优点太多了:

  1. 参数很少:比如 7B 的模型,LoRA 只需要训练几百万的参数,比全量微调少了三个数量级。

  2. 没有推理延迟:训练完之后,把 A 和 B 合并到 W 上,推理的时候和原来的模型一模一样,没有任何延迟。

  3. 切换任务很方便:不同的下游任务,只需要不同的 A 和 B,切换的时候,只需要把 A 和 B 换了就行,不需要换整个模型。

  4. 效果很好:很多实验证明,LoRA 的效果和全量微调差不多,甚至更好。

现在 LoRA 已经成为了最主流的 PEFT 方法,HuggingFace 的 PEFT 库也支持 LoRA,你只需要几行代码,就能用 LoRA 微调大模型了。


第五章:实战项目 1:基于 GPT2 搭建医疗问诊机器人,手把手教你做对话模型

讲完了理论,我们来做第一个实战项目,基于 GPT2 搭建一个医疗问诊的机器人,这个项目我们会从数据处理,到模型搭建,到训练,到交互,一步步带你完成,让你真正学会怎么训练一个对话模型。

5.1 项目背景

现在智能医疗是一个很火的方向,我们想要做一个医疗问诊的机器人,能够回答患者的问题,给患者提供医疗的建议,这个项目我们用 GPT2 来训练,因为 GPT2 是一个生成模型,很适合做对话任务。

5.2 数据处理

首先,我们需要处理数据,我们用的数据集是儿科疾病问诊的数据集,里面有 10 万多条问诊的数据,每一条都有患者的问题和医生的回答。

5.2.1 数据格式转换

首先,我们把 Excel 的数据转换成对话的文本格式,每一个对话,就是患者的问题,然后是医生的回答,中间用换行分开,不同的对话之间用空行分开:

import pandas as pd 
from tqdm import tqdm 

def read_csv2txt():
    data = pd.read_excel('./data/儿科疾病问诊信息.xlsx')
    print(data.head())
    data_list = data.values.tolist()
    for data in tqdm(data_list):
        try:
            question = data[2]
            answer = data[3]
            str1 = question + '\n' + answer
            with open('./data/train.txt', 'a')as f:
                f.write(str1 + '\n\n')
        except:
            continue 

read_csv2txt()
5.2.2 数据 Tokenize

接下来,我们把文本转换成 token,因为 GPT2 需要输入的是 token 的 id,我们用 BERT 的 tokenizer 来处理,因为我们用的是中文的 GPT2:

# 导入分词器 
from transformers import BertTokenizerFast 
# 将数据保存为pkl文件,方便下次读取 
import pickle
# 读取数据的进度条展示 
from tqdm import tqdm 

def preprocess(train_txt_path, train_pkl_path):
    """
 对原始语料进行tokenize,将每段对话处理成如下形式:"[CLS]utterance1[SEP]utterance2[SEP]"
 """    
    '''初始化tokenizer,使用BertTokenizerFast.  从预训练的中文Bert模型(bert-base-chinese)创建一个tokenizer对象'''
    tokenizer = BertTokenizerFast.from_pretrained('bert-base-chinese',
                                                  sep_token="[SEP]",
                                                  pad_token="[PAD]",
                                                  cls_token="[CLS]")

    sep_id = tokenizer.sep_token_id  # 获取分隔符[SEP]的token ID
    cls_id = tokenizer.cls_token_id  # 获取起始符[CLS]的token ID

    # 读取训练数据集
    with open(train_txt_path, 'rb') as f:
        data = f.read().decode("utf-8")  # 以UTF-8编码读取文件内容

    # 根据换行符区分不同的对话段落,需要区分Windows和Linux环境下的换行符
    if "\r\n" in data:
        train_data = data.split("\r\n\r\n")
    else:
        train_data = data.split("\n\n")

    print(len(train_data))  # 打印对话段落数量
    # 开始进行tokenize
    # 保存所有的对话数据,每条数据的格式为:"[CLS]seq1[SEP]seq2[SEP]"
    dialogue_len = []  # 记录所有对话tokenize之后的长度,用于统计中位数与均值
    dialogue_list = []  # 记录所有对话

    for index, dialogue in enumerate(tqdm(train_data)):
        if "\r\n" in data:
            sequences = dialogue.split("\r\n")
        else:
            sequences = dialogue.split("\n")

        input_ids = [cls_id]  # 每个dialogue以[CLS]开头
        for sequence in sequences:
            # 将每个对话句子进行tokenize,并将结果拼接到input_ids列表中
            input_ids += tokenizer.encode(sequence, add_special_tokens=False)  
            input_ids.append(sep_id)  # 每个seq之后添加[SEP],表示seqs会话结束

        dialogue_len.append(len(input_ids))  # 将对话的tokenize后的长度添加到对话长度列表中
        dialogue_list.append(input_ids)  # 将tokenize后的对话添加到对话列表中

    print(f'dialogue_len--->{dialogue_len}')  # 打印对话长度列表
    print(f'dialogue_list--->{dialogue_list}')  # 打印

    # 保存pkl文件数据
    with open(train_pkl_path, "wb") as f:
        pickle.dump(dialogue_list, f)
5.2.3 数据封装

接下来,我们把数据封装成 PyTorch 的 Dataset 和 DataLoader,这样我们就能用 PyTorch 来训练了:

from torch.utils.data import Dataset  # 导入Dataset模块,用于定义自定义数据集 
import torch  # 导入torch模块,用于处理张量和构建神经网络 

class MyDataset(Dataset):
    """
 自定义数据集类,继承自Dataset类 
 """

    def __init__(self, input_list, max_len):
        """
 初始化函数,用于设置数据集的属性
 :param input_list: 输入列表,包含所有对话的tokenize后的输入序列 
 :param max_len: 最大序列长度,用于对输入进行截断或填充 
 """
        self.input_list = input_list  # 将输入列表赋值给数据集的input_list属性
        self.max_len = max_len  # 将最大序列长度赋值给数据集的max_len属性

    def __len__(self):         
        """
 获取数据集的长度
 :return: 数据集的长度 
 """
        return len(self.input_list)  # 返回数据集的长度

    def __getitem__(self, index):         
        """
 根据给定索引获取数据集中的一个样本
 :param index: 样本的索引
 :return: 样本的输入序列张量 
 """
        input_ids = self.input_list[index]  # 获取给定索引处的输入序列
        input_ids = input_ids[:self.max_len]  # 根据最大序列长度对输入进行截断或填充
        input_ids = torch.tensor(input_ids, dtype=torch.long)  # 将输入序列转换为long类型
        return input_ids  # 返回样本的输入序列张量

然后是 DataLoader:

# 导入rnn_utils模块,用于处理可变长度序列的填充和排序 
import torch.nn.utils.rnn as rnn_utils  

# 导入Dataset和DataLoader模块,用于加载和处理数据集 
from torch.utils.data import Dataset, DataLoader 

import pickle  # 导入pickle模块,用于序列化和反序列化Python对象 
from dataset import *  # 导入自定义的数据集类 

def load_dataset(train_path):
    """
 加载训练集和验证集
 :param train_path: 训练数据集路径
 :return: 训练数据集和验证数据集 
 """
    with open(train_path, "rb") as f:
        input_list = pickle.load(f)  # 从文件中加载输入列表

    # 划分训练集与验证集
    print(len(input_list))  # 打印输入列表的长度
    input_list_train = input_list[200:]  # 将输入列表划分为训练集部分
    input_list_val = input_list[:200]  # 将输入列表划分为验证集部分

    train_dataset = MyDataset(input_list_train, 200)  # 创建训练数据集对象
    val_dataset = MyDataset(input_list_val, 200)  # 创建验证数据集对象
    return train_dataset, val_dataset 

def collate_fn(batch):
    """
 自定义的collate_fn函数,用于将数据集中的样本进行批处理
 :param batch: 样本列表
 :return: 经过填充的输入序列张量和标签序列张量 
 """
     # 对输入序列进行填充,使其长度一致
    input_ids = rnn_utils.pad_sequence(batch, batch_first=True, padding_value=0) 

    # 对标签序列进行填充,使其长度一致
    labels = rnn_utils.pad_sequence(batch, batch_first=True, padding_value=-100) 

    return input_ids, labels 

def get_dataloader(train_path):
    """
 获取训练数据集和验证数据集的DataLoader对象
 :param train_path: 训练数据集路径
 :return: 训练数据集的DataLoader对象和验证数据集的DataLoader对象 
 """
    # 加载训练数据集和验证数据集
    train_dataset, val_dataset = load_dataset(train_path) 

    # 创建训练数据集的DataLoader对象
    train_dataloader = DataLoader(train_dataset,
                                  batch_size=4,
                                  shuffle=True,
                                  collate_fn=collate_fn,
                                  drop_last=True)  
    # 创建验证数据集的DataLoader对象
    validate_dataloader = DataLoader(val_dataset,
                                     batch_size=4,
                                     shuffle=True,
                                     collate_fn=collate_fn,
                                     drop_last=True)  
    return train_dataloader, validate_dataloader

5.3 模型搭建

接下来,我们搭建模型,我们用的是 GPT2 的预训练模型,HuggingFace 已经帮我们实现好了,我们只需要加载它就行:

from transformers import GPT2LMHeadModel, GPT2Config
# 创建模型 
if params.pretrained_model:  
    # 加载预训练模型
    model = GPT2LMHeadModel.from_pretrained(params.pretrained_model) 
else:  
    # 初始化模型
    model_config = GPT2Config.from_json_file(params.config_json)
    model = GPT2LMHeadModel(config=model_config)

我们的 GPT2 模型的参数是:

  • n_embd: 768,隐藏层的维度

  • n_head: 12,注意力头的数量

  • n_layer: 12,Transformer 层的数量

  • n_positions: 1024,最大的序列长度

  • vocab_size: 21128,词表的大小

5.4 模型训练和交互

接下来,我们训练模型,训练的代码很简单,就是标准的 PyTorch 训练代码:

import torch
import os
from datetime import datetime
import transformers 
from transformers import GPT2LMHeadModel, GPT2Config 
from transformers import BertTokenizerFast 
from functions_tools import *
from parameter_config import *
from data_preprocess.dataloader import *
from pytorch_tools import EarlyStopping 

def train_epoch(model,
                train_dataloader,
                optimizer, scheduler,
                epoch, args):
    model.train()
    device = args.device
    # 对于ignore_index的label token不计算梯度
    ignore_index = args.ignore_index
    epoch_start_time = datetime.now()
    total_loss = 0  # 记录下整个epoch的loss的总和

    # epoch_correct_num:每个epoch中,output预测正确的word的数量
    # epoch_total_num: 每个epoch中,output预测的word的总数量
    epoch_correct_num, epoch_total_num = 0, 0

    for batch_idx, (input_ids, labels) in enumerate(train_dataloader):
        input_ids = input_ids.to(device)
        labels = labels.to(device)
        outputs = model.forward(input_ids, labels=labels)

        logits = outputs.logits
        loss = outputs.loss
        loss = loss.mean()

        # 统计该batch的预测token的正确数与总数
        batch_correct_num, batch_total_num = calculate_acc(logits, labels, ignore_index=ignore_index)

        # 统计该epoch的预测token的正确数与总数
        epoch_correct_num += batch_correct_num
        epoch_total_num += batch_total_num
        # 计算该batch的accuracy
        batch_acc = batch_correct_num / batch_total_num

        total_loss += loss.item()
        if args.gradient_accumulation_steps > 1:
            loss = loss / args.gradient_accumulation_steps

        loss.backward()
        # 梯度裁剪 # 避免梯度爆炸的方式。
        torch.nn.utils.clip_grad_norm_(model.parameters(), args.max_grad_norm)

        # 进行一定step的梯度累计之后,更新参数
        if (batch_idx + 1) % args.gradient_accumulation_steps == 0:
            # 更新参数
            optimizer.step()
            # 更新学习率
            scheduler.step()
            # 清空梯度信息
            optimizer.zero_grad()

        if (batch_idx + 1) % args.loss_step == 0:
            print(
                "batch {} of epoch {}, loss {}, batch_acc {},",
                batch_idx, epoch, loss.item(), batch_acc)

    # 整个epoch的平均loss
    avg_loss = total_loss / len(train_dataloader)
    # 整个epoch的准确率
    avg_acc = epoch_correct_num / epoch_total_num

    print('epoch: {}, avg_loss: {}, avg_acc: {}'.format(epoch, avg_loss, avg_acc))

训练完之后,我们就可以和模型交互了,输入患者的问题,模型就会输出医生的回答,一个医疗问诊机器人就做好了!


第六章:实战项目 2:小样本电商评论文本分类,用 Prompt-Tuning 解决小样本问题

接下来,我们来做第二个实战项目,小样本的电商评论文本分类,我们只有 63 条标注数据,怎么训练一个好的分类模型?这时候我们就可以用 Prompt-Tuning,用很少的标注数据,就能达到很好的效果。

6.1 项目背景

在电商领域,我们有很多用户的评论,我们想要把这些评论分类到不同的类别,比如水果、衣服、书籍等,但是标注数据很贵,我们只有 63 条标注数据,590 条验证数据,这么少的数据,传统的微调根本没法训练,很容易过拟合,这时候我们就可以用 Prompt-Tuning,用很少的标注数据,就能训练一个好的分类模型。

我们会做两个版本:一个是基于 BERT+PET 的硬模板版本,一个是基于 BERT+P-Tuning 的软模板版本,你可以对比一下它们的效果。

6.2 基于 BERT+PET 的硬模板方案

首先,我们来做 PET 的版本,PET 用的是人工设计的硬模板。

6.2.1 数据准备

我们的数据集很简单,训练集只有 63 条,每一行是标签和评论:

水果 脆脆的,甜味可以,可能时间有点长了,水分不是很足。
平板 华为机器肯定不错,但第一次碰上京东最糟糕的服务,以后不想到京东购物了。
书籍 为什么不认真的检查一下, 发这么一本脏脏的书给顾客呢!

验证集有 590 条,用来测试我们的模型。

我们的模板是人工设计的:这是一条\{MASK\}评论:\{textA\}。,也就是把评论放进去,然后让模型预测 [MASK] 的位置的词,这个词就是我们的标签。

我们的 Verbalizer 是标签到词的映射,因为我们的标签就是类别,所以我们直接用标签本身就行,比如水果的标签词就是 “水果”,衣服的就是 “衣服”。

6.2.2 模板处理

首先,我们处理模板,把输入的文本和模板拼起来,转换成模型的输入:

class HardTemplate(object):
    """
 硬模板,人工定义句子和[MASK]之间的位置关系。
 """

    def __init__(self, prompt: str):
        """
 Args:
 prompt (str): prompt格式定义字符串, e.g. -> "这是一条{MASK}评论:{textA}。"
 """
        self.prompt = prompt
        self.inputs_list = []                       # 根据文字prompt拆分为各part的列表
        self.custom_tokens = set(['MASK'])          # 从prompt中解析出的自定义token集合
        self.prompt_analysis()                         # 解析prompt模板

    def prompt_analysis(self):
        """
 将prompt文字模板拆解为可映射的数据结构。

 Examples:
 prompt -> "这是一条{MASK}评论:{textA}。"
 inputs_list -> ['这', '是', '一', '条', 'MASK', '评', '论', ':', 'textA', '。']
 custom_tokens -> {'textA', 'MASK'}
 """
        idx = 0
        while idx < len(self.prompt):
            str_part = ''
            if self.prompt[idx] not in ['{', '}']:
                self.inputs_list.append(self.prompt[idx])
            if self.prompt[idx] == '{':                  # 进入自定义字段
                idx += 1
                while self.prompt[idx] != '}':
                    str_part += self.prompt[idx]
                    idx += 1
            elif self.prompt[idx] == '}':
                raise ValueError("Unmatched bracket '}', check your prompt.")
            if str_part:
                self.inputs_list.append(str_part)
                # 将所有自定义字段存储,后续会检测输入信息是否完整
                self.custom_tokens.add(str_part)  
            idx += 1

    def __call__(self,
                 inputs_dict: dict,
                 tokenizer,
                 mask_length,
                 max_seq_len=512):
        """
 输入一个样本,转换为符合模板的格式。

 Args:
 inputs_dict (dict): prompt中的参数字典, e.g. -> {
 "textA": "这个手机也太卡了", 
 "MASK": "[MASK]"
 }
 tokenizer: 用于encoding文本
 mask_length (int): MASK token 的长度

 Returns:
 dict -> {
 'text': '[CLS]这是一条[MASK]评论:这个手机也太卡了。[SEP]',
 'input_ids': [1, 47, 10, 7, 304, 3, 480, 279, 74, 47, 27, 247, 98, 105, 512, 777, 15, 12043, 2],
 'token_type_ids': [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], 
 'attention_mask': [1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1],
 'mask_position': [0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
 }
 """
        # 定义输出格式
        outputs = {
            'text': '', 
            'input_ids': [],
            'token_type_ids': [],
            'attention_mask': [],
            'mask_position': []
        }

        str_formated = ''
        for value in self.inputs_list:
            if value in self.custom_tokens:
                if value == 'MASK':
                    str_formated += inputs_dict[value] * mask_length
                else:
                    str_formated += inputs_dict[value]
            else:
                str_formated += value
        # print(f'str_formated-->{str_formated}')
        encoded = tokenizer(text=str_formated,
                            truncation=True,
                            max_length=max_seq_len,
                            padding='max_length')
        # print(f'encoded--->{encoded}')
        outputs['input_ids'] = encoded['input_ids']
        outputs['token_type_ids'] = encoded['token_type_ids']
        outputs['attention_mask'] = encoded['attention_mask']
        token_list = tokenizer.convert_ids_to_tokens(encoded['input_ids'])
        outputs['text'] = ''.join(token_list)
        mask_token_id = tokenizer.convert_tokens_to_ids(['[MASK]'])[0]
        condition = np.array(outputs['input_ids']) == mask_token_id
        mask_position = np.where(condition)[0].tolist()
        outputs['mask_position'] = mask_position
        return outputs
6.2.3 数据预处理

接下来,我们把样本转换成模型的输入:

def convert_example(
        examples: dict,
        tokenizer,
        max_seq_len: int,
        max_label_len: int,
        hard_template: HardTemplate,
        train_mode=True,
        return_tensor=False) -> dict:
    """
 将样本数据转换为模型接收的输入数据。

 Args:
 examples (dict): 训练数据样本, e.g. -> {
 "text": [
 '手机 这个手机也太卡了。',
 '体育 世界杯为何迟迟不见宣传',
 ...
 ]
 }
 max_seq_len (int): 句子的最大长度,若没有达到最大长度,则padding为最大长度
 max_label_len (int): 最大label长度,若没有达到最大长度,则padding为最大长度
 hard_template: 模板类。
 train_mode (bool): 训练阶段 or 推理阶段。
 return_tensor (bool): 是否返回tensor类型,如不是,则返回numpy类型。

 Returns:
 dict (str: np.array) -> tokenized_output = {
 'input_ids': [[1, 47, 10, 7, 304, 3, 3, 3, 3, 47, 27, 247, 98, 105, 512, 777, 15, 12043, 2], ...],
 'token_type_ids': [[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0], ...],
 'attention_mask': [[1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1, 1], ...],
 'mask_positions': [[5, 6, 7, 8], ...],
 'mask_labels': [[2372, 3442, 0, 0], 
 [2643, 4434, 2334, 0], ...]
 }
 """
    tokenized_output = {
        'input_ids': [],
        'token_type_ids': [],
        'attention_mask': [],
        'mask_positions': [],
        'mask_labels': []
    }

    for i, example in enumerate(examples['text']):
        if train_mode:
            label, content = example.strip().split('\t')
        else:
            content = example.strip()

        inputs_dict = {
            'textA': content,
            'MASK': '[MASK]'
        }
        encoded_inputs = hard_template(
            inputs_dict=inputs_dict,
            tokenizer=tokenizer,
            max_seq_len=max_seq_len,
            mask_length=max_label_len)
        tokenized_output['input_ids'].append(encoded_inputs["input_ids"])
        tokenized_output['token_type_ids'].append(encoded_inputs["token_type_ids"])
        tokenized_output['attention_mask'].append(encoded_inputs["attention_mask"])
        tokenized_output['mask_positions'].append(encoded_inputs["mask_position"])

        if train_mode:
            label_encoded = tokenizer(text=[label])  # 将label补到最大长度
            # print(f'label_encoded-->{label}')
            label_encoded = label_encoded['input_ids'][1:-1]
            label_encoded = label_encoded[:max_label_len]
            add_pad = [tokenizer.pad_token_id] * (max_label_len - len(label_encoded))
            label_encoded = label_encoded + add_pad
            tokenized_output['mask_labels'].append(label_encoded)

    for k, v in tokenized_output.items():
        if return_tensor:
            tokenized_output[k] = torch.LongTensor(v)
        else:
            tokenized_output[k] = np.array(v)

    return tokenized_output
6.2.4 工具函数

然后,我们需要一些工具函数,比如 Verbalizer,用来处理标签的映射,还有损失函数,评估函数等。

Verbalizer 的代码:

class Verbalizer(object):
    """
 Verbalizer类,用于将一个Label对应到其子Label的映射。
 """

    def __init__(self, verbalizer_file: str, tokenizer, max_label_len: int):
        """
 Args:
 verbalizer_file (str): verbalizer文件存放地址。
 tokenizer: 分词器,用于文本和id之间的转换。
 max_label_len: 标签长度,若大于则截断,若小于则补齐
 """
        self.tokenizer = tokenizer
        self.label_dict = self.load_label_dict(verbalizer_file)
        self.max_label_len = max_label_len

    def load_label_dict(self, verbalizer_file: str):
        """
 读取本地文件,构建verbalizer字典。
 Args:
 verbalizer_file (str): verbalizer文件存放地址。
 Returns:
 dict -> {
 '体育': ['篮球', '足球','网球', '排球', ...],
 '酒店': ['宾馆', '旅馆', '旅店', '酒店', ...],
 ...
 }
 """
        label_dict = {}
        with open(verbalizer_file, 'r', encoding='utf8') as f:
            for line in f.readlines():
                label, sub_labels = line.strip().split('\t')
                label_dict[label] = list(set(sub_labels.split(',')))
        return label_dict

    def find_sub_labels(self, label: Union[list, str]):
        """
 通过标签找到对应所有的子标签。
 Args:
 label (Union[list, str]): 标签, 文本型 或 id_list, e.g. -> '体育' or [860, 5509]

 Returns:
 dict -> {
 'sub_labels': ['足球', '网球'], 
 'token_ids': [[6639, 4413], [5381, 4413]]
 }
 """
        if type(label) == list:    # 如果传入为id_list, 则通过tokenizer进行文本转换
            while self.tokenizer.pad_token_id in label:
                label.remove(self.tokenizer.pad_token_id)
            label = ''.join(self.tokenizer.convert_ids_to_tokens(label))
        # print(f'label-->{label}')
        if label not in self.label_dict:
            raise ValueError(f'Lable Error: "{label}" not in label_dict')

        sub_labels = self.label_dict[label]
        ret = {'sub_labels': sub_labels}
        token_ids = [_id[1:-1] for _id in self.tokenizer(sub_labels)['input_ids']]
        # print(f'token_ids-->{token_ids}')
        for i in range(len(token_ids)):
            token_ids[i] = token_ids[i][:self.max_label_len]  # 对标签进行截断与补齐
            if len(token_ids[i]) < self.max_label_len:
                token_ids[i] = token_ids[i] + [self.tokenizer.pad_token_id] * (self.max_label_len - len(token_ids[i]))
        ret['token_ids'] = token_ids
        return ret

    def batch_find_sub_labels(self, label: List[Union[list, str]]):
        """
 批量找到子标签。

 Args:
 label (List[list, str]): 标签列表, [[4510, 5554], [860, 5509]] or ['体育', '电脑']

 Returns:
 list -> [
 {
 'sub_labels': ['足球', '网球'], 
 'token_ids': [[6639, 4413], [5381, 4413]]
 },
 ...
 ]
 """
        return [self.find_sub_labels(l) for l in label]

    def find_main_label(self, sub_label: List[Union[list, str]], hard_mapping=True):
        """
 通过子标签找到父标签。

 Args:
 sub_label (List[Union[list, str]]): 子标签, 文本型 或 id_list, e.g. -> '苹果' or [5741, 3362]
 hard_mapping (bool): 当生成的词语不存在时,是否一定要匹配到一个最相似的label。

 Returns:
 dict -> {
 'label': '水果', 
 'token_ids': [3717, 3362]
 }
 """
        if type(sub_label) == list:     # 如果传入为id_list, 则通过tokenizer转回来
            pad_token_id = self.tokenizer.pad_token_id
            while pad_token_id in sub_label:           # 移除[PAD]token
                sub_label.remove(pad_token_id)
            sub_label = ''.join(self.tokenizer.convert_ids_to_tokens(sub_label))
        # print(sub_label)
        main_label = '无'
        for label, s_labels in self.label_dict.items():
            if sub_label in s_labels:
                main_label = label
                break

        if main_label == '无' and hard_mapping:
            main_label = self.hard_mapping(sub_label)
        # print(main_label)
        ret = {
            'label': main_label,
            'token_ids': self.tokenizer(main_label)['input_ids'][1:-1]
        }
        return ret

    def batch_find_main_label(self, sub_label: List[Union[list, str]], hard_mapping=True):
        """
 批量通过子标签找父标签。

 Args:
 sub_label (List[Union[list, str]]): 子标签列表, ['苹果', ...] or [[5741, 3362], ...]

 Returns:
 list: [
 {
 'label': '水果', 
 'token_ids': [3717, 3362]
 },
 ...
 ]
 """
        return [self.find_main_label(l, hard_mapping) for l in sub_label]

损失函数的代码:

def mlm_loss(logits, mask_positions, sub_mask_labels,
             cross_entropy_criterion, device):
    """
 计算指定位置的mask token的output与label之间的cross entropy loss。

 Args:
 logits (torch.tensor): 模型原始输出 -> (batch, seq_len, vocab_size)
 mask_positions (torch.tensor): mask token的位置 -> (batch, mask_label_num)
 sub_mask_labels (list): mask token的sub label, 由于每个label的sub_label数目不同,所以 这里是个变长的list,
 e.g. -> [
 [[2398, 3352]],
 [[2398, 3352], [3819, 3861]]
 ]
 cross_entropy_criterion (CrossEntropyLoss): CE Loss计算器
 device (str): cpu还是gpu

 Returns:
 torch.tensor: CE Loss
 """
    batch_size, seq_len, vocab_size = logits.size()
    loss = None
    for single_value in zip(logits, sub_mask_labels, mask_positions):
        single_logits = single_value[0]
                single_sub_mask_labels = single_value[1]
        single_mask_positions = single_value[2]

        # single_mask_logits形状:(mask_label_num, vocab_size)
        single_mask_logits = single_logits[single_mask_positions] 

        # single_mask_logits按照子标签的长度进行复制:
        # single_mask_logits形状-->(sub_label_num, mask_label_num, vocab_size)
        single_mask_logits = single_mask_logits.repeat(len(single_sub_mask_labels), 1,
                                                       1)  
        #single_mask_logits改变形状:(sub_label_num * mask_label_num, vocab_size)
        #模型预测的结果
        single_mask_logits = single_mask_logits.reshape(-1, vocab_size)

        # single_sub_mask_labels形状:(sub_label_num, mask_label_num)
        single_sub_mask_labels = torch.LongTensor(single_sub_mask_labels).to(device)  

        # single_sub_mask_labels形状: # (sub_label_num * mask_label_num)
        single_sub_mask_labels = single_sub_mask_labels.reshape(-1, 1).squeeze() 

        if not single_sub_mask_labels.size():  # 处理单token维度下维度缺失的问题
            single_sub_mask_labels = single_sub_mask_labels.unsqueeze(dim=0)

        cur_loss = cross_entropy_criterion(single_mask_logits, single_sub_mask_labels)
        cur_loss = cur_loss / len(single_sub_mask_labels)

        if not loss:
            loss = cur_loss
        else:
            loss += cur_loss

    loss = loss / batch_size
    return loss

评估函数的代码:

class ClassEvaluator(object):

    def __init__(self):
        self.goldens = []
        self.predictions = []

    def add_batch(self, pred_batch: List[List], gold_batch: List[List]):
        """
 添加一个batch中的prediction和gold列表,用于后续统一计算。

 Args:
 pred_batch (list): 模型预测标签列表, e.g. -> [0, 0, 1, 2, 0, ...] or [['体', '育'], ['财', '经'], ...]
 gold_batch (list): 真实标签标签列表, e.g. -> [1, 0, 1, 2, 0, ...] or [['体', '育'], ['财', '经'], ...]
 """
        assert len(pred_batch) == len(gold_batch)

        # 若遇到多个子标签构成一个标签的情况
        if type(gold_batch[0]) in [list, tuple]:  
            # 将所有的label拼接为一个整label: ['体', '育'] -> '体育'
            pred_batch = [','.join([str(e) for e in ele]) for ele in pred_batch]  
            gold_batch = [','.join([str(e) for e in ele]) for ele in gold_batch]

        self.goldens.extend(gold_batch)
        self.predictions.extend(pred_batch)

    def compute(self, round_num=2) -> dict:
        """
 根据当前类中累积的变量值,计算当前的P, R, F1。

 Args:
 round_num (int): 计算结果保留小数点后几位, 默认小数点后2位。

 Returns:
 dict -> {
 'accuracy': 准确率,
 'precision': 精准率,
 'recall': 召回率,
 'f1': f1值,
 'class_metrics': {
 '0': {
 'precision': 该类别下的precision,
 'recall': 该类别下的recall,
 'f1': 该类别下的f1
 },
 ...
 }
 }
 """
        classes, class_metrics, res = sorted(list(set(self.goldens) | set(self.predictions))), {}, {}

        # 构建全局指标
        res['accuracy'] = round(accuracy_score(self.goldens, self.predictions), round_num)  

        res['precision'] = round(precision_score(self.goldens, self.predictions, average='weighted'), round_num)

        res['recall'] = round(recall_score(self.goldens, self.predictions, average='weighted'), round_num)

        res['f1'] = round(f1_score(self.goldens, self.predictions, average='weighted'), round_num)

        try:
            conf_matrix = np.array(confusion_matrix(self.goldens, self.predictions))  # (n_class, n_class)
            assert conf_matrix.shape[0] == len(classes)
            for i in range(conf_matrix.shape[0]):  # 构建每个class的指标
                precision = 0 if sum(conf_matrix[:, i]) == 0 else conf_matrix[i, i] / sum(conf_matrix[:, i])
                recall = 0 if sum(conf_matrix[i, :]) == 0 else conf_matrix[i, i] / sum(conf_matrix[:, i])
                f1 = 0 if (precision + recall) == 0 else 2 * precision * recall / (precision + recall)
                class_metrics[classes[i]] = {
                    'precision': round(precision, round_num),
                    'recall': round(recall, round_num),
                    'f1': round(f1, round_num)
                }
            res['class_metrics'] = class_metrics
        except Exception as e:
            print(f'[Warning] Something wrong when calculate class_metrics: {e}')
            print(f'-> goldens: {set(self.goldens)}')
            print(f'-> predictions: {set(self.predictions)}')
            print(f'-> diff elements: {set(self.predictions) - set(self.goldens)}')
            res['class_metrics'] = {}

        return res

    def reset(self):
        """
 重置积累的数值。
 """
        self.goldens = []
        self.predictions = []
6.2.5 模型训练

有了这些工具函数,我们就可以训练模型了,训练的时候,我们只需要计算 MLM 的损失,因为我们把分类任务转换成了 MLM 任务,训练的代码和之前的很类似,我们只需要冻结 BERT 的大部分参数?不,PET 其实是微调 BERT 的所有参数?不对,不对,PET 是在小样本的情况下,微调 BERT 的所有参数,但是因为我们把任务转换成了 MLM 任务,所以即使是小样本,也不会过拟合,效果很好。

训练完之后,我们就能得到一个分类模型,在 590 条验证集上,准确率能达到 80% 以上,而传统的微调,在 63 条数据上,准确率只有 50% 左右,这就是 Prompt-Tuning 的威力!

6.3 基于 BERT+P-Tuning 的软模板方案

PET 的效果已经很好了,但是它用的是人工设计的模板,不稳定,这时候我们可以用 P-Tuning,用可学习的软模板,效果会更好,而且更稳定。

P-Tuning 的代码和 PET 的很类似,唯一的区别就是,模板不再是人工设计的了,而是用可学习的伪 token,训练的时候,我们只训练这些伪 token 的参数,BERT 的参数是固定的,这样就只需要微调很少的参数,就能达到很好的效果。

P-Tuning 的工具函数和 PET 的差不多,Verbalizer、损失函数、评估函数都是一样的,唯一的区别就是数据处理的时候,不再用人工的模板,而是用可学习的伪 token,训练的时候,训练这些伪 token 的参数。

P-Tuning 的效果比 PET 更好,因为它的模板是可以学习的,能够找到最优的模板,而且更稳定,不会因为人工模板的不同而效果波动。


写在最后:大模型时代,我们该如何学习?

到这里,这份 2 万字的 LLM 从入门到实战的指南就结束了,我们从最基础的 LLM 的概念,到主流的模型架构,到 Prompt 工程,到参数高效微调,再到两个完整的实战项目,一步步带你掌握了大模型的核心知识。

大模型时代,学习的方式变了,我们不再需要从零开始训练模型,我们只需要学会如何用好现有的大模型,如何用 Prompt 引导它,如何用微调适配它,如何把它落地到我们的项目里。

希望这份指南能够帮到你,如果你跟着这份指南,把所有的代码都跑一遍,你就能真正掌握大模型的开发,从一个新手,变成一个能够动手做项目的开发者。

最后,如果你觉得这份指南对你有帮助,欢迎点赞、收藏、转发,让更多的人看到,我们一起学习,一起进步,在大模型的时代,抓住这个技术的浪潮!

Logo

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念,把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起,为开发者提供从开发、训练到部署的一站式体验。

更多推荐