用Keras从零实现Transformer中英机器翻译的完整实践指南

简介：基于Python与Keras-Transformer的中英文双向机器翻译系统，包含完整可执行程序、源代码与技术文档，可直接运行部署，适用毕业设计、课程实践和项目原型开发等场景。资源包共二十一个文件，主体为Py源码、数据获取与训练翻译Notebook、序列化pkl中间数据、h5预训练权重以及md与txt说明文档，压缩包大小约十三点四七MB，目录结构清晰，便于按模块复用与二次开发。目前已有七十二人学习，代码经过多轮完整性验证与功能测试，具备可靠技术基准，可作为开发起点进行功能扩展与性能优化。该实现侧重Keras-Transformer标准接口的应用层开发，同时也包含中文简繁转换、语料预处理等辅助模块。若与基于LSTM的同类翻译项目配合学习，在完全一致的训练数据集和预处理流程下，可直观对比Transformer与LSTM在机器翻译任务中的表现差异，帮助理解不同神经网络架构的适用特性。

1. 中英文机器翻译为什么绕不开Transformer：一个能落地的Keras方案

做中英文机器翻译，很多人第一反应是LSTM/Seq2Seq，但如果你真的拿它去翻一句带长从句的新闻稿，就会发现漏译、错序、越翻越短是常态。Transformer用自注意力把任意两个位置直接联系起来，长距离依赖不再是靠记忆硬扛，而是靠注意力矩阵显式建模；配合Keras这种能快速改结构、看中间张量的框架，你可以在一台普通GPU上从头训练一个能用的中英翻译系统，而不是只调HuggingFace的现成接口。这篇笔记面向的是想自己跑通数据、词表、训练、推理全流程的人：你需要会基本Python和Keras，但不需要懂论文里的所有数学。我会把每个组件的设计理由、最小实现、参数取值和踩坑记录都摊开讲，让这个「附源码及文档」的项目真正能被你复现，而不是变成黑匣子。

2. 从数据集到词表：训练翻译模型前要把文本处理成什么样子

2.1 平行语料的选择与预处理：别让清洗毁掉你的BLEU

常见做法是用公开的中英平行语料，比如WMT、UN平行语料或较干净的AI Challenger翻译数据集。选语料时先看两个指标：句子对齐是否可靠，领域是否单一。混入新闻、口语、法律文本会严重拉低单领域效果，所以我的经验是先按领域切分成子集，训练时只用其中一个，否则你会在验证集上看到BLEU永远上不去。

拿到原始语料后，第一件事不是分词，而是清洗。以下这段是我的标准预处理脚本：

import re import unicodedata def clean_pair(en, zh): # 英文侧：统一半角、去掉控制字符、折叠多余空格 en = unicodedata.normalize("NFKC", en) en = re.sub(r"[\x00-\x1f\x7f]", "", en) en = re.sub(r"\s+", " ", en).strip() # 中文侧：转全角标点、去控制符、合并连续空格（中文里空格通常无意义） zh = unicodedata.normalize("NFKC", zh) zh = re.sub(r"[\x00-\x1f\x7f]", "", zh) zh = re.sub(r"\s+", " ", zh).strip() # 过滤明显坏样本：太长的、比例失衡的、空行 if not en or not zh: return None if len(en.split()) > 80 or len(zh) > 160: return None en_len = len(en.split()) zh_len = len(zh) if zh_len / max(en_len, 1) > 8 or zh_len / max(en_len, 1) < 1.5: return None return en, zh

这段代码做的事是把英文NFKC归一化（把全角英文字符转成半角），把中文控制符清掉，再用长度比例过滤掉明显没对齐的句子对。为什么中文侧不能简单按空格分词？因为中文分词需要额外工具；Transformer的输入本质上是一串整数ID，我们完全可以按字符切分中文，这样不需要引入jieba，也能避免分词错误在翻译阶段被放大。所以上面的清洗里不处理中文分词，只处理非法字符和比例过滤。

比例过滤阈值我按经验取1.5到8之间。中文单字信息密度高，一个英文词对应1.5到2个中文字符正常；如果一条中文短而英文很长，多半是没对齐；反过来中文很长英文很短，可能是把多句话拼在一起了。你可以根据自己语料统计分布再调这个区间，但别直接删掉所有偏离均值的样本，那会让模型丢掉正常的长句能力。

2.2 构建中英文词表：子词切分与最小频次阈值

词表是机器翻译的命门。对英文，我建议用子词切分（BPE或WordPiece），因为它能把「look」和「looked」拆成共享子词，大幅降低未登录词。中文则可以直接按字符建词表，常用中文字符大概五六千，加上特殊符号，词表规模很小。用Keras的Tokenizer可以快速做字符级词表：

