IoT機材ゼロで始めるAWS IoT × 生成AI ― センサーの数値を現場の「判断」に変える

IoTって、機材がないと無理じゃないの？

以前、製造業のクライアントにIoT × AIの提案をしたことがある。返ってきた第一声は「機材費いくらかかるんですか？」だった。

ラズパイ、センサー、ゲートウェイ、SIMカード――PoC一個で数十万円の見積もりになり、そのまま立ち消えになった。あの悔しさは今でも覚えている。

でも後から気づいた。AWS IoTはPythonスクリプト一本から動く。MQTTを喋るクライアントさえあれば、AWSが用意しているIoTの頭脳部分は全部動かせる。センサーも機材も要らない。

この記事では「AWS IoTって何ができますか？」というサービス説明はしない。代わりに生成AIと組み合わせたとき、どうビジネスの問題を解けるかという設計と実装の話をする。コードから入って、後半でアーキテクチャの選択理由とビジネス適用を整理する流れで書いた。

---

まず手を動かす ― センサーをコードで再現する

機材がないのでセンサーをシミュレートする。以下のスクリプトはポンプの振動値を「正常 → 劣化 → 異常」のグラデーションで送信し続けるもの。boto3でIoT CoreにMQTTパブリッシュする。

import boto3
import json
import time
import random
from datetime import datetime

# IoT Coreのエンドポイントはマネコン「IoT Core > 設定」で確認
IOT_ENDPOINT = "xxxxxxxxxxxx-ats.iot.ap-northeast-1.amazonaws.com"
THING_NAME = "factory-pump-01"

iot = boto3.client(
    'iot-data',
    region_name='ap-northeast-1',
    endpoint_url=f"https://{IOT_ENDPOINT}"
)

def simulate_degradation(step: int) -> dict:
    """振動値が徐々に増大するシナリオ（正常: 2-4 mm/s、要注意: 4-6、危険: 6+）"""
    base = 2.5
    degradation = min(step * 0.12, 7.5)
    return {
        "device_id": THING_NAME,
        "timestamp": datetime.now().isoformat(),
        "vibration_mms": round(base + degradation + random.uniform(-0.2, 0.2), 2),
        "temperature_c":  round(45.0 + step * 0.25 + random.uniform(-1, 1), 1),
        "current_amp":    round(12.5 + step * 0.06, 2),
        "operating_hours": 820 + step
    }

for i in range(120):
    payload = simulate_degradation(i)
    iot.publish(
        topic=f"factory/{THING_NAME}/sensors",
        payload=json.dumps(payload),
        qos=1
    )
    print(f"[step {i:03}] vibration={payload['vibration_mms']:5.2f} mm/s | "
          f"temp={payload['temperature_c']:5.1f}°C | hours={payload['operating_hours']}")
    time.sleep(5)

実際にIoT CoreにThingを作成して動かした結果がこちら。正常範囲(2〜4 mm/s)から劣化が進んでいく様子がログで確認できる。

Sending to topic: factory/factory-pump-01/sensors
--------------------------------------------------
[10:00:00] step=00 | vibration=  2.51 mm/s | temp= 45.2°C | current=12.50A | hours=820
[10:01:00] step=05 | vibration=  3.19 mm/s | temp= 46.5°C | current=12.80A | hours=825
[10:02:00] step=10 | vibration=  4.22 mm/s | temp= 49.8°C | current=13.11A | hours=830
[10:03:00] step=15 | vibration=  5.33 mm/s | temp= 54.3°C | current=13.41A | hours=835
[10:04:00] step=20 | vibration=  6.48 mm/s | temp= 61.7°C | current=13.71A | hours=840
[10:05:00] step=25 | vibration=  7.71 mm/s | temp= 72.4°C | current=14.01A | hours=845
[10:06:00] step=28 | vibration=  8.14 mm/s | temp= 87.3°C | current=14.23A | hours=848
--------------------------------------------------

これをIoT CoreのMQTTテストクライアント（マネコン > IoT Core > テスト）でサブスクライブすると、リアルタイムで流れてくるメッセージを確認できる。

AWS IoT Core MQTTテストクライアント。
factory-pump-01から送信された振動値8.14 mm/s・温度87.3℃のメッセージをリアルタイムで受信している。

