概要
異常検知（Anomaly Detection）とは、データのパターンや挙動が「正常」と定義された状態から逸脱している点を自動的に検出するAI・機械学習技術です。製造業における製品の欠陥検出・機械の予知保全・IoTセンサーの異常監視など、組み込みシステムの分野で特に重要な技術です。
異常検知は以下の三種類に分類されます。
| 種類 | 内容 | 例 |
|---|---|---|
| 点異常（Point Anomaly） | 単一データ点が異常 | センサー値の突発的な異常値 |
| 文脈異常（Contextual Anomaly） | 文脈（時系列・位置）から見て異常 | 夏の低温値（単体では正常） |
| 集団異常（Collective Anomaly） | 複数点のパターンが異常 | 機器の振動が徐々に増大 |
エッジAIにおける異常検知の大きな特徴は「正常データだけで学習できる」点です。製造現場では良品のデータは大量に集まりますが、不良品のデータは少ないため、正常分布を学習して外れ値を検出するアプローチが非常に有効です。
歴史・背景
異常検知の研究は統計学的手法から始まり、機械学習・ディープラーニングへと発展してきました。
手法の変遷：
- 1960〜1980年代：Mahalanobis距離・統計的仮説検定による外れ値検出
- 1990〜2000年代：SVM（One-Class SVM）・kNN（k近傍法）の応用
- 2000年代：Isolation Forest、LOF（局所外れ値因子法）の提案
- 2014年：オートエンコーダ（Autoencoder）を使った異常検知が実用化
- 2016年：VAE（変分オートエンコーダ）による確率的異常スコア算出
- 2019年：SPADE・PaDiM・PatchCoreなど工業外観検査向け高精度手法の登場
- 2021年：MVTec Anomaly Detection Datasetが公開され、ベンチマーク整備
- 2022年以降：エッジデバイス向け軽量異常検知モデルの研究活発化
技術仕様
異常検知アルゴリズムの分類
異常検知手法
├── 統計的手法
│ ├── z-score（標準偏差ベース）
│ ├── Mahalanobis距離
│ └── CUSUM（累積和制御図）
│
├── 機械学習ベース
│ ├── Isolation Forest（孤立森）
│ ├── One-Class SVM
│ ├── LOF（局所外れ値因子）
│ └── K-Means クラスタリング
│
└── ディープラーニングベース
├── Autoencoder（再構成誤差）
├── VAE（変分オートエンコーダ）
├── GAN（生成的敵対ネットワーク）
├── PatchCore（特徴マップ距離）
└── LSTM/TCN（時系列異常検知）異常スコアの計算
異常検知では各データ点に「異常スコア」を割り当て、閾値を超えた場合に異常と判定します。
import numpy as np
def compute_anomaly_score(normal_features, test_features, method='mahalanobis'):
"""異常スコアを計算"""
if method == 'mahalanobis':
# Mahalanobis距離: 分散・共分散を考慮した距離
mean = np.mean(normal_features, axis=0)
cov = np.cov(normal_features.T)
cov_inv = np.linalg.inv(cov + 1e-6 * np.eye(len(mean)))
scores = []
for feat in test_features:
diff = feat - mean
score = np.sqrt(diff @ cov_inv @ diff)
scores.append(score)
return np.array(scores)
elif method == 'knn':
# k近傍距離: 正常データセットの最近傍k点への平均距離
from sklearn.neighbors import NearestNeighbors
k = 5
nbrs = NearestNeighbors(n_neighbors=k).fit(normal_features)
distances, _ = nbrs.kneighbors(test_features)
return distances.mean(axis=1)
elif method == 'reconstruction':
# 再構成誤差（オートエンコーダ用）
# normal_features: 再構成後の特徴量
# test_features: 元の特徴量
return np.mean((normal_features - test_features) ** 2, axis=1)
# 閾値設定（正常データの99パーセンタイル）
def determine_threshold(normal_scores, percentile=99):
threshold = np.percentile(normal_scores, percentile)
return thresholdオートエンコーダによる異常検知
オートエンコーダは正常データのみで学習し、入力データを低次元に圧縮（エンコード）した後に復元（デコード）します。異常データは正常パターンに沿った復元ができないため、再構成誤差が大きくなります。
import tensorflow as tf
def build_autoencoder(input_dim, latent_dim=16):
"""1次元センサーデータ用オートエンコーダ"""
# エンコーダ
encoder = tf.keras.Sequential([
tf.keras.layers.Dense(64, activation='relu',
input_shape=(input_dim,)),
tf.keras.layers.Dense(32, activation='relu'),
tf.keras.layers.Dense(latent_dim, activation='relu')
])
# デコーダ
decoder = tf.keras.Sequential([
tf.keras.layers.Dense(32, activation='relu',
input_shape=(latent_dim,)),
tf.keras.layers.Dense(64, activation='relu'),
tf.keras.layers.Dense(input_dim, activation='linear')
])
# オートエンコーダ全体
input_layer = tf.keras.Input(shape=(input_dim,))
encoded = encoder(input_layer)
decoded = decoder(encoded)
autoencoder = tf.keras.Model(input_layer, decoded)
autoencoder.compile(optimizer='adam', loss='mse')
return autoencoder, encoder, decoder
# 正常データのみで学習
autoencoder, encoder, decoder = build_autoencoder(
input_dim=128 # センサーの特徴量次元
)
autoencoder.fit(
normal_data, normal_data, # 入力=出力（自己教師あり）
epochs=50,
batch_size=32,
validation_split=0.1
)
# 推論：再構成誤差を異常スコアとして使用
def predict_anomaly(autoencoder, data, threshold):
reconstructed = autoencoder.predict(data)
mse = np.mean((data - reconstructed) ** 2, axis=1)
is_anomaly = mse > threshold
return is_anomaly, mse動作原理
時系列データの異常検知（IoTセンサー）
振動・温度・電流などの時系列センサーデータに対しては、滑動窓（スライディングウィンドウ）を使って特徴量を抽出します。
import numpy as np
from scipy import stats
def extract_features_from_window(window_data):
"""時系列ウィンドウから統計的特徴量を抽出"""
features = {
'mean': np.mean(window_data),
'std': np.std(window_data),
'max': np.max(window_data),
'min': np.min(window_data),
'rms': np.sqrt(np.mean(window_data**2)),
'peak_to_peak': np.max(window_data) - np.min(window_data),
'skewness': stats.skew(window_data),
'kurtosis': stats.kurtosis(window_data),
'energy': np.sum(window_data**2),
# 周波数ドメイン特徴量（振動解析向け）
'dominant_freq': get_dominant_frequency(window_data),
}
return np.array(list(features.values()))
def get_dominant_frequency(data, sample_rate=1000):
"""FFTで支配的な周波数を取得"""
fft = np.fft.fft(data)
freqs = np.fft.fftfreq(len(data), 1/sample_rate)
magnitude = np.abs(fft[:len(fft)//2])
dominant_freq = freqs[np.argmax(magnitude)]
return dominant_freq
# リアルタイム処理ループ（C擬似コード）// マイコンでの異常検知ループ（C言語）
#define WINDOW_SIZE 128
#define FEATURE_DIM 10
#define ANOMALY_THRESHOLD 0.85f
float sensor_buffer[WINDOW_SIZE];
float features[FEATURE_DIM];
int buffer_idx = 0;
void sensor_interrupt_handler(float new_value) {
sensor_buffer[buffer_idx++] = new_value;
if (buffer_idx >= WINDOW_SIZE) {
// ウィンドウが満杯になったら特徴量抽出
extract_features(sensor_buffer, features, WINDOW_SIZE);
// 推論実行（TFLite Micro）
float anomaly_score = run_anomaly_model(features);
if (anomaly_score > ANOMALY_THRESHOLD) {
trigger_alert(anomaly_score);
}
// ウィンドウをシフト（50%オーバーラップ）
memmove(sensor_buffer,
sensor_buffer + WINDOW_SIZE/2,
sizeof(float) * WINDOW_SIZE/2);
buffer_idx = WINDOW_SIZE / 2;
}
}画像異常検知（外観検査）
工業製品の外観検査では、正常品の画像のみで学習し、異常部位を検出します。
正常品画像のみで学習（教師なし/半教師あり）
↓
テスト画像を入力
↓
特徴マップを抽出（ResNet, EfficientNetのバックボーン）
↓
正常特徴分布との距離を計算
↓
異常ヒートマップ生成（どのピクセルが異常か）
↓
閾値処理 → 異常箇所の特定・可視化用途・ユースケース
予知保全（Predictive Maintenance）
製造設備の振動センサー・温度センサー・電流センサーのデータを常時モニタリングし、故障の前兆を検知します。
正常運転時のパターンを学習
↓
軸受けの異常振動 → 異常スコア上昇 → 保全アラート
モータ過熱 → 異常スコア上昇 → 緊急停止コスト効果：
- 計画外の生産停止を削減（1回の停止で数百万〜数億円の損失）
- 過剰な定期メンテナンスを削減
外観検査（Visual Inspection）
半導体・電子部品・食品・医薬品などの製造ラインで高速・高精度な外観検査を実施します。
| 産業 | 検出対象 | 要求速度 |
|---|---|---|
| 半導体 | パターン欠陥・ボイド | 数ms/枚 |
| プリント基板 | はんだ不良・部品欠品 | 30FPS |
| 食品 | 異物混入・形状異常 | 数百ms/枚 |
| 薬品 | ラベル印刷ミス | 数十ms/枚 |
ネットワーク・セキュリティ（IoTデバイス）
IoTデバイスの通信パターンを学習し、マルウェア感染や不正アクセスを検知します。
医療機器
ウェアラブルデバイスでの不整脈検知・血糖値の異常変動検知など。
実装・開発のポイント
閾値の決め方
閾値設定は異常検知システムの実用性を左右する重要なパラメータです。
def find_optimal_threshold(normal_scores, anomaly_scores):
"""ROC曲線からF1スコアが最大の閾値を探索"""
from sklearn.metrics import roc_curve, f1_score
import numpy as np
# 正常=0、異常=1
labels = np.array([0] * len(normal_scores) +
[1] * len(anomaly_scores))
scores = np.concatenate([normal_scores, anomaly_scores])
# 様々な閾値でF1スコアを計算
thresholds = np.percentile(scores, np.arange(50, 100, 1))
best_f1 = 0
best_threshold = thresholds[0]
for threshold in thresholds:
preds = (scores > threshold).astype(int)
f1 = f1_score(labels, preds)
if f1 > best_f1:
best_f1 = f1
best_threshold = threshold
print(f"最適閾値: {best_threshold:.4f}, F1: {best_f1:.4f}")
return best_threshold
# 実運用では正常データの99パーセンタイルを閾値にすることも多い
threshold_99 = np.percentile(normal_scores, 99)
print(f"99パーセンタイル閾値: {threshold_99:.4f}")モデルの継続的更新
生産環境では機械の状態が変化するため、正常データを継続的に追加してモデルを更新することが重要です。
class AdaptiveAnomalyDetector:
"""正常データを継続的に学習する適応型異常検知器"""
def __init__(self, window_size=1000):
self.normal_buffer = []
self.window_size = window_size
self.model = IsolationForest(contamination=0.01)
def update(self, new_normal_data):
"""新しい正常データでモデルを更新"""
self.normal_buffer.extend(new_normal_data)
# バッファが満杯になったら古いデータを削除
if len(self.normal_buffer) > self.window_size:
self.normal_buffer = self.normal_buffer[-self.window_size:]
# モデルを再学習（定期的に実施）
if len(self.normal_buffer) >= 100:
self.model.fit(np.array(self.normal_buffer))
def predict(self, data):
"""異常スコアを返す"""
score = -self.model.score_samples([data])[0]
return score他技術との比較
| 比較軸 | 異常検知 | 物体検出 | 画像分類 |
|---|---|---|---|
| 学習データ | 正常データのみで可 | 大量のラベル付きデータ必要 | ラベル付きデータ必要 |
| 学習コスト | 低〜中 | 高 | 中 |
| 異常の種類 | 未知の異常に対応 | 既知クラスのみ | 既知クラスのみ |
| 解釈性 | 中（異常スコア） | 高（BOX座標） | 中（確率） |
| リアルタイム性 | 高（軽量モデル多い） | 中〜高 | 高 |
異常検知はエッジAIの中でも特に産業IoT向けの重要な応用分野です。TinyML・TensorFlow Liteなどと組み合わせることでマイコン上での常時監視が実現し、推論アクセラレータを活用すれば高速な外観検査も可能になります。