📷

Vision-Launguageモデルで走行データベースと動画検索システムを作る

2024/01/17に公開

Turing株式会社の自動運転チームでインターンしている東大B3の大野です。

自動運転チームでは、完全自動運転の実現を目指して自動運転AIを開発しています。モデル開発の際に、「雨の日に高速を走っていて先行車がいない」や「交差点で歩行者がいる中、右折している」など、特定の状況の走行データが必要になることがあります。

今回私は、動画に対して天気や歩行者の数などのラベルをデータベース化し、検索できるシステムを、Vision-Languageモデルを使って開発しました。この記事では、このシステムの作成にあたって取り組んだことについて説明します。

作成したGUI

課題

Turingでは、走行パートナーの方々とともに、大量の走行データを収集してきました。走行データには、車両に載せたカメラによる動画や、その際の車両のログ(速度やステアリング角、位置情報など)が含まれます。また、すべてのデータをAWSのS3で保管してきました。

これらの走行データを活用するために私達は以前、Vision-LanguageモデルとGPTを活用して、動画から切り出した画像を自由なテキストで意味検索ができるシステムを開発しました。たとえば、”traffic light”や”tunnel”などのテキストによる検索が出来ます。この動画検索システムによって、モデルの評価などを行う際の生産性が大きく向上しました。

https://zenn.dev/turing_motors/articles/ai-movie-searcher

しかし、この検索システムにはテキスト検索にしか対応しておらず、数値による絞り込みを併用できないという問題がありました。例えば「信号があって速度20km/h以下で走行するデータ」が欲しいとします。速度はログから検索できますが、信号機の有無は動画から推論するしかないためVision-Languageモデルの使用が必要です。しかし、Vision-Languageモデルを使った検索と、ログの数値データによる検索を併用する方法がありませんでした。また、このシステムはデータベースを使用しておらず、データを増やす場合に、手間がかかるという問題もありました。

そこで、天気や信号機の有無など画像から得られる情報の検索と、速度や位置など数値データによる検索を両立できるデータベースの開発を行うことになりました。

まずはテキスト検索データベースを作成した

以前開発した検索システムは、次のような仕組みでした。

  1. 動画から切り出した画像に対して、複数のVision-Langaugeモデルを使用して、キャプションを生成する。
  2. 1で生成した複数のキャプションをGPTに渡して文章化する。
  3. 文章をOpenAIのAPIでベクトル化する。
  4. 検索文も同様にベクトル化し、faissというライブラリを用いて最近傍探索を行う。

そこで、ステップ2で生成した説明テキストをそのままデータベースに入れ、ステップ3や4のベクトル化や最近傍探索はデータベース上で行うというアプローチを考えました。BigQueryでは最近、BigQuery上でテキストのベクトル化や距離計算ができる機能が公開されました。

https://cloud.google.com/bigquery/docs/text-embedding-semantic-search?hl=ja

この機能を利用し、次の手順でデータベースを作成しました。

  1. 画像のファイル名と、説明テキストをBigQueryにアップロードする。
  2. 説明テキストに対して、埋め込みベクトルを計算して、新しい列に保存する。
  3. 埋め込みベクトルに対して、k-meansクラスタを作成する。

このデータベースに対して、以下の手順で検索が行えます。

  1. 検索テキスト(例では”traffic light”)の埋め込みベクトルを計算する。
  2. そのベクトルが所属するk-meansクラスタのidを求める。
  3. そのクラスタに所属するベクトルのうち、検索テキストのベクトルと距離が近いものを求める。
WITH tar_text_embedding AS (
    SELECT
        text_embedding
    FROM
        ML.GENERATE_TEXT_EMBEDDING(
            MODEL `driving_data.embedding_model`,
            (
                select
                    'traffic light' AS content
            ),
            STRUCT(TRUE AS flatten_json_output)
        )
),
predict AS (
    SELECT
        centroid_id,
        text_embedding
    FROM
        ML.PREDICT(
            MODEL `driving_data.clustering`,
            (
                SELECT
                    text_embedding
                FROM
                    tar_text_embedding
            )
        )
)
SELECT
    c.id AS id,
    c.content AS content,
    c.centroid_id AS centroid_id,
    ML.DISTANCE(s.text_embedding, c.text_embedding, 'COSINE') AS distance
FROM
    predict AS s,
    `driving_data.image_description` AS c
WHERE
    s.centroid_id = c.centroid_id
