生の監視映像からトレーニング可能なデータセットまでをわずか数分で：TwelveLabsとFiftyOneで構築する自動ビデオキュレーター

はじめに

工場内のIPカメラからは毎年、数千テラバイトもの監視映像データが流れていきます。そのデータの中に、先週火曜日に発生した「あわや大惨事」の原因となったフォークリフトの違反運転、夜勤シフト時に発生したPPE（個人用保護具）の着用違反の傾向、そして6週間ものあいだ誰も気づかなかった非常口の閉鎖状況などが記録されています。映像は存在しています。しかし、ラベルが付けられていません。

この、膨大な生の映像データと実用可能なトレーニングデータ（教師データ）との差は、労働安全向けのコンピュータビジョン（画像認識）を実用化するうえでの最大の障害となっています。これまでの解決法としては、AWS SageMaker Ground TruthやScale AI、あるいは専門のアノテーションチームの力を借り、ビデオを一枚一枚、作業員がチェックしていくという方法が使われてきました。この方法は効果的ですが、動画レビュー1時間あたり25〜50ドルのコストがかかり、ビデオの数に従って予算も増えてしまいます。カメラの台数が2倍になれば、アノテーションのための予算も2倍になります。

こうした開発プロセスを180度転換できたらどうでしょうか？ 人が映像を一コマずつチェックしてインシデントを探すのではなく、映像自体の意味で「自動的に整理」して同じような出来事をグループ分けし、その内容を説明してくれれば、これほど便利なことはありません。

このチュートリアルでは、まさにこの「自動分類と説明」を実現します。TwelveLabsのビデオ理解モデルと、FiftyOneのデータ管理プラットフォームを組み合わせることで、以下のような開発パイプラインを構築します。

1. TwelveLabs APIを用いて生の監視映像を取り込んでインデックス化する（インジェストとインデックス作成）

2. 映像のコンテンツを密集したベクトルとして表現し、視覚・音声・状況など複数の情報を一度にキャプチャするマルチモーダル埋め込み（Marengo 3.0）を抽出する

3. あらかじめカテゴリが定義されていなくても、KMeansを用いて安全に関する類似インシデントを自動でグルーピングする（セマンティック・クラスタリング）

4. 生成型ビデオ分析AI（Pegasus 1.2）を活用し、抽象的なクラスタID（例：クラスタAなど）を、人が読めるような安全分類（「フォークリフトの違反」など）に変換する（各クラスタの自動ラベリング）

5. 品質検証のためにFiftyOne上で結果を可視化し、すぐに学習に使用できるPyTorch形式のデータセットとして出力する

重要なポイントは次の点です。TwelveLabsはビデオを「フレームごと」に処理しません。Marengoはビデオ全体を「ボリュームとしてのビデオ（video as volume）」のように圧縮されたマルチモーダルのデータ（音声、テキスト、モーション、視覚要素を含む一つの埋め込み空間）として表現します。そのため、セマンティック・クラスタリングが非常によく機能します。「見た目は異なる」ものの「同じ意味を持つ」クリップ（たとえば、異なる2つのカメラアングルから撮影された、フォークリフトが歩行者に接近してしまった2つのシーン）が、埋め込みベクトル空間上で非常に近い場所に位置づけられるからです。

完成したアプリケーションの簡単なデモについては、以下の動画をご覧ください。

前提条件

このプロジェクトには、次の3点が必要です。

1. Python 3.8以上：Pythonをダウンロードする

2. TwelveLabs APIキー：認証ドキュメント（無料枠も利用可能です）

3. HuggingFace アカウント（任意）：本プロジェクトで使用している、こちらの監視映像データセットを使用する場合にのみ必要です — Voxel51/Safe_and_Unsafe_Behaviours

リポジトリをクローンして、必要な依存ライブラリをインストールします。

>> git clone https://github.com/nathanchess/visual-ai-worker-safety-kit
>> cd

