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