# ======================================== # 必要なライブラリのインストール # ======================================== # transformers: Hugging FaceのNLPライブラリ # fugashi, ipadic: 日本語の形態素解析（BERTのトークナイザーに必要） !pip install transformers==4.35.0 fugashi==1.3.0 ipadic==1.0.0 -q # scikit-learn: 機械学習の評価指標 # seaborn: 可視化（混同行列など） !pip install scikit-learn seaborn -q # gradio: デモアプリ作成用（オプション） !pip install gradio -q

# ======================================== # ライブラリのインポート # ======================================== import torch import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns import os import re import warnings warnings.filterwarnings(‘ignore’) # scikit-learn: データ分割と評価 from sklearn.model_selection import train_test_split from sklearn.metrics import classification_report, confusion_matrix from sklearn.preprocessing import LabelEncoder # Hugging Face Transformers from transformers import ( BertJapaneseTokenizer, # 日本語BERT用トークナイザー BertForSequenceClassification, # 分類用BERT Trainer, # 学習ループ TrainingArguments, # 学習設定 EarlyStoppingCallback # 早期終了 ) # PyTorch from torch.utils.data import Dataset # GPU確認 device = torch.device(‘cuda’ if torch.cuda.is_available() else ‘cpu’) print(f”Using device: {device}”) if torch.cuda.is_available(): print(f”GPU: {torch.cuda.get_device_name(0)}”)

# ======================================== # Livedoorニュースコーパスのダウンロード # ======================================== import tarfile import urllib.request # データのURL url = ‘https://www.rondhuit.com/download/ldcc-20140209.tar.gz’ filename = ‘ldcc-20140209.tar.gz’ # ダウンロード print(“ダウンロード中…”) urllib.request.urlretrieve(url, filename) print(“ダウンロード完了！”) # tar.gzファイルを解凍 print(“解凍中…”) with tarfile.open(filename, ‘r:gz’) as tar: tar.extractall() print(“解凍完了！”)

# ======================================== # ディレクトリ構造の確認 # ======================================== !ls -la text/

# ======================================== # データ読み込み関数の定義 # ======================================== def load_livedoor_corpus(base_dir=’text’): “”” Livedoorニュースコーパスを読み込む関数 Args: base_dir: データが保存されているディレクトリ Returns: DataFrame: テキストとカテゴリを含むデータ “”” # カテゴリのフォルダ名一覧 categories = [ ‘dokujo-tsushin’, ‘it-life-hack’, ‘kaden-channel’, ‘livedoor-homme’, ‘movie-enter’, ‘peachy’, ‘smax’, ‘sports-watch’, ‘topic-news’ ] # フォルダ名 → 日本語名の対応表 category_names = { ‘dokujo-tsushin’: ‘独女通信’, ‘it-life-hack’: ‘ITライフハック’, ‘kaden-channel’: ‘家電チャンネル’, ‘livedoor-homme’: ‘livedoor HOMME’, ‘movie-enter’: ‘MOVIE ENTER’, ‘peachy’: ‘Peachy’, ‘smax’: ‘エスマックス’, ‘sports-watch’: ‘Sports Watch’, ‘topic-news’: ‘トピックニュース’ } data = [] # 読み込んだデータを格納するリスト # 各カテゴリのフォルダを処理 for category in categories: category_dir = os.path.join(base_dir, category) # フォルダ内のテキストファイル一覧を取得 # LICENSE.txtは除外 files = [f for f in os.listdir(category_dir) if f.endswith(‘.txt’) and f != ‘LICENSE.txt’] # 各ファイルを読み込み for file in files: file_path = os.path.join(category_dir, file) try: with open(file_path, ‘r’, encoding=’utf-8′) as f: lines = f.readlines() # ファイル形式: # 1行目: URL（不要） # 2行目: 日付（不要） # 3行目: タイトル # 4行目以降: 本文 if len(lines) >= 4: title = lines[2].strip() body = ”.join(lines[3:]).strip() # タイトルと本文を結合 text = title + ‘ ‘ + body data.append({ ‘text’: text, ‘category’: category, ‘category_name’: category_names[category] }) except Exception as e: print(f”Error reading {file_path}: {e}”) continue # DataFrameに変換 df = pd.DataFrame(data) return df

