GPTに画像をラベリングさせて、サクッと分類モデルを作ってみた【アノテ地獄からの解放】

![ChatGPT Image Jun 15, 2025, 09_21_30 AM.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/235259/78e9b67b-1670-4ddc-9c6b-71916f0eba59.png)

# — ChatGPT Visionで画像仕分け → EfficientNetで高速画像分類器を作るまで —

## はじめに

近年、**大規模言語モデル（LLM）**が圧倒的な認識力・判断力を持ち、ゼロショット・ワンショットタスクにおいても驚異的な精度を叩き出すようになってきました。しかし、こうしたモデルを**毎回推論に用いるのは高コストかつ低速**です。

一方で、分類対象が明確なタスクでは、**小型で軽量な画像分類モデル**（**たとえばEfficientNetなど**）**の方が、**高速かつオンデバイス動作にも適している**というメリットがあります。

本記事では、**ChatGPTのVisionモデル（GPT-4o）**を使って、画像を自動でラベル分類・仕分けし、その仕分け結果を元に**EfficientNetベースの画像分類モデルを訓練する一連のフロー**をご紹介します。

---

## 背景と目的

- **LLM（ChatGPT）**には圧倒的なゼロショット判断力がある  
- ただし**毎回LLMで推論するのは高コスト・高レイテンシ**  
- タスクが定義済みなら、**軽量な分類モデルに“知見”を移し替える方が現実的**

この流れに沿って、以下のようなワークフローを構築しました：

> ### Vision LLMによる画像ラベリング → 学習データ生成 → 軽量モデルで再学習 → 高速・安価な分類器完成

---

## ステップ1：GPT-4o Visionによる画像仕分け

```shell
pip install "openai>=1.25.0" backoff
```

```python
#!/usr/bin/env python3
"""
classify_and_sort.py
--------------------
指定フォルダ以下の画像を GPT-4o Vision で分類し、
<dst>/<predicted_label>/ へ移動するワンショットスクリプト。

例:
  export OPENAI_API_KEY="sk-..."   # GPT-4o 用 API キー
  python classify_and_sort.py \
         --src ./images_raw \
         --dst ./images_sorted \
         --labels "cat,dog,bird,hamster" \
         --model gpt-4o \
         --concurrency 4
"""

from __future__ import annotations
import argparse
import asyncio
import base64
import json
import mimetypes
import os
import shutil
import sys
import uuid
from pathlib import Path
from typing import List

import backoff                # pip install backoff
from openai import AsyncOpenAI # pip install "openai>=1.25.0"

# ---------- ユーティリティ ----------

def parse_labels(arg: str) -> List[str]:
    """
    --labels で渡された文字列をリストへ変換。
    - JSON 形式ならそのまま
    - それ以外はカンマ区切り
    """
    try:
        data = json.loads(arg)
        if isinstance(data, list):
            return [str(s).strip() for s in data if str(s).strip()]
    except json.JSONDecodeError:
        pass
    return [s.strip() for s in arg.split(",") if s.strip()]

def to_data_uri(path: Path) -> str:
    """画像ファイルを Base64 data URI へエンコード。"""
    mime, _ = mimetypes.guess_type(path)
    if mime is None or not mime.startswith("image/"):
        raise ValueError(f"{path} is not an image")
    b64 = base64.b64encode(path.read_bytes()).decode("ascii")
    return f"data:{mime};base64,{b64}"

SYSTEM_PROMPT = (
    "あなたは厳格な画像分類器です。\n"
    "次のラベル一覧から **最も適切なラベルを 1 語だけ** ラベルのみで返答してください。\n"
    "他の出力やフォーマットは不要です。"
)

def build_messages(img_data_uri: str, labels: List[str]) -> list:
    """OpenAI Chat API 用 message 生成。"""
    return [
        {
            "role": "system",
            "content": SYSTEM_PROMPT + "\nラベル一覧: " + ", ".join(labels),
        },
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "この画像のラベルを 1 語で答えて"},
                {"type": "image_url", "image_url": {"url": img_data_uri}},
            ],
        },
    ]

# ---------- GPT 呼び出し ----------

async def classify_file(
    client: AsyncOpenAI,
    path: Path,
    labels: List[str],
    model: str,
    temp: float = 0.0,
) -> str:
    """画像 1 枚を GPT-4o Vision で分類し、ラベル文字列を返す。"""
    messages = build_messages(to_data_uri(path), labels)

    @backoff.on_exception(
        backoff.expo,
        (Exception,),  # openai.RateLimitError, openai.APIError などまとめて
        max_time=120,
    )
    async def _call():
        resp = await client.chat.completions.create(
            model=model,
            messages=messages,
            temperature=temp,
            max_tokens=5,
        )
        return resp.choices[0].message.content.strip()

    return await _call()

# ---------- ワーカー / メイン ----------

async def worker(
    queue: asyncio.Queue,
    client: AsyncOpenAI,
    labels: List[str],
    dst_root: Path,
    model: str,
):
    """非同期ワーカー: queue から画像パスを取り出し分類 → 移動。"""
    while True:
        path = await queue.get()
        if path is None:
            queue.task_done()
            break
        try:
            label = await classify_file(client, path, labels, model)
            # 移動先ディレクトリ作成
            tgt_dir = dst_root / label
            tgt_dir.mkdir(parents=True, exist_ok=True)

            # 同名衝突を避けて移動
            tgt_name = path.name
            if (tgt_dir / tgt_name).exists():
                stem, suf = os.path.splitext(tgt_name)
                tgt_name = f"{stem}_{uuid.uuid4().hex[:8]}{suf}"
            shutil.move(str(path), tgt_dir / tgt_name)
            print(f"[{label}] {path.relative_to(dst_root.parent)} → {label}/")
        except Exception as e:
            print(f"Error processing {path}: {e}", file=sys.stderr)
        finally:
            queue.task_done()

async def main(args):
    src = Path(args.src).expanduser().resolve()
    dst = Path(args.dst).expanduser().resolve()
    dst.mkdir(parents=True, exist_ok=True)

    labels = parse_labels(args.labels)
    if not labels:
        raise SystemExit("No labels parsed from --labels")

    client_kwargs = {}
    if args.base_url:
        # 例: Azure OpenAI を使う場合
        client_kwargs = {"base_url": args.base_url, "api_key": os.getenv("OPENAI_API_KEY")}
    client = AsyncOpenAI(**client_kwargs)

    queue: asyncio.Queue[Path] = asyncio.Queue()

    # 再帰的に画像を列挙
    for p in src.rglob("*"):
        if p.is_file() and mimetypes.guess_type(p)[0]?.startswith("image/"):
            queue.put_nowait(p)

    workers = [
        asyncio.create_task(worker(queue, client, labels, dst, args.model))
        for _ in range(args.concurrency)
    ]
    await queue.join()

    # 終了フラグ
    for _ in workers:
        queue.put_nowait(None)
    await asyncio.gather(*workers)

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Classify images via GPT-4o and sort into folders.")
    parser.add_argument("--src", required=True, help="Input root directory containing images")
    parser.add_argument("--dst", required=True, help="Output directory to store sorted images")
    parser.add_argument("--labels", required=True,
                        help='Comma separated list OR JSON array of labels (e.g. "cat,dog" or \'["cat","dog"]\')')
    parser.add_argument("--model", default="gpt-4o", help="OpenAI vision-capable model name")
    parser.add_argument("--concurrency", type=int, default=2, help="Number of concurrent API calls")
    parser.add_argument("--base-url", help="Custom OpenAI base URL (e.g. Azure endpoint)", default=None)
    args = parser.parse_args()

    asyncio.run(main(args))

```

