使用CLIP構建視頻搜索引擎

作者：Guy Ross 2023-01-03 15:42:29

CLIP(Contrastive Language-Image Pre-training)是一種機器學習技術，它可以準確理解和分類圖像和自然語言文本，這對圖像和語言處理具有深遠的影響，并且已經被用作流行的擴散模型DALL-E的底層機制。在這篇文章中，我們將介紹如何調整CLIP來輔助視頻搜索。

這篇文章將不深入研究CLIP模型的技術細節，而是展示CLIP的另外一個實際應用(除了擴散模型外)。

首先我們要知道：CLIP使用圖像解碼器和文本編碼器來預測數據集中哪些圖像與哪些文本是匹配的。

使用CLIP進行搜索

通過使用來自hugging face的預訓練CLIP模型，我們可以構建一個簡單而強大的視頻搜索引擎，并且具有自然語言能力，而且不需要進行特征工程的處理。

我們需要用到以下的軟件

Python≥= 3.8，ffmpeg，opencv

通過文本搜索視頻的技術有很多。我們可以將搜索引擎將由兩部分組成，索引和搜索。

索引

視頻索引通常涉及人工和機器過程的結合。人類通過在標題、標簽和描述中添加相關關鍵字來預處理視頻，而自動化過程則是提取視覺和聽覺特征，例如物體檢測和音頻轉錄。用戶交互指標等等，這樣可以記錄視頻的哪些部分是最相關的，以及它們保持相關性的時間。所有這些步驟都有助于創建視頻內容的可搜索索引。

索引過程的概述如下

將視頻分割成多個場景
為框架取樣場景
幀處理后進行像素嵌入
索引建立存儲

將視頻分成多個場景

為什么場景檢測很重要?視頻由場景組成，而場景由相似的幀組成。如果我們只對視頻中的任意場景進行采樣，可能會錯過整個視頻中的關鍵幀。

所以我們就需要準確地識別和定位視頻中的特定事件或動作。例如，如果我搜索“公園里的狗”，而我正在搜索的視頻包含多個場景，例如一個男人騎自行車的場景和一個公園里的狗的場景，場景檢測可以讓我識別出與搜索查詢最接近的場景。

可以使用“scene detect”python包來進行這個操作。

 mport scenedetect as sd
 
 video_path = '' # path to video on machine
 
 video = sd.open_video(video_path)
 sm = sd.SceneManager()
         
 sm.add_detector(sd.ContentDetector(threshold=27.0))
 sm.detect_scenes(video)
 
 scenes = sm.get_scene_list()

對場景的幀進行采樣

然后就需要使用cv2對視頻進行幀采樣。

 import cv2
 
 cap = cv2.VideoCapture(video_path)
 
 every_n = 2 # number of samples per scene
 
 scenes_frame_samples = []    
 for scene_idx in range(len(scenes)):
    scene_length = abs(scenes[scene_idx][0].frame_num - scenes[scene_idx][1].frame_num)
    every_n = round(scene_length/no_of_samples)
    local_samples = [(every_n * n) + scenes[scene_idx][0].frame_num for n in range(3)]
             
    scenes_frame_samples.append(local_samples)

將幀轉換為像素嵌入

在收集樣本之后，我們需要將它們計算成CLIP模型可用的東西。

首先需要將每個樣本轉換為圖像張量嵌入。

 from transformers import CLIPProcessor
 from PIL import Image
 
 clip_processor = CLIPProcessor.from_pretrained("openai/clip-vit-base-patch32")
 
 def clip_embeddings(image):
    inputs = clip_processor(images=image, return_tensors="pt", padding=True)
    input_tokens = {
        k: v for k, v in inputs.items()
    }
 
    return input_tokens['pixel_values']
 
 # ...
 scene_clip_embeddings = [] # to hold the scene embeddings in the next step
 
 for scene_idx in range(len(scenes_frame_samples)):
    scene_samples = scenes_frame_samples[scene_idx]
 
    pixel_tensors = [] # holds all of the clip embeddings for each of the samples
    for frame_sample in scene_samples:
        cap.set(1, frame_sample)
        ret, frame = cap.read()
        if not ret:
            print('failed to read', ret, frame_sample, scene_idx, frame)
            break
 
          pil_image = Image.fromarray(frame)
                 
          clip_pixel_values = clip_embeddings(pil_image)
          pixel_tensors.append(clip_pixel_values)

下一步就是平均同一場景中的所有樣本，這樣可以降低樣本的維數，而且還可以解決單個樣本中存在噪聲的問題。

 import torch
 import uuid
 
 def save_tensor(t):
    path = f'/tmp/{uuid.uuid4()}'
    torch.save(t, path)
 
    return path
 
 # ..
 avg_tensor = torch.mean(torch.stack(pixel_tensors), dim=0)
 scene_clip_embeddings.append(save_tensor(avg_tensor))

這樣就獲得了一個CLIP嵌入的表示視頻內容的的張量列表。

存儲索引

對于底層索引存儲，我們使用LevelDB(LevelDB是由谷歌維護的鍵/值庫)。我們搜索引擎的架構將包括 3 個獨立的索引：

視頻場景索引：哪些場景屬于特定視頻
場景嵌入索引：保存特定的場景數據
視頻元數據索引：保存視頻的元數據。