# ======================================== # データの読み込み実行 # ======================================== print(“データを読み込み中…”) df = load_livedoor_corpus() print(f”総記事数: {len(df)}”) print(f”カテゴリ数: {df[‘category_name’].nunique()}”) print(“\n=== カテゴリ別記事数 ===”) print(df[‘category_name’].value_counts())

# ======================================== # テキストクリーニング関数 # ======================================== def clean_text(text): “”” テキストをクリーニングする関数 処理内容: 1. URLの除去 2. HTMLタグの除去 3. 改行を空白に変換 4. 複数の空白を1つに 5. 前後の空白を削除 “”” # URLを除去（http:// または https:// で始まる文字列） text = re.sub(r’https?://\S+’, ”, text) # HTMLタグを除去（<>で囲まれた部分） text = re.sub(r’<[^>]+>’, ”, text) # 改行を空白に変換 text = text.replace(‘\n’, ‘ ‘) text = text.replace(‘\r’, ‘ ‘) # 複数の空白を1つにまとめる text = re.sub(r’\s+’, ‘ ‘, text) # 前後の空白を削除 text = text.strip() return text

# ======================================== # クリーニングの適用 # ======================================== print(“テキストをクリーニング中…”) df[‘text_cleaned’] = df[‘text’].apply(clean_text) # クリーニング前後の比較 print(“\n=== クリーニング前 ===”) print(df[‘text’].iloc[0][:150]) print(“\n=== クリーニング後 ===”) print(df[‘text_cleaned’].iloc[0][:150]) # テキストの長さ統計 df[‘text_length’] = df[‘text_cleaned’].apply(len) print(“\n=== テキスト長の統計 ===”) print(f”平均: {df[‘text_length’].mean():.0f}文字”) print(f”最小: {df[‘text_length’].min()}文字”) print(f”最大: {df[‘text_length’].max()}文字”)

# ======================================== # ラベルのエンコーディング # ======================================== # LabelEncoder: カテゴリ名を数値（0, 1, 2, …）に変換 le = LabelEncoder() df[‘label’] = le.fit_transform(df[‘category_name’]) # ラベルと名前の対応を保存 # id2label: 数値 → カテゴリ名 # label2id: カテゴリ名 → 数値 id2label = {i: label for i, label in enumerate(le.classes_)} label2id = {label: i for i, label in id2label.items()} print(“=== ラベルマッピング ===”) for i, label in id2label.items(): print(f” {i}: {label}”)

# ======================================== # Train / Validation / Test に分割 # ======================================== # 1回目の分割: Train+Val (80%) と Test (20%) # stratify=df[‘label’]: カテゴリの比率を維持 train_val_df, test_df = train_test_split( df, test_size=0.2, stratify=df[‘label’], random_state=42 ) # 2回目の分割: Train (75%) と Val (25%) # → 全体でいうと Train 60%, Val 20%, Test 20% train_df, val_df = train_test_split( train_val_df, test_size=0.25, stratify=train_val_df[‘label’], random_state=42 ) print(“=== データ分割結果 ===”) print(f”Train: {len(train_df):,} ({len(train_df)/len(df)*100:.1f}%)”) print(f”Val: {len(val_df):,} ({len(val_df)/len(df)*100:.1f}%)”) print(f”Test: {len(test_df):,} ({len(test_df)/len(df)*100:.1f}%)”)

