개요 : 아래 코드는 다중 센서 시계열 데이터를 입력으로 하여 Transformer를 통해 단일 값을 예측하는 회귀 모델입니다. Positional Encoding을 통해 시간 정보를 반영하며, Self-Attention 기반 인코더를 사용하여 시점 간 및 센서 간 상호작용을 효과적으로 학습합니다.
🔹 전체 코드
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, Dataset
import numpy as np
# ----------------------
# 1. Positional Encoding
# ----------------------
class PositionalEncoding(nn.Module):
def __init__(self, d_model, max_len=500):
super().__init__()
pe = torch.zeros(max_len, d_model)
position = torch.arange(0, max_len).unsqueeze(1).float()
div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-np.log(10000.0) / d_model))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
self.pe = pe.unsqueeze(0)
def forward(self, x):
x = x + self.pe[:, :x.size(1), :].to(x.device)
return x
# ----------------------
# 2. Transformer 기반 회귀 모델
# ----------------------
class SensorTransformer(nn.Module):
def __init__(self, num_sensors, d_model=64, nhead=4, num_layers=2, output_dim=1):
super().__init__()
self.input_proj = nn.Linear(num_sensors, d_model)
self.pos_encoder = PositionalEncoding(d_model)
encoder_layer = nn.TransformerEncoderLayer(d_model=d_model, nhead=nhead, batch_first=True)
self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
self.regressor = nn.Linear(d_model, output_dim)
def forward(self, x):
x = self.input_proj(x)
x = self.pos_encoder(x)
x = self.encoder(x)
x = x.mean(dim=1)
out = self.regressor(x)
return out
# ----------------------
# 3. 예제용 Dummy Dataset
# ----------------------
class DummySensorDataset(Dataset):
def __init__(self, num_samples=1000, time_steps=50, num_sensors=10):
self.X = torch.randn(num_samples, time_steps, num_sensors)
self.Y = self.X.mean(dim=[1, 2], keepdim=True) + torch.randn(num_samples, 1) * 0.1
def __len__(self):
return len(self.X)
def __getitem__(self, idx):
return self.X[idx], self.Y[idx]
# ----------------------
# 4. 학습 루프
# ----------------------
def train_model(model, train_loader, optimizer, criterion, epochs=10):
model.train()
for epoch in range(epochs):
total_loss = 0
for x, y in train_loader:
x, y = x.to(device), y.to(device)
optimizer.zero_grad()
pred = model(x)
loss = criterion(pred, y)
loss.backward()
optimizer.step()
total_loss += loss.item()
print(f"Epoch {epoch+1}/{epochs} - Loss: {total_loss/len(train_loader):.4f}")
# ----------------------
# 5. 테스트 루프 (추론)
# ----------------------
def test_model(model, test_loader):
model.eval()
preds, targets = [], []
with torch.no_grad():
for x, y in test_loader:
x = x.to(device)
output = model(x)
preds.append(output.cpu())
targets.append(y)
return torch.cat(preds), torch.cat(targets)
# ----------------------
# 6. 실행 파트
# ----------------------
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
num_sensors = 10
time_steps = 50
output_dim = 1
train_dataset = DummySensorDataset(num_samples=800, time_steps=time_steps, num_sensors=num_sensors)
test_dataset = DummySensorDataset(num_samples=200, time_steps=time_steps, num_sensors=num_sensors)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=32)
model = SensorTransformer(num_sensors=num_sensors, d_model=64, nhead=4, num_layers=2, output_dim=output_dim).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.MSELoss()
train_model(model, train_loader, optimizer, criterion, epochs=10)
preds, targets = test_model(model, test_loader)
print(f"예측값 (앞 5개): {preds[:5].squeeze().numpy()}")
print(f"실제값 (앞 5개): {targets[:5].squeeze().numpy()}")
주요 구성 요소 설명
① PositionalEncoding: Transformer는 입력 순서를 인식하지 못하므로, 시간 순서 정보를 인코딩하여 입력에 더해줍니다. 사인/코사인 함수를 이용한 위치 임베딩을 사용합니다.
② SensorTransformer: 입력 시계열 데이터를 선형 변환 후 위치 인코딩을 추가하고, Self-Attention 기반 Transformer Encoder를 통과시켜 시계열 관계를 학습합니다. 최종 출력은 평균 풀링 후 선형 회귀층을 통해 Y를 예측합니다.
③ DummySensorDataset: 실제 센서 데이터가 없는 경우를 위한 예제용 데이터셋입니다. 랜덤한 센서값과 해당 평균값(Y)을 기반으로 합니다.
④ train_model: 모델 학습을 위한 루프입니다. MSE 손실을 최소화하며, Adam 옵티마이저를 사용해 파라미터를 갱신합니다.
⑤ test_model: 학습된 모델의 추론을 위한 함수입니다. 예측값과 실제값을 반환합니다.
⑥ 실행 파트: 모델 및 데이터셋 초기화, 학습 및 추론을 포함한 전체 파이프라인 실행 파트입니다.
요약 : 이 코드는 다양한 센서 데이터를 기반으로 Transformer 모델을 이용해 단일 수치 Y를 예측하는 데 적합합니다. 시계열 구조와 센서 간 상관관계를 효과적으로 학습할 수 있어 이상 탐지, 품질 예측 등 실무 문제에 적용할 수 있습니다.
| 구성 요소 | 역할 | 비고 |
|---|---|---|
| PositionalEncoding | 시계열 순서 반영 | Transformer의 필수 요소 |
| Transformer Encoder | 시점 간 관계 학습 | Self-Attention 구조 |
| Pooling + Regressor | 전체 시계열 압축 후 Y 예측 | 평균 풀링 + 선형 회귀 |
| Dummy Dataset | 테스트용 시계열 샘플 생성 | 평균값 기반 Y 생성 |
| Train/Test Loop | 모델 학습 및 검증 | PyTorch 기본 구조 |