サーキュラーエコノミーを実装する ― AIで「ゴミを見分ける」廃棄物画像分類モデルをPyTorchで作る

はじめに

「リサイクルしている気持ち」と「実際に資源が循環しているかどうか」の間には、まだ大きな溝があります。その溝を埋めるコア技術が、AIの画像認識 × ロボットアーム による廃棄物の自動仕分け（AI Sorting）です。

本記事では、このテーマをエンジニア目線で咀嚼し、「ゴミの写真を素材クラスに分類するAIモデル」をPyTorchで最短構築する手順 を解説します。AMP Roboticsが展開しているような仕分けロボットの「眼」の部分を、手元のPCで再現するイメージです。

参考にした解説記事：「捨てる」という概念が消える日——AIとロボットが回す「永久に資源が循環する経済」 | AI Robotics Quantum Lab

この記事のゴールは次のとおりです。

• 公開データセット（TrashNetなど）を使い、6〜10クラスの廃棄物分類モデル を学習する
• 転移学習（ResNet18 / MobileNetV3） で短時間・少量データでも高精度を狙う
• 推論スクリプトとWebデモ（Gradio）を用意し、ロボット制御や実運用につなげられる土台 を作る
• デジタルプロダクトパスポート（DPP）連携やハイパースペクトル拡張の設計方針も整理する

対象読者は、PythonとPyTorchの基本に触れたことがあるエンジニア・データサイエンティストです。

---

1. 全体像 ― 何を作るのか

構築するパイプラインはシンプルです。

[廃棄物画像] → [前処理] → [CNN (ResNet18)] → [クラス確率] → [分別指示/ロボット制御]

1. データ：TrashNet（glass / paper / cardboard / plastic / metal / trash の6クラス）
2. モデル：ImageNet事前学習済みResNet18の最終層を差し替えて転移学習
3. 学習：GPUなしでも数十分、Colab T4なら数分で収束する規模
4. 推論：単一画像の分類スクリプトとGradioデモ
5. 拡張：ハイパースペクトル／DPP／ロボット連携の設計ポイント

---

2. 環境構築

2.1 推奨環境

• OS: macOS / Linux / WSL2 / Windows
• Python: 3.10 以上
• GPU: 任意（CPUでも可、GPUなら大幅短縮）

2.2 仮想環境とライブラリ

mkdir circular-ai && cd circular-ai
python3 -m venv .venv
source .venv/bin/activate

pip install --upgrade pip
pip install torch torchvision         # PyTorch本体
pip install scikit-learn matplotlib   # 評価・可視化
pip install pillow tqdm               # 画像IO・進捗
pip install gradio                    # Webデモ

GPUを使う場合は、PyTorch公式インストーラ でCUDA版のコマンドを生成してください。

2.3 プロジェクト構成

circular-ai/
├── data/
│   └── trashnet/              # 展開したデータセット
│       ├── cardboard/
│       ├── glass/
│       ├── metal/
│       ├── paper/
│       ├── plastic/
│       └── trash/
├── src/
│   ├── dataset.py
│   ├── model.py
│   ├── train.py
│   ├── evaluate.py
│   └── infer.py
├── app.py                     # Gradioデモ
└── README.md

---

3. データセットの準備

3.1 TrashNetを入手する

TrashNetはStanfordの学生が公開した、廃棄物画像の古典的データセットです（約2,500枚・6クラス）。

# 例: GitHub経由でzipを取得して data/ に展開
mkdir -p data && cd data
git clone https://github.com/garythung/trashnet.git
unzip trashnet/data/dataset-resized.zip -d trashnet-extracted
mv trashnet-extracted/dataset-resized trashnet
cd ..

⚠️ 配布形態は時期により変わる可能性があります。ライセンス条件を確認の上、研究・検証目的で利用してください。

自前の撮影画像を追加する場合は、data/trashnet/<class_name>/ にJPEGを追加するだけで拡張できます。

3.2 データセットクラス

# src/dataset.py
from pathlib import Path
from torch.utils.data import Dataset
from torchvision import transforms
from PIL import Image