# ======================================== # モデルとトークナイザーの読み込み # ======================================== # 東北大学が公開している日本語BERTモデル model_name = ‘cl-tohoku/bert-base-japanese-v2’ # トークナイザーの読み込み # BertJapaneseTokenizer: 日本語に対応（MeCabで形態素解析） print(“トークナイザーを読み込み中…”) tokenizer = BertJapaneseTokenizer.from_pretrained(model_name) # モデルの読み込み # BertForSequenceClassification: 文分類用のBERT # num_labels: 分類するカテゴリ数（9） print(“モデルを読み込み中…”) model = BertForSequenceClassification.from_pretrained( model_name, num_labels=len(id2label), # 9カテゴリ id2label=id2label, # 数値 → ラベル label2id=label2id # ラベル → 数値 ) # GPUに転送 model = model.to(device) print(f”\nモデル: {model_name}”) print(f”パラメータ数: {sum(p.numel() for p in model.parameters()):,}”)

# ======================================== # データセットクラスの定義 # ======================================== class NewsDataset(Dataset): “”” ニュース記事のデータセット Args: texts: テキストのリスト labels: ラベルのリスト tokenizer: トークナイザー max_length: 最大トークン数 “”” def __init__(self, texts, labels, tokenizer, max_length=256): self.texts = texts self.labels = labels self.tokenizer = tokenizer self.max_length = max_length def __len__(self): “””データ数を返す””” return len(self.texts) def __getitem__(self, idx): “””指定したインデックスのデータを返す””” text = self.texts[idx] label = self.labels[idx] # トークン化 # max_length: 最大トークン数（長い場合は切り詰め） # padding: 最大長までパディング # truncation: 最大長を超えた場合は切り詰め encoding = self.tokenizer( text, max_length=self.max_length, padding=’max_length’, truncation=True, return_tensors=’pt’ ) # 返すデータ return { ‘input_ids’: encoding[‘input_ids’].flatten(), ‘attention_mask’: encoding[‘attention_mask’].flatten(), ‘labels’: torch.tensor(label, dtype=torch.long) }

# ======================================== # データセットの作成 # ======================================== print(“データセットを作成中…”) # max_length=256: メモリ節約のため（本来は512まで可能） train_dataset = NewsDataset( train_df[‘text_cleaned’].tolist(), train_df[‘label’].tolist(), tokenizer, max_length=256 ) val_dataset = NewsDataset( val_df[‘text_cleaned’].tolist(), val_df[‘label’].tolist(), tokenizer, max_length=256 ) test_dataset = NewsDataset( test_df[‘text_cleaned’].tolist(), test_df[‘label’].tolist(), tokenizer, max_length=256 ) print(f”Train: {len(train_dataset)} samples”) print(f”Val: {len(val_dataset)} samples”) print(f”Test: {len(test_dataset)} samples”) # サンプル確認 sample = train_dataset[0] print(f”\nサンプルの形状:”) print(f” input_ids: {sample[‘input_ids’].shape}”) print(f” attention_mask: {sample[‘attention_mask’].shape}”) print(f” label: {sample[‘labels’].item()} ({id2label[sample[‘labels’].item()]})”)

# ======================================== # 評価関数の定義 # ======================================== from sklearn.metrics import accuracy_score, precision_recall_fscore_support def compute_metrics(pred): “”” 評価指標を計算する関数 Args: pred: Trainerからの予測結果 Returns: dict: 各評価指標 “”” # 正解ラベル labels = pred.label_ids # 予測ラベル（最も確率が高いクラス） preds = pred.predictions.argmax(-1) # Accuracy（正解率） accuracy = accuracy_score(labels, preds) # Precision, Recall, F1（マクロ平均） # average=’macro’: 全カテゴリの平均（不均衡データに適切） precision, recall, f1, _ = precision_recall_fscore_support( labels, preds, average=’macro’ ) return { ‘accuracy’: accuracy, ‘precision’: precision, ‘recall’: recall, ‘f1’: f1 } print(“評価関数を定義しました”)

