# ======================================== # 必要なライブラリのインストール # ======================================== !pip install transformers==4.35.0 datasets==2.14.0 -q

# ======================================== # ライブラリのインポート # ======================================== import torch from transformers import BertTokenizerFast, BertForQuestionAnswering from transformers import Trainer, TrainingArguments from datasets import load_dataset import numpy as np # GPU確認 device = torch.device(‘cuda’ if torch.cuda.is_available() else ‘cpu’) print(f”Using device: {device}”)

# ======================================== # SQuADデータセットの読み込み # ======================================== # load_dataset: Hugging Faceからデータセットをダウンロード # ‘squad’: SQuAD v1.1データセット squad = load_dataset(‘squad’) print(“=== データセットの構造 ===”) print(squad)

# ======================================== # サンプルデータの確認 # ======================================== # 1つ目のサンプルを取得 example = squad[‘train’][0] print(“=== サンプルデータ ===”) print(f”タイトル: {example[‘title’]}”) print(f”\n文書（context）:”) print(f” {example[‘context’][:300]}…”) print(f”\n質問: {example[‘question’]}”) print(f”\n回答: {example[‘answers’][‘text’][0]}”) print(f”回答の開始位置: {example[‘answers’][‘answer_start’][0]}”)

# ======================================== # 回答位置の確認 # ======================================== context = example[‘context’] answer_text = example[‘answers’][‘text’][0] answer_start = example[‘answers’][‘answer_start’][0] # 回答の終了位置を計算 answer_end = answer_start + len(answer_text) print(“=== 回答位置の確認 ===”) print(f”回答テキスト: ‘{answer_text}'”) print(f”開始位置: {answer_start}”) print(f”終了位置: {answer_end}”) # 実際に抽出してみる extracted = context[answer_start:answer_end] print(f”\n抽出結果: ‘{extracted}'”) print(f”一致確認: {extracted == answer_text}”)

# ======================================== # トークナイザーの準備 # ======================================== # bert-base-uncased: 小文字化された英語BERT model_name = ‘bert-base-uncased’ # BertTokenizerFast: 高速版トークナイザー # offset_mappingを取得するためにFast版が必要 tokenizer = BertTokenizerFast.from_pretrained(model_name) print(f”トークナイザー: {model_name}”) print(f”語彙サイズ: {tokenizer.vocab_size}”)

# ======================================== # トークン化の例 # ======================================== # サンプルデータ question = “Where was Einstein born?” context = “Albert Einstein was born in Germany in 1879.” # トークン化 # return_offsets_mapping=True: 各トークンの文字範囲を取得 inputs = tokenizer( question, context, return_offsets_mapping=True ) # トークンを確認 tokens = tokenizer.convert_ids_to_tokens(inputs[‘input_ids’]) print(“=== トークン化の結果 ===”) print(f”トークン数: {len(tokens)}”) print(f”\nトークン一覧:”) for i, (token, offset) in enumerate(zip(tokens, inputs[‘offset_mapping’])): print(f” {i:2d}: {token:15s} 文字範囲: {offset}”)

# ======================================== # 前処理関数の作成 # ======================================== def prepare_train_features(examples): “”” 訓練データの前処理を行う関数 Args: examples: データセットのバッチ Returns: トークン化されたデータ（start/end位置付き） “”” # トークン化 tokenized = tokenizer( examples[‘question’], # 質問 examples[‘context’], # 文書 truncation=’only_second’, # 文書のみ切り詰め max_length=384, # 最大384トークン stride=128, # 長い文書は128トークンずつオーバーラップ return_overflowing_tokens=True, # 長い文書を分割 return_offsets_mapping=True, # 文字位置を取得 padding=’max_length’ # 最大長までパディング ) # 分割されたサンプルと元のサンプルの対応 sample_mapping = tokenized.pop(‘overflow_to_sample_mapping’) offset_mapping = tokenized.pop(‘offset_mapping’) tokenized[‘start_positions’] = [] tokenized[‘end_positions’] = [] for i, offsets in enumerate(offset_mapping): # 元のサンプルのインデックス sample_idx = sample_mapping[i] answers = examples[‘answers’][sample_idx] # 回答がない場合は[CLS]位置（0）に設定 if len(answers[‘answer_start’]) == 0: tokenized[‘start_positions’].append(0) tokenized[‘end_positions’].append(0) continue # 回答の文字位置 start_char = answers[‘answer_start’][0] end_char = start_char + len(answers[‘text’][0]) # 文書部分の開始トークンを探す # token_type_ids: 0=質問, 1=文書 sequence_ids = tokenized.sequence_ids(i) # 文書の開始・終了インデックスを取得 context_start = 0 while sequence_ids[context_start] != 1: context_start += 1 context_end = len(sequence_ids) – 1 while sequence_ids[context_end] != 1: context_end -= 1 # 回答が文書範囲外の場合は[CLS]位置に設定 if offsets[context_start][0] > end_char or offsets[context_end][1] < start_char: tokenized['start_positions'].append(0) tokenized['end_positions'].append(0) continue # 開始トークン位置を探す token_start = context_start while token_start <= context_end and offsets[token_start][0] <= start_char: token_start += 1 token_start -= 1 # 終了トークン位置を探す token_end = context_end while token_end >= context_start and offsets[token_end][1] >= end_char: token_end -= 1 token_end += 1 tokenized[‘start_positions’].append(token_start) tokenized[‘end_positions’].append(token_end) return tokenized print(“前処理関数を定義しました”)

