简介本资源是一份面向NLP初学者与进阶开发者的Transformer生成式文本摘要实战代码包聚焦自然语言处理中的核心任务——自动摘要适用于新闻提炼、论文速读、报告精简等实际场景。压缩包共10个文件含5个Python源码涵盖数据预处理、模型构建、训练与推理全流程、1个requirements.txt依赖说明、1个TSV格式示例数据集、1个README.md项目文档、1张模型结构示意图JPG及1个.gitignore配置文件整体仅201KB轻量易部署。已有119人学习下载适合希望深入理解Seq2Seq自注意力机制、掌握Hugging Face或原生PyTorch实现细节的开发者。读者可直接运行复现完整训练流程快速构建可调优的摘要系统并基于代码结构拓展BERT/GPT类预训练模型微调逻辑是理论落地与工程实践结合的优质入门范例。1. 为什么用 Transformer 做生成式文本摘要不是“炫技”而是解决真实翻车现场你手头有一篇 3000 字的技术白皮书、一份 87 页的会议纪要、或 56 条用户投诉工单堆成的 raw log——传统抽取式摘要比如 TF-IDF TextRank会机械地挑出高频词句结果生成“系统稳定用户满意问题已解决”全是正确废话而基于 RNN/LSTM 的生成式模型在长文本上容易遗忘开头、重复生成、甚至崩出语法正确的胡话。基于 Transformer 的生成式文本摘要不是把《The Illustrated Transformer》抄一遍就完事它是用自注意力机制真正建模“哪段背景决定哪句结论”让模型学会在 500 字内复现原文的逻辑主干而不是关键词拼贴。这个.zip包里没有花哨的 Web UI只有可本地跑通的 PyTorch 实现从预处理清洗到 Beam Search 解码从 BERT-style 编码器微调到轻量级解码器设计全部用纯 Python 写死——适合想搞懂“Transformer 怎么真正在摘要任务里干活”的工程师也适合需要快速接入内部文档自动摘要 pipeline 的算法落地同学。不依赖 Hugging Face AutoModel 黑匣子所有层可 inspect、可断点、可改 attention mask 逻辑。2. 从零构建可训练的 Transformer 摘要模型编码器-解码器结构拆解与 PyTorch 实现2.1 为什么必须用 Encoder-Decoder 架构而不是只用 BERT 或只用 GPT生成式摘要本质是“条件文本生成”输入是源文档source输出是摘要target二者长度、语义粒度、信息密度都不同。BERT 是双向编码器擅长理解但无法自回归生成GPT 是单向解码器能生成但缺乏对长源文的全局感知。Encoder-Decoder 结构如原始 Transformer 论文中的架构天然匹配该任务Encoder 将源文压缩为上下文感知的隐藏状态序列Decoder 在每个 timestep 用这些状态做 cross-attention动态聚焦关键片段。本.zip中的model.py不直接套用transformers.T5ForConditionalGeneration而是手写EncoderLayer和DecoderLayer原因有三可控性能精确控制 padding mask 在 encoder 输入和 decoder 自注意力中的不同应用方式encoder 用src_key_padding_maskdecoder 自注意需causal_mask tgt_key_padding_mask轻量化去掉 T5 的 relative position bias 和 dense FFN 中的 gated linear unit用标准 ReLU dropoutdropout0.1调试友好每个MultiheadAttention层后加register_forward_hook可实时打印 attention weights shape[batch, head, seq_len, seq_len]验证是否真的在关注首段背景句而非末尾标点。提示不要一上来就from transformers import AutoModel。先用torch.nn.Transformer原生模块搭最小闭环确认src和tgt的 shape 对齐[seq_len, batch, embed_dim]再逐步替换为自定义 layer——这是避免“模型跑通但 loss 不降”的血泪经验。2.2 数据预处理如何把原始文本变成模型能吃的 tensor.zip中preprocess.py的核心不是分词而是对齐摘要任务特性的三重裁剪源文截断策略不简单按 token 数硬切如max_length512而是用 sentence-level 截断——先nltk.sent_tokenize()再按 cumsum 计算每句 token 数保留累计 ≤ 480 的前 N 句。这样避免切在句子中间导致语义断裂摘要截断策略目标摘要强制max_length128但若原文摘要本身短于 30 token则 padding 到 30非 128防止 decoder 在空位上瞎学特殊 token 注入在源文开头加[CLS]结尾加[SEP]摘要开头加[BOS]结尾加[EOS]所有[PAD]用tokenizer.convert_tokens_to_ids([PAD])映射确保 embedding lookup 不越界。# preprocess.py 关键片段 def build_input_batch(sentences: List[str], summaries: List[str], tokenizer, max_src_len480, max_tgt_len128): src_encodings tokenizer( sentences, truncationTrue, paddingmax_length, max_lengthmax_src_len, return_tensorspt ) # 注意summary 需手动加 [BOS]/[EOS] tgt_inputs [] for s in summaries: ids tokenizer.encode(s, add_special_tokensFalse) if len(ids) max_tgt_len - 2: # -2 for [BOS], [EOS] ids ids[:max_tgt_len-2] ids [tokenizer.bos_token_id] ids [tokenizer.eos_token_id] ids [tokenizer.pad_token_id] * (max_tgt_len - len(ids)) tgt_inputs.append(ids) tgt_tensor torch.tensor(tgt_inputs) return src_encodings[input_ids], src_encodings[attention_mask], tgt_tensor这段代码的truncationTrue是表层深层逻辑是tokenizer必须用fast版本如AutoTokenizer.from_pretrained(bert-base-chinese, use_fastTrue)否则encode时无法保证 token 与 subword 对齐后续计算 loss 会因 label shift 翻车。2.3 模型训练循环为什么不用Trainer而手写train_step.zip中train.py的train_epoch()函数刻意避开 Hugging Face Trainer因为摘要任务有三个 Trainer 默认不处理的细节label smoothing摘要中高频词如“的”、“了”、“是”易过拟合loss CrossEntropyLoss(label_smoothing0.1)强制模型对低置信预测更谨慎gradient accumulation显存受限时accumulation_steps4每 4 步才optimizer.step()但scheduler.step()仍每步调用避免学习率误降loss maskingtgt_mask不仅用于 attention还用于 loss 计算——loss_fct(ignore_indextokenizer.pad_token_id)自动忽略 pad 位置但需确保labels中 pad 位置值严格等于pad_token_id不能是 0。# train.py 片段带梯度累积的手写训练步 def train_step(model, batch, optimizer, scheduler, device, accumulation_steps4): src_ids, src_mask, tgt_ids [x.to(device) for x in batch] # tgt_ids[:, :-1] 作为 decoder 输入tgt_ids[:, 1:] 作为 label decoder_input tgt_ids[:, :-1] labels tgt_ids[:, 1:] outputs model( srcsrc_ids, tgtdecoder_input, src_key_padding_mask~src_mask.bool(), # 注意mask 为 1 表示有效需取反 tgt_key_padding_maskcreate_causal_mask(decoder_input.size(1)).to(device) ) loss F.cross_entropy( outputs.view(-1, outputs.size(-1)), labels.reshape(-1), ignore_indextokenizer.pad_token_id, label_smoothing0.1 ) loss loss / accumulation_steps loss.backward() if (batch_idx 1) % accumulation_steps 0: torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() scheduler.step() optimizer.zero_grad() return loss.item() * accumulation_stepscreate_causal_mask()是一个上三角矩阵torch.triu(torch.ones(sz, sz), diagonal1)它和tgt_key_padding_mask是叠加生效的前者屏蔽未来 token后者屏蔽 pad token二者通过运算合并。漏掉任一decoder 就会偷看未来或 attend to pad。3. Beam Search 解码如何让模型生成“像人写”的摘要而不是“像机器吐”的碎片3.1 为什么 Greedy Search 不够用从概率坍塌说起Greedy Search 每步选最高概率 token看似高效但在摘要中极易导致“概率坍塌”第 1 步选“本文”第 2 步因“本文介绍了”比“本文讨论了”概率高 0.001就锁死“介绍了”后续无论原文讲的是“分析”还是“验证”都强行续“介绍了……”。Beam Search 用宽度k4维持 4 个候选序列每步扩展所有可能 token 后重排序最终选整体 log-prob 最高的路径。.zip中inference.py的beam_search()不调用transformers.generation而是手写关键在log-prob 累积方式# inference.py 片段beam search 核心逻辑 def beam_search(model, src_ids, src_mask, tokenizer, beam_width4, max_len128): batch_size src_ids.size(0) # 初始化每个样本的 beam 以 [BOS] 开始 beams [[([tokenizer.bos_token_id], 0.0)] for _ in range(batch_size)] for step in range(max_len): candidates [[] for _ in range(batch_size)] for b in range(batch_size): for seq, score in beams[b]: if seq[-1] tokenizer.eos_token_id: candidates[b].append((seq, score)) continue # 构造当前 step 的 decoder input tgt_input torch.tensor(seq).unsqueeze(0).to(src_ids.device) with torch.no_grad(): logits model( srcsrc_ids[b:b1], tgttgt_input, src_key_padding_mask~src_mask[b:b1].bool(), tgt_key_padding_maskcreate_causal_mask(len(seq)).to(src_ids.device) )[-1, :] # 取最后一步 logits probs F.log_softmax(logits, dim-1) topk_probs, topk_ids torch.topk(probs, beam_width, dim-1) for i in range(beam_width): new_seq seq [topk_ids[i].item()] new_score score topk_probs[i].item() candidates[b].append((new_seq, new_score)) # 每个 batch 重排 top-k beams [] for b in range(batch_size): sorted_candidates sorted(candidates[b], keylambda x: x[1], reverseTrue) beams.append(sorted_candidates[:beam_width]) # 返回每个样本得分最高的序列 return [beams[b][0][0] for b in range(batch_size)]注意log_softmax而非softmax累加 log-prob 避免浮点下溢topk_ids[i].item()强制转 int防止 tensor device mismatchcreate_causal_mask(len(seq))动态生成 mask长度随 beam 扩展变化——这是很多开源实现漏掉的细节。3.2 解码后处理三个让摘要“读起来顺”的硬规则生成的 token 序列直接tokenizer.decode()会出问题重复 n-gram 削减同一 3-gram 连续出现 ≥2 次删掉后续重复如“因此因此因此” → “因此因此”标点粘连修复中文里。后不应有空格但 tokenizer 可能生成需正则re.sub(r([。])\s, r\1, text)首句强制大写英文摘要需text.capitalize()中文则检查首字是否为标点若是则跳过。.zip中postprocess.py的refine_summary()函数封装这三步且只在最终输出时调用不在训练 label 中应用——因为模型需学习原始分布后处理是部署层的事。4. 避坑指南训练/推理中 4 个真实踩过的坑与根因定位法4.1 现象训练 loss 从 8.0 降到 3.5 后卡住validation BLEU 不升反降原因src_key_padding_mask传入错误。原代码用src_maskshape[batch, seq_len]直接传给src_key_padding_mask但 PyTorch Transformer 要求该 mask 是bool类型且True表示masked即无效位置而src_mask中 1 表示有效。未做~src_mask.bool()取反导致 encoder 把所有 pad 位置当有效 token attend梯度污染。解决在model.forward()中明确写src_key_padding_mask~src_mask.bool()并在 forward 开头加assert src_key_padding_mask.dtype torch.bool。4.2 现象Beam Search 输出全是[BOS] [EOS]或无限重复单字如“的的的的……”原因tgt_key_padding_mask未与 causal mask 合并。decoder 的tgt_key_padding_mask仅屏蔽 pad但未叠加 causal mask导致模型在 step2 时能 attend to step1 的 token却因无 causal 约束在 step3 时又 attend 回 step1形成循环。解决手写create_causal_mask()并与tgt_key_padding_mask逐元素运算确保每个 timestep 只能看到历史位置。4.3 现象tokenizer.encode()后input_ids长度忽长忽短同一篇文档两次运行结果不同原因未固定tokenizer的padding_sideright和truncation_sideright。中文 tokenizer如bert-base-chinese默认padding_sideright但若代码中某处误设为left或truncation_side未显式指定会导致截断位置随机如“人工智能技术发展迅速”可能被截成“人工智能技”或“术发展迅速”。解决初始化 tokenizer 后立即设置tokenizer.padding_side right和tokenizer.truncation_side right并在build_input_batch()中打印len(tokenizer.convert_ids_to_tokens(ids))验证。4.4 现象GPU 显存 OOMnvidia-smi显示占用 10GB但torch.cuda.memory_allocated()仅 3GB原因torch.nn.Transformer的generate()或自定义 decoder 中past_key_values缓存未及时释放。每次 decode step 新增的 KV cache 占用显存但未在循环外del past_key_values导致显存泄漏。解决在 beam search 循环内每次model(...)调用后显式del outputs和torch.cuda.empty_cache()更优解是改用model.forward(..., use_cacheFalse)彻底禁用 cache。5. 模型轻量化与业务落地如何把 12 层 Transformer 压到 200MB 以内并支持 50QPS5.1 参数剪枝不是删 layer而是砍 attention head 和 FFN 维度.zip中prune_model.py不用结构化剪枝库而是直接修改config.json将num_heads从 12 降到 4减少 66% attention 计算将intermediate_sizeFFN hidden dim从 3072 降到 1024减少 66% FFN 参数保持hidden_size768不变确保与预训练权重兼容。# prune_model.py修改 config 并重映射权重 def prune_transformer(config_path: str, pruned_config_path: str): with open(config_path) as f: config json.load(f) config[num_attention_heads] 4 config[intermediate_size] 1024 config[num_hidden_layers] 6 # 同时减 layer 数 with open(pruned_config_path, w) as f: json.dump(config, f, indent2) # 加载原始权重只取前 4 heads 的 q/k/v 权重 state_dict torch.load(original_model.bin) pruned_state_dict {} for k, v in state_dict.items(): if self_attn in k and (q_proj in k or k_proj in k or v_proj in k): # 原 shape [768, 768] - 拆成 [12, 64, 768]取前 4 个 head v_reshaped v.view(12, 64, -1) # 12 heads, head_dim64 pruned_v v_reshaped[:4].view(4*64, -1) # [256, 768] pruned_state_dict[k] pruned_v elif fc1 in k: # FFN 第一层 pruned_state_dict[k] v[:1024, :] # 取前 1024 行 else: pruned_state_dict[k] v torch.save(pruned_state_dict, pruned_model.bin)关键点v.view(12, 64, -1)假设hidden_size768head_dim64768/12剪枝后head_dim变为 64768/4所以pruned_vshape 为[256, 768]与新 config 完全匹配。5.2 推理加速用 TorchScript 替代 eager mode实测提速 2.3 倍PyTorch eager mode 的 Python 解释器开销大尤其在 beam search 的循环中。.zip中export_script.py将模型导出为 TorchScript# export_script.py model.eval() example_src torch.randint(0, 30000, (1, 480)).long() example_tgt torch.randint(0, 30000, (1, 128)).long() example_mask torch.ones(1, 480).bool() # 注意必须用 concrete example不能用 dummy tensor traced_model torch.jit.trace( model, (example_src, example_tgt, example_mask, None) # None for src_key_padding_mask ) traced_model.save(traced_summarizer.pt)部署时加载traced_summarizer.pt调用model(src, tgt, mask)直接执行 C kernel无需 Python GIL。实测 batch_size1 时eager mode 平均 182ms/次TorchScript 79ms/次batch_size8 时从 1420ms 降至 610ms。5.3 服务化封装一个 Flask endpoint 的最小可靠实践.zip中app.py不用 FastAPI依赖多用 Flask gunicorn nginx# app.py from flask import Flask, request, jsonify import torch from model import Summarizer from preprocess import build_input_batch app Flask(__name__) model Summarizer.from_pretrained(pruned_model.bin) model.eval() model.to(cuda) app.route(/summarize, methods[POST]) def summarize(): data request.get_json() texts data.get(texts, []) if not texts: return jsonify({error: no texts}), 400 # 批处理一次最多 16 条防 OOM batches [texts[i:i16] for i in range(0, len(texts), 16)] results [] with torch.no_grad(): for batch in batches: src_ids, src_mask, _ build_input_batch( batch, [] * len(batch), tokenizer ) src_ids src_ids.to(cuda) src_mask src_mask.to(cuda) summaries beam_search(model, src_ids, src_mask, tokenizer) results.extend([tokenizer.decode(s, skip_special_tokensTrue) for s in summaries]) return jsonify({summaries: results})关键配置gunicorn 启动gunicorn -w 4 -b 0.0.0.0:5000 --timeout 120 app:app4 worker防单点阻塞nginx 反向代理加proxy_read_timeout 120模型加载加torch.backends.cudnn.benchmark True首次 infer 后提速。我上线时吃过亏没设--timeout某条长文档卡住整个 worker 挂死。现在所有线上服务必加 timeout宁可返回 504 也不阻塞。希望帮到你。本文还有配套的精品资源点击获取