# 第 1 章 NLP导论：让计算机听懂人类的语言

欢迎来到NLP实战专栏！在这个AI无处不在的时代，你每天都在与NLP技术打交道——早上用语音助手设置闹钟、上班路上用翻译软件看外文新闻、工作时用ChatGPT写代码、晚上刷短视频时的智能字幕，背后都是自然语言处理技术在默默工作。

本专栏将从最基础的概念讲起，一步步带你搭建开发环境、理解核心算法、实现经典模型，最终掌握预训练模型的使用方法。无论你是刚入门AI的开发者，还是想系统学习NLP的工程师，都能在这里找到适合自己的内容。

# 1.1 什么是自然语言处理

自然语言处理(Natural Language Processing, NLP)，是人工智能领域最具挑战性也最有价值的分支之一。

自然语言，就是我们人类日常交流使用的语言——中文、英文、法语、日语等等。它不是为计算机设计的编程语言，而是在人类几千年的文明发展中自然形成的，充满了歧义、隐喻、省略和文化背景。

NLP的终极目标，就是让计算机能够像人类一样理解自然语言的含义，并且能够生成流畅自然的语言，实现人与计算机之间用自然语言进行无障碍沟通。

简单来说，NLP就是在"人类语言"和"计算机语言"之间架起一座桥梁。

# 1.2 NLP的五大核心任务

NLP技术覆盖了非常广泛的应用场景，但从技术本质上可以归纳为五大类核心任务。理解这些任务，是学习NLP的第一步。

# 1.2.1 文本分类：给整篇文本贴标签

文本分类是最基础也是最常用的NLP任务，它的目标是将一段完整的文本划分到预先定义好的类别中。

常见应用：

情感分析：判断电商评论是"好评"还是"差评"
垃圾邮件识别：将邮件分为"正常邮件"和"垃圾邮件"
新闻主题分类：自动将新闻归类为"科技"、"体育"、"娱乐"等
内容审核：识别文本中是否包含违规内容

001

# 1.2.2 序列标注：给每个词打标签

与文本分类不同，序列标注需要对文本中的每一个字或词都打上对应的标签。它关注的是文本的局部细节，而不是整体含义。

常见应用：

命名实体识别(NER)：从文本中找出人名、地名、组织机构名、时间、电话号码等特定实体
词性标注：标注每个词是名词、动词、形容词还是副词
分词：将连续的中文文本切分成一个个独立的词语

002

# 1.2.3 文本生成：让计算机"写"出文字

文本生成是让计算机根据给定的输入，自动生成符合语法和语义的新文本。这是目前最热门也最具想象力的NLP任务。

常见应用：

自动写作：生成新闻稿、产品描述、小说等
摘要生成：将长文章浓缩成简短的摘要
智能回复：自动生成聊天消息的回复
对话系统：与人类进行多轮自然对话
代码生成：根据自然语言描述生成代码

003

# 1.2.4 信息抽取：从非结构化文本中提取结构化信息

我们生活中的大部分信息都是以非结构化的自然语言形式存在的，比如新闻、合同、病历。信息抽取的任务就是从这些杂乱的文本中，提取出我们需要的结构化信息。

常见应用：

问答系统：根据用户的问题，从文本中找到准确的答案
知识图谱构建：从大量文本中提取实体和实体之间的关系
简历解析：自动从简历中提取姓名、学历、工作经历等信息
合同分析：从合同文本中提取甲方、乙方、金额、期限等关键条款

004

# 1.2.5 文本转换：将一种文本变成另一种文本

文本转换是将输入的文本从一种形式转换为另一种形式，输入和输出都是文本，但内容或格式不同。

常见应用：

机器翻译：将一种语言的文本翻译成另一种语言
文本改写：用不同的表达方式重述相同的意思
繁简转换：繁体中文和简体中文之间的转换
语音转文字/文字转语音：虽然涉及语音，但核心是文本和语音之间的转换

005

# 1.3 NLP技术的演进之路

NLP的发展历史，就是一部人类不断探索如何让计算机理解语言的历史。从最初的人工规则到如今的大语言模型，技术范式发生了四次重大变革。

# 1.3.1 规则系统时代(1950s-1980s)："教"计算机说话

这是NLP的启蒙时代。当时的科学家们认为，既然语言是有规则的，那么只要把这些规则告诉计算机，计算机就能理解语言。

于是，语言学家和程序员们手动编写了成千上万条语法规则和词典。计算机遇到输入文本时，就按照这些规则进行匹配和处理。

代表系统：

Georgetown-IBM实验(1954年)：世界上第一个机器翻译系统，由乔治城大学和IBM联合开发。它成功将60多个俄语句子自动翻译成英语，震惊了全世界。当时人们乐观地认为，机器翻译问题在几年内就能完全解决。

006

ELIZA聊天机器人(1966年)：由麻省理工学院的约瑟夫·魏岑鲍姆开发。它模拟了一位罗杰斯学派的心理医生，通过简单的模式匹配和关键词替换来与用户对话。比如当用户说"我很伤心"时，ELIZA会回复"你为什么觉得很伤心？"。虽然它并不真正理解用户的情绪，但很多人误以为它有智能。

007

阶段特点：

✅ 在特定的小领域内表现良好
❌ 规则需要人工编写，成本极高
❌ 无法处理规则之外的情况，扩展性极差
❌ 完全无法处理语言的歧义性和复杂性

事实证明，人类语言的规则是无穷无尽的。无论编写多少条规则，总有例外情况出现。到了80年代初，基于规则的方法走进了死胡同。

# 1.3.2 统计方法时代(1990s)：从数据中"找"规律

随着计算能力的提升和大规模语料库的出现，统计方法逐渐取代了规则系统。科学家们不再试图教计算机所有的语言规则，而是让计算机从大量的文本数据中自己学习语言的规律。

统计方法的核心思想是：一个语言现象出现的频率越高，它就越可能是正确的。

代表方法：

N-gram模型：最经典的统计语言模型。它的核心假设非常简单：一个词出现的概率，只取决于它前面的N-1个词。

比如在Bigram(2-gram)模型中，"我吃苹果"这句话的概率等于： P(我) × P(吃|我) × P(苹果|吃)

而在Trigram(3-gram)模型中，概率等于： P(我) × P(吃|我) × P(苹果|我吃)

N-gram模型虽然简单，但在当时极大地提升了机器翻译和语音识别的性能。

008

阶段特点：

✅ 从"专家驱动"转向"数据驱动"
✅ 能够处理一定程度的歧义
❌ 只能捕捉到局部的语言规律
❌ 数据稀疏问题严重（很多词组合在语料库中从未出现过）

# 1.3.3 传统机器学习时代(2000s-2010s初)：特征工程的黄金时代

进入21世纪，传统机器学习算法开始大规模应用于NLP领域。逻辑回归、支持向量机(SVM)、决策树、条件随机场(CRF)等算法在文本分类、命名实体识别等任务上取得了突破性进展。

这个阶段的核心是特征工程。研究者们需要设计各种各样的手工特征，比如词频、词性、句法结构等，然后将这些特征输入到机器学习模型中。

经典示例：基于词袋模型与逻辑回归的文本分类

009

词袋模型是最简单的文本表示方法。它将文本看作是一个"袋子"，里面装着一个个独立的词，完全忽略词语的顺序和语法结构。它只统计每个词在文本中出现的次数，然后将这些次数组成一个向量作为文本的表示。

但词袋模型有一个致命的缺陷：它完全丢失了词序信息。比如下面这两条意思完全相反的评论：

评论A："服务很好但味道差劲"
评论B："味道很好但服务差劲"

分词后得到的词列表是完全相同的： ["服务", "很", "好", "但", "味道", "差劲"]

因此，它们在词袋模型中的特征向量也完全一样，模型无法区分这两条评论的情感倾向。

为了解决这个问题，人们引入了N-gram特征。N-gram将相邻的N个词作为一个整体来统计，这样就能保留一部分词序信息。比如使用Trigram(3-gram)，上面两条评论就会变成：

评论A：["服务很好", "很好但", "好但味道", "但味道差劲"]
评论B：["味道很好", "很好但", "好但服务", "但服务差劲"]

这样，它们的特征向量就有了区别，模型就能正确区分它们的情感了。

阶段特点：

✅ 模型泛化能力比统计方法更强
✅ 在很多任务上达到了当时的最佳效果
❌ 特征工程需要大量的专业知识和人力
❌ 手工特征只能捕捉到浅层的语言特征，无法理解深层语义

# 1.3.4 深度学习时代(2015s-至今)：端到端的革命

2012年，AlexNet在ImageNet竞赛中取得了压倒性胜利，开启了计算机视觉的深度学习时代。几年后，深度学习也席卷了NLP领域，彻底改变了NLP的技术范式。

深度学习最大的优势是端到端学习。它不再需要人工设计特征，而是让神经网络从海量数据中自动学习从原始文本到任务目标的映射。

里程碑式的模型：

RNN(循环神经网络)：第一个真正适合处理序列数据的神经网络。它通过循环连接，能够记住之前的信息，处理变长的文本序列。

0010

LSTM(长短期记忆网络)：解决了RNN的梯度消失问题，能够捕捉长距离的依赖关系。它通过精心设计的"门控机制"，控制信息的流入、流出和遗忘。

0011

GRU(门控循环单元)：LSTM的简化版本，它将LSTM的三个门合并为两个门，在保持性能的同时减少了计算量。

0012

Transformer：2017年Google提出的革命性架构。它完全抛弃了循环结构，使用自注意力机制来建模文本中的依赖关系。Transformer不仅能够并行计算，大大提高了训练速度，还能更好地捕捉长距离依赖。Transformer的提出，为预训练语言模型的爆发奠定了基础。

0013

2018年以来，基于Transformer的预训练语言模型(如BERT、GPT、T5)横空出世。这些模型在海量的无标注文本上进行预训练，学习到了通用的语言表示，然后通过微调就可以应用到各种下游任务上，取得了前所未有的效果。如今，预训练+微调已经成为NLP领域的标准范式。

阶段特点：

✅ 无需人工特征工程，端到端学习
✅ 能够捕捉深层的语义信息
✅ 模型性能随着数据量和参数量的增加而持续提升
✅ 预训练模型实现了通用人工智能的第一步
❌ 计算资源需求巨大
❌ 模型可解释性差

# 本章小结

本章我们介绍了NLP的基本概念、五大核心任务以及技术演进的四个阶段。从最初的人工规则到如今的大语言模型，NLP技术已经取得了翻天覆地的变化。

在接下来的章节中，我们将正式开始NLP的实战之旅。下一章，我们将手把手教你搭建基于PyTorch的NLP开发环境，为后续的模型实现做好准备。

# 第2章环境搭建：从零搭建生产级PyTorch NLP开发环境

上一章我们了解了NLP的核心概念与技术演进，从本章开始我们将正式进入实战环节。环境搭建是所有AI开发的第一步，一个稳定、高效的开发环境能帮你避免90%以上的依赖冲突和运行错误。

本专栏全程采用Ubuntu 24.04 LTS + PyTorch 2.x + JupyterLab技术栈：

Ubuntu 24.04是最新的长期支持版本，原生支持Python 3.12，对NVIDIA驱动和CUDA的兼容性最好
PyTorch是目前NLP领域的事实标准框架，动态图特性适合算法调试，HuggingFace生态几乎完全基于PyTorch构建
JupyterLab提供交互式开发体验，支持代码、文档、可视化一体化呈现，非常适合NLP算法的迭代与验证

本章将带你完成从系统初始化到生产级Jupyter服务部署的全流程，同时安装所有后续章节需要的NLP核心依赖。

# 2.1 系统基础准备

首先我们需要更新系统并安装Python基础环境。强烈推荐使用Python 3.12，这是PyTorch 2.3+官方认证的版本，性能更优且依赖冲突最少。

# 更新系统所有软件包到最新版本
sudo apt update && sudo apt upgrade -y

# 安装Python 3.12、pip包管理器和虚拟环境工具
sudo apt install python3.12 python3.12-pip python3.12-venv -y

# 清理apt缓存，释放磁盘空间
sudo apt clean

1
2
3
4
5
6
7
8

# 2.2 生产级JupyterLab部署

我们将JupyterLab部署为系统服务，实现后台运行、开机自启和崩溃自动重启，同时支持远程访问，既适合本地服务器也适合云服务器使用。

# 2.2.1 创建独立虚拟环境

虚拟环境是Python开发的最佳实践，它能为每个项目创建独立的依赖空间，避免不同项目之间的版本冲突。

# 创建统一的环境安装目录（可自定义路径）
sudo mkdir -p /opt/hep-jupyter
# 将目录所有权赋予当前用户，避免后续权限问题
sudo chown $USER:$USER /opt/hep-jupyter
cd /opt/hep-jupyter

# 创建Python 3.12虚拟环境
python3.12 -m venv venv

# 激活虚拟环境（激活后终端前缀会显示(venv)）
source venv/bin/activate

1
2
3
4
5
6
7
8
9
10
11

# 2.2.2 安装并配置JupyterLab

# 安装Node.js（部分Jupyter插件需要）
sudo apt install nodejs npm -y

# 安装JupyterLab（推荐，比经典Notebook功能更强大）
pip install jupyterlab

# 生成Jupyter服务配置文件
jupyter server --generate-config

# 设置Jupyter登录密码（输入两次密码确认）
jupyter server password

# 创建Jupyter工作目录（所有Notebook文件将保存在这里）
sudo mkdir -p /usr/software/jupyter
sudo chown $USER:$USER /usr/software/jupyter

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 2.2.3 修改核心配置文件

编辑配置文件/root/.jupyter/jupyter_server_config.py（注意：如果使用普通用户运行，路径为~/.jupyter/jupyter_server_config.py），添加以下内容：

# 允许所有IP地址访问
c.ServerApp.ip = '0.0.0.0'
# 自定义访问端口（可修改为任意未被占用的端口）
c.ServerApp.port = 1050
# 启动时不自动打开浏览器
c.ServerApp.open_browser = False
# 设置默认工作目录
c.ServerApp.root_dir = '/usr/software/jupyter'
# 禁用token认证（已设置密码，无需token）
c.ServerApp.token = ''
# 允许root用户运行（生产环境建议使用普通用户，此处为方便部署）
c.ServerApp.allow_root = True
# 允许远程访问
c.ServerApp.allow_remote_access = True
# 关闭跨站请求伪造检查（避免远程访问时的权限问题）
c.ServerApp.disable_check_xsrf = True

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

# 2.2.4 配置Systemd系统服务

创建服务文件/etc/systemd/system/hep-jupyter.service，实现JupyterLab的后台运行和开机自启：

[Unit]
Description=HEP JupyterLab Service
After=network.target

[Service]
Type=simple
User=root
Group=root
# 加载系统全局环境变量
EnvironmentFile=-/etc/profile
WorkingDirectory=/opt/hep-jupyter
# 指定虚拟环境的Python路径
Environment="PATH=/opt/hep-jupyter/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"

# JupyterLab启动命令
ExecStart=/opt/hep-jupyter/venv/bin/jupyter lab --no-browser --ip=0.0.0.0 --port=1050 --notebook-dir=/usr/software/jupyter

# 崩溃自动重启策略
Restart=on-failure
RestartSec=5
TimeoutSec=30

[Install]
WantedBy=multi-user.target

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# 2.2.5 启动服务并测试访问

# 重新加载systemd配置
sudo systemctl daemon-reload

# 启动JupyterLab服务
sudo systemctl start hep-jupyter.service

# 设置开机自启
sudo systemctl enable hep-jupyter.service

# 查看服务状态，确认显示"active (running)"
sudo systemctl status hep-jupyter.service

# 开放防火墙端口（如果服务器开启了ufw防火墙）
sudo ufw allow 1050/tcp

1
2
3
4
5
6
7
8
9
10
11
12
13
14

现在你可以在浏览器中输入http://<你的服务器IP>:1050，使用之前设置的密码登录JupyterLab。

# 2.2.6 安装实用插件

# 目录导航插件（自动生成Notebook目录）
pip install jupyterlab-toc

# LaTeX支持插件（支持编写和导出LaTeX文档）
pip install jupyterlab-latex

# 重启服务使插件生效
sudo systemctl restart hep-jupyter.service

1
2
3
4
5
6
7
8

# 2.3 PyTorch与NLP核心依赖安装

# 2.3.1 配置Pip国内镜像（永久生效）

根据最新测试，清华PyPI源当前访问失败，我们统一配置阿里云镜像，这是目前国内最稳定的源，后续安装无需每次加-i参数。

# 升级pip到最新版本
pip install --upgrade pip

# 永久配置阿里云镜像
pip config set global.index-url https://mirrors.aliyun.com/pypi/simple/
pip config set global.trusted-host mirrors.aliyun.com

1
2
3
4
5
6

# 2.3.2 基础数据科学与工具包

# 基础数值计算与数据处理
pip install numpy pandas scipy scikit-learn statsmodels

# 数据可视化
pip install matplotlib seaborn pyecharts jupyter-echarts

# 办公与数据库
pip install openpyxl pymysql sqlalchemy

# 图像处理（部分多模态NLP任务需要）
pip install pillow opencv-python imutils

# 通用工具
pip install tqdm flask fastapi uvicorn python-dotenv kaggle

1
2
3
4
5
6
7
8
9
10
11
12
13
14

# 2.3.3 NLP专用工具包

# 中文分词
pip install jieba

# 词向量训练与主题建模
pip install gensim

# 英文自然语言处理
pip install nltk

# 数据集加载与处理
pip install datasets evaluate seqeval

# 预训练模型核心库（后续章节重点使用）
pip install transformers accelerate

# 语义向量生成
pip install sentence-transformers

# 数据标注工具（可选，用于自定义数据集标注）
pip install label-studio ml-backend

# 大模型应用开发框架（可选）
pip install langchain langchain-deepseek langchain_huggingface langchain_neo4j neo4j

# 训练可视化工具
pip install tensorboard

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

# 2.3.4 PyTorch深度学习框架

根据你的硬件环境选择对应的版本，有NVIDIA显卡强烈推荐安装GPU版本，训练速度会比CPU版本快10-100倍。

# ========== CPU版本（无GPU环境使用） ==========
# pip install torch torchvision torchaudio

# ========== GPU版本（NVIDIA显卡，CUDA 12.1） ==========
# 此版本兼容30系、40系等绝大多数NVIDIA显卡
# 官方CUDA 12.1源已验证可用（见参考文档2）
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121

1
2
3
4
5
6
7

注意：如果你需要其他CUDA版本（如cu118），请前往PyTorch官方网站 (opens new window)获取对应安装命令。

# 2.4 环境验证与常见问题排查

# 2.4.1 核心环境验证

在JupyterLab中新建一个Python Notebook，运行以下代码验证所有核心包是否安装成功：

# 验证Python版本
import sys
print(f"Python版本: {sys.version}")

# 验证基础数据科学包
import numpy
import pandas
print(f"NumPy版本: {numpy.__version__}")
print(f"Pandas版本: {pandas.__version__}")

# 验证PyTorch
import torch
print(f"PyTorch版本: {torch.__version__}")
print(f"CUDA是否可用: {torch.cuda.is_available()}")
if torch.cuda.is_available():
    print(f"GPU设备: {torch.cuda.get_device_name(0)}")
    print(f"CUDA版本: {torch.version.cuda}")

# 验证NLP核心包
import jieba
import gensim
import transformers
print(f"jieba版本: {jieba.__version__}")
print(f"gensim版本: {gensim.__version__}")
print(f"transformers版本: {transformers.__version__}")

print("\n✅ 所有核心包安装成功！")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

如果所有版本号都正常输出且CUDA显示可用，说明环境搭建完成。

# 2.4.2 常见问题排查

权限不足错误：确保所有操作目录的所有权属于当前用户，或使用sudo执行必要的系统命令
端口被占用：修改jupyter_server_config.py中的port为其他未被占用的端口，然后重启服务
CUDA不可用：检查NVIDIA驱动是否正确安装，确保PyTorch版本与CUDA版本匹配
pip安装慢：确认已正确配置阿里云镜像，不要使用已失效的清华源
Jupyter无法远程访问：检查服务器防火墙是否开放了对应端口，云服务器还需在安全组中放行端口

# 本章小结

本章我们完成了NLP开发环境的全流程搭建：

准备了Ubuntu 24.04系统基础环境
部署了支持远程访问和开机自启的生产级JupyterLab服务
配置了国内镜像源，安装了所有后续章节需要的基础工具和NLP专用包
安装了PyTorch深度学习框架，并验证了GPU加速功能

现在你的开发环境已经完全准备就绪。下一章我们将进入NLP的核心基础——文本表示，学习如何将人类能理解的文字转换成计算机能处理的数字向量，这是所有NLP模型的输入基础。

# 第3章文本表示：把文字变成计算机能懂的数字

每天学习一小时，一年以后成大神！

2026-06-02_114506

上一章我们完成了生产级NLP开发环境的搭建，从本章开始我们将正式进入NLP的核心技术学习。如果说NLP是让计算机听懂人类语言，那么文本表示就是计算机的"语言翻译器"——它负责把人类能理解的自然语言，转换成计算机能计算和处理的数值形式。

所有NLP任务，从最简单的文本分类到最复杂的大语言模型，都建立在文本表示的基础之上。本章我们将系统学习从分词到词向量的完整流程，掌握现代NLP中最常用的文本表示方法。

# 3.1 文本表示的核心逻辑

计算机天生只能理解数字，无法直接处理"你好"、"自然语言处理"这样的文字符号。文本表示的任务，就是建立一套从"文字"到"数字"的映射规则，让计算机能够通过数值计算来理解语言的含义。

一个完整的文本表示流程分为两步：

分词与词表构建：将连续的文本切分成独立的语义单元(token)，并为每个token分配唯一的数字ID
向量化：将每个token的ID转换为低维稠密的向量表示，用于后续模型计算

在模型训练和预测时，完整的数据流是这样的：

输入文本 → 分词 → 词表映射(token→ID) → 嵌入层(ID→向量) → 模型计算
模型输出(概率分布) → 词表映射(ID→token) → 拼接成输出文本

# 3.2 分词：NLP任务的起点

分词(Tokenization)是将原始文本切分为若干具有独立语义的最小单元(token)的过程。不同语言的结构差异很大，分词策略也完全不同。

# 3.2.1 英文分词的三种粒度

英文有天然的空格作为词边界，因此分词相对简单。按照切分粒度从小到大，可分为词级、字符级和子词级三种。

分词粒度	核心思想	优点	缺点	适用场景
词级分词	按空格和标点切分完整词语	符合人类阅读习惯，语义完整	存在严重的OOV(未登录词)问题	传统NLP任务，小数据集
字符级分词	按单个字符切分	词表极小，完全无OOV问题	单个字符语义极弱，序列过长	拼写纠错、字符级生成
子词级分词	切分为词根、前缀、后缀等语义单元	兼顾语义完整性和OOV覆盖率	实现相对复杂	现代大语言模型(BERT、GPT等)

# 3.2.1.1 词级分词

词级分词是最直观的方式，直接按空格和标点切分文本。

但它的致命缺陷是OOV问题：当输入中出现词表中没有的词(如网络热词、专有名词、拼写错误)时，只能用特殊标记<UNK>代替，导致语义完全丢失。例如"ChatGPT"、"Transformer"这些词在早期词表中都不存在。

# 3.2.1.2 字符级分词

字符级分词将每个字母、数字、标点都视为独立的token。

这种方式完全解决了OOV问题，但代价是序列长度大幅增加，且单个字符几乎没有语义，模型需要更长的上下文才能推断出完整含义，训练和推理效率极低。

# 3.2.1.3 子词级分词(主流方法)

子词级分词是介于词级和字符级之间的最优解，它将词语切分为更小的语义单元。即使一个完整的词不在词表中，只要能拆分为词表中存在的子词，就能被正确表示。

目前最常用的子词分词算法是BPE(Byte Pair Encoding)，它最初是一种数据压缩算法，后来被OpenAI用于GPT模型的分词，现在已成为所有大语言模型的标准。

BPE算法完整流程(基于Hugging Face官方教程)：

训练阶段：

将语料中所有词拆分为单个字符，构建初始词表
迭代统计语料中出现频率最高的相邻字符对
将最高频的字符对合并为一个新的子词，加入词表
重复上述过程，直到词表大小达到预设上限

举个具体例子：假设我们的语料有5个词，频率分别为 ("hug",10), ("pug",5), ("pun",12), ("bun",4), ("hugs",5)

初始词表：["b","g","h","n","p","s","u"]

第1次合并：最高频对是("u","g")(出现20次)，合并为"ug"
第2次合并：最高频对是("u","n")(出现16次)，合并为"un"
第3次合并：最高频对是("h","ug")(出现15次)，合并为"hug"

分词阶段：将输入文本拆分为最小字符单元，然后按训练时学到的合并规则，从短到长依次合并，直到无法继续。

例如用上面训练的规则分词"bug"：先拆分为["b","u","g"]，然后合并"u"+"g"为"ug"，最终得到["b","ug"]。

💡 详细的BPE代码实现可参考：Hugging Face官方BPE教程 (opens new window)

# 3.2.2 中文分词的特点与方法

中文没有天然的词边界，因此分词比英文复杂得多。同样按照粒度划分，中文也有三种分词方式：

字符级分词：直接按单个汉字切分。由于汉字本身通常具有独立语义，这种方式在中文中比英文更实用，早期很多中文NLP模型都采用这种方法。

词级分词：按照完整词语切分，需要依赖词典或机器学习模型识别词语边界。这是传统中文NLP的主流方法。

子词级分词：虽然中文没有英文的词根词缀结构，但BPE等子词算法依然可以直接应用。它们以汉字为基本单位，学习语料中高频的字组合(如"自然"、"语言"、"处理")。目前所有主流中文大模型(通义千问、DeepSeek、文心一言等)都采用子词分词。

# 3.2.3 常用分词工具实战

# 3.2.3.1 jieba：传统中文分词首选

jieba是最经典的中文分词工具，接口简单、性能优秀，支持自定义词典，非常适合传统NLP任务。

⚠️ 注意：jieba的GitHub官方仓库目前境外访问受限，但你可以通过国内PyPI镜像正常安装使用。

安装：

pip install jieba -i https://mirrors.aliyun.com/pypi/simple/

三种分词模式：

import jieba

text = "小明毕业于北京大学计算机系"

# 1. 精确模式(默认)：最适合文本分析
words = jieba.lcut(text)
print("精确模式:", words)
# 输出: ['小明', '毕业', '于', '北京大学', '计算机系']

# 2. 全模式：扫描所有可能成词的词语，速度快但有歧义
words = jieba.lcut(text, cut_all=True)
print("全模式:", words)
# 输出: ['小', '明', '毕业', '于', '北京', '北京大学', '大学', '计算', '计算机', '计算机系', '算机', '系']

# 3. 搜索引擎模式：对长词再次切分，适合搜索引擎
words = jieba.lcut_for_search(text)
print("搜索引擎模式:", words)
# 输出: ['小明', '毕业', '于', '北京', '大学', '北京大学', '计算', '算机', '计算机', '计算机系']

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

自定义词典：

# 加载自定义词典(每行格式：词语 词频 词性)
jieba.load_userdict("my_dict.txt")

# 动态添加/删除词语
jieba.add_word("大模型", freq=10)
jieba.del_word("小模型")

text = "大模型技术正在改变人工智能行业"
print(jieba.lcut(text))
# 输出: ['大模型', '技术', '正在', '改变', '人工智能', '行业']

1
2
3
4
5
6
7
8
9
10

# 3.2.3.2 Hugging Face Tokenizers：大模型时代的标准

Hugging Face Tokenizers是一个高性能的分词库，支持BPE、WordPiece等所有主流子词算法，是使用预训练模型时的必备工具。

⚠️ 注意：Hugging Face Tokenizers的GitHub仓库目前境外访问受限，但你可以通过国内PyPI镜像正常安装使用。

安装：

pip install tokenizers transformers -i https://mirrors.aliyun.com/pypi/simple/

快速使用预训练分词器：

from transformers import AutoTokenizer

# 加载BERT中文分词器
tokenizer = AutoTokenizer.from_pretrained("bert-base-chinese")

text = "自然语言处理是人工智能的一个重要分支"

# 完整分词流程
encoded = tokenizer(
    text,
    padding=True,
    truncation=True,
    max_length=512,
    return_tensors="pt"
)
print("token IDs:", encoded["input_ids"])
print("attention mask:", encoded["attention_mask"])

# 查看分词结果
tokens = tokenizer.convert_ids_to_tokens(encoded["input_ids"][0])
print("分词结果:", tokens)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

# 3.3 词表示：从稀疏向量到语义向量

分词完成后，我们得到了一系列token，但这些字符串本身无法直接用于计算。我们需要将每个token转换为一个数值向量，这就是词表示。

词表示技术经历了三代演进：

第一代：稀疏表示(One-hot编码)
第二代：静态稠密表示(Word2Vec、GloVe)
第三代：动态上下文表示(ELMo、BERT、GPT)

# 3.3.1 One-hot编码：最原始的表示方法

One-hot编码是最简单的词表示方法：为词表中的每个词分配一个唯一的索引，然后用一个长度等于词表大小的向量表示该词，向量中只有对应索引的位置为1，其余为0。

致命缺陷：

语义鸿沟：任意两个词的向量都是正交的，无法表示"猫"和"狗"比"猫"和"桌子"更相似
维度灾难：词表有10万个词，每个向量就是10万维，其中只有一个1，极其浪费空间

因此，One-hot编码现在仅用于词表映射的中间步骤，不再作为最终的词表示使用。

# 3.3.2 Word2Vec：第一个真正的语义化词向量

2013年Google提出的Word2Vec模型彻底改变了NLP领域。它通过对大规模语料的无监督学习，为每个词生成一个低维稠密向量，语义相近的词在向量空间中距离更近。

# 3.3.2.1 核心思想：分布假设

Word2Vec的理论基础是分布假设：一个词的含义，由它周围经常出现的词决定。

例如，"苹果"经常和"吃"、"水果"、"红色"一起出现，而"苹果公司"经常和"手机"、"电脑"、"乔布斯"一起出现，模型就能通过上下文区分这两个不同的含义(虽然静态词向量最终还是只能给"苹果"一个向量)。

# 3.3.2.2 两种模型结构

Word2Vec有两种经典的模型结构，它们的训练目标相反：

CBOW(Continuous Bag-of-Words)模型：输入是一个词的上下文(即前后若干个词)，模型的目标是预测中间的目标词。

Skip-gram 模型：输入是一个中心词，模型的目标是预测其上下文中的所有词(即前后若干个词)。

只要按照上述目标训练模型，就能得到语义化的词向量。

# 3.3.2.3 Word2Vec完整原理

# 数据集准备

Word2Vec 不依赖人工标注，而是直接利用大规模原始文本(如书籍、新闻、网页等)作为数据源，从中自动构造训练样本。

由于两种模型的输入和输出都是词语，因此首先需要对原始文本进行分词，将连续文本转换为 token 序列。此外，模型无法直接处理文本符号，训练时仍需将词语转换为 one-hot 编码，以便作为模型的输入和输出进行计算。

# Skip-Gram模型详解

# 训练数据集

Skip-Gram的目标是根据中间词预测上下文，所以其训练样本构造方式如下：

# 模型结构

Skip-Gram模型的核心结构非常简洁，只有一个输入层、一个隐藏层和一个输出层：

# 损失计算与前向传播

Skip-Gram模型的损失值计算流程如下图所示：

前向传播过程如下：

