AIサービスを「利用する」のではなく「作ってみる」

— GPT時代に“自分たち専用AI”を docker-compose up で立ち上げるまで—

はじめに：なぜ「作る」のか

ChatGPT のような汎用大規模モデルは、広い文脈理解や文章生成に強みがある一方で、「この画像にアニサキスがいるか？」「製造ラインの食品にカビが生えていないか？」「金属片に微細なヒビはあるか？」といった現場固有・ニッチな判定は苦手です。理由はシンプルで、そのニッチに十分な学習データが入っていないから。
さらに、たとえば Google Cloud Vision や一般的な OCR でも、日本語の帳票や崩れたレイアウト、スキャン品質のバラつきが絡むと実務品質に届かないことが珍しくありません。汎用AIの“平均点”は高いが、皆さんの現場で必要な“合格点”とはズレる。このギャップが「導入したいが信用できない」という法人の心理につながります。

ここで効くのが、自社データで教師あり学習し、現場タスクに最適化した専用モデルを作るアプローチです。
結論から言えば、「OK / NG」の正例・負例を地道に集め、最小限のMLOpsとセットで配備すれば、汎用GPTより**実用度の高い“うちの業務に効くAI”**になります。

この記事では、例としてアニサキス有無の画像判定を題材に、

• docker-compose up で起動できる判定Webアプリ（API＋簡易UI）の全ファイル
• 学習・推論パイプライン（PyTorch/ResNet18 転移学習）
• データ設計・評価・運用の勘所（どのくらいの枚数から“妥当”と言えるのか）
  を、システム会社の視点でまとめます。すべてローカル/オンプレ/クラウドどこでも動かせる構成です。

---

まずは完成像（アーキテクチャ）

• frontend: 最小のHTML（画像アップロード→推論結果を表示）
• backend: FastAPI（/predict で推論、/train で転移学習、/metrics で評価、/health で稼働確認）
• model: PyTorch ResNet18 をベースに最後の層を付け替え、OK/NG 2値分類
• storage: 学習済みモデル（model/model.pt）とデータ（data/ok, data/ng）はボリューム永続化
• docker-compose: ワンコマンドでバックエンド＋静的フロントを起動

ディレクトリ構成

anisakis-ai/
├─ docker-compose.yml
├─ .env.example
├─ README.md
├─ backend/
│  ├─ Dockerfile
│  ├─ requirements.txt
│  ├─ app.py
│  ├─ train.py
│  ├─ dataset.py
│  ├─ inference.py
│  └─ utils.py
├─ frontend/
│  └─ index.html
├─ model/
│  └─ (model.pt が保存される)
└─ data/
   ├─ ok/   (アニサキス無しの画像をここに)
   └─ ng/   (アニサキス有りの画像をここに)

---

セットアップ（先に手順）

1. この一式を任意ディレクトリに保存
2. cp .env.example .env（必要ならポート等を調整）
3. 画像データを data/ok, data/ng に配置
4. 学習（2パターン）
   • APIで学習: POST /train を叩く
   • 一発学習: docker compose run --rm backend python train.py
5. 本番起動: docker compose up -d
6. ブラウザで http://localhost:8080 にアクセス → 画像アップ → 結果表示

---

全ファイル（コピペ可）

docker-compose.yml

version: "3.9"
services:
  backend:
    build: ./backend
    container_name: anisakis-backend
    env_file:
      - .env
    volumes:
      - ./data:/app/data
      - ./model:/app/model
    ports:
      - "${BACKEND_PORT_HOST:-8000}:8000"
    restart: unless-stopped

  frontend:
    image: nginx:alpine
    container_name: anisakis-frontend
    volumes:
      - ./frontend:/usr/share/nginx/html:ro
    ports:
      - "${FRONTEND_PORT_HOST:-8080}:80"
    depends_on:
      - backend
    restart: unless-stopped

.env.example

