Transformerを使ったテキスト要約の実装ガイド¶

from datasets import load_dataset

# CNN/DailyMailデータセットを読み込み
datasets = load_dataset("cnn_dailymail", "3.0.0")

datasets

# データの構造を確認
datasets["train"][1]

from transformers import T5Tokenizer

# T5モデルのトークナイザーを読み込み
model_checkpoint = "t5-small"
tokenizer = T5Tokenizer.from_pretrained(model_checkpoint)

# トークナイザーのテスト
inputs = tokenizer("I loved reading the Hunger Games!")
inputs

# トークンに変換して内容を確認
tokenizer.convert_ids_to_tokens(inputs.input_ids)

# 前処理関数の定義
def preprocess_function(examples):
    # T5にはタスクプレフィックスが必要
    inputs = ["summarize: " + doc for doc in examples["article"]]
    targets = examples["highlights"]

    # 入力とターゲットをトークン化
    model_inputs = tokenizer(
        inputs, 
        max_length=512,  # 最大入力長
        truncation=True, 
        padding=False
    )

    # ターゲットテキストをトークン化
    with tokenizer.as_target_tokenizer():
        labels = tokenizer(
            targets, 
            max_length=128,  # 要約は通常短い
            truncation=True, 
            padding=False
        )

    model_inputs["labels"] = labels["input_ids"]
    return model_inputs

# 前処理を適用
tokenized_dataset = datasets.map(preprocess_function, batched=True)

from transformers import T5ForConditionalGeneration, T5Tokenizer
import torch

# デバイスの設定とモデルの読み込み
device = "mps" if torch.mps.is_available() else "cpu"
model_name = "t5-small"
tokenizer = T5Tokenizer.from_pretrained(model_name)
model = T5ForConditionalGeneration.from_pretrained(model_name).to(device)

from transformers import Seq2SeqTrainingArguments

# 学習引数の設定
training_args = Seq2SeqTrainingArguments(
    output_dir="./t5-small-cnn-summarization",
    eval_strategy="steps",
    eval_steps=1000,
    save_steps=1000,
    logging_steps=500,
    per_device_train_batch_size=8,
    per_device_eval_batch_size=8,
    num_train_epochs=3,
    learning_rate=3e-4,
    warmup_steps=1000,
    weight_decay=0.01,
    fp16=False,
    save_total_limit=3,
    load_best_model_at_end=True,
    metric_for_best_model="rouge1",
    greater_is_better=True,

    # Seq2Seq固有のパラメータ
    predict_with_generate=True,  # 評価にgenerate()を使用
    generation_max_length=128,   # 最大生成長
    generation_num_beams=4,      # より良い品質のためのビームサーチ

    dataloader_num_workers=4,
    report_to=None,
)

from rouge_score import rouge_scorer
import numpy as np

# 評価関数の定義
def compute_metrics(eval_pred):
    predictions, labels = eval_pred

    # 予測をデコード
    decoded_preds = tokenizer.batch_decode(predictions, skip_special_tokens=True)

    # ラベルの-100を置換してデコード
    labels = np.where(labels != -100, labels, tokenizer.pad_token_id)
    decoded_labels = tokenizer.batch_decode(labels, skip_special_tokens=True)

    # ROUGEスコアを計算
    scorer = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)

    rouge_scores = []
    for pred, ref in zip(decoded_preds, decoded_labels):
        scores = scorer.score(ref, pred)
        rouge_scores.append(scores)

    # 平均スコアを計算
    result = {}
    for key in ['rouge1', 'rouge2', 'rougeL']:
        scores = [score[key].fmeasure for score in rouge_scores]
        result[key] = np.mean(scores) * 100

    return {k: round(v, 2) for k, v in result.items()}

from transformers import DataCollatorForSeq2Seq

# データ照合器の設定
data_collator = DataCollatorForSeq2Seq(tokenizer, model=model, padding=True)

from transformers import Seq2SeqTrainer

# Trainerの代わりにSeq2SeqTrainerを使用
trainer = Seq2SeqTrainer(  # 単なるTrainerではない
    model=model,
    args=training_args,
    train_dataset=tokenized_dataset["train"],
    eval_dataset=tokenized_dataset["validation"],
    processing_class=tokenizer,
    data_collator=data_collator,
    compute_metrics=compute_metrics,
)

# 学習開始（コメントアウト済み - デモンストレーション用）
# trainer.train()

# データセットの形式をPyTorchテンソルに設定
tokenized_dataset.set_format("torch")

# モデルを再読み込み
model = T5ForConditionalGeneration.from_pretrained(model_name).to(device)

from torch.utils.data import DataLoader

# 前処理を適用（列を除去して新しいものに）
tokenized_dataset = datasets.map(preprocess_function, batched=True, remove_columns=datasets["train"].column_names)

# データ照合器
data_collator = DataCollatorForSeq2Seq(tokenizer=tokenizer, model=model)