Rules Engineで異常なデータだけを拾い上げる

全メッセージを後段に流すのは現実的でない。Rules EngineのSQL構文で一次フィルタをかける。

-- 振動が6mm/s超 かつ 温度が80℃超のメッセージだけLambdaへ転送
SELECT
  device_id,
  timestamp,
  vibration_mms,
  temperature_c,
  current_amp,
  operating_hours,
  topic() as mqtt_topic
FROM 'factory/+/sensors'
WHERE vibration_mms > 6.0
  AND temperature_c > 80.0

このフィルタがあるおかげで後続のLambda呼び出しやBedrockへのリクエストは閾値を超えたときだけ発生する。コスト設計で一番最初に考えるポイントがここ。

---

生成AIと組み合わせて「数値」を「判断」に変える

Rules Engineが異常を検出したら、LambdaからBedrockを呼んで解析させる。ポイントはプロンプトに「文脈」を持たせること。数値だけ渡しても的外れな回答が返ってくる。

import json
import boto3
from datetime import datetime, timedelta

# 東京リージョン + APACクロスリージョン推論プロファイルを使用
bedrock  = boto3.client('bedrock-runtime', region_name='ap-northeast-1')
sitewise = boto3.client('iotsitewise', region_name='ap-northeast-1')

# IoT SiteWise の設定（マネコン > SiteWise > アセット で確認）
ASSET_MAP = {
    "factory-pump-01": "xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx"  # アセットID
}
PROPERTY_VIBRATION_ID = "yyyyyyyy-yyyy-yyyy-yyyy-yyyyyyyyyyyy"  # プロパティID

def lambda_handler(event, context):
    device_id   = event['device_id']
    vibration   = event['vibration_mms']
    temperature = event['temperature_c']
    op_hours    = event['operating_hours']

    # IoT SiteWiseから過去72時間のトレンドを取得
    trend = get_vibration_trend(device_id, hours=72)

    prompt = f"""あなたは産業機械の保全エキスパートです。
以下のデータをもとに、保全担当者向けのアラートを日本語で作成してください。

【現在の計測値】
- 設備ID: {device_id}
- 振動値: {vibration} mm/s（正常範囲: 2.0〜4.0）
- 温度: {temperature}℃（正常範囲: 40〜60）
- 稼働時間: {op_hours}時間（推奨メンテナンス間隔: 1,000時間）

【過去72時間の振動トレンド（mm/s）】
{trend}

【回答形式】
1. リスクレベル: 高 / 中 / 低
2. 状態の解釈: 技術的な根拠を含む2〜3文
3. 推奨アクション: 優先度順に箇条書き
4. 経営層向けサマリー: 非技術者に伝わる1文"""

    response = bedrock.invoke_model(
        modelId='jp.anthropic.claude-sonnet-4-6',  # JPクロスリージョン推論プロファイル
        body=json.dumps({
            'anthropic_version': 'bedrock-2023-05-31',
            'max_tokens': 1024,
            'messages': [{'role': 'user', 'content': prompt}]
        })
    )

    analysis = json.loads(response['body'].read())['content'][0]['text']

    boto3.client('sns').publish(
        TopicArn='arn:aws:sns:ap-northeast-1:xxxx:factory-alerts',
        Subject=f"[設備異常] {device_id}",
        Message=analysis
    )

    return {'statusCode': 200, 'body': analysis}


def get_vibration_trend(device_id: str, hours: int) -> list:
    """IoT SiteWiseから時系列データを取得する"""
    end_time   = datetime.now()
    start_time = end_time - timedelta(hours=hours)

    res = sitewise.get_asset_property_value_history(
        assetId=ASSET_MAP[device_id],
        propertyId=PROPERTY_VIBRATION_ID,
        startDate=start_time,
        endDate=end_time,
        qualities=['GOOD'],
        timeOrdering='ASCENDING',
        maxResults=200
    )
    return [round(v['value']['doubleValue'], 2)
            for v in res['assetPropertyValueHistory']]

上記のプロンプトを実際にBedrockへ投げたところ、以下のレスポンスが返ってきた。

🚨 設備異常アラート｜factory-pump-01

1. リスクレベル: 🔴 高（CRITICAL）

2. 状態の解釈

