简介这款基于Python与Keras Transformer模型的中英机器翻译项目是一套已调试可直接运行的完整源码包主要面向计算机、通信、人工智能、自动化等相关专业的学生、教师及从业者适用于毕业设计、期末课程设计、课程大作业及个人进阶学习。压缩包共17个文件仅7.42MB包含Python脚本、Jupyter Notebook实操演示、预训练模型权重h5、序列化中间数据pkl与说明文档md/README资源覆盖语料获取、繁简转换、数据预处理、Transformer模型训练、中英翻译演示等完整流程代码经过调试测试确保可运行。项目中搭配get_data.ipynb和traintranslate.ipynb两份笔记本可直观操作从数据处理到训练翻译的路径附带的language模块包含中文转换工具便于理解与二次开发。目前已有102人学习下载目录划分明确既适合零基础学生快速跑通项目也为能力较强读者提供了调整模型、扩展功能的良好基础。1. 中文机器翻译用 kreas-transformer 自己搭比调 API 更能看清翻译的本质一段中文要翻成英文大多数人第一反应是接现成 API 或者加载大模型但这套基于 python 开发、基于 kreas-transformer也就是 Keras Transformer的中英文机器翻译源码我建议你先别急着嫌“老”。原因很反直觉Keras 已经帮我们封装了绝大多数张量运算真正的工程难点反而集中在数据预处理、mask 构造和自回归推理这三个环节。把这套能直接跑的源码在本地跑通再回去看那些“一行调用”的翻译服务你会知道黑匣子里发生的到底是什么。适合谁想搞懂 Transformer 翻译链路的学生、需要在离线环境做中英翻译的开发者以及准备把翻译模型接入自己业务但不想盲目堆大模型参数的一线工程师。2. 为什么用 Keras 搭中英翻译 Transformer数据管道先于模型2.1 中英文对齐语料清洗、切分与 token 化的先后顺序常见做法是先拿开源的中英平行语料比如新闻领域或日常口语领域的句子对。别一上来就做模型先把语料清洗干净。我一般会做三个动作按句号、问号、感叹号拆句去掉肉眼可见的 HTML 标签过滤掉长度极端或者中英比例严重失衡的句子对。比如中文只有 5 个字英文却有 200 个词这种翻译模型学起来成本极高不如直接丢掉。清洗完之后做 tokenization。中文不能按空格分最少要按字切更合理的是用 jieba 分词英文按空格分之后还要做小写化。这里有个容易踩的坑中英共用一套 tokenizer 会互相干扰建议分别建词表。词表大小直接影响模型体积和训练速度按经验# 词表与模型基本参数先跑通再调大 VOCAB_EN 16000 # 英文词表大小 VOCAB_ZH 20000 # 中文词表大小 MAX_LEN 64 # 序列统一长度 D_MODEL 128 # transformer 隐层维度 NUM_HEAD 4 # 多头注意力头数 DFF 512 # feed-forward 中间层维度 DROPOUT 0.1 # 正则化比例 BATCH_SIZE 32这段参数是“能用”的起步配置不是“最好”的配置。在显存足够的机器上D_MODEL 提到 256、NUM_HEAD 提到 8效果会明显上升但训练时间也会涨。词表大小的选择逻辑是在训练语料上统计词频覆盖掉 95% 以上的 token 即可剩下的全部映射成unk。不要盲目把词表设得很大否则 embedding 层占内存不说低频词学不出有效向量等于白占参数。2.2 把句子对变成模型输入训练时的三个张量Transformer 翻译模型训练时一个句子对要生成三条输入序列。encoder 输入是源语言句子decoder 输入是目标语言句子右移一位后加上start训练标签是目标语言句子也就是 decoder 输入去掉start后的内容。这样设计的目的是让 decoder 在预测第 i 个词时只能看到前 i-1 个真实词这就是 teacher forcing。def encode_batch(en_lines, zh_lines): en_tokens [en_tokenizer.encode(line) [END_ID] for line in en_lines] zh_tokens [[START_ID] zh_tokenizer.encode(line) [END_ID] for line in zh_lines] # 统一长度短补 0长截断 en_padded tf.keras.preprocessing.sequence.pad_sequences( en_tokens, maxlenMAX_LEN, paddingpost, truncatingpost) zh_padded tf.keras.preprocessing.sequence.pad_sequences( zh_tokens, maxlenMAX_LEN 1, paddingpost, truncatingpost) decoder_input zh_padded[:, :-1] # 去掉最后一个 END target zh_padded[:, 1:] # 去掉最前面的 START # 返回 encoder_input, decoder_input, target 三个张量 return en_padded, decoder_input, target注意这里 pad 用 0而pad的 token id 也是 0所以后面构造 mask 时只需要判断“是否等于 0”就够了。truncatingpost表示超长部分从尾部截断保留句子前部语义对翻译任务来说比截头部合理。decoder_input 长度是 MAX_LENtarget 长度也是 MAX_LEN刚好对齐。2.3 为什么用 Keras 而不是纯 TensorFlow 手写这里有个取舍纯 TensorFlow 手写 Transformer 不是不行但需要自己管理变量初始化、梯度更新和 checkpoint代码量至少多三分之一。Keras 的层抽象恰好把 MultiHeadAttention、LayerNormalization、Dropout 这些高频组件封装成可以直接调用的层同时保留call()里自定义 mask 的能力。对翻译这个任务来说mask 是最容易出错的地方Keras 子类化 API 可以让我们在层内部显式处理 mask调试比函数式 API 直观很多。另外项目名是 kreas-transformer本质就是 Keras 的 BogoTransformer 的实现。本地验证时留意 TensorFlow 的版本tf 2.10 以上内置的keras.layers.MultiHeadAttention接口稳定低于 2.4 的版本则建议直接升级否则一些参数名对不上。3. 手写 Transformer 翻译模型位置编码、多头注意力与 mask3.1 位置编码没有它Transformer 只是一袋词袋Transformer 没有循环结构无法天然感知词的先后顺序所以要在 embedding 上叠加位置信息。常见做法是用 sin/cos 函数生成固定位置编码而不是让模型学习位置向量。原因很简单固定编码可以处理训练时长度的序列且不同频率的波形让模型更容易区分相对位置。import numpy as np import tensorflow as tf from tensorflow import keras class PositionalEncoding(keras.layers.Layer): def __init__(self, d_model, max_len): super().__init__() # 预计算 shape 为 (1, max_len, d_model) 的位置编码 pos np.arange(max_len)[:, None] dim np.arange(d_model)[None, :] even np.sin(pos / np.power(10000, 2 * (dim // 2) / d_model)) odd np.cos(pos / np.power(10000, 2 * (dim // 2) / d_model)) pe np.where(dim % 2 0, even, odd).astype(float32) self.pe tf.constant(pe[None, ...]) def call(self, x): # x shape: (batch, seq_len, d_model) return x self.pe[:, :tf.shape(x)[1], :]代码里最关键的是np.where(dim % 2 0, even, odd)这对应原始论文里奇偶维度交替使用 sin 和 cos 的公式。10000是频率底数一般不需要改。这个层在训练和推理时可以共用因为位置编码只和序列长度有关和 batch 大小无关。需要注意的是叠加位置编码后要做 Dropout通常放在__init__里初始化的keras.layers.Dropout(0.1)之后。3.2 Encoder 块多头注意力的 mask 是 padding maskEncoder 的每个 block 包含多头注意力和前馈网络两层之间都接残差连接与 LayerNorm。这里最容易被忽略的是Encoder 内部做 self-attention 时每个 token 都可以看其他所有 token因此只需要 padding mask不需要因果 mask。所谓 padding mask就是把值为 0 的位置替换成一个极大的负数让 softmax 在那些位置上的注意力权重趋近于 0。def create_padding_mask(seq): # seq 中 pad 的位置记为 1参与注意力时被屏蔽 mask tf.cast(tf.equal(seq, 0), tf.float32) return mask[:, None, None, :] # (batch, 1, 1, seq_len) class EncoderBlock(keras.layers.Layer): def __init__(self, d_model, num_heads, dff, dropout): super().__init__() self.attn keras.layers.MultiHeadAttention( num_headsnum_heads, key_dimd_model) self.ffn keras.Sequential([ keras.layers.Dense(dff, activationrelu), keras.layers.Dense(d_model), ]) self.ln1 keras.layers.LayerNormalization(epsilon1e-6) self.ln2 keras.layers.LayerNormalization(epsilon1e-6) self.drop1 keras.layers.Dropout(dropout) self.drop2 keras.layers.Dropout(dropout) def call(self, x, padding_mask): attn_out self.attn(x, x, attention_maskpadding_mask) x self.ln1(x self.drop1(attn_out)) # 残差连接在 LayerNorm 之前 ffn_out self.ffn(x) return self.ln2(x self.drop2(ffn_out))keras.layers.MultiHeadAttention的attention_mask参数要求是 float mask1 的位置保留注意力0 的位置屏蔽。上面create_padding_mask返回的矩阵中pad 位是 1这看起来冲突了对吧这是 Keras 的习惯问题它内部会把attention_mask按“0 表示屏蔽”处理。所以实际使用时要在注意力层外面手动转换。我通常改为传入1 - pad_mask也就是有效位置为 1。这个细节如果不注意训练能跑但注意力永远学不到有效信息Loss 掉到 4.0 左右就神秘持平。3.3 Decoder 块两层注意力两类 mask 叠加Decoder 比 Encoder 多一层 cross-attention它的输入来自 Encoder 输出。Decoder 的 self-attention 需要一个因果 mask确保每个位置只能看它自己和之前的位置否则就是作弊。因果 mask 的构造方式是生成一个上三角矩阵把右上角置为 1 后交给 Keras 做屏蔽。def create_causal_mask(seq_len): # 下三角为 0上三角为 1 mask 1 - tf.linalg.band_part(tf.ones((seq_len, seq_len)), -1, 0) return mask[None, None, ...] # shape (1, 1, seq_len, seq_len) class DecoderBlock(keras.layers.Layer): def __init__(self, d_model, num_heads, dff, dropout): super().__init__() self.self_attn keras.layers.MultiHeadAttention( num_headsnum_heads, key_dimd_model) self.cross_attn keras.layers.MultiHeadAttention( num_headsnum_heads, key_dimd_model) self.ffn keras.Sequential([ keras.layers.Dense(dff, activationrelu), keras.layers.Dense(d_model), ]) self.ln1 keras.layers.LayerNormalization(epsilon1e-6) self.ln2 keras.layers.LayerNormalization(epsilon1e-6) self.ln3 keras.layers.LayerNormalization(epsilon1e-6) self.drop1 keras.layers.Dropout(dropout) self.drop2 keras.layers.Dropout(dropout) self.drop3 keras.layers.Dropout(dropout) def call(self, x, enc_output, padding_mask, combined_mask, causal_mask): # Decoder 自身注意力用 causal_mask 屏蔽未来 token self_attn_out self.self_attn( x, x, attention_maskcombined_mask) x self.ln1(x self.drop1(self_attn_out)) # cross-attention 里 query 来自 decoderkey/value 来自 encoder cross_attn_out self.cross_attn( x, enc_output, enc_output, attention_maskpadding_mask) x self.ln2(x self.drop2(cross_attn_out)) ffn_out self.ffn(x) return self.ln3(x self.drop3(ffn_out))这里有一个实现细节Decoder 的 self-attention 需要同时屏蔽 pad 和未来 token所以上面代码里出现了combined_mask。常见做法是把 padding mask 和 causal mask 做逐位取最大也就是tf.maximum(padding_mask, causal_mask)这样在任何一种需要屏蔽的位置上都会生效。cross-attention 只需要屏蔽 Encoder 输出里的 pad 位置因为未来 token 这个约束只对 Decoder 自身有效。两个 mask 千万不要混用否则翻译结果会出现“看到答案再作答”的诡异现象训练 loss 低得离谱推理时却驴唇不对马嘴。3.4 组装成整体模型输入三张量输出一个概率分布整个 Transformer 模型接受三个输入encoder 输入序列、decoder 输入序列以及对应的 mask。这里 mask 是动态计算出来的不是固定参数。我习惯用一个 Keras Model 把所有块串起来这样model.fit()可以自动处理数据管道。class Transformer(keras.Model): def __init__(self, vocab_en, vocab_zh, d_model, num_heads, dff, max_len, dropout): super().__init__() self.embed_en keras.layers.Embedding(vocab_en, d_model) self.embed_zh keras.layers.Embedding(vocab_zh, d_model) self.pos_enc PositionalEncoding(d_model, max_len) self.encoder_blocks [EncoderBlock(d_model, num_heads, dff, dropout) for _ in range(2)] # 先用 2 层 self.decoder_blocks [DecoderBlock(d_model, num_heads, dff, dropout) for _ in range(2)] self.final_layer keras.layers.Dense(vocab_zh) def call(self, en_input, zh_input, trainingFalse): # 输入天然要求 [batch, seq_len]embedding 后叠加位置编码 enc_x self.pos_enc(self.embed_en(en_input)) enc_pad_mask create_padding_mask(en_input) for block in self.encoder_blocks: enc_x block(enc_x, mask_convert(enc_pad_mask)) zh_pad_mask create_padding_mask(zh_input) causal_mask create_causal_mask(tf.shape(zh_input)[1]) combined_mask tf.maximum(zh_pad_mask, causal_mask) dec_x self.pos_enc(self.embed_zh(zh_input)) for block in self.decoder_blocks: dec_x block(dec_x, enc_x, mask_convert(enc_pad_mask), mask_convert(combined_mask), causal_mask) return self.final_layer(dec_x)mask_convert其实就是tf.cast(mask 0, tf.float32) * -1e9新版 Keras 可以直接传 bool mask但为了兼容性我还是转成 float。上面的代码刻意把层数压到 2 层是因为这套源码要让你“先跑起来”。把层数加到 6 层、D_MODEL 提到 512就是真正可用规模但显存和训练时长会成倍增长。先让模型学会拟合一小批数据再慢慢加容量这是最稳的调模型路线。4. 训练与推理让模型真正开始翻译而不是背诵4.1 损失函数与学习率计划训练翻译模型有两个默认选择损失函数用 SparseCategoricalCrossentropy且必须打开from_logitsTrue因为模型最后一层没有接 Softmax。如果忘记这个参数你会看到 Loss 一开始就在 7 以上而且怎么调都降不下去原因是 Keras 默认认为输出已经是概率分布又在内部做了一次 softmax 归一化。学习率不能固定Transformer 论文里给出的 Wamup 学习率计划务必照抄先线性增长再按步数倒数衰减这个方案对翻译任务非常关键。class TransformerSchedule(keras.optimizers.schedules.LearningRateSchedule): def __init__(self, d_model, warmup_steps4000): 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.math.minimum(arg1, arg2) schedule TransformerSchedule(D_MODEL) optimizer keras.optimizers.Adam(schedule, beta_10.9, beta_20.98, epsilon1e-9) loss_obj keras.losses.SparseCategoricalCrossentropy(from_logitsTrue)warmup_steps4000不是硬性标准小数据集可以降到 1000-2000太大反而会让模型前期收敛过慢。beta_2从默认的 0.999 改成 0.98是针对 Transformer 这类梯度稀疏任务的常见调整作用是在训练早期让优化器更敏感地响应梯度变化。不要小看这个数字很多复现项目 loss 曲线一开始就震荡十有八九没改beta_2。4.2 训练循环与模型保存数据集较小的时候直接用model.fit配合 tf.data API 最省事。注意每轮训练要打乱数据顺序并且建议按序列长度做简单的 bucket padding否则短句和长句混在一个 batch 里pad 比例过高模型大部分计算都浪费在无效位置上。下面是最小可跑的代码train_ds tf.data.Dataset.from_tensor_slices( (en_padded, decoder_input, target)) train_ds (train_ds .shuffle(10000) .batch(BATCH_SIZE) .prefetch(tf.data.AUTOTUNE)) model Transformer(VOCAB_EN, VOCAB_ZH, D_MODEL, NUM_HEAD, DFF, MAX_LEN, DROPOUT) model.compile(optimizeroptimizer, lossloss_obj, metrics[keras.metrics.SparseCategoricalAccuracy(nameacc)]) # 训练时额外传入一个空 dict 即可因为 mask 在 call 里现算 model.fit(train_ds, epochs20) # 保存权重不保存完整模型结构便于后续升级代码 model.save_weights(transformer_zh_en.h5)model.fit的优点是自带进度条、验证集评估和 early stopping 回调。如果你发现 loss 降得很慢先不要怀疑模型结构优先检查数据集 token 化有没有问题比如中文句子被错误地按英文空格切分或者unk占比超过 15%。这些数据层面的问题远远比模型结构更影响最终效果。保存权重用 h5 格式兼容性最好新版 Keras 的.keras格式也能用但如果你要在老项目中加载h5 是后悔药一样的存在。4.3 推理阶段真正翻译时的自动回归解码训练时模型一次性给出整句话的预测因为用了 teacher forcing。推理时没有标准答案只能用贪心解码或者 beam search把上一个预测词作为下一个 decoder 输入。这是整个项目里最容易写出 bug 的地方。def greedy_decode(model, en_sentence, max_lenMAX_LEN): en_tokens en_tokenizer.encode(en_sentence) en_tokens tf.constant([en_tokens], dtypetf.int32) # decoder 初始输入只有 start zh_tokens tf.constant([[START_ID]], dtypetf.int32) for _ in range(max_len): pred model(en_tokens, zh_tokens, trainingFalse) next_token tf.argmax(pred[:, -1, :], axis-1)[-1] zh_tokens tf.concat([zh_tokens, [[next_token]]], axis-1) if next_token END_ID: break return zh_tokenizer.decode(zh_tokens[0].numpy())循环里有两处要解释一是 decoder 的输入长度一直在变但位置编码层会根据当前长度自动截取对应位置向量所以模型可以处理变长序列二是每次都要重新完整前向计算一遍没有 KV Cache所以推理速度慢。这里有条件的项目可以缓存 Encoder 输出因为源语言不变Encoder 只需要跑一次Decoder 每次迭代仍然要重算但省掉 Encoder 后速度也已经能快近一倍。5. 直接跑项目时最常翻车的 4 个位置与排查顺序5.1 Loss 卡在 6 到 7 之间一动不动甚至越训越高现象很典型训练和验证 Loss 都高得离谱模型输出几乎全是unk。原因分两类。第一类是from_logits设置错误模型输出了 logits损失函数却把它当概率分布处理。第二类是词表严重不匹配比如训练时中文词表用的是全角字符推理时用户输入半角标点导致同一句话 token 化结果完全不一样。解决方法是先打印一个 batch 的模型输出看数值范围。logits 通常是正负几十的实数概率值则集中在 0 到 1 之间。如果数据没问题再检查 tokenizer 是否有未登录词占比过高的警告。源码包里通常会附一个debug_dataset.py脚本跑一遍就能看到对齐后的句子对是否错位。5.2 训练 loss 很低但翻译结果是复读机这是我见过最隐蔽的坑。模型在训练时准确率高达 90% 以上一到推理阶段就反复输出同一个词比如把“你好世界”翻译成“hello hello hello hello”。原因是训练和推理的输入分布不一致训练时 decoder 看到的是真实前词推理时看到的是自己预测的词一旦第一步预测错后续全部被带偏。缓解手段有三种一是推理时用 beam search维护多条候选路径避免一条道走到黑二是数据量大的话在训练中随机把部分真实 token 替换成模型预测 token这叫 scheduled sampling三是降低 dropout推理时会默认关闭 dropout训练时如果 dropout 太大整体模型会过度依赖训练阶段的随机噪声。实操中先用 beam search不要一上来就动训练流程。5.3 Loss 曲线锯齿状每个 batch 差异巨大训练时如果设置了验证集会看到每个 epoch 内 acc 忽高忽低看起来毫无规律。原因是 batch 内部句子长度差异太大有的句子只有 5 个 token有的 60 个padding 后有效信息占比波动剧烈。比如一个 batch 里全是长句pad 少loss 就低另一个 batch 里全是短句大量 padloss 就高。解决方法是给句子按长度分桶。把长度差在 5 个 token 以内的句子放到同一个 bucket再从 bucket 里取样组 batch。tf.data 里可以用bucket_by_sequence_length或者更简单地在数据预处理时先按长度排序再按 batch_size 切片每个 batch 内最大长度差不超过 8 个 token。这个操作不改任何模型代码经常能让 loss 曲线立刻平滑下来。5.4 中文“词”没有被真正切出来如果模型对“人工智能”这种词翻译得七零八落多半不是模型问题是中文切分问题。直接用字符级 token 的表会让模型需要更多步数才能学到词级语义。理想方案是 jieba 分词 子词融合但更省事的是用 SentPiece 在中文语料上做 unigram 模型词表大小 8000 到 20000 都比较常见。注意不要拿英文的 BPE 模型直接套中文中文里单字即语义单元BPE 合并后容易产生没有意义的整词碎片。解决方法是先把中文文本用 jieba.cut 切分再把切分结果送入 tokenizer。源码里如果 tokenizer 是直接用字符扫的改成先切分再编码通常能立刻提升几个百分点的 BLEU。6. 把翻译质量往上再拔一档BLEU、Beam Search 与调优顺序6.1 极简 Beam Search 写法和效果差异贪心解码每一步只挑概率最大的词一旦开头选错后面很难回头。Beam Search 的做法是每一步保留 top-k 条候选路径最终从 k 条路径里选整句概率最高的一条。k3 到 5 时性价比最高k 再大推理时间会明显增加但效果提升趋于平缓。def beam_decode(model, en_tokens, beam_size4, max_len64): start [[START_ID]] scores [0.0] for _ in range(max_len): all_candidates [] for seq, score in zip(start, scores): zh_input tf.constant([seq], dtypetf.int32) pred model(en_tokens, zh_input, trainingFalse) log_probs tf.nn.log_softmax(pred[0, -1, :]) # 取 log 概率 top_k tf.math.top_k(log_probs, kbeam_size) for token, logp in zip(top_k.indices.numpy(), top_k.values.numpy()): all_candidates.append((seq [token], score logp)) # 按总分排序只保留 beam_size 条 all_candidates.sort(keylambda x: x[1], reverseTrue) start [cand[0] for cand in all_candidates[:beam_size]] scores [cand[1] for cand in all_candidates[:beam_size]] if any(seq[-1] END_ID for seq in start): break return start[scores.index(max(scores))]需要注意的是这里用log_softmax而不是 softmax因为多步概率相乘时浮点数会下溢累加 log 概率才稳定。beam search 做出来之后你还会发现一个问题beam size 越大句子越倾向安全、平淡的翻译因为概率累加惩罚了长句。这时候可以引入 length normalization常见的做法是除以(seq_len ** 0.7)这段修正会让偏短的译文明显改善这也是很多项目直接把论文里的 alpha0.7 拿来做默认值的原因。6.2 评估别只盯着“翻得顺不顺”人工看几条翻译结果容易产生错觉最好用 BLEU 做客观评估。最简单的计算方式是sacrebleu库把参考译文和模型译文按行喂进去。需要注意 BLEU 对中文这种分词方式敏感中译英时官方标准通常按空格分词。建议固定 200 句测试集每次改动后都跑同一份记录 BLEU 和 chrF 两个指标。BLEU 提升 1 个点可能对应一次成功的调参也可能是运气波动至少要重复三次取平均。我做这类中英翻译项目的习惯是先把英译中和中译英都跑一个 2 层的迷你模型确认 pipeline 完整、loss 能降再逐步加深到 4 层、6 层并调大 D_MODEL每次改动只动一个变。这套基于 python 的 kreas-transformer 源码适合作为起步骨架但真正要投入生产还需要加入领域语料、长度标准化和更细的评估集。先把上面几个坑绕过去你的第一版翻译模型很快就能跑出像样的句子希望帮到你。本文还有配套的精品资源点击获取