简介:这是一套基于Python开发的古文到现代文机器翻译项目源码,面向毕业设计、课程设计与项目开发场景,尤其适合有一定Python基础、希望在NLP翻译方向快速落地参考实现的开发者。项目源码已经过严格测试,可直接运行并在此基础上扩展功能。资源包共12个文件,由11个Python脚本和1个Markdown说明文档组成,脚本覆盖数据预处理、TFRecord数据集操作、输入向量化、词表构建、神经网络搭建、注意力机制训练与测试、服务端通信等功能模块,压缩包整体仅19KB,代码轻量精炼,便于通读和二次修改。目前已有199人学习下载。通过阅读源码可掌握从原始语料到模型训练、再到译文输出的完整流程,理解注意力机制在古文与现代文转换中的作用;搭配README中的说明,可快速搭建运行环境,适合作为课设、毕设的工程基础或进一步改进的起点。
1. 古文机器翻译:一份能跑通毕设的 seq2seq+attention 源码包
做机器翻译方向的课程设计或毕业设计,最尴尬的不是模型写不出来,而是拿到一份源码不知道从哪个文件开始跑。这份基于 python 的古文到现代文机器翻译源码包,解决的正是这个问题:语料解析、文本清洗、词典构建、向量化、TFRecord 序列化、attention 训练、socket 演示服务,一条链路全部到位。它适合三类人——正在做毕设需要完整可复现工程的学生,课程设计要交可演示项目的开发者,以及想基于 seq2seq+attention 改造自己语料的人。文件数量不多,但每个脚本职责明确,按数据流跑一遍就能摸清整套翻译系统的骨架。
2. 先按数据流拆源码包:11 个 py 文件的调用顺序与最小跑通命令
拿到源码第一步不是打开训练代码慢慢读,而是数文件、排顺序。这个项目里的文件分工非常清楚,数据先经过一串预处理脚本变成磁盘上的 TFRecord,再进入训练脚本,最后测试脚本和服务脚本分别接住训练产物。顺序错了,后面每一步都会报「文件不存在」或者「shape 对不上」。
2.1 每个文件干什么:职责与输入输出对照表
把全部文件按阶段拆开看,依赖关系就很直观了:
| 文件 | 阶段 | 主要职责 | 常见输入 | 常见输出 |
|---|---|---|---|---|
| ParseXml.py | 语料解析 | 从 XML 中提取古文-现代文平行句对 | 原始 XML 语料 | 句对列表文件 |
| Pre-treatment.py | 文本清洗 | 去标签残留、统一字符、截断长度 | 平行句对 | 清洗后语料 |
| Text_vocab.py | 词典构建 | 统计频次,生成 word2id / id2word | 清洗后语料 | vocab 文件 |
| Input_vec.py | 向量化 | 把文本字符映射成 id 序列或向量 | vocab 文件 + 语料 | 向量化数组 |
| Senten_id_seq.py | 序列生成 | 把句子整理成定长 batch,处理 padding | 向量化结果 | 训练矩阵 |
| TFRecord_operate.py | 数据序列化 | 把 batch 写成 TFRecord,供训练读取 | batch 矩阵 | .tfrecord 文件 |
| neural_network.py | 模型定义 | 定义 encoder / decoder / attention 结构 | 超参数 | 模型对象 |
| Attention_train.py | 训练入口 | 加载 TFRecord,跑训练循环,保存权重 | TFRecord + 超参数 | checkpoint |
| attention_test.py | 测试出口 | 加载权重,对输入的句子做翻译 | checkpoint + 输入句 | 译文文本 |
| ServerSocket.py | 服务封装 | 用 socket 接收请求,返回译文 | 模型权重 | 翻译服务 |
| test.py | 冒烟测试 | 用少量样本快速验证链路通不通 | 少量样本 | 打印译文 |
| README.md | 文档 | 记录运行顺序与参数说明 | - | - |
按依赖关系看,数据流其实是一条直线加三个出口:ParseXml → Pre-treatment → Text_vocab → Input_vec → Senten_id_seq → TFRecord_operate,之后分支到 Attention_train(训练)、attention_test(测试)、ServerSocket(演示服务)。neural_network.py 是训练时被 import 的模型定义,test.py 是脱离主链路的冒烟测试脚本。这个结构符合一线工程的常规拆法——数据准备单独一段,模型实验单独一段,两者通过 TFRecord 解耦。
2.2 为什么中间绕一道 TFRecord:从「每 epoch 重处理」到「离线序列化」
第一次跑这种项目的人常有一个疑问:为什么不直接把 numpy 数组 feed 进模型,非要转 TFRecord?如果只是几百条样本的 demo,直接喂内存确实可以;但翻译模型的训练数据往往是几万到几十万句对,如果每个 epoch 都重新走一遍文本清洗和字符映射,大部分时间都耗在重复处理上,而且 shuffle 也不彻底。
TFRecord 的价值在于把「数据处理」和「模型训练」完全拆开:数据只处理一次,落盘后训练阶段用 TFRecordDataset 读取,由框架负责 shuffle、prefetch 和并行读取。对毕设答辩来说,这条链路本身也是一个很好的工作点——能讲清楚为什么用离线序列化,比单纯背 attention 公式更能让老师认可工作量。
提示:TFRecord 不是模型的一部分,是数据管道的一部分。如果 README 没有写清中间文件路径,直接看每个 py 文件开头的 argparse 默认值,路径参数一般都写在脚本里。
2.3 最小复现命令:按文件依赖顺序跑下来的实操
环境上,这类项目大概率是 TensorFlow 1.x 的写法,建议先用老版本环境复现,不要一上来就装最新的 TensorFlow:
# 建议用 python 3.6/3.7 + tensorflow 1.x 复现,先把环境定下来 conda create -n guwen python=3.6 conda activate guwen pip install tensorflow==1.15 numpy # 1) 原始语料 -> 平行句对 python ParseXml.py # 2) 平行句对 -> 清洗后文本 python Pre-treatment.py # 3) 清洗后文本 -> vocab 文件 python Text_vocab.py # 4) 文本 + vocab -> 向量化结果 python Input_vec.py # 5) 向量 -> 定长 batch 序列 python Senten_id_seq.py # 6) batch -> TFRecord python TFRecord_operate.py # 7) 训练与测试 python Attention_train.py python attention_test.py # 8) 启动演示服务 python ServerSocket.py这个顺序不能乱:每个脚本都依赖上一个脚本产出的中间文件。常见错误是直接跳到最后跑 Attention_train.py,结果报「找不到 vocab 文件」或「TFRecord 路径为空」。判断链路是否跑通有个简单标准——跑出 checkpoint 文件且训练 loss 在打印,就说明数据管道没问题,接下来才值得调模型参数。
3. 语料解析与词典构建:ParseXml、Pre-treatment、Text_vocab 的配合方式
语料质量决定了翻译质量的上限,模型只是在逼近这个上限。这个项目里语料从 XML 到词典要过三道关:先把结构化的 XML 变成平行句对,再做文本清洗,最后统计出词典。这三步每一步都直接影响后续训练效果,也是最容易「看着没问题、实际埋雷」的地方。
3.1 XML 语料怎么变成平行句对
古文语料常见组织形式是 XML,因为 XML 能保留篇章结构、出处和注释信息,比纯文本更适合做平行语料管理。项目里的 ParseXml.py 负责把这些标签结构拆成源文-目标文句对,核心逻辑通常是这样:
import xml.etree.ElementTree as ET def parse_corpus(xml_path): tree = ET.parse(xml_path) root = tree.getroot() pairs = [] # 常见古文平行语料里,一个句对长这样: # <pair id="1"><source>学而时习之</source><target>学习并且经常温习</target></pair> for pair in root.iter('pair'): src = pair.findtext('source', default='').strip() tgt = pair.findtext('target', default='').strip() if src and tgt: # 过滤空句对 pairs.append((src, tgt)) return pairs这段代码的逻辑很直接:遍历所有 pair 节点,用 findtext 取出 source 和 target 子节点的文本,strip 掉首尾空白,空的直接丢弃。参数上有个高频坑——如果 XML 带命名空间(很多公开数据集都有 xmlns),findtext 可能返回 None,需要在解析前处理命名空间,或者用root.find('pair', ns)传入命名空间字典。我一般会先用print(root.tag)看一下根标签长什么样,再决定解析方式,这能省掉大半天的排查时间。
3.2 清洗细节:全角字符、标签残留与长度截断
拿到平行句对后,下一步是清洗。Pre-treatment.py 一般会做这几件事:去掉 XML 解析残留的尖括号内容、统一全角半角字符、按需截断句子长度。古文场景里有个特殊点:古文原文没有现代标点,语料里的标点是后人加的,清洗时往往要直接去掉:
import re def clean_text(text, max_len=64, keep_punct=False): text = text.replace(',', ',').replace('。', '.') # 全角标点转半角 text = re.sub(r'<[^>]+>', '', text) # 去掉标签残留 if not keep_punct: text = re.sub(r'[^\w\u4e00-\u9fff]', '', text) # 只保留中文和字母数字 return text[:max_len]参数上,max_len 直接影响显存占用和训练速度:设 64 是常见值,如果语料以短句为主,32 就够,batch 也能开更大;keep_punct 在古文场景建议设 False,因为古文标点是后人标注,不是原文内容,保留反而会干扰模型对齐。这个脚本做完之后,建议顺手打印几条清洗结果看一眼——这一步能发现 80% 的编码问题,比如繁体没转、空格没去干净之类的。
3.3 vocab 怎么定:字级还是词级,UNK 留给谁
词典构建是整条数据链路里最像玄学的一步。古文场景和现代文不一样,现代文可以用 jieba 这类现成分词器,古文缺少高质量分词工具,「词」的边界本身就是语法问题。因此常见做法是切到字级,让模型自己学边界。Text_vocab.py 的核心逻辑通常是频次统计加截断:
from collections import Counter def build_vocab(sentences, min_count=2, vocab_size=30000): counter = Counter() for sent in sentences: counter.update(list(sent)) # 字级切分,不引入分词器 # 四个特殊 token 固定占前 4 位 vocab = ['<pad>', '<unk>', '<bos>', '<eos>'] + [ w for w, c in counter.most_common(vocab_size - 4) if c >= min_count ] word2id = {w: i for i, w in enumerate(vocab)} id2word = {i: w for w, i in word2id.items()} return word2id, id2word两个参数要特别留意。min_count 默认 2 表示出现次数少于 2 的字直接进 UNK;古文生僻字、人名地名很多,如果发现测试时 UNK 刷屏,把 min_count 降到 1,代价是 vocab 容量被生僻字占掉一些。vocab_size 控制词典上限,字级语料 30000 基本够用,如果你的语料里有大量异体字,适当拉大到 50000。字级模型的好处是 OOV 少,坏处是序列变长、训练变慢,但古文语料普遍不长,这个代价可以接受。
4. 向量化与 TFRecord:Input_vec、Senten_id_seq、TFRecord_operate 的序列化链路
清洗和词典构建完成后,数据还是「文本」形态,接下来要做的是把它变成模型能吃的数值。这一章讲的是三个脚本的配合:Input_vec 负责字符到 id 的映射,Senten_id_seq 负责把 id 序列对齐成 batch,TFRecord_operate 负责把 batch 落盘。很多人把这三步当成黑匣子直接用,其实每一步都藏着影响训练效果的细节。
4.1 向量化的两种做法:查表还是 one-hot
Input_vec.py 的作用是把句子变成模型输入。常见实现有两种:one-hot 向量和 id 查表。one-hot 的问题很直接——字级词典 30000 的话,每个 token 是一个 30000 维向量,数据量和内存都扛不住;所以实际项目基本都走查表路线,字符映射成整数 id,真正的高维向量交给模型里的 embedding 层去学:
import numpy as np def input_vec(sentences, word2id, max_len=64): # 句子转 id 数组,不足补 pad_id=0,超长截断 vec = np.zeros((len(sentences), max_len), dtype=np.int32) for i, sent in enumerate(sentences): ids = [word2id.get(ch, word2id['<unk>']) for ch in sent[:max_len]] vec[i, :len(ids)] = ids return vec注意这里用word2id.get(ch, word2id['<unk>'])而不是word2id[ch]:前者遇到未登录字会落到 UNK 的 id,后者直接抛 KeyError。线上环境里宁可让 UNK 兜底,也不能让整个训练崩掉。dtype 用 int32 而不是 int64 是为了省显存,embedding 层会把它当索引查表。
4.2 对齐与错位:decoder 输入和目标只差一位
Senten_id_seq.py 负责把成对的句子拼成训练 batch。seq2seq 训练时不是「输入整句、输出整句」,而是 teacher forcing:每个解码时间步输入前一个目标词,预测下一个词。所以对每个句对要构造出三份数据——encoder 输入、decoder 输入、decoder 目标,后两者错开一位:
def make_batch(pairs, word2id, batch_size=32): pad_id = word2id['<pad>'] bos_id = word2id['<bos>'] eos_id = word2id['<eos>'] batch_src, batch_tgt_in, batch_tgt_out = [], [], [] for src, tgt in pairs: src_ids = [word2id.get(c, word2id['<unk>']) for c in src] tgt_ids = [word2id.get(c, word2id['<unk>']) for c in tgt] # decoder 输入:句子开头放 <bos> tgt_in = [bos_id] + tgt_ids # decoder 目标:句子结尾放 <eos> tgt_out = tgt_ids + [eos_id] # 三个序列统一补到当前 batch 的最大长度 cur_len = max(len(src_ids), len(tgt_in)) batch_src.append(src_ids + [pad_id] * (cur_len - len(src_ids))) batch_tgt_in.append(tgt_in + [pad_id] * (cur_len - len(tgt_in))) batch_tgt_out.append(tgt_out + [pad_id] * (cur_len - len(tgt_out))) if len(batch_src) == batch_size: yield np.array(batch_src), np.array(batch_tgt_in), np.array(batch_tgt_out) batch_src, batch_tgt_in, batch_tgt_out = [], [], [] # 最后不足一个 batch 的样本也要 yield 出去 if batch_src: yield np.array(batch_src), np.array(batch_tgt_in), np.array(batch_tgt_out)这里的错位是理解重点:tgt_in 在开头插入 bos,tgt_out 在结尾追加 eos,两者长度一致但内容右移一位。模型看到「学」预测「学」的下一个字,看到「学而时习之」预测「学习并且经常温习」,本质就是在学条件概率。padding 时三个序列统一补到同一长度,这个 pad 位置在后面计算 loss 时必须掩掉,否则模型会拼命学习预测「pad」,这是第 5 章要展开的坑。batch_size 我一般从 32 起调,显存不够先降 batch 而不是砍 max_len。
4.3 写 TFRecord 的固定套路与读取参数
最后一步是把 batch 序列化到磁盘。TFRecord_operate.py 的核心逻辑是构造 Example 对象再写入:
import tensorflow as tf def write_tfrecord(path, src_ids_list, tgt_ids_list): # 注意:TF 1.x 用 tf.python_io.TFRecordWriter; # TF 2.x 换成 tf.io.TFRecordWriter,写法基本一致 with tf.python_io.TFRecordWriter(path) as writer: for s, t in zip(src_ids_list, tgt_ids_list): example = tf.train.Example(features=tf.train.Features(feature={ 'src': tf.train.Feature(int64_list=tf.train.Int64List(value=s)), 'tgt': tf.train.Feature(int64_list=tf.train.Int64List(value=t)), })) writer.write(example.SerializeToString())这里最容易翻车的点是 feature 名:写入端叫src/tgt,读取端解析时必须严格同名,大小写和拼写差一个字符都读不出来。读取时一般配合tf.data.TFRecordDataset,用 shuffle buffer 打乱数据顺序。shuffle buffer 的大小会影响随机性,我一般设为总样本数的五分之一左右,太小的话每个 epoch 的样本顺序变化不大,模型容易记住语料顺序而不是语言规律。
5. 避坑:四类实际翻车现场与排查过程
数据链路和训练代码都跑通之后,真正的噩梦才开始。这一章记录的是最常遇到的四类翻车现场,每一条都是真实排查过的,按「现象 → 原因 → 解决」写。建议先收藏,遇到对应报错再回来看。
5.1 训练期的两个大坑:loss 横盘与全场 UNK
坑 1:loss 一直横在 4.5 左右,几十个 epoch 不动。
现象:Attention_train.py 跑起来,前几个 epoch loss 快速降到 4.5 附近,然后就死水一潭,偶尔抖动但整体不降。
原因:常见原因有三个。第一,loss 没有对 padding 位置做 mask,模型花大量精力「学习」预测 pad token,真实位置的梯度被稀释;第二,学习率设得太大,attention 权重在震荡中始终收敛不了;第三,TFRecord 读取时没做 shuffle,每个 epoch 的样本顺序完全相同,模型学到了语料排序而不是语言规律。
解决:先在训练代码里找 loss 计算处,确认是否乘了 mask(mask 与 padding 位置对应,把 pad 位置的 loss 置 0);再把学习率降到 0.001 以下,必要时加 warmup;最后在 dataset 读取时加大 shuffle buffer。这三步做完,loss 通常会从 4.x 平滑降到 3.0 以下,才算进入正常训练状态。
坑 2:测试输出里全是 UNK,常见字能翻译,人名地名全军覆没。
现象:attention_test.py 输入「孔子登东山而小鲁」,输出变成「 登上 山而 」。
原因:Text_vocab.py 的 min_count 设得太高,生僻字出现次数不足被过滤成了 UNK;解码阶段又把 UNK 原样输出,没有做任何兜底。
解决:字典层把 min_count 从默认值降到 1~2,让生僻字保留在 vocab 里;解码层在输出端过滤 UNK——当 beam search 选中 UNK 时,用 attention 权重最高的源端字符回填。这个方法虽然朴素,但实测能明显降低 UNK 比例,尤其是人名地名密集的古文句子。
5.2 数据层和服务层的两个小坑:TF 版本迁移与端口占用
坑 3:import tensorflow 后直接报AttributeError: module 'tensorflow' has no attribute 'Session'。
现象:在自己电脑上复现时,一跑 Attention_train.py 就报错,提示 Session 或 placeholder 不存在。
原因:这是典型的 TF 1.x 代码跑在 TF 2.x 环境。源码里的 Session、placeholder、get_variable 都是 1.x 接口,2.x 默认 eager 模式里这些都没了。
解决:最省心的是配老环境,python 3.6/3.7 + tensorflow 1.15,几乎不用改代码;如果必须留在 TF 2.x,在入口文件加一行tf.compat.v1.disable_eager_execution(),并把tf.Session、tf.placeholder改成tf.compat.v1.Session、tf.compat.v1.placeholder。我一般建议毕设直接走老环境,改 compat 接口容易引发连锁报错,不值得为版本问题耽误进度。
坑 4:ServerSocket.py 第二次启动报OSError: [Errno 98] Address already in use。
现象:第一次运行 socket 服务正常,Ctrl+C 中断后再次运行,bind 直接报端口被占。
原因:上一次进程没完全退出,端口还被内核占用。socket 默认不会设置地址复用,短时间内重启必然冲突。
解决:先找到占用进程再决定杀不杀。Mac/Linux 用lsof -i:端口号拿到 PID,Windows 用netstat -ano | findstr 端口号,确认后kill -9 PID。自己改代码的话,在 bind 之前加一行s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1),这个问题就从根上解决了。
6. 验证与二次开发:把 attention 结果变成答辩能用的数据
模型训练完,最后一步是验证和展示。毕设答辩最常被问的一句话是「你这个翻译效果到底怎么评价」。如果只放几个翻译样例,说服力不够,至少要有一个量化指标。常见的做法是算 BLEU 近似值——统计译文和参考译文的 n-gram 重叠率:
def bleu_approx(pred, ref, n=2): # 简易 n-gram 重叠率,适合快速抽查,正式评估建议用 nltk/sacrebleu pred_tokens = list(pred) ref_tokens = list(ref) if len(pred_tokens) < n: return 0.0 pred_ngrams = set(zip(*[pred_tokens[i:] for i in range(n)])) ref_ngrams = set(zip(*[ref_tokens[i:] for i in range(n)])) overlap = len(pred_ngrams & ref_ngrams) return overlap / max(1, len(pred_ngrams))这个函数在测试集上跑一遍,输出一个 0~1 之间的分数,写进论文的对比表格里。如果想证明 attention 确实学到了对齐关系,还可以把测试句的 attention 权重矩阵保存下来画成热力图,选一两个「古文词 → 现代文词」对齐明显的例子贴在论文里,比大段文字描述直观得多。
跑通之后再往上做,有四个比较实际的改造方向。第一,换自己的语料:《论语》《战国策》这些公有领域古籍整理成 pair 结构的 XML,整个 pipeline 不用动,换文件路径就能重新训练;第二,改网络结构:在 neural_network.py 里把单向 RNN 换成双向 GRU,或者加深 attention 层数,这是毕设里最好讲的工作量;第三,把 socket 服务外面包一层 Flask,变成 HTTP 接口,前端页面一接就是完整的演示系统;第四,做数据增强:对现代文一侧做同义改写或简单回译,扩充训练集规模。
我自己的习惯是,接到任何源码先跑一遍 README,再按数据流把脚本挨个跑通,最后才动模型参数。这个顺序帮我避开了很多「改了半天不知道改的是哪一层」的尴尬。希望这篇拆解能帮到你。
本文还有配套的精品资源,点击获取