# ======================================== # 学習設定（TrainingArguments） # ======================================== training_args = TrainingArguments( # 出力ディレクトリ output_dir=’./results’, # 学習パラメータ num_train_epochs=3, # エポック数 per_device_train_batch_size=16, # 訓練バッチサイズ per_device_eval_batch_size=32, # 評価バッチサイズ learning_rate=2e-5, # 学習率 weight_decay=0.01, # 重み減衰（正則化） warmup_steps=500, # ウォームアップステップ数 # ログと評価 logging_steps=100, # ログ出力間隔 evaluation_strategy=’steps’, # 評価タイミング eval_steps=200, # 評価間隔 # モデル保存 save_strategy=’steps’, # 保存タイミング save_steps=200, # 保存間隔 save_total_limit=2, # 保存するチェックポイント数 load_best_model_at_end=True, # 最良モデルを最後に読み込み metric_for_best_model=’f1′, # 最良モデルの基準 # その他 fp16=True, # 混合精度訓練（高速化） report_to=’none’ # ログ出力先（wandbなし） ) print(“学習設定:”) print(f” エポック数: {training_args.num_train_epochs}”) print(f” バッチサイズ: {training_args.per_device_train_batch_size}”) print(f” 学習率: {training_args.learning_rate}”)

# ======================================== # Trainerの作成と学習 # ======================================== # Trainerの初期化 trainer = Trainer( model=model, args=training_args, train_dataset=train_dataset, eval_dataset=val_dataset, compute_metrics=compute_metrics, callbacks=[EarlyStoppingCallback(early_stopping_patience=3)] ) # 学習開始 print(“=” * 60) print(“学習を開始します…”) print(“=” * 60) trainer.train() print(“\n” + “=” * 60) print(“学習完了！”) print(“=” * 60)

# ======================================== # モデルの保存 # ======================================== print(“モデルを保存中…”) trainer.save_model(‘./best_model’) tokenizer.save_pretrained(‘./best_model’) print(“モデルを ./best_model に保存しました”)

# ======================================== # テストセットでの評価 # ======================================== print(“テストセットで評価中…”) test_results = trainer.evaluate(test_dataset) print(“\n=== テスト結果 ===”) print(f”Accuracy: {test_results[‘eval_accuracy’]:.4f}”) print(f”Precision: {test_results[‘eval_precision’]:.4f}”) print(f”Recall: {test_results[‘eval_recall’]:.4f}”) print(f”F1-score: {test_results[‘eval_f1’]:.4f}”)

# ======================================== # 詳細な分類レポート # ======================================== # 予測を生成 predictions = trainer.predict(test_dataset) pred_labels = predictions.predictions.argmax(-1) true_labels = predictions.label_ids # 分類レポート print(“=== 分類レポート ===”) print(classification_report( true_labels, pred_labels, target_names=[id2label[i] for i in range(len(id2label))], digits=4 ))

# ======================================== # 混同行列の可視化 # ======================================== # 混同行列の計算 cm = confusion_matrix(true_labels, pred_labels) # 可視化 plt.figure(figsize=(12, 10)) sns.heatmap( cm, annot=True, # 数値を表示 fmt=’d’, # 整数形式 cmap=’Blues’, # カラーマップ xticklabels=[id2label[i] for i in range(len(id2label))], yticklabels=[id2label[i] for i in range(len(id2label))] ) plt.xlabel(‘予測ラベル’, fontsize=12) plt.ylabel(‘正解ラベル’, fontsize=12) plt.title(‘混同行列 – 日本語ニュース分類’, fontsize=14) plt.xticks(rotation=45, ha=’right’) plt.tight_layout() plt.savefig(‘confusion_matrix.png’, dpi=100) plt.show() print(“混同行列を confusion_matrix.png に保存しました”)