# ======================================== # データセット全体の前処理 # ======================================== # 小さいサブセットで試す（開発用） small_train = squad[‘train’].select(range(1000)) small_valid = squad[‘validation’].select(range(200)) print(“前処理を開始…”) tokenized_train = small_train.map( prepare_train_features, batched=True, remove_columns=small_train.column_names ) tokenized_valid = small_valid.map( prepare_train_features, batched=True, remove_columns=small_valid.column_names ) print(“前処理完了！”) print(f”\n訓練データ: {len(tokenized_train)} サンプル”) print(f”検証データ: {len(tokenized_valid)} サンプル”)

# ======================================== # モデルの読み込み # ======================================== # BertForQuestionAnswering: QA用のBERTモデル model = BertForQuestionAnswering.from_pretrained(model_name) # GPUに転送 model = model.to(device) print(f”モデル: {model_name}”) print(f”パラメータ数: {sum(p.numel() for p in model.parameters()):,}”)

# ======================================== # 訓練設定 # ======================================== training_args = TrainingArguments( output_dir=’./qa_results’, # 出力ディレクトリ num_train_epochs=3, # エポック数 per_device_train_batch_size=8, # バッチサイズ（メモリに応じて調整） per_device_eval_batch_size=16, # 評価時バッチサイズ learning_rate=3e-5, # 学習率 weight_decay=0.01, # 重み減衰 warmup_steps=100, # ウォームアップ logging_steps=50, # ログ出力間隔 evaluation_strategy=’epoch’, # 評価タイミング save_strategy=’epoch’, # 保存タイミング load_best_model_at_end=True, # 最良モデルを保持 fp16=True # 混合精度訓練 ) print(“訓練設定を定義しました”)

# ======================================== # 訓練の実行 # ======================================== # Trainerの初期化 trainer = Trainer( model=model, args=training_args, train_dataset=tokenized_train, eval_dataset=tokenized_valid ) # 訓練開始 print(“=” * 50) print(“訓練を開始します…”) print(“=” * 50) trainer.train() print(“\n訓練完了！”)

# ======================================== # 推論関数の作成 # ======================================== def answer_question(question, context, model, tokenizer, device): “”” 質問に対する回答を予測する関数 Args: question: 質問文 context: 文書（回答が含まれるテキスト） model: 訓練済みモデル tokenizer: トークナイザー device: デバイス Returns: 回答テキストと信頼度 “”” # モデルを評価モードに model.eval() # トークン化 inputs = tokenizer( question, context, return_tensors=’pt’, # PyTorchテンソルで返す max_length=384, truncation=’only_second’, # 文書のみ切り詰め padding=’max_length’ ) # GPUに転送 inputs = {k: v.to(device) for k, v in inputs.items()} # 予測 with torch.no_grad(): outputs = model(**inputs) # 開始・終了位置のスコアを取得 start_logits = outputs.start_logits[0] end_logits = outputs.end_logits[0] # 最もスコアが高い位置を取得 start_idx = torch.argmax(start_logits).item() end_idx = torch.argmax(end_logits).item() # start <= end の制約をチェック if start_idx > end_idx: # 制約違反の場合、次点の候補を探す start_idx, end_idx = end_idx, start_idx # スコア（信頼度） confidence = (start_logits[start_idx] + end_logits[end_idx]).item() # トークンから回答を取得 tokens = tokenizer.convert_ids_to_tokens(inputs[‘input_ids’][0]) answer_tokens = tokens[start_idx:end_idx + 1] # トークンをテキストに変換 answer = tokenizer.convert_tokens_to_string(answer_tokens) # [CLS]や[SEP]などの特殊トークンを除去 answer = answer.replace(‘[CLS]’, ”).replace(‘[SEP]’, ”).strip() return { ‘answer’: answer, ‘confidence’: confidence, ‘start_idx’: start_idx, ‘end_idx’: end_idx } print(“推論関数を定義しました”)