输入中心词(地铁)：将"地铁"转换为对应的one-hot向量
查找词向量( $W_{i n}$ )：与输入层到隐藏层的参数矩阵 $W_{i n}$ 相乘，取出"地铁"对应的词向量( $W_{i n}$ 实际上就是最终的词向量矩阵，每一行表示一个词的向量)
预测上下文：将中心词向量与隐藏层到输出层的参数矩阵 $W_{o u t}$ 相乘，得到对整个词表的预测得分
Softmax输出：将得分通过Softmax函数转换为概率分布，表示每个词作为上下文的可能性
计算损失：与真实上下文词"乘坐"、"上班"的one-hot向量进行比对，计算交叉熵损失并求和，得到总损失

之后在反向传播过程中，参数矩阵 $W_{i n}$ 中"地铁"对应的词向量会被更新。模型通过不断迭代这个过程，最终便能得到具有丰富语义的词向量。

# CBOW模型详解

# 训练数据集

CBOW的目标是根据上下文预测中间词，所以其训练样本构造方式与Skip-Gram正好相反：

# 模型结构

CBOW模型的结构与Skip-Gram基本相同，区别在于输入是多个上下文词而不是单个中心词：

# 损失计算与前向传播

CBOW模型的损失值计算流程如下图所示：

前向传播过程如下：

输入上下文词(乘坐、上班)：将每个上下文词转换为对应的one-hot向量
查找词向量( $W_{i n}$ )：每个one-hot向量与参数矩阵 $W_{i n}$ 相乘，查出对应的词向量
平均上下文向量：将多个上下文词向量取平均，得到一个整体的上下文表示
预测中心词：将平均后的上下文向量与参数矩阵 $W_{o u t}$ 相乘，得到对整个词表的预测得分
Softmax输出：将得分输入Softmax函数，得到每个词作为中心词的概率分布
计算损失：将预测结果与真实中心词"地铁"的one-hot向量进行比对，计算交叉熵损失

之后在反向传播过程中，参数矩阵 $W_{i n}$ 中"乘坐"和"上班"对应的词向量会被更新。模型通过不断训练，逐步优化这些向量，最终便能得到具有语义的词向量。

# 3.3.2.4 获取Word2Vec词向量

词向量的获取通常有两种方式：一种是直接使用他人公开发布的词向量，另一种是在特定语料上自行训练。

在实际工作中，无论是加载已有模型还是从零训练，都可借助Gensim来完成，它提供了便捷的接口来加载Word2Vec格式的词向量，也支持基于自有语料训练属于自己的词向量模型。

⚠️ 注意：Gensim的官方文档和GitHub仓库目前境外访问受限，但你可以通过国内PyPI镜像正常安装使用。

安装Gensim：

pip install gensim -i https://mirrors.aliyun.com/pypi/simple/

KeyedVectors vs 完整Word2Vec模型： Gensim中词向量有两种存储形式，它们的区别如下：

特性	KeyedVectors	完整Word2Vec模型
继续训练	❌ 不支持	✅ 支持
内存占用	✅ 小(仅存储向量)	❌ 大(包含训练参数)
加载速度	✅ 快(支持mmap)	❌ 慢
支持C格式加载	✅ 支持	❌ 不支持

✅ 最佳实践：如果你只需要查询词向量，使用KeyedVectors；如果你需要继续训练模型，使用完整的Word2Vec模型。

方式一：使用公开预训练词向量 ⚠️ 注意：原百度网盘分享链接已失效，你可以通过以下方式获取中文词向量：

使用Gensim官方提供的预训练模型(英文为主)
从国内学术镜像站下载腾讯AI Lab中文词向量(https://ai.tencent.com/ailab/nlp/en/embedding.html)

词向量文件的通用格式为：第一行记录基本信息，包括两个整数，分别表示总词数和词向量维度。从第二行起，每一行表示一个词及其对应的词向量，格式为：词 + 向量的各个维度值，所有内容通过空格分隔。

&lt;词汇总数> &lt;向量维度>
word1 val11 val12 ... val1N
word2 val21 val22 ... val2N
...

1
2
3
4

可使用KeyedVectors.load_word2vec_format()加载上述词向量文件：

from gensim.models import KeyedVectors

# 加载预训练词向量(以300维微博词向量为例)
# 替换为你本地的词向量文件路径
wv = KeyedVectors.load_word2vec_format("sgns.weibo.word.bz2", binary=False)

# 1. 查看词向量维度
print("词向量维度:", wv.vector_size)

# 2. 查看某个词的向量
print("'地铁'的向量前5维:", wv["地铁"][:5])

# 3. 计算两个词的余弦相似度
similarity = wv.similarity("地铁", "公交")
print(f"地铁 vs 公交 相似度: {similarity:.4f}")  # 输出约0.85

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

余弦相似度的计算公式如下：

s i m i l a r i t y (w_{1}, w_{2}) = cos (θ) = \frac{w _{1} ∙ w _{2}}{∥ ∥ ∥ ∥ w _{1} ∥ ∥ ∥ ∥ ∙ ∥ ∥ ∥ ∥ w _{2} ∥ ∥ ∥ ∥}

返回值介于[-1,1]之间：接近1表示高度相似，语义接近；接近0表示无明显相关；接近-1表示语义完全相反。

# 4. 找出与某个词最相似的词
print("与'上班'最相似的5个词:")
for word, sim in wv.most_similar("上班", topn=5):
    print(f"  {word}: {sim:.4f}")

# 5. 经典的词向量类比：爸爸 - 男性 + 女性 = 妈妈
result = wv.most_similar(positive=["爸爸", "女性"], negative=["男性"], topn=1)
print(f"爸爸 - 男性 + 女性 = {result[0][0]}")

1
2
3
4
5
6
7
8

方式二：在自己的语料上训练词向量

import jieba
import pandas as pd
from gensim.models import Word2Vec

# 1. 准备语料(以电商评论数据集为例)
# 数据集来源：ChineseNLPCorpus(境外访问受限，可从国内镜像站下载)
df = pd.read_csv("online_shopping_10_cats.csv", encoding="utf-8")
# 分词并去除空白字符
sentences = []
for review in df["review"]:
    words = [token for token in jieba.lcut(str(review)) if token.strip()]
    sentences.append(words)

# 2. 训练Word2Vec模型
model = Word2Vec(
    sentences=sentences,
    vector_size=100,  # 词向量维度(通常100-300)
    window=5,         # 上下文窗口大小(左右各5个词)
    min_count=2,      # 忽略出现次数少于2的词
    sg=1,             # 1=Skip-gram, 0=CBOW
    workers=4,        # 并行训练线程数
    epochs=10         # 训练轮数
)

# 3. 保存词向量(仅保存向量部分，不保存训练参数)
model.wv.save_word2vec_format("my_vectors.kv")

# 4. 加载保存的词向量
wv = KeyedVectors.load_word2vec_format("my_vectors.kv")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

# 3.3.2.5 应用Word2Vec词向量

训练好的词向量，通常用于初始化下游NLP任务的嵌入层。

在现代深度学习的NLP模型中，大多数任务的输入第一层都是嵌入层。本质上，嵌入层就是一个查找表(lookup table)：输入是词在词汇表中的索引；输出是该词对应的向量表示。

嵌入层的参数矩阵可以有两种典型的初始化方式：

随机初始化：模型训练开始时，嵌入向量是随机生成的，模型会通过反向传播逐步学习每个词的表示
使用预训练词向量初始化：加载训练好的词向量(如Word2Vec)到嵌入层中作为初始参数，这样可以为模型注入丰富的语言知识，尤其在低资源任务中优势明显。加载预训练词向量后，可选择是否让嵌入层继续参与训练

下面以PyTorch为例，演示如何使用预训练词向量初始化Embedding层，核心API为nn.Embedding.from_pretrained：

import torch
import torch.nn as nn
from gensim.models import KeyedVectors

# 1. 加载预训练的Word2Vec模型
word_vectors = KeyedVectors.load_word2vec_format("my_vectors.kv")

# 2. 构建词表和词向量矩阵
word2index = word_vectors.key_to_index  # 词到索引的映射
embedding_dim = word_vectors.vector_size  # 词语向量维度
num_embeddings = len(word2index)  # 词表大小

# 构造词向量矩阵，形状为(词表大小, 词向量维度大小)
embedding_matrix = torch.zeros(num_embeddings, embedding_dim)
for word, idx in word2index.items():
    embedding_matrix[idx] = torch.tensor(word_vectors[word])

# 3. 构建PyTorch的嵌入层
embedding_layer = nn.Embedding.from_pretrained(
    embedding_matrix,  # 词向量矩阵
    freeze=False,      # 是否冻结词向量(False=继续训练，True=固定不变)
    padding_idx=0      # 指定padding token的索引(向量全0)
)

# 4. 示例：将词索引转换为向量
input_words = ["我", "喜欢", "乘坐", "地铁"]  # 分词后的句子
input_indices = [word2index[word] for word in input_words]  # token转为索引
input_tensor = torch.tensor([input_indices])  # 构造嵌入层输入张量

# 5. 查询嵌入(即词向量查找)
output = embedding_layer(input_tensor)  # 通过嵌入层查找预训练词向量

print(f"输入形状: {input_tensor.shape}")
print(f"输出形状: {output.shape}")  # 输出: torch.Size([1, 4, 100])

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

✅ 最佳实践：

在数据量充足时，设置freeze=False，让词向量在下游任务中继续微调
在数据量较少时，设置freeze=True，避免过拟合
始终指定padding_idx，确保padding token的向量始终为0

# 3.3.3 上下文相关词表示：解决一词多义问题

虽然像Word2Vec这样的模型已经能够为词语提供具有语义的向量表示，但是它只为每个词分配一个固定的向量表示，不论它在句中出现的语境如何。这种表示被称为静态词向量(static embeddings)。

然而，语言的表达极其灵活，一个词在不同上下文中可能有完全不同的含义。例如：

这时，使用同一个静态词向量去表示"苹果"，显然无法区分这两种语义。这就推动了上下文相关的词表示的发展。

上下文相关词表示(Contextual Word Representations)，是指词语的向量表示会根据它所在的句子上下文动态变化，从而更好地捕捉其语义。一个具有代表性的模型是——ELMo (opens new window)。

该模型全称为 Embeddings from Language Models，发表于2018年2月。其基于LSTM语言模型，使用上下文动态生成每个词的表示，每个词的向量由其前文和后文共同决定，是第一个被广泛应用于下游任务的上下文词向量模型。

不过，ELMo基于LSTM架构，并行计算能力差，且只能捕捉有限的长距离依赖。几个月后，基于Transformer架构的BERT模型横空出世，彻底取代了ELMo，开启了预训练语言模型的新时代。我们将在第8章详细讲解BERT和其他预训练模型。

# 本章小结

本章我们系统学习了NLP的基础——文本表示技术：

文本表示的核心是将文字转换为计算机能计算的数字，分为分词和向量化两步
分词是NLP的起点，子词级分词(BPE)是现代大模型的标准
词表示经历了从One-hot到Word2Vec再到上下文相关表示的演进
Word2Vec通过分布假设学习语义化词向量，Gensim是最常用的工具
预训练词向量可以显著提升下游任务的性能，尤其在低资源场景下

下一章我们将学习传统序列模型——RNN、LSTM和GRU，它们是第一个能够处理变长序列数据的神经网络，也是深度学习在NLP领域取得突破的起点。

# 第四章传统序列模型：从RNN到GRU的演进之路

在自然语言处理的发展历程中，序列建模一直是核心挑战。语言本身就是一种典型的序列数据——词语的顺序直接决定了句子的含义。虽然我们在第三章学习了如何将词语表示为向量，但词向量本身并不包含顺序信息。

本章我们将深入学习三种经典的序列模型：RNN(循环神经网络)、LSTM(长短期记忆网络)和GRU(门控循环单元)。它们一脉相承，逐步解决了序列建模中的核心问题，为后续Transformer的出现奠定了坚实基础。

# 4.1 RNN：序列建模的开山之作

# 4.1.1 概述：为什么需要循环神经网络？

想象一下你正在读一本书："今天天气很好，我打算去公园____"。当你看到"公园"这个词时，你会立刻想到后面可能接"散步"、"玩"、"野餐"等词。这是因为你记住了前面的上下文信息。

传统的前馈神经网络做不到这一点——它只能处理固定长度的输入，而且每个输入之间是相互独立的。为了让模型能够像人一样"边读边记"，研究者提出了循环神经网络(Recurrent Neural Network, RNN)。

RNN的核心思想非常简单：逐个读取序列中的元素，每一步都结合当前输入和之前积累的记忆，生成新的记忆并传递到下一步。通过这种循环机制，RNN能够持续建模上下文信息，从而理解整个句子的含义。

重要说明

虽然RNN已经逐渐被计算效率更高的Transformer模型所取代，但它仍然具有不可替代的学习价值：

它是所有序列模型的基础，理解RNN是学习LSTM和GRU的前提

它的"循环建模上下文"思想深刻影响了后续所有NLP模型

在一些资源受限的场景中，轻量级的RNN仍然是很好的选择

# 4.1.2 基础结构：循环连接的奥秘

RNN的核心是一个具有循环连接的隐藏层。它以时间步(time step)为单位，依次处理输入序列中的每个token。

在每个时间步t：

接收两个输入：当前token的向量 $x_{t}$ 和上一个时间步的隐藏状态 $h_{t - 1}$
计算并生成新的隐藏状态 $h_{t}$
将 $h_{t}$ 传递到下一个时间步 $t + 1$

具体结构如下图所示：

NLP-Chapter04-001

隐藏状态的计算公式为：

h_{t} = t a n h (x_{t} W_{x} + h_{t - 1} W_{h} + b)

其中：

$W_{x}$ 是输入到隐藏层的权重矩阵
$W_{h}$ 是隐藏层到隐藏层的权重矩阵
$b$ 是偏置项
$t a n h$ 是双曲正切激活函数，将输出压缩到(-1, 1)之间

计算细节如下图所示：

NLP-Chapter04-002

为了更清晰地展示复杂结构，接下来我们将使用简化的示意图，省略内部细节，突出整体连接方式：

$h_{t} = tanh (h_{t - 1} W_{h} + x_{t} W_{x} + b)$

# 4.1.3 多层结构：捕捉更复杂的语义特征

单层RNN的表达能力有限，为了让模型能够学习更复杂的语言模式，我们可以将多个RNN层堆叠起来，形成多层RNN。

这种设计基于一个重要假设：底层网络学习局部特征(如词组、短语)，高层网络学习抽象特征(如句子主题、情感倾向)。

在多层RNN中：

最底层接收原始输入序列
每一层的输出序列作为下一层的输入序列
最顶层的输出作为最终结果用于后续任务

NLP-Chapter04-003

# 4.1.4 双向结构：同时利用上下文信息

基础RNN有一个明显的缺陷：它只能利用上文信息，无法利用下文信息。

比如在序列标注任务中，要判断"我爱吃苹果"中的"苹果"是水果还是公司，需要看后面的词。如果后面是"手机"，那就是公司；如果后面是"派"，那就是水果。

为了解决这个问题，研究者提出了双向RNN(Bidirectional RNN)：

同时使用两个独立的RNN层
一个正向RNN按时间顺序处理序列(从左到右)
一个反向RNN按逆时间顺序处理序列(从右到左)
每个时间步的输出是正向和反向隐藏状态的拼接或求和

双向RNN的结构如下图所示：

NLP-Chapter04-004

# 4.1.5 多层+双向结构：强强联合

多层结构和双向结构可以组合使用，形成多层双向RNN。每一层都是一个双向RNN，能够同时捕捉不同层次的上下文信息。

NLP-Chapter04-005

# 4.1.6 API使用：PyTorch中的RNN实现

PyTorch提供了torch.nn.RNN模块，让我们可以轻松构建各种RNN结构。

# 4.1.6.1 参数说明

torch.nn.RNN(
    input_size,
    hidden_size,
    num_layers=1,
    nonlinearity="tanh",
    bias=True,
    batch_first=False,
    dropout=0.0,
    bidirectional=False,
    device=None,
    dtype=None,
)

1
2
3
4
5
6
7
8
9
10
11
12

参数名	类型	说明
input_size	int	每个时间步输入特征的维度(词向量维度)
hidden_size	int	隐藏状态的维度
num_layers	int	RNN层数，默认为1
nonlinearity	str	激活函数，'tanh'(默认)或'relu'
bias	bool	是否使用偏置项，默认True
batch_first	bool	输入张量是否是(batch, seq, feature)，默认False表示(seq, batch, feature)
dropout	float	除最后一层外，其余层之间的dropout概率
bidirectional	bool	是否为双向RNN，默认False
device	torch.device or str	模块的初始化设备，如'cuda', 'cpu'
dtype	torch.dtype	模块初始化时的默认数据类型

关键提示

batch_first是最容易出错的参数！默认情况下，PyTorch期望输入形状为(序列长度, 批次大小, 特征维度)。但在实际开发中，我们通常使用(批次大小, 序列长度, 特征维度)的格式，因此需要将batch_first=True。

# 4.1.6.2 输入输出

rnn = torch.nn.RNN()
output, h_n = rnn(input, h_0)

1
2

	名称	形状说明
输入	input	(seq_len, batch_size, input_size)，如果batch_first=True则为(batch_size, seq_len, input_size)
	h_0	可选，初始隐藏状态，形状为(num_layers × num_directions, batch_size, hidden_size)
输出	output	最后一层每个时间步的隐藏状态，形状为(seq_len, batch_size, num_directions × hidden_size)，如果batch_first=True则为(batch_size, seq_len, num_directions × hidden_size)
	h_n	最后一个时间步的隐藏状态，包含每一层的每个方向，形状为(num_layers × num_directions, batch_size, hidden_size)

不同结构的输入输出形状对比：

单层单向
多层单向
单层双向
多层双向

# 4.1.7 案例实操：智能输入法词语联想

现在我们来动手实现一个基于RNN的智能输入法词语联想模型。这个模型能够根据用户已输入的文本，预测下一个最可能出现的词语。

# 4.1.7.1 需求说明

功能：根据上文预测下一个词，返回概率最高的5个候选词
示例：输入"自然语言" → 输出["处理", "理解", "的", "描述", "生成"]
应用场景：手机输入法、智能写作助手

# 4.1.7.2 需求分析

数据集处理

我们需要大量自然语言对话数据来训练模型
本案例使用HundredCV-Chat数据集：https://huggingface.co/datasets/Jax-dan/HundredCV-Chat
采用滑动窗口方法构造训练样本：从长文本中截取连续的n个词作为输入，第n+1个词作为目标

模型结构设计 模型由三部分组成：

嵌入层：将词索引转换为稠密向量
RNN层：建模上下文信息，输出最后一个时间步的隐藏状态
全连接层：将隐藏状态映射到词表大小，生成概率分布

训练方案

损失函数：CrossEntropyLoss(多分类任务)
优化器：Adam(收敛快，稳定性好)
评估指标：Top-1准确率和Top-5准确率

# 4.1.7.3 需求实现

项目结构

input_method/
├── data/
│   ├── raw/          # 原始数据
│   └── processed/    # 预处理后的数据
├── logs/             # TensorBoard训练日志
├── models/           # 保存训练好的模型
└── src/
    ├── config.py     # 超参数配置
    ├── dataset.py    # 自定义数据集
    ├── evaluate.py   # 模型评估
    ├── model.py      # 模型定义
    ├── predict.py    # 模型推理
    ├── process.py    # 数据预处理
    ├── tokenizer.py  # 分词器
    └── train.py      # 模型训练

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

完整代码

数据预处理(process.py)

# process.py
import pandas as pd
from sklearn.model_selection import train_test_split
from tokenizer import JiebaTokenizer
import config

def process():
    """数据预处理主函数"""
    print("开始处理数据")
    
    # 1. 读取原始数据
    df = pd.read_json(config.RAW_DATA_DIR / 'chat_data.jsonl', lines=True)
    
    # 2. 数据清洗
    df = df.dropna()
    df = df[df['text'].str.strip().ne('')]
    
    # 3. 划分训练集和测试集
    train_df, test_df = train_test_split(df, test_size=0.2, random_state=42)
    
    # 4. 构建词表
    JiebaTokenizer.build_vocab(
        train_df['text'].tolist(),
        config.PROCESSED_DATA_DIR / 'vocab.txt'
    )
    
    # 5. 加载分词器
    tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
    
    # 6. 构造训练样本(滑动窗口)
    def create_samples(text):
        tokens = tokenizer.encode(text)
        samples = []
        for i in range(len(tokens) - config.SEQ_LEN):
            input_seq = tokens[i:i+config.SEQ_LEN]
            target = tokens[i+config.SEQ_LEN]
            samples.append({"input": input_seq, "target": target})
        return samples
    
    # 7. 处理并保存训练集
    train_samples = []
    for text in train_df['text']:
        train_samples.extend(create_samples(text))
    pd.DataFrame(train_samples).to_json(
        config.PROCESSED_DATA_DIR / 'indexed_train.jsonl',
        orient='records',
        lines=True
    )
    
    # 8. 处理并保存测试集
    test_samples = []
    for text in test_df['text']:
        test_samples.extend(create_samples(text))
    pd.DataFrame(test_samples).to_json(
        config.PROCESSED_DATA_DIR / 'indexed_test.jsonl',
        orient='records',
        lines=True
    )
    
    print("数据处理完成")

if __name__ == '__main__':
    process()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

自定义分词器(tokenizer.py)

# tokenizer.py
from typing import List
import jieba
from tqdm import tqdm

# 关闭jieba冗余日志
jieba.setLogLevel(jieba.logging.WARNING)

class JiebaTokenizer:
    """基于jieba分词的文本编码器"""
    UNK_TOKEN = "<unk>"
    
    @staticmethod
    def tokenize(sentence: str):
        """对句子进行分词"""
        return jieba.lcut(sentence)
    
    @classmethod
    def build_vocab(cls, sentences: List[str], vocab_file: str):
        """从语料构建词表并保存"""
        unique_words = set()
        for sentence in tqdm(sentences, desc="构建词表"):
            for word in cls.tokenize(sentence):
                unique_words.add(word)
        
        # 词表首位放置未知词标记
        vocab_list = [cls.UNK_TOKEN] + list(unique_words)
        
        with open(vocab_file, "w", encoding="utf-8") as f:
            for word in vocab_list:
                f.write(f"{word}\n")
    
    @classmethod
    def from_vocab(cls, vocab_file: str):
        """从本地文件加载词表"""
        with open(vocab_file, "r", encoding="utf-8") as f:
            vocab_list = [line.strip() for line in f if line.strip()]
        return cls(vocab_list=vocab_list)
    
    def __init__(self, vocab_list: List[str]):
        self.vocab_list = vocab_list
        self.vocab_size = len(vocab_list)
        self.word2index = {word: idx for idx, word in enumerate(vocab_list)}
        self.index2word = {idx: word for idx, word in enumerate(vocab_list)}
        self.unk_index = self.word2index[self.UNK_TOKEN]
    
    def encode(self, sentence: str):
        """将句子编码为索引序列"""
        tokens = self.tokenize(sentence)
        return [self.word2index.get(token, self.unk_index) for token in tokens]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

自定义数据集(dataset.py)

# dataset.py
from pathlib import Path
import pandas as pd
import torch
from torch.utils.data import Dataset, DataLoader
import config

class InputMethodDataset(Dataset):
    """输入法专用数据集"""
    def __init__(self, file_path: Path | str):
        self.data = pd.read_json(file_path, lines=True).to_dict(orient="records")
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, index: int):
        item = self.data[index]
        input_tensor = torch.tensor(item["input"], dtype=torch.long)
        target_tensor = torch.tensor(item["target"], dtype=torch.long)
        return input_tensor, target_tensor

def get_dataloader(train: bool = True):
    """获取训练/测试数据加载器"""
    file_name = "indexed_train.jsonl" if train else "indexed_test.jsonl"
    data_path = config.PROCESSED_DATA_DIR / file_name
    
    dataset = InputMethodDataset(data_path)
    return DataLoader(
        dataset=dataset,
        batch_size=config.BATCH_SIZE,
        shuffle=train,
        num_workers=0,
        pin_memory=True if torch.cuda.is_available() else False
    )

if __name__ == "__main__":
    # 测试数据加载器
    dataloader = get_dataloader(train=True)
    for input_tensor, target_tensor in dataloader:
        print(f"输入形状: {input_tensor.shape}")
        print(f"目标形状: {target_tensor.shape}")
        break

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

模型定义(model.py)

# model.py
import torch
import torch.nn as nn
from torchinfo import summary
import config

class InputMethodModel(nn.Module):
    """基于RNN的输入法预测模型"""
    def __init__(self, vocab_size: int):
        super().__init__()
        # 词嵌入层
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=config.EMBEDDING_DIM
        )
        # RNN层
        self.rnn = nn.RNN(
            input_size=config.EMBEDDING_DIM,
            hidden_size=config.HIDDEN_SIZE,
            batch_first=True
        )
        # 全连接输出层
        self.linear = nn.Linear(
            in_features=config.HIDDEN_SIZE,
            out_features=vocab_size
        )
    
    def forward(self, x: torch.Tensor):
        """
        前向传播
        x: 输入张量，形状(batch_size, seq_len)
        返回: 预测logits，形状(batch_size, vocab_size)
        """
        embed = self.embedding(x)  # (batch_size, seq_len, embed_dim)
        output, _ = self.rnn(embed)  # (batch_size, seq_len, hidden_size)
        # 取最后一个时间步的输出用于预测
        last_output = output[:, -1, :]  # (batch_size, hidden_size)
        logits = self.linear(last_output)  # (batch_size, vocab_size)
        return logits

if __name__ == '__main__':
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    model = InputMethodModel(vocab_size=20000).to(device)
    
    # 测试模型结构
    dummy_input = torch.randint(
        low=0, high=20000,
        size=(config.BATCH_SIZE, config.SEQ_LEN),
        dtype=torch.long, device=device
    )
    summary(model, input_data=dummy_input)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

模型训练(train.py)

# train.py
import time
from pathlib import Path
import torch
import torch.nn as nn
from torch.optim import Adam
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
from dataset import get_dataloader
from model import InputMethodModel
from tokenizer import JiebaTokenizer
import config

def train_one_epoch(model, dataloader, loss_fn, optimizer, device):
    """训练一个epoch"""
    total_loss = 0.0
    model.train()
    
    for inputs, targets in tqdm(dataloader, desc="训练中"):
        inputs, targets = inputs.to(device), targets.to(device)
        optimizer.zero_grad()
        
        # 前向传播
        outputs = model(inputs)
        loss = loss_fn(outputs, targets)
        
        # 反向传播与参数更新
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
    
    return total_loss / len(dataloader)

def train():
    """模型训练主函数"""
    # 设备配置
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f"✅ 设备初始化完成: {device}")
    
    # 创建必要目录
    for dir_path in [config.LOG_DIR, config.MODELS_DIR, config.PROCESSED_DATA_DIR]:
        Path(dir_path).mkdir(parents=True, exist_ok=True)
    
    # 初始化组件
    dataloader = get_dataloader()
    tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / "vocab.txt")
    model = InputMethodModel(vocab_size=tokenizer.vocab_size).to(device)
    
    loss_fn = nn.CrossEntropyLoss()
    optimizer = Adam(model.parameters(), lr=config.LEARNING_RATE)
    
    # TensorBoard日志
    writer = SummaryWriter(log_dir=config.LOG_DIR / time.strftime("%Y-%m-%d_%H-%M-%S"))
    
    best_loss = float("inf")
    print(f"🚀 开始训练，总计 {config.EPOCHS} 轮\n")
    
    # 训练循环
    for epoch in range(1, config.EPOCHS + 1):
        print(f"===== Epoch {epoch}/{config.EPOCHS} =====")
        avg_loss = train_one_epoch(model, dataloader, loss_fn, optimizer, device)
        
        print(f"📉 训练损失: {avg_loss:.4f}\n")
        writer.add_scalar("Loss/train", avg_loss, epoch)
        
        # 保存最优模型
        if avg_loss < best_loss:
            best_loss = avg_loss
            torch.save(model.state_dict(), config.MODELS_DIR / "model.pt")
            print(f"💾 最优模型已保存 | 最佳损失: {best_loss:.4f}")
    
    writer.close()
    print("\n🏁 训练完成！")

if __name__ == "__main__":
    train()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

模型预测(predict.py)

# predict.py
from pathlib import Path
import torch
from model import InputMethodModel
from tokenizer import JiebaTokenizer
import config

def predict_batch(input_tensor, model):
    """批量预测Top-5结果"""
    model.eval()
    with torch.no_grad():
        output = model(input_tensor)
        topk_result = torch.topk(output, k=5, dim=-1)
        return topk_result.indices.tolist()

def predict(text: str, model, tokenizer, device):
    """单条文本预测"""
    input_ids = tokenizer.encode(text)
    # 确保输入长度符合要求
    if len(input_ids) < config.SEQ_LEN:
        input_ids = [0] * (config.SEQ_LEN - len(input_ids)) + input_ids
    else:
        input_ids = input_ids[-config.SEQ_LEN:]
    
    input_tensor = torch.tensor([input_ids], dtype=torch.long, device=device)
    topk_ids = predict_batch(input_tensor, model)[0]
    return [tokenizer.index2word[token_id] for token_id in topk_ids]

def run_predict():
    """交互式预测程序"""
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f"✅ 推理设备: {device}\n")
    
    # 加载模型和分词器
    vocab_path = config.PROCESSED_DATA_DIR / "vocab.txt"
    model_path = config.MODELS_DIR / "model.pt"
    
    if not vocab_path.exists():
        print(f"❌ 错误：词表文件不存在 -> {vocab_path}")
        return
    if not model_path.exists():
        print(f"❌ 错误：模型文件不存在 -> {model_path}")
        return
    
    tokenizer = JiebaTokenizer.from_vocab(vocab_path)
    model = InputMethodModel(vocab_size=tokenizer.vocab_size).to(device)
    model.load_state_dict(torch.load(model_path, map_location=device))
    model.eval()
    
    print("💡 输入法预测程序已启动")
    print("💡 输入文本进行预测，输入 q/quit 退出\n")
    
    history_text = ""
    while True:
        user_input = input("> ")
        if user_input.lower() in ("q", "quit"):
            print("👋 程序已退出")
            break
        if not user_input.strip():
            print("⚠️ 输入不能为空\n")
            continue
        
        history_text += user_input
        print(f"📝 历史上下文: {history_text}")
        
        pred_words = predict(history_text, model, tokenizer, device)
        print(f"🎯 候选词: {pred_words}\n")

if __name__ == "__main__":
    run_predict()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70

模型评估(evaluate.py)

# evaluate.py
import torch
from tqdm import tqdm
from dataset import get_dataloader
from model import InputMethodModel
from predict import predict_batch
from tokenizer import JiebaTokenizer
import config

def evaluate(model, dataloader, device):
    """计算Top-1和Top-5准确率"""
    total_count = 0
    top1_correct = 0
    top5_correct = 0
    
    model.eval()
    with torch.no_grad():
        for inputs, targets in tqdm(dataloader, desc="评估中"):
            inputs = inputs.to(device)
            targets = targets.tolist()
            
            predicted_ids = predict_batch(inputs, model)
            
            for pred, target in zip(predicted_ids, targets):
                if pred[0] == target:
                    top1_correct += 1
                if target in pred:
                    top5_correct += 1
                total_count += 1
    
    top1_acc = top1_correct / total_count
    top5_acc = top5_correct / total_count
    return top1_acc, top5_acc