アーキテクチャの概要

コードを書く前に、パイプラインの全体像を確認しておきましょう。大まかには「TwelveLabsが処理（埋め込み表現とラベルの生成）」を担当し、「FiftyOneがプラットフォーム（可視化、フィルタリング、エクスポートなど）」を提供します。

フルスクリーン表示（Lucid App）：[Voxel51 x TwelveLabs] - セマンティック・データセット監視ツール 

Marengo 3.0は、ビデオを豊かなマルチモーダル埋め込み（視覚、音声、文脈情報を捉えたもの）へと圧縮するTwelveLabs社のエンコーダーモデルです。Pegasus 1.2は、その圧縮された表現を学習・推論して、インシデントの認識、行動パターンの分類、および自然言語による説明表現を行います。FiftyOneはそれらの出力を受け取り、インタラクティブな可視化や、意味（類似性）によるデータ閲覧、そしてPyTorchへ向けた直接のエクスポートを行うための役割を担います。

完成したソースコードは、こちらのGitHubリポジトリにあります。以下のチュートリアルは、main.pyの構成に沿って進めます。

ステップ1：インテリジェントな動画のインポート

まずは、Hugging Faceから監視データセットをロードします。FiftyOneの ViewField (F としてインポート) を使うと、元のデータを書き換えることなく宣言的なフィルタ（条件式）を作成できます。今回は「学習用データ (training split) であること」「クリップの長さが4秒以上であること（これより短いと価値ある埋め込みデータを抽出できません）」「インデックス作成がまだ行われていないこと」の3つの条件を設定します。

from fiftyone import ViewField as F
from fiftyone.utils.huggingface import load_from_hub

def load_or_create_dataset(dataset_name):
    # Load directly from Hugging Face Hub
    return load_from_hub("Voxel51/Safe_and_Unsafe_Behaviours", name=dataset_name)

def ingest_videos(client, index_id, dataset, min_duration=4.0):
    # Create a view: Train split ONLY, Duration >= 4s, NOT yet indexed
    base_view = (
        dataset
        .match_tags("train")
        .match(F("metadata.duration") >= min_duration)
        .match(~F("tl_video_id").exists())  # Idempotency check
    )
    
    # ... iteration logic ...

特に、~F("tl_video_id").exists() という指定方法に注目してください。実際の開発現場では、新しく追加された動画に対して、このスクリプトを毎日定期的に実行する場合があります。この記述があれば、同じ動画を（二重に課金されて）再度 TwelveLabs へアップロードしてインデックス化することを防げます。複雑な管理コード（ブックキーピング用のスクリプト）を書く必要はありません。データ処理の段階で「冪等性（べきとうせい）」を保証しています。

ステップ2：埋め込み（Embedding）の抽出とFiftyOneへの登録

準備したビューを使って各ビデオファイルをループ処理で TwelveLabs にアップロードし、インデックス作成を依頼します。FiftyOneの iter_samples(autosave=True) を使用すると、データベース（DB）への保存をバッググラウンドでバッチ処理してくれるため、何千ものクリップを高速で処理したい時に、DBとのやり取りを最小限に抑えられます。

# Iterate through the filtered view
    for sample in label_view.iter_samples(autosave=True, progress=True):
        try:
            # Upload to TwelveLabs and save the returned Video ID to the sample
            sample["tl_video_id"] = index_video_to_twelvelabs(
                client, index_id, sample
            )
            print(f"  ✓ {sample.filename}")
        except Exception as e:
            print(f"  ✗ {sample.filename}: {e}")

インデックス作成が終われば、各ビデオに対応するMarengo 3.0の埋め込みベクトル（Embedding）を取得できます。これらは、ビデオの中に起きていること（視覚情報、音声情報、そしてそれらの関連性）が詰まったデータです。単体のフレーム（画像）のみに基づいた埋め込みモデル（一秒ごとに単純に評価するアプローチ）とは異なり、Marengoは時間の流れ（連続性）をしっかりキャプチャできます。例えば「作業員が安全ゴーグルを外した」という状況と「作業員が安全ゴーグルを装着した」状況は、切り抜かれた一瞬（フレーム）の画像は似てしまいますが、Marengoの埋め込み空間においては、状況が正反対として区別されます。