BACKEND_PORT_HOST=8000
FRONTEND_PORT_HOST=8080
NUM_EPOCHS=5
BATCH_SIZE=16
LEARNING_RATE=0.0005

backend/Dockerfile

FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . /app

EXPOSE 8000
CMD ["uvicorn", "app:api", "--host", "0.0.0.0", "--port", "8000"]

backend/requirements.txt

fastapi==0.111.0
uvicorn[standard]==0.30.1
pydantic==2.7.4
python-multipart==0.0.9
torch==2.3.1
torchvision==0.18.1
Pillow==10.3.0
scikit-learn==1.5.0

backend/utils.py

import os, torch, random, numpy as np

def get_device():
    return torch.device("cuda" if torch.cuda.is_available() else "cpu")

def seed_everything(seed: int = 42):
    random.seed(seed); np.random.seed(seed); torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

def ensure_dirs():
    os.makedirs("model", exist_ok=True)
    os.makedirs("data/ok", exist_ok=True)
    os.makedirs("data/ng", exist_ok=True)

backend/dataset.py

import os
from torchvision import datasets, transforms
from torch.utils.data import DataLoader, random_split

def make_transforms(img_size=224):
    train_tf = transforms.Compose([
        transforms.Resize((img_size, img_size)),
        transforms.RandomHorizontalFlip(),
        transforms.ColorJitter(0.1, 0.1, 0.1, 0.05),
        transforms.ToTensor(),
        transforms.Normalize((0.485,0.456,0.406),(0.229,0.224,0.225)),
    ])
    val_tf = transforms.Compose([
        transforms.Resize((img_size, img_size)),
        transforms.ToTensor(),
        transforms.Normalize((0.485,0.456,0.406),(0.229,0.224,0.225)),
    ])
    return train_tf, val_tf

def make_loaders(data_dir="data", batch_size=16, val_ratio=0.2, num_workers=2):
    train_tf, val_tf = make_transforms()
    full = datasets.ImageFolder(root=data_dir, transform=train_tf)
    n_val = int(len(full) * val_ratio)
    n_train = len(full) - n_val
    train_ds, val_ds = random_split(full, [n_train, n_val])
    # val は transform を差し替え
    val_ds.dataset.transform = val_tf

    train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True, num_workers=num_workers)
    val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False, num_workers=num_workers)
    class_names = full.classes  # ["ng","ok"] など
    return train_loader, val_loader, class_names

backend/train.py

import os, torch, time
from torch import nn, optim
from torchvision import models
from sklearn.metrics import classification_report, confusion_matrix
from dataset import make_loaders
from utils import get_device, seed_everything, ensure_dirs

def build_model(n_classes=2):
    base = models.resnet18(weights=models.ResNet18_Weights.DEFAULT)
    for p in base.parameters():
        p.requires_grad = False
    # 最終層だけ学習
    in_feat = base.fc.in_features
    base.fc = nn.Sequential(
        nn.Linear(in_feat, 128),
        nn.ReLU(),
        nn.Dropout(0.2),
        nn.Linear(128, n_classes)
    )
    return base

def train_one_epoch(model, loader, criterion, optimizer, device):
    model.train(); total_loss=0; correct=0; total=0
    for x, y in loader:
        x, y = x.to(device), y.to(device)
        optimizer.zero_grad()
        logits = model(x)
        loss = criterion(logits, y)
        loss.backward(); optimizer.step()
        total_loss += float(loss) * x.size(0)
        preds = logits.argmax(1)
        correct += (preds==y).sum().item()
        total += x.size(0)
    return total_loss/total, correct/total

@torch.no_grad()
def evaluate(model, loader, criterion, device):
    model.eval(); total_loss=0; correct=0; total=0
    all_y=[]; all_p=[]
    for x, y in loader:
        x, y = x.to(device), y.to(device)
        logits = model(x)
        loss = criterion(logits, y)
        total_loss += float(loss) * x.size(0)
        preds = logits.argmax(1)
        correct += (preds==y).sum().item(); total += x.size(0)
        all_y += y.cpu().tolist(); all_p += preds.cpu().tolist()
    rep = classification_report(all_y, all_p, output_dict=True, zero_division=0)
    cm = confusion_matrix(all_y, all_p).tolist()
    return total_loss/total, correct/total, rep, cm