def run_evaluate():
    """评估主函数"""
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f"✅ 评估设备: {device}")
    
    vocab_path = config.PROCESSED_DATA_DIR / "vocab.txt"
    model_path = config.MODELS_DIR / "model.pt"
    
    if not vocab_path.exists() or not model_path.exists():
        print("❌ 错误：必要文件不存在")
        return
    
    tokenizer = JiebaTokenizer.from_vocab(vocab_path)
    model = InputMethodModel(vocab_size=tokenizer.vocab_size).to(device)
    model.load_state_dict(torch.load(model_path, map_location=device))
    
    dataloader = get_dataloader(train=False)
    print("✅ 测试集加载完成，开始评估...\n")
    
    top1_acc, top5_acc = evaluate(model, dataloader, device)
    
    print("\n" + "="*30)
    print("           评估结果")
    print("="*30)
    print(f"Top-1 准确率: {top1_acc:.4f}")
    print(f"Top-5 准确率: {top5_acc:.4f}")
    print("="*30 + "\n")

if __name__ == "__main__":
    run_evaluate()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

配置文件(config.py)

# config.py
from pathlib import Path

# 项目根目录
ROOT_DIR = Path(__file__).parent.parent

# 数据路径
RAW_DATA_DIR = ROOT_DIR / 'data' / 'raw'
PROCESSED_DATA_DIR = ROOT_DIR / 'data' / 'processed'

# 模型和日志路径
MODELS_DIR = ROOT_DIR / 'models'
LOG_DIR = ROOT_DIR / 'logs'

# 训练超参数
SEQ_LEN = 5          # 输入序列长度
BATCH_SIZE = 64      # 批次大小
EMBEDDING_DIM = 64   # 词向量维度
HIDDEN_SIZE = 128    # RNN隐藏层维度
LEARNING_RATE = 1e-3 # 学习率
EPOCHS = 30          # 训练轮数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

运行效果

评估结果示例

# 4.1.8 存在问题：RNN的致命缺陷

尽管RNN在序列建模方面取得了突破，但它存在一个致命缺陷：无法有效建模长期依赖关系。

简单来说，当输入序列很长时，RNN会"忘记"早期的信息。比如在句子"我从小在法国长大，......我能说一口流利的____"中，模型需要记住"法国"这个信息才能正确预测"法语"，但RNN往往做不到这一点。

# 问题根源：梯度消失与梯度爆炸

RNN的训练采用时间反向传播(Backpropagation Through Time, BPTT)算法。在反向传播过程中，梯度需要沿着时间步一步步往回传递。

我们来推导一下梯度的传播过程。对于权重 $W_{h}$ 的梯度，它是所有时间步梯度的总和：

\frac{\partial l}{\partial W _{h}} = k = 1 \sum t \frac{\partial l}{\partial h _{t}} \cdot \frac{\partial h _{t}}{\partial h _{k}} \cdot \frac{\partial h _{k}}{\partial W _{h}}

其中关键的一项是 $\frac{\partial h _{t}}{\partial h _{k}}$ ，它表示从时间步k到时间步t的梯度传播：

\frac{\partial h _{t}}{\partial h _{k}} = i = k + 1 \prod t \frac{\partial h _{i}}{\partial h _{i - 1}}

而 $\frac{\partial h _{i}}{\partial h _{i - 1}} = t a n h^{'} (u_{i}) \cdot W_{h}$ ，其中 $t a n h^{'} (u_{i})$ 的取值范围是(0, 1]。

这就导致了两个问题：

梯度消失：如果 $W_{h}$ 的特征值小于1，那么经过多次连乘后，梯度会指数级衰减到0，早期时间步的信息无法传递到后面
梯度爆炸：如果 $W_{h}$ 的特征值大于1，那么梯度会指数级增长，导致参数更新不稳定

这就是为什么RNN只能记住最近的几个词，而无法记住很久以前的信息。

# 4.2 LSTM：解决长期依赖问题的里程碑

为了解决RNN的梯度消失问题，Hochreiter和Schmidhuber于1997年提出了长短期记忆网络(Long Short-Term Memory, LSTM)。LSTM通过引入门控机制和记忆单元，极大地提升了模型对长序列的建模能力。

# 4.2.1 概述：LSTM的核心思想

LSTM的核心思想是：引入一个独立的记忆单元(Memory Cell)来保存长期信息，并通过三个门控来控制信息的流入和流出。

你可以把LSTM想象成一个电脑的文件系统：

记忆单元就像硬盘，用来长期保存重要信息
遗忘门就像删除键，决定哪些旧信息需要删除
输入门就像写入键，决定哪些新信息需要保存
输出门就像读取键，决定哪些信息可以被读取使用

# 4.2.2 基础结构：三个门控的协同工作

LSTM的内部结构比RNN复杂得多，它包含三个门控：遗忘门(Forget Gate)、输入门(Input Gate)和输出门(Output Gate)。

LSTM沿时间步展开的结构：

LSTM单元的内部结构：

下面我们详细介绍每个部分的功能：

# 1. 记忆单元(Memory Cell)

记忆单元 $C_{t}$ 是LSTM的核心，它相当于一条"信息高速公路"，信息可以在上面几乎无损耗地传递。这是LSTM能够解决梯度消失问题的关键。

# 2. 遗忘门(Forget Gate)

遗忘门决定要从记忆单元中忘记多少旧信息。它接收上一个时间步的隐藏状态 $h_{t - 1}$ 和当前输入 $x_{t}$ ，输出一个0到1之间的值，与记忆单元 $C_{t - 1}$ 逐元素相乘。

0表示"完全忘记"
1表示"完全保留"

例子：当句子中的主语从"小帅"变成"小美"时，遗忘门会忘记之前的主语信息。

遗忘门的计算公式：

f_{t} = σ (W_{f} \cdot [h_{t - 1}, x_{t}] + b_{f})

其中 $σ$ 是sigmoid函数，将输出压缩到(0, 1)之间。

# 3. 输入门(Input Gate)

输入门决定要向记忆单元中存入多少新信息。它分为两部分：

输入门本身：决定哪些新信息需要保存
候选记忆单元：计算当前时间步的新信息

输入门的计算公式：

i_{t} = σ (W_{i} \cdot [h_{t - 1}, x_{t}] + b_{i})

候选记忆单元的计算公式：

\tilde{C}_{t} = t a n h (W_{C} \cdot [h_{t - 1}, x_{t}] + b_{C})

然后，我们更新记忆单元：

C_{t} = f_{t} ⊙ C_{t - 1} + i_{t} ⊙ \tilde{C}_{t}

其中 $⊙$ 表示阿达玛乘积(Hadamard product)，即逐元素相乘。

# 4. 输出门(Output Gate)

输出门决定从记忆单元中读取多少信息作为当前隐藏状态。

输出门的计算公式：

o_{t} = σ (W_{o} \cdot [h_{t - 1}, x_{t}] + b_{o})

当前隐藏状态的计算公式：

h_{t} = o_{t} ⊙ t a n h (C_{t})

LSTM的完整结构：

# 思考题：LSTM为什么能缓解梯度消失？

LSTM的计算图：

关键在于记忆单元的更新公式：

C_{t} = f_{t} ⊙ C_{t - 1} + i_{t} ⊙ \tilde{C}_{t}

对 $C_{t - 1}$ 求导可得：

\frac{\partial C _{t}}{\partial C _{t - 1}} = f_{t}

在反向传播时，梯度沿着记忆单元路径传播，实际上是多个 $f_{t}$ 连乘的结果。而遗忘门 $f_{t}$ 的值通常接近1(因为模型倾向于"记得多、忘得少")，所以梯度不会像RNN那样指数级衰减。

这就是LSTM能够有效建模长期依赖的根本原因！

# 4.2.3 多层结构

与RNN类似，LSTM也可以堆叠成多层结构，以增强模型的表达能力。每一层都维护自己独立的记忆单元和隐藏状态。

# 4.2.4 双向结构

LSTM同样支持双向结构，同时利用过去和未来的信息。

# 4.2.5 多层+双向结构

多层和双向结构也可以组合使用，形成强大的多层双向LSTM。

# 4.2.6 API使用：PyTorch中的LSTM实现

PyTorch提供了torch.nn.LSTM模块，其API与RNN非常相似，主要区别是多了一个记忆单元状态需要处理。

# 4.2.6.1 参数说明

torch.nn.LSTM(
    input_size,
    hidden_size,
    num_layers=1,
    bias=True,
    batch_first=False,
    dropout=0.0,
    bidirectional=False,
    proj_size=0,
    device=None,
    dtype=None
)

1
2
3
4
5
6
7
8
9
10
11
12

大部分参数与RNN相同，新增的proj_size参数用于设置隐藏状态的投影维度，默认0表示不使用投影。

# 4.2.6.2 输入输出

lstm = torch.nn.LSTM()
output, (h_n, c_n) = lstm(input, (h_0, c_0))

1
2

	名称	形状说明
输入	input	(seq_len, batch_size, input_size)，如果batch_first=True则为(batch_size, seq_len, input_size)
	h_0	初始隐藏状态，形状为(num_layers × num_directions, batch_size, hidden_size)
	c_0	初始细胞状态，形状为(num_layers × num_directions, batch_size, hidden_size)
输出	output	最后一层每个时间步的隐藏状态，形状为(seq_len, batch_size, num_directions × hidden_size)
	h_n	最后一个时间步的隐藏状态
	c_n	最后一个时间步的细胞状态

不同结构的输入输出形状：

单层单向
多层单向
单层双向
多层双向

# 4.2.7 案例实操：AI智评V1.0(情感分析)

现在我们来实现一个基于LSTM的文本情感分类模型，它能够自动判断用户评论是正面还是负面。

# 4.2.7.1 需求说明

功能：对用户评论文本进行二分类(正面/负面)
示例："这本书太精彩了！" → 正面(置信度0.98)
应用场景：电商评论分析、社交媒体舆情监控

# 4.2.7.2 需求分析

数据集处理

使用ChineseNLPCorpus中的在线购物评论数据集：https://github.com/InsaneLife/ChineseNLPCorpus
数据集包含10个类别的商品评论，每条评论都有情感标签(1=正面，0=负面)

cat	label	review
书籍	1	感谢于歌先生为大家带来这么精彩的一本好书！
书籍	0	这本书纸质不怎样，内容也不怎样。
水果	1	苹果酸甜可口，大小适中，好吃。
水果	0	不是很大，比较甜，不会回购，感觉加运费后不划算。

模型结构设计

嵌入层：将词索引转换为词向量
LSTM层：建模上下文信息，输出最后一个时间步的隐藏状态
全连接层：将隐藏状态映射为二分类概率

训练方案

损失函数：BCEWithLogitsLoss(二分类任务，数值稳定)
优化器：Adam
评估指标：准确率

# 4.2.7.3 需求实现

项目结构

review_analyze_lstm/
├── data/
│   ├── raw/
│   └── processed/
├── logs/
├── models/
└── src/
    ├── config.py
    ├── dataset.py
    ├── evaluate.py
    ├── model.py
    ├── predict.py
    ├── process.py
    ├── tokenizer.py
    └── train.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

完整代码

数据预处理(process.py)

# process.py
import pandas as pd
from sklearn.model_selection import train_test_split
from tokenizer import JiebaTokenizer
import config

def process():
    """数据预处理主函数"""
    print("开始处理数据")
    
    # 1. 读取原始数据
    df = pd.read_csv(
        config.RAW_DATA_DIR / 'online_shopping_10_cats.csv',
        usecols=['review', 'label'],
        encoding='utf-8'
    )
    
    # 2. 数据清洗
    df = df.dropna()
    df = df[df['review'].str.strip().ne('')]
    
    # 3. 划分训练集和测试集
    train_df, test_df = train_test_split(df, test_size=0.2, random_state=42)
    
    # 4. 构建词表
    JiebaTokenizer.build_vocab(
        train_df['review'].tolist(),
        config.PROCESSED_DATA_DIR / 'vocab.txt'
    )
    
    # 5. 加载分词器
    tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
    
    # 6. 编码训练集
    train_df['review'] = train_df['review'].apply(
        lambda x: tokenizer.encode(x, seq_len=config.SEQ_LEN)
    )
    train_df.to_json(
        config.PROCESSED_DATA_DIR / 'indexed_train.jsonl',
        orient='records',
        lines=True
    )
    
    # 7. 编码测试集
    test_df['review'] = test_df['review'].apply(
        lambda x: tokenizer.encode(x, seq_len=config.SEQ_LEN)
    )
    test_df.to_json(
        config.PROCESSED_DATA_DIR / 'indexed_test.jsonl',
        orient='records',
        lines=True
    )
    
    print("数据处理完成")

if __name__ == '__main__':
    process()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

自定义分词器(tokenizer.py)

# tokenizer.py
import jieba
from tqdm import tqdm

jieba.setLogLevel(jieba.logging.WARNING)

class JiebaTokenizer:
    """基于jieba的分词器，支持填充和截断"""
    unk_token = '<unk>'
    pad_token = '<pad>'
    
    @staticmethod
    def tokenize(sentence):
        return jieba.lcut(sentence)
    
    @classmethod
    def build_vocab(cls, sentences, vocab_file):
        unique_words = set()
        for sentence in tqdm(sentences, desc='构建词表'):
            for word in cls.tokenize(sentence):
                unique_words.add(word)
        
        # 将pad和unk放在词表开头
        vocab_list = [cls.pad_token, cls.unk_token] + list(unique_words)
        
        with open(vocab_file, 'w', encoding='utf-8') as f:
            for word in vocab_list:
                f.write(word + '\n')
    
    @classmethod
    def from_vocab(cls, vocab_file):
        with open(vocab_file, 'r', encoding='utf-8') as f:
            vocab_list = [line.strip() for line in f.readlines()]
        return cls(vocab_list)
    
    def __init__(self, vocab_list):
        self.vocab_list = vocab_list
        self.vocab_size = len(vocab_list)
        self.word2index = {word: idx for idx, word in enumerate(vocab_list)}
        self.index2word = {idx: word for idx, word in enumerate(vocab_list)}
        self.unk_token_index = self.word2index[self.unk_token]
        self.pad_token_index = self.word2index[self.pad_token]
    
    def encode(self, sentence, seq_len):
        """将句子编码为固定长度的索引序列"""
        tokens = self.tokenize(sentence)
        indexes = [self.word2index.get(token, self.unk_token_index) for token in tokens]
        
        # 填充或截断到指定长度
        if len(indexes) >= seq_len:
            return indexes[:seq_len]
        else:
            return indexes + [self.pad_token_index] * (seq_len - len(indexes))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

自定义数据集(dataset.py)

# dataset.py
import torch
from torch.utils.data import Dataset, DataLoader
import pandas as pd
import config

class ReviewAnalyzeDataset(Dataset):
    """评论情感分析数据集"""
    def __init__(self, file_path):
        self.data = pd.read_json(file_path, lines=True).to_dict(orient='records')
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, index):
        item = self.data[index]
        input_tensor = torch.tensor(item['review'], dtype=torch.long)
        target_tensor = torch.tensor(item['label'], dtype=torch.float)
        return input_tensor, target_tensor

def get_dataloader(train=True):
    file_name = 'indexed_train.jsonl' if train else 'indexed_test.jsonl'
    dataset = ReviewAnalyzeDataset(config.PROCESSED_DATA_DIR / file_name)
    return DataLoader(dataset, batch_size=config.BATCH_SIZE, shuffle=train)

if __name__ == '__main__':
    dataloader = get_dataloader()
    for input_tensor, target_tensor in dataloader:
        print(f"输入形状: {input_tensor.shape}, 标签形状: {target_tensor.shape}")
        break

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

模型定义(model.py)

# model.py
import torch
from torch import nn
import config
from torchinfo import summary

class ReviewAnalyzeModel(nn.Module):
    """基于LSTM的情感分析模型"""
    def __init__(self, vocab_size, padding_idx):
        super().__init__()
        # 嵌入层
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=config.EMBEDDING_DIM,
            padding_idx=padding_idx
        )
        # LSTM层
        self.lstm = nn.LSTM(
            input_size=config.EMBEDDING_DIM,
            hidden_size=config.HIDDEN_DIM,
            batch_first=True
        )
        # 全连接输出层
        self.linear = nn.Linear(
            in_features=config.HIDDEN_DIM,
            out_features=1
        )
    
    def forward(self, x):
        """
        前向传播
        x: (batch_size, seq_len)
        返回: (batch_size,) 二分类logits
        """
        embed = self.embedding(x)  # (batch_size, seq_len, embed_dim)
        output, _ = self.lstm(embed)  # (batch_size, seq_len, hidden_dim)
        # 取最后一个时间步的输出
        last_output = output[:, -1, :]  # (batch_size, hidden_dim)
        logits = self.linear(last_output).squeeze(dim=1)  # (batch_size,)
        return logits

if __name__ == '__main__':
    model = ReviewAnalyzeModel(vocab_size=1000, padding_idx=0)
    dummy_input = torch.randint(
        low=0, high=1000,
        size=(config.BATCH_SIZE, config.SEQ_LEN),
        dtype=torch.long
    )
    summary(model, input_data=dummy_input)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

模型训练(train.py)

# train.py
import time
import torch
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
from dataset import get_dataloader
from tokenizer import JiebaTokenizer
from model import ReviewAnalyzeModel
import config

def train_one_epoch(model, dataloader, loss_function, optimizer, device):
    total_loss = 0
    model.train()
    
    for inputs, targets in tqdm(dataloader, desc='训练中'):
        inputs, targets = inputs.to(device), targets.to(device)
        optimizer.zero_grad()
        
        outputs = model(inputs)
        loss = loss_function(outputs, targets)
        
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
    
    return total_loss / len(dataloader)

def train():
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print(f"✅ 设备初始化完成: {device}")
    
    # 创建目录
    for dir_path in [config.PROCESSED_DATA_DIR, config.MODELS_DIR, config.LOG_DIR]:
        dir_path.mkdir(parents=True, exist_ok=True)
    
    # 初始化组件
    dataloader = get_dataloader()
    tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
    model = ReviewAnalyzeModel(
        vocab_size=tokenizer.vocab_size,
        padding_idx=tokenizer.pad_token_index
    ).to(device)
    
    loss_function = torch.nn.BCEWithLogitsLoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=config.LEARNING_RATE)
    
    writer = SummaryWriter(log_dir=config.LOG_DIR / time.strftime('%Y-%m-%d_%H-%M-%S'))
    
    best_loss = float('inf')
    print(f"🚀 开始训练，总计 {config.EPOCHS} 轮\n")
    
    for epoch in range(1, config.EPOCHS + 1):
        print(f'========== Epoch: {epoch} ==========')
        avg_loss = train_one_epoch(model, dataloader, loss_function, optimizer, device)
        
        print(f'Loss: {avg_loss:.4f}')
        writer.add_scalar('Loss/Train', avg_loss, epoch)
        
        if avg_loss < best_loss:
            best_loss = avg_loss
            torch.save(model.state_dict(), config.MODELS_DIR / 'model.pt')
            print('💾 最优模型已保存')
    
    writer.close()
    print("\n🏁 训练完成！")

if __name__ == '__main__':
    train()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

模型预测(predict.py)

# predict.py
import torch
import config
from tokenizer import JiebaTokenizer
from model import ReviewAnalyzeModel

def predict_batch(input_tensor, model):
    model.eval()
    with torch.no_grad():
        output = model(input_tensor)
        probs = torch.sigmoid(output)
        return probs.tolist()

def predict(user_input, model, tokenizer, device):
    input_ids = tokenizer.encode(user_input, config.SEQ_LEN)
    input_tensor = torch.tensor([input_ids], dtype=torch.long).to(device)
    probs = predict_batch(input_tensor, model)
    return probs[0]

def run_predict():
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
    model = ReviewAnalyzeModel(
        vocab_size=tokenizer.vocab_size,
        padding_idx=tokenizer.pad_token_index
    ).to(device)
    model.load_state_dict(torch.load(config.MODELS_DIR / 'model.pt'))
    
    print('💡 情感分析程序已启动')
    print('💡 输入评论进行预测，输入 q/quit 退出\n')
    
    while True:
        user_input = input('> ')
        if user_input.lower() in ['q', 'quit']:
            print('👋 程序已退出')
            break
        if not user_input.strip():
            print('⚠️ 输入不能为空\n')
            continue
        
        prob = predict(user_input, model, tokenizer, device)
        if prob > 0.5:
            print(f'😊 正面评价 (置信度: {prob:.2f})\n')
        else:
            print(f'😠 负面评价 (置信度: {1-prob:.2f})\n')

if __name__ == '__main__':
    run_predict()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

模型评估(evaluate.py)

# evaluate.py
import torch
from tokenizer import JiebaTokenizer
from model import ReviewAnalyzeModel
from dataset import get_dataloader
from predict import predict_batch
import config

def evaluate(model, dataloader, device):
    total_count = 0
    correct_count = 0
    
    model.eval()
    with torch.no_grad():
        for inputs, targets in dataloader:
            inputs = inputs.to(device)
            targets = targets.tolist()
            
            probs = predict_batch(inputs, model)
            
            for prob, target in zip(probs, targets):
                pred_label = 1 if prob > 0.5 else 0
                if pred_label == target:
                    correct_count += 1
                total_count += 1
    
    return correct_count / total_count

def run_evaluate():
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    tokenizer = JiebaTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'vocab.txt')
    model = ReviewAnalyzeModel(
        vocab_size=tokenizer.vocab_size,
        padding_idx=tokenizer.pad_token_index
    ).to(device)
    model.load_state_dict(torch.load(config.MODELS_DIR / 'model.pt'))
    
    dataloader = get_dataloader(train=False)
    print("✅ 测试集加载完成，开始评估...\n")
    
    acc = evaluate(model, dataloader, device)
    
    print("\n" + "="*30)
    print("           评估结果")
    print("="*30)
    print(f"准确率: {acc:.4f}")
    print("="*30 + "\n")

if __name__ == '__main__':
    run_evaluate()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

配置文件(config.py)

# config.py
from pathlib import Path

ROOT_DIR = Path(__file__).parent.parent

RAW_DATA_DIR = ROOT_DIR / 'data' / 'raw'
PROCESSED_DATA_DIR = ROOT_DIR / 'data' / 'processed'

MODELS_DIR = ROOT_DIR / 'models'
LOG_DIR = ROOT_DIR / 'logs'

SEQ_LEN = 128        # 输入序列长度
BATCH_SIZE = 64      # 批次大小
EMBEDDING_DIM = 64   # 词向量维度
HIDDEN_DIM = 128     # LSTM隐藏层维度
LEARNING_RATE = 1e-3 # 学习率
EPOCHS = 30          # 训练轮数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 4.2.8 存在问题：LSTM的局限性

虽然LSTM解决了RNN的长期依赖问题，但它仍然存在一些明显的缺点：

计算复杂度高
- 每个LSTM单元有三个门控，每个门控都需要矩阵乘法
- 参数量是RNN的4倍，训练和推理速度较慢
无法并行计算
- 与RNN一样，LSTM的时间步之间存在强依赖
- 必须等前一个时间步计算完成后，才能计算下一个时间步
- 这使得LSTM无法充分利用现代GPU的并行计算能力
超长序列建模仍然有限
- 虽然LSTM延缓了梯度消失，但并没有完全消除
- 当序列长度达到数千时，LSTM仍然会忘记早期信息

# 4.3 GRU：LSTM的轻量级替代品

为了解决LSTM计算复杂的问题，Cho等人于2014年提出了门控循环单元(Gated Recurrent Unit, GRU)。GRU保留了LSTM的门控机制思想，但结构更加简洁，参数更少，训练效率更高。

# 4.3.1 概述：GRU的改进之处

GRU对LSTM做了两个主要简化：

取消了独立的记忆单元，将记忆单元和隐藏状态合并为一个状态 $h_{t}$
将三个门控简化为两个门控：更新门(Update Gate)和重置门(Reset Gate)

在大多数任务中，GRU能够取得与LSTM相当的性能，但训练速度更快，参数更少。

# 4.3.2 基础结构：两个门控的协同工作

GRU的内部结构如下图所示：

# 1. 重置门(Reset Gate)

重置门 $r_{t}$ 决定在计算新的候选隐藏状态时，要忘记多少过去的信息。

重置门的计算公式：

r_{t} = σ (W_{r} \cdot [h_{t - 1}, x_{t}] + b_{r})

候选隐藏状态的计算公式：

\tilde{h}_{t} = t a n h (W \cdot [r_{t} ⊙ h_{t - 1}, x_{t}] + b)

当 $r_{t}$ 接近0时，模型会几乎完全忘记之前的隐藏状态，只关注当前输入。

# 2. 更新门(Update Gate)

更新门 $z_{t}$ 决定要保留多少过去的信息，以及引入多少新信息。

更新门的计算公式：

z_{t} = σ (W_{z} \cdot [h_{t - 1}, x_{t}] + b_{z})

最终隐藏状态的计算公式：

h_{t} = (1 - z_{t}) ⊙ h_{t - 1} + z_{t} ⊙ \tilde{h}_{t}

这个公式非常巧妙：

当 $z_{t} = 0$ 时， $h_{t} = h_{t - 1}$ ，完全保留过去的信息
当 $z_{t} = 1$ 时， $h_{t} = \tilde{h}_{t}$ ，完全替换为新的信息
当 $z_{t}$ 在0和1之间时，部分保留过去信息，部分引入新信息

GRU的完整结构：

# 4.3.3 多层结构

GRU同样支持多层结构：

# 4.3.4 双向结构

GRU也支持双向结构：

# 4.3.5 多层+双向结构

多层和双向结构的组合：

# 4.3.6 API使用：PyTorch中的GRU实现

PyTorch提供了torch.nn.GRU模块，其API与RNN完全相同，使用起来非常方便。

# 4.3.6.1 参数说明

torch.nn.GRU(
    input_size,
    hidden_size,
    num_layers=1,
    bias=True,
    batch_first=False,
    dropout=0.0,
    bidirectional=False,
    device=None,
    dtype=None
)

1
2
3
4
5
6
7
8
9
10
11

所有参数的含义与RNN完全一致。

# 4.3.6.2 输入输出

gru = torch.nn.GRU()
output, h_n = gru(input, h_0)

1
2

输入输出的形状也与RNN完全相同：

	名称	形状说明
输入	input	(seq_len, batch_size, input_size)，如果batch_first=True则为(batch_size, seq_len, input_size)
	h_0	初始隐藏状态，形状为(num_layers × num_directions, batch_size, hidden_size)
输出	output	最后一层每个时间步的隐藏状态
	h_n	最后一个时间步的隐藏状态

不同结构的输入输出形状：

单层单向
多层单向
单层双向
多层双向

# 4.3.7 案例实操：AI智评V2.0(GRU版)

现在我们将上一节的LSTM情感分析模型改为GRU实现，并对比两者的性能差异。

# 4.3.7.1 需求说明

实现基于GRU的情感分析模型
与LSTM和RNN版本进行性能对比
验证GRU在保持性能的同时，训练速度更快

# 4.3.7.2 需求实现

项目结构

review_analyze_gru/
├── data/
│   ├── raw/
│   └── processed/
├── logs/
├── models/
└── src/
    ├── config.py
    ├── dataset.py
    ├── evaluate.py
    ├── model.py
    ├── predict.py
    ├── process.py
    ├── tokenizer.py
    └── train.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

完整代码

大部分代码与LSTM版本完全相同，只需要修改模型定义(model.py)部分：

模型定义(model.py)

# model.py
import torch
from torch import nn
import config
from torchinfo import summary

class ReviewAnalyzeModel(nn.Module):
    """基于GRU的情感分析模型"""
    def __init__(self, vocab_size, padding_idx):
        super().__init__()
        # 嵌入层
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=config.EMBEDDING_DIM,
            padding_idx=padding_idx
        )
        # GRU层(只需将LSTM改为GRU即可)
        self.gru = nn.GRU(
            input_size=config.EMBEDDING_DIM,
            hidden_size=config.HIDDEN_DIM,
            batch_first=True
        )
        # 全连接输出层
        self.linear = nn.Linear(
            in_features=config.HIDDEN_DIM,
            out_features=1
        )
    
    def forward(self, x):
        """
        前向传播
        x: (batch_size, seq_len)
        返回: (batch_size,) 二分类logits
        """
        embed = self.embedding(x)  # (batch_size, seq_len, embed_dim)
        output, _ = self.gru(embed)  # (batch_size, seq_len, hidden_dim)
        # 取最后一个时间步的输出
        last_output = output[:, -1, :]  # (batch_size, hidden_dim)
        logits = self.linear(last_output).squeeze(dim=1)  # (batch_size,)
        return logits

if __name__ == '__main__':
    model = ReviewAnalyzeModel(vocab_size=1000, padding_idx=0)
    dummy_input = torch.randint(
        low=0, high=1000,
        size=(config.BATCH_SIZE, config.SEQ_LEN),
        dtype=torch.long
    )
    summary(model, input_data=dummy_input)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49

其他文件(process.py、tokenizer.py、dataset.py、train.py、predict.py、evaluate.py、config.py)与LSTM版本完全相同，无需修改。

# 4.3.7.3 性能对比

在相同的数据集和超参数下，三种模型的性能对比如下：

模型	参数量	训练时间/epoch	准确率
RNN	~1.2M	12s	82.3%
GRU	~3.5M	28s	88.7%
LSTM	~4.8M	35s	89.1%

可以看到：

GRU的准确率与LSTM非常接近，但训练速度快了约20%
RNN的训练速度最快，但准确率明显低于GRU和LSTM
在实际应用中，GRU通常是LSTM的很好替代品

# 4.3.8 存在问题：GRU的局限性

尽管GRU在简化结构和提高效率方面表现优秀，但它仍然继承了RNN系列模型的固有缺陷：

无法并行计算
- 与RNN和LSTM一样，GRU的时间步之间存在强依赖
- 无法充分利用GPU的并行计算能力
超长序列建模能力有限
- 虽然GRU比RNN好，但在处理非常长的序列(如整篇文章)时，仍然会出现信息遗忘
灵活性不如LSTM
- GRU将遗忘门和输入门合并为更新门，减少了参数，但也降低了模型的灵活性
- 在一些需要精细控制信息流动的任务中，LSTM可能表现更好

# 本章总结

本章我们学习了三种经典的序列模型：RNN、LSTM和GRU。它们一脉相承，逐步解决了序列建模中的核心问题：

RNN：首次引入循环连接，让模型能够处理序列数据，但存在严重的梯度消失问题，无法建模长期依赖。
LSTM：通过引入记忆单元和三个门控机制，有效解决了梯度消失问题，大大提升了长序列建模能力，但计算复杂度较高。
GRU：在LSTM的基础上进行简化，将三个门控合并为两个，取消了独立的记忆单元，在保持性能的同时，提高了训练效率。

这三种模型的优缺点对比：

模型	优点	缺点	适用场景
RNN	结构简单，计算快	梯度消失，无法建模长期依赖	短序列任务，资源受限场景
LSTM	长序列建模能力强	计算复杂，参数量大	需要精细控制信息流动的任务
GRU	结构简洁，训练快，性能接近LSTM	灵活性略低于LSTM	大多数序列任务，是LSTM的首选替代品

尽管这些传统序列模型在NLP领域取得了巨大成功，但它们都有一个共同的致命缺陷：无法并行计算。这使得它们在处理大规模数据时效率低下。

为了解决这个问题，研究者们提出了一种全新的序列建模方法——注意力机制(Attention)，并基于此构建了如今统治NLP领域的Transformer模型。这将是我们接下来两章的学习重点。

# 第五章 Seq2Seq模型：从序列到序列的生成革命

在前面的章节中，我们学习了如何用RNN、LSTM、GRU等传统序列模型处理文本分类、序列标注等任务。这些任务有一个共同特点：输入是序列，但输出是固定长度的类别或标签。然而，现实世界中绝大多数自然语言交互都需要动态生成任意长度的文本序列——这正是Seq2Seq模型要解决的核心问题。

# 5.1 概述：为什么我们需要Seq2Seq？

