NLP实战必看！文本摘要模型开发与应用全流程，附可直接复用代码

python
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split

# 1. 读取数据集（以csv格式为例，包含"text"[原文]和"summary"[摘要]字段）
df = pd.read_csv("news_summary_dataset.csv")

# 2. 数据清洗：去除空值、过长/过短文本
df = df.dropna(subset=["text", "summary"]) # 删除空值
# 过滤文本长度（根据实际场景调整阈值）
df = df[(df["text"].str.len() > 100) & (df["text"].str.len() < 2000)] # 原文100-2000字
df = df[(df["summary"].str.len() > 20) & (df["summary"].str.len() < 300)] # 摘要20-300字
# 去除重复数据
df = df.drop_duplicates(subset=["text"])

# 3. 格式转换与数据集划分（按8:1:1划分训练/验证/测试集）
train_df, temp_df = train_test_split(df, test_size=0.2, random_state=42)
val_df, test_df = train_test_split(temp_df, test_size=0.5, random_state=42)

# 4. 保存划分后的数据集（用于后续训练）
train_df.to_csv("train_data.csv", index=False)
val_df.to_csv("val_data.csv", index=False)
test_df.to_csv("test_data.csv", index=False)

print(f"训练集数量：{len(train_df)}")
print(f"验证集数量：{len(val_df)}")
print(f"测试集数量：{len(test_df)}")
print("数据准备完成！")

python
from transformers import T5Tokenizer, T5ForConditionalGeneration

# 选择T5-small模型（新手首选，轻量化易训练）
model_name = "t5-small"
# 加载tokenizer和模型
tokenizer = T5Tokenizer.from_pretrained(model_name)
model = T5ForConditionalGeneration.from_pretrained(model_name)

# 查看模型基本信息
print(f"模型参数量：{sum(p.numel() for p in model.parameters()):,}") # T5-small约6000万参数，显存要求低

# 测试模型基础功能（生成摘要示例）
def test_base_model(text):
# T5模型需要在输入文本前添加任务前缀
input_text = f"summarize: {text}"
# 编码文本
inputs = tokenizer(input_text, max_length=512, truncation=True, return_tensors="pt")
# 生成摘要（max_length为摘要最大长度）
outputs = model.generate(
**inputs,
max_length=150,
min_length=20,
num_beams=4, # 束搜索，提升摘要质量
early_stopping=True
)
# 解码生成结果
summary = tokenizer.decode(outputs[0], skip_special_tokens=True)
return summary

# 测试示例
test_text = "自然语言处理（NLP）是人工智能领域的重要分支，致力于让计算机理解和生成人类语言。文本摘要作为NLP的核心应用之一，能够快速从海量文本中提炼核心信息，帮助用户节省阅读时间，提高信息获取效率。近年来，随着预训练语言模型的发展，文本摘要模型的效果得到了大幅提升，广泛应用于新闻、医疗、金融等多个行业。"
print("测试文本：", test_text)
print("基础模型生成摘要：", test_base_model(test_text))

python
import torch
from torch.utils.data import Dataset, DataLoader
from transformers import T5Tokenizer, T5ForConditionalGeneration, Trainer, TrainingArguments

# 1. 定义自定义数据集类
class SummaryDataset(Dataset):
def __init__(self, df, tokenizer, max_source_length=512, max_target_length=150):
self.df = df
self.tokenizer = tokenizer
self.max_source_length = max_source_length
self.max_target_length = max_target_length

def __len__(self):
return len(self.df)

def __getitem__(self, idx):
text = self.df.iloc[idx]["text"]
summary = self.df.iloc[idx]["summary"]

# 编码输入文本（添加T5任务前缀）
inputs = self.tokenizer(
f"summarize: {text}",
max_length=self.max_source_length,
truncation=True,
padding="max_length",
return_tensors="pt"
)

# 编码目标摘要
labels = self.tokenizer(
summary,
max_length=self.max_target_length,
truncation=True,
padding="max_length",
return_tensors="pt"
)

# 去除多余的维度
inputs = {k: v.squeeze(0) for k, v in inputs.items()}
labels = labels["input_ids"].squeeze(0)

# 将padding部分的label设为-100（Trainer会自动忽略）
labels[labels == self.tokenizer.pad_token_id] = -100

inputs["labels"] = labels
return inputs

# 2. 加载数据和模型
train_df = pd.read_csv("train_data.csv")
val_df = pd.read_csv("val_data.csv")
tokenizer = T5Tokenizer.from_pretrained("t5-small")
model = T5ForConditionalGeneration.from_pretrained("t5-small")

# 3. 创建数据集和数据加载器
train_dataset = SummaryDataset(train_df, tokenizer)
val_dataset = SummaryDataset(val_df, tokenizer)

# 4. 定义训练参数
training_args = TrainingArguments(
output_dir="./t5-summary-model", # 模型保存路径
per_device_train_batch_size=8, # 训练批次大小（根据显存调整，16G显存可设8-16）
per_device_eval_batch_size=8,
num_train_epochs=5, # 训练轮数
learning_rate=5e-5, # 学习率
weight_decay=0.01, # 权重衰减，防止过拟合
logging_dir="./logs", # 日志保存路径
logging_steps=100,
evaluation_strategy="epoch", # 每轮结束后评估
save_strategy="epoch", # 每轮结束后保存模型
load_best_model_at_end=True, # 训练结束后加载最优模型
metric_for_best_model="loss",
fp16=torch.cuda.is_available(), # 开启混合精度训练，加速训练
)