CLASSES = ["cardboard", "glass", "metal", "paper", "plastic", "trash"]


def build_transforms(train: bool = True):
    if train:
        return transforms.Compose([
            transforms.Resize((256, 256)),
            transforms.RandomResizedCrop(224, scale=(0.7, 1.0)),
            transforms.RandomHorizontalFlip(),
            transforms.ColorJitter(0.2, 0.2, 0.2),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406],
                                 std=[0.229, 0.224, 0.225]),
        ])
    return transforms.Compose([
        transforms.Resize((224, 224)),
        transforms.ToTensor(),
        transforms.Normalize(mean=[0.485, 0.456, 0.406],
                             std=[0.229, 0.224, 0.225]),
    ])


class TrashDataset(Dataset):
    def __init__(self, root: str, split_files: list[Path], transform=None):
        self.samples = []
        self.transform = transform
        for f in split_files:
            label = CLASSES.index(f.parent.name)
            self.samples.append((f, label))

    def __len__(self) -> int:
        return len(self.samples)

    def __getitem__(self, idx: int):
        path, label = self.samples[idx]
        img = Image.open(path).convert("RGB")
        if self.transform:
            img = self.transform(img)
        return img, label


def split_dataset(root: str, val_ratio: float = 0.15, test_ratio: float = 0.15, seed: int = 42):
    import random
    random.seed(seed)
    root_path = Path(root)
    files = [p for c in CLASSES for p in (root_path / c).glob("*.jpg")]
    random.shuffle(files)
    n = len(files)
    n_val = int(n * val_ratio)
    n_test = int(n * test_ratio)
    val = files[:n_val]
    test = files[n_val:n_val + n_test]
    train = files[n_val + n_test:]
    return train, val, test

---

4. モデル定義 ― 転移学習で小規模データに対応

# src/model.py
import torch
from torch import nn
from torchvision import models


def build_model(num_classes: int = 6, pretrained: bool = True) -> nn.Module:
    weights = models.ResNet18_Weights.IMAGENET1K_V1 if pretrained else None
    model = models.resnet18(weights=weights)
    in_features = model.fc.in_features
    model.fc = nn.Sequential(
        nn.Dropout(0.3),
        nn.Linear(in_features, num_classes),
    )
    return model


def freeze_backbone(model: nn.Module) -> None:
    """最終層以外を凍結（少データ時に有効）"""
    for name, param in model.named_parameters():
        if not name.startswith("fc."):
            param.requires_grad = False

MobileNetV3を使いたい場合は models.mobilenet_v3_small(...) に差し替え、最終層を model.classifier[-1] に向けて同様に置き換えればOKです。

---

5. 学習スクリプト

# src/train.py
import torch
from torch import nn
from torch.utils.data import DataLoader
from tqdm import tqdm

from src.dataset import TrashDataset, build_transforms, split_dataset, CLASSES
from src.model import build_model, freeze_backbone


def train(
    data_root: str = "data/trashnet",
    epochs: int = 10,
    batch_size: int = 32,
    lr: float = 3e-4,
    freeze: bool = True,
    device: str = "cuda" if torch.cuda.is_available() else "cpu",
):
    train_files, val_files, test_files = split_dataset(data_root)
    train_ds = TrashDataset(data_root, train_files, build_transforms(train=True))
    val_ds = TrashDataset(data_root, val_files, build_transforms(train=False))

    train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True, num_workers=2)
    val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False, num_workers=2)

    model = build_model(num_classes=len(CLASSES)).to(device)
    if freeze:
        freeze_backbone(model)

    criterion = nn.CrossEntropyLoss()
    optim = torch.optim.AdamW(
        [p for p in model.parameters() if p.requires_grad],
        lr=lr, weight_decay=1e-4,
    )
    scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optim, T_max=epochs)

    best_acc = 0.0
    for epoch in range(1, epochs + 1):
        model.train()
        running = 0.0
        for x, y in tqdm(train_loader, desc=f"epoch {epoch}"):
            x, y = x.to(device), y.to(device)
            optim.zero_grad()
            out = model(x)
            loss = criterion(out, y)
            loss.backward()
            optim.step()
            running += loss.item() * x.size(0)
        scheduler.step()
        train_loss = running / len(train_ds)

        # ----- validation -----
        model.eval()
        correct = 0
        with torch.no_grad():
            for x, y in val_loader:
                x, y = x.to(device), y.to(device)
                pred = model(x).argmax(dim=1)
                correct += (pred == y).sum().item()
        val_acc = correct / len(val_ds)

        print(f"epoch {epoch}: train_loss={train_loss:.4f} val_acc={val_acc:.4f}")
        if val_acc > best_acc:
            best_acc = val_acc
            torch.save(model.state_dict(), "best_model.pt")

    print(f"Best val acc: {best_acc:.4f}")