ORDER BY
    distance ASC
LIMIT
    20

以前の検索システムでは検索時に、ベクトル生成のためにOpenAIのAPIを呼び出したり、ベクトルの近傍探索のライブラリや作成済インデックスを読み込んだり、とリソースが分散していましたが、BigQuery MLを使うことによって、検索のすべてをBigQuery上に集約できるようになりました。また、テキストと埋め込みベクトルを管理するテーブルとは別に、数値データを管理するテーブルを作成し、検索時にJoinすることによって、テキスト検索と同時に、数値データによる絞り込みも同時に行えるようになりました。

方針転換

しかし、テキスト検索には次のような問題がありました。

  1. 文脈のある長い検索テキストでないと、欲しいデータが得られない場合がある。
  2. テキスト生成で使用したモデルで検出できない項目で検索したくなった場合、モデルを変え、一からテキスト生成する必要が出る。
  3. 埋め込みベクトルはサイズが大きいため、データ数が増えた場合にクエリ料金が多く発生する。

まず、1について、例えば”people”で検索をすると、次のように、人の写っていない画像がヒットしてしまいました。これは、GPTが文章化する際に、歩道橋に対して”where people can safely cross the road”という修飾句をつけ、検索時に修飾句の”people”に引っかかったことが原因でした。また、文脈のある”car is stopping in front of traffic light”というテキストで検索しても、街灯のある駐車場に車が止まっている画像がヒットする場合がありました。

誤ってヒットした画像

このように、正しく画像を文章化しても検索に失敗するのは、使用した埋め込みモデルの性能の問題でもあり、以前の検索システムで使用したOpenAIのAPIを使用すれば一部は解決できます。ただ、BigQueryのクエリでそのまま使える埋め込みモデルの中では、十分の性能を持ったものはありませんでした。

次に、2について、以前の検索システムで文章化に使用したモデル(BLIP-2、GRiT)では、十分な精度で分類、検出できない項目がありました。例えば、信号機の色は検出できません。また、天気については”rainy”では十分に検索できず、”wet road”で道路が濡れている画像を検索する必要がありました。これらの項目について高い精度で検索したくなった場合は、新たなモデルを導入するだけでなく、その推論結果を元にテキストや埋め込みベクトルの生成を一からやり直す必要が出て、経済的・人的コストが生じます。

最後に3について、BigQueryはクエリで参照されるデータ量に対して、$5.00 per TBのクエリコストが生じるため、データサイズには配慮が必要です。埋め込みベクトルは、デフォルトモデルではは768次元です。現在のデータ数では問題ありませんが、もし10^8件程度のデータ数になった場合、1クエリあたり500円程度かかってしまいます。

そこで、事前に選定した検索したい項目ごとに、分類や検出を行って、その結果をデータベースに入れるという方法に方針転換しました。この方法であれば、次のようにテキスト検索の課題を解決することができます。

  1. 項目ごとにストレートに検索でき、検索テキストの出来に結果が左右されない。
  2. 項目の追加は、その項目に特化した新たなモデル、プロンプトを検討し、カラムを追加すればよく、今までに作成した項目を壊さなくて良い。
  3. 1項目あたりのデータは、boolかintでサイズが小さいので、クエリコスト、ストレージコストを抑えられる。

また、テキスト検索と比べて検索の自由度は下がりますが、テキスト検索にしても、運用上は限られたテキストでしか検索しないことがわかったので、項目を絞ってしまってもデメリットが小さいと考えました。

項目ごとに分類、検出方法を考える

検索したい項目として、実際に検索システムを使用するメンバーに、主に次のようなものに設定してもらいました。

  • 画像全体の分類
    • 天気:晴れ、曇り、雨
    • 場所:市街地、高速道路、トンネル、駐車場
    • 逆光かどうか
    • 交差点にいるかどうか
    • 混雑しているかどうか
  • 物体検知
    • 歩行者の有無、多さ
    • ラバーポールの有無
    • 信号機、道路標識の有無
  • 物体検知+分類
    • 車線分類:中央線の有無、色、薄さ
    • 先行車:先行車の有無、先行車の種類

まず、項目ごとに、どのように分類、検出できるか考えました。はじめは、ResNet-50やEfficientDetなどのフレームワークを用いて、モデルの訓練をすることを検討しました。しかし、天気、歩行者、信号機、道路標識以外の項目に関しては、条件を満たす既存データセットが存在せず自力でアノテーションが必要ななため、断念しました。