def main():
    seed_everything(42); ensure_dirs()
    epochs = int(os.getenv("NUM_EPOCHS", 5))
    batch = int(os.getenv("BATCH_SIZE", 16))
    lr = float(os.getenv("LEARNING_RATE", 5e-4))
    device = get_device()

    train_loader, val_loader, class_names = make_loaders(batch_size=batch)
    model = build_model(n_classes=len(class_names)).to(device)
    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.fc.parameters(), lr=lr)

    history = []
    for ep in range(1, epochs+1):
        t0=time.time()
        tr_loss, tr_acc = train_one_epoch(model, train_loader, criterion, optimizer, device)
        va_loss, va_acc, rep, cm = evaluate(model, val_loader, criterion, device)
        history.append({"epoch": ep, "train_loss": tr_loss, "train_acc": tr_acc,
                        "val_loss": va_loss, "val_acc": va_acc})
        print(f"[{ep}/{epochs}] train_acc={tr_acc:.3f} val_acc={va_acc:.3f} ({time.time()-t0:.1f}s)")

    torch.save({
        "model_state": model.state_dict(),
        "class_names": class_names,
        "history": history
    }, "model/model.pt")

    print("== Final Validation Report ==")
    print(rep)
    print("== Confusion Matrix ==")
    print(cm)

if __name__ == "__main__":
    main()

backend/inference.py

import io, torch
from PIL import Image
from torchvision import transforms
from utils import get_device
from torchvision.models import resnet18, ResNet18_Weights
from torch import nn

class Predictor:
    def __init__(self, model_path="model/model.pt"):
        ckpt = torch.load(model_path, map_location="cpu")
        self.class_names = ckpt["class_names"]
        base = resnet18(weights=ResNet18_Weights.DEFAULT)
        for p in base.parameters(): p.requires_grad = False
        in_feat = base.fc.in_features
        base.fc = nn.Sequential(
            nn.Linear(in_feat, 128),
            nn.ReLU(),
            nn.Dropout(0.2),
            nn.Linear(128, len(self.class_names))
        )
        base.load_state_dict(ckpt["model_state"])
        self.model = base.eval().to(get_device())

        self.tf = transforms.Compose([
            transforms.Resize((224,224)),
            transforms.ToTensor(),
            transforms.Normalize((0.485,0.456,0.406),(0.229,0.224,0.225)),
        ])

    @torch.no_grad()
    def predict(self, file_bytes: bytes):
        img = Image.open(io.BytesIO(file_bytes)).convert("RGB")
        x = self.tf(img).unsqueeze(0).to(get_device())
        logits = self.model(x)
        probs = torch.softmax(logits, dim=1).cpu().numpy()[0]
        idx = probs.argmax()
        return {
            "label": self.class_names[idx],
            "probabilities": {self.class_names[i]: float(p) for i,p in enumerate(probs)}
        }

backend/app.py

import os
from fastapi import FastAPI, UploadFile, File
from fastapi.responses import JSONResponse
from pydantic import BaseModel
from utils import ensure_dirs
from inference import Predictor
import subprocess, json

api = FastAPI(title="Anisakis Detector API", version="1.0.0")

@api.on_event("startup")
def _startup():
    ensure_dirs()

@api.get("/health")
def health():
    return {"status":"ok"}

# 予め学習済みモデルがある前提でロード。なければエラーハンドリング
def _get_predictor():
    if not os.path.exists("model/model.pt"):
        return None
    return Predictor("model/model.pt")

@api.post("/predict")
async def predict(file: UploadFile = File(...)):
    pred = _get_predictor()
    if pred is None:
        return JSONResponse({"error":"model not found. Train first."}, status_code=400)
    b = await file.read()
    out = pred.predict(b)
    return out