# ======================================== # 推論クラスの定義 # ======================================== class NewsClassifier: “”” ニュース記事分類器 Args: model_path: 保存したモデルのパス “”” def __init__(self, model_path=’./best_model’): self.device = torch.device(‘cuda’ if torch.cuda.is_available() else ‘cpu’) print(f”モデルを読み込み中: {model_path}”) # トークナイザーとモデルの読み込み self.tokenizer = BertJapaneseTokenizer.from_pretrained(model_path) self.model = BertForSequenceClassification.from_pretrained(model_path) self.model.to(self.device) self.model.eval() # 評価モードに設定 # ラベルマッピング self.id2label = self.model.config.id2label print(“読み込み完了！”) def predict(self, text, return_probs=False): “”” 単一テキストを分類 Args: text: 分類するテキスト return_probs: 全カテゴリの確率を返すか Returns: dict: カテゴリと信頼度 “”” # テキストクリーニング text = clean_text(text) # トークン化 inputs = self.tokenizer( text, max_length=256, padding=’max_length’, truncation=True, return_tensors=’pt’ ) inputs = {k: v.to(self.device) for k, v in inputs.items()} # 推論（勾配計算なし） with torch.no_grad(): outputs = self.model(**inputs) logits = outputs.logits # 確率に変換 probs = torch.softmax(logits, dim=-1)[0] # 最も確率が高いカテゴリ pred_id = torch.argmax(probs).item() pred_label = self.id2label[pred_id] confidence = probs[pred_id].item() result = { ‘category’: pred_label, ‘confidence’: confidence } # 全カテゴリの確率を追加 if return_probs: result[‘all_probabilities’] = { self.id2label[i]: prob.item() for i, prob in enumerate(probs) } return result

# ======================================== # 分類器のテスト # ======================================== # 分類器の初期化 classifier = NewsClassifier(‘./best_model’)

# ======================================== # テスト記事で分類 # ======================================== test_articles = [ “プロ野球の試合で、ホームランが飛び交う激しい試合が繰り広げられた。”, “新しいスマートフォンが発売され、高性能なカメラ機能が話題となっている。”, “話題の映画が公開され、初週興行収入で1位を獲得した。”, “最新のノートパソコンが発売。薄型軽量で持ち運びに便利。”, “美容に関する新しいトレンドが女性の間で人気を集めている。” ] print(“=== 分類結果 ===\n”) for i, article in enumerate(test_articles, 1): result = classifier.predict(article, return_probs=True) print(f”[記事 {i}]”) print(f”テキスト: {article}”) print(f”カテゴリ: {result[‘category’]}”) print(f”信頼度: {result[‘confidence’]:.2%}”) # 上位3カテゴリを表示 top3 = sorted( result[‘all_probabilities’].items(), key=lambda x: x[1], reverse=True )[:3] print(“上位3カテゴリ:”) for cat, prob in top3: print(f” {cat}: {prob:.2%}”) print()

# ======================================== # Gradioデモアプリ # ======================================== import gradio as gr def classify_news(text): “”” Gradio用の分類関数 “”” if not text.strip(): return “テキストを入力してください。”, {} result = classifier.predict(text, return_probs=True) # 結果のフォーマット output_text = f”**カテゴリ:** {result[‘category’]}\n” output_text += f”**信頼度:** {result[‘confidence’]:.2%}” return output_text, result[‘all_probabilities’] # Gradioインターフェース demo = gr.Interface( fn=classify_news, inputs=gr.Textbox( lines=5, placeholder=”ニュース記事を入力してください…”, label=”ニュース記事” ), outputs=[ gr.Textbox(label=”分類結果”), gr.Label(label=”各カテゴリの確率”, num_top_classes=9) ], title=”📰 日本語ニュース分類システム”, description=”ニュース記事を入力すると、自動的にカテゴリを分類します。”, examples=[ [“プロ野球の試合で、ホームランが飛び交う激しい試合が繰り広げられた。”], [“新しいスマートフォンが発売され、高性能なカメラ機能が話題となっている。”], [“話題の映画が公開され、初週興行収入で1位を獲得した。”] ] ) # アプリの起動 print(“Gradioアプリを起動中…”) demo.launch(share=True) # share=True で公開URL生成