CLIP(Contrastive Language-Image
Pre-training)是一種機器學習技術，它可以準確理解和分類圖像和自然語言文本，這對圖像和語言處理具有深遠的影響，并且已經被用作流行的擴散模型DALL-E的底層機制。在這篇文章中，我們將介紹如何調整CLIP來輔助視頻搜索。
這篇文章將不深入研究CLIP模型的技術細節(jié)，而是展示CLIP的另外一個實際應用(除了擴散模型外)。
首先我們要知道：CLIP使用圖像解碼器和文本編碼器來預測數(shù)據集中哪些圖像與哪些文本是匹配的。
使用CLIP進行搜索
通過使用來自hugging
face的預訓練CLIP模型，我們可以構建一個簡單而強大的視頻搜索引擎，并且具有自然語言能力，而且不需要進行特征工程的處理。
我們需要用到以下的軟件
Python≥= 3.8，ffmpeg，opencv
通過文本搜索視頻的技術有很多。我們可以將搜索引擎將由兩部分組成，索引和搜索。
索引
視頻索引通常涉及人工和機器過程的結合。人類通過在標題、標簽和描述中添加相關關鍵字來預處理視頻，而自動化過程則是提取視覺和聽覺特征，例如物體檢測和音頻轉錄。用戶交互指標等等，這樣可以記錄視頻的哪些部分是最相關的，以及它們保持相關性的時間。所有這些步驟都有助于創(chuàng)建視頻內容的可搜索索引。
索引過程的概述如下
- 將視頻分割成多個場景
- 為框架取樣場景
- 幀處理后進行像素嵌入
- 索引建立存儲
將視頻分成多個場景
為什么場景檢測很重要?視頻由場景組成，而場景由相似的幀組成。如果我們只對視頻中的任意場景進行采樣，可能會錯過整個視頻中的關鍵幀。
所以我們就需要準確地識別和定位視頻中的特定事件或動作。例如，如果我搜索“公園里的狗”，而我正在搜索的視頻包含多個場景，例如一個男人騎自行車的場景和一個公園里的狗的場景，場景檢測可以讓我識別出與搜索查詢最接近的場景。
可以使用“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)
下一步就是平均同一場景中的所有樣本，這樣可以降低樣本的維數(shù)，而且還可以解決單個樣本中存在噪聲的問題。
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 個獨立的索引：
- 視頻場景索引：哪些場景屬于特定視頻
- 場景嵌入索引：保存特定的場景數(shù)據
- 視頻元數(shù)據索引：保存視頻的元數(shù)據。
我們將首先將視頻中所有計算出的元數(shù)據以及視頻的唯一標識符，插入到元數(shù)據索引中，這一步都是現(xiàn)成的，非常簡單。
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,
})
然后在場景嵌入索引中創(chuàng)建一個新條目保存視頻中的每個像素嵌入，還需要一個唯一的標識符來識別每個場景。
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)
搜索
現(xiàn)在我們有了一種將視頻的索引，下面就可以根據模型輸出對它們進行搜索和排序。
第一步需要遍歷場景索引中的所有記錄。然后，創(chuàng)建一個視頻中所有視頻和匹配場景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)
然后訪問模型輸出中的“l(fā)ogits_per_image”獲得模型的輸出。
Logits本質上是對網絡的原始非標準化預測。由于我們只提供一個文本字符串和一個表示視頻中的場景的張量，所以logit的結構將是一個單值預測。
logits_per_image = outputs.logits_per_image
probs = logits_per_image.squeeze()
prob_for_tensor = probs.item()
將每次迭代的概率相加，并在運算結束時將其除以張量的總數(shù)來獲得視頻的平均概率。
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)
最后在得到每個視頻的概率并對概率進行排序后，返回請求的搜索結果數(shù)目。
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']
})
就是這樣!現(xiàn)在就可以輸入一些視頻并測試搜索結果。
總結
通過CLIP可以輕松地創(chuàng)建一個頻搜索引擎。使用預訓練的CLIP模型和谷歌的LevelDB，我們可以對視頻進行索引和處理，并使用自然語言輸入進行搜索。通過這個搜索引擎使用戶可以輕松地找到相關的視頻，最主要的是我們并不需要大量的預處理或特征工程。
那么我們還能有什么改進呢?
- 使用場景的時間戳來確定最佳場景。
- 修改預測讓他在計算集群上運行。
- 使用向量搜索引擎，例如Milvus 替代LevelDB
- 在索引的基礎上建立推薦系統(tǒng)
- 等等
最后：可以在這里找到本文的代碼:https://github.com/GuyARoss/CLIP-video-search/tree/article-01。
以及這個修改版本:https://github.com/GuyARoss/CLIP-video-search。
