LLM Fine-tuning完全ガイド2025 - LoRA/QLoRAでコストを90%削減する実装方法
OpenAI、Anthropic、Google等の主要プロバイダーのFine-tuning方法から、LoRA/QLoRAによる効率的な実装まで、2025年最新のLLMファインチューニング手法を徹底解説。実践的なコード例とコスト削減戦略を紹介します。
機械学習フレームワークの二大巨頭、PyTorchとTensorFlowを2025年の視点で徹底比較。最新機能、パフォーマンス、企業採用事例、選定基準まで、AI開発者が知るべきすべてを網羅的に解説します。
2025 年、ai 開発の現場で最も重要な選択の 1 つが、機械学習フレームワークの選定です。 PyTorch と TensorFlow は、それぞれ独自の強みを持ち、異なるユースケースで輝きを放っています。 本記事では、最新の統計データと実践例をもとに、両フレームワークを徹底的に比較分析します。
2025 年 6 月現在、機械学習フレームワーク市場は大きな転換期を迎えています。
指標 | PyTorch | TensorFlow | その他 |
---|---|---|---|
研究論文での採用率 | 80% | 15% | 5% |
企業プロダクション採用率 | 55% | 70% | 25% |
GitHub スター数 | 79k | 183k | - |
新規プロジェクト採用率 | 65% | 30% | 5% |
開発者満足度 | 92% | 78% | varies |
プロダクション対応が本格化
torch.compile による劇的な高速化
LLMやDiffusion ModelでPyTorchが主流に
新規プロジェクトの過半数がPyTorchを選択
用途に応じた使い分けが一般化
PyTorch 2.0 の最大の特徴は、torch.compile
による動的グラフの高速化です。
import torch
import torch.nn as nn
import time
class SimpleModel(nn.Module):
def __init__(self):
super().__init__()
self.linear1 = nn.Linear(784, 512)
self.relu = nn.ReLU()
self.linear2 = nn.Linear(512, 10)
def forward(self, x):
x = self.linear1(x)
x = self.relu(x)
x = self.linear2(x)
return x
# 通常の実行
model = SimpleModel().cuda()
x = torch.randn(1000, 784).cuda()
# ウォームアップ
for _ in range(10):
_ = model(x)
# 計測
start = time.time()
for _ in range(1000):
output = model(x)
torch.cuda.synchronize()
end = time.time()
print(f"実行時間: {end - start:.3f}秒")
# 実行時間: 2.156秒
import torch
import torch.nn as nn
import time
class SimpleModel(nn.Module):
def __init__(self):
super().__init__()
self.linear1 = nn.Linear(784, 512)
self.relu = nn.ReLU()
self.linear2 = nn.Linear(512, 10)
def forward(self, x):
x = self.linear1(x)
x = self.relu(x)
x = self.linear2(x)
return x
# torch.compile で最適化
model = SimpleModel().cuda()
model = torch.compile(model, mode="reduce-overhead")
x = torch.randn(1000, 784).cuda()
# ウォームアップ(コンパイル実行)
for _ in range(10):
_ = model(x)
# 計測
start = time.time()
for _ in range(1000):
output = model(x)
torch.cuda.synchronize()
end = time.time()
print(f"実行時間: {end - start:.3f}秒")
# 実行時間: 0.892秒(2.4倍高速化)
import torch
import torch.nn as nn
import time
class SimpleModel(nn.Module):
def __init__(self):
super().__init__()
self.linear1 = nn.Linear(784, 512)
self.relu = nn.ReLU()
self.linear2 = nn.Linear(512, 10)
def forward(self, x):
x = self.linear1(x)
x = self.relu(x)
x = self.linear2(x)
return x
# 通常の実行
model = SimpleModel().cuda()
x = torch.randn(1000, 784).cuda()
# ウォームアップ
for _ in range(10):
_ = model(x)
# 計測
start = time.time()
for _ in range(1000):
output = model(x)
torch.cuda.synchronize()
end = time.time()
print(f"実行時間: {end - start:.3f}秒")
# 実行時間: 2.156秒
import torch
import torch.nn as nn
import time
class SimpleModel(nn.Module):
def __init__(self):
super().__init__()
self.linear1 = nn.Linear(784, 512)
self.relu = nn.ReLU()
self.linear2 = nn.Linear(512, 10)
def forward(self, x):
x = self.linear1(x)
x = self.relu(x)
x = self.linear2(x)
return x
# torch.compile で最適化
model = SimpleModel().cuda()
model = torch.compile(model, mode="reduce-overhead")
x = torch.randn(1000, 784).cuda()
# ウォームアップ(コンパイル実行)
for _ in range(10):
_ = model(x)
# 計測
start = time.time()
for _ in range(1000):
output = model(x)
torch.cuda.synchronize()
end = time.time()
print(f"実行時間: {end - start:.3f}秒")
# 実行時間: 0.892秒(2.4倍高速化)
torch.compile
は PyTorch 2.0 の目玉機能で、動的グラフを静的に最適化します。
主な最適化モード:
# デフォルトモード(バランス重視)
model = torch.compile(model)
# 推論最適化(最速)
model = torch.compile(model, mode="reduce-overhead")
# メモリ効率重視
model = torch.compile(model, mode="max-autotune")
# カスタムバックエンド
model = torch.compile(model, backend="inductor")
パフォーマンス向上例:
PyTorch 2.0 では Transformer モデルの最適化が大幅に改善されました。
import torch.nn as nn
# Flash Attention 統合
class OptimizedTransformer(nn.Module):
def __init__(self, d_model=768, nhead=12):
super().__init__()
self.transformer = nn.TransformerEncoder(
nn.TransformerEncoderLayer(
d_model=d_model,
nhead=nhead,
batch_first=True,
# PyTorch 2.0 の新機能
activation="gelu",
norm_first=True
),
num_layers=12,
enable_nested_tensor=True # 高速化
)
def forward(self, x, mask=None):
# Better Transformer が自動適用
return self.transformer(x, mask=mask)
メリット:
PyTorch 2.0 の分散学習機能は大幅に強化されています。
import torch.distributed as dist
from torch.nn.parallel import DistributedDataParallel as DDP
from torch.distributed.fsdp import FullyShardedDataParallel as FSDP
# FSDP(完全シャード型データ並列)
def setup_fsdp_model(model):
return FSDP(
model,
auto_wrap_policy=transformer_auto_wrap_policy,
mixed_precision=MixedPrecision(
param_dtype=torch.float16,
reduce_dtype=torch.float16,
buffer_dtype=torch.float16,
),
sharding_strategy=ShardingStrategy.FULL_SHARD,
device_id=torch.cuda.current_device(),
)
# Tensor Parallel(新機能)
from torch.distributed.tensor.parallel import (
parallelize_module,
ColwiseParallel,
RowwiseParallel,
)
def setup_tensor_parallel(model):
return parallelize_module(
model,
device_mesh,
{"fc1": ColwiseParallel(), "fc2": RowwiseParallel()}
)
PyTorch 2.0 では量子化がより使いやすくなりました。
import torch
from torch.ao.quantization import (
get_default_qconfig_mapping,
quantize_fx,
)
# PyTorch 2.0 の新しい量子化API
def quantize_model(model):
# QAT(Quantization Aware Training)
qconfig_mapping = get_default_qconfig_mapping("qnnpack")
# モデル準備
model.eval()
example_inputs = torch.randn(1, 3, 224, 224)
# FX グラフモードで量子化
prepared_model = prepare_fx(
model,
qconfig_mapping,
example_inputs
)
# キャリブレーション
calibrate(prepared_model, data_loader)
# 量子化実行
quantized_model = convert_fx(prepared_model)
return quantized_model
# INT8 動的量子化(推論専用)
quantized_model = torch.quantization.quantize_dynamic(
model,
{torch.nn.Linear, torch.nn.Conv2d},
dtype=torch.qint8
)
効果:
チャートを読み込み中...
TensorFlow 2.x では Keras が標準 api として完全統合されています。
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
# Functional API での柔軟なモデル構築
def create_advanced_model():
inputs = keras.Input(shape=(224, 224, 3))
# 事前学習済みバックボーン
base_model = keras.applications.EfficientNetV2B0(
include_top=False,
input_tensor=inputs,
weights='imagenet'
)
# カスタムヘッド
x = base_model.output
x = layers.GlobalAveragePooling2D()(x)
x = layers.Dense(512, activation='relu')(x)
x = layers.Dropout(0.5)(x)
# マルチタスク出力
classification_output = layers.Dense(
1000,
activation='softmax',
name='classification'
)(x)
regression_output = layers.Dense(
4,
activation='linear',
name='bbox_regression'
)(x)
model = keras.Model(
inputs=inputs,
outputs=[classification_output, regression_output]
)
return model
# カスタムトレーニングループ
@tf.function
def train_step(images, labels, model, optimizer):
with tf.GradientTape() as tape:
predictions = model(images, training=True)
loss = compute_loss(labels, predictions)
gradients = tape.gradient(loss, model.trainable_variables)
optimizer.apply_gradients(zip(gradients, model.trainable_variables))
return loss
TensorFlow は Google TPU に最適化されており、大規模モデルの学習で圧倒的な性能を発揮します。
import tensorflow as tf
# TPU 戦略設定
resolver = tf.distribute.cluster_resolver.TPUClusterResolver()
tf.config.experimental_connect_to_cluster(resolver)
tf.tpu.experimental.initialize_tpu_system(resolver)
strategy = tf.distribute.TPUStrategy(resolver)
# TPU 最適化モデル
with strategy.scope():
model = create_model()
# Mixed Precision Training
policy = tf.keras.mixed_precision.Policy('mixed_bfloat16')
tf.keras.mixed_precision.set_global_policy(policy)
# XLA コンパイル最適化
model.compile(
optimizer=tf.keras.optimizers.Adam(learning_rate=1e-4),
loss='sparse_categorical_crossentropy',
metrics=['accuracy'],
jit_compile=True # XLA 有効化
)
# TPU 向けデータパイプライン
def create_tpu_dataset(batch_size):
dataset = tf.data.TFRecordDataset(filenames)
# TPU 最適化
dataset = dataset.batch(batch_size, drop_remainder=True)
dataset = dataset.prefetch(tf.data.AUTOTUNE)
# シャーディング
options = tf.data.Options()
options.experimental_distribute.auto_shard_policy = (
tf.data.experimental.AutoShardPolicy.DATA
)
dataset = dataset.with_options(options)
return dataset
TPU パフォーマンス:
本番環境への展開で TensorFlow Serving は業界標準となっています。
# モデルの保存(SavedModel形式)
def export_model(model, export_path):
# シグネチャ定義
@tf.function
def serving_fn(inputs):
return model(inputs, training=False)
signatures = {
'serving_default': serving_fn.get_concrete_function(
tf.TensorSpec(shape=[None, 224, 224, 3], dtype=tf.float32)
)
}
# エクスポート
tf.saved_model.save(
model,
export_path,
signatures=signatures
)
# TF Serving 設定(Docker)
"""
docker run -p 8501:8501 \
--mount type=bind,source=/path/to/model,target=/models/my_model \
-e MODEL_NAME=my_model \
tensorflow/serving:latest
"""
# クライアント側の推論
import requests
import json
def predict_with_tf_serving(image_data):
url = 'http://localhost:8501/v1/models/my_model:predict'
# バッチ推論対応
data = json.dumps({
"signature_name": "serving_default",
"instances": image_data.tolist()
})
response = requests.post(url, data=data)
predictions = response.json()['predictions']
return predictions
TF Serving の利点:
TensorFlow Lite による軽量化とエッジデバイス展開。
import tensorflow as tf
# TFLite 変換
def convert_to_tflite(saved_model_dir):
converter = tf.lite.TFLiteConverter.from_saved_model(
saved_model_dir
)
# 最適化オプション
converter.optimizations = [tf.lite.Optimize.DEFAULT]
# INT8 量子化
def representative_dataset():
for _ in range(100):
data = np.random.rand(1, 224, 224, 3).astype(np.float32)
yield [data]
converter.representative_dataset = representative_dataset
converter.target_spec.supported_ops = [
tf.lite.OpsSet.TFLITE_BUILTINS_INT8
]
# 変換実行
tflite_model = converter.convert()
# 保存
with open('model.tflite', 'wb') as f:
f.write(tflite_model)
return tflite_model
# Android/iOS での推論(Kotlin例)
"""
class ImageClassifier(private val context: Context) {
private lateinit var interpreter: Interpreter
fun initialize() {
val model = loadModelFile("model.tflite")
val options = Interpreter.Options().apply {
setNumThreads(4)
setUseNNAPI(true) // Neural Networks API
setUseGPU(true) // GPU Delegate
}
interpreter = Interpreter(model, options)
}
fun classify(bitmap: Bitmap): FloatArray {
val input = preprocessImage(bitmap)
val output = Array(1) { FloatArray(1000) }
interpreter.run(input, output)
return output[0]
}
}
"""
モバイル展開の実績:
モデル | バッチサイズ | PyTorch 2.0 | TensorFlow 2.15 | 速度比 |
---|---|---|---|---|
ResNet-50 | 128 | 312 img/s | 298 img/s | PyTorch +4.7% |
BERT-Base | 32 | 215 seq/s | 189 seq/s | PyTorch +13.8% |
GPT-2 | 16 | 142 seq/s | 168 seq/s | TF +18.3% |
ViT-B/16 | 64 | 187 img/s | 173 img/s | PyTorch +8.1% |
EfficientNet-B7 | 32 | 89 img/s | 102 img/s | TF +14.6% |
Stable Diffusion | 1 | 2.3 img/s | 1.8 img/s | PyTorch +27.8% |
# PyTorch のメモリ管理
import torch
import torch.cuda as cuda
# メモリ使用量計測
def measure_memory_pytorch():
# 初期状態
cuda.empty_cache()
initial = cuda.memory_allocated()
# モデル作成
model = create_large_model().cuda()
model_memory = cuda.memory_allocated() - initial
# バッチ処理
batch = torch.randn(32, 3, 512, 512).cuda()
output = model(batch)
peak_memory = cuda.max_memory_allocated()
return {
'model_memory': model_memory / 1024**3, # GB
'peak_memory': peak_memory / 1024**3,
'reserved': cuda.memory_reserved() / 1024**3
}
# Gradient Checkpointing
from torch.utils.checkpoint import checkpoint
class MemoryEfficientModel(nn.Module):
def forward(self, x):
# メモリ節約のため中間層をチェックポイント
x = checkpoint(self.layer1, x)
x = checkpoint(self.layer2, x)
return self.layer3(x)
# TensorFlow のメモリ管理
import tensorflow as tf
# メモリ使用量計測
def measure_memory_tensorflow():
# GPU メモリ成長を制限
gpus = tf.config.experimental.list_physical_devices('GPU')
tf.config.experimental.set_memory_growth(gpus[0], True)
# メモリ使用量追跡
tf.debugging.set_log_device_placement(True)
with tf.device('/GPU:0'):
# モデル作成
model = create_large_model()
# メモリプロファイリング
tf.profiler.experimental.start('logdir')
# バッチ処理
batch = tf.random.normal([32, 512, 512, 3])
output = model(batch, training=False)
tf.profiler.experimental.stop()
# メモリ統計取得
return tf.config.experimental.get_memory_info('GPU:0')
# Gradient Tape でのメモリ最適化
@tf.function
def train_step_memory_efficient(inputs, labels):
with tf.GradientTape() as tape:
# 勾配累積でメモリ節約
predictions = model(inputs, training=True)
loss = loss_fn(labels, predictions)
# 勾配計算を分割
gradients = tape.gradient(loss, model.trainable_variables)
optimizer.apply_gradients(zip(gradients, model.trainable_variables))
# PyTorch のメモリ管理
import torch
import torch.cuda as cuda
# メモリ使用量計測
def measure_memory_pytorch():
# 初期状態
cuda.empty_cache()
initial = cuda.memory_allocated()
# モデル作成
model = create_large_model().cuda()
model_memory = cuda.memory_allocated() - initial
# バッチ処理
batch = torch.randn(32, 3, 512, 512).cuda()
output = model(batch)
peak_memory = cuda.max_memory_allocated()
return {
'model_memory': model_memory / 1024**3, # GB
'peak_memory': peak_memory / 1024**3,
'reserved': cuda.memory_reserved() / 1024**3
}
# Gradient Checkpointing
from torch.utils.checkpoint import checkpoint
class MemoryEfficientModel(nn.Module):
def forward(self, x):
# メモリ節約のため中間層をチェックポイント
x = checkpoint(self.layer1, x)
x = checkpoint(self.layer2, x)
return self.layer3(x)
# TensorFlow のメモリ管理
import tensorflow as tf
# メモリ使用量計測
def measure_memory_tensorflow():
# GPU メモリ成長を制限
gpus = tf.config.experimental.list_physical_devices('GPU')
tf.config.experimental.set_memory_growth(gpus[0], True)
# メモリ使用量追跡
tf.debugging.set_log_device_placement(True)
with tf.device('/GPU:0'):
# モデル作成
model = create_large_model()
# メモリプロファイリング
tf.profiler.experimental.start('logdir')
# バッチ処理
batch = tf.random.normal([32, 512, 512, 3])
output = model(batch, training=False)
tf.profiler.experimental.stop()
# メモリ統計取得
return tf.config.experimental.get_memory_info('GPU:0')
# Gradient Tape でのメモリ最適化
@tf.function
def train_step_memory_efficient(inputs, labels):
with tf.GradientTape() as tape:
# 勾配累積でメモリ節約
predictions = model(inputs, training=True)
loss = loss_fn(labels, predictions)
# 勾配計算を分割
gradients = tape.gradient(loss, model.trainable_variables)
optimizer.apply_gradients(zip(gradients, model.trainable_variables))
チャートを読み込み中...
import torch
from torch.profiler import profile, ProfilerActivity
# PyTorch Profiler
def profile_model(model, input_data):
with profile(
activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA],
record_shapes=True,
profile_memory=True,
with_stack=True
) as prof:
with torch.no_grad():
for _ in range(100):
model(input_data)
# 結果表示
print(prof.key_averages().table(sort_by="cuda_time_total", row_limit=10))
# TensorBoard 出力
prof.export_chrome_trace("trace.json")
# 勾配フロー可視化
def check_gradients(model):
for name, param in model.named_parameters():
if param.grad is not None:
grad_norm = param.grad.data.norm(2).item()
print(f"{name}: grad_norm = {grad_norm:.4f}")
# 勾配消失/爆発チェック
if grad_norm < 1e-5:
print(f"Warning: Gradient vanishing in {name}")
elif grad_norm > 100:
print(f"Warning: Gradient explosion in {name}")
# フック機能でのデバッグ
def add_debug_hooks(model):
def forward_hook(module, input, output):
print(f"{module.__class__.__name__} output shape: {output.shape}")
if torch.isnan(output).any():
print(f"NaN detected in {module.__class__.__name__}")
for module in model.modules():
module.register_forward_hook(forward_hook)
import tensorflow as tf
# TensorFlow Debugger V2
def debug_model(model, dataset):
# デバッグモード有効化
tf.debugging.enable_check_numerics()
# TensorBoard Profiler
log_dir = "logs/profile"
with tf.profiler.experimental.Profile(log_dir):
for step, (x, y) in enumerate(dataset.take(100)):
with tf.profiler.experimental.Trace('train', step_num=step):
# 訓練ステップ
loss = train_step(x, y)
# 詳細なデバッグ情報
@tf.function
def debug_forward_pass(inputs):
tf.print("Input shape:", tf.shape(inputs))
# 中間出力の監視
with tf.name_scope("debug"):
for layer in model.layers:
inputs = layer(inputs)
tf.print(f"{layer.name} output stats:")
tf.print(" Mean:", tf.reduce_mean(inputs))
tf.print(" Std:", tf.nn.moments(inputs, axes=[0, 1, 2])[1])
# NaN/Inf チェック
tf.debugging.assert_all_finite(
inputs,
f"NaN/Inf in {layer.name}"
)
return inputs
# Model Analyzer
def analyze_model(model):
# モデル構造の可視化
tf.keras.utils.plot_model(
model,
to_file='model.png',
show_shapes=True,
show_layer_names=True
)
# パラメータ統計
total_params = model.count_params()
trainable_params = sum(
tf.size(w).numpy() for w in model.trainable_weights
)
print(f"Total parameters: {total_params:,}")
print(f"Trainable parameters: {trainable_params:,}")
PyTorch は研究者の思考に近い形でコードを書けるため、 新しいアイデアを試すのに最適です。一方、TensorFlow は プロダクション環境での信頼性と拡張性において優れています。
カテゴリ | PyTorch | TensorFlow | 特徴 |
---|---|---|---|
NLP | Transformers (HuggingFace) | TF Text | PyTorchが圧倒的に充実 |
Vision | torchvision, timm | TF Vision | 両者とも充実 |
強化学習 | Stable Baselines3 | TF Agents | PyTorchがやや優勢 |
グラフNN | PyTorch Geometric | TF GNN | PyTorchが先行 |
MLOps | PyTorch Lightning | TFX | TensorFlowが成熟 |
モバイル | PyTorch Mobile | TF Lite | TensorFlowが優勢 |
チャートを読み込み中...
# PyTorch Lightning による構造化
import pytorch_lightning as pl
from pytorch_lightning.callbacks import ModelCheckpoint, EarlyStopping
class LitModel(pl.LightningModule):
def __init__(self, config):
super().__init__()
self.save_hyperparameters()
self.model = create_model(config)
self.criterion = nn.CrossEntropyLoss()
def forward(self, x):
return self.model(x)
def training_step(self, batch, batch_idx):
x, y = batch
y_hat = self(x)
loss = self.criterion(y_hat, y)
# 自動ログ
self.log('train_loss', loss, prog_bar=True)
return loss
def validation_step(self, batch, batch_idx):
x, y = batch
y_hat = self(x)
loss = self.criterion(y_hat, y)
acc = (y_hat.argmax(1) == y).float().mean()
self.log('val_loss', loss)
self.log('val_acc', acc)
return loss
def configure_optimizers(self):
optimizer = torch.optim.AdamW(
self.parameters(),
lr=self.hparams.learning_rate
)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
optimizer,
T_max=self.hparams.max_epochs
)
return [optimizer], [scheduler]
# 訓練実行
trainer = pl.Trainer(
max_epochs=100,
accelerator='gpu',
devices=4,
strategy='ddp',
precision=16,
callbacks=[
ModelCheckpoint(monitor='val_acc', mode='max'),
EarlyStopping(monitor='val_loss', patience=10)
]
)
trainer.fit(model, train_loader, val_loader)
# TensorFlow/Keras による構造化
import tensorflow as tf
from tensorflow import keras
class TFModel(keras.Model):
def __init__(self, config):
super().__init__()
self.config = config
self.base_model = create_model(config)
# メトリクス
self.loss_tracker = keras.metrics.Mean(name="loss")
self.acc_tracker = keras.metrics.SparseCategoricalAccuracy(name="acc")
def call(self, inputs, training=None):
return self.base_model(inputs, training=training)
@tf.function
def train_step(self, data):
x, y = data
with tf.GradientTape() as tape:
y_pred = self(x, training=True)
loss = self.compiled_loss(y, y_pred)
# 勾配計算と更新
gradients = tape.gradient(loss, self.trainable_variables)
self.optimizer.apply_gradients(
zip(gradients, self.trainable_variables)
)
# メトリクス更新
self.loss_tracker.update_state(loss)
self.acc_tracker.update_state(y, y_pred)
return {
"loss": self.loss_tracker.result(),
"accuracy": self.acc_tracker.result()
}
def test_step(self, data):
x, y = data
y_pred = self(x, training=False)
loss = self.compiled_loss(y, y_pred)
self.loss_tracker.update_state(loss)
self.acc_tracker.update_state(y, y_pred)
return {
"loss": self.loss_tracker.result(),
"accuracy": self.acc_tracker.result()
}
# 分散訓練設定
strategy = tf.distribute.MirroredStrategy()
with strategy.scope():
model = TFModel(config)
model.compile(
optimizer=keras.optimizers.Adam(learning_rate=1e-3),
loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy']
)
# 訓練実行
model.fit(
train_dataset,
validation_data=val_dataset,
epochs=100,
callbacks=[
keras.callbacks.ModelCheckpoint(
'best_model',
monitor='val_accuracy',
save_best_only=True
),
keras.callbacks.EarlyStopping(
monitor='val_loss',
patience=10
),
keras.callbacks.TensorBoard(log_dir='./logs')
]
)
# PyTorch Lightning による構造化
import pytorch_lightning as pl
from pytorch_lightning.callbacks import ModelCheckpoint, EarlyStopping
class LitModel(pl.LightningModule):
def __init__(self, config):
super().__init__()
self.save_hyperparameters()
self.model = create_model(config)
self.criterion = nn.CrossEntropyLoss()
def forward(self, x):
return self.model(x)
def training_step(self, batch, batch_idx):
x, y = batch
y_hat = self(x)
loss = self.criterion(y_hat, y)
# 自動ログ
self.log('train_loss', loss, prog_bar=True)
return loss
def validation_step(self, batch, batch_idx):
x, y = batch
y_hat = self(x)
loss = self.criterion(y_hat, y)
acc = (y_hat.argmax(1) == y).float().mean()
self.log('val_loss', loss)
self.log('val_acc', acc)
return loss
def configure_optimizers(self):
optimizer = torch.optim.AdamW(
self.parameters(),
lr=self.hparams.learning_rate
)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
optimizer,
T_max=self.hparams.max_epochs
)
return [optimizer], [scheduler]
# 訓練実行
trainer = pl.Trainer(
max_epochs=100,
accelerator='gpu',
devices=4,
strategy='ddp',
precision=16,
callbacks=[
ModelCheckpoint(monitor='val_acc', mode='max'),
EarlyStopping(monitor='val_loss', patience=10)
]
)
trainer.fit(model, train_loader, val_loader)
# TensorFlow/Keras による構造化
import tensorflow as tf
from tensorflow import keras
class TFModel(keras.Model):
def __init__(self, config):
super().__init__()
self.config = config
self.base_model = create_model(config)
# メトリクス
self.loss_tracker = keras.metrics.Mean(name="loss")
self.acc_tracker = keras.metrics.SparseCategoricalAccuracy(name="acc")
def call(self, inputs, training=None):
return self.base_model(inputs, training=training)
@tf.function
def train_step(self, data):
x, y = data
with tf.GradientTape() as tape:
y_pred = self(x, training=True)
loss = self.compiled_loss(y, y_pred)
# 勾配計算と更新
gradients = tape.gradient(loss, self.trainable_variables)
self.optimizer.apply_gradients(
zip(gradients, self.trainable_variables)
)
# メトリクス更新
self.loss_tracker.update_state(loss)
self.acc_tracker.update_state(y, y_pred)
return {
"loss": self.loss_tracker.result(),
"accuracy": self.acc_tracker.result()
}
def test_step(self, data):
x, y = data
y_pred = self(x, training=False)
loss = self.compiled_loss(y, y_pred)
self.loss_tracker.update_state(loss)
self.acc_tracker.update_state(y, y_pred)
return {
"loss": self.loss_tracker.result(),
"accuracy": self.acc_tracker.result()
}
# 分散訓練設定
strategy = tf.distribute.MirroredStrategy()
with strategy.scope():
model = TFModel(config)
model.compile(
optimizer=keras.optimizers.Adam(learning_rate=1e-3),
loss=keras.losses.SparseCategoricalCrossentropy(from_logits=True),
metrics=['accuracy']
)
# 訓練実行
model.fit(
train_dataset,
validation_data=val_dataset,
epochs=100,
callbacks=[
keras.callbacks.ModelCheckpoint(
'best_model',
monitor='val_accuracy',
save_best_only=True
),
keras.callbacks.EarlyStopping(
monitor='val_loss',
patience=10
),
keras.callbacks.TensorBoard(log_dir='./logs')
]
)
評価項目 | 重要度 | PyTorch | TensorFlow | 推奨 |
---|---|---|---|---|
研究開発速度 | 高 | ★★★★★ | ★★★ | PyTorch |
本番環境対応 | 高 | ★★★ | ★★★★★ | TensorFlow |
学習曲線 | 中 | ★★★★★ | ★★★ | PyTorch |
パフォーマンス | 高 | ★★★★ | ★★★★ | 両者同等 |
エコシステム | 中 | ★★★★ | ★★★★★ | 用途次第 |
コミュニティ | 中 | ★★★★★ | ★★★★ | PyTorch |
企業サポート | 低 | ★★★ | ★★★★★ | TensorFlow |
# PyTorch から ONNX エクスポート
import torch
import onnx
def export_pytorch_to_onnx(model, dummy_input, onnx_path):
torch.onnx.export(
model,
dummy_input,
onnx_path,
export_params=True,
opset_version=15,
do_constant_folding=True,
input_names=['input'],
output_names=['output'],
dynamic_axes={
'input': {0: 'batch_size'},
'output': {0: 'batch_size'}
}
)
# TensorFlow での ONNX モデル読み込み
import onnx
import onnx_tf
def load_onnx_in_tensorflow(onnx_path):
onnx_model = onnx.load(onnx_path)
tf_rep = onnx_tf.backend.prepare(onnx_model)
tf_rep.export_graph('tf_model')
# TensorFlow モデルとして読み込み
imported = tf.saved_model.load('tf_model')
return imported
2025 年現在、PyTorch と TensorFlow はそれぞれ異なる強みを持ち、 用途に応じた使い分けが重要です。
2025 年の ai 開発において、フレームワークの選択は もはや宗教戦争ではありません。プロジェクトの要件に 最適なツールを選ぶことが、成功への鍵となります。
最終的には、チームのスキルセット、プロジェクトの要件、 そして将来の拡張性を考慮して選択することが重要です。 多くの組織では、研究開発に PyTorch、本番環境に TensorFlow という ハイブリッドアプローチを採用し始めています。