从零构建中英翻译系统：Python与Keras-Transformer实战指南

简介这是一套面向高校毕业设计、课程实践与项目原型开发的中英文双向机器翻译系统源码包基于Python与Keras-Transformer框架构建采用模块化方式封装标准Transformer组件适合具备一定深度学习基础、希望动手实践神经机器翻译的学习者。压缩包共21个文件约13.47MB包含py主程序与ipynb实验笔记、h5训练权重、pkl词表与中间数据、pyc与zbak备份文件以及md说明文档覆盖数据预处理、模型训练到翻译推理的完整链路。资源提供可执行程序、源代码与技术文档代码经过完整性与功能性验证可直接部署运行并支持在此基础上扩展功能与优化性能。已有72人学习下载。建议与基于LSTM的机器翻译项目对照使用两者训练数据与预处理流程一致便于直观比较Transformer与LSTM在翻译任务中的表现差异深入理解不同网络架构的特性。1. 从零搭一套中英翻译为什么我选 Python Keras-Transformer 而不是调 API线上翻译 API 用起来确实省事但一旦你要做垂直领域术语定制、要离线部署、要控制单次推理成本调用外部接口就会变成瓶颈。我去年接过一个跨境电商商品标题翻译的需求术语表里全是「加厚保暖」「防泼水」这类词通用 API 翻出来要么丢词要么语序乱最后只能自己训一个。这套基于 Python 与 Keras-Transformer 的中英文机器翻译系统核心就是用 Keras 3 把 Transformer 的编码器-解码器结构手写出来配合中英平行语料做训练最终导出一个能本地跑的翻译模型。它适合两类人一是想搞懂 Transformer 内部到底怎么算的工程师二是需要一个可定制、可离线、能塞进自己术语表的翻译底座。整条链路从环境配置、数据预处理、模型搭建、训练调参到推理部署每一步我都会给出能直接抄的代码和参数说明踩过的坑也一并写清楚。2. 环境准备与中英平行语料处理从 python 安装到分词器落地2.1 用 conda 隔离环境避开 Keras 3 与 TensorFlow 的版本打架很多人第一步就翻车pip install keras装完发现后端是 TensorFlow 2.x而 Keras 3 的多后端机制要求你显式指定后端。我一般用 conda 建一个干净环境把 Python 版本锁在 3.10因为 3.11 之后部分科学计算库的 wheel 还没跟上。conda create -n mt-keras python3.10 -y conda activate mt-keras pip install tensorflow2.16.1 keras3.3.3 numpy1.26.4 pip install sentencepiece0.2.0 pandas2.2.2 tqdm4.66.4逻辑说明TensorFlow 2.16 是第一个原生支持 Keras 3 的稳定版低于这个版本 Keras 3 会报后端不兼容。sentencepiece用来做子词分词中英混合语料用 BPE 比按字切分更稳。参数上Python 3.10 是当前生态兼容性最好的版本numpy 锁 1.26 是因为 2.x 改了 ABI部分旧 wheel 会崩。装完验证一下后端import keras print(keras.__version__) # 应输出 3.3.3 print(keras.backend.backend()) # 应输出 tensorflow如果第二行输出tensorflow说明后端绑定成功。要是输出别的检查环境变量KERAS_BACKEND有没有被全局设成 jax 或 torch。2.2 中英平行语料的清洗与长度过滤公开的中英平行语料常见的有 WMT、OPUS 里的 News-Commentary、TED2020 等。我拿到的原始文件通常是制表符分隔的en\tzh两列但里面混着大量空行、HTML 标签、中英不对齐的脏数据。清洗脚本我一般这么写import re import pandas as pd def clean_pair(en, zh): en re.sub(r[^], , en).strip() zh re.sub(r[^], , zh).strip() if not en or not zh: return None # 过滤长度比异常的对中英字符数比一般在 0.3~3 之间 ratio len(zh) / max(len(en), 1) if ratio 0.3 or ratio 3.0: return None # 过滤纯符号或纯数字 if re.fullmatch(r[\W\d_], en) or re.fullmatch(r[\W\d_], zh): return None return en, zh pairs [] with open(raw_corpus.tsv, encodingutf-8) as f: for line in f: parts line.rstrip(\n).split(\t) if len(parts) ! 2: continue res clean_pair(parts[0], parts[1]) if res: pairs.append(res) df pd.DataFrame(pairs, columns[en, zh]) df.drop_duplicates(subset[en], inplaceTrue) df.to_csv(clean_corpus.csv, indexFalse) print(f清洗后剩余 {len(df)} 对)逻辑说明长度比过滤是最有效的一刀中英翻译里如果中文长度是英文的 5 倍以上基本是噪声。去重按英文侧去因为同一句英文可能对应多个中文译法保留一个就够训练用。参数上0.3 和 3.0 这两个阈值是我在 200 万对语料上试出来的再放宽会引入大量低质对收紧到 0.5~2.0 又会让短句被误杀。2.3 用 SentencePiece 训练中英联合分词模型Transformer 的输入是 token id 序列中英混合语料不能简单按空格切。我一般训练一个联合的 SentencePiece 模型词表大小设 32000这样中英共享子词单元低频词也能被拆成子词而不是全变 UNK。import sentencepiece as spm # 把清洗后的中英文分别写成两个纯文本文件 with open(all_en.txt, w, encodingutf-8) as f: f.write(\n.join(df[en].tolist())) with open(all_zh.txt, w, encodingutf-8) as f: f.write(\n.join(df[zh].tolist())) # 合并训练联合词表 import itertools with open(all_joint.txt, w, encodingutf-8) as f: for en, zh in zip(df[en], df[zh]): f.write(en \n) f.write(zh \n) spm.SentencePieceTrainer.train( inputall_joint.txt, model_prefixspm_joint, vocab_size32000, model_typebpe, character_coverage0.9995, pad_id0, unk_id1, bos_id2, eos_id3, pad_piecepad, unk_pieceunk, bos_pieces, eos_piece/s, num_threads8 )逻辑说明character_coverage0.9995对中文很重要覆盖不到的生僻字会变 UNK设 1.0 又会让词表被生僻字撑大。model_typebpe比 unigram 在中英混合场景下更稳因为 BPE 的合并规则对两种语言都公平。pad/unk/bos/eos 的 id 固定成 0/1/2/3后面模型里直接按这个索引取 embedding不用再查。训练完会得到spm_joint.model和spm_joint.vocab加载验证sp spm.SentencePieceProcessor(model_filespm_joint.model) ids sp.encode(I love machine translation, out_typeint) print(ids) print(sp.decode(ids))如果 decode 回来和原文一致说明分词器可用。注意中文侧不要额外加空格SentencePiece 对中文是按字符处理的加了空格反而会切碎。3. Keras-Transformer 模型搭建编码器、解码器与注意力掩码3.1 位置编码为什么不能直接用可学习 EmbeddingTransformer 本身没有循环结构位置信息全靠位置编码注入。Keras 官方教程里常用Embedding(input_dimmax_len, output_dimd_model)做可学习位置编码但我在中英翻译任务上对比过正弦位置编码在长句上泛化更好因为训练集里没出现过的长度可学习编码直接没见过正弦编码还能靠公式外推。import numpy as np import tensorflow as tf from tensorflow import keras from keras import layers def sinusoidal_pos_encoding(max_len, d_model): positions np.arange(max_len)[:, np.newaxis] dims np.arange(d_model)[np.newaxis, :] angles positions / np.power(10000, (2 * (dims // 2)) / d_model) angles[:, 0::2] np.sin(angles[:, 0::2]) angles[:, 1::2] np.cos(angles[:, 1::2]) return tf.cast(angles[np.newaxis, :, :], dtypetf.float32)逻辑说明dims // 2保证奇偶维度用不同的频率偶数维 sin、奇数维 cos这样任意两个位置的点积只和相对距离有关。参数上max_len设 128 够用中英商品标题和短句基本不超过这个长度长文档翻译再往上加。3.2 多头注意力与 Padding Mask 的正确写法Transformer 的注意力层里mask 写错是最隐蔽的 bug——loss 会降但推理时输出全是重复词。中英翻译里有两类 mask编码器的 padding mask 和解码器的 causal mask 加 padding mask。class MultiHeadAttention(layers.Layer): def __init__(self, d_model, num_heads): super().__init__() self.num_heads num_heads self.d_model d_model assert d_model % num_heads 0 self.depth d_model // num_heads self.wq layers.Dense(d_model) self.wk layers.Dense(d_model) self.wv layers.Dense(d_model) self.dense layers.Dense(d_model) def split_heads(self, x, batch_size): x tf.reshape(x, (batch_size, -1, self.num_heads, self.depth)) return tf.transpose(x, perm[0, 2, 1, 3]) def call(self, v, k, q, mask): batch_size tf.shape(q)[0] q self.split_heads(self.wq(q), batch_size) k self.split_heads(self.wk(k), batch_size) v self.split_heads(self.wv(v), batch_size) scale tf.math.sqrt(tf.cast(self.depth, tf.float32)) scores tf.matmul(q, k, transpose_bTrue) / scale if mask is not None: scores (mask * -1e9) weights tf.nn.softmax(scores, axis-1) out tf.matmul(weights, v) out tf.transpose(out, perm[0, 2, 1, 3]) out tf.reshape(out, (batch_size, -1, self.d_model)) return self.dense(out)逻辑说明mask * -1e9是加性掩码的标准写法被 mask 的位置 softmax 后趋近 0。注意 mask 的形状要能广播到(batch, heads, seq_q, seq_k)我一般传(batch, 1, 1, seq_k)的 padding mask 和(1, 1, seq_q, seq_k)的 causal mask两者相加后统一传入。参数上-1e9不要用-inf否则 softmax 里会出现 NaN。3.3 编码器-解码器整体组装与训练配置把注意力、前馈网络、残差连接、LayerNorm 拼成 EncoderLayer 和 DecoderLayer再堆 N 层。我一般用 4 层编码器加 4 层解码器d_model 256heads 8这个规模在单张 12G 显存的卡上 batch_size 能开到 64。def build_transformer(vocab_size, max_len128, d_model256, num_heads8, dff1024, num_layers4, dropout0.1): inputs keras.Input(shape(None,), dtypeint32, nameencoder_input) dec_inputs keras.Input(shape(None,), dtypeint32, namedecoder_input) enc_emb layers.Embedding(vocab_size, d_model)(inputs) enc_emb * tf.math.sqrt(tf.cast(d_model, tf.float32)) enc_emb sinusoidal_pos_encoding(max_len, d_model)[:, :tf.shape(inputs)[1], :] enc_emb layers.Dropout(dropout)(enc_emb) dec_emb layers.Embedding(vocab_size, d_model)(dec_inputs) dec_emb * tf.math.sqrt(tf.cast(d_model, tf.float32)) dec_emb sinusoidal_pos_encoding(max_len, d_model)[:, :tf.shape(dec_inputs)[1], :] dec_emb layers.Dropout(dropout)(dec_emb) # 这里省略 EncoderLayer/DecoderLayer 的堆叠细节核心是残差LayerNorm enc_output enc_emb for _ in range(num_layers): enc_output EncoderLayer(d_model, num_heads, dff, dropout)(enc_output) dec_output dec_emb for _ in range(num_layers): dec_output DecoderLayer(d_model, num_heads, dff, dropout)( dec_output, enc_output) outputs layers.Dense(vocab_size, namefinal_logits)(dec_output) return keras.Model([inputs, dec_inputs], outputs) model build_transformer(vocab_size32000) model.compile( optimizerkeras.optimizers.Adam(learning_rate1e-4, beta_10.9, beta_20.98), losskeras.losses.SparseCategoricalCrossentropy( from_logitsTrue, ignore_class0), metrics[accuracy] ) model.summary()逻辑说明embedding 乘sqrt(d_model)是为了和位置编码的量级对齐不加这个缩放位置编码会淹没词嵌入。ignore_class0让 padding 位置的 loss 不参与梯度这是必须的否则模型会学会预测 pad。学习率用 1e-4 配 Adam 的 0.98 beta_2是 Transformer 原论文的 warmup 策略简化版实际训练时前 4000 步可以再加线性 warmup。4. 训练、推理与避坑从 loss 曲线到贪心解码4.1 训练循环里 teacher forcing 的输入构造训练时解码器输入是右移一位的目标序列第一个 token 是s最后一个 token 对应的标签是/s。这个错位如果搞反模型永远学不会。def make_dataset(df, sp, max_len128, batch_size64): en_ids [sp.encode(x, out_typeint)[:max_len-2] for x in df[en]] zh_ids [sp.encode(x, out_typeint)[:max_len-2] for x in df[zh]] enc_in keras.preprocessing.sequence.pad_sequences( en_ids, maxlenmax_len, paddingpost, value0) dec_in keras.preprocessing.sequence.pad_sequences( [[2] z for z in zh_ids], maxlenmax_len, paddingpost, value0) dec_tgt keras.preprocessing.sequence.pad_sequences( [z [3] for z in zh_ids], maxlenmax_len, paddingpost, value0) ds tf.data.Dataset.from_tensor_slices( ((enc_in, dec_in), dec_tgt)) return ds.shuffle(10000).batch(batch_size).prefetch(tf.data.AUTOTUNE) train_ds make_dataset(df, sp) model.fit(train_ds, epochs20)逻辑说明dec_in前面加sid2dec_tgt后面加/sid3这样第 t 步的输入预测第 t 步的输出错位一位。paddingpost配合ignore_class0才能正确忽略 pad。参数上 batch_size 64 是 12G 显存下的稳妥值显存够可以上 128但学习率要相应调大。4.2 贪心解码与 beam search 的取舍推理时最简单的贪心解码每步取概率最大的 token实现简单但容易陷入重复。beam search 保留 top-k 候选质量更好但慢 k 倍。我一般先用贪心验证模型是否收敛上线再换 beam search。def greedy_decode(model, sp, sentence, max_len128): enc_in keras.preprocessing.sequence.pad_sequences( [sp.encode(sentence, out_typeint)], maxlenmax_len, paddingpost, value0) dec_in tf.constant([[2]]) # s result [] for _ in range(max_len): logits model([enc_in, dec_in], trainingFalse) next_id tf.argmax(logits[:, -1, :], axis-1).numpy()[0] if next_id 3: # /s break result.append(next_id) dec_in tf.concat([dec_in, [[next_id]]], axis-1) return sp.decode(result) print(greedy_decode(model, sp, I love machine translation))逻辑说明每步把已生成的 token 拼到dec_in后面取最后一个位置的 logits 做 argmax。遇到/s就停。注意trainingFalse必须显式传否则 dropout 会在推理时生效输出会随机抖动。4.3 训练不收敛、输出重复、显存溢出三个高频翻车点现象一loss 降到 2.0 左右就不动了。原因通常是学习率太大导致在局部震荡或者 mask 写错让模型看到了未来 token。解决先把学习率降到 5e-5 跑 2000 步看 loss 是否继续降再检查 causal mask 是否真的把上三角遮住了打印scores矩阵看右上角是否接近 0。现象二推理输出「的的的的」重复。这是贪心解码的经典问题模型在某个位置卡在高频 token 上。解决加 repetition penalty对已生成的 token 的 logits 减一个惩罚项或者直接换 beam searchbeam 宽度设 4 就能明显缓解。现象三训练到一半 OOM。中英长句 padding 到 128 后注意力矩阵是batch × heads × 128 × 128batch 64 时显存占用不小。解决把max_len按实际语料的分位数设比如 95 分位是 80就设 80 而不是 128或者用梯度累积模拟大 batch。5. 模型导出与术语定制让翻译系统真正能落地训练完的模型要落地绕不开两件事导出成可独立加载的格式以及支持术语表注入。Keras 3 的model.save(mt_model.keras)会保存结构和权重加载时只要环境里有相同的自定义层定义就能恢复。但自定义层如果没注册加载会报Unknown layer。我一般给每个自定义层加keras.saving.register_keras_serializable()装饰器这样导出后换台机器也能直接keras.models.load_model(mt_model.keras)。术语定制是这套方案相比调 API 最大的价值。做法是在解码阶段做 constrained decoding把术语表里英文短语对应的中文 token 序列预先编码当解码器生成到该英文短语的起始位置时强制把候选 token 限制在术语的中文 token 上。实现上不用改模型结构只在greedy_decode的 argmax 前加一层过滤def decode_with_glossary(model, sp, sentence, glossary, max_len128): # glossary: {waterproof: 防泼水, ...} enc_in keras.preprocessing.sequence.pad_sequences( [sp.encode(sentence, out_typeint)], maxlenmax_len, paddingpost, value0) dec_in tf.constant([[2]]) result [] for step in range(max_len): logits model([enc_in, dec_in], trainingFalse)[:, -1, :] # 检查当前已生成的中文是否匹配术语触发条件 for en_term, zh_term in glossary.items(): if en_term in sentence and not any( t in sp.decode(result) for t in [zh_term]): zh_ids sp.encode(zh_term, out_typeint) if len(result) len(zh_ids) or result[-len(zh_ids):] ! zh_ids: # 强制下一个 token 走术语序列 if len(result) len(zh_ids): next_id zh_ids[len(result)] result.append(next_id) dec_in tf.concat([dec_in, [[next_id]]], axis-1) continue next_id tf.argmax(logits, axis-1).numpy()[0] if next_id 3: break result.append(next_id) dec_in tf.concat([dec_in, [[next_id]]], axis-1) return sp.decode(result)逻辑说明这段代码在每步解码前检查术语表如果英文原文里出现了术语且中文还没生成对应译法就强制按术语的中文 token 序列走。参数上术语表不宜过大超过 50 条后强制解码会明显拖慢速度而且多条术语重叠时优先级要自己定。实测在商品标题场景下术语命中率能从基线的 60% 提到 95% 以上。验证方法上我习惯用 BLEU 加人工抽检双轨。BLEU 用sacrebleu算但 BLEU 高不代表读起来顺所以每轮训练后我会随机抽 50 句做人工评分重点看术语有没有翻对、语序有没有乱。这套流程跑下来从零到能用的中英翻译模型单卡大概两天。最后说个血泪教训别一上来就堆层数4 层能收敛的语料加到 8 层只会让你多等一倍时间还不一定更好。先把数据清洗和 mask 写对比调模型结构重要得多。希望帮到你。本文还有配套的精品资源点击获取