if __name__ == "__main__":
    train()

実行は以下のとおり。

python -m src.train

TrashNet＋ResNet18の転移学習で、Val Accuracyは85〜92%前後 が一つの目安です。凍結を外してフル微調整するとさらに伸びる一方、過学習しやすいので早期終了やデータ拡張を忘れずに。

---

6. 評価 ― 混同行列でクラス間の誤りを把握する

# src/evaluate.py
import torch
import numpy as np
from sklearn.metrics import classification_report, confusion_matrix
from torch.utils.data import DataLoader

from src.dataset import TrashDataset, build_transforms, split_dataset, CLASSES
from src.model import build_model


def evaluate(data_root="data/trashnet", weights="best_model.pt"):
    _, _, test_files = split_dataset(data_root)
    test_ds = TrashDataset(data_root, test_files, build_transforms(train=False))
    loader = DataLoader(test_ds, batch_size=32, num_workers=2)

    device = "cuda" if torch.cuda.is_available() else "cpu"
    model = build_model(len(CLASSES)).to(device)
    model.load_state_dict(torch.load(weights, map_location=device))
    model.eval()

    all_y, all_p = [], []
    with torch.no_grad():
        for x, y in loader:
            x = x.to(device)
            p = model(x).argmax(dim=1).cpu().numpy()
            all_p.append(p)
            all_y.append(y.numpy())
    y_true = np.concatenate(all_y)
    y_pred = np.concatenate(all_p)

    print(classification_report(y_true, y_pred, target_names=CLASSES))
    print("Confusion matrix:")
    print(confusion_matrix(y_true, y_pred))


if __name__ == "__main__":
    evaluate()

plastic と trash のように見た目が近いクラスで誤分類が集中するのが典型です。誤りの傾向を見ながらデータ拡張や追加撮影の戦略を立てます。

---

7. 推論スクリプト

# src/infer.py
import sys
import torch
from PIL import Image

from src.dataset import build_transforms, CLASSES
from src.model import build_model


def predict(image_path: str, weights: str = "best_model.pt") -> dict:
    device = "cuda" if torch.cuda.is_available() else "cpu"
    model = build_model(len(CLASSES)).to(device)
    model.load_state_dict(torch.load(weights, map_location=device))
    model.eval()

    tf = build_transforms(train=False)
    img = Image.open(image_path).convert("RGB")
    x = tf(img).unsqueeze(0).to(device)

    with torch.no_grad():
        logits = model(x)
        probs = torch.softmax(logits, dim=1).cpu().numpy()[0]

    ranked = sorted(zip(CLASSES, probs.tolist()), key=lambda kv: -kv[1])
    return {"top1": ranked[0][0], "probs": ranked}


if __name__ == "__main__":
    print(predict(sys.argv[1]))

---

8. Gradioでサッと触れるデモを作る

# app.py
import gradio as gr
import torch
from PIL import Image

from src.dataset import build_transforms, CLASSES
from src.model import build_model


device = "cuda" if torch.cuda.is_available() else "cpu"
model = build_model(len(CLASSES)).to(device)
model.load_state_dict(torch.load("best_model.pt", map_location=device))
model.eval()
tf = build_transforms(train=False)