class TrainResponse(BaseModel):
    ok: bool
    stdout: str

@api.post("/train", response_model=TrainResponse)
def train():
    # 別プロセスで学習（簡易実装）
    try:
        res = subprocess.run(["python", "train.py"], capture_output=True, text=True, check=True)
        return TrainResponse(ok=True, stdout=res.stdout + "\n" + res.stderr)
    except subprocess.CalledProcessError as e:
        return JSONResponse({"ok":False, "stdout": e.stdout + "\n" + e.stderr}, status_code=500)

@api.get("/metrics")
def metrics():
    # 学習履歴を返すだけの簡易版
    if not os.path.exists("model/model.pt"):
        return JSONResponse({"error":"model not found."}, status_code=400)
    import torch
    ckpt = torch.load("model/model.pt", map_location="cpu")
    return {"history": ckpt.get("history", []), "class_names": ckpt.get("class_names", [])}

frontend/index.html

<!doctype html>
<html lang="ja">
<head>
  <meta charset="utf-8">
  <title>Anisakis Detector</title>
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <style>
    body{font-family:system-ui,-apple-system,Segoe UI,Roboto,Helvetica,Arial,sans-serif; margin:2rem;}
    header{margin-bottom:1rem}
    .card{border:1px solid #eee; border-radius:12px; padding:1rem; max-width:640px;}
    .row{display:flex; gap:1rem; align-items:center; flex-wrap:wrap;}
    .btn{padding:.6rem 1rem; border-radius:8px; border:1px solid #333; background:#fff; cursor:pointer}
    pre{background:#f7f7f7; padding:.8rem; border-radius:8px; overflow:auto}
  </style>
</head>
<body>
  <header>
    <h1>アニサキス画像判定デモ</h1>
    <p>1) 学習 → 2) 画像アップロード → 3) 結果表示</p>
  </header>

  <section class="card">
    <h2>1) 学習（転移学習）</h2>
    <p><code>data/ok</code> と <code>data/ng</code> に画像を入れてから実行してください。</p>
    <button class="btn" id="train">/train を実行</button>
    <pre id="trainLog"></pre>
  </section>

  <section class="card" style="margin-top:1rem;">
    <h2>2) 推論</h2>
    <div class="row">
      <input type="file" id="file" accept="image/*">
      <button class="btn" id="predict">/predict</button>
    </div>
    <pre id="result"></pre>
  </section>

  <section class="card" style="margin-top:1rem;">
    <h2>3) 学習履歴</h2>
    <button class="btn" id="metricsBtn">/metrics</button>
    <pre id="metrics"></pre>
  </section>

  <script>
    const API = location.origin.replace(/:8080$/,'') + ":8000";

    document.getElementById('train').onclick = async ()=>{
      const r = await fetch(API+"/train", {method:'POST'});
      const j = await r.json();
      document.getElementById('trainLog').textContent = JSON.stringify(j,null,2);
    };

    document.getElementById('predict').onclick = async ()=>{
      const f = document.getElementById('file').files[0];
      if(!f){ alert('画像を選択してください'); return; }
      const fd = new FormData(); fd.append('file', f);
      const r = await fetch(API+"/predict", {method:'POST', body: fd});
      const j = await r.json();
      document.getElementById('result').textContent = JSON.stringify(j,null,2);
    };

    document.getElementById('metricsBtn').onclick = async ()=>{
      const r = await fetch(API+"/metrics");
      const j = await r.json();
      document.getElementById('metrics').textContent = JSON.stringify(j,null,2);
    };
  </script>
</body>
</html>

README.md（要点）

# Anisakis Detector

## Quick Start
1) `data/ok` と `data/ng` に画像を置く  
2) `cp .env.example .env`（必要ならNUM_EPOCHS等を調整）  
3) 学習:
   - `docker compose run --rm backend python train.py`
   - または `POST /train`  