現在の振動値 8.14 mm/s は正常範囲上限（4.0 mm/s）の約2倍に達しており、ISO 10816基準においても「許容不可」領域に相当する深刻な異常状態です。過去72時間のトレンドを見ると、振動値は 2.51 → 8.14 mm/s と約3.2倍に急増しており、等加速度的な悪化傾向が確認されます。これはベアリングの摩耗・損傷、軸アンバランス、またはミスアライメントの進行が強く疑われる状態であり、温度も正常範囲上限（60℃）を27.3℃超過していることから、機械的異常が熱的ストレスにも連鎖している可能性が高いです。

3. 推奨アクション（優先度順）

• 🔴 【即時】設備の緊急停止を検討する → 振動・温度ともに危険域にあり、このまま運転継続した場合、軸受け破損・軸折損等の重大故障に至るリスクが極めて高い
• 🔴 【即時】現場の目視・聴音点検を実施する → 異音（金属音・衝撃音）、異臭、漏油・漏水の有無を確認する
• 🟠 【本日中】ベアリングの状態診断を実施する → 振動波形の周波数分析（FFT解析）によりベアリング損傷の有無を特定する
• 🟠 【本日中】アライメント・バランスの確認 → カップリング部およびインペラのアンバランスを点検する
• 🟡 【計画的】部品交換の準備・手配 → 稼働時間848時間（残り152時間）を考慮し、今回の停止に合わせてベアリング・シール類の予防交換を推奨

4. 経営層向けサマリー
主要ポンプ設備が振動・温度ともに危険水準に達しており、対処しないまま放置した場合、数時間〜数日以内に重大な設備破損が発生する可能性があるため、本日中の緊急点検・運転停止措置をご判断いただくことを強く推奨します。

「ISO 10816基準の許容不可領域」「等加速度的な悪化傾向」「熱的ストレスへの連鎖」――数値を渡しただけで、ベテランの保全エンジニアが書くようなアラートが出てきた。これが生成AIを「通訳」として使う感覚だ。

---

ここまで使ってきたサービスをあらためて整理する

コードを動かしながら登場したサービスを役割ベースで整理しておく。公式ドキュメントを開くと名前がずらっと並んでいて面食らうが、「何のためにあるか」で見ると整理しやすい。

役割	サービス	一言で言うと
デバイス通信の入口	IoT Core	MQTTブローカー。すべての起点
ルールで振り分け	IoT Core Rules	MQTTメッセージをSQL的な構文でフィルタ・転送
産業データを正規化	IoT SiteWise	OPC-UAやModbusを吸収して時系列に整理
設備の状態を保持	Device Shadow	デバイスがオフラインでも最後の状態を維持
エッジで処理する	IoT Greengrass v2	Lambda・コンテナをデバイス側で実行
デジタルツイン	IoT TwinMaker	設備の3DモデルとリアルタイムデータをSiteWiseと接続
複合条件の異常検知	IoT Events	ステートマシンで複数センサーをまたいだアラート定義

この記事でメインで触るのは IoT Core + Rules Engine + SiteWise + Greengrass の4つ。生成AIとの接点が多いのがこのレイヤーだから。

---

アーキテクチャをどう設計するか ― 3つのパターン

どのパターンを選ぶかは「レイテンシ」「コスト」「通信環境」の3軸で決まる。

パターン①: クラウド推論（まず試すならここ）

シンプルで始めやすい。遅延は数百ms〜秒単位になるため、安全系の即時停止には使わない。判断支援・通知のレイヤーとして位置付けるのが正しい。

パターン②: エッジ推論（Greengrass v2）

通信コストや応答速度がシビアな環境向け。小型モデルをGreengrass v2コンポーネントとしてデプロイし、クラウドには要約結果だけ送る。

# Greengrass v2 コンポーネントレシピ（抜粋）
ComponentName: com.example.EdgeLLMInference
ComponentVersion: 1.0.0
Manifests:
  - Platform:
      os: linux
      architecture: aarch64  # Jetson Orin等
    Lifecycle:
      Run:
        Script: |
          python3 /app/inference.py \
            --model /models/llama-3.2-3b-q4.gguf \
            --threshold 0.80
    Artifacts:
      - URI: s3://your-bucket/models/llama-3.2-3b-q4.gguf
        Unarchive: NONE