# 5. 初始化Trainer并开始训练
trainer = Trainer(
model=model,
args=training_args,
train_dataset=train_dataset,
eval_dataset=val_dataset,
)

# 开始训练
trainer.train()

# 保存最优模型
model.save_pretrained("./best-t5-summary-model")
tokenizer.save_pretrained("./best-t5-summary-model")
print("模型微调完成，最优模型已保存！")

python
from rouge import Rouge
from transformers import T5Tokenizer, T5ForConditionalGeneration

# 1. 加载最优模型和tokenizer
model = T5ForConditionalGeneration.from_pretrained("./best-t5-summary-model")
tokenizer = T5Tokenizer.from_pretrained("./best-t5-summary-model")
model.eval() # 切换到评估模式

# 2. 定义评估函数
def evaluate_model(test_df, model, tokenizer):
rouge = Rouge()
predictions = []
references = []

with torch.no_grad(): # 关闭梯度计算，加速评估
for idx, row in test_df.iterrows():
text = row["text"]
reference = row["summary"]

# 生成摘要
input_text = f"summarize: {text}"
inputs = tokenizer(input_text, max_length=512, truncation=True, return_tensors="pt")
outputs = model.generate(
**inputs,
max_length=150,
min_length=20,
num_beams=4,
early_stopping=True
)
prediction = tokenizer.decode(outputs[0], skip_special_tokens=True)

predictions.append(prediction)
references.append(reference)

# 计算ROUGE指标
scores = rouge.get_scores(predictions, references, avg=True)
print("ROUGE指标评估结果：")
print(f"ROUGE-1: {scores['rouge-1']['f']:.4f}")
print(f"ROUGE-2: {scores['rouge-2']['f']:.4f}")
print(f"ROUGE-L: {scores['rouge-L']['f']:.4f}")

return predictions, references, scores

# 加载测试集并评估
test_df = pd.read_csv("test_data.csv")
predictions, references, scores = evaluate_model(test_df, model, tokenizer)

# 优化生成参数（提升摘要质量）
def optimize_summary(text, model, tokenizer):
input_text = f"summarize: {text}"
inputs = tokenizer(input_text, max_length=512, truncation=True, return_tensors="pt")
outputs = model.generate(
**inputs,
max_length=150,
min_length=20,
num_beams=6, # 增加束搜索数量
temperature=0.8, # 调整温度，控制生成多样性
top_p=0.9, # nucleus sampling，提升生成质量
early_stopping=True
)
optimized_summary = tokenizer.decode(outputs[0], skip_special_tokens=True)
return optimized_summary

# 测试优化效果
test_text = test_df.iloc[0]["text"]
original_summary = test_df.iloc[0]["summary"]
optimized_summary = optimize_summary(test_text, model, tokenizer)
print("\n优化前生成摘要：", test_base_model(test_text))
print("优化后生成摘要：", optimized_summary)
print("真实摘要：", original_summary)

python
from flask import Flask, request, jsonify
from transformers import T5Tokenizer, T5ForConditionalGeneration
import torch

# 1. 初始化Flask应用
app = Flask(__name__)

# 2. 加载模型和tokenizer（全局加载，避免重复加载）
model = T5ForConditionalGeneration.from_pretrained("./best-t5-summary-model")
tokenizer = T5Tokenizer.from_pretrained("./best-t5-summary-model")
model.eval()
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model.to(device) # 移到GPU（如有），加速推理

# 3. 定义摘要生成接口
@app.route("/generate_summary", methods=["POST"])
def generate_summary():
try:
# 获取请求参数
data = request.get_json()
text = data.get("text")
max_length = data.get("max_length", 150)
min_length = data.get("min_length", 20)

# 参数校验
if not text:
return jsonify({"code": 400, "msg": "请传入需要摘要的文本", "data": None})

# 生成摘要
input_text = f"summarize: {text}"
inputs = tokenizer(
input_text,
max_length=512,
truncation=True,
return_tensors="pt"
).to(device)

with torch.no_grad():
outputs = model.generate(
**inputs,
max_length=max_length,
min_length=min_length,
num_beams=6,
temperature=0.8,
top_p=0.9,
early_stopping=True
)

summary = tokenizer.decode(outputs[0], skip_special_tokens=True)

# 返回结果
return jsonify({
"code": 200,
"msg": "摘要生成成功",
"data": {"summary": summary}
})

except Exception as e:
return jsonify({
"code": 500,
"msg": f"服务器错误：{str(e)}",
"data": None
})

# 4. 启动服务
if __name__ == "__main__":
print("文本摘要API服务启动中...")
app.run(host="0.0.0.0", port=5000, debug=False) # 生产环境关闭debug
print("服务启动成功，访问 http://localhost:5000/generate_summary 调用接口")