from tensorflow.keras.preprocessing.text import Tokenizer def build_vocab(texts, lang, vocab_size=16000): tokenizer = Tokenizer(num_words=vocab_size, filters='', lower=(lang == 'en'), oov_token='[UNK]') tokenizer.fit_on_texts(texts) return tokenizer

这里有两处关键设置。第一，filters='' 表示不自动过滤标点，因为翻译需要保留逗号、句号、问号。第二，中文侧的lower=False，英文侧True，避免中文里不存在的大小写问题。vocab_size对于字符级中文可以设到8000左右，对英文BPE可以设到16000或32000。如果语料只有几十万句，32000词表会有大量低频词，模型学不好，反而掉BLEU；我一般先用16000跑基线。

词表构建完要手动插入控制符。序列两端需要[START]和[END]，解码时才能知道什么时候开始、什么时候停止。用Tokenizer时，oov_token='[UNK]'已经占了索引1，索引0是保留的，所以我会在索引2和3分别插入[START]和[END]，并把现有索引全部加2。这一步很坑，很多人漏掉对齐导致训练标签错位。

2.3 数据管道：用tf.data把样本喂给Keras的注意点

清洗和词表构建完成后，要用 tf.data 搭建高效的训练管道。不要用Python生成器逐条feed，那样GPU利用率会惨不忍睹。下面的代码把变长序列按batch内最大长度做padding，并生成decoder输入和标签的错位关系：

import tensorflow as tf def encode_pair(en, zh, en_tok, zh_tok): en_ids = en_tok.texts_to_sequences([en])[0] zh_ids = zh_tok.texts_to_sequences([zh])[0] # decoder输入：开头加 [START]，结尾不加；标签：开头不加，结尾加 [END] enc_in = [en_tok.word_index['[START]']] + en_ids dec_in = [zh_tok.word_index['[START]']] + zh_ids dec_tgt = zh_ids + [zh_tok.word_index['[END]']] return enc_in, dec_in, dec_tgt def make_dataset(pairs, en_tok, zh_tok, batch_size=64): ds = tf.data.Dataset.from_generator( lambda: (encode_pair(en, zh, en_tok, zh_tok) for en, zh in pairs), output_types=(tf.int32, tf.int32, tf.int32)) ds = ds.padded_batch( batch_size, padded_shapes=([None], [None], [None]), padding_values=(0, 0, 0)) return ds.prefetch(tf.data.AUTOTUNE)

注意padded_batch的padding值设成0，对应词表里没有词的那个保留位。在训练时，注意力掩码会把padding位置遮掉，所以0这个值具体是什么不重要，只要它不出现在真实词表里就行。from_generator在单机单卡够用；如果数据量大，建议先preprocess成TFRecord，否则每次重启训练都要重新执行一遍Python生成器，很浪费时间。

还有一个常常被忽略的点：padded_batch默认按batch内最长序列padding，这会导致同一个batch里长短差距大时浪费算力。可以先用bucket_by_sequence_length分桶再padding，Keras内置支持不太直接，但我建议先用简单方案跑通，后期再优化，不要一开始就上复杂的分桶策略。

3. 用Keras搭建Transformer翻译模型：核心组件与参数对照

3.1 为什么选Keras而不是直接写底层：从复现到维护的权衡

很多人一听说Transformer就想去手写矩阵运算，其实直接用Keras的MultiHeadAttention层能省掉大量验证时间。Keras的Layer封装了注意力权重、mask处理、加性偏置，底层用的是TensorFlow优化过的算子，训练速度比自己拼einsum快很多，而且不容易在维度顺序上搞错。这个项目定位是「系统实现」而非「论文复现」，所以我建议用Keras的API组合组件，只在必要处自定义Layer。

Keras对Transformer这类结构有个明显优势：动态图模式下可以打印每一层的输出张量，方便排查维度问题。比如编码器输出(batch, seq_len, d_model)，解码器交叉注意力要用的key/value都来自它；你一旦把维度搞错，报错信息会直接指出。另外，Keras的Model类可以同时管理encoder和decoder两个子模型，训练时用一个fit调用结束，推理时再单独调用子模型。

3.2 Encoder与Decoder的Keras实现要点

下面是一个精简版的Transformer编码器层实现，包含掩码多头注意力和前馈网络。这是整个系统的核心骨架：