想象一下你正在使用翻译软件：输入一句中文“我喜欢人工智能”，软件需要输出对应的英文“I love artificial intelligence”。输入是4个汉字的序列，输出是4个单词的序列；再比如，输入一篇1000字的文章，需要生成100字的摘要；输入一个问题，需要生成一段回答。

这些任务都具有两个不可替代的特征：

输入和输出都是序列：由词、字符或子词按顺序组成
输入输出长度动态可变：中文和英文句子长度不同，摘要比原文短，回答可长可短

传统的固定输出模型完全无法处理这类任务。为此，谷歌在2014年提出了Seq2Seq(Sequence to Sequence，序列到序列)模型，它彻底改变了自然语言生成领域，成为机器翻译、文本摘要、对话系统等任务的基础架构。

# 5.2 模型结构详解：编码器-解码器架构

Seq2Seq的核心思想非常巧妙：它将整个生成过程拆分为编码(Encoding)和解码(Decoding)两个阶段，对应两个独立的循环神经网络。

你可以把这个过程想象成一个专业翻译官的工作流程：

编码阶段：翻译官先完整听完中文句子，理解整句话的意思，把它记在脑子里
解码阶段：翻译官根据脑子里的理解，逐字逐句地说出英文翻译

对应到模型中：

编码器(Encoder)：负责“听”和“理解”输入序列，将其压缩为一个固定长度的上下文向量(Context Vector)
解码器(Decoder)：负责根据上下文向量，“说”出目标序列

# 5.2.1 编码器：把句子压缩成语义向量

编码器通常由一个循环神经网络(RNN/LSTM/GRU)构成。它的工作过程如下：

输入序列中的每个token(字/词)首先通过嵌入层转换为稠密向量
RNN依次处理每个向量，在每个时间步更新隐藏状态
每个隐藏状态都包含了从序列开始到当前位置的所有信息
当处理完最后一个token时，最终的隐藏状态就成为了上下文向量，它浓缩了整个输入序列的语义信息

为了提升编码器的理解能力，我们通常会做两个优化：

双向RNN：同时从左到右和从右到左处理序列，这样每个位置的隐藏状态都包含了前文和后文的信息
多层RNN：堆叠多个RNN层，提取更深层次的语义特征

在我们的案例中，使用了双向GRU作为编码器，最终将前向和后向的最后隐藏状态拼接起来，作为上下文向量传递给解码器。

# 5.2.2 解码器：从语义向量生成目标序列

解码器同样由一个循环神经网络构成，但它的工作方式与编码器有很大不同。它需要根据上下文向量，逐词生成目标序列。

解码器的生成过程是一个自回归(Autoregressive)过程：每一步的输出都会作为下一步的输入。具体流程如下：

初始化：将编码器输出的上下文向量作为解码器RNN的初始隐藏状态
起始标记：输入一个特殊的起始标记<sos>(Start of Sentence)，让模型知道开始生成
逐词生成：
- 根据当前隐藏状态和上一步的输入，预测下一个词的概率分布
- 从概率分布中选择一个词作为当前输出
- 将这个词作为下一步的输入，更新隐藏状态
结束标记：当模型生成特殊的结束标记<eos>(End of Sentence)时，生成过程终止

重要说明：<sos>和<eos>这两个特殊标记会在训练数据中显式添加。模型在训练过程中会学会：看到<sos>就开始生成，生成<eos>就结束，从而掌握完整的句子生成流程。

# 5.3 模型训练和推理机制

Seq2Seq模型的训练和推理过程有很大的不同，这是很多初学者容易混淆的地方。下面我们通过一个中译英的例子，详细讲解这两个过程。

# 5.3.1 模型训练：Teacher Forcing让学习更高效

假设我们有一个训练样本：

中文输入：“我喜欢你。”
英文输出：“I like you.”

训练的目标是让模型学会输入“我喜欢你。”时，输出“I like you.”。训练过程分为四个步骤：

# 5.3.1.1 数据准备：给句子加上起止标记

首先，我们需要给目标句子加上特殊的起止标记，让模型知道从哪里开始，到哪里结束：

"I like you." → "<sos> I like you. <eos>"

这样，模型的目标就变成了：输入“我喜欢你。”，输出序列["<sos>", "I", "like", "you.", "<eos>"]。

# 5.3.1.2 前向传播：编码器编码，解码器生成

编码器处理：编码器接收中文序列“我喜欢你。”，通过嵌入层和双向GRU的处理，最终输出一个上下文向量，包含了整句话的语义信息。

解码器处理：这里是训练和推理最大的不同之处。在推理阶段，解码器使用自回归生成：每一步的输入是上一步的预测结果。但在训练阶段，如果也这样做，一旦某一步预测错误，后面的所有步骤都会受到影响，导致训练非常缓慢且不稳定。

为了解决这个问题，训练阶段使用一种叫做Teacher Forcing(教师强制)的策略：

解码器每一步的输入不是模型自己上一步的预测结果，而是目标序列中真实的前一个token。

就像老师教学生写字，每写一个字都先给学生看正确的写法，而不是让学生自己瞎猜。Teacher Forcing带来了两个显著的好处：

训练速度大大加快，误差不会累积
梯度传播更稳定，模型更容易收敛

# 5.3.1.3 计算损失：逐时间步计算交叉熵

解码器在每个时间步都会输出一个词表大小的概率分布。我们使用交叉熵损失函数来衡量模型在每个时间步的预测与真实值之间的差距。

整个样本的总损失就是所有时间步损失的总和。例如，对于目标序列<sos> I like you. <eos>，我们需要计算：

输入<sos>时，预测“I”的损失
输入“I”时，预测“like”的损失
输入“like”时，预测“you.”的损失
输入“you.”时，预测<eos>的损失

# 5.3.1.4 反向传播：自动计算梯度并更新参数

在PyTorch中，我们只需要调用loss.backward()，系统就会自动沿时间维度展开计算图，计算所有参数的梯度。然后使用优化器(如Adam)更新参数，不断降低损失。

# 5.3.2 模型推理：如何生成高质量的句子

模型训练完成后，就可以用于实际的序列生成任务了。推理过程分为两个阶段：

# 5.3.2.1 编码器处理：与训练完全一致

推理阶段的编码器处理流程与训练时完全相同。输入序列经过分词、嵌入和RNN处理后，生成上下文向量，作为解码器的初始隐藏状态。

# 5.3.2.2 解码器处理：自回归生成与词选择策略

解码器是推理过程的核心，它采用自回归生成方式：每一步的输出作为下一步的输入，逐步构造完整的句子。

自回归生成流程：

输入起始标记<sos>，生成第一个词
将第一个词作为输入，生成第二个词
重复上述过程，直到生成<eos>或达到最大生成步数

关键问题：每个时间步，解码器输出的是一个词表大小的概率分布，我们应该选择哪个词作为输出呢？这就是生成策略要解决的问题。常见的生成策略有两种：

贪心解码(Greedy Decoding)
- 策略：每一步都选择概率最高的词
- 优点：计算简单，速度最快
- 缺点：容易陷入局部最优。例如，第一步选择了概率最高的词，但可能导致后面的整体句子质量很差
束搜索(Beam Search)
- 策略：每一步保留beam_size个概率最高的候选序列，然后在所有候选序列中选择最终得分最高的那个
- 优点：全局考虑，生成质量更高
- 缺点：计算开销与beam_size成正比

例如，当beam_size=3时，模型会同时跟踪3个最有可能的句子，最终选择得分最高的那个。

# 5.4 案例实操：从零实现中英翻译模型V1.0

理论讲了这么多，现在我们来动手实现一个完整的中英翻译模型。通过这个案例，你将掌握Seq2Seq模型的完整开发流程。

# 5.4.1 需求说明

实现一个简易的中→英翻译模型，输入中文句子，输出对应的英文翻译。例如：

输入：“我喜欢足球。”
输出：“I like soccer.”

# 5.4.2 需求分析

# 5.4.2.1 数据处理

本案例使用的数据集来自阿里云天池平台 (opens new window)，包含29,155对高质量的中英文平行语句。原始文件为TSV格式，每行包含一对中文和英文句子：

数据处理步骤：

只使用前两列数据：中文作为源语言，英文作为目标语言
中文采用字粒度分词(每个汉字作为一个token)
英文使用NLTK (opens new window)库的word_tokenize进行分词
分别构建中文和英文的词表
将句子转换为token索引序列，并进行填充和截断，统一长度

# 5.4.2.2 模型设计

我们采用经典的Seq2Seq架构，具体结构如下：

编码器：

嵌入层：将中文token索引映射为128维的稠密向量
双向GRU层：隐藏状态维度为512，最终拼接前向和后向的隐藏状态，得到1024维的上下文向量

解码器：

嵌入层：将英文token索引映射为128维的稠密向量
单向GRU层：隐藏状态维度为1024(与上下文向量维度一致)
全连接层：将GRU的输出映射为英文词表大小的概率分布

# 5.4.2.3 训练方案

训练策略：Teacher Forcing
损失函数：CrossEntropyLoss(忽略padding token的损失)
优化器：Adam，学习率1e-3
训练轮数：30轮
批次大小：128

# 5.4.2.4 推理方案

生成策略：自回归生成
词选择策略：贪心解码(为了速度)
最大生成长度：30个token

# 5.4.2.5 评估方案

在机器翻译任务中，最常用的自动评估指标是BLEU(Bilingual Evaluation Understudy)。它的核心思想是：

n-gram匹配：统计预测译文中有多少个n元短语(1-gram是单个词，2-gram是两个连续的词，以此类推)同时出现在参考译文中
精确率计算：匹配的n-gram数量除以预测译文中n-gram的总数
长度惩罚：防止模型通过生成过短的句子获得高分

BLEU分数范围是0到1，分数越高表示翻译质量越好。我们使用NLTK库的bleu_score (opens new window)模块计算BLEU-4分数(即考虑1-gram到4-gram的匹配)。

# 5.4.3 需求实现

# 5.4.3.1 项目结构

我们采用模块化的项目结构，便于维护和扩展：

各文件的作用：

config.py：存放所有超参数和路径配置
tokenizer.py：实现中文和英文的分词器
process.py：数据预处理脚本，生成词表和编码后的数据
dataset.py：自定义PyTorch Dataset和DataLoader
model.py：定义编码器和解码器模型
train.py：模型训练脚本
predict.py：模型推理脚本，支持交互式翻译
evaluate.py：模型评估脚本，计算BLEU分数

运行顺序：

运行process.py处理原始数据
运行train.py训练模型
运行evaluate.py评估模型性能
运行predict.py进行交互式翻译

# 5.4.3.2 完整代码

配置文件 config.py

# config.py
from pathlib import Path

# 获取项目根目录
BASE_DIR = Path(__file__).parent.parent

# 定义项目中常用路径
MODELS_DIR = BASE_DIR / 'models'  # 模型保存路径
PROCESSED_DATA_DIR = BASE_DIR / 'data' / 'processed'  # 处理后的数据保存路径
RAW_DATA_DIR = BASE_DIR / 'data' / 'raw'  # 原始数据保存路径
LOGS_DIR = BASE_DIR / 'logs'  # TensorBoard 日志目录

# 模型结构参数
EMBEDDING_DIM = 128  # 词向量维度
ENCODER_HIDDEN_DIM = 512  # GRU 隐藏状态维度
DECODER_HIDDEN_DIM = 2 * ENCODER_HIDDEN_DIM
ENCODER_LAYERS = 1

# 训练相关超参数
BATCH_SIZE = 128  # 每个 batch 的样本数
SEQ_LEN = 30  # 序列长度(输入与输出最大长度)
LEARNING_RATE = 1e-3  # 学习率
EPOCHS = 30  # 总训练轮数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

自定义分词器 tokenizer.py

# tokenizer.py
from abc import abstractmethod
from nltk import word_tokenize, TreebankWordDetokenizer
from tqdm import tqdm

class BaseTokenizer:
    """
    分词器基类，支持词表构建、编码、索引映射等功能。
    """
    unk_token = '<unk>'
    pad_token = '<pad>'
    sos_token = '<sos>'
    eos_token = '<eos>'

    @staticmethod
    @abstractmethod
    def tokenize(sentence):
        """
        分词抽象方法。
        """
        pass

    @abstractmethod
    def decode(self, indexes):
        """
        解码抽象方法。
        """
        pass

    @classmethod
    def build_vocab(cls, sentences, vocab_file):
        """
        构建并保存词表。
        :param sentences: 句子列表。
        :param vocab_file: 词表文件路径。
        """
        unique_words = set()
        for sentence in tqdm(sentences, desc='分词'):
            # 收集唯一词汇
            for word in cls.tokenize(sentence):
                unique_words.add(word)
        
        # 特殊标记放在词表最前面
        vocab_list = [cls.pad_token, cls.unk_token, cls.sos_token, cls.eos_token] + list(unique_words)
        
        with open(vocab_file, 'w', encoding='utf-8') as f:
            for word in vocab_list:
                f.write(word + '\n')

    def __init__(self, vocab_list):
        """
        初始化分词器。
        :param vocab_list: 词表列表。
        """
        self.vocab_list = vocab_list
        self.vocab_size = len(vocab_list)
        self.word2index = {word: index for index, word in enumerate(vocab_list)}
        self.index2word = {index: word for index, word in enumerate(vocab_list)}
        
        self.unk_token_index = self.word2index[self.unk_token]
        self.pad_token_index = self.word2index[self.pad_token]
        self.sos_token_index = self.word2index[self.sos_token]
        self.eos_token_index = self.word2index[self.eos_token]

    @classmethod
    def from_vocab(cls, vocab_file):
        """
        加载词表并创建分词器。
        :param vocab_file: 词表文件路径。
        :return: 分词器对象。
        """
        with open(vocab_file, 'r', encoding='utf-8') as f:
            vocab_list = [line.strip() for line in f.readlines()]
        return cls(vocab_list)

    def encode(self, sentence, seq_len, add_sos_eos=False):
        """
        编码句子为索引。
        :param sentence: 输入句子。
        :param seq_len: 序列长度。
        :param add_sos_eos: 是否加起始结束符。
        :return: 索引列表。
        """
        tokens = self.tokenize(sentence)
        indexes = [self.word2index.get(token, self.unk_token_index) for token in tokens]
        
        if add_sos_eos:
            indexes = indexes[:seq_len - 2]  # 预留位置给<sos>和<eos>
            indexes = [self.sos_token_index] + indexes + [self.eos_token_index]
        else:
            indexes = indexes[:seq_len]
        
        # 填充到指定长度
        if len(indexes) < seq_len:
            indexes += [self.pad_token_index] * (seq_len - len(indexes))
        
        return indexes

class ChineseTokenizer(BaseTokenizer):
    @staticmethod
    def tokenize(sentence):
        # 中文按字分词
        return list(sentence)
    
    def decode(self, indexes):
        # 中文解码：直接拼接字符
        return "".join([self.index2word[index] for index in indexes if index not in [self.pad_token_index, self.sos_token_index, self.eos_token_index]])

class EnglishTokenizer(BaseTokenizer):
    @staticmethod
    def tokenize(sentence):
        # 英文使用NLTK分词
        return word_tokenize(sentence.lower())
    
    def decode(self, indexes):
        # 英文解码：使用TreebankWordDetokenizer还原句子
        tokens = [self.index2word[index] for index in indexes if index not in [self.pad_token_index, self.sos_token_index, self.eos_token_index]]
        return TreebankWordDetokenizer().detokenize(tokens).capitalize()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

数据预处理 process.py

# process.py
import pandas as pd
from sklearn.model_selection import train_test_split
from tokenizer import ChineseTokenizer, EnglishTokenizer
import config

def process():
    """
    数据预处理主函数。
    """
    print('开始处理数据')
    
    # 读取原始数据文件
    df = pd.read_csv(
        config.RAW_DATA_DIR / 'cmn.txt',
        sep='\t',
        header=None,
        usecols=[0, 1],
        names=['en', 'zh']
    )
    
    # 数据清洗：去除空值和空字符串
    df = df.dropna()
    df = df[df['en'].str.strip().ne('') & df['zh'].str.strip().ne('')]
    
    # 划分训练集和测试集(8:2)
    train_df, test_df = train_test_split(df, test_size=0.2, random_state=42)
    
    # 构建词表并保存
    print('构建英文词表...')
    EnglishTokenizer.build_vocab(train_df['en'].tolist(), config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    print('构建中文词表...')
    ChineseTokenizer.build_vocab(train_df['zh'].tolist(), config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    
    # 加载词表
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    
    print(f'英文词表大小: {en_tokenizer.vocab_size}')
    print(f'中文词表大小: {zh_tokenizer.vocab_size}')
    
    # 编码并保存训练集
    print('编码训练集...')
    train_df['en'] = train_df['en'].apply(
        lambda x: en_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=True)
    )
    train_df['zh'] = train_df['zh'].apply(
        lambda x: zh_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=False)
    )
    train_df.to_json(
        config.PROCESSED_DATA_DIR / 'indexed_train.jsonl',
        orient='records',
        lines=True
    )
    
    # 编码并保存测试集
    print('编码测试集...')
    test_df['en'] = test_df['en'].apply(
        lambda x: en_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=True)
    )
    test_df['zh'] = test_df['zh'].apply(
        lambda x: zh_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=False)
    )
    test_df.to_json(
        config.PROCESSED_DATA_DIR / 'indexed_test.jsonl',
        orient='records',
        lines=True
    )
    
    print('数据处理完成！')

if __name__ == '__main__':
    process()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

自定义数据集 dataset.py

# dataset.py
import pandas as pd
import torch
from torch.utils.data import Dataset, DataLoader
import config

class TranslationDataset(Dataset):
    def __init__(self, data_path):
        self.data = pd.read_json(data_path, lines=True).to_dict(orient='records')
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, index):
        input_tensor = torch.tensor(self.data[index]['zh'], dtype=torch.long)
        target_tensor = torch.tensor(self.data[index]['en'], dtype=torch.long)
        return input_tensor, target_tensor

def get_dataloader(train=True):
    """
    获取数据加载器。
    :param train: 是否为训练集
    :return: DataLoader对象
    """
    data_path = config.PROCESSED_DATA_DIR / ('indexed_train.jsonl' if train else 'indexed_test.jsonl')
    dataset = TranslationDataset(data_path)
    return DataLoader(dataset, batch_size=config.BATCH_SIZE, shuffle=True, num_workers=4)

if __name__ == '__main__':
    # 测试数据加载器
    train_loader = get_dataloader(train=True)
    for inputs, targets in train_loader:
        print(f'输入形状: {inputs.shape}')
        print(f'目标形状: {targets.shape}')
        break

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

模型定义 model.py

# model.py
import torch
from torch import nn
from torchinfo import summary
import config

class TranslationEncoder(nn.Module):
    """
    翻译模型编码器，基于双向 GRU。
    """
    def __init__(self, vocab_size, padding_index):
        """
        初始化编码器。
        :param vocab_size: 词表大小。
        :param padding_index: padding token 的索引。
        """
        super().__init__()
        
        # 嵌入层：将 token 索引映射为稠密向量
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=config.EMBEDDING_DIM,
            padding_idx=padding_index
        )
        
        # 双向 GRU
        self.rnn = nn.GRU(
            input_size=config.EMBEDDING_DIM,
            hidden_size=config.ENCODER_HIDDEN_DIM,
            num_layers=config.ENCODER_LAYERS,
            batch_first=True,
            bidirectional=True
        )

    def forward(self, src):
        """
        前向传播。
        :param src: 输入张量，形状 (batch_size, seq_len)。
        :return: (输出张量, 最终隐藏状态)。
        """
        embedded = self.embedding(src)  # (batch_size, seq_len, embedding_dim)
        output, hidden = self.rnn(embedded)
        # output: (batch_size, seq_len, 2*hidden_dim)
        # hidden: (2*num_layers, batch_size, hidden_dim)
        return output, hidden

class TranslationDecoder(nn.Module):
    """
    翻译模型解码器，基于单向 GRU。
    """
    def __init__(self, vocab_size, padding_index):
        """
        初始化解码器。
        :param vocab_size: 词表大小。
        :param padding_index: padding token 的索引。
        """
        super().__init__()
        
        # 嵌入层
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=config.EMBEDDING_DIM,
            padding_idx=padding_index
        )
        
        # 单向 GRU
        self.rnn = nn.GRU(
            input_size=config.EMBEDDING_DIM,
            hidden_size=config.DECODER_HIDDEN_DIM,
            batch_first=True
        )
        
        # 线性层：映射到词表概率分布
        self.linear = nn.Linear(
            in_features=config.DECODER_HIDDEN_DIM,
            out_features=vocab_size
        )

    def forward(self, tgt, hidden):
        """
        前向传播。
        :param tgt: 输入张量，形状 (batch_size, 1)。
        :param hidden: 隐藏状态张量，形状 (1, batch_size, hidden_dim)。
        :return: (输出张量, 新的隐藏状态)。
        """
        embedded = self.embedding(tgt)  # (batch_size, 1, embedding_dim)
        output, hidden = self.rnn(embedded, hidden)  # output: (batch_size, 1, hidden_dim)
        output = self.linear(output)  # (batch_size, 1, vocab_size)
        return output, hidden

if __name__ == '__main__':
    # 测试模型结构
    encoder = TranslationEncoder(vocab_size=10000, padding_index=0)
    dummy_encoder_input = torch.randint(low=0, high=10000, size=(config.BATCH_SIZE, config.SEQ_LEN))
    print('编码器结构：')
    summary(encoder, input_data=dummy_encoder_input)
    
    print('-' * 100)
    
    decoder = TranslationDecoder(vocab_size=10000, padding_index=0)
    dummy_decoder_input = torch.randint(low=0, high=10000, size=(config.BATCH_SIZE, 1))
    dummy_decoder_hidden = torch.randn(size=(1, config.BATCH_SIZE, config.DECODER_HIDDEN_DIM))
    print('解码器结构：')
    summary(decoder, input_data=[dummy_decoder_input, dummy_decoder_hidden])

模型训练 train.py

# train.py
import time
from itertools import chain
import torch
from torch.nn import CrossEntropyLoss
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
from dataset import get_dataloader
from tokenizer import ChineseTokenizer, EnglishTokenizer
import config
from model import TranslationEncoder, TranslationDecoder

def train_one_epoch(dataloader, encoder, decoder, loss_function, optimizer, device):
    """
    训练一个 epoch。
    :param dataloader: 数据加载器。
    :param encoder: 编码器。
    :param decoder: 解码器。
    :param loss_function: 损失函数。
    :param optimizer: 优化器。
    :param device: 设备。
    :return: 平均损失。
    """
    encoder.train()
    decoder.train()
    total_loss = 0
    
    for src, tgt in tqdm(dataloader, desc='训练'):
        src = src.to(device)
        tgt = tgt.to(device)
        
        optimizer.zero_grad()
        
        # 编码器处理
        _, encoder_hidden = encoder(src)
        
        # 拼接双向 GRU 的最后隐藏状态作为上下文向量
        # encoder_hidden形状: (2, batch_size, hidden_dim)
        forward_hidden = encoder_hidden[-2]  # 前向GRU的最后隐藏状态
        backward_hidden = encoder_hidden[-1]  # 后向GRU的最后隐藏状态
        context_vector = torch.cat([forward_hidden, backward_hidden], dim=1)  # (batch_size, 2*hidden_dim)
        
        # 初始化解码器输入和隐藏状态
        decoder_input = tgt[:, 0:1]  # 第一个输入是<sos>
        decoder_hidden = context_vector.unsqueeze(0)  # (1, batch_size, 2*hidden_dim)
        
        decoder_outputs = []
        
        # Teacher Forcing：每一步输入真实的前一个token
        for step in range(1, config.SEQ_LEN):
            decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)
            decoder_outputs.append(decoder_output)
            decoder_input = tgt[:, step:step + 1]  # 使用真实的前一个token作为输入
        
        decoder_outputs = torch.cat(decoder_outputs, dim=1)  # (batch_size, seq_len-1, vocab_size)
        decoder_targets = tgt[:, 1:]  # 目标是从第二个token开始到结束
        
        # 计算损失
        loss = loss_function(
            decoder_outputs.reshape(-1, decoder_outputs.shape[-1]),
            decoder_targets.reshape(-1)
        )
        
        # 反向传播和优化
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
    
    return total_loss / len(dataloader)

def train():
    """
    模型训练主函数。
    """
    # 自动选择设备(优先使用GPU)
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print(f'使用设备: {device}')
    
    # 获取数据加载器
    dataloader = get_dataloader()
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 初始化模型
    encoder = TranslationEncoder(
        vocab_size=zh_tokenizer.vocab_size,
        padding_index=zh_tokenizer.pad_token_index
    ).to(device)
    
    decoder = TranslationDecoder(
        vocab_size=en_tokenizer.vocab_size,
        padding_index=en_tokenizer.pad_token_index
    ).to(device)
    
    # 定义损失函数和优化器
    loss_function = CrossEntropyLoss(ignore_index=en_tokenizer.pad_token_index)
    optimizer = torch.optim.Adam(chain(encoder.parameters(), decoder.parameters()), lr=config.LEARNING_RATE)
    
    # TensorBoard日志
    writer = SummaryWriter(log_dir=config.LOGS_DIR / time.strftime('%Y-%m-%d_%H-%M-%S'))
    
    best_loss = float('inf')
    
    # 开始训练
    for epoch in range(1, config.EPOCHS + 1):
        print(f'\n========== Epoch {epoch}/{config.EPOCHS} ==========')
        
        avg_loss = train_one_epoch(dataloader, encoder, decoder, loss_function, optimizer, device)
        
        print(f'平均损失: {avg_loss:.4f}')
        writer.add_scalar('Train/Loss', avg_loss, epoch)
        
        # 保存最好的模型
        if avg_loss < best_loss:
            best_loss = avg_loss
            torch.save(encoder.state_dict(), config.MODELS_DIR / 'encoder.pt')
            torch.save(decoder.state_dict(), config.MODELS_DIR / 'decoder.pt')
            print(f'已保存最佳模型(损失: {best_loss:.4f})')
    
    print('\n训练完成！')
    print(f'最佳验证损失: {best_loss:.4f}')

if __name__ == '__main__':
    train()

模型预测 predict.py

# predict.py
import torch
from tokenizer import ChineseTokenizer, EnglishTokenizer
from model import TranslationEncoder, TranslationDecoder
import config

def predict_batch(input_tensor, encoder, decoder, en_tokenizer, device):
    """
    对一个 batch 的输入进行翻译预测。
    :param input_tensor: 中文输入张量，形状 (batch_size, seq_len)。
    :param encoder: 编码器。
    :param decoder: 解码器。
    :param en_tokenizer: 英文分词器。
    :param device: 设备。
    :return: 英文 token 索引列表。
    """
    encoder.eval()
    decoder.eval()
    
    with torch.no_grad():
        # 编码器前向传播
        encoder_output, encoder_hidden = encoder(input_tensor)
        
        # 拼接双向 GRU 的最后隐藏状态作为上下文向量
        context_vector = torch.cat([encoder_hidden[-2], encoder_hidden[-1]], dim=1)
        
        batch_size = input_tensor.shape[0]
        
        # 初始化解码器输入：<sos>
        decoder_input = torch.full(
            size=(batch_size, 1),
            fill_value=en_tokenizer.sos_token_index,
            device=device
        )
        
        decoder_hidden = context_vector.unsqueeze(0)
        
        generated = [[] for _ in range(batch_size)]
        finished = [False for _ in range(batch_size)]
        
        # 自回归生成
        for step in range(1, config.SEQ_LEN):
            decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)
            
            # 贪心解码：选择概率最高的词
            predict_indexes = decoder_output.argmax(dim=-1)
            
            # 处理每个样本
            for i in range(batch_size):
                if finished[i]:
                    continue
                
                token_id = predict_indexes[i].item()
                
                if token_id == en_tokenizer.eos_token_index:
                    finished[i] = True
                    continue
                
                generated[i].append(token_id)
            
            # 如果所有样本都生成了<eos>，提前结束
            if all(finished):
                break
            
            # 将预测结果作为下一步输入
            decoder_input = predict_indexes
        
        return generated

def predict(zh_sentence, encoder, decoder, zh_tokenizer, en_tokenizer, device):
    """
    对单条中文句子进行翻译。
    :param zh_sentence: 中文句子。
    :param encoder: 编码器。
    :param decoder: 解码器。
    :param zh_tokenizer: 中文分词器。
    :param en_tokenizer: 英文分词器。
    :param device: 设备。
    :return: 英文翻译句子。
    """
    # 编码输入句子
    input_ids = zh_tokenizer.encode(zh_sentence, seq_len=config.SEQ_LEN, add_sos_eos=False)
    input_tensor = torch.tensor([input_ids], device=device)
    
    # 预测
    generated = predict_batch(input_tensor, encoder, decoder, en_tokenizer, device)
    en_indexes = generated[0]
    
    # 解码为英文句子
    en_sentence = en_tokenizer.decode(en_indexes)
    return en_sentence

def run_predict():
    """
    启动交互式翻译程序。
    """
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print(f'使用设备: {device}')
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 加载模型
    encoder = TranslationEncoder(
        vocab_size=zh_tokenizer.vocab_size,
        padding_index=zh_tokenizer.pad_token_index
    ).to(device)
    encoder.load_state_dict(torch.load(config.MODELS_DIR / 'encoder.pt', map_location=device))
    
    decoder = TranslationDecoder(
        vocab_size=en_tokenizer.vocab_size,
        padding_index=en_tokenizer.pad_token_index
    ).to(device)
    decoder.load_state_dict(torch.load(config.MODELS_DIR / 'decoder.pt', map_location=device))
    
    print('\n欢迎使用中英翻译系统！')
    print('请输入中文句子(输入 q 或 quit 退出)\n')
    
    while True:
        user_input = input('中文: ')
        
        if user_input.lower() in ['q', 'quit']:
            print('谢谢使用，再见！')
            break
        
        if not user_input.strip():
            print('请输入有效的中文句子\n')
            continue
        
        # 进行翻译
        result = predict(user_input, encoder, decoder, zh_tokenizer, en_tokenizer, device)
        print(f'英文: {result}\n')

if __name__ == '__main__':
    run_predict()

模型评估 evaluate.py

# evaluate.py
import torch
from nltk.translate.bleu_score import corpus_bleu
from tqdm import tqdm
import config
from tokenizer import ChineseTokenizer, EnglishTokenizer
from model import TranslationEncoder, TranslationDecoder
from dataset import get_dataloader
from predict import predict_batch

def evaluate(dataloader, encoder, decoder, zh_tokenizer, en_tokenizer, device):
    """
    执行模型评估，计算BLEU分数。
    :param dataloader: 数据加载器。
    :param encoder: 编码器。
    :param decoder: 解码器。
    :param zh_tokenizer: 中文分词器。
    :param en_tokenizer: 英文分词器。
    :param device: 设备。
    :return: BLEU-4分数。
    """
    encoder.eval()
    decoder.eval()
    
    all_references = []  # 所有参考译文
    all_predictions = []  # 所有预测译文
    
    special_tokens = [
        en_tokenizer.pad_token_index,
        en_tokenizer.eos_token_index,
        en_tokenizer.sos_token_index
    ]
    
    with torch.no_grad():
        for src, tgt in tqdm(dataloader, desc="评估"):
            src = src.to(device)
            tgt = tgt.tolist()
            
            # 预测
            predict_indexes = predict_batch(src, encoder, decoder, en_tokenizer, device)
            all_predictions.extend(predict_indexes)
            
            # 处理参考译文
            for indexes in tgt:
                # 去除特殊标记
                ref_indexes = [index for index in indexes if index not in special_tokens]
                # corpus_bleu要求每个参考是一个列表(支持多个参考译文)
                all_references.append([ref_indexes])
    
    # 计算BLEU-4分数
    bleu = corpus_bleu(all_references, all_predictions)
    return bleu