我們將首先將視頻中所有計算出的元數據以及視頻的唯一標識符，插入到元數據索引中，這一步都是現成的，非常簡單。

 import leveldb
 import uuid
 
 def insert_video_metadata(videoID, data):
    b = json.dumps(data)
 
    level_instance = leveldb.LevelDB('./dbs/videometadata_index')
    level_instance.Put(videoID.encode('utf-8'), b.encode('utf-8'))
 
 # ...
 video_id = str(uuid.uuid4())
 insert_video_metadata(video_id, {
    'VideoURI': video_path,
 })

然后在場景嵌入索引中創建一個新條目保存視頻中的每個像素嵌入，還需要一個唯一的標識符來識別每個場景。

import leveldb
 import uuid
 
 def insert_scene_embeddings(sceneID, data):
    level_instance = leveldb.LevelDB('./dbs/scene_embedding_index')
    level_instance.Put(sceneID.encode('utf-8'), data)
 
 # ...
 for f in scene_clip_embeddings:
    scene_id = str(uuid.uuid4())
     
    with open(f, mode='rb') as file:
        content = file.read()
             
        insert_scene_embeddings(scene_id, content)

最后，我們需要保存哪些場景屬于哪個視頻。

 import leveldb
 import uuid
 
 def insert_video_scene(videoID, sceneIds):
    b = ",".join(sceneIds)
     
    level_instance = leveldb.LevelDB('./dbs/scene_index')
    level_instance.Put(videoID.encode('utf-8'), b.encode('utf-8'))
 
 # ...
 scene_ids = []
 for f in scene_clip_embeddings:
    # .. as shown in previous step
    scene_ids.append(scene_id)
    scene_embedding_index.insert(scene_id, content)
 
 scene_index.insert(video_id, scene_ids)

搜索

現在我們有了一種將視頻的索引，下面就可以根據模型輸出對它們進行搜索和排序。

第一步需要遍歷場景索引中的所有記錄。然后，創建一個視頻中所有視頻和匹配場景id的列表。

 records = []
 
 level_instance = leveldb.LevelDB('./dbs/scene_index')
 
 for k, v in level_instance.RangeIter():    
    record = (k.decode('utf-8'), str(v.decode('utf-8')).split(','))
    records.append(record)

下一步需要收集每個視頻中存在的所有場景嵌入張量。

import leveldb

def get_tensor_by_scene_id(id):
    level_instance = leveldb.LevelDB('./dbs/scene_embedding_index')
    b = level_instance.Get(bytes(id,'utf-8'))

    return BytesIO(b)

for r in records:
    tensors = [get_tensor_by_scene_id(id) for id in r[1]]

在我們有了組成視頻的所有張量之后，我們可以把它傳遞到模型中。該模型的輸入是“pixel_values”，表示視頻場景的張量。

import torch
from transformers import CLIPProcessor, CLIPModel

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

inputs = processor(text=text, return_tensors="pt", padding=True)

for tensor in tensors:
    image_tensor = torch.load(tensor)
    inputs['pixel_values'] = image_tensor   
    outputs = model(**inputs)

然后訪問模型輸出中的“logits_per_image”獲得模型的輸出。

Logits本質上是對網絡的原始非標準化預測。由于我們只提供一個文本字符串和一個表示視頻中的場景的張量，所以logit的結構將是一個單值預測。

logits_per_image = outputs.logits_per_image    
probs = logits_per_image.squeeze()

prob_for_tensor = probs.item()

將每次迭代的概率相加，并在運算結束時將其除以張量的總數來獲得視頻的平均概率。

def clip_scenes_avg(tensors, text):
    avg_sum = 0.0

    for tensor in tensors:
        # ... previous code snippets
        probs = probs.item()
        avg_sum += probs.item()
   
    return avg_sum / len(tensors)

最后在得到每個視頻的概率并對概率進行排序后，返回請求的搜索結果數目。

import leveldb
import json

top_n = 1 # number of search results we want back

def video_metadata_by_id(id):
    level_instance = leveldb.LevelDB('./dbs/videometadata_index')
    b = level_instance.Get(bytes(id,'utf-8'))
    return json.loads(b.decode('utf-8'))

results = []
for r in records:
    # .. collect scene tensors
  
    # r[0]: video id
    return (clip_scenes_avg, r[0]) 

sorted = list(results)
sorted.sort(key=lambda x: x[0], reverse=True)
            
results = []
for s in sorted[:top_n]:
    data = video_metadata_by_id(s[1])
                
    results.append({
        'video_id': s[1],
        'score': s[0],
        'video_uri': data['VideoURI']
     })

就是這樣!現在就可以輸入一些視頻并測試搜索結果。

總結

通過CLIP可以輕松地創建一個頻搜索引擎。使用預訓練的CLIP模型和谷歌的LevelDB，我們可以對視頻進行索引和處理，并使用自然語言輸入進行搜索。通過這個搜索引擎使用戶可以輕松地找到相關的視頻，最主要的是我們并不需要大量的預處理或特征工程。

那么我們還能有什么改進呢?

使用場景的時間戳來確定最佳場景。
修改預測讓他在計算集群上運行。
使用向量搜索引擎，例如Milvus 替代LevelDB
在索引的基礎上建立推薦系統
等等

最后：

可以在這里找到本文的代碼:https://github.com/GuyARoss/CLIP-video-search/tree/article-01。

以及這個修改版本:https://github.com/GuyARoss/CLIP-video-search。

責任編輯：華軒來源： DeepHub IMBA

機器學習視頻搜索