from tensorflow.keras import layers, Model class TransformerEncoderLayer(layers.Layer): def __init__(self, d_model, num_heads, dff, rate=0.1): super().__init__() self.mha = layers.MultiHeadAttention(num_heads=num_heads, key_dim=d_model // num_heads) self.ffn = tf.keras.Sequential([ layers.Dense(dff, activation='relu'), layers.Dense(d_model) ]) self.layernorm1 = layers.LayerNormalization(epsilon=1e-6) self.layernorm2 = layers.LayerNormalization(epsilon=1e-6) self.dropout1 = layers.Dropout(rate) self.dropout2 = layers.Dropout(rate) def call(self, x, mask, training=False): attn_output = self.mha(x, x, x, attention_mask=mask, training=training) attn_output = self.dropout1(attn_output, training=training) out1 = self.layernorm1(x + attn_output) # 残差先加后norm ffn_output = self.ffn(out1) ffn_output = self.dropout2(ffn_output, training=training) return self.layernorm2(out1 + ffn_output)

这里用的是Pre-LN还是Post-LN？我用的是Post-LN，即先残差相加再做LayerNormalization。实际训练时，Pre-LN（先norm再相加）更稳定，尤其在batch size较大时。如果你想减少训练初期loss震荡，可以在call里改成先norm再计算注意力。不过Post-LN收敛后的BLEU通常会略高一点，这是取舍问题，我建议基线用Post-LN，不好收敛再切Pre-LN。

Decoder层比Encoder多一个交叉注意力子层。它的第一个多头注意力要用masked self-attention，挡住未来位置；第二个多头注意力的query来自解码器，key/value来自编码器输出。Keras的MultiHeadAttention支持attention_mask参数，但交叉注意力的mask需要你自己拼形状，后面在第4章专门讲。

3.3 多头注意力与位置编码：参数怎么设才不会玄学翻车

参数设置直接决定模型能不能训练起来。我常用的基线配置如下表：

d_model必须能被num_heads整除。这个约束是因为每个头的key维度是d_model // num_heads，如果不整除，Keras会直接报错。d_model用256而不是更大的512，是为了在单卡上更快迭代；等基线跑通后，再把d_model调到512、层数加到6，BLEU通常能再涨1到2个点。

位置编码我用的是经典正弦函数版本，而不是可学习的位置嵌入，因为正弦编码对序列长度没有硬上限，推理时遇到比训练更长的句子不会因嵌入矩阵越界报错。下面是实现：

def positional_encoding(max_len, d_model): pos = tf.range(max_len)[:, tf.newaxis] i = tf.range(d_model)[tf.newaxis, :] angle = pos / tf.pow(10000.0, (2 * (i // 2)) / tf.cast(d_model, tf.float32)) angles = tf.where(i % 2 == 0, tf.sin(angle), tf.cos(angle)) return angles[tf.newaxis, ...]

这里用了i // 2来构造不同频率的正弦/余弦对。注意tf.pow(10000.0, ...)里的除法和d_model需要强转成float32，否则整数除法会截断，生成的位置编码会乱掉。你可以在模型外面验证一下：positional_encoding(100, 256)输出的形状应该是(1, 100, 256)，且第0行是全0（因为sin(0)=0）。位置编码通常是加到词嵌入上，不是拼接；拼接会让维度膨胀且破坏注意力关系。

4. 训练与推理：学习率调度、掩码和波束搜索

4.1 自定义学习率调度与损失函数：训练不炸的底线

Transformer训练对学习率极敏感。热门做法是用Transformer论文里的Warmup调度：先线性上升到峰值，再按step的倒数平方根衰减。用Keras实现很简单：

class WarmupScheduler(tf.keras.optimizers.schedules.LearningRateSchedule): def __init__(self, d_model, warmup_steps=4000): super().__init__() self.d_model = tf.cast(d_model, tf.float32) self.warmup_steps = warmup_steps def __call__(self, step): step = tf.cast(step, tf.float32) arg1 = tf.math.rsqrt(step) arg2 = step * (self.warmup_steps ** -1.5) return tf.math.rsqrt(self.d_model) * tf.minimum(arg1, arg2)

这里rsqrt(step)在step=0时会变成无穷大，但实际训练时step从1开始，所以不用担心。warmup_steps默认4000，对这个小规模系统来说偏大，我一般改成2000，否则前期学习率太小，loss下降很慢。如果你发现训练初期loss震荡剧烈，把warmup_steps加大；如果loss下降太慢，调小。这是最好调的旋钮。

损失函数使用带padding屏蔽的稀疏交叉熵。不能直接对所有token求平均，因为padding位置的损失会把模型拉偏。在Keras里可以用以下方式：

def masked_loss(y_true, y_pred): loss = tf.keras.losses.sparse_categorical_crossentropy(y_true, y_pred) mask = tf.cast(tf.not_equal(y_true, 0), tf.float32) loss = loss * mask return tf.reduce_sum(loss) / tf.reduce_sum(mask)

注意mask的作用是把标签为0的位置（padding）的loss移除，但这里有个隐患：如果整个batch里某个样本的目标序列恰好全部是0（极少见），分母会变成0。我习惯给分母加一个tf.reduce_sum(mask) + 1e-9避免除零。

4.2 推理阶段的掩码处理：为什么译文会越翻越乱

训练时Keras的MultiHeadAttention会帮你计算mask，但推理时必须手动构造。推理是自回归的：每一步用一个start token开始，把当前已生成的词立即拼回decoder输入，再整体过一遍模型。此时必须遮挡未来位置，否则模型会偷看答案。

看这段推理循环：

def decode_step(dec_input, enc_output, enc_padding_mask, combined_mask, model): # dec_input形状 (1, current_len) decoder_out, _ = model.layers[1](dec_input, enc_output, False, combined_mask, enc_padding_mask) # 取最后一个位置的logits next_probs = model.layers[-1](decoder_out[:, -1:, :]) return next_probs

combined_mask要同时挡住padding和未来位置。构造方法是用一个下三角矩阵与padding mask做逻辑与。Keras的MultiHeadAttention在call里传attention_mask时，它会自动扩展维度，所以你传给它的mask形状必须是(batch, from_seq, to_seq)。最常见的问题是把mask形状传成(batch, seq)，务必检查。

这里特别容易翻车：训练时解码器的输入和标签是错位一步的，所以模型在第t步看到的是位置0到t-1的token，预测的是位置t的目标token。推理时你把已有输出作为输入，模型天然也是预测下一个token，只要mask逻辑正确，不需要额外偏移。如果发现整句输出全是重复词，十有八九是未来mask没生效，模型把当前位置以后的内容也当成上下文了。

4.3 波束搜索实现：beam size怎么定

贪心解码一次只走一条路，很容易在某个错词上走死。波束搜索保留多条候选，每步扩展所有候选的top-k概率，最后选整体得分最高的。实现要点如下：

def beam_search(enc_output, start_token, end_token, beam_size=4, max_len=60): beams = [{"tokens": [start_token], "score": 0.0}] for _ in range(max_len): new_beams = [] for beam in beams: if beam["tokens"][-1] == end_token: new_beams.append(beam) continue dec_input = tf.expand_dims(beam["tokens"], 0) # 计算下一个token概率 next_probs = predict_next(dec_input, enc_output) log_probs = tf.math.log(tf.maximum(next_probs, 1e-10)) top_k = tf.math.top_k(log_probs, k=beam_size) for i in range(beam_size): new_beams.append({ "tokens": beam["tokens"] + [int(top_k.indices[0, i])], "score": beam["score"] + float(top_k.values[0, i]) }) # 按score排序保留beam_size个 beams = sorted(new_beams, key=lambda b: b["score"], reverse=True)[:beam_size] return beams[0]["tokens"]

beam size不是越大越好。我实测在4到8之间增益明显，从8到16几乎没提升，反而生成速度变慢，而且偶尔会翻出更长但更绕的句子。所以中英翻译项目我默认beam=4。还有长度惩罚问题：beam search天然倾向短句，因为概率是累乘的，句子越长log概率越负。如果不加处理，模型会输出过短的句子。常见做法是在score里加length_penalty，让长句的累积分数除以一个随长度增长的系数，但系数设不好又会让结果拖沓。我的建议是先用无惩罚跑通，看长句质量再决定是否加。

5. 避坑/常见问题/排查：Keras-Transformer翻译系统最常见的5个坑

5.1 现象：训练loss下降但BLEU不动

模型在训练集上loss一直降，验证集loss也降，但翻译结果的BLEU就是原地踏步。原因是验证时用的是贪心解码，而训练时用的是Teacher Forcing（每一步都给真实前文）。模型学会了依赖真实历史，但推理时一旦第一步出错，错误会顺着上下文传播，导致整句崩盘。解决办法是先在训练后期加入「计划采样」——以一定概率把真实token替换成模型自己的预测token，或者干脆先在训练时用较小的mask。更直接的做法是：检查验证集是否和训练集领域一致。如果领域混杂，BLEU会很低，不代表模型坏了。

5.2 现象：显存不足/OOM

训练时batch size设64，d_model设为512，一上来就爆显存。原因是Transformer的注意力复杂度是O(n²)，加上padding浪费，长句会占大量显存。解决路径有三步：先调小batch size到16或8，观察loss是否还能下降；再把max_len限制在80以内；最后使用混合精度（mixed_float16）训练，显存能省接近一半。Keras设置混合精度只需要在fit前加一句：

tf.keras.mixed_precision.set_global_policy('mixed_float16')

但注意混合精度下loss计算可能不稳定，最好在loss函数里用tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)，让内部处理精度。

5.3 现象：预测时输出全是[EOS]

模型第一个词就输出结束标记。这通常不是因为模型笨，而是因为词表构建时把[EOS]放进了padding mask之外，导致模型在预测位置0时看到的上下文几乎全是padding，而[EOS]在训练集里作为所有句子最后一个token，频率很高，模型学会了「胡乱结束」。解决方法是检查训练时decoder输入的mask是否把[EOS]位置也遮住了——不应该遮，但padding要遮；同时确认标签里[EOS]的位置没有被当成padding去掉。另外，如果语料里空句子太多，[EOS]概率会被拉高，建议清洗时过滤掉长度小于4的词对。

5.4 现象：位置编码相加后效果反而变差

我在一个简化版模型里把位置编码直接加到嵌入层，结果BLEU比去掉位置编码还低。排查后发现是嵌入向量初始化范围太大，位置编码的值域在[-1,1]，两者相加后，位置信息被嵌入向量的噪声淹没了。解决办法是把词嵌入用tf.keras.layers.Embedding默认的方差缩放初始化，并在相加前将嵌入乘以sqrt(d_model)。代码里embedding * tf.math.sqrt(tf.cast(d_model, tf.float32))再add位置编码，这一步做不做，收敛速度差别明显。

5.5 现象：Keras版本不同导致模型不兼容

在TensorFlow 2.10上保存的.h5模型，换到2.15后load_model报Unknown layer或MultHeadAttention属性找不到。原因是在不同版本里自定义层序列化名称不一致。解决办法是不要用model.save("model.h5")只存权重后再重建结构，或者直接保存为.keras新格式：

model.save("mt_model.keras") # 加载时需重新注册自定义层或用keras.saving.load_model

如果你在项目文档里依赖了旧模型的.h5文件，建议同时导出config.json和weights.h5，加载时先用from_config重建结构，再加载权重，这样能跨小版本。我踩过这个坑后，所有版本迭代前都会在文档里注明TensorFlow版本号。

6. 进阶验证：用BLEU和案例句给翻译系统做一次体检

6.1 计算BLEU的脚本与阈值判断

光看loss不能说明翻译质量，建议用sacrebleu计算BLEU。它会把中英文按官方切分方式做tokenize，避免你的分词方式影响分数。跑测试集：

sacrebleu test.zh --score-only -m bleu < pred.zh

这里的pred.zh是模型对所有英文测试句的输出。一个从零训练、4层d_model=256的小模型在WMT测试集上BLEU能达到20左右；如果只在一个单一领域的小语料上训练，25以上不算稀奇。低于15就说明语料或训练有问题。你也可以用BLEURT做参考，但BLEU足以判断方向。

6.2 三个必测案例句：数字、术语、长句

我每次训练完必测这三类句子：包含数字的句子，看数字是否保留；包含专业名词的句子，看词表是否覆盖；还有超过40个词的长句，看是否漏译。例如：

如果数字翻错，多半是词表把数字拆碎了；如果专有名词变成[UNK]，说明词表该加这个术语；如果长句漏译，则检查注意力部分是否没覆盖到所有编码位置。这些检查比BLEU更能定位问题来源。

6.3 从BLEU到实用：保存模型、导出词表与后续微调

跑完测试，把词表和模型捆绑保存，否则换个环境就废了。我会把en_tok和zh_tok用tokenizer_to_json导出，模型单独存.keras。后续要微调领域数据，只需在现有权重上继续fit，但学习率要调小到原来的十分之一，warmup_steps也调小，否则会把已有参数冲乱。另一个实用技巧是写一个简单的翻译函数，把预处理、词表映射、beam search全部封装，让业务方直接调用translate("string")，而不是暴露内部数据结构。

最后说一个我自己的习惯：每次调完参，我都会把训练日志、验证BLEU、测试输出样例放同一个目录，标好日期。下次有人说「模型怎么变差了」，翻日志十分钟就能定位是哪次改动引入的。这是最土但最有效的版本管理手段，希望这些落地细节能帮你在自己的中英翻译项目上少走弯路。

本文还有配套的精品资源，点击获取