日本語テキスト生成の精度が低い・コストが高い問題をT5で効率的に解決する方法

# 従来のアプローチ
classification_model = load_bert_classifier()      # 文書分類用
summarization_model = load_summarization_model()   # 要約用
title_generation_model = load_title_model()        # タイトル生成用

# 問題：
# - 3つのモデルを個別管理
# - メモリ使用量が3倍
# - 開発・保守コストが高い

pip install torch transformers pytorch_lightning sentencepiece

from transformers import T5ForConditionalGeneration, T5Tokenizer

# 日本語T5モデルの読み込み
MODEL_NAME = "sonoisa/t5-base-japanese"
tokenizer = T5Tokenizer.from_pretrained(MODEL_NAME, is_fast=True)
model = T5ForConditionalGeneration.from_pretrained(MODEL_NAME)

def generate_text(task_prefix, input_text, max_length=128):
    # タスク指定付きの入力
    input_text = f"{task_prefix}: {input_text}"
    
    # トークン化
    input_ids = tokenizer.encode(input_text, return_tensors="pt", max_length=512, truncation=True)
    
    # テキスト生成
    output_ids = model.generate(
        input_ids,
        max_length=max_length,
        temperature=1.0,
        repetition_penalty=2.0,
        do_sample=True
    )
    
    # デコード
    output_text = tokenizer.decode(output_ids[0], skip_special_tokens=True)
    return output_text

# 文書要約
summary = generate_text("要約", "長い文書の内容...")

# タイトル生成  
title = generate_text("タイトル生成", "記事の本文...")

# 文書分類
category = generate_text("分類", "分類したい文書...")

from torch.utils.data import Dataset

class TsvDataset(Dataset):
    def __init__(self, tokenizer, data_path, input_max_len=512, target_max_len=128):
        self.tokenizer = tokenizer
        self.input_max_len = input_max_len
        self.target_max_len = target_max_len
        self.inputs = []
        self.targets = []
        self._load_data(data_path)
    
    def _load_data(self, data_path):
        with open(data_path, "r", encoding="utf-8") as f:
            for line in f:
                target, input_text = line.strip().split("\t")
                
                # タスク指定付きの入力形式
                input_text = f"要約: {input_text}"
                
                # トークン化
                tokenized_input = self.tokenizer.encode_plus(
                    input_text,
                    max_length=self.input_max_len,
                    padding="max_length",
                    truncation=True,
                    return_tensors="pt"
                )
                
                tokenized_target = self.tokenizer.encode_plus(
                    target,
                    max_length=self.target_max_len,
                    padding="max_length", 
                    truncation=True,
                    return_tensors="pt"
                )
                
                self.inputs.append(tokenized_input)
                self.targets.append(tokenized_target)

# データセットの作成
train_dataset = TsvDataset(tokenizer, "train.tsv")

import pytorch_lightning as pl

class T5FineTuner(pl.LightningModule):
    def __init__(self, model_name, learning_rate=3e-4):
        super().__init__()
        self.model = T5ForConditionalGeneration.from_pretrained(model_name)
        self.tokenizer = T5Tokenizer.from_pretrained(model_name)
        self.learning_rate = learning_rate
    
    def forward(self, input_ids, attention_mask, labels=None):
        return self.model(
            input_ids=input_ids,
            attention_mask=attention_mask,
            labels=labels
        )
    
    def training_step(self, batch, batch_idx):
        input_ids = batch["input_ids"]
        attention_mask = batch["attention_mask"]
        labels = batch["labels"]
        
        outputs = self(input_ids, attention_mask, labels)
        loss = outputs.loss
        
        self.log("train_loss", loss)
        return loss
    
    def configure_optimizers(self):
        return torch.optim.AdamW(self.parameters(), lr=self.learning_rate)

# 学習の実行
model = T5FineTuner(MODEL_NAME)
trainer = pl.Trainer(max_epochs=8, gpus=1)
trainer.fit(model, train_dataloader)

def news_summarizer(article_text):
    # 複数の要約タイプを生成
    short_summary = generate_text("短信要約", article_text, max_length=64)
    title = generate_text("タイトル生成", article_text, max_length=32)
    
    return {
        "summary": short_summary,
        "title": title
    }

# 使用例
article = "長いニュース記事の本文..."
result = news_summarizer(article)
print(f"タイトル: {result['title']}")
print(f"要約: {result['summary']}")

def document_processor(document):
    # 文書分類
    category = generate_text("分類", document, max_length=16)
    
    # カテゴリに応じた要約
    if "技術" in category:
        summary = generate_text("技術要約", document)
    elif "ビジネス" in category:
        summary = generate_text("ビジネス要約", document)
    else:
        summary = generate_text("一般要約", document)
    
    return {
        "category": category,
        "summary": summary
    }