# ======================================== # 予測の実行 # ======================================== # テスト用の文書 context = “”” Albert Einstein was a German-born theoretical physicist who developed the theory of relativity. He was born in Ulm, Germany in 1879 and died in Princeton, New Jersey in 1955. Einstein is best known for his mass-energy equivalence formula E = mc². “”” # 質問リスト questions = [ “Where was Einstein born?”, “When was Einstein born?”, “When did Einstein die?”, “What is Einstein known for?” ] print(“=== 質問応答の結果 ===\n”) for question in questions: result = answer_question(question, context, model, tokenizer, device) print(f”質問: {question}”) print(f”回答: {result[‘answer’]}”) print(f”信頼度: {result[‘confidence’]:.2f}”) print(“-” * 50)

# ======================================== # Hugging Face Pipelineの使用 # ======================================== from transformers import pipeline # QAパイプライン（事前学習済みモデル） # distilbert-base-cased-distilled-squad: 軽量で高速なモデル qa_pipeline = pipeline( ‘question-answering’, model=’distilbert-base-cased-distilled-squad’, device=0 if torch.cuda.is_available() else -1 ) print(“QAパイプラインを読み込みました”)

# ======================================== # Pipelineでの予測 # ======================================== context = “”” The Apollo 11 mission was the first manned mission to land on the Moon. It was launched on July 16, 1969, and Neil Armstrong became the first person to step onto the lunar surface on July 20, 1969. “”” questions = [ “When was Apollo 11 launched?”, “Who was the first person on the Moon?”, “What was Apollo 11?” ] print(“=== Pipeline QA ===\n”) for question in questions: # Pipelineは辞書形式で入力 result = qa_pipeline(question=question, context=context) print(f”質問: {question}”) print(f”回答: {result[‘answer’]}”) print(f”スコア: {result[‘score’]:.4f}”) print(“-” * 50)

# ======================================== # 評価関数の実装 # ======================================== import string from collections import Counter def normalize_answer(text): “”” 回答を正規化する関数 – 小文字化 – 句読点除去 – 冠詞除去 – 空白正規化 “”” # 小文字化 text = text.lower() # 句読点除去 text = ”.join(ch for ch in text if ch not in string.punctuation) # 冠詞除去 articles = [‘a’, ‘an’, ‘the’] words = text.split() words = [w for w in words if w not in articles] # 空白正規化 text = ‘ ‘.join(words) return text def compute_exact_match(prediction, ground_truth): “”” Exact Matchを計算 “”” return int(normalize_answer(prediction) == normalize_answer(ground_truth)) def compute_f1(prediction, ground_truth): “”” F1スコアを計算 “”” pred_tokens = normalize_answer(prediction).split() truth_tokens = normalize_answer(ground_truth).split() # 共通トークン common = Counter(pred_tokens) & Counter(truth_tokens) num_common = sum(common.values()) if len(pred_tokens) == 0 or len(truth_tokens) == 0: return int(pred_tokens == truth_tokens) if num_common == 0: return 0 precision = num_common / len(pred_tokens) recall = num_common / len(truth_tokens) f1 = 2 * precision * recall / (precision + recall) return f1 print(“評価関数を定義しました”)

# ======================================== # 評価例 # ======================================== # テストケース test_cases = [ {“prediction”: “Albert Einstein”, “ground_truth”: “Albert Einstein”}, {“prediction”: “Einstein”, “ground_truth”: “Albert Einstein”}, {“prediction”: “the theory of relativity”, “ground_truth”: “theory of relativity”}, {“prediction”: “1879”, “ground_truth”: “1879”}, {“prediction”: “Germany”, “ground_truth”: “Ulm, Germany”}, ] print(“=== 評価例 ===\n”) for case in test_cases: pred = case[“prediction”] truth = case[“ground_truth”] em = compute_exact_match(pred, truth) f1 = compute_f1(pred, truth) print(f”予測: ‘{pred}'”) print(f”正解: ‘{truth}'”) print(f”EM: {em}, F1: {f1:.4f}”) print(“-” * 40)