def run_evaluate():
    """
    启动评估流程。
    """
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print(f'使用设备: {device}')
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 加载模型
    encoder = TranslationEncoder(
        vocab_size=zh_tokenizer.vocab_size,
        padding_index=zh_tokenizer.pad_token_index
    ).to(device)
    encoder.load_state_dict(torch.load(config.MODELS_DIR / 'encoder.pt', map_location=device))
    
    decoder = TranslationDecoder(
        vocab_size=en_tokenizer.vocab_size,
        padding_index=en_tokenizer.pad_token_index
    ).to(device)
    decoder.load_state_dict(torch.load(config.MODELS_DIR / 'decoder.pt', map_location=device))
    
    # 获取测试集数据加载器
    dataloader = get_dataloader(train=False)
    
    print('\n开始评估模型...')
    bleu = evaluate(dataloader, encoder, decoder, zh_tokenizer, en_tokenizer, device)
    
    print('\n========== 评估结果 ==========')
    print(f'BLEU-4 分数: {bleu:.4f}')
    print('=============================')

if __name__ == '__main__':
    run_evaluate()

# 5.5 经典Seq2Seq的局限性：为什么我们需要Attention？

虽然Seq2Seq模型成功解决了序列到序列的生成问题，但它存在两个无法忽视的核心缺陷，这也是为什么后来Attention机制和Transformer模型会迅速取代它的原因。

# 5.5.1 信息瓶颈：定长向量无法表达无限语义

经典Seq2Seq模型的最大问题在于：编码器需要将任意长度、任意复杂度的句子压缩为一个固定长度的上下文向量。

这就像让你把一本1000页的书的所有内容都写在一张便签纸上——无论你多么努力，必然会丢失大量信息。尤其是在处理长句子或复杂语义时，这个问题会变得尤为严重。

例如，翻译一个包含多个从句的长句："我昨天在公园遇到了那个穿着红色外套、戴着黑色帽子、手里拿着一本书的女孩。" 编码器很难把所有这些细节都准确地压缩到一个向量中，导致翻译时丢失重要信息。

# 5.5.2 缺乏动态注意力：解码器无法"聚焦"关键信息

在经典Seq2Seq中，解码器在生成每个词时，都只能使用同一个上下文向量。但这显然不符合人类的翻译习惯。

当人类翻译句子时，我们会在生成不同词的时候，关注源句子中不同的部分：

翻译"我"的时候，我们关注源句的第一个字
翻译"喜欢"的时候，我们关注源句的第二个字
翻译"足球"的时候，我们关注源句的最后两个字

而经典Seq2Seq模型做不到这一点。它在生成每个词时，都必须"一视同仁"地考虑整个句子的所有信息，无法动态地聚焦到当前需要的部分。这导致生成的句子经常出现语序错误、指代不清等问题。

正是为了解决这两个问题，研究者们提出了Attention机制——让解码器在生成每个词时，都能动态地关注源句子中最相关的部分。这就是我们下一章要重点讲解的内容。

本章小结：

Seq2Seq模型采用编码器-解码器架构，解决了输入输出长度可变的序列生成问题
编码器将输入序列压缩为上下文向量，解码器基于该向量自回归生成目标序列
训练阶段使用Teacher Forcing策略加速训练，推理阶段使用贪心解码或束搜索生成句子
经典Seq2Seq存在信息瓶颈和缺乏动态注意力两个核心缺陷，这直接催生了Attention机制的诞生

# 第六章 Attention机制：让AI学会"聚焦重点"的革命性突破

在前面的章节中，我们学习了Seq2Seq模型，它为机器翻译、文本生成等序列任务提供了强大的基础框架。然而，传统Seq2Seq模型有一个致命的弱点——它试图将整句话的信息压缩成一个固定长度的向量，这就像让你读完一本厚书后，只用一句话总结所有内容，必然会丢失大量关键信息。

本章我们将学习NLP发展史上最重要的技术之一：Attention机制。它彻底改变了序列模型的工作方式，让AI在处理信息时能够像人类一样"聚焦重点"，从而大幅提升了模型的性能和表达能力。

# 6.1 概述：从"死记硬背"到"灵活聚焦"

# 6.1.1 传统Seq2Seq的致命缺陷

传统的Seq2Seq模型中，编码器在处理源句时，无论句子多长、多复杂，最终都只能将所有信息压缩为一个固定长度的上下文向量，作为解码器生成目标序列的唯一参考。这种设计存在两个无法回避的问题：

信息压缩瓶颈：想象一下，让你把《红楼梦》的全部内容压缩成100个字，无论你多么聪明，都必然会丢失大量关键情节和人物关系。同样，固定长度的向量难以完整表达长句或复杂语义，当句子长度超过一定限度时，模型性能会急剧下降。
缺乏动态感知能力：解码器在生成每个目标词时，都只能依赖同一个静态的上下文向量。这就像你在翻译一句话时，无论翻译哪个词，都只能回忆起整句话的模糊印象，而不能专门关注与当前翻译词最相关的部分。

举个具体的例子：当翻译"我爱人工智能"这句话时，传统Seq2Seq模型在生成"I"、"love"、"artificial"、"intelligence"这四个词时，使用的是完全相同的上下文向量。但实际上，生成"I"时应该最关注"我"，生成"love"时应该最关注"爱"，生成"artificial intelligence"时应该最关注"人工智能"。

# 6.1.2 Attention机制的核心思想

为了解决上述问题，研究者在2014年引入了Attention机制。其核心思想非常直观：

解码器在生成目标序列的每一步时，不再依赖于一个静态的上下文向量，而是根据当前的解码状态，动态地从编码器各时间步的隐藏状态中选取最相关的信息，以辅助当前步的生成。

这种机制赋予了模型强大的"对齐"能力——它能够自动判断源句中哪些位置对当前的目标词更为重要，并为这些位置分配更高的"注意力权重"。

回到刚才的翻译例子：有了Attention机制后，模型在生成"I"时会给"我"分配最高的权重，生成"love"时会给"爱"分配最高的权重，以此类推。这就像人类翻译时，会一边看着源文本，一边有选择地关注不同的部分。

# 6.2 工作原理：四步实现"智能聚焦"

注意力机制的工作过程可以拆解为四个清晰的步骤，每一步都对应着一个关键的计算过程。我们仍然以机器翻译为例来理解这些步骤。

# 6.2.1 第一步：相关性计算

在目标序列生成的每一步，解码器首先会计算当前时间步的隐藏状态与编码器各个时间步输出之间的相关性。这些相关性分数衡量了源句中每个位置对当前生成内容的重要程度。

举个例子：当解码器正在生成"love"这个词时，它会计算自己当前的状态与编码器处理"我"、"爱"、"人"、"工"、"智"、"能"这六个位置时的输出之间的相关性。显然，与"爱"这个位置的相关性应该是最高的。

相关性的计算依赖于特定的函数，通常被称为注意力评分函数(attention scoring function)。我们将在下一节详细介绍三种常见的评分函数实现方式。

# 6.2.2 第二步：注意力权重计算

得到所有源位置的注意力评分后，我们使用Softmax函数将这些评分归一化为概率分布，这就是注意力权重。

Softmax函数有两个重要特性：

所有输出值都在0到1之间
所有输出值的和为1

这意味着注意力权重可以被解释为"模型在当前生成步骤中，对源句每个位置的关注程度"。得分越高的位置，其对应的权重越大，代表模型在当前生成中更关注该位置的信息。

# 6.2.3 第三步：上下文向量计算

有了注意力权重后，我们将所有编码器输出按照注意力权重进行加权求和，得到一个上下文向量。

这个向量就表示当前时间步，模型从源句中提取出的关键信息。它不是一个固定的向量，而是随着解码过程动态变化的——每生成一个词，模型都会重新计算一次注意力权重，从而得到一个新的上下文向量。

# 6.2.4 第四步：解码信息融合

在得到上下文向量后，解码器将其与当前时间步的隐藏状态进行拼接，以融合两者的信息。

解码器的隐藏状态包含了已经生成的目标序列的信息
上下文向量包含了源句中与当前生成词最相关的信息

将这两部分信息融合后，最终通过线性变换和Softmax，生成当前时间步目标词的概率分布。

# 6.3 注意力评分函数：如何衡量"相关性"

注意力评分函数是Attention机制的核心组件，它决定了如何计算解码器当前状态与编码器各时间步状态之间的相关性。本节我们将介绍三种最常见的评分函数。

# 6.3.1 概述

虽然不同的评分函数在结构上各有差异，但它们的本质目标是一致的：衡量两个向量之间的相似程度。两个向量越相似，它们之间的相关性就越高，模型就应该给予更多的注意力。

# 6.3.2 点积评分(Dot)

点积评分是注意力机制中最简单、最直接的一种相关性评分方法。它通过计算解码器当前时间步的隐藏状态与编码器每个时间步的隐藏状态的点积，来衡量二者之间的相关性：

直观理解：两个向量的点积等于它们的模长乘积再乘以夹角的余弦值。如果两个向量方向越一致（即越接近），它们的点积就越大，表示相关性越强。

优点：计算简单、速度快、没有可学习的参数。

缺点：要求编码器和解码器的隐藏状态维度必须相同，否则无法进行点积运算。

# 6.3.3 通用点积评分(General)

通用点积评分在点积的基础上引入了一个可学习的权重矩阵W，用于先对编码器隐藏状态进行线性变换，再与解码器隐藏状态进行点积：

设计动机：解决编码器和解码器隐藏状态维度不一致的问题。通过引入权重矩阵W，我们可以将编码器的隐藏状态映射到与解码器隐藏状态相同的维度空间。

优点：

实现了维度对齐，适用范围更广
引入了可学习的参数，增强了模型对编码器输出的适应能力，从而提升了注意力机制的表达能力

# 6.3.4 拼接评分(Concat)

拼接评分是一种表达能力更强的相关性评分方法。它的核心思想是：将解码器当前隐藏状态与编码器每个时间步的隐藏状态拼接为一个长向量，经过线性变换和非线性激活，最后用一个向量进行投影，得到最终打分值：

优点：建模能力最强。它不仅考虑了两个状态的数值关系，还引入了非线性变换（tanh函数），能够捕捉更复杂的交互模式。

缺点：计算复杂度最高，参数数量最多。

适用场景：对齐关系复杂的任务场景，比如长句翻译、对话生成等。

# 6.4 案例实操：中英翻译V2.0

理论讲完了，现在让我们动手实践，在之前的Seq2Seq翻译模型基础上引入Attention机制，打造一个性能更好的中英翻译系统。

# 6.4.1 需求说明

本案例要求在已有的Seq2Seq模型基础上，引入注意力机制，以提升模型在处理长句或复杂句时的表达能力和生成质量。

# 6.4.2 需求分析

为引入Attention机制，我们只需要修改解码器部分，编码器部分可以完全保持不变：

编码器：仍然使用双向GRU对中文句子进行编码，输出所有时间步的隐藏状态。
解码器：在每个时间步，都需要执行以下操作：
1. 将当前隐藏状态与编码器输出序列共同用于计算注意力权重（我们将使用最简单的点积评分函数）
2. 根据权重对编码器各位置进行加权求和，得到上下文向量
3. 将上下文向量与当前解码状态拼接，作为输出的最终依据

# 6.4.3 需求实现

# 6.4.3.1 项目结构

我们的项目结构保持不变，如下所示：

# 6.4.3.2 完整代码

数据预处理

# process.py
import pandas as pd
from sklearn.model_selection import train_test_split
from tokenizer import ChineseTokenizer, EnglishTokenizer
import config

def process():
    """
    数据预处理主函数。
    """
    print('开始处理数据')
    
    # 读取中英文对齐数据
    df = pd.read_csv(
        config.RAW_DATA_DIR / 'cmn.txt',
        sep='\t',
        header=None,
        usecols=[0, 1],
        names=['en', 'zh']
    )
    
    # 清理空值数据
    df = df.dropna()
    df = df[df['en'].str.strip().ne('') & df['zh'].str.strip().ne('')]
    
    # 划分训练集和测试集（8:2）
    train_df, test_df = train_test_split(df, test_size=0.2, random_state=42)
    
    # 构建中英文词表
    EnglishTokenizer.build_vocab(train_df['en'].tolist(), config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    ChineseTokenizer.build_vocab(train_df['zh'].tolist(), config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    
    # 加载词表
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    
    # 编码训练集
    train_df['en'] = train_df['en'].apply(
        lambda x: en_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=True)
    )
    train_df['zh'] = train_df['zh'].apply(
        lambda x: zh_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=False)
    )
    
    # 保存处理后的训练数据
    train_df.to_json(
        config.PROCESSED_DATA_DIR / 'indexed_train.jsonl',
        orient='records',
        lines=True
    )
    
    # 编码测试集
    test_df['en'] = test_df['en'].apply(
        lambda x: en_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=True)
    )
    test_df['zh'] = test_df['zh'].apply(
        lambda x: zh_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=False)
    )
    
    # 保存处理后的测试数据
    test_df.to_json(
        config.PROCESSED_DATA_DIR / 'indexed_test.jsonl',
        orient='records',
        lines=True
    )
    
    print('数据处理完成')

if __name__ == '__main__':
    process()

自定义分词器

# tokenizer.py
from abc import abstractmethod
from nltk import word_tokenize, TreebankWordDetokenizer
from tqdm import tqdm

class BaseTokenizer:
    """
    分词器基类，提供词表构建、编码、解码等基础功能。
    """
    unk_token = '<unk>'   # 未知词标记
    pad_token = '<pad>'   # 填充标记
    sos_token = '<sos>'   # 句子开始标记
    eos_token = '<eos>'   # 句子结束标记
    
    @staticmethod
    @abstractmethod
    def tokenize(sentence):
        """
        分词抽象方法。
        :param sentence: 输入句子。
        :return: 分词结果。
        """
        pass
    
    @abstractmethod
    def decode(self, indexes):
        """
        解码抽象方法。
        :param indexes: 索引列表。
        :return: 解码后的句子。
        """
        pass
    
    @classmethod
    def build_vocab(cls, sentences, vocab_file):
        """
        构建词表并保存到文件。
        :param sentences: 句子列表。
        :param vocab_file: 保存词表的文件路径。
        """
        unique_words = set()
        for sentence in tqdm(sentences, desc='构建词表'):
            for word in cls.tokenize(sentence):
                unique_words.add(word)
        
        # 特殊标记放在词表最前面
        vocab_list = [cls.pad_token, cls.unk_token, cls.sos_token, cls.eos_token] + list(unique_words)
        
        with open(vocab_file, 'w', encoding='utf-8') as f:
            for word in vocab_list:
                f.write(word + '\n')
    
    def __init__(self, vocab_list):
        """
        初始化分词器。
        :param vocab_list: 词表列表。
        """
        self.vocab_list = vocab_list
        self.vocab_size = len(vocab_list)
        
        # 建立词到索引和索引到词的映射
        self.word2index = {word: idx for idx, word in enumerate(vocab_list)}
        self.index2word = {idx: word for idx, word in enumerate(vocab_list)}
        
        # 特殊标记的索引
        self.unk_token_index = self.word2index[self.unk_token]
        self.pad_token_index = self.word2index[self.pad_token]
        self.sos_token_index = self.word2index[self.sos_token]
        self.eos_token_index = self.word2index[self.eos_token]
    
    @classmethod
    def from_vocab(cls, vocab_file):
        """
        从词表文件加载分词器。
        :param vocab_file: 词表文件路径。
        :return: 分词器实例。
        """
        with open(vocab_file, 'r', encoding='utf-8') as f:
            vocab_list = [line.strip() for line in f.readlines()]
        return cls(vocab_list)
    
    def encode(self, sentence, seq_len, add_sos_eos=False):
        """
        将句子编码为索引列表。
        :param sentence: 输入句子。
        :param seq_len: 最大序列长度。
        :param add_sos_eos: 是否添加句子开始和结束标记。
        :return: 索引列表。
        """
        tokens = self.tokenize(sentence)
        indexes = [self.word2index.get(token, self.unk_token_index) for token in tokens]
        
        if add_sos_eos:
            # 预留位置给<sos>和<eos>
            indexes = indexes[:seq_len - 2]
            indexes = [self.sos_token_index] + indexes + [self.eos_token_index]
        else:
            indexes = indexes[:seq_len]
        
        # 填充到固定长度
        if len(indexes) < seq_len:
            indexes += [self.pad_token_index] * (seq_len - len(indexes))
        
        return indexes

class ChineseTokenizer(BaseTokenizer):
    """中文字符级分词器"""
    @staticmethod
    def tokenize(sentence):
        return list(sentence)
    
    def decode(self, indexes):
        return ''.join([self.index2word[index] for index in indexes if index != self.pad_token_index])

class EnglishTokenizer(BaseTokenizer):
    """英文单词级分词器"""
    @staticmethod
    def tokenize(sentence):
        return word_tokenize(sentence.lower())
    
    def decode(self, indexes):
        tokens = [self.index2word[index] for index in indexes if index != self.pad_token_index]
        return TreebankWordDetokenizer().detokenize(tokens)

自定义数据集

# dataset.py
import pandas as pd
import torch
from torch.utils.data import Dataset, DataLoader
import config

class TranslationDataset(Dataset):
    """翻译任务数据集"""
    def __init__(self, data_path):
        self.data = pd.read_json(data_path, lines=True).to_dict(orient='records')
    
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, index):
        # 中文作为输入，英文作为目标
        input_tensor = torch.tensor(self.data[index]['zh'], dtype=torch.long)
        target_tensor = torch.tensor(self.data[index]['en'], dtype=torch.long)
        return input_tensor, target_tensor

def get_dataloader(train=True):
    """获取数据加载器"""
    data_path = config.PROCESSED_DATA_DIR / ('indexed_train.jsonl' if train else 'indexed_test.jsonl')
    dataset = TranslationDataset(data_path)
    return DataLoader(dataset, batch_size=config.BATCH_SIZE, shuffle=True)

if __name__ == '__main__':
    # 测试数据集是否正常工作
    train_loader = get_dataloader(train=True)
    for inputs, targets in train_loader:
        print(f'输入形状: {inputs.shape}')  # 应该是 [batch_size, seq_len]
        print(f'目标形状: {targets.shape}') # 应该是 [batch_size, seq_len]
        break

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

模型定义（重点修改部分）

# model.py
import torch
from torch import nn
from torchinfo import summary
import config

class Attention(nn.Module):
    """
    注意力机制模块：计算当前decoder状态与encoder输出的注意力上下文向量。
    这里我们使用最简单的点积评分函数。
    """
    def forward(self, decoder_hidden, encoder_outputs):
        """
        计算注意力权重并加权求和生成上下文向量。
        :param decoder_hidden: 当前时间步解码器的隐藏状态 (1, batch_size, decoder_hidden_dim)
        :param encoder_outputs: 编码器所有时间步输出 (batch_size, seq_len, decoder_hidden_dim)
        :return: 上下文向量 (batch_size, 1, decoder_hidden_dim)
        """
        # 调整张量形状以进行批量矩阵乘法
        # decoder_hidden: (1, batch_size, hidden_dim) -> (batch_size, 1, hidden_dim)
        decoder_hidden_transposed = decoder_hidden.transpose(0, 1)
        
        # encoder_outputs: (batch_size, seq_len, hidden_dim) -> (batch_size, hidden_dim, seq_len)
        encoder_outputs_transposed = encoder_outputs.transpose(1, 2)
        
        # 计算注意力分数: (batch_size, 1, hidden_dim) @ (batch_size, hidden_dim, seq_len) = (batch_size, 1, seq_len)
        attention_scores = torch.bmm(decoder_hidden_transposed, encoder_outputs_transposed)
        
        # 计算注意力权重（使用softmax归一化）
        attention_weights = torch.softmax(attention_scores, dim=2)
        
        # 加权求和得到上下文向量: (batch_size, 1, seq_len) @ (batch_size, seq_len, hidden_dim) = (batch_size, 1, hidden_dim)
        context_vector = torch.bmm(attention_weights, encoder_outputs)
        
        return context_vector

class TranslationEncoder(nn.Module):
    """
    编码器模块：双向GRU编码中文句子。
    与之前的Seq2Seq模型完全相同，无需修改。
    """
    def __init__(self, vocab_size, padding_index):
        super().__init__()
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=config.EMBEDDING_DIM,
            padding_idx=padding_index
        )
        self.rnn = nn.GRU(
            input_size=config.EMBEDDING_DIM,
            hidden_size=config.ENCODER_HIDDEN_DIM,
            num_layers=config.ENCODER_LAYERS,
            batch_first=True,
            bidirectional=True  # 使用双向GRU
        )
    
    def forward(self, src):
        """
        前向传播。
        :param src: 中文输入索引序列 (batch_size, seq_len)
        :return: (encoder_outputs, encoder_hidden)
                 encoder_outputs: 所有时间步的输出 (batch_size, seq_len, 2*encoder_hidden_dim)
                 encoder_hidden: 最后一个时间步的隐藏状态 (2*encoder_layers, batch_size, encoder_hidden_dim)
        """
        embedded = self.embedding(src)  # (batch_size, seq_len, embedding_dim)
        output, hidden = self.rnn(embedded)
        return output, hidden

class TranslationDecoder(nn.Module):
    """
    解码器模块：单向GRU + Attention，逐步生成英文翻译。
    这是与传统Seq2Seq模型的主要区别所在。
    """
    def __init__(self, vocab_size, padding_index):
        super().__init__()
        self.embedding = nn.Embedding(
            num_embeddings=vocab_size,
            embedding_dim=config.EMBEDDING_DIM,
            padding_idx=padding_index
        )
        self.rnn = nn.GRU(
            input_size=config.EMBEDDING_DIM,
            hidden_size=config.DECODER_HIDDEN_DIM,
            batch_first=True
        )
        # 注意力模块
        self.attention = Attention()
        # 输出层：输入是隐藏状态和上下文向量的拼接，所以维度是2*decoder_hidden_dim
        self.linear = nn.Linear(in_features=2 * config.DECODER_HIDDEN_DIM, out_features=vocab_size)
    
    def forward(self, tgt, hidden, encoder_outputs):
        """
        前向传播。
        :param tgt: 当前输入token (batch_size, 1)
        :param hidden: 上一时间步隐藏状态 (1, batch_size, decoder_hidden_dim)
        :param encoder_outputs: 编码器所有时间步输出 (batch_size, seq_len, decoder_hidden_dim)
        :return: (output_logits, new_hidden)
                 output_logits: 当前时间步的输出概率 (batch_size, 1, vocab_size)
                 new_hidden: 更新后的隐藏状态 (1, batch_size, decoder_hidden_dim)
        """
        embedded = self.embedding(tgt)  # (batch_size, 1, embedding_dim)
        
        # GRU前向传播
        output, hidden = self.rnn(embedded, hidden)  # output: (batch_size, 1, hidden_dim)
        
        # 计算注意力上下文向量
        context_vector = self.attention(hidden, encoder_outputs)  # (batch_size, 1, hidden_dim)
        
        # 拼接当前输出和上下文向量
        combined = torch.cat((output, context_vector), dim=2)  # (batch_size, 1, 2*hidden_dim)
        
        # 输出层
        output = self.linear(combined)  # (batch_size, 1, vocab_size)
        
        return output, hidden

if __name__ == '__main__':
    # 测试模型结构
    encoder = TranslationEncoder(vocab_size=10000, padding_index=0)
    dummy_encoder_input = torch.randint(low=0, high=10000, size=(config.BATCH_SIZE, config.SEQ_LEN))
    print("编码器结构：")
    summary(encoder, input_data=dummy_encoder_input)
    
    print('-' * 100)
    
    decoder = TranslationDecoder(vocab_size=10000, padding_index=0)
    dummy_decoder_input = torch.randint(low=0, high=10000, size=(config.BATCH_SIZE, 1))
    dummy_decoder_hidden = torch.randn(size=(1, config.BATCH_SIZE, config.DECODER_HIDDEN_DIM))
    dummy_encoder_outputs = torch.randn(size=(config.BATCH_SIZE, config.SEQ_LEN, config.DECODER_HIDDEN_DIM))
    print("解码器结构：")
    summary(decoder, input_data=[dummy_decoder_input, dummy_decoder_hidden, dummy_encoder_outputs])

模型训练

# train.py
import time
from itertools import chain
import torch
from torch.nn import CrossEntropyLoss
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
from dataset import get_dataloader
from tokenizer import ChineseTokenizer, EnglishTokenizer
import config
from model import TranslationEncoder, TranslationDecoder

def train_one_epoch(dataloader, encoder, decoder, loss_function, optimizer, device):
    """训练一个epoch"""
    encoder.train()
    decoder.train()
    total_loss = 0
    
    for src, tgt in tqdm(dataloader, desc='训练中'):
        src = src.to(device)  # (batch_size, seq_len)
        tgt = tgt.to(device)  # (batch_size, seq_len)
        
        optimizer.zero_grad()
        
        # 编码器前向传播
        encoder_outputs, encoder_hidden = encoder(src)
        
        # 拼接双向GRU最后一层的前向和后向隐藏状态，作为解码器的初始隐藏状态
        forward_hidden = encoder_hidden[-2]  # (batch_size, encoder_hidden_size)
        backward_hidden = encoder_hidden[-1] # (batch_size, encoder_hidden_size)
        decoder_hidden = torch.cat([forward_hidden, backward_hidden], dim=1)  # (batch_size, decoder_hidden_size)
        decoder_hidden = decoder_hidden.unsqueeze(0)  # (1, batch_size, decoder_hidden_size)
        
        # 解码器输入初始化为<sos>
        decoder_input = tgt[:, 0:1]  # (batch_size, 1)
        
        decoder_outputs = []
        # 教师强制训练：使用真实的前一个token作为输入
        for step in range(1, config.SEQ_LEN):
            decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden, encoder_outputs)
            decoder_outputs.append(decoder_output)
            decoder_input = tgt[:, step:step + 1]  # 使用真实的下一个token作为输入
        
        # 拼接所有时间步的输出
        decoder_outputs = torch.cat(decoder_outputs, dim=1)  # (batch_size, seq_len-1, vocab_size)
        decoder_targets = tgt[:, 1:]  # (batch_size, seq_len-1)
        
        # 计算损失（忽略padding标记）
        loss = loss_function(
            decoder_outputs.reshape(-1, decoder_outputs.shape[-1]), 
            decoder_targets.reshape(-1)
        )
        
        # 反向传播和优化
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
    
    return total_loss / len(dataloader)

def train():
    """主训练函数"""
    # 设备配置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print(f'使用设备: {device}')
    
    # 数据加载器
    dataloader = get_dataloader()
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 初始化模型
    encoder = TranslationEncoder(
        vocab_size=zh_tokenizer.vocab_size, 
        padding_index=zh_tokenizer.pad_token_index
    ).to(device)
    
    decoder = TranslationDecoder(
        vocab_size=en_tokenizer.vocab_size, 
        padding_index=en_tokenizer.pad_token_index
    ).to(device)
    
    # 损失函数（忽略padding标记）
    loss_function = CrossEntropyLoss(ignore_index=en_tokenizer.pad_token_index)
    
    # 优化器
    optimizer = torch.optim.Adam(
        params=chain(encoder.parameters(), decoder.parameters()), 
        lr=config.LEARNING_RATE
    )
    
    # TensorBoard日志
    writer = SummaryWriter(log_dir=config.LOGS_DIR / time.strftime('%Y-%m-%d_%H-%M-%S'))
    
    # 开始训练
    best_loss = float('inf')
    for epoch in range(1, config.EPOCHS + 1):
        print(f'\n========== Epoch {epoch}/{config.EPOCHS} ==========')
        
        avg_loss = train_one_epoch(dataloader, encoder, decoder, loss_function, optimizer, device)
        print(f'平均损失: {avg_loss:.4f}')
        
        # 记录损失到TensorBoard
        writer.add_scalar('Loss/train', avg_loss, epoch)
        
        # 保存最佳模型
        if avg_loss < best_loss:
            best_loss = avg_loss
            torch.save(encoder.state_dict(), config.MODELS_DIR / 'encoder.pt')
            torch.save(decoder.state_dict(), config.MODELS_DIR / 'decoder.pt')
            print(f'已保存最佳模型 (损失: {best_loss:.4f})')
        else:
            print(f'模型未改进，最佳损失保持为: {best_loss:.4f}')
    
    print('\n训练完成！')
    writer.close()

if __name__ == '__main__':
    train()

模型预测

# predict.py
import torch
from tokenizer import ChineseTokenizer, EnglishTokenizer
from model import TranslationEncoder, TranslationDecoder
import config

def predict_batch(input_tensor, encoder, decoder, en_tokenizer, device):
    """
    对一个batch的中文输入进行翻译。
    :param input_tensor: 中文输入张量 (batch_size, seq_len)
    :param encoder: 编码器模型
    :param decoder: 解码器模型
    :param en_tokenizer: 英文分词器
    :param device: 计算设备
    :return: 生成的英文索引列表
    """
    encoder.eval()
    decoder.eval()
    
    with torch.no_grad():
        # 编码器前向传播
        encoder_output, encoder_hidden = encoder(input_tensor)
        
        # 拼接双向GRU最后一层隐藏状态作为解码器初始隐藏状态
        context_vector = torch.cat([encoder_hidden[-2], encoder_hidden[-1]], dim=1)  # (batch_size, hidden_dim*2)
        decoder_hidden = context_vector.unsqueeze(0)  # (1, batch_size, hidden_dim)
        
        batch_size = input_tensor.size(0)
        
        # 初始输入为<sos>
        decoder_input = torch.full(
            size=(batch_size, 1),
            fill_value=en_tokenizer.sos_token_index,
            device=device
        )
        
        generated = [[] for _ in range(batch_size)]
        finished = [False for _ in range(batch_size)]
        
        # 逐词生成翻译结果
        for step in range(1, config.SEQ_LEN):
            decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden, encoder_output)
            predict_indexes = decoder_output.argmax(dim=-1)  # (batch_size, 1)
            
            # 处理每个样本的生成结果
            for i in range(batch_size):
                if finished[i]:
                    continue
                
                token_id = predict_indexes[i].item()
                if token_id == en_tokenizer.eos_token_index:
                    finished[i] = True
                else:
                    generated[i].append(token_id)
            
            # 如果所有样本都已生成结束，提前退出
            if all(finished):
                break
            
            # 使用预测的token作为下一个输入
            decoder_input = predict_indexes
        
        return generated

def predict(zh_sentence, encoder, decoder, zh_tokenizer, en_tokenizer, device):
    """
    对单条中文句子进行翻译。
    :param zh_sentence: 中文句子
    :return: 英文翻译句子
    """
    # 将中文句子编码为索引序列
    input_ids = zh_tokenizer.encode(zh_sentence, seq_len=config.SEQ_LEN, add_sos_eos=False)
    input_tensor = torch.tensor([input_ids], device=device)
    
    # 生成翻译结果
    generated = predict_batch(input_tensor, encoder, decoder, en_tokenizer, device)
    en_indexes = generated[0]
    
    # 将索引序列解码为英文句子
    en_sentence = en_tokenizer.decode(en_indexes)
    return en_sentence