ステップ3：教師なしのセマンティック・クラスタリング

このプロジェクトの核となる部分です。事前定義されたカテゴリをまだ持っていなくても問題ありません。どの程度の分類タイプが含まれているか分からない状況であっても、抽出された埋め込みベクトル群に対してKMeansクラスタリングを適用すれば、似たもの同士がオートで自然にグループ分けされていきます。

def cluster_and_label(client, dataset, embeddings, num_clusters=8):
    print(f"Clustering {len(embeddings)} videos into {num_clusters} clusters...")

    # 1. Cluster embeddings
    kmeans = KMeans(n_clusters=num_clusters, random_state=0)
    cluster_labels = kmeans.fit_predict(embeddings)

    # 2. Map clusters to semantic labels using Pegasus (Generative AI)
    cluster_label_map = {}
    for cluster_idx in np.unique(cluster_labels):
         # We pick a representative video from the cluster and ask Pegasus to label it
         representative_video_id = get_video_id_for_cluster(cluster_idx)
         cluster_label_map[cluster_idx] = generate_label(client, representative_video_id)

    # 3. Batch update FiftyOne samples
    indexed_view = dataset.exists("tl_video_id")
    classifications = [Classification(label=cluster_label_map[c]) for c in cluster_labels]
    indexed_view.set_values("pred_cluster", classifications)

データの一斉書き込みをサポートする set_values() メソッドを使用しています。素朴にアプローチしてしまい、1つひとつのサンプルごとに sample.save() を繰り返すと、数千ものサンプルを処理する場合には途方に暮れる時間がかかります。大量のデータを扱う際に、このような細かな技術の選択がとても大きな違いを生み出すことになります。

なぜセマンティック・クラスタリングがこれほどスムーズに進むのでしょうか？ それは先ほどご紹介した通り、Marengoが動画を一連の文脈（マルチモーダルな表現）として捉えているからです。同じ「通路がフォークリフトで塞がれている」シーンがあったとして、それらが異なる時間帯や異なるアングル、周りのノイズが異なる状況で撮影されていたとしても、その深い意味（文脈）において情報が同じであるため、埋め込みスペースの中で自動的に同一グループとして近くに集まります。従来の単純な見かけ（ピクセル位置）の類似だけでは、別のカテゴリへ分散されてしまう状況を防ぐことができます。

ステップ4：Pegasusによるゼロショット自動ラベリング

確かに「クラスタ 3」のようなラベルは数学的に識別するには役立ちますが、運用（現場）をサポートする目的としては何の役にも立ちません。そこで、実用的な「フォークリフト通行規制違反」のような言葉へ変換するために、Pegasus 1.2の出番となります。

定義されたクラスタの中から「代表的なビデオ」を取り出し、Pegasusへ渡すようにプロンプトを構成します。Pegasusは、1フレームでの単純な画像情報や文字起こしだけではなく、映像の「流れ全体」を考慮に入れて、ラベルに相当する適切なテキストを返してくれます。これまでに、その具体的な監視映像そのものをPegasusに見せて学習させたことは一度もありません（＝ゼロショット）。それでも、決められた安全基準に合った表現を見事に提示してくれます。

CLUSTER_LABEL_PROMPT = """
Analyze this workplace safety video and classify it as exactly ONE of the following labels.
UNSAFE BEHAVIORS:
- Safe Walkway Violation
- Unauthorized Intervention
- Opened Panel Cover
...
Return ONLY the exact label name.
"""