4) `docker compose up -d` → フロント `http://localhost:8080`

## API
- `GET /health`
- `POST /train`
- `POST /predict` (multipart/form-data, key=`file`)
- `GET /metrics`

## Tips
- class_names は `ImageFolder` のディレクトリ名順です（`data/ng`, `data/ok`）。
- GPUがあれば自動利用します（CUDA）。

---

ここからが“現場で勝つ”ための要所

1) データ設計：OK/NGの定義を“運用可能な粒度”に

• 定義の明確化：「アニサキスあり＝NG」「いない＝OK」を誰が見ても同じ基準で判定できるよう、**判断例集（ガイドライン）**を必ず作成します（境界事例・反事例込み）。
• 撮影標準化：解像度・照明・角度・距離をテンプレ化し、**学習と本番の分布差（ドメインギャップ）**を最小化。
• メタ情報：撮影条件、ロット、ライン、検査者IDなどを CSV 等で付与しておくと、原因分析・ドリフト監視が楽になります。

2) データ枚数：どれくらいあれば“妥当”？

二値判定の精度pの推定に関して、単純化すると二項分布の信頼区間の考え方が使えます。
誤差（95%信頼区間の半幅）を m とすると、おおよそ
n ≈ p(1-p) × (1.96 / m)^2
最悪ケース p=0.5 とすると n ≈ 0.25 × (1.96/m)^2。

• 誤差 m=±5% → n ≈ 0.25 × (1.96/0.05)^2 ≈ 384（1クラスあたりの目安として400枚程度）
• 誤差 m=±3% → n ≈ 0.25 × (1.96/0.03)^2 ≈ 1067（1クラスあたり ~1,000枚強）

もちろん 学習に必要な枚数は、被写体の多様度や条件のばらつきで増減しますが、最低でも各クラス数百枚、理想は1,000枚以上を目標にしつつ、学習曲線（データ量 vs 精度）を見ながら増やすのが実務的です。
また、クラス不均衡が強い場合（NGが希少）、ハードマイニングや重み付け、データ拡張、サンプリングで補正すると学習が安定します。

3) 評価：業務KPIで測る

• **再現率（Recall）**重視が多い：NG見逃しが致命的なら、Recall(NG) を最優先に。
• しきい値最適化：ソフトマックス確率のしきい値を ROC/PR カーブで調整。
• 誤判定の費用：FP/FP のコストを明示（例：疑わしきは人検査回し＝追加の人件費）。
• エッジケース分析：誤判定画像をカテゴリ化（照明悪化/ピンボケ/反射/異物…）し、データに還流。

4) 運用：現場に届くMLOpsミニマム

• モデルバージョン管理：model/model.pt にバージョン・学習条件・データスナップショットIDをメタとして保存。
• 再学習の定期運用：新NGケースが発生したら即アノテーション→継ぎ足し学習。
• 監視：本番分布の統計（輝度・コントラスト等）と学習時分布のドリフト監視。
• フェイルセーフ：低信頼度時は人手に回すルールをAPI側で返す（confidence < t なら「要再検」）。

---

使い方（学習→起動→推論）

1. data/ok, data/ng に画像を配置（最低でも各20〜50で動作確認、評価は数百〜）
2. docker compose run --rm backend python backend/train.py
   • NUM_EPOCHS=10 など .env で調整
3. docker compose up -d
4. ブラウザで http://localhost:8080 → 画像をアップロード → 結果 JSON が表示
5. GET /metrics で学習履歴（epochごとのacc/loss）を確認

---

コード解説のポイント

転移学習（ResNet18）

• 既存の汎用画像表現を活かし、最終層のみ学習。データがそこまで多くなくても収束が速い。
• 実運用では、**最後の数ブロックを微調整（Fine-tune）**した方が精度が上がるケースも多い。

データローダ（datasets.ImageFolder）