def classify(image: Image.Image):
    x = tf(image.convert("RGB")).unsqueeze(0).to(device)
    with torch.no_grad():
        probs = torch.softmax(model(x), dim=1).cpu().numpy()[0]
    return {c: float(p) for c, p in zip(CLASSES, probs)}


demo = gr.Interface(
    fn=classify,
    inputs=gr.Image(type="pil"),
    outputs=gr.Label(num_top_classes=3),
    title="Circular AI: Waste Classifier",
    description="画像からPET/紙/金属/ガラスなどの素材クラスを推定します。",
)


if __name__ == "__main__":
    demo.launch()

python app.py

ブラウザから画像をドラッグ＆ドロップするだけで、分類結果と確率が確認できます。スマホから撮ってその場で試すと手応えをつかみやすいです。

---

9. 実運用への拡張ポイント

9.1 ハイパースペクトル画像対応

RGBだけでは「白いプラスチック」の材質（PET/PP/PS/PVC）は区別しにくいです。ハイパースペクトル（近赤外〜紫外の数十〜数百バンド）を使う場合は、入力チャネル数を変更し、最初のConv層を付け替えます。

model.conv1 = nn.Conv2d(in_channels=BAND_COUNT, out_channels=64,
                        kernel_size=7, stride=2, padding=3, bias=False)

バンド数が多いときは、1D CNN＋2D CNNのハイブリッド や バンド選択（特徴量重要度） を併用すると学習が安定します。

9.2 デジタルプロダクトパスポート（DPP）との連携

QRコード/RFIDの読み取り結果を、画像分類と並列に扱うとリッチな判定ができます。設計のイメージ：

• 画像モデル：素材クラスと汚れの度合いを推定
• DPP：メーカー・素材配合・想定リサイクル経路
• ルールエンジン：両者を突き合わせて「最適な搬送先」を決定

スキーマは EU DPP Regulation 系のドラフトを参考に、product_id / material_composition / disposal_instructions など最低限のフィールドから始めるのが現実的です。

9.3 ロボットアーム制御との接続

ROS 2（rclpy）/ PythonドライバでUR・Dobot・myCobotなどに接続し、分類結果をトピックとしてpublishすれば、アームが対応ビンへ投入できます。class_id と confidence を渡し、低信頼時は「不明ビン」に落とすフェイルセーフ方針が安全です。

9.4 MLOpsで「学び続ける」現場AIにする

• 現場で誤分類された画像を継続収集（Active Learning）
• 週次で再学習し、Val Accuracyが基準を満たしたものだけデプロイ
• モデルバージョン・データ版・評価指標をMLflowなどで追跡
• 性能低下（データドリフト）の検知と、しきい値を割ったらアラート

---

10. 法令・倫理に関する注意

• 廃棄物処理は国・自治体単位で規制があります（日本では廃棄物処理法）。実ラインに組み込む場合は、処理業許可や施設要件の確認が必須です。
• 撮影画像にプライバシー情報（個人を特定できる文書・ラベル）が含まれるケースがあります。公開データに使うときはマスキング処理を。
• モデルの誤分類による資源損失・事故リスクを評価し、人間のレビュー工程を残す運用から始めるのが安全です。

---

まとめ

• サーキュラーエコノミーの最初のボトルネック「廃棄物の仕分け」は、標準的なCNNと転移学習 で最小限の装備から取り組める
• TrashNet＋ResNet18の組み合わせなら、数百行のPython でロボットの「眼」に相当する分類器を構築できる
• ハイパースペクトル・DPP・ロボット制御と組み合わせることで、工場・自治体レベルの産業共生システム に拡張できる
• 「もったいない」を個人の美徳から 計算可能なシステム に翻訳する作業は、今日の手元のPCから始められる

「捨てる」を設計ミスと定義するなら、その設計ミスを潰していく仕事は、私たちエンジニアの領分です。まずは手元で小さなゴミ分類モデルを動かし、自社のサプライチェーンや自治体の現場 に持ち込んでみてください。