```bash
python classify_and_sort.py \
  --src ./images_raw \
  --dst ./images_sorted \
  --labels "cat,dog,bird,hamster" \
  --model gpt-4o \
  --concurrency 4
```

このスクリプトは以下のように動作します：

1. 指定ディレクトリ内のすべての画像を走査
2. GPT-4oのVision APIを使い、各画像に対して「これは何か？」を問い合わせ
3. 推論結果に応じて、画像をそれぞれのフォルダに分類（`cat/`, `dog/`, ...）

### メリット
- 手動アノテーション不要
- 精度が非常に高い（Zero-shot）
- 少量データからでも十分な分類が可能

---


## ステップ2：EfficientNetで画像分類モデルを訓練

```python
from torchvision import datasets, transforms, models
from torch.utils.data import DataLoader
import torch.nn as nn
import torch.optim as optim

data_dir = "./images_sorted"
batch_size = 32

transform = transforms.Compose([
    transforms.Resize((224, 224)),
    transforms.ToTensor(),
])

dataset = datasets.ImageFolder(data_dir, transform=transform)
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)

model = models.efficientnet_b0(pretrained=True)
model.classifier[1] = nn.Linear(model.classifier[1].in_features, len(dataset.classes))

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = model.to(device)

optimizer = optim.Adam(model.parameters(), lr=1e-4)
criterion = nn.CrossEntropyLoss()

for epoch in range(5):
    model.train()
    for inputs, labels in dataloader:
        inputs, labels = inputs.to(device), labels.to(device)
        optimizer.zero_grad()
        outputs = model(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
```

---

## ステップ3：小型モデルによるリアルタイム推論

```python
torch.save(model.state_dict(), "classifier.pth")
```

```python
from PIL import Image

def predict(image_path):
    image = Image.open(image_path)
    image = transform(image).unsqueeze(0).to(device)
    model.eval()
    with torch.no_grad():
        output = model(image)
    pred = torch.argmax(output, 1)
    return dataset.classes[pred.item()]
```

---

## 考察：なぜこの手法が有効か？

| 観点 | GPT-4o | EfficientNet |
|------|--------|--------------|
| 精度 | 高い（ゼロショット） | 高い（ファインチューニングで） |
| コスト | 高（API課金） | 低（ローカル可） |
| 推論速度 | 遅い（数秒〜） | 非常に速い（ms） |
| 適用範囲 | 広い | 限定的（学習内容に依存） |

このように、**GPTの判断力を教師として使い、EfficientNetに“知見を蒸留”する構造**は、以下のような場面で極めて有効です：

- LLMの推論コストを下げたい  
- モバイル・エッジデバイス上で高速推論したい  
- そもそも大量にアノテーションする余力がない

---

## まとめ

以下の流れで**LLM→軽量モデルへの知識転送の実例**を紹介しました：

1. GPT-4o Visionを使って画像にラベルを付与・仕分け  
2. 分類済みデータでEfficientNetをファインチューニング  
3. 軽量・高速・オンデバイス可能な分類器を実現

---

## Appendix：使用したライブラリと環境

- Python 3.10+
- PyTorch 2.0+
- torchvision
- OpenAI Python SDK (`openai>=1.25.0`)
- backoff, tqdm など

---

## 今後の展望

- ラベルの自動修正や2段階分類による精度向上  
- クラス間の混同がある場合の再ラベル支援  
- Vision LLM × 多クラス分類（>50クラス）への拡張

---

🐣
****
フリーランスエンジニアです。
お仕事のご相談こちらまで
rockyshikoku@gmail.com

Core MLを使ったアプリを作っています。
機械学習関連の情報を発信しています。

[Twitter](https://twitter.com/JackdeS11)
[Medium](https://medium.com/@rockyshikoku)
[GitHub](https://github.com/john-rocky)