そこで、既存のVision Language系のモデルを使用し、プロンプトの工夫によって、分類や検出を行うことにしました。実際に使用したモデルを説明します。

BLIP

BLIPとは、Image CaptioningやVisual Question Answering、Image-text matchingなど、様々なマルチモーダルタスクを行えるフレームワークです。今回はVisual Question Answeringを使用しました。

Hugging FaceのTransformersライブラリに含まれているため、次のように簡単に使えます。また、パラメータ数が385Mのモデルを使用したため、メモリが16Gのノートパソコン(CPU)でも、高速に動かすことが出来ました。

from transformers import AutoProcessor, BlipForQuestionAnswering

model = BlipForQuestionAnswering.from_pretrained("Salesforce/blip-vqa-base")
processor = AutoProcessor.from_pretrained("Salesforce/blip-vqa-base")

def blip_inference(image, prompt: str) -> str:
    inputs = processor(images=image, text=prompt, return_tensors="pt")
    outputs = model.generate(**inputs)
    result = processor.decode(outputs[0], skip_special_tokens=True)
    return result.lower()

例えば、混雑しているかは、"Is the road crowed?”というプロンプトで分類できます。混雑しているを真とした場合のPrecisionは97%でした。

prompt = "Is the road crowed?"
ans = blip_inference(image, prompt)
print(ans) # yes/no

yes
no

また、細かい物体の数を数えることもできます。次のプロンプトでは、ラバーポールの数を数えています。

prompt = "Is there red and white striped poles on the road?"
ans = blip_inference(image, prompt)
print(ans) # number

answer = 15

answer = 0

また、検出したものに対して、分類することもできます。次のプロンプトでは、中央線の色を分類しています。

prompt = "Is there a center line on the road?"
ans = blip_inference(image, prompt)
print(ans) # yes / no

prompt = "What is the color of the center line on the road?"
ans = blip_inference(image, prompt)
print(ans) # white / yellow

中央線があるか: yes, 色: yellow

中央線があるか: yes, 色: white

中央線があるか: no

このように、どの項目に関しても、そこそこの精度で分類、検出することができました。しかし、100%の精度で分類、検出できるわけではありません。画像全体の分類系では、場所は「高速」、天気は「雨」に過度に分類される傾向がありました。また、物体検出系に関しては、過検出がたまに起こりました。特に「雨」、「逆光」、「ラバーポール」に関しては、データセット全体に対する、Trueデータの割合が小さいため、過検出の割合が10%程度でも、検索結果のうちTrue画像の割合が3割以下になってしまうという問題が生じました。検索する側の気持ちとしては、「ほしい項目のTrueデータの割合が小さいからこそ、検索で取得したい」ので改善が必要です。そこで、物体検出系はDetic、場所・天気の分類はCLIPという別のモデルを使用しました。

CLIP

CLIPとは、image-text similarityやZero-shot画像分類を行えるマルチモーダルモデルです。

CLIPに関しても、BLIPと同様、Hugging FaceのTransformersライブラリに含まれているため、次のように簡単に使えます。また、メモリが16GBのノートパソコン(CPU)でも、高速に動かすことが出来ました。

from transformers import CLIPProcessor, CLIPModel

model = CLIPModel.from_pretrained("openai/clip-vit-base-patch32")
processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")

def clip_inference(image, texts: list[str]) -> list[float]:
    inputs = processor(text=texts, images=image,
                       return_tensors="pt", padding=True)
    outputs = model(**inputs)
    logits_per_image = outputs.logits_per_image
    probs = logits_per_image.softmax(dim=1)
    return probs.tolist()[0]

例えば、場所は、次のようなプロンプトで分類できます。

prompts = ["the photo in highway", "the photo in tunnel",
           "the photo in parking lot", "the photo in urban area"]
probs = clip_inference(image, prompts)
print(probs)

次の画像では、[0.0857, 0.9002, 0.0099, 0.0041]が出力され、トンネルで取られた画像である可能性が一番高いと、判定できました。

トンネル

CLIPでもBLIPと同様、高速だと判定されやすい傾向がありました。ただ、CLIPは確率が得られるので、ラベルごとに閾値を設定して、分類結果を調整しました。その結果、市街地は100%、高速道路、トンネルは90%、駐車場は75%のPrecisionを達成できました。