# データローダーを作成
train_dataloader = DataLoader(
    tokenized_dataset["train"],
    batch_size=8,
    shuffle=True,
    collate_fn=data_collator,
)

eval_dataloader = DataLoader(
    tokenized_dataset["validation"],
    batch_size=8,
    shuffle=False,
    collate_fn=data_collator,
)

from transformers import get_linear_schedule_with_warmup
from torch.optim import AdamW

# オプティマイザーとスケジューラー
optimizer = AdamW(model.parameters(), lr=3e-4, weight_decay=0.01)

num_epochs = 3
num_training_steps = num_epochs * len(train_dataloader)
num_warmup_steps = min(1000, num_training_steps // 10)

lr_scheduler = get_linear_schedule_with_warmup(
    optimizer=optimizer,
    num_warmup_steps=num_warmup_steps,
    num_training_steps=num_training_steps
)

from accelerate import Accelerator

# acceleratorを初期化
accelerator = Accelerator()

# acceleratorですべてを準備
model, optimizer, train_dataloader, eval_dataloader, lr_scheduler = accelerator.prepare(
    model, optimizer, train_dataloader, eval_dataloader, lr_scheduler
)

print(f"デバイス: {accelerator.device}")
print(f"学習ステップ数: {num_training_steps}")
print(f"ウォームアップステップ数: {num_warmup_steps}")

from tqdm.auto import tqdm
import os

# 評価関数の定義
def evaluate_model():
    model.eval()
    rouge_scorer_obj = rouge_scorer.RougeScorer(['rouge1', 'rouge2', 'rougeL'], use_stemmer=True)

    all_predictions = []
    all_references = []
    eval_loss = 0

    for batch in tqdm(eval_dataloader, desc="評価中"):
        with torch.no_grad():
            # 損失を計算
            outputs = model(**batch)
            eval_loss += outputs.loss.item()

            # 予測を生成
            generated_tokens = accelerator.unwrap_model(model).generate(
                input_ids=batch["input_ids"],
                attention_mask=batch["attention_mask"],
                max_length=128,
                num_beams=4,
                early_stopping=True
            )

            # 予測と参照をデコード
            predictions = tokenizer.batch_decode(generated_tokens, skip_special_tokens=True)
            labels = batch["labels"]
            labels = torch.where(labels != -100, labels, tokenizer.pad_token_id)
            references = tokenizer.batch_decode(labels, skip_special_tokens=True)

            all_predictions.extend(predictions)
            all_references.extend(references)

    # ROUGEスコアを計算
    rouge_scores = []
    for pred, ref in zip(all_predictions, all_references):
        scores = rouge_scorer_obj.score(ref, pred)
        rouge_scores.append(scores)

    # 平均スコア
    rouge_results = {}
    for key in ['rouge1', 'rouge2', 'rougeL']:
        scores = [score[key].fmeasure for score in rouge_scores]
        rouge_results[key] = np.mean(scores) * 100

    avg_eval_loss = eval_loss / len(eval_dataloader)

    return avg_eval_loss, rouge_results

# 学習ループ
model.train()
progress_bar = tqdm(range(num_training_steps), desc="学習中")

step = 0
best_rouge1 = 0
save_dir = "./t5-small-cnn-accelerate"
os.makedirs(save_dir, exist_ok=True)

for epoch in range(num_epochs):
    print(f"\nエポック {epoch + 1}/{num_epochs}")

    for batch in train_dataloader:
        # 順伝播
        outputs = model(**batch)
        loss = outputs.loss

        # 逆伝播
        accelerator.backward(loss)

        optimizer.step()
        lr_scheduler.step()
        optimizer.zero_grad()

        progress_bar.update(1)
        step += 1

        # ログ出力
        if step % 500 == 0:
            print(f"ステップ {step}: 損失 = {loss.item():.4f}, 学習率 = {lr_scheduler.get_last_lr()[0]:.2e}")

        # 評価
        if step % 1000 == 0:
            print(f"\nステップ {step}で評価中...")
            eval_loss, rouge_results = evaluate_model()

            print(f"評価損失: {eval_loss:.4f}")
            print(f"ROUGE-1: {rouge_results['rouge1']:.2f}")
            print(f"ROUGE-2: {rouge_results['rouge2']:.2f}")
            print(f"ROUGE-L: {rouge_results['rougeL']:.2f}")

            # 最良モデルを保存
            if rouge_results['rouge1'] > best_rouge1:
                best_rouge1 = rouge_results['rouge1']
                accelerator.wait_for_everyone()
                unwrapped_model = accelerator.unwrap_model(model)
                unwrapped_model.save_pretrained(
                    save_dir, 
                    is_main_process=accelerator.is_main_process,
                    save_function=accelerator.save
                )
                if accelerator.is_main_process:
                    tokenizer.save_pretrained(save_dir)
                print(f"ROUGE-1: {best_rouge1:.2f}で最良モデルを保存しました")

            model.train()

print("\n学習完了!")
print(f"最良ROUGE-1スコア: {best_rouge1:.2f}")

# 最終評価
print("\n最終評価...")
eval_loss, rouge_results = evaluate_model()
print(f"最終結果:")
print(f"ROUGE-1: {rouge_results['rouge1']:.2f}")
print(f"ROUGE-2: {rouge_results['rouge2']:.2f}")  
print(f"ROUGE-L: {rouge_results['rougeL']:.2f}")

# 学習済みモデルを読み込み
from transformers import T5ForConditionalGeneration, T5Tokenizer

model_path = "./t5-small-cnn-accelerate"
tokenizer = T5Tokenizer.from_pretrained(model_path)
model = T5ForConditionalGeneration.from_pretrained(model_path)

def generate_summary(text, max_length=128):
    """
    テキストの要約を生成する関数

    Args:
        text (str): 要約したいテキスト
        max_length (int): 要約の最大長

    Returns:
        str: 生成された要約
    """
    # T5用のプレフィックスを追加
    input_text = f"summarize: {text}"

    # トークン化
    inputs = tokenizer(
        input_text, 
        return_tensors="pt", 
        max_length=512, 
        truncation=True
    )

    # 要約を生成
    with torch.no_grad():
        summary_ids = model.generate(
            inputs["input_ids"],
            attention_mask=inputs["attention_mask"],
            max_length=max_length,
            num_beams=4,
            early_stopping=True,
            no_repeat_ngram_size=2
        )

    # デコードして返す
    summary = tokenizer.decode(summary_ids[0], skip_special_tokens=True)
    return summary

# 使用例
sample_text = """
人工知能（AI）の発展は、私たちの生活に大きな変化をもたらしています。
機械学習、特に深層学習の技術により、画像認識、音声認識、自然言語処理
などの分野で画期的な進歩が見られています。医療分野では、AIが診断支援や
薬物発見に活用され、教育分野では個別学習システムが開発されています。
しかし、AIの普及に伴い、雇用への影響やプライバシーの問題など、
社会的な課題も浮上しています。これらの課題に対処しながら、
AIの恩恵を最大化することが重要です。
"""

summary = generate_summary(sample_text)
print("生成された要約:")
print(summary)

def advanced_generate_summary(text, num_beams=4, temperature=0.7):
    """
    高度な生成パラメータを使用した要約生成
    """
    input_text = f"summarize: {text}"
    inputs = tokenizer(input_text, return_tensors="pt", max_length=512, truncation=True)

    summary_ids = model.generate(
        inputs["input_ids"],
        attention_mask=inputs["attention_mask"],
        max_length=128,
        min_length=20,
        num_beams=num_beams,
        temperature=temperature,
        do_sample=True,
        early_stopping=True,
        no_repeat_ngram_size=3,
        repetition_penalty=1.2
    )

    return tokenizer.decode(summary_ids[0], skip_special_tokens=True)

def summarize_long_document(text, chunk_size=400, overlap=50):
    """
    長い文書を分割して要約する関数

    Args:
        text (str): 長い文書
        chunk_size (int): チャンクサイズ（単語数）
        overlap (int): チャンク間のオーバーラップ（単語数）
    """
    words = text.split()
    chunks = []

    # 文書を重複ありでチャンクに分割
    for i in range(0, len(words), chunk_size - overlap):
        chunk = ' '.join(words[i:i + chunk_size])
        chunks.append(chunk)

    # 各チャンクを要約
    chunk_summaries = []
    for chunk in chunks:
        summary = generate_summary(chunk, max_length=64)
        chunk_summaries.append(summary)

    # チャンク要約を結合して最終要約を生成
    combined_summary = ' '.join(chunk_summaries)
    final_summary = generate_summary(combined_summary, max_length=128)

    return final_summary

import requests
from datetime import datetime

class NewsSummarizationSystem:
    """ニュース記事自動要約システム"""

    def __init__(self, model_path):
        self.tokenizer = T5Tokenizer.from_pretrained(model_path)
        self.model = T5ForConditionalGeneration.from_pretrained(model_path)

    def summarize_article(self, article_text, target_length=100):
        """記事を指定された長さで要約"""
        return generate_summary(article_text, max_length=target_length)

    def batch_summarize(self, articles):
        """複数記事の一括要約"""
        summaries = []
        for article in articles:
            try:
                summary = self.summarize_article(article)
                summaries.append(summary)
            except Exception as e:
                summaries.append(f"要約エラー: {str(e)}")
        return summaries

    def export_summaries(self, articles, summaries, filename):
        """要約結果をファイルに出力"""
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(f"要約レポート - {datetime.now().strftime('%Y-%m-%d %H:%M')}\n\n")
            for i, (article, summary) in enumerate(zip(articles, summaries)):
                f.write(f"記事 {i+1}:\n")
                f.write(f"要約: {summary}\n")
                f.write(f"元記事の長さ: {len(article.split())} 語\n")
                f.write("-" * 50 + "\n\n")

# 使用例
summarizer = NewsSummarizationSystem("./t5-small-cnn-accelerate")