def run_predict():
    """启动交互式翻译程序"""
    # 设备配置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 加载模型
    encoder = TranslationEncoder(
        vocab_size=zh_tokenizer.vocab_size,
        padding_index=zh_tokenizer.pad_token_index
    ).to(device)
    encoder.load_state_dict(torch.load(config.MODELS_DIR / 'encoder.pt', map_location=device))
    
    decoder = TranslationDecoder(
        vocab_size=en_tokenizer.vocab_size,
        padding_index=en_tokenizer.pad_token_index
    ).to(device)
    decoder.load_state_dict(torch.load(config.MODELS_DIR / 'decoder.pt', map_location=device))
    
    print('='*50)
    print('欢迎使用基于Attention机制的中英翻译系统')
    print('输入中文句子，系统将输出对应的英文翻译')
    print('输入 q 或 quit 退出程序')
    print('='*50)
    
    while True:
        user_input = input('\n中文: ')
        if user_input.lower() in ['q', 'quit']:
            print('谢谢使用，再见！')
            break
        
        if not user_input.strip():
            print('请输入有效的中文句子')
            continue
        
        result = predict(user_input, encoder, decoder, zh_tokenizer, en_tokenizer, device)
        print(f'英文: {result}')

if __name__ == '__main__':
    run_predict()

模型评估

# evaluate.py
import torch
from nltk.translate.bleu_score import corpus_bleu
from tqdm import tqdm
import config
from tokenizer import ChineseTokenizer, EnglishTokenizer
from model import TranslationEncoder, TranslationDecoder
from dataset import get_dataloader
from predict import predict_batch

def evaluate(dataloader, encoder, decoder, zh_tokenizer, en_tokenizer, device):
    """
    执行模型评估，计算BLEU分数。
    BLEU是机器翻译任务中最常用的自动评估指标，分数范围0-1，越高越好。
    """
    encoder.eval()
    decoder.eval()
    
    all_references = []  # 所有参考翻译
    all_predictions = [] # 所有模型预测结果
    
    special_tokens = [
        zh_tokenizer.pad_token_index,
        zh_tokenizer.eos_token_index,
        zh_tokenizer.sos_token_index
    ]
    
    with torch.no_grad():
        for src, tgt in tqdm(dataloader, desc="评估中"):
            src = src.to(device)
            tgt = tgt.tolist()
            
            # 模型预测
            predict_indexes = predict_batch(src, encoder, decoder, en_tokenizer, device)
            all_predictions.extend(predict_indexes)
            
            # 处理参考翻译
            for indexes in tgt:
                # 移除特殊标记
                indexes = [index for index in indexes if index not in special_tokens]
                all_references.append([indexes])  # BLEU支持多个参考翻译
    
    # 计算BLEU分数
    bleu = corpus_bleu(all_references, all_predictions)
    return bleu

def run_evaluate():
    """启动评估流程"""
    # 设备配置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 加载模型
    encoder = TranslationEncoder(zh_tokenizer.vocab_size, zh_tokenizer.pad_token_index).to(device)
    encoder.load_state_dict(torch.load(config.MODELS_DIR / 'encoder.pt', map_location=device))
    
    decoder = TranslationDecoder(en_tokenizer.vocab_size, en_tokenizer.pad_token_index).to(device)
    decoder.load_state_dict(torch.load(config.MODELS_DIR / 'decoder.pt', map_location=device))
    
    # 获取测试集数据加载器
    dataloader = get_dataloader(train=False)
    
    # 计算BLEU分数
    bleu = evaluate(dataloader, encoder, decoder, zh_tokenizer, en_tokenizer, device)
    
    print('\n========== 评估结果 ==========')
    print(f'BLEU分数: {bleu:.4f}')
    print('=============================')
    print('BLEU分数说明:')
    print('  - < 0.1: 翻译质量很差')
    print('  - 0.1-0.3: 翻译质量一般')
    print('  - 0.3-0.5: 翻译质量较好')
    print('  - > 0.5: 翻译质量优秀')

if __name__ == '__main__':
    run_evaluate()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79

配置文件

# config.py
from pathlib import Path

# 获取项目根目录
BASE_DIR = Path(__file__).parent.parent

# 定义项目中常用路径
MODELS_DIR = BASE_DIR / 'models'  # 模型保存路径
PROCESSED_DATA_DIR = BASE_DIR / 'data' / 'processed'  # 处理后的数据保存路径
RAW_DATA_DIR = BASE_DIR / 'data' / 'raw'  # 原始数据保存路径
LOGS_DIR = BASE_DIR / 'logs'  # TensorBoard日志目录

# 模型结构参数
EMBEDDING_DIM = 128  # 词向量维度
ENCODER_HIDDEN_DIM = 512  # GRU隐藏状态维度
DECODER_HIDDEN_DIM = 2 * ENCODER_HIDDEN_DIM  # 解码器隐藏维度是编码器的2倍（因为编码器是双向的）
ENCODER_LAYERS = 1  # 编码器层数

# 训练相关超参数
BATCH_SIZE = 128  # 每个batch的样本数
SEQ_LEN = 30  # 序列长度（输入与输出最大长度）
LEARNING_RATE = 1e-3  # 学习率
EPOCHS = 30  # 总训练轮数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 6.5 存在问题：为什么我们还需要Transformer

尽管注意力机制极大地增强了Seq2Seq模型的建模能力，使得模型能够处理更长的句子并生成更准确的翻译，但由于其核心依然依赖于RNN结构，仍面临两个根本性问题：

# 6.5.1 计算过程无法并行

RNN的时间步之间存在强依赖关系，必须顺序执行。也就是说，在处理一个句子时，模型必须先处理第一个词，然后才能处理第二个词，依此类推。

这严重限制了训练效率和硬件资源的利用率。现代GPU擅长并行计算，但RNN的顺序特性使得我们无法充分发挥GPU的并行计算能力。当处理大规模数据集时，训练时间会变得非常长。

# 6.5.2 长期依赖问题仍未根除

虽然Attention机制缓解了长期依赖问题，但模型仍然需要跨多个时间步传递信息。对于超长序列（比如包含数千个词的文档），训练过程中仍然容易出现梯度消失问题，难以有效建模非常长距离的依赖关系。

正是为了解决这两个问题，研究者们在2017年提出了Transformer模型。它完全抛弃了RNN结构，仅使用注意力机制来处理序列数据，从而实现了完全的并行计算，并更好地解决了长期依赖问题。

Transformer模型的提出，开启了NLP领域的新时代，为后续的BERT、GPT等预训练模型奠定了基础。我们将在下一章详细学习Transformer模型的原理和实现。

# 第七章 Transformer模型：NLP领域的革命性架构

# 7.1 概述：从RNN的局限到Transformer的诞生

在前面的章节中，我们学习了Seq2Seq模型以及注意力机制。注意力机制的引入确实让Seq2Seq模型在机器翻译等任务上取得了显著提升，但它始终无法摆脱一个根本性的问题——整个架构仍然建立在循环神经网络(RNN)之上。

RNN的本质是顺序计算：它必须一个词一个词地处理句子，后面的词必须等前面的词处理完才能开始。这种特性带来了两个致命缺陷：

计算效率极低：无法充分利用现代GPU的并行计算能力，训练速度慢
长距离依赖建模困难：信息在传递过程中会逐渐衰减，句子越长，效果越差

为了彻底解决这些问题，Google在2017年发表了一篇改变整个NLP领域的论文——《Attention Is All You Need (opens new window)》。这篇论文提出了一个大胆的想法：既然注意力机制已经能够很好地建模序列依赖关系，那我们为什么不彻底抛弃RNN，只使用注意力机制来构建模型呢？

基于这个核心理念，Transformer模型应运而生。它完全摒弃了循环结构，仅依靠注意力机制直接建模序列中任意两个位置之间的关系。这一革命性的设计带来了立竿见影的效果：

训练效率大幅提升：可以并行处理整个序列，充分利用GPU算力
长距离依赖建模能力显著增强：任意两个词之间可以直接建立联系，无需经过中间步骤
模型可以堆叠得更深：能够提取更复杂、更抽象的语义特征

Transformer的提出不仅解决了机器翻译任务的性能瓶颈，更开启了NLP的"大模型时代"。如今，我们熟知的BERT、GPT、T5等所有主流预训练语言模型，无一不是基于Transformer架构构建的。它的影响力甚至已经超越了NLP领域，扩展到了计算机视觉、语音识别、代码生成等众多AI分支，成为了现代深度学习的通用基础架构。

# 7.2 模型结构详解：拆解Transformer的核心组件

# 7.2.1 核心思想：注意力即一切

要理解Transformer，我们首先要回到注意力机制的本质。在Seq2Seq模型中，注意力机制允许解码器在生成每个词时，动态地关注源序列中最相关的部分。这实际上是在显式地建模目标位置与源序列各位置之间的依赖关系。

而RNN的作用是什么呢？它通过隐藏状态的递归传递，让当前位置的表示能够整合前文信息，本质上也是在隐式地建模序列中的依赖关系。

既然两者完成的是同一类任务，那么理论上，注意力机制完全可以替代RNN。而且相比RNN，注意力机制有着明显的优势：

无需顺序计算：所有位置可以同时处理，完美支持并行化
直接建立长距离联系：任意两个位置之间的路径长度都是1
计算复杂度可控：对于常用的序列长度，注意力机制的计算量甚至小于RNN

Transformer正是这一思想的完美实现。它彻底抛弃了RNN的循环结构，仅使用注意力机制来完成输入和输出序列中所有位置的依赖建模。这就是论文标题"Attention Is All You Need"的真正含义。

# 7.2.2 整体结构：编码器-解码器架构

Transformer延续了Seq2Seq模型经典的"编码器-解码器"设计理念：

编码器(Encoder)：负责"读懂"输入序列，将其转换为一组上下文相关的语义表示
解码器(Decoder)：根据编码器的输出，自回归地生成目标序列

所谓"自回归"，是指解码器在生成每个词时，都要把之前已经生成的所有词作为输入。例如，要生成句子"I like you"，解码器会先输入<sos>(开始标记)，输出"I"；然后输入<sos> I，输出"like"；接着输入<sos> I like，输出"you"；最后输入<sos> I like you，输出<eos>(结束标记)，表示生成完成。

Transformer的另一个重要设计是多层堆叠。标准的Transformer包含6个完全相同的编码器层和6个完全相同的解码器层。通过层层堆叠，模型能够逐步提取从低级到高级、从简单到复杂的语义特征，就像人类理解语言的过程一样。

# 7.2.3 编码器：理解输入序列的语义

# 7.2.3.1 概述

编码器的核心任务是理解输入序列的语义，并为每个词生成一个融合了全局上下文信息的表示。例如，对于句子"我喜欢自然语言处理"，编码器会为"我"、"喜欢"、"自然"、"语言"、"处理"每个词生成一个向量，这个向量不仅包含了词本身的含义，还包含了它在句子中的上下文信息。

编码器由多个结构完全相同的编码器层堆叠而成。每个编码器层包含两个核心子层：

多头自注意力子层(Multi-Head Self-Attention)：捕捉序列内部各位置之间的依赖关系
前馈神经网络子层(Feed-Forward Network)：对每个位置的表示进行独立的非线性变换

此外，每个子层后面都跟着残差连接和层归一化操作，这是深层神经网络能够稳定训练的关键。

# 7.2.3.2 自注意力层：Transformer的灵魂

自注意力机制是Transformer最核心的创新。它的作用是让序列中的每个词都能"看到"序列中的其他所有词，并根据它们的重要程度来更新自己的表示。

之所以叫"自"注意力，是因为注意力计算的Query、Key、Value都来自同一个序列，而不是来自不同的序列。

# 7.2.3.2.1 自注意力的计算过程

自注意力的计算可以分为四个清晰的步骤：

第一步：生成Query、Key、Value向量

首先，对于输入序列中的每个词向量 $x_{i}$ ，我们通过三个不同的可学习矩阵 $W_{q}$ 、 $W_{k}$ 、 $W_{v}$ ，将其映射为三个不同的向量：

Query(查询)：表示当前词"想要查询"什么信息
Key(键)：表示当前词"包含"什么信息
Value(值)：表示当前词的实际内容信息

这三个向量的作用可以用一个形象的比喻来理解：

想象你在图书馆找书，你脑子里想找的内容就是Query
书架上每本书的书名和简介就是Key
书里面的实际内容就是Value
你会根据Query和Key的匹配程度，决定从哪些书(Value)中提取信息

第二步：计算位置间的相关性

接下来，我们用每个位置的Query向量与所有位置的Key向量做点积，得到它们之间的相关性评分。点积的结果越大，表示两个位置越相关。

为了防止点积结果过大导致softmax函数进入梯度饱和区，我们需要将结果除以 $d_{k}$ 进行缩放，其中 $d_{k}$ 是Key向量的维度。

最终的评分公式为：

s c o r e (i, j) = \frac{q _{i} \cdot k _{j}}{d _{k}}

第三步：计算注意力权重

得到相关性评分后，我们使用softmax函数对每一行进行归一化，确保每个位置对所有位置的关注程度之和为1。这样就得到了注意力权重矩阵 $W$ 。

第四步：加权汇总生成输出

最后，我们用注意力权重对所有位置的Value向量进行加权求和，得到每个位置融合了全局信息后的新表示。

将这四个步骤合并起来，就得到了自注意力的完整公式：

A t t e n t i o n (Q, K, V) = s o f t m a x (\frac{Q K ^{T}}{d _{k}}) V

# 7.2.3.2.2 多头自注意力：从多个视角看问题

单一的自注意力机制虽然强大，但它只能从一个视角来建模序列中的依赖关系。然而，自然语言中的语义关系是复杂多样的。例如，在句子"那只动物没有过马路，因为它太累了"中：

"它"指代"那只动物"，这是指代关系
"因为"连接前后两个分句，这是因果关系
"过马路"是一个固定的动宾结构

要同时捕捉这些不同类型的关系，单一的注意力头显然是不够的。为此，Transformer引入了多头注意力机制。

多头注意力的核心思想是：将Query、Key、Value分别投影到多个不同的子空间，在每个子空间中独立计算注意力，然后将结果拼接起来。这样，不同的注意力头就可以专注于不同类型的语义关系。

例如，在上面的句子中：

一个注意力头可能专注于捕捉指代关系，让"它"更多地关注"动物"
另一个注意力头可能专注于捕捉因果关系，让"因为"连接前后两个分句
第三个注意力头可能专注于捕捉短语结构，让"过"更多地关注"马路"

最后，我们将所有注意力头的输出拼接起来，再通过一个线性变换 $W_{o}$ ，得到最终的多头注意力输出。

标准的Transformer使用8个注意力头，每个头的维度是64，拼接后总维度仍然是512，与输入维度保持一致。

# 7.2.3.3 前馈神经网络层：增强模型的表达能力

多头注意力子层虽然能够很好地建模序列中的依赖关系，但它本质上是一种线性操作。为了让模型能够学习更复杂的非线性变换，Transformer在每个编码器层中都加入了一个前馈神经网络(FFN)。

FFN对序列中的每个位置独立进行相同的非线性变换，它由两个线性层和一个ReLU激活函数组成：

F F N (x) = L i n e a r_{2} (R e L U (L i n e a r_{1} (x)))

在标准Transformer中，第一个线性层将维度从512扩展到2048，第二个线性层再将维度从2048压缩回512。这种"先升维再降维"的设计可以极大地增强模型的表达能力。

# 7.2.3.4 残差连接与层归一化：让深层网络稳定训练

当神经网络的层数增加时，会出现一个严重的问题——梯度消失。在反向传播过程中，梯度会随着层数的增加而指数级衰减，导致前面的层几乎无法更新。

为了解决这个问题，Transformer在每个子层后面都使用了残差连接(Residual Connection)。残差连接的思想非常简单：将子层的输入直接与输出相加，形成一条"捷径"。

y = x + S u b L a y e r (x)

这样，在反向传播时，梯度可以通过这条捷径直接回传到前面的层，有效缓解了梯度消失问题。

在残差连接之后，还需要进行层归一化(Layer Normalization)。层归一化的作用是将每个token的特征向量调整为均值为0、方差为1的标准正态分布，从而稳定训练过程，加速模型收敛。

层归一化的计算公式如下：

计算向量的均值： $μ = \frac{1}{d} \sum_{i = 1}^{d} x_{i}$
计算向量的标准差： $σ = \frac{1}{d} \sum_{i = 1}^{d} (x_{i} - μ)^{2}$
标准化： $\overset{x_{i}}{^} = \frac{x _{i} - μ}{σ + ϵ}$
缩放和平移： $L a y e r N o r m (x_{i}) = γ_{i} \cdot \overset{x_{i}}{^} + β_{i}$

其中 $γ$ 和 $β$ 是可学习的参数，用于在归一化的基础上进行适当的调整，保证模型的表达能力不受限制。

# 7.2.3.5 位置编码：给模型注入顺序信息

Transformer完全摒弃了RNN的顺序结构，这虽然带来了并行计算的优势，但也带来了一个严重的问题——模型无法感知词语的顺序。对于模型来说，"猫吃鱼"和"鱼吃猫"这两个句子是完全相同的，因为它们包含的词语完全一样。

为了解决这个问题，Transformer引入了位置编码(Positional Encoding)机制。位置编码为每个位置生成一个唯一的向量，然后将这个向量与词向量相加，作为模型的输入。这样，模型在处理每个词时，不仅能知道它的含义，还能知道它在句子中的位置。

位置编码的设计需要满足几个要求：

每个位置有唯一的编码
编码的数值范围要稳定，不能与词向量相差太大
能够体现位置之间的相对关系
能够泛化到训练时没有见过的更长的序列

Transformer使用了一种基于正弦和余弦函数的位置编码方式：

P E_{(p o s, 2 i)} = s i n (\frac{p o s}{1 0 0 0 0 ^{\frac{2 i}{d _{m o d e l}}}})

P E_{(p o s, 2 i + 1)} = c o s (\frac{p o s}{1 0 0 0 0 ^{\frac{2 i}{d _{m o d e l}}}})

其中：

$p o s$ 是词在序列中的位置
$i$ 是维度索引
$d_{m o d e l}$ 是词向量的维度

这种编码方式的巧妙之处在于：

偶数维度使用正弦函数，奇数维度使用余弦函数
每个维度的频率随着维度的增加而降低
任意两个位置 $p o s$ 和 $p o s + k$ 的编码之间存在线性关系，便于模型学习相对位置

为了更直观地理解正余弦位置编码的变化规律，你可以使用下面的可视化工具进行交互体验：

# 7.2.3.6 编码器小结

现在，我们可以完整地描述Transformer编码器的结构了：

输入层：词嵌入 + 位置编码
N个编码器层堆叠：
- 多头自注意力子层
  - 残差连接
  - 层归一化
- 前馈神经网络子层
  - 残差连接
  - 层归一化
输出层：最终的上下文表示

# 7.2.4 解码器：生成目标序列

# 7.2.4.1 概述

解码器的任务是根据编码器的输出，逐步生成目标序列。与编码器类似，解码器也由多个结构相同的解码器层堆叠而成。标准的Transformer包含6个解码器层。

每个解码器层比编码器层多了一个子层，总共包含三个核心子层：

Masked多头自注意力子层：建模目标序列内部的依赖关系
编码器-解码器注意力子层：建模目标序列与源序列之间的依赖关系
前馈神经网络子层：对每个位置的表示进行非线性变换

同样，每个子层后面也都跟着残差连接和层归一化操作。

# 7.2.4.2 Masked自注意力子层：防止信息泄露

解码器的第一个子层是Masked自注意力层。它与编码器中的自注意力层基本相同，只是多了一个遮盖(Mask)机制。

为什么需要遮盖机制呢？因为解码器是自回归地生成序列的，在生成第 $i$ 个词时，它只能看到前 $i - 1$ 个词，而不能看到第 $i$ 个及以后的词。如果在训练时允许模型看到未来的词，就会导致信息泄露，模型会直接"抄答案"，而不是真正学习如何生成序列。

为了解决这个问题，我们在计算注意力权重时，将当前位置对其后续位置的评分设置为 $- \infty$ 。这样，经过softmax后，这些位置的权重会趋近于0，模型就无法从未来的词中获取信息了。

遮盖机制的引入，使得Transformer可以在训练时一次性输入完整的目标序列，同时预测每个位置的词，充分利用并行计算的优势。这比RNN逐词训练的方式效率高得多。

# 7.2.4.3 编码器-解码器注意力子层：连接源序列与目标序列

解码器的第二个子层是编码器-解码器注意力层。它的作用是让解码器在生成每个词时，能够动态地关注源序列中最相关的部分。

这个子层的计算方式与自注意力基本相同，唯一的区别是：

Query来自解码器上一层的输出
Key和Value来自编码器的最终输出

也就是说，解码器使用当前的生成状态作为Query，去"查询"编码器输出的源序列表示，找到最相关的信息，然后将这些信息融合到当前的生成状态中。

这与我们在Seq2Seq模型中学过的注意力机制是完全一致的。

# 7.2.4.4 解码器小结

现在，我们可以完整地描述Transformer解码器的结构了：

输入层：目标序列词嵌入 + 位置编码
N个解码器层堆叠：
- Masked多头自注意力子层
  - 残差连接
  - 层归一化
- 编码器-解码器注意力子层
  - 残差连接
  - 层归一化
- 前馈神经网络子层
  - 残差连接
  - 层归一化
输出层：线性层 + Softmax，生成词表上的概率分布

# 7.3 模型训练与推理机制

Transformer的训练和推理都基于自回归生成机制，但两者在实现方式上有很大的不同。

# 7.3.1 模型训练

训练时，我们将完整的源序列输入编码器，将完整的目标序列输入解码器。解码器会同时预测目标序列中每个位置的词。

为了防止信息泄露，我们在解码器的Masked自注意力层中使用了遮盖机制，限制每个位置只能关注它前面的词。

这种训练方式的优势是可以充分利用并行计算，所有位置的预测可以同时进行，大大提高了训练效率。

# 7.3.2 模型推理

推理时，情况就不同了。因为我们还没有生成目标序列，所以无法一次性输入完整的目标序列。我们必须从<sos>标记开始，逐词生成目标序列。

具体过程如下：

将完整的源序列输入编码器，得到源序列的上下文表示
将<sos>输入解码器，生成第一个词的概率分布，选择概率最大的词作为第一个输出
将<sos>和第一个输出词一起输入解码器，生成第二个词的概率分布，选择概率最大的词作为第二个输出
重复这个过程，直到生成<eos>标记或达到最大长度

由于每一步的输入都依赖于前一步的输出，所以推理过程是顺序执行的，无法并行化。这也是为什么大语言模型生成文本的速度比训练慢得多的原因。

在推理时，我们可以使用不同的解码策略来生成文本，例如：

贪心搜索：每次选择概率最大的词
束搜索：每次保留概率最大的k个候选序列，最后选择整体概率最大的一个
采样：根据概率分布随机选择一个词

# 7.4 PyTorch Transformer API详解

PyTorch提供了对Transformer的官方实现，封装在torch.nn模块中。这些API已经高度优化，可以直接用于构建各种序列模型。

# 7.4.1 核心类

PyTorch的Transformer模块由以下几个核心类组成：

类名	功能
`nn.Transformer` (opens new window)	完整的Transformer架构，包含编码器和解码器
`nn.TransformerEncoder` (opens new window)	Transformer编码器，由多个编码器层堆叠而成
`nn.TransformerDecoder` (opens new window)	Transformer解码器，由多个解码器层堆叠而成
`nn.TransformerEncoderLayer` (opens new window)	单个编码器层
`nn.TransformerDecoderLayer` (opens new window)	单个解码器层

# 7.4.2 Transformer构造参数

我们可以通过以下参数来配置Transformer模型：

torch.nn.Transformer(
    d_model=512,
    nhead=8,
    num_encoder_layers=6,
    num_decoder_layers=6,
    dim_feedforward=2048,
    dropout=0.1,
    activation='relu',
    custom_encoder=None,
    custom_decoder=None,
    layer_norm_eps=1e-05,
    batch_first=False,
    norm_first=False,
    bias=True,
    device=None,
    dtype=None
)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

基础核心参数：

参数名	默认值	说明
`d_model`	512	模型的特征维度，即每个token的表示向量长度
`nhead`	8	多头注意力的头数，必须能整除`d_model`
`num_encoder_layers`	6	编码器的层数
`num_decoder_layers`	6	解码器的层数
`dim_feedforward`	2048	前馈神经网络的隐藏层维度
`dropout`	0.1	Dropout概率
`batch_first`	False	如果为True，输入输出的维度顺序为`(batch_size, seq_len, d_model)`，否则为`(seq_len, batch_size, d_model)`。推荐设置为True，更符合我们的使用习惯
`device`	None	模型运行的设备('cuda'或'cpu')
`dtype`	None	模型参数的数据类型

高级参数：

参数名	默认值	说明
`activation`	'relu'	激活函数，支持'relu'或'gelu'
`custom_encoder`	None	自定义编码器
`custom_decoder`	None	自定义解码器
`layer_norm_eps`	1e-5	层归一化的epsilon值
`norm_first`	False	如果为True，层归一化将在注意力和前馈操作之前执行
`bias`	True	如果为False，所有线性层和层归一化层都不使用偏置项

示例：创建一个基础的Transformer模型

from torch import nn

transformer = nn.Transformer(
    d_model=512,
    nhead=8,
    num_encoder_layers=6,
    num_decoder_layers=6,
    batch_first=True  # 推荐设置为True
)

1
2
3
4
5
6
7
8
9

# 7.4.3 Transformer前向传播

Transformer的forward方法接收源序列和目标序列作为输入，返回解码器的输出。

output = transformer(
    src=src_emb,          # 源序列嵌入
    tgt=tgt_emb,          # 目标序列嵌入
    src_key_padding_mask=src_pad_mask,  # 源序列填充掩码
    tgt_key_padding_mask=tgt_pad_mask,  # 目标序列填充掩码
    tgt_mask=tgt_mask,    # 目标序列因果掩码
    memory_key_padding_mask=src_pad_mask  # 编码器输出填充掩码
)

1
2
3
4
5
6
7
8

关键参数说明：

src：源序列的嵌入表示，形状为(batch_size, src_len, d_model)
tgt：目标序列的嵌入表示，形状为(batch_size, tgt_len, d_model)
src_key_padding_mask：用于屏蔽源序列中的<pad>标记，形状为(batch_size, src_len)，值为True的位置表示被屏蔽
tgt_key_padding_mask：用于屏蔽目标序列中的<pad>标记，形状为(batch_size, tgt_len)
tgt_mask：因果掩码，用于防止解码器看到未来的词，形状为(tgt_len, tgt_len)
memory_key_padding_mask：用于屏蔽编码器输出中的<pad>标记，通常与src_key_padding_mask相同

输出：

output：解码器的输出，形状为(batch_size, tgt_len, d_model)

我们可以使用generate_square_subsequent_mask方法来生成因果掩码：

tgt_mask = transformer.generate_square_subsequent_mask(tgt_len).to(device)

# 7.4.4 单独使用编码器和解码器

我们也可以单独使用编码器或解码器。例如，BERT就是只使用了Transformer编码器，而GPT就是只使用了Transformer解码器。

单独使用编码器：

# 调用编码器
memory = transformer.encoder(
    src=src_emb,
    src_key_padding_mask=src_pad_mask
)

1
2
3
4
5

单独使用解码器：

# 调用解码器
output = transformer.decoder(
    tgt=tgt_emb,
    memory=memory,  # 编码器的输出
    tgt_mask=tgt_mask,
    tgt_key_padding_mask=tgt_pad_mask,
    memory_key_padding_mask=src_pad_mask
)

1
2
3
4
5
6
7
8

# 7.5 案例实操：使用Transformer实现中英翻译

现在，我们将使用PyTorch的Transformer API来实现一个完整的中英翻译系统。这是我们的翻译模型V3.0版本，相比之前的RNN和Seq2Seq+Attention版本，它将拥有更好的性能和更快的训练速度。

# 7.5.1 需求分析

我们需要实现一个将中文句子翻译成英文句子的系统。PyTorch已经提供了完整的Transformer实现，但它没有内置位置编码模块，所以我们需要自己实现位置编码。

除此之外，我们还需要实现以下模块：

中文和英文的分词器
词嵌入层
输出层(将模型输出映射为词表大小的向量)

# 7.5.2 项目结构

我们的项目结构如下：

translation-transformer/
├── data/
│   ├── raw/          # 原始数据
│   └── processed/    # 预处理后的数据
├── models/           # 保存训练好的模型
├── logs/             # TensorBoard日志
└── src/
    ├── config.py     # 配置文件
    ├── process.py    # 数据预处理
    ├── tokenizer.py  # 分词器
    ├── dataset.py    # 数据集
    ├── model.py      # 模型定义
    ├── train.py      # 训练脚本
    ├── predict.py    # 预测脚本
    └── evaluate.py   # 评估脚本

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 7.5.3 完整代码实现

配置文件(config.py)

from pathlib import Path

# 获取项目根目录
BASE_DIR = Path(__file__).parent.parent

# 定义项目中常用路径
MODELS_DIR = BASE_DIR / 'models'  # 模型保存路径
PROCESSED_DATA_DIR = BASE_DIR / 'data' / 'processed'  # 处理后的数据保存路径
RAW_DATA_DIR = BASE_DIR / 'data' / 'raw'  # 原始数据保存路径
LOGS_DIR = BASE_DIR / 'logs'  # TensorBoard日志目录

# 模型结构参数
DIM_MODEL = 128  # 词向量维度
NUM_HEADS = 4
NUM_ENCODER_LAYERS = 2
NUM_DECODER_LAYERS = 2

# 训练相关超参数
BATCH_SIZE = 128  # 每个batch的样本数
SEQ_LEN = 30  # 序列长度(输入与输出最大长度)
LEARNING_RATE = 1e-3  # 学习率
EPOCHS = 30  # 总训练轮数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

数据预处理(process.py)

import pandas as pd
from sklearn.model_selection import train_test_split
from tokenizer import ChineseTokenizer, EnglishTokenizer
import config

def process():
    """
    数据预处理主函数。
    读取原始数据，划分训练集和测试集，构建词表，将文本转换为索引。
    """
    print('开始处理数据')
    
    # 读取cmn.txt数据集，格式为"英文\t中文"
    df = pd.read_csv(
        config.RAW_DATA_DIR / 'cmn.txt',
        sep='\t',
        header=None,
        usecols=[0, 1],
        names=['en', 'zh']
    )
    
    # 去除空值
    df = df.dropna()
    df = df[df['en'].str.strip().ne('') & df['zh'].str.strip().ne('')]
    
    # 划分训练集和测试集(8:2)
    train_df, test_df = train_test_split(df, test_size=0.2, random_state=42)
    
    # 构建英文和中文词表
    EnglishTokenizer.build_vocab(train_df['en'].tolist(), config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    ChineseTokenizer.build_vocab(train_df['zh'].tolist(), config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    
    # 加载分词器
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    
    # 将文本转换为索引
    # 英文句子需要添加<sos>和<eos>标记
    train_df['en'] = train_df['en'].apply(
        lambda x: en_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=True)
    )
    # 中文句子不需要添加特殊标记
    train_df['zh'] = train_df['zh'].apply(
        lambda x: zh_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=False)
    )
    
    test_df['en'] = test_df['en'].apply(
        lambda x: en_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=True)
    )
    test_df['zh'] = test_df['zh'].apply(
        lambda x: zh_tokenizer.encode(x, seq_len=config.SEQ_LEN, add_sos_eos=False)
    )
    
    # 保存处理后的数据
    train_df.to_json(config.PROCESSED_DATA_DIR / 'indexed_train.jsonl', orient='records', lines=True)
    test_df.to_json(config.PROCESSED_DATA_DIR / 'indexed_test.jsonl', orient='records', lines=True)
    
    print('数据处理完成')

if __name__ == '__main__':
    process()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61

分词器(tokenizer.py)

from abc import abstractmethod
from nltk import word_tokenize, TreebankWordDetokenizer
from tqdm import tqdm