上記のプロンプトを用いて、Pegasusが規定のカテゴリ以外の回答を出力しないように制限を仕掛けてあります。これは意図的に行っている選択です。なぜなら実際の工場運用においては、自由に入力されたテキストよりも、すでにある安全報告の規約（タクソノミー）と同じ分類にマッピングしたいからになります。利用する場合にはプロンプトをお使いの会社の安全マニュアルなどの定義項目に変更してあげるだけで完了します。モデルの再学習などは一切、必要ありません。

ステップ5：UMAPによる埋め込みデータの視覚的な理解

もちろんデータとラベルで機械的な仕分けは進みますが、実際に目で見て確認する（視覚化）ことはデータサイエンスや製品開発において極めて重要です。FiftyOneのBrainモジュールを使うと、多次元の複雑な情報を2次元のUMAPプロットとして描画してくれます。それぞれの点が「特定の動画」を示し、色がPegasusによって自動アサインされた安全カテゴリを示します。

この散布図チャートを読む際には次の点に着目してみましょう。きれいに固まっている箇所があるならば、それはAIモデルが「非常口のブロック」と「保護具ルール違反」を自信を持って仕分けできていることを示します。一方で、重なり合っている箇所があれば、おそらくその2つのカテゴリ表記（例えば、同じ腰痛リスクに対して「不適切な重量物の持ち上げ」と「エルゴノミクス違反」というカテゴリ設定など）があいまいであるために、境界線が引きづらくなっていることを示唆しています。また、他のクラスの中心部からぽつんと極端に外れて存在するような点（Outliers）は、イレギュラーな状況が発生していたり、ラベリングミスが発生している可能性が高いので、スタッフがそのビデオを直接確認することが推奨されます。

この可視化ステップは派手さ（デモ目的）のためだけでなく、モデルの作成（アノテーションの段階）における品質チェックのために本質的に重要なアプローチになります。ここでの「5分の目視確認」だけで、後の何時間ものモデル学習エラー対応のデバッグを防いでくれます。

ステップ6：PyTorchデータローダーの作成

きれいに自動分類できたデータは、実際にトレーニング（学習）に回せてこそ真価を発揮します。FiftyOneの to_torch() メソッドへ渡す GetItem カスタムクラスを用意し、あらかじめ作成しておいたEmbedding（埋め込み）情報とラベルの対応を素早くテンソル（Tensor）化してトレーニングフローに流しこみます。

from fiftyone.utils.torch import GetItem
import torch

class WorkerSafetyGetItem(GetItem):
    def __call__(self, d):
        return {
            "embedding": torch.tensor(d.get("tl_embedding"), dtype=torch.float32),
            "label_idx": torch.tensor(
                self.label_to_idx.get(d.get("ground_truth").label, -1),
                dtype=torch.long
            ),
        }

def create_dataloader(dataset, batch_size=4):
    indexed_view = dataset.exists("tl_video_id")
    
    # Create a PyTorch dataset directly from the FiftyOne view
    torch_dataset = indexed_view.to_torch(WorkerSafetyGetItem(LABEL_TO_IDX))
    
    return DataLoader(torch_dataset, batch_size=batch_size, shuffle=True)

これにより、すでに事前計算済みの映像の特徴量データと正解ラベルのセット（DataLoader）が高速に取得可能です。既にMarengoが抽出を完了しているデータを使うため、通常のコンピュータビジョン（画像認識）プロジェクトにおいてもっともコンピュータ資源と時間を食う「特徴量抽出（映像からピクセル情報をパースしていく部分）」のステップを完全にスキップして学習をスタートできます。これにより、この埋め込みベクトルの上に数レイヤーの非常に軽量な中間クラス（MLPや線形分類器など）を用意して学習（ファインチューニング）することが数秒で可能になります。また、ラベルはPegasusにより作成された高品質なものであるため、不正確なフォルダ分けによるミスの心配もありません。

サンプル用のデータセットは、こちらのGitHubリポジトリに用意されています。

