简介这是一套面向高校毕业设计、课程实践与项目原型开发的中英文机器翻译系统源码包基于Python与Keras-Transformer框架构建采用模块化方式封装标准Transformer组件实现中英文双向翻译功能。资源共21个文件压缩包约13.47MB包含py主程序与ipynb实验笔记、h5训练权重、pkl词典与中间数据、pyc与zbak备份文件及md说明文档覆盖数据预处理、模型训练到翻译推理的完整链路。已有72人学习下载。代码经过多轮完整性与功能性验证可直接部署运行适合作为开发基础进行功能扩展与性能优化。作者建议与基于LSTM的机器翻译项目协同学习两者训练数据与预处理流程一致便于对比Transformer与LSTM在翻译任务中的表现差异深入理解不同神经网络架构的特性。1. 从零搭一套中英翻译为什么我选 Python Keras-Transformer 而不是调 API线上翻译接口用起来确实省事但一旦你要做垂直领域适配、要控制成本、要把模型塞进内网环境调 API 这条路就走不通了。我去年接的一个跨境电商商品描述翻译需求客户明确要求模型必须跑在自己的服务器上数据不能出内网这时候基于 Python 与 Keras-Transformer 的中英文机器翻译系统就成了一个绕不开的方案。Python 负责数据处理和训练调度Keras-Transformer 负责序列到序列的建模整套东西从分词、编码、训练到推理全部可以在一台带 GPU 的机器上跑通。这篇文章面向的是有 Python 基础、想自己动手训一个翻译模型的工程师。我不会只讲 Transformer 的注意力公式而是把重点放在环境怎么配、数据怎么洗、模型怎么搭、训练怎么调、推理怎么部署、坑在哪里。如果你之前只跑过python入门级别的脚本跟着走也能把最小可用版本跑起来如果你已经用过其他框架训过 seq2seq这里关于 Keras 实现细节和参数选择的经验能帮你少走弯路。整套方案的核心价值在于可控——模型权重在你手里词表在你手里推理服务也在你手里。2. 环境准备与数据管道把中英文语料变成模型能吃的张量2.1 用 conda 隔离环境避开 TensorFlow 与 Keras 版本打架Keras-Transformer 这类实现对 TensorFlow 版本比较敏感我一般用 conda 建独立环境避免和系统里已有的python安装版本冲突。下面这套命令在 Ubuntu 和 Windows WSL 下都验证过conda create -n mt-keras python3.9 -y conda activate mt-keras pip install tensorflow2.11.0 keras2.11.0 numpy pandas tqdm pip install sentencepiece # 用于子词分词逻辑说明TensorFlow 2.11 自带 Keras 2.11这个组合对自定义 Transformer 层的支持比较稳定。参数上Python 选 3.9 是因为 3.10 以上在某些 CUDA 版本下会有兼容问题。装完后用python -c import tensorflow as tf; print(tf.__version__)验证输出2.11.0即可。如果你用vscode配置python环境记得在 VSCode 里把解释器切到mt-keras这个 conda 环境否则终端里跑通了、编辑器里还是报模块找不到。注意不要混用 pip 和 conda 装 TensorFlow我见过有人 conda 装了 TF 又 pip 装了一遍结果 GPU 识别不出来排查了半天。2.2 语料清洗中英文平行语料的四个过滤规则原始平行语料里噪声很多直接喂给模型会导致 loss 震荡。我一般按下面四条规则过滤规则阈值原因句对长度比中文/英文长度比在 0.3~3.0 之间过滤掉明显不对齐的句对单句最大长度中文不超过 150 字英文不超过 100 词控制显存占用空白与乱码去除含连续特殊符号的句子避免词表被噪声污染重复句对去重后保留唯一句对防止模型过拟合高频句import re def clean_pair(zh, en): zh zh.strip() en en.strip() if not zh or not en: return None # 长度比过滤 ratio len(zh) / max(len(en.split()), 1) if ratio 0.3 or ratio 3.0: return None # 长度上限 if len(zh) 150 or len(en.split()) 100: return None # 特殊符号过滤 if re.search(r[^\w\s\u4e00-\u9fff.,!?;:\\-], zh en): return None return zh, en逻辑说明clean_pair返回None表示该句对被丢弃。参数上长度比阈值 0.3~3.0 是我在 200 万句对上的经验值如果你的语料领域特殊比如法律文本中英文长度差异大可以放宽到 0.2~4.0。特殊符号正则里保留了中英文常用标点去掉的是表情、控制字符等。2.3 用 SentencePiece 训练子词词表统一中英文切分中英文混合场景下按字切分中文、按词切分英文会导致词表爆炸。我一般用 SentencePiece 训练一个共享的 BPE 词表中英文都走同一套子词切分import sentencepiece as spm # 合并中英文语料到一个文件每行一句 spm.SentencePieceTrainer.train( inputall_corpus.txt, model_prefixmt_spm, vocab_size32000, model_typebpe, character_coverage0.9995, pad_id0, unk_id1, bos_id2, eos_id3 )逻辑说明vocab_size32000是翻译任务的常用值太小会导致未登录词多太大则 embedding 矩阵占显存。character_coverage0.9995保证覆盖中文里的生僻字。训练完后会得到mt_spm.model和mt_spm.vocab两个文件后续编码解码都用它。参数上pad_id到eos_id的顺序要和模型里的定义一致否则训练时 mask 会出错。3. Keras-Transformer 模型搭建编码器、解码器与注意力层的实现细节3.1 位置编码为什么不能直接用可学习 embeddingTransformer 本身没有循环结构位置信息必须显式注入。Keras 里常见两种做法一种是可学习的位置 embedding另一种是正弦位置编码。我在翻译任务上更倾向正弦编码原因是训练语料长度分布不均时可学习 embedding 在长句上泛化差。下面是我用的位置编码层import tensorflow as tf import numpy as np class PositionalEncoding(tf.keras.layers.Layer): def __init__(self, max_len, d_model): super().__init__() self.max_len max_len self.d_model d_model def build(self, input_shape): pos np.arange(self.max_len)[:, np.newaxis] i np.arange(self.d_model)[np.newaxis, :] angle pos / np.power(10000, (2 * (i // 2)) / self.d_model) angle[:, 0::2] np.sin(angle[:, 0::2]) angle[:, 1::2] np.cos(angle[:, 1::2]) self.pos_encoding tf.constant(angle[np.newaxis, ...], dtypetf.float32) def call(self, x): seq_len tf.shape(x)[1] return x self.pos_encoding[:, :seq_len, :]逻辑说明build里预计算好位置编码矩阵call里按实际序列长度切片相加。参数上max_len设成 200 足够覆盖大部分句对d_model要和 embedding 维度一致。注意angle[:, 0::2]和angle[:, 1::2]分别处理偶数和奇数维度这是正弦编码的标准写法写反了模型也能训但收敛会慢。3.2 多头注意力把 Q、K、V 的维度拆对多头注意力是 Transformer 的核心Keras 实现时最容易搞错的是维度拆分。我一般把d_model拆成num_heads份每份depth d_model // num_headsclass MultiHeadAttention(tf.keras.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 tf.keras.layers.Dense(d_model) self.wk tf.keras.layers.Dense(d_model) self.wv tf.keras.layers.Dense(d_model) self.dense tf.keras.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) scaled_attention, _ self.scaled_dot_product_attention(q, k, v, mask) scaled_attention tf.transpose(scaled_attention, perm[0, 2, 1, 3]) concat tf.reshape(scaled_attention, (batch_size, -1, self.d_model)) return self.dense(concat)逻辑说明split_heads把(batch, seq_len, d_model)变成(batch, num_heads, seq_len, depth)这样每个头独立算注意力。scaled_dot_product_attention里做softmax(QK^T / sqrt(depth))mask 用来屏蔽 padding 和未来位置。参数上num_heads一般取 8d_model取 512这样depth64。如果显存不够优先降d_model而不是num_heads头数太少注意力表达会变弱。3.3 编码器与解码器堆叠层数、Dropout 与残差连接编码器和解码器各堆 6 层是 Transformer 原论文的配置我在翻译任务上一般保持 6 层但会根据语料规模调整。每层里残差连接和 LayerNorm 的顺序会影响训练稳定性class EncoderLayer(tf.keras.layers.Layer): def __init__(self, d_model, num_heads, dff, dropout_rate0.1): super().__init__() self.mha MultiHeadAttention(d_model, num_heads) self.ffn tf.keras.Sequential([ tf.keras.layers.Dense(dff, activationrelu), tf.keras.layers.Dense(d_model) ]) self.layernorm1 tf.keras.layers.LayerNormalization(epsilon1e-6) self.layernorm2 tf.keras.layers.LayerNormalization(epsilon1e-6) self.dropout1 tf.keras.layers.Dropout(dropout_rate) self.dropout2 tf.keras.layers.Dropout(dropout_rate) def call(self, x, training, mask): attn_output self.mha(x, x, x, mask) attn_output self.dropout1(attn_output, trainingtraining) out1 self.layernorm1(x attn_output) ffn_output self.ffn(out1) ffn_output self.dropout2(ffn_output, trainingtraining) return self.layernorm2(out1 ffn_output)逻辑说明这里用的是 Post-LN 结构即LayerNorm(x sublayer(x))。dff是前馈网络中间层维度一般取4 * d_model 2048。dropout_rate0.1是默认值语料小于 50 万句对时可以提到 0.2 防过拟合。残差连接保证梯度能传到底层LayerNorm 的epsilon1e-6避免除零。4. 训练与调参loss 不降、显存爆了、BLEU 上不去怎么排查4.1 学习率预热与 Adam 优化器配置Transformer 训练初期 loss 容易震荡标准做法是学习率预热。我一般用自定义 scheduleclass CustomSchedule(tf.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) optimizer tf.keras.optimizers.Adam( CustomSchedule(512), beta_10.9, beta_20.98, epsilon1e-9 )逻辑说明warmup_steps4000表示前 4000 步学习率线性上升之后按步数平方根倒数下降。beta_20.98比默认的 0.999 更适合 Transformer因为梯度方差较大。参数上如果 batch size 调大warmup_steps 可以相应减小到 2000。4.2 显存不够时的三个降级策略训练翻译模型时显存爆了是常态。我按优先级用这三个策略第一把batch_size从 64 降到 32 或 16配合梯度累积模拟大 batch第二把max_len从 200 截到 128长句单独处理第三把d_model从 512 降到 256但 BLEU 会掉 1~2 个点。梯度累积的写法accum_steps 4 for step, (inp, tar) in enumerate(dataset): with tf.GradientTape() as tape: predictions model(inp, tar, trainingTrue) loss loss_function(tar, predictions) loss loss / accum_steps gradients tape.gradient(loss, model.trainable_variables) if (step 1) % accum_steps 0: optimizer.apply_gradients(zip(gradients, model.trainable_variables))逻辑说明accum_steps4表示每 4 个 batch 更新一次参数等效 batch size 翻 4 倍。注意 loss 要除以accum_steps否则梯度会放大。4.3 BLEU 评估为什么训练 loss 降了翻译质量却没提升训练 loss 用的是 teacher forcing即解码器输入是真实标签而推理时用的是模型自己生成的 token两者分布不一致这就是 exposure bias。我一般每训练 5 个 epoch 做一次 BLEU 评估from nltk.translate.bleu_score import corpus_bleu def evaluate_bleu(model, dataset, sp_model): references, hypotheses [], [] for inp, tar in dataset: output greedy_decode(model, inp, sp_model) for i in range(len(output)): ref sp_model.decode(tar[i].numpy().tolist()).split() hyp sp_model.decode(output[i].numpy().tolist()).split() references.append([ref]) hypotheses.append(hyp) return corpus_bleu(references, hypotheses)逻辑说明greedy_decode是逐 token 生成corpus_bleu计算语料级 BLEU。参数上如果 BLEU 在 20 以上说明模型基本可用30 以上算不错。注意评估时要关掉 dropout即trainingFalse。5. 避坑与排查五个让我加班到凌晨的翻车现场5.1 现象loss 一直是 nan训练几步就崩原因学习率太大或者位置编码里出现了除零。我遇到过一次是d_model设成了 0 导致rsqrt报错还有一次是 warmup_steps 设成 0 导致学习率无穷大。解决检查d_model和warmup_steps是否为正整数学习率 schedule 里加tf.math.maximum(step, 1.0)防止 step 为 0。5.2 现象模型只输出 BOS 然后重复同一个词原因解码器 mask 写错了导致注意力看到了未来位置模型学不到有效对齐。解决检查look_ahead_mask和padding_mask的组合逻辑确保解码器自注意力里上三角被屏蔽。我一般用tf.linalg.band_part生成 mask比手动构造不容易错。5.3 现象GPU 利用率只有 20%训练速度慢原因数据管道成了瓶颈tf.data没有预取。解决在 dataset 后面加.prefetch(tf.data.AUTOTUNE)和.cache()如果内存够的话。另外batch_size太小也会导致 GPU 等数据我一般把batch_size提到 64 以上。5.4 现象推理时翻译结果比训练时短很多原因解码时 EOS 概率过高模型过早停止。解决在 greedy decode 里加长度惩罚或者用 beam search 代替 greedy。我一般用 beam width4 的 beam searchBLEU 能涨 2~3 个点。5.5 现象保存的模型加载后推理结果全乱原因自定义层没有实现get_config加载时权重对不上。解决给每个自定义 Layer 实现get_config方法保存时用model.save(mt_model)而不是只存权重。加载时用tf.keras.models.load_model(mt_model, custom_objects{...})。6. 推理部署与效果验证把模型变成能用的翻译服务训练完模型只是第一步真正要用起来还得包一层推理服务。我一般用 FastAPI 起一个 HTTP 接口把模型加载到内存里请求进来直接调greedy_decode或beam_search。下面是一个最小可用的推理服务from fastapi import FastAPI import tensorflow as tf import sentencepiece as spm app FastAPI() model tf.keras.models.load_model(mt_model, custom_objects{...}) sp spm.SentencePieceProcessor(model_filemt_spm.model) def translate(text, max_len100): tokens sp.encode(text, out_typeint) tokens [2] tokens [3] # BOS tokens EOS encoder_input tf.constant([tokens]) output beam_search(model, encoder_input, sp, max_len) return sp.decode(output) app.post(/translate) def do_translate(payload: dict): return {result: translate(payload[text])}逻辑说明beam_search里维护 beam width 个候选序列每步选概率最高的 top-k 扩展。参数上max_len100控制输出长度beam_width4是速度和质量的折中。部署时用uvicorn main:app --host 0.0.0.0 --port 8000启动压测可以用ab或wrk。验证翻译质量不能只看 BLEU我一般还会人工抽 50 句看三类问题漏译、重复翻译、专有名词错误。漏译通常是max_len设太小重复翻译是解码策略问题专有名词错误需要在训练语料里补充领域词典。如果要做python生成exe可执行文件给客户离线用可以用 PyInstaller 把推理脚本打包但模型文件要单独放不然 exe 会很大。最后说一个我踩过的坑有次客户反馈翻译服务跑了一周后内存暴涨排查发现是每次请求都重新加载了 SentencePiece 模型。后来改成全局加载一次内存就稳了。这种问题不写日志根本看不出来所以推理服务里一定要加请求日志和内存监控。希望帮到你。本文还有配套的精品资源点击获取