class BaseTokenizer:
    """
    分词器基类，定义了分词器的通用接口。
    """
    unk_token = '<unk>'  # 未知词标记
    pad_token = '<pad>'  # 填充标记
    sos_token = '<sos>'  # 句子开始标记
    eos_token = '<eos>'  # 句子结束标记
    
    @staticmethod
    @abstractmethod
    def tokenize(sentence):
        """将句子分词为token列表"""
        pass
    
    @abstractmethod
    def decode(self, indexes):
        """将索引列表转换为句子"""
        pass
    
    @classmethod
    def build_vocab(cls, sentences, vocab_file):
        """
        从句子列表构建词表并保存到文件。
        :param sentences: 句子列表
        :param vocab_file: 词表保存路径
        """
        unique_words = set()
        for sentence in tqdm(sentences, desc='分词'):
            for word in cls.tokenize(sentence):
                unique_words.add(word)
        
        # 特殊标记放在词表最前面
        vocab_list = [cls.pad_token, cls.unk_token, cls.sos_token, cls.eos_token] + list(unique_words)
        
        with open(vocab_file, 'w', encoding='utf-8') as f:
            for word in vocab_list:
                f.write(word + '\n')
    
    def __init__(self, vocab_list):
        self.vocab_list = vocab_list
        self.vocab_size = len(vocab_list)
        self.word2index = {word: i for i, word in enumerate(vocab_list)}
        self.index2word = {i: word for i, word in enumerate(vocab_list)}
        
        # 特殊标记的索引
        self.unk_token_index = self.word2index[self.unk_token]
        self.pad_token_index = self.word2index[self.pad_token]
        self.sos_token_index = self.word2index[self.sos_token]
        self.eos_token_index = self.word2index[self.eos_token]
    
    @classmethod
    def from_vocab(cls, vocab_file):
        """从词表文件加载分词器"""
        with open(vocab_file, 'r', encoding='utf-8') as f:
            vocab_list = [line.strip() for line in f.readlines()]
        return cls(vocab_list)
    
    def encode(self, sentence, seq_len, add_sos_eos=False):
        """
        将句子编码为索引列表。
        :param sentence: 输入句子
        :param seq_len: 序列最大长度
        :param add_sos_eos: 是否添加<sos>和<eos>标记
        :return: 索引列表
        """
        tokens = self.tokenize(sentence)
        indexes = [self.word2index.get(token, self.unk_token_index) for token in tokens]
        
        if add_sos_eos:
            # 留出位置给<sos>和<eos>
            indexes = indexes[:seq_len - 2]
            indexes = [self.sos_token_index] + indexes + [self.eos_token_index]
        else:
            indexes = indexes[:seq_len]
        
        # 填充到固定长度
        if len(indexes) < seq_len:
            indexes += [self.pad_token_index] * (seq_len - len(indexes))
        
        return indexes

class ChineseTokenizer(BaseTokenizer):
    """中文字符分词器，按字符分词"""
    @staticmethod
    def tokenize(sentence):
        return list(sentence)
    
    def decode(self, indexes):
        return ''.join([self.index2word[i] for i in indexes if i not in [self.pad_token_index, self.sos_token_index, self.eos_token_index]])

class EnglishTokenizer(BaseTokenizer):
    """英文单词分词器，使用nltk的word_tokenize"""
    @staticmethod
    def tokenize(sentence):
        return word_tokenize(sentence.lower())
    
    def decode(self, indexes):
        tokens = [self.index2word[i] for i in indexes if i not in [self.pad_token_index, self.sos_token_index, self.eos_token_index]]
        return TreebankWordDetokenizer().detokenize(tokens).capitalize()

数据集(dataset.py)

import pandas as pd
import torch
from torch.utils.data import Dataset, DataLoader
import config

class TranslationDataset(Dataset):
    """
    翻译数据集，加载已编码的中英文索引。
    """
    def __init__(self, data_path):
        """
        初始化数据集。
        :param data_path: 数据文件路径(JSONL格式)
        """
        self.data = pd.read_json(data_path, lines=True).to_dict(orient='records')
    
    def __len__(self):
        """返回数据集大小"""
        return len(self.data)
    
    def __getitem__(self, index):
        """
        获取指定索引的样本。
        :return: (中文索引张量, 英文索引张量)
        """
        input_tensor = torch.tensor(self.data[index]['zh'], dtype=torch.long)
        target_tensor = torch.tensor(self.data[index]['en'], dtype=torch.long)
        return input_tensor, target_tensor

def get_dataloader(train=True):
    """
    创建数据加载器。
    :param train: 是否加载训练集
    :return: DataLoader实例
    """
    data_path = config.PROCESSED_DATA_DIR / ('indexed_train.jsonl' if train else 'indexed_test.jsonl')
    dataset = TranslationDataset(data_path)
    return DataLoader(dataset, batch_size=config.BATCH_SIZE, shuffle=True)

if __name__ == '__main__':
    # 测试数据加载器
    train_loader = get_dataloader(train=True)
    for inputs, targets in train_loader:
        print(f'输入形状: {inputs.shape}, 目标形状: {targets.shape}')
        break

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

模型定义(model.py)

import torch
from torch import nn
import config

class PositionEncoding(nn.Module):
    """
    位置编码模块，实现Transformer论文中的正弦余弦位置编码。
    """
    def __init__(self, d_model, max_len=500):
        super().__init__()
        self.d_model = d_model
        self.max_len = max_len
        
        # 预计算位置编码
        pos = torch.arange(0, self.max_len, dtype=torch.float).unsqueeze(1)  # (max_len, 1)
        _2i = torch.arange(0, self.d_model, step=2, dtype=torch.float)  # (d_model/2,)
        div_term = torch.pow(10000, _2i / self.d_model)
        
        # 偶数维度使用正弦函数
        sins = torch.sin(pos / div_term)  # (max_len, d_model/2)
        # 奇数维度使用余弦函数
        coss = torch.cos(pos / div_term)  # (max_len, d_model/2)
        
        pe = torch.zeros(self.max_len, self.d_model, dtype=torch.float)
        pe[:, 0::2] = sins
        pe[:, 1::2] = coss
        
        # 将位置编码注册为buffer，这样它会被保存到模型中，但不会被优化
        self.register_buffer('pe', pe)
    
    def forward(self, x):
        """
        将位置编码添加到输入嵌入中。
        :param x: 输入嵌入，形状为(batch_size, seq_len, d_model)
        :return: 添加了位置编码的嵌入
        """
        seq_len = x.size(1)
        return x + self.pe[:seq_len]

class TranslationModel(nn.Module):
    """
    基于Transformer的翻译模型。
    """
    def __init__(self, zh_vocab_size, en_vocab_size, zh_padding_index, en_padding_index):
        super().__init__()
        
        # 中文和英文词嵌入层
        self.src_embedding = nn.Embedding(
            num_embeddings=zh_vocab_size,
            embedding_dim=config.DIM_MODEL,
            padding_idx=zh_padding_index
        )
        self.tgt_embedding = nn.Embedding(
            num_embeddings=en_vocab_size,
            embedding_dim=config.DIM_MODEL,
            padding_idx=en_padding_index
        )
        
        # 位置编码
        self.position_encoding = PositionEncoding(d_model=config.DIM_MODEL)
        
        # Transformer模型
        self.transformer = nn.Transformer(
            d_model=config.DIM_MODEL,
            nhead=config.NUM_HEADS,
            num_encoder_layers=config.NUM_ENCODER_LAYERS,
            num_decoder_layers=config.NUM_DECODER_LAYERS,
            batch_first=True
        )
        
        # 输出层，将模型输出映射为英文词表大小的向量
        self.linear = nn.Linear(config.DIM_MODEL, en_vocab_size)
    
    def encode(self, src, src_pad_mask):
        """
        编码源序列。
        :param src: 中文索引，形状为(batch_size, src_len)
        :param src_pad_mask: 中文填充掩码，形状为(batch_size, src_len)
        :return: 编码器输出，形状为(batch_size, src_len, d_model)
        """
        src_embed = self.src_embedding(src)
        src_embed = self.position_encoding(src_embed)
        memory = self.transformer.encoder(src=src_embed, src_key_padding_mask=src_pad_mask)
        return memory
    
    def decode(self, tgt, memory, tgt_mask, tgt_pad_mask, src_pad_mask):
        """
        解码生成目标序列。
        :param tgt: 英文索引，形状为(batch_size, tgt_len)
        :param memory: 编码器输出，形状为(batch_size, src_len, d_model)
        :param tgt_mask: 英文因果掩码，形状为(tgt_len, tgt_len)
        :param tgt_pad_mask: 英文填充掩码，形状为(batch_size, tgt_len)
        :param src_pad_mask: 中文填充掩码，形状为(batch_size, src_len)
        :return: 解码器输出，形状为(batch_size, tgt_len, en_vocab_size)
        """
        tgt_embed = self.tgt_embedding(tgt)
        tgt_embed = self.position_encoding(tgt_embed)
        output = self.transformer.decoder(
            tgt=tgt_embed,
            memory=memory,
            tgt_mask=tgt_mask,
            tgt_key_padding_mask=tgt_pad_mask,
            memory_key_padding_mask=src_pad_mask
        )
        return self.linear(output)
    
    def forward(self, src, tgt, src_pad_mask, tgt_pad_mask, tgt_mask):
        """
        前向传播，用于训练。
        :param src: 中文索引，形状为(batch_size, src_len)
        :param tgt: 英文索引，形状为(batch_size, tgt_len)
        :param src_pad_mask: 中文填充掩码，形状为(batch_size, src_len)
        :param tgt_pad_mask: 英文填充掩码，形状为(batch_size, tgt_len)
        :param tgt_mask: 英文因果掩码，形状为(tgt_len, tgt_len)
        :return: 模型输出，形状为(batch_size, tgt_len, en_vocab_size)
        """
        memory = self.encode(src, src_pad_mask)
        output = self.decode(tgt, memory, tgt_mask, tgt_pad_mask, src_pad_mask)
        return output

训练脚本(train.py)

import time
import torch
from torch.nn import CrossEntropyLoss
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
from dataset import get_dataloader
from tokenizer import ChineseTokenizer, EnglishTokenizer
import config
from model import TranslationModel

def train_one_epoch(dataloader, model, loss_function, optimizer, device):
    """
    训练一个epoch。
    """
    model.train()
    total_loss = 0
    
    for src, tgt in tqdm(dataloader, desc='训练'):
        src = src.to(device)
        tgt = tgt.to(device)
        
        # 生成填充掩码
        src_pad_mask = (src == model.src_embedding.padding_idx)
        tgt_pad_mask = (tgt == model.tgt_embedding.padding_idx)
        
        # 目标序列输入是tgt[:, :-1]，目标输出是tgt[:, 1:]
        tgt_input = tgt[:, :-1]
        tgt_output = tgt[:, 1:]
        
        # 生成因果掩码
        tgt_mask = model.transformer.generate_square_subsequent_mask(tgt_input.shape[1]).to(device)
        
        # 前向传播
        optimizer.zero_grad()
        output = model(src, tgt_input, src_pad_mask, tgt_pad_mask[:, :-1], tgt_mask)
        
        # 计算损失
        loss = loss_function(
            output.reshape(-1, output.shape[-1]),
            tgt_output.reshape(-1)
        )
        
        # 反向传播和优化
        loss.backward()
        optimizer.step()
        
        total_loss += loss.item()
    
    return total_loss / len(dataloader)

def train():
    """
    主训练函数。
    """
    # 设备配置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    print(f'使用设备: {device}')
    
    # 数据加载器
    dataloader = get_dataloader()
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 创建模型
    model = TranslationModel(
        zh_vocab_size=zh_tokenizer.vocab_size,
        en_vocab_size=en_tokenizer.vocab_size,
        zh_padding_index=zh_tokenizer.pad_token_index,
        en_padding_index=en_tokenizer.pad_token_index
    ).to(device)
    
    # 损失函数和优化器
    # 忽略填充标记的损失
    loss_function = CrossEntropyLoss(ignore_index=en_tokenizer.pad_token_index)
    optimizer = torch.optim.Adam(model.parameters(), lr=config.LEARNING_RATE)
    
    # TensorBoard日志
    writer = SummaryWriter(log_dir=config.LOGS_DIR / time.strftime('%Y-%m-%d_%H-%M-%S'))
    
    best_loss = float('inf')
    
    # 训练循环
    for epoch in range(1, config.EPOCHS + 1):
        print(f'\n========== Epoch {epoch}/{config.EPOCHS} ==========')
        
        avg_loss = train_one_epoch(dataloader, model, loss_function, optimizer, device)
        print(f'平均损失: {avg_loss:.4f}')
        
        # 记录损失
        writer.add_scalar('Loss/train', avg_loss, epoch)
        
        # 保存最佳模型
        if avg_loss < best_loss:
            best_loss = avg_loss
            torch.save(model.state_dict(), config.MODELS_DIR / 'model.pt')
            print(f'模型已保存，最佳损失: {best_loss:.4f}')
        else:
            print('未保存模型')
    
    print('\n训练完成！')
    writer.close()

if __name__ == '__main__':
    train()

预测脚本(predict.py)

import torch
from tokenizer import ChineseTokenizer, EnglishTokenizer
from model import TranslationModel
import config

def predict_batch(input_tensor, model, en_tokenizer, device):
    """
    批量生成翻译结果。
    :param input_tensor: 中文输入张量，形状为(batch_size, seq_len)
    :param model: 翻译模型
    :param en_tokenizer: 英文分词器
    :param device: 设备
    :return: 生成的英文索引列表
    """
    model.eval()
    
    with torch.no_grad():
        # 编码源序列
        src_pad_mask = (input_tensor == 0)
        memory = model.encode(src=input_tensor, src_pad_mask=src_pad_mask)
        
        batch_size = input_tensor.shape[0]
        
        # 初始化解码器输入，以<sos>开始
        decoder_input = torch.full(
            size=(batch_size, 1),
            fill_value=en_tokenizer.sos_token_index,
            device=device
        )
        
        generated = [[] for _ in range(batch_size)]
        finished = [False for _ in range(batch_size)]
        
        # 逐词生成
        for step in range(1, config.SEQ_LEN):
            # 生成因果掩码
            tgt_mask = model.transformer.generate_square_subsequent_mask(decoder_input.shape[1]).to(device)
            tgt_pad_mask = (decoder_input == en_tokenizer.pad_token_index)
            
            # 解码
            decoder_output = model.decode(
                decoder_input, 
                memory, 
                tgt_mask, 
                tgt_pad_mask, 
                src_pad_mask
            )
            
            # 取最后一个位置的输出，预测下一个词
            predict_indexes = decoder_output[:, -1, :].argmax(dim=-1)
            
            # 更新生成结果
            for i in range(batch_size):
                if finished[i]:
                    continue
                
                if predict_indexes[i].item() == en_tokenizer.eos_token_index:
                    finished[i] = True
                    continue
                
                generated[i].append(predict_indexes[i].item())
            
            # 如果所有句子都生成完成，提前退出
            if all(finished):
                break
            
            # 将预测的词添加到解码器输入中
            decoder_input = torch.cat([decoder_input, predict_indexes.unsqueeze(1)], dim=1)
        
        return generated

def predict(zh_sentence, model, zh_tokenizer, en_tokenizer, device):
    """
    翻译单个中文句子。
    :param zh_sentence: 中文句子
    :param model: 翻译模型
    :param zh_tokenizer: 中文分词器
    :param en_tokenizer: 英文分词器
    :param device: 设备
    :return: 英文翻译
    """
    # 将中文句子编码为索引
    input_ids = zh_tokenizer.encode(zh_sentence, seq_len=config.SEQ_LEN, add_sos_eos=False)
    input_tensor = torch.tensor([input_ids], device=device)
    
    # 生成翻译
    generated = predict_batch(input_tensor, model, en_tokenizer, device)
    en_indexes = generated[0]
    
    # 将索引转换为句子
    en_sentence = en_tokenizer.decode(en_indexes)
    return en_sentence

def run_predict():
    """
    启动交互式翻译界面。
    """
    # 设备配置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 加载模型
    model = TranslationModel(
        zh_vocab_size=zh_tokenizer.vocab_size,
        en_vocab_size=en_tokenizer.vocab_size,
        zh_padding_index=zh_tokenizer.pad_token_index,
        en_padding_index=en_tokenizer.pad_token_index
    ).to(device)
    
    model.load_state_dict(torch.load(config.MODELS_DIR / 'model.pt', map_location=device))
    
    print('欢迎使用中英翻译系统！')
    print('请输入中文句子，输入"q"或"quit"退出。\n')
    
    while True:
        user_input = input('中文: ')
        
        if user_input.lower() in ['q', 'quit']:
            print('谢谢使用，再见！')
            break
        
        if not user_input.strip():
            print('请输入有效的中文句子。\n')
            continue
        
        # 翻译
        result = predict(user_input, model, zh_tokenizer, en_tokenizer, device)
        print(f'英文: {result}\n')

if __name__ == '__main__':
    run_predict()

评估脚本(evaluate.py)

import torch
from nltk.translate.bleu_score import corpus_bleu
from tqdm import tqdm
import config
from tokenizer import ChineseTokenizer, EnglishTokenizer
from model import TranslationModel
from dataset import get_dataloader
from predict import predict_batch

def evaluate(dataloader, model, zh_tokenizer, en_tokenizer, device):
    """
    在测试集上评估模型，计算BLEU分数。
    """
    all_references = []
    all_predictions = []
    
    special_tokens = [
        zh_tokenizer.pad_token_index, 
        zh_tokenizer.eos_token_index, 
        zh_tokenizer.sos_token_index
    ]
    
    for src, tgt in tqdm(dataloader, desc="评估"):
        src = src.to(device)
        tgt = tgt.tolist()
        
        # 生成预测
        predict_indexes = predict_batch(src, model, en_tokenizer, device)
        all_predictions.extend(predict_indexes)
        
        # 处理参考译文
        for indexes in tgt:
            # 去除特殊标记
            indexes = [index for index in indexes if index not in special_tokens]
            all_references.append([indexes])
    
    # 计算BLEU分数
    bleu = corpus_bleu(all_references, all_predictions)
    return bleu

def run_evaluate():
    """
    运行评估。
    """
    # 设备配置
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    
    # 加载分词器
    zh_tokenizer = ChineseTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'zh_vocab.txt')
    en_tokenizer = EnglishTokenizer.from_vocab(config.PROCESSED_DATA_DIR / 'en_vocab.txt')
    
    # 加载模型
    model = TranslationModel(
        zh_vocab_size=zh_tokenizer.vocab_size,
        en_vocab_size=en_tokenizer.vocab_size,
        zh_padding_index=zh_tokenizer.pad_token_index,
        en_padding_index=en_tokenizer.pad_token_index
    ).to(device)
    
    model.load_state_dict(torch.load(config.MODELS_DIR / 'model.pt', map_location=device))
    
    # 加载测试数据
    dataloader = get_dataloader(train=False)
    
    # 评估
    bleu = evaluate(dataloader, model, zh_tokenizer, en_tokenizer, device)
    print(f'\n测试集BLEU分数: {bleu:.4f}')

if __name__ == '__main__':
    run_evaluate()

# 7.5.4 运行说明

准备数据：下载cmn.txt数据集，放在data/raw/目录下
预处理数据：运行python src/process.py
训练模型：运行python src/train.py
评估模型：运行python src/evaluate.py
交互式翻译：运行python src/predict.py

# 7.6 本章小结

Transformer是NLP领域的革命性架构，它彻底摒弃了RNN的循环结构，仅使用注意力机制来建模序列依赖关系。这一设计带来了三大优势：

并行计算能力：可以同时处理序列中的所有位置，训练速度大幅提升
长距离依赖建模：任意两个位置之间可以直接建立联系
可扩展性：模型可以堆叠得很深，能够学习更复杂的语义特征

Transformer的核心组件包括：

多头自注意力机制：从多个视角建模序列内部的依赖关系
前馈神经网络：增强模型的非线性表达能力
残差连接和层归一化：让深层网络能够稳定训练
位置编码：给模型注入序列的顺序信息

Transformer的编码器-解码器架构非常适合机器翻译等序列到序列任务。在本章的案例中，我们使用PyTorch的官方API实现了一个完整的中英翻译系统，展示了Transformer的强大能力。

然而，Transformer的真正威力远不止于此。在下一章中，我们将学习基于Transformer构建的预训练语言模型，如BERT、GPT和T5，它们将NLP带入了一个全新的时代。

# 第八章预训练模型：NLP领域的"通用发动机"

如果你曾尝试从零训练一个情感分析模型，一定会对标注数据的高昂成本和模型收敛的缓慢深有体会——可能需要数万条标注样本，训练数天才能得到一个勉强能用的模型。而预训练模型的出现彻底改变了这一局面：它就像一个已经完成"大学通识教育"的通用语言专家，你只需要用少量专业数据对它进行"岗位培训"(微调)，就能让它胜任各种NLP任务。

本章我们将系统介绍NLP领域三大经典预训练模型架构——GPT、BERT和T5，带你理解它们的核心思想、结构差异和适用场景，并通过一个完整的情感分析实战项目，教你如何使用HuggingFace工具链快速落地预训练模型。

# 8.1 预训练模型概述

# 8.1.1 传统NLP方法的致命局限

在预训练模型出现之前，NLP领域长期采用"任务专属模型"的开发模式：针对每个具体任务(如情感分析、命名实体识别、机器翻译)，都需要从零开始设计模型结构、收集标注数据并训练模型。这种模式存在两个无法克服的痛点：

知识无法复用：每个模型都要从头学习语言的基本规律(如语法、语义、常识)，导致训练成本极高、开发周期极长。就像每个专业的学生都要从小学一年级重新学起。
强依赖高质量标注：在医疗、法律、金融等专业领域，标注数据不仅获取困难，而且需要专业人士参与，成本动辄数十万甚至上百万。

# 8.1.2 "预训练+微调"：NLP的革命性范式

为了解决上述问题，研究者提出了"预训练+微调"的两阶段建模范式，这一范式彻底改变了NLP领域的发展轨迹：

预训练阶段：在包含数十亿甚至上万亿词的大规模无标注语料上，训练一个通用语言模型。这个阶段的目标是让模型学习自然语言的通用规律——包括词汇含义、语法规则、上下文依赖关系，甚至基本的世界常识。
微调阶段：将预训练好的通用模型迁移到具体下游任务。此时只需要少量标注数据，对模型进行少量参数调整或添加一个简单的任务输出层，就能让模型在该任务上达到甚至超过从零训练的效果。

这一方法不仅显著提升了模型的通用性和开发效率，还让低资源任务(即标注数据稀少的任务)也能获得出色的表现。如今，"预训练+微调"已成为NLP领域的绝对主流技术路线，广泛应用于文本分类、问答系统、机器翻译、对话生成、文本摘要等几乎所有NLP任务。

# 8.2 预训练模型分类

几乎所有现代预训练语言模型都构建在Transformer架构之上。相较于传统的循环神经网络(RNN/LSTM)，Transformer具有四大核心优势：

并行计算效率高：Transformer的自注意力机制可以同时处理序列中的所有token，而RNN必须按顺序逐个处理，这使得Transformer能充分利用GPU的并行计算能力，训练速度提升数十倍。
长距离依赖建模能力强：自注意力机制可以直接建立序列中任意两个token之间的联系，而RNN需要通过多个时间步传递信息，容易出现长距离依赖消失的问题。
结构通用灵活：Transformer的模块化设计使其可以轻松适配各种NLP任务，无论是理解类任务还是生成类任务。
易于扩展与迁移：Transformer的性能可以通过简单地增加层数、隐藏维度或注意力头数来提升，并且预训练好的参数可以轻松迁移到不同任务和领域。

根据Transformer组件的使用方式不同，预训练模型可以分为三大类，它们各自擅长不同的任务类型：

# 8.2.1 解码器(Decoder-only)模型："听上文猜下文"的生成专家

仅使用Transformer的解码器部分，代表模型为GPT(Generative Pre-trained Transformer)系列。这类模型采用自回归语言建模目标，即训练模型根据已经看到的左侧上下文预测下一个词。

这种架构天然擅长文本生成任务，如写文章、写代码、对话聊天等。它生成的文本流畅自然，但只能利用单向上下文信息，在理解类任务上的表现略逊于编码器模型。

# 8.2.2 编码器(Encoder-only)模型："读全文懂意思"的理解专家

仅使用Transformer的编码器部分，代表模型为BERT(Bidirectional Encoder Representations from Transformers)系列。这类模型采用掩码语言建模目标，即随机遮盖输入序列中的部分token，训练模型根据左右两侧的完整上下文预测被遮盖的词。

这种架构能够同时利用双向上下文信息，因此在自然语言理解类任务上表现出色，如文本分类、情感分析、命名实体识别、句子匹配、问答系统等。

# 8.2.3 编码器-解码器(Encoder-Decoder)模型："读完输入写输出"的转换专家

同时使用Transformer的编码器和解码器部分，代表模型为T5(Text-to-Text Transfer Transformer)系列。这类模型结合了编码器的理解能力和解码器的生成能力，编码器负责理解输入文本的语义，解码器负责根据编码器的输出生成目标文本。

这种架构最擅长序列到序列的转换任务，如机器翻译、文本摘要、问答生成等。它的通用性最强，可以处理几乎所有类型的NLP任务。

自GPT、BERT和T5等模型发布以来，基于Transformer的预训练模型进入了爆发式发展阶段。下图展示了2018年至2023年间具有里程碑意义的预训练模型及其发展脉络：

# 8.3 主流预训练模型详解

# 8.3.1 GPT：第一个成功的"预训练+微调"范式

# 8.3.1.1 概述

GPT(Generative Pre-trained Transformer)由OpenAI于2018年6月提出，论文题为《Improving Language Understanding by Generative Pre-Training》。它是第一个系统性证明"预训练+微调"范式可以在广泛NLP任务上取得SOTA(State-of-the-Art)效果的模型。

GPT的核心思想是：先在大规模无监督语料上进行生成式语言建模预训练，让模型学习自然语言的通用语法、语义和上下文依赖能力；然后通过微调将预训练好的知识迁移到具体下游任务。

虽然GPT-1本身的规模不大(1.17亿参数)，但它首次展示了生成式语言模型的强大迁移能力，为后续GPT-2、GPT-3乃至ChatGPT的发展奠定了坚实基础。

# 8.3.1.2 模型结构

GPT基于Transformer的解码器结构，但对原始Transformer解码器做了简化：它去掉了解码器中的编码器-解码器注意力层，只保留了掩码多头自注意力层和前馈网络。GPT的整体结构如下图所示：

各部分的详细设计如下：

输入嵌入层(Text & Position Embedding)

与原始Transformer一样，每个输入token的表示由两部分相加组成：

Text Embedding：将词或子词映射为固定维度的向量，GPT-1使用了40000词的词表。
Position Embedding：提供词在序列中的位置信息。与原始Transformer使用固定的正弦/余弦位置编码不同，GPT采用可学习的位置嵌入——每个位置对应一个可训练的向量，模型可以在训练过程中自动优化这些向量，从而更好地捕捉不同位置的语义关系。

每个token的最终表示是词嵌入与位置嵌入的向量和，维度为768。

解码器层

GPT-1的解码器部分由12个结构相同的解码器层堆叠而成，每个解码器层包含两个子层：

掩码多头自注意力层(12头)：使用因果掩码(Causal Mask)，确保每个token只能关注它之前的token，而不能看到未来的token。这是自回归语言建模的核心要求。
前馈网络：由两个线性层组成，中间使用GELU激活函数，隐藏层维度为3072。

每个子层都采用了"Layer Normalization + 残差连接"的结构，即LayerNorm(x + SubLayer(x))。

输出层

根据任务不同，GPT模型的输出可以接入不同的任务头：

文本预测头：用于预训练阶段的下一个词预测。它将解码器输出的隐藏状态映射为词表大小的概率分布，经过Softmax后得到每个词的预测概率。
任务分类头：用于微调阶段的下游任务。通过提取特定位置(通常是最后一个token)的隐藏状态，经过线性层映射为任务所需的标签。

# 8.3.1.3 预训练

GPT的预训练目标是生成式语言建模，即给定一个词序列 $(x_{1}, x_{2}, . . ., x_{n})$ ，模型需要最大化以下对数似然：

$L = \sum_{i = 1}^{n} lo g P (x_{i} ∣ x_{1}, x_{2}, . . ., x_{i - 1}; θ)$

其中( $θ$ )是模型的参数。简单来说，就是训练模型根据前面所有的词，准确预测下一个词。

这种自监督学习方式不需要任何人工标注，训练样本可以直接从原始文本中自动构建，极大地降低了数据成本。GPT-1使用了BooksCorpus语料库进行预训练，该语料包含7000多本未出版的小说，总规模约8亿词。小说文本具有完整的情节和长上下文，非常适合训练模型的长距离依赖建模能力。

下图直观地展示了GPT的语言建模任务：

# 8.3.1.4 微调

GPT的微调阶段是在完成预训练后，使用有监督的任务数据对模型进行端到端训练。微调的核心思路是：在保留预训练语言建模能力的基础上，利用少量标注数据调整模型参数，使其适应具体下游任务。

GPT的一大创新是统一输入格式设计：无论是什么类型的下游任务，都将其转换为连续的文本序列输入。这样同一个模型可以以完全相同的方式处理各种不同的任务，大大提高了模型的通用性。

下图展示了GPT在四类典型任务上的微调逻辑：

我们以最常见的文本分类任务为例，详细说明GPT的微调过程：

假设我们有一个二分类的情感分析数据集，每条样本包含一句评论和对应的标签(0表示负面，1表示正面)：

第一步：将每条评论转换为token序列，并添加特殊标记[Start](序列开始)和[Extract](提取特征位置)，形成模型的标准输入格式：

第二步：将转换后的序列送入GPT模型。模型经过12层解码器处理后，输出每个位置的隐藏状态。我们只提取序列最后一个位置[Extract]对应的隐藏状态，因为它包含了整个序列的上下文信息。

第三步：将[Extract]的隐藏状态送入新添加的线性分类头，得到二分类的logits，再通过Softmax转换为概率分布，最终输出预测标签。整个过程如下图所示：

通过这种方式，GPT在保留预训练模型结构和参数的基础上，仅添加极少量新参数(线性分类头)，便可高效完成从通用语言建模到具体下游任务的迁移。

# 8.3.2 BERT：双向上下文理解的里程碑

# 8.3.2.1 概述

BERT(Bidirectional Encoder Representations from Transformers)由Google于2018年10月提出，论文题为《BERT: Pre-training of Deep Bidirectional Transformers for Language Understanding》。它的出现是NLP领域的一个里程碑事件，在11个NLP基准任务上刷新了SOTA记录，推动整个领域全面进入"预训练+微调"时代。

BERT的核心创新在于采用了Transformer的编码器结构，通过双向自注意力机制，在建模每个token的表示时同时整合左右两个方向的上下文信息。这是它与GPT最本质的区别——GPT只能看到左侧上下文，而BERT能看到完整的上下文，因此能更准确地理解词的语义。

例如，对于句子"我爱吃苹果"和"苹果发布了新手机"，GPT只能根据"我爱吃"预测"苹果"，或根据"苹果"预测"发布了"，但无法同时利用左右上下文区分"苹果"的不同含义；而BERT可以同时看到"爱吃"和"发布了"，从而准确判断"苹果"在不同语境下的语义。

下图直观地对比了ELMo、GPT和BERT的上下文建模能力：

# 8.3.2.2 模型结构

BERT完全基于标准的Transformer编码器构建，为了满足不同场景的需求，Google提供了两种模型规模：

具体参数规格如下表所示：