これがなぜ大きな変革なのか：仕分けられたデータがビジネス知性に変わる理由

作成したこのパイプラインによって、組織に一体どのようなビジネス変革や、生産現場におけるインサイド（洞察）が得られるのか考えてみましょう。

超高速なモデル開発サイクル。 通常、生の監視映像からトレーニング済みモデルを作成しようとすれば、外部委託でのアノテーション作業の手配や、幾度もの手直しフィードバックの会議、膨大な時間・コストが必要になります。今回のこのパイプラインを活用すれば、1つのスクリプトを走らせるだけで、すでに埋め込みが作成済みのモデル向けラベリングデータを作成できます。安全管理を担当するチームが新しい課題へ対応する際のスピードが画期的に高速化されます。

組織課題（パターン化）の発見。 工場の安全管理責任者にとって、このデータのクラスタ図（ステップ 5）は、ただの「データ品質の検証ツール」にとどまりません。立派な「状況監視（ダッシュボード）」に進化します。仮にKMeansが「非常口のブロック、閉鎖」の箇所について異常に大きなデータ（クラスター）を作りだしていたら、それは一過性のインシデント（一回誰かがミスをして物を置いてしまった）だけではなく、現場（プロセス）に根本的な問題が存在していることを示します（単なる警告書やマニュアル改訂だけでは解決しない、構造的な配置問題などがあるはずです）。これを毎日、あるいは毎週自動で走らせるように設定すれば、工場の監視映像は「ただ有事に提出するだけの録画テープ（パッシブ型）」から「毎日状況改善を促すためのビジネスインサイド（アクティブ型）」に変貌します。インシデントカテゴリが時間経過とともに減っているかどうかや、安全対策の取り組みが実際の変化を生んでいるかの追跡、またインシデント密度が最も高いシフト（時間帯）や工場エリア（ゾーン）などを明確に数値で特定できます。

アノテーションコストの圧倒的削減。 安全要件（特に誤検知などが人命に関わる用途）での人手による動画アノテーション作業代は、一般的に検証対象ビデオ1時間あたり25〜50ドルの支払いが伴います。このパイプラインはそれを「API利用における計算リソース代」に代替してくれます。カメラが何十台もある大規模な工場をサポートするならば、その節約金額だけでもかなりの規模になります。さらに重要なことに、ラベリング結果のブレがなくなります。作業を依頼した担当者によって判定基準が揺れることがないため、一貫したデータ品質を手に入れることができます。

おわりに

このチュートリアルでは、「生の監視映像を取り出す」ところから「最終的に安全予測に対応したPyTorchデータローダーとして出力する」までの、すべて自動化された美しいパイプラインを作成しました。TwelveLabs社のMarengo 3.0が一番難しいところ（マルチモーダルにビデオから連続性をコードに落とし出す点）を担い、Pegasus 1.2がそれを現場に即した言葉にマッピングしました。そしてFiftyOneが、開発者がすぐに使えるように整理（可視化や出力）してくれました。

完全なコードは、こちらのGitHub リポジトリで公開されています。ぜひクローンし、ご自身の監視映像データで走らせ、目的のプロンプト（ラベリング定義）へ修正していろいろと試してみてください。

TwelveLabsを活用すれば、これ以外にも例えば「ビデオコンテンツ内検索」「コンプライアンス監視の自動制御」「映像の自動見せ場カット編集」など、多くのアプリケーションを開発することができます。詳細を知りたい場合は、ぜひTwelveLabs 公式ドキュメントへお越しいただくか、チームに直接連絡を頂き、皆さんの要件についてお聞かせください。

参考資料（リソース）

• TwelveLabs 公式ドキュメント

• FiftyOne 公式ドキュメント

• FiftyOne ビジュアライゼーション表示チートシート

• FiftyOne フィルタリング利用チートシート

• GitHubリポジトリ（お使いのコード）：visual-ai-worker-safety-kit