• data/クラス名/画像 の単純な構成でスタートでき、アノテーションコストが低い。
• 将来的に bbox/segmentation が必要なら Detectron2 や Ultralytics YOLO で領域検出に拡張可。

API 設計（FastAPI）

• /train: 簡易に別プロセス実行。実戦投入ではキューイング（RQ/Celery）や非同期ワーカー化。
• /predict: 画像1枚の即時推論→ label と probabilities を返す。
• /metrics: 可視化用に JSON で履歴を返す。フロントでグラフ化すればOK。

---

よくある落とし穴と対策

• 本番の画像分布が学習時と違う → 撮影標準の徹底、またはドメイン拡張（薄暗い/過露出/ブレ）
• NGが希少で学習が進まない → クラス重み、オーバーサンプリング、合成データ（慎重に）
• 誤検知を嫌って閾値を上げすぎ → Recallを犠牲にしてないか？現場KPIで最適化
• 精度は高いのに現場で使われない → UI/UXの細部（アップロード速度、サムネ、再検ボタン）
• 再学習が回らない → 誤判定の即ラベル返送フロー（人手確認UI→学習リストへ）を作る

---

セキュリティ・ガバナンス

• データ保護：画像は個人情報や機密が含まれることも。保存方針とマスキングを明文化。
• 監査証跡：誰がいつモデルを更新・デプロイしたか、差分と効果を残す。
• 責任分界：AIは判定支援。最終判断の所在（人/機械）を契約・運用に明記。
• 再現性：同一データ・同一条件で同じ結果が出るよう、seed固定・環境のコンテナ化は必須。

---

「どのくらい集めれば妥当？」をもう少しだけ具体的に

• 最初のPoC：各クラス 200〜500枚で学習→しきい値最適化→誤判定分析
• 実務導入ライン：各クラス 1,000〜5,000枚（条件バラつきが多いほど上振れ）
• 継続運用：月次で誤判定30〜100例を再学習に取り込むサイクル
• 精度の「妥当性」= 数学（信頼区間）× 業務（コスト最適化）。NG見逃し1件のコストを金額換算し、最適なしきい値を意思決定しましょう。

---

ここまでのまとめ

• 汎用AIは強い。ただし現場ニッチには学習データが足りない。
• 自社データ×教師あり学習で、「うちの業務に効く」AIを短期間で作れる。
• 本記事の一式は、docker-compose up でAPI＋UIを起動、/train で学習、/predict で推論。
• データの数は「多いほど良い」が、評価・しきい値・運用サイクルを回すことが実用化の鍵。

---

次にやるべきこと（ロードマップ）

1. 評価ダッシュボード（ROC/PR、混同行列、データドリフト）
2. 再学習オートメーション（誤判定の即アノテーション→学習→A/Bロールアウト）
3. 推論最適化（ONNX/TensorRT、エッジ実装）
4. 検出モデル化（領域検出で「ここにアニサキス」とハイライト）
5. 監査とリスク管理（モデルカード、データカード、バージョニング）

---

おわりに：作れば、信用が生まれる

「AIサービスを利用する」から「AIを作って現場へ届ける」へ。
“自分たちのデータ”で鍛えたモデルは、業務KPIと整合し、説明もでき、改善も回せます。
GPT時代の正解は、汎用AI×専用AIのハイブリッド。
本記事の一式は、その第一歩です。ぜひ叩き台としてお使いください。

---

付録：運用FAQ（抜粋）

• Q. GPU必須？
  A. なくてもOK（CPUで学習は時間増）。推論は十分実用的。大量学習はGPU推奨。
• Q. 学習済みモデルはどこに？
  A. model/model.pt。/metrics で履歴を確認。
• Q. データはどう増やす？
  A. 現場で誤判定した画像を即「OK/NG」ラベル付けして継ぎ足し学習。
• Q. クラウドに置いて問題ない？
  A. 法務・情報管理の基準に従って。必要ならオンプレ運用に切替可能（コンテナ同じ）。