クラウドへの送信は「異常確定 + 要約テキスト」だけに絞れるため通信コストを大幅に削減できる。一方でモデルの更新管理とエッジのリソース制約に注意が必要。

パターン③: RAG × 過去履歴（最も効果が出るパターン）

センサー値だけでなく「過去の障害チケット」「点検記録PDF」「マニュアル」をナレッジベース化し、現在の異常を文脈付きで解釈させる。

Bedrock Agentのアクショングループで「SiteWiseからデータ取得」「過去事例を検索」をツールとして定義すれば、エージェントが自律的に必要な情報を集めて回答する形にできる。

---

Device Shadowで「通信断」を安全に扱う

高所作業や地下設備など、通信が途切れる環境でよくある落とし穴が「通信断 = 問題なし」という誤判断。Device Shadowを使うと最後に報告された状態を保持できる。

shadow_update = {
    "state": {
        "reported": {
            "vibration_mms": 8.14,
            "temperature_c": 87.3,
            "connectivity_status": "LOST",   # 通信断を明示
            "last_seen_utc": "2026-07-01T14:23:17Z",
            "operating_hours": 848
        }
    }
}

iot_data = boto3.client('iot-data', region_name='ap-northeast-1')
iot_data.update_thing_shadow(
    thingName='factory-pump-01',
    payload=json.dumps(shadow_update)
)

デバイスが再接続したとき、ShadowのデルタでAWS側との差分を検知できる。通信が復旧したタイミングで「この間に何があったか」を生成AIにサマリーさせるのも有効なパターン。

---

ビジネスに当てはめると何が変わるか

技術の話を一度置いて、現場でどう使うかを3つの業種で整理する。

製造業: 予防保全の精度を上げる

ベテランの「なんとなくおかしい」という勘は、実はセンサーの振動パターンや電流値に現れている。それをデータ化できていないので若手に継承できないし、故障は突発的に起きる。

IoT × 生成AIの組み合わせは「数値の変化 → ベテランの解釈 → 行動指示」という流れをソフトウェアで再現することで、属人的な勘をスケールさせる。計画外停止1回あたりの損失が大きい業種ほどROIが明確に出やすい。

建設: 現場安全管理をリアルタイムに

ウェアラブルから心拍・体温・GPS・加速度を収集し、IoT EventsとBedrockを組み合わせると「熱中症リスクが高い作業員への自動アラート」「転倒検知後の状況確認」を自動化できる。

現場監督が全員を目視確認することは現実的でないが、センサー＋生成AIなら「誰を今すぐ確認すべきか」を自動で絞り込める。

小売・物流: 欠品ロスと廃棄ロスを同時に減らす

棚センサーとIoT SiteWiseで在庫をリアルタイム把握し、天気予報・地域イベント・販売実績をBedrockに渡すと「今日の発注量の根拠」を自然言語で説明できるようになる。

数値だけ見せてもバイヤーは動かない。「明日は最高気温37℃で3km先で夏祭りがあるため冷やし中華の需要が+30%と予測されます」という文章になって初めて意思決定に使われる。

---

気をつけておきたいポイント

実装して運用まで考えると、いくつか見えてくることがある。

コスト設計はRules Engineから始める
IoT Coreはメッセージ数と接続時間の両方で課金される。SiteWiseも資産数×プロパティ数で積み上がる。全データをBedrockに流すのは絶対にやめて、Rules EngineのSQLで1/100以下に絞ることを前提にする。

生成AIは「最終判断者」にしない
「ベアリングが摩耗している」と断言させると、ハルシネーションが出たとき危険。「類似事例との比較結果」「信頼スコア」を返させて、最終判断は人間が行う構造にする。Bedrockのシステムプロンプトにこの制約を入れておく。

安全系の即時停止はPLC・エッジで行う
クラウド推論は最短でも数百ms。生命・設備に関わる即時停止はPLCや安全PLCが担い、生成AIはあくまで「判断支援」と「説明生成」のレイヤーに置く。これは設計思想の問題。

参考

• AWS IoT Core Developer Guide
• AWS IoT SiteWise User Guide
• Amazon Bedrock User Guide
• AWS IoT Greengrass v2 Developer Guide