if probs[0] > 0.9:
   return Place.HIGHWAY
elif probs[1] > 0.7:
   return Place.TUNNEL
elif probs[2] > 0.7:
   return Place.PARKING
elif probs[3] > 0.25:
   return Place.STREET
else:
   return Place.OTHER

天気に関しても同様に実装し、BLIPではPrecisionが低かった雨でも、約90%のPrecisionを達成しました。

このように、CLIPでは、画像全体の特徴をつかめ、さらに閾値を調整できるというメリットがあり、場所と天気の分類に利用できました。一方で、中央線や人などの検出や、混雑状況など、画像の一部分の認識が必要なものの分類はできなかったので、BLIPやDeticに任せました。

Detic

Deticは物体検出モデルで、CLIPと組み合わせることによって、画像中の任意のクラスを、テキストを使って検出すること(Zero-Shot物体検出)ができます。

Deticは次のように使うことが出来ます。BLIP、CLIPと違って、Transformerライブラリに含まれていないので、こちらに従って、環境構築が必要です。CPUでも動かせますが、1枚あたり2~3秒かかったので、GPUで動かしました。

また、公式のdemo.pyを使用すると画像に対する検出結果の可視化が行われます。今回必要だったのは可視化結果ではなく検出個数であったので、こちらのスクリプトを作成しました。

from detectron2.data import MetadataCatalog, DatasetCatalog
from detectron2.config import get_cfg
from detectron2.engine.defaults import DefaultPredictor
from centernet.config import add_centernet_config
from detic.config import add_detic_config
from detic.modeling.utils import reset_cls_test
from detic.modeling.text.text_encoder import build_text_encoder
import sys
import time
sys.path.insert(0, 'third_party/CenterNet2/')

def get_clip_embeddings(vocabulary, prompt='a '):
    text_encoder = build_text_encoder(pretrain=True)
    text_encoder.eval()
    texts = [prompt + x for x in vocabulary]
    emb = text_encoder(texts).detach().permute(1, 0).contiguous().cpu()
    return emb

def get_detic_predictor(custom_vocabulary):
    cfg = get_cfg()
    add_centernet_config(cfg)
    add_detic_config(cfg)
    cfg.merge_from_file(
        "configs/Detic_LCOCOI21k_CLIP_SwinB_896b32_4x_ft4x_max-size.yaml")
    cfg.MODEL.WEIGHTS = 'https://dl.fbaipublicfiles.com/detic/Detic_LCOCOI21k_CLIP_SwinB_896b32_4x_ft4x_max-size.pth'
    cfg.MODEL.ROI_HEADS.SCORE_THRESH_TEST = 0.5  # set threshold for this model
    cfg.MODEL.ROI_BOX_HEAD.ZEROSHOT_WEIGHT_PATH = 'rand'
    cfg.MODEL.ROI_HEADS.ONE_CLASS_PER_PROPOSAL = True
    # cfg.MODEL.DEVICE='cpu' # uncomment this to use cpu-only mode.
    predictor = DefaultPredictor(cfg)
    metadata = MetadataCatalog.get(str(time.time()))

    metadata.thing_classes = custom_vocabulary.split(',')
    classifier = get_clip_embeddings(metadata.thing_classes)
    num_classes = len(metadata.thing_classes)
    reset_cls_test(predictor.model, classifier, num_classes)
    return predictor

各項目について、次のようなプロンプトを使用しました。ラバーポールに関しては、日本に特有のためか、”rubber pole”では検出できないため、”red and white striped pole”というプロンプトを使用しました。


person_predictor = get_detic_predictor("person")
rubber_pole_predictor = get_detic_predictor("red and white striped pole")
traffic_light_predictor = get_detic_predictor("traffic light")
traffic_sign_predictor = get_detic_predictor("traffic sign")

def people(image) -> int:
    res = person_predictor(image)
    return len(res["instances"])

def rubber_pole(image) -> int:
    res = rubber_pole_predictor(image)
    return len(res["instances"])

def traffic_light(image) -> int:
    res = traffic_light_predictor(image)
    return len(res["instances"])

def traffic_sign(image) -> int:
    res = traffic_sign_predictor(image)
    return len(res["instances"])