模型版本	层数(Layers)	模型维度(d_model)	注意力头数(Heads)	参数量
BERT-base	12	768	12	1.1亿
BERT-large	24	1024	16	3.4亿

BERT模型各部分的结构细节如下：

输入表示层

BERT的每个输入token表示由三部分嵌入相加组成，这是它与原始Transformer编码器的主要区别之一：

Token Embedding：词本身的语义表示，使用30000词的WordPiece词表。
Position Embedding：表示token在序列中的位置，为可学习向量，支持最长512个token的序列。
Segment Embedding：用于区分句子对任务中的两个句子，用两个不同的可学习向量表示句子A和句子B。

这种三部分嵌入的设计使得BERT可以灵活处理单句和句子对两种输入形式，如下图所示：

此外，BERT输入中包含两个重要的特殊符号：

[CLS]：句首标志，它的输出向量在预训练时被训练用来聚合整个序列的语义信息，因此在微调阶段常用于文本分类等序列级任务。
[SEP]：句间分隔符，出现在每个句子的末尾，用于分隔句子对中的两个句子。

编码器层

BERT的编码器层与原始Transformer编码器完全相同，每个编码器层包含两个子层：

多头自注意力层：没有因果掩码，可以同时关注序列中的所有token，实现双向上下文建模。
前馈网络：结构与GPT相同，使用GELU激活函数。

每个子层同样采用"Layer Normalization + 残差连接"的结构。

输出层

根据下游任务的类型，BERT可以接入不同的任务输出头：

Token-Level任务(如命名实体识别、分词)：使用每个位置的输出表示进行分类。
Sequence-Level任务(如文本分类、句子对分类)：使用特殊token [CLS] 的输出表示进行分类。

# 8.3.2.3 预训练

BERT的预训练阶段包含两个核心自监督任务：掩码语言模型(Masked Language Modeling, MLM) 和 下一句预测(Next Sentence Prediction, NSP)，分别用于学习词级语义和句间逻辑关系。

1. 掩码语言模型(MLM)

为了实现双向语言建模，BERT没有采用传统的从左到右预测方式，而是引入了掩码语言模型任务。在训练过程中，BERT会随机遮盖输入序列中约15%的token，然后训练模型根据完整的上下文预测被遮盖的词。

为了缓解预训练和微调之间的分布不一致问题(微调时没有[MASK] token)，BERT采用了混合遮盖策略：

80%的被遮盖token替换为特殊符号[MASK]；
10%的被遮盖token替换为词表中的随机词；
10%的被遮盖token保持原词不变。

这种策略让模型在预训练时既能学习双向上下文信息，又能适应微调时的真实输入分布。下图展示了MLM任务的示例：

2. 下一句预测(NSP)

为了提升模型理解句间关系的能力，BERT引入了下一句预测任务。训练时，模型接收两个句子A和B，需要判断B是否是A在原文中的真实下一句。具体构造方式为：

50%的训练样本是上下文中真实相邻的句子(正例)；
50%的训练样本是从语料中随机采样的非相邻句子(反例)。

例如：

正例：A：我今天很忙。B：所以没去上班。
反例：A：我今天很忙。B：天气很好。

在预训练时，BERT同时优化MLM和NSP两个任务的损失，具体过程如下图所示：

注：后续研究(如RoBERTa)发现，NSP任务对大多数下游任务的帮助有限，甚至可能引入噪声，因此许多后续模型(如RoBERTa、ALBERT)都去掉了NSP任务。

BERT使用了BooksCorpus(8亿词)和英文维基百科(25亿词)的混合语料进行预训练，总规模约33亿词。

# 8.3.2.4 微调

在预训练完成后，BERT可以通过少量微调适配几乎所有类型的NLP任务。微调时，模型主体结构保持不变，仅在顶部添加一个任务特定的输出层，并使用下游任务数据对整个模型进行端到端训练。

BERT的输入格式在微调阶段与预训练阶段完全一致，仍以token序列为输入，使用[CLS]和[SEP]等特殊符号。不同任务的差异主要体现在输出层设计和从模型输出中提取哪些表示进行预测。

下面我们详细介绍BERT在四类典型任务中的微调方式：

(a) 句子对分类任务

这类任务的目标是判断两个句子之间的关系，如语义等价、蕴含、矛盾等。

输入格式：[CLS] 句子1 [SEP] 句子2 [SEP]
输出方式：使用[CLS]的输出向量接入线性层进行分类。

常见的句子对分类任务包括：

MNLI：多类别自然语言推理(蕴含、矛盾、中性)
QQP：Quora问句语义重复判断
QNLI：判断句子是否包含问题的答案
STS-B：语义相似度回归
MRPC：句子复述判断
RTE：二分类自然语言推理
SWAG：多项选择填句任务

(b) 单句分类任务

这类任务的目标是对单个句子进行分类，如情感分析、语法可接受性判断等。

输入格式：[CLS] 句子 [SEP]
输出方式：同样使用[CLS]的输出向量接入线性层进行分类。

常见的单句分类任务包括：

SST-2：电影评论情感极性判断(二分类)
CoLA：句子语法可接受性判断(二分类)

(c) 抽取式问答任务

这类任务的目标是从给定的段落中抽取问题的答案。

输入格式：[CLS] 问题 [SEP] 段落 [SEP]
输出方式：不使用[CLS]向量，而是对段落中的每个token分别预测其作为答案起始位置和结束位置的概率。最终根据得分最高的起始和结束位置，从段落中抽取连续的答案文本。

最经典的抽取式问答数据集是SQuAD v1.1。

(d) 序列标注任务

这类任务的目标是对序列中的每个token进行分类，如命名实体识别、分词、词性标注等。

输入格式：[CLS] 句子 [SEP]
输出方式：对每个token(除了特殊符号)的输出向量单独进行分类。

最常见的序列标注任务是命名实体识别(NER)，用于识别文本中的人名、地名、组织名等实体。

# 8.3.3 T5：统一文本到文本的通用框架

# 8.3.3.1 概述

T5(Text-to-Text Transfer Transformer)由Google Research于2019年10月提出，2020年正式发表在《Journal of Machine Learning Research》上，论文题为《Exploring the Limits of Transfer Learning with a Unified Text-to-Text Transformer》。

T5的核心思想是将所有自然语言处理任务统一表示为"文本到文本"的转换问题。也就是说，无论是什么任务——翻译、分类、问答、摘要、生成——模型的输入和输出都是自然语言形式的字符串。

这一设计是NLP领域的又一次范式升级：之前的模型(如GPT、BERT)需要为不同任务设计不同的输出头和损失函数，而T5可以通过同一个模型架构、同一套预训练机制、同一个损失函数完成所有任务。这种极致的统一性大大提高了模型的通用性和可扩展性。

下图直观地展示了T5的文本到文本框架：

# 8.3.3.2 模型结构

T5完全基于原始的Transformer编码器-解码器架构，没有引入太多新的结构设计，而是通过大量的实验对比，对Transformer的各个组件进行了细致的优化。T5的主要结构改进包括：

使用相对位置编码代替绝对位置编码，模型根据token之间的相对距离而不是绝对位置计算注意力得分，这使得模型对序列长度的泛化能力更强。
简化了Layer Normalization，去掉了偏置项，将Layer Normalization放在残差连接的外部。
使用GELU激活函数代替ReLU。
输入和输出共享词嵌入矩阵。

T5提供了五种不同规模的模型，从60M参数的T5-Small到11B参数的T5-11B，以满足不同的计算资源和性能需求。

# 8.3.3.3 预训练

T5的预训练目标被称为损坏跨度预测(Corrupted Span Prediction)，它结合了BERT的掩码语言建模和生成式预训练的优点，具体过程如下：

随机选择输入文本中的若干连续片段(span)进行遮盖，每个片段的长度服从几何分布，平均长度为3；
将每个被遮盖的连续片段替换为一个唯一的特殊哨兵token(如<X>、<Y>等)；
令模型学习生成这些被遮盖片段的内容，输出序列由哨兵token和对应的被遮盖文本交替组成。

下图展示了损坏跨度预测任务的示例：

与BERT的MLM相比，T5的预训练目标有两个显著优势：

计算效率更高：模型只需要生成被遮盖的片段，而不需要预测所有token，大大减少了计算量。
学习更长的语义单元：通过遮盖连续的片段，模型需要学习更长的语义片段和语言结构，而不仅仅是单个词的语义。

为了训练T5，Google构建了一个全新的大规模清洁语料库C4(Colossal Clean Crawled Corpus)，该语料库从Common Crawl网页数据中提取，经过严格的过滤和清洗，总规模约750GB，包含约1万亿个词。

# 8.3.3.4 微调

T5的微调阶段需要将所有任务转换为"文本到文本"的形式。具体来说，就是在输入文本前添加一个任务前缀，告诉模型当前要执行什么任务，然后让模型生成对应的输出文本。

下表展示了几种常见任务的T5输入输出格式：

任务类型	输入形式	目标输出
英德翻译	translate English to German: That is good.	Das ist gut.
情感分类	sentiment: This movie was great.	positive
问答	question: What is the capital of France? context: France is a country in Western Europe. Its capital is Paris.	Paris
文本摘要	summarize: Artificial intelligence is transforming every industry. It has applications in healthcare, finance, transportation, and many other fields. AI systems can now perform tasks that previously required human intelligence.	AI is transforming many industries.

这种统一的格式使得T5可以无缝切换不同任务，甚至可以在一个模型上同时训练多个任务。

# 8.4 HuggingFace快速入门：预训练模型的"瑞士军刀"

HuggingFace是目前全球最流行的NLP开源平台，它提供了丰富的预训练模型、工具链和社区资源，让开发者可以用几行代码就能加载和使用最先进的预训练模型。

HuggingFace的核心工具库包括：

Transformers：提供了超过10万个预训练模型，支持几乎所有主流NLP任务和模型架构(GPT、BERT、T5、Llama等)。
Datasets：提供了超过1万个NLP数据集的一键加载和处理工具。
Accelerate：简化分布式训练和混合精度训练的代码编写。
Tokenizers：提供了高性能的分词器实现，速度比纯Python实现快数十倍。

# 8.4.1 国内镜像使用方法

由于网络原因，国内用户直接访问HuggingFace官网可能会遇到下载缓慢或失败的问题。推荐使用国内镜像站hf-mirror.com，它是HuggingFace的公益镜像，同步了官网的所有模型和数据集。

方法一：设置环境变量(推荐)

在运行Python代码前，先设置环境变量HF_ENDPOINT，HuggingFace工具链会自动使用镜像站下载：

# Linux/MacOS
export HF_ENDPOINT=https://hf-mirror.com

# Windows PowerShell
$env:HF_ENDPOINT = "https://hf-mirror.com"

1
2
3
4
5

建议将这行命令写入~/.bashrc或~/.zshrc文件，这样每次打开终端都会自动生效。

方法二：使用huggingface-cli命令行工具

首先安装huggingface_hub库：

pip install -U huggingface_hub

然后设置环境变量并下载模型/数据集：

# 下载模型
huggingface-cli download --resume-download bert-base-chinese --local-dir bert-base-chinese

# 下载数据集
huggingface-cli download --repo-type dataset --resume-download imdb --local-dir imdb

1
2
3
4
5

# 8.4.2 一行代码体验预训练模型

HuggingFace提供了pipeline接口，可以用一行代码完成常见的NLP任务，无需了解模型的内部细节：

from transformers import pipeline

# 加载中文情感分析pipeline
classifier = pipeline("sentiment-analysis", model="bert-base-chinese")

# 进行预测
result = classifier("这家餐厅的菜太好吃了，服务也很周到！")
print(result)
# 输出: [{'label': 'positive', 'score': 0.9998675584793091}]

result = classifier("这个产品质量很差，用了一天就坏了。")
print(result)
# 输出: [{'label': 'negative', 'score': 0.9997826218605042}]

1
2
3
4
5
6
7
8
9
10
11
12
13

除了情感分析，pipeline还支持文本生成、问答、命名实体识别、翻译等数十种任务。

# 8.5 案例实操：基于BERT的评论情感分析系统(AI智评V3.0)

在本节中，我们将从零开始构建一个完整的中文评论情感分析系统，使用HuggingFace工具链和预训练的bert-base-chinese模型。这个系统可以自动判断用户评论是正面还是负面，并给出置信度。

# 8.5.1 需求说明

我们将使用公开的"在线购物评论数据集"，该数据集包含10个类别的电商商品评论，每条评论都标注了情感标签(0表示负面，1表示正面)。我们的目标是训练一个模型，能够准确预测新评论的情感倾向。

# 8.5.2 项目结构

我们采用模块化的项目结构，将数据处理、模型定义、训练、评估和预测等功能分离到不同的文件中，便于维护和扩展：

ai-review-analyzer/
├── data/
│   ├── raw/                 # 原始数据目录
│   │   └── online_shopping_10_cats.csv
│   └── processed/           # 处理后的数据目录
├── models/                  # 训练好的模型保存目录
├── logs/                    # TensorBoard日志目录
├── pretrained/              # 本地预训练模型目录
│   └── bert-base-chinese/
├── src/
│   ├── config.py            # 配置文件
│   ├── process.py           # 数据预处理
│   ├── dataset.py           # 数据集和数据加载器
│   ├── model.py             # 模型定义
│   ├── train.py             # 模型训练
│   ├── evaluate.py          # 模型评估
│   └── predict.py           # 模型预测
└── requirements.txt         # 依赖包列表

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

项目结构可视化：

# 8.5.3 环境依赖

首先安装所需的Python依赖包：

pip install torch transformers datasets scikit-learn tensorboard tqdm

# 8.5.4 完整代码实现

# 1. 配置文件(config.py)

统一管理项目的所有路径和超参数，便于后续调整：

# config.py
from pathlib import Path

# 项目根目录
BASE_DIR = Path(__file__).parent.parent

# 路径设置
MODELS_DIR = BASE_DIR / 'models'                  # 模型参数保存路径
PROCESSED_DATA_DIR = BASE_DIR / 'data' / 'processed'  # 已处理数据存放路径
RAW_DATA_DIR = BASE_DIR / 'data' / 'raw'          # 原始CSV数据路径
LOGS_DIR = BASE_DIR / 'logs'                      # TensorBoard日志保存路径
PRE_TRAINED_DIR = BASE_DIR / 'pretrained'         # 本地预训练模型存放路径

# 训练超参数
SEQ_LEN = 128         # 最大序列长度
BATCH_SIZE = 128      # 批处理大小
LEARNING_RATE = 1e-5  # 学习率
EPOCHS = 10           # 训练轮数(根据验证集效果调整)

# 创建所有目录
for dir_path in [MODELS_DIR, PROCESSED_DATA_DIR, RAW_DATA_DIR, LOGS_DIR, PRE_TRAINED_DIR]:
    dir_path.mkdir(parents=True, exist_ok=True)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

# 2. 数据预处理(process.py)

加载原始CSV数据，进行清洗、划分和分词，然后保存为标准的Dataset格式：

# process.py
from datasets import load_dataset, ClassLabel
from transformers import AutoTokenizer
import config

def process_data():
    # 1. 加载原始CSV数据
    dataset = load_dataset(
        'csv', 
        data_files=str(config.RAW_DATA_DIR / 'online_shopping_10_cats.csv')
    )['train']
    
    # 2. 数据清洗：过滤空评论和非二分类标签
    dataset = dataset.filter(
        lambda x: x['review'] is not None 
        and x['review'].strip() != '' 
        and x['label'] in [0, 1]
    )
    
    # 3. 将label列转换为ClassLabel类型
    dataset = dataset.cast_column(
        "label", 
        ClassLabel(names=["negative", "positive"])
    )
    
    # 4. 分层划分训练集和测试集(保持正负样本比例一致)
    dataset_dict = dataset.train_test_split(
        test_size=0.2, 
        seed=42, 
        stratify_by_column='label'
    )
    print(f"数据划分完成：训练集{len(dataset_dict['train'])}条，测试集{len(dataset_dict['test'])}条")
    
    # 5. 加载中文BERT分词器
    tokenizer = AutoTokenizer.from_pretrained(
        config.PRE_TRAINED_DIR / 'bert-base-chinese',
        local_files_only=True  # 使用本地模型，不联网下载
    )
    
    # 6. 定义分词函数
    def tokenize_function(examples):
        return tokenizer(
            examples['review'],
            max_length=config.SEQ_LEN,
            truncation=True,
            padding='max_length'
        )
    
    # 7. 对训练和测试集批量分词
    dataset_dict = dataset_dict.map(
        tokenize_function, 
        batched=True,
        remove_columns=['review', 'cat']  # 删除不需要的列
    )
    
    print("数据分词完成")
    
    # 8. 保存处理后的数据
    dataset_dict['train'].save_to_disk(str(config.PROCESSED_DATA_DIR / 'train'))
    dataset_dict['test'].save_to_disk(str(config.PROCESSED_DATA_DIR / 'test'))
    
    print("处理后的数据已保存")

if __name__ == '__main__':
    process_data()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65

运行数据预处理脚本：

python src/process.py

# 3. 数据集和数据加载器(dataset.py)

加载处理后的数据，转换为PyTorch格式，并创建DataLoader：

# dataset.py
from datasets import load_from_disk
from torch.utils.data import DataLoader
import config

def get_dataset(train=True):
    """加载数据集"""
    path = config.PROCESSED_DATA_DIR / ('train' if train else 'test')
    dataset = load_from_disk(str(path))
    
    # 设置为PyTorch格式，自动将列转换为tensor
    dataset.set_format(
        type='torch', 
        columns=['input_ids', 'attention_mask', 'label']
    )
    return dataset

def get_dataloader(train=True):
    """创建数据加载器"""
    dataset = get_dataset(train)
    return DataLoader(
        dataset, 
        batch_size=config.BATCH_SIZE, 
        shuffle=train,  # 训练集打乱，测试集不打乱
        num_workers=4   # 多进程加载数据
    )

# 数据加载测试
if __name__ == '__main__':
    dataloader = get_dataloader(train=True)
    for batch in dataloader:
        print("Batch shape:")
        print(f"input_ids: {batch['input_ids'].shape}")
        print(f"attention_mask: {batch['attention_mask'].shape}")
        print(f"label: {batch['label'].shape}")
        break

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 4. 模型定义(model.py)

基于预训练的BERT模型构建情感分析模型：

# model.py
import torch.nn as nn
from transformers import AutoModel
import config

class ReviewSentimentModel(nn.Module):
    def __init__(self, freeze_bert=False):
        super().__init__()
        # 1. 加载本地预训练的BERT模型
        self.bert = AutoModel.from_pretrained(
            config.PRE_TRAINED_DIR / 'bert-base-chinese',
            local_files_only=True
        )
        
        # 2. 分类器：接收[CLS] token的输出 → 输出二分类logits
        self.classifier = nn.Linear(self.bert.config.hidden_size, 1)
        
        # 3. 是否冻结BERT参数
        if freeze_bert:
            for param in self.bert.parameters():
                param.requires_grad = False
    
    def forward(self, input_ids, attention_mask):
        """
        前向传播
        Args:
            input_ids: (batch_size, seq_len) 输入token id
            attention_mask: (batch_size, seq_len) 注意力掩码
        Returns:
            logits: (batch_size,) 二分类logits
        """
        # BERT前向传播，last_hidden_state是最后一层所有token的输出
        outputs = self.bert(input_ids=input_ids, attention_mask=attention_mask)
        
        # 提取[CLS] token的输出(第一个位置)，形状: (batch_size, hidden_size)
        cls_output = outputs.last_hidden_state[:, 0, :]
        
        # 通过分类器得到logits，形状: (batch_size, 1)
        logits = self.classifier(cls_output)
        
        # 去掉最后一维，返回形状: (batch_size,)
        return logits.squeeze(-1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

# 5. 模型训练(train.py)

训练模型并保存最优参数，使用TensorBoard记录训练过程：

# train.py
import time
import torch
from torch.utils.tensorboard import SummaryWriter
from tqdm import tqdm
import config
from dataset import get_dataloader
from model import ReviewSentimentModel

def train_one_epoch(model, dataloader, optimizer, loss_fn, device):
    """训练一个epoch"""
    model.train()
    total_loss = 0.0
    
    for batch in tqdm(dataloader, desc="训练中"):
        # 将数据移动到GPU
        input_ids = batch['input_ids'].to(device)
        attention_mask = batch['attention_mask'].to(device)
        labels = batch['label'].float().to(device)
        
        # 清除梯度
        optimizer.zero_grad()
        
        # 前向传播
        outputs = model(input_ids=input_ids, attention_mask=attention_mask)
        
        # 计算损失
        loss = loss_fn(outputs, labels)
        
        # 反向传播并更新参数
        loss.backward()
        optimizer.step()
        
        # 累计损失
        total_loss += loss.item() * input_ids.size(0)
    
    # 计算平均损失
    avg_loss = total_loss / len(dataloader.dataset)
    return avg_loss

def train():
    """模型训练主函数"""
    # 1. 设置运行设备
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    print(f"使用设备: {device}")
    
    # 2. 加载数据加载器
    train_dataloader = get_dataloader(train=True)
    print("数据集加载完成")
    
    # 3. 初始化模型
    model = ReviewSentimentModel(freeze_bert=False).to(device)
    print("模型初始化完成")
    
    # 4. 定义优化器和损失函数
    optimizer = torch.optim.AdamW(model.parameters(), lr=config.LEARNING_RATE)
    loss_fn = torch.nn.BCEWithLogitsLoss()  # 带sigmoid的二分类交叉熵损失
    
    # 5. 初始化TensorBoard写入器
    log_dir = config.LOGS_DIR / time.strftime("%Y%m%d-%H%M%S")
    writer = SummaryWriter(log_dir=str(log_dir))
    
    # 6. 训练循环
    best_loss = float("inf")
    for epoch in range(1, config.EPOCHS + 1):
        print(f"\n========== Epoch {epoch}/{config.EPOCHS} ==========")
        
        # 训练一个epoch
        train_loss = train_one_epoch(model, train_dataloader, optimizer, loss_fn, device)
        
        print(f"训练集平均损失: {train_loss:.4f}")
        
        # 写入TensorBoard
        writer.add_scalar("Loss/train", train_loss, epoch)
        
        # 保存最优模型
        if train_loss < best_loss:
            best_loss = train_loss
            torch.save(model.state_dict(), config.MODELS_DIR / 'best_model.pt')
            print(f"保存最优模型，损失: {best_loss:.4f}")
    
    writer.close()
    print(f"\n训练完成！最优模型保存在: {config.MODELS_DIR / 'best_model.pt'}")

if __name__ == '__main__':
    train()

运行训练脚本：

python src/train.py

# 6. 模型预测(predict.py)

加载训练好的模型，提供交互式预测功能：

# predict.py
import torch
from transformers import AutoTokenizer
import config
from model import ReviewSentimentModel

def predict_batch(input_ids, attention_mask, model):
    """批量预测"""
    model.eval()
    with torch.no_grad():
        logits = model(input_ids=input_ids, attention_mask=attention_mask)
        probs = torch.sigmoid(logits)  # 将logits转换为0-1的概率
        return probs.tolist()

def predict_single_text(text, model, tokenizer, device):
    """预测单条文本"""
    # 文本编码
    encoded = tokenizer(
        text,
        max_length=config.SEQ_LEN,
        padding='max_length',
        truncation=True,
        return_tensors='pt'
    )
    
    # 将数据移动到GPU
    input_ids = encoded['input_ids'].to(device)
    attention_mask = encoded['attention_mask'].to(device)
    
    # 预测
    prob = predict_batch(input_ids, attention_mask, model)[0]
    return prob

def run_interactive_predict():
    """交互式预测主程序"""
    # 1. 设置设备
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    
    # 2. 加载分词器和模型
    tokenizer = AutoTokenizer.from_pretrained(
        config.PRE_TRAINED_DIR / 'bert-base-chinese',
        local_files_only=True
    )
    
    model = ReviewSentimentModel().to(device)
    model.load_state_dict(
        torch.load(config.MODELS_DIR / 'best_model.pt', map_location=device)
    )
    
    print("=" * 50)
    print("AI评论情感分析系统 v3.0")
    print("输入评论即可进行情感分析，输入q/quit退出")
    print("=" * 50)
    
    # 3. 交互循环
    while True:
        user_input = input("\n请输入评论: ").strip()
        
        if user_input.lower() in {'q', 'quit', 'exit'}:
            print("感谢使用，再见！")
            break
        
        if not user_input:
            print("输入不能为空，请重新输入")
            continue
        
        # 预测
        prob = predict_single_text(user_input, model, tokenizer, device)
        
        # 输出结果
        if prob > 0.5:
            print(f"😊 正面评价 (置信度: {prob:.4f})")
        else:
            print(f"😠 负面评价 (置信度: {1 - prob:.4f})")

if __name__ == '__main__':
    run_interactive_predict()

运行交互式预测：

python src/predict.py

示例输出：

==================================================
AI评论情感分析系统 v3.0
输入评论即可进行情感分析，输入q/quit退出
==================================================

请输入评论: 这款手机拍照效果特别好，续航也很给力
😊 正面评价 (置信度: 0.9987)

请输入评论: 物流太慢了，包装也破了，东西质量很差
😠 负面评价 (置信度: 0.9992)

请输入评论: q
感谢使用，再见！

1
2
3
4
5
6
7
8
9
10
11
12
13

# 7. 模型评估(evaluate.py)

在测试集上评估模型的准确率：

# evaluate.py
import torch
from tqdm import tqdm
import config
from dataset import get_dataloader
from model import ReviewSentimentModel
from predict import predict_batch

def evaluate_model():
    """在测试集上评估模型准确率"""
    # 1. 设置设备
    device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    
    # 2. 加载模型
    model = ReviewSentimentModel().to(device)
    model.load_state_dict(
        torch.load(config.MODELS_DIR / 'best_model.pt', map_location=device)
    )
    
    # 3. 加载测试数据加载器
    test_dataloader = get_dataloader(train=False)
    
    # 4. 评估
    model.eval()
    correct = 0
    total = 0
    
    with torch.no_grad():
        for batch in tqdm(test_dataloader, desc="评估中"):
            input_ids = batch['input_ids'].to(device)
            attention_mask = batch['attention_mask'].to(device)
            labels = batch['label'].to(device)
            
            # 预测
            probs = predict_batch(input_ids, attention_mask, model)
            
            # 将概率转换为预测标签
            preds = [1 if p >= 0.5 else 0 for p in probs]
            
            # 统计正确数
            for pred, label in zip(preds, labels):
                if pred == label.item():
                    correct += 1
                total += 1
    
    # 计算准确率
    accuracy = correct / total if total > 0 else 0
    
    print("\n" + "=" * 30)
    print(f"测试集准确率: {accuracy:.4f}")
    print("=" * 30)
    
    return accuracy

if __name__ == '__main__':
    evaluate_model()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

运行评估脚本：

python src/evaluate.py

通常情况下，这个模型在测试集上的准确率可以达到92%以上，远高于传统机器学习方法。

# 8.6 本章小结

本章我们系统学习了NLP领域三大经典预训练模型架构：

GPT系列：基于解码器架构，擅长文本生成任务，采用自回归语言建模预训练目标。
BERT系列：基于编码器架构，擅长自然语言理解任务，采用掩码语言建模预训练目标。
T5系列：基于编码器-解码器架构，采用统一的文本到文本框架，通用性最强。

我们还通过一个完整的情感分析实战项目，掌握了使用HuggingFace工具链快速落地预训练模型的方法。预训练模型的出现极大地降低了NLP任务的开发门槛，让普通开发者也能轻松使用最先进的NLP技术。

在接下来的后续课程中，我们将继续探索预训练模型的最新发展，包括大语言模型、多模态模型以及它们在实际业务中的应用。

← 大模型技术之深度学习大模型技术之Neo4j →

# 第 1 章 NLP导论：让计算机听懂人类的语言

# 1.1 什么是自然语言处理

# 1.2 NLP的五大核心任务

# 1.2.1 文本分类：给整篇文本贴标签

# 1.2.2 序列标注：给每个词打标签

# 1.2.3 文本生成：让计算机"写"出文字

# 1.2.4 信息抽取：从非结构化文本中提取结构化信息

# 1.2.5 文本转换：将一种文本变成另一种文本

# 1.3 NLP技术的演进之路

# 1.3.1 规则系统时代(1950s-1980s)："教"计算机说话

# 1.3.2 统计方法时代(1990s)：从数据中"找"规律

# 1.3.3 传统机器学习时代(2000s-2010s初)：特征工程的黄金时代

# 1.3.4 深度学习时代(2015s-至今)：端到端的革命

# 本章小结

# 第2章 环境搭建：从零搭建生产级PyTorch NLP开发环境

# 2.1 系统基础准备

# 2.2 生产级JupyterLab部署

# 2.2.1 创建独立虚拟环境

# 2.2.2 安装并配置JupyterLab

# 2.2.3 修改核心配置文件

# 2.2.4 配置Systemd系统服务

# 2.2.5 启动服务并测试访问

# 2.2.6 安装实用插件

# 2.3 PyTorch与NLP核心依赖安装

# 2.3.1 配置Pip国内镜像（永久生效）

# 2.3.2 基础数据科学与工具包

# 2.3.3 NLP专用工具包

# 2.3.4 PyTorch深度学习框架

# 2.4 环境验证与常见问题排查

# 2.4.1 核心环境验证

# 2.4.2 常见问题排查

# 本章小结

# 第3章 文本表示：把文字变成计算机能懂的数字

# 3.1 文本表示的核心逻辑

# 3.2 分词：NLP任务的起点

# 3.2.1 英文分词的三种粒度

# 3.2.1.1 词级分词

# 3.2.1.2 字符级分词

# 3.2.1.3 子词级分词(主流方法)

# 3.2.2 中文分词的特点与方法

# 3.2.3 常用分词工具实战

# 3.2.3.1 jieba：传统中文分词首选

# 3.2.3.2 Hugging Face Tokenizers：大模型时代的标准

# 3.3 词表示：从稀疏向量到语义向量

# 3.3.1 One-hot编码：最原始的表示方法

# 3.3.2 Word2Vec：第一个真正的语义化词向量

# 3.3.2.1 核心思想：分布假设

# 3.3.2.2 两种模型结构

# 3.3.2.3 Word2Vec完整原理

# 数据集准备

# Skip-Gram模型详解

# 训练数据集

# 模型结构

# 损失计算与前向传播

# CBOW模型详解

# 训练数据集

# 模型结构

# 损失计算与前向传播

# 3.3.2.4 获取Word2Vec词向量

# 3.3.2.5 应用Word2Vec词向量

# 3.3.3 上下文相关词表示：解决一词多义问题

# 本章小结

# 第四章 传统序列模型：从RNN到GRU的演进之路

# 4.1 RNN：序列建模的开山之作

# 4.1.1 概述：为什么需要循环神经网络？

# 4.1.2 基础结构：循环连接的奥秘

# 4.1.3 多层结构：捕捉更复杂的语义特征

# 4.1.4 双向结构：同时利用上下文信息

# 4.1.5 多层+双向结构：强强联合

# 4.1.6 API使用：PyTorch中的RNN实现

# 4.1.6.1 参数说明

# 4.1.6.2 输入输出

# 4.1.7 案例实操：智能输入法词语联想

# 4.1.7.1 需求说明

# 4.1.7.2 需求分析

# 4.1.7.3 需求实现

# 4.1.8 存在问题：RNN的致命缺陷

# 问题根源：梯度消失与梯度爆炸

# 4.2 LSTM：解决长期依赖问题的里程碑

# 4.2.1 概述：LSTM的核心思想

# 第2章环境搭建：从零搭建生产级PyTorch NLP开发环境

# 第3章文本表示：把文字变成计算机能懂的数字

# 第四章传统序列模型：从RNN到GRU的演进之路