見栄えのため、検出件数の代わりに可視化結果を貼ります。人、信号機、道路標識に関しては、ほぼ完璧に検出できました。ラバーポールに関しては、検出漏れはありますが、1画像中に複数ラバーポールがある場合、そのうちの少なくとも1つは検出できる傾向にあったため、画像中の有無の判定には問題がないと判断しました。交通標識に関しても、すべての種類の交通標識を認識できるわけではありませんが、検索上は問題ない程度の検出ができました。


ラバーポール
交通標識

また、どの項目も過検出が極めて稀だったため、Trueデータの割合が小さいラバーポールでも、約90%のPrecisionを達成できました。また、白線の検知はできなかったので、BLIPに任せました。

結果

精度を評価した結果、各項目について次の方法をとりました。また、逆光など、十分なPrecisionを得られなかった項目については、一旦断念しました。今後、いいモデルやプロンプトを思いついたら、データベースに追加する予定です。

  • 画像全体の分類
    • 天気:CLIP
    • 場所:CLIP
    • 逆光:断念
    • 交差点にいるかどうか:BLIP
    • 混雑しているかどうか:BLIP
  • 物体検知
    • 歩行者の有無、多さ:Detic
    • ラバーポールの有無:Detic
    • 信号機、道路標識の有無:Detic
  • 物体検知+分類
    • 車線分類:中央線の有無、色はBLIP。薄さは断念。
    • 先行車:先行車の有無はBLIP。先行車の種類の分類は断念。

BigQueryにアップロード

約5万件の動画から、10sごとに、1枚あたり約5枚のフレームを切り出し、約25万枚画像に対して、BLIP、CLIP、Deticによる推論を実行しました。そして、その結果をBigQueryにアップロードしました。

job_config = bigquery.LoadJobConfig(
    schema=[
        bigquery.SchemaField("file_id", "STRING", mode="NULLABLE"),
        bigquery.SchemaField("frame_id", "INTEGER", mode="NULLABLE"),
        bigquery.SchemaField("weather", "INTEGER", mode="NULLABLE"),
        bigquery.SchemaField("place", "INTEGER", mode="NULLABLE"),
        bigquery.SchemaField("center_line_color", "INTEGER", mode="NULLABLE"),
        bigquery.SchemaField("time_zone", "BOOLEAN", mode="NULLABLE"),
        bigquery.SchemaField("congestion", "BOOLEAN", mode="NULLABLE"),
        bigquery.SchemaField("leading_car", "BOOLEAN", mode="NULLABLE"),
        bigquery.SchemaField("junction", "BOOLEAN", mode="NULLABLE"),
        bigquery.SchemaField("people", "INTEGER", mode="NULLABLE"),
        bigquery.SchemaField("rubber_pole", "INTEGER", mode="NULLABLE"),
        bigquery.SchemaField("traffic_light", "INTEGER", mode="NULLABLE"),
        bigquery.SchemaField("traffic_sign", "INTEGER", mode="NULLABLE")
    ],
    source_format=bigquery.SourceFormat.CSV,
)
with open(file_path, "rb") as source_file:
    load_job = client.load_table_from_file(
        source_file,
        table_id,
        job_config=job_config,
    )
print(load_job.result())

また、数値データに関しても、必要な項目をパースして、BigQueryにアップロードしました。これによって、ログから検索できる定量的な情報と、本来人が見ないとわからない情報がひとつのデータベースで扱えるようなって利便性が向上しました。

GUIで使いやすくする

BigQueryには、ファイル名、frame idと、各項目のラベル(推論結果)のみを格納しているため、検索後に対象の画像や動画にアクセスするには、手動でS3から対象ファイルを検索してダウンロードする必要があって、手間がかかりました。

そこで、クエリ、S3からの画像の取得、S3からの動画のダウンロードリンクの取得を自動化するGUIをstreamlitで作成し、使い勝手を良くしました。また、EC2インスタンス上で動かし、社内のWiFiからアクセスできるようにして、利用者の環境構築を不要にしました。

作成したGUI

まとめ

動画検索システムについて、項目ごとにモデル、プロンプトを調整して分類、検出をおこない、データベース及び検索用GUIを作成した例について紹介しました。項目の追加や変更が簡単なので、今後新たな分類方法を思いついたり、モデルの精度が向上したときに、検索システムを進化させていけそうです。

Turing では自動運転モデルの学習や、自動運転を支える基盤モデルの作成、EV開発などを、これからも熱く取り組んでいきます。興味がある方は、Turing の公式 Web サイト採用情報などをご覧ください。

Tech Blog - Turing

Discussion