关于人工智能:5分钟搭建一个粗粒度视频去重系统

Jupyter Notebook 教程: How to Build a Video Deduplication System

「视频去重」能够在海量的视频数据中实现侵权片段或者删除掉反复冗余的内容。随着抖音、快手、Bilibili 等视频平台的衰亡和火爆，相似视频这样的非结构化数据在数量上有了极大的增长。

视频平台存储着大量的视频资源，但其中会蕴含很多反复的视频数据。视频资源的冗余会带来两个问题：

1. 反复的视频会占用大部分的存储空间。据统计，一个未压缩的时长 1 分钟的 4k 视频可能须要 40GB 的存储空间。
2. 反复的视频资源不利于用户的体验。视频平台通常会利用举荐算法将视频推送给用户，然而当视频资源重复时，用户会被屡次推送高度类似、甚至完全相同的视频内容。

因而，为了更无效地治理视频和晋升用户体验，「视频去重」是视频平台必不可少的一项工作！

这篇文章将教你如何利用 Milvus 和 Towhee 搭建一个粗粒度「视频去重」零碎！该零碎的外围思路就是通过 Towhee 利用预训练的神经网络模型提取视频的特征向量，并将其存储在向量数据库 Milvus 中，而后 比拟查问对象的特征向量与数据库中的向量，从而实现判断视频之间的相似性。

01

装置工具包

在开始之前，咱们须要装置环境所依赖的包。咱们用到了以下工具：

• Towhee : 用于构建模型推理流水线的框架，对于老手十分敌对。
• Milvus : 用于存储向量并创立索引的数据库，简略好上手
• Pillow：图像处理罕用的 Python 库。
• Pandas：一个基于 Python 的疾速、灵便且易于应用的开源数据分析和操作工具。

python -m pip install -q pymilvus towhee pillow pandas ipython

02

筹备数据集

咱们在这里选用了 VCDB core dataset 的子集作为咱们的数据。VCDB 是一个罕用于「视频去重」工作的数据集，蕴含了超过 10 万个 Web 视频，以及 9,000 多个手动找到的复制片段对。

该数据集由两局部组成：VCDB core dataset 和 VCDB background dataset。其中 core dataset 的数据来自在 YouTube 和 MetaCafe，选自 28 个精心筛选的查问，共计 528 个视频，总时长大概 27 小时。通过大量的人工筛选，数据集中收录了 9236 对局部视频的正本。这些正本比照原视频通过了不同的变换模式，包含“插入图案”、“录像”、“比例变动”、“画中画”等。

咱们从中选取的视频形容了 20 个事件，每个事件蕴含大概 5 个内容雷同或类似的视频，总共约占 1.3G。

首先，咱们下载并解压数据：

curl -L https://github.com/towhee-io/examples/releases/download/data/VCDB_core_sample.zip -O
unzip -q -o VCDB_core_sample.zip

而后，咱们利用以下代码简略地察看这些视频：

import random
from pathlib import Path
import torch
import pandas as pd
random.seed(6)

root_dir = './VCDB_core_sample'


min_sample_num = 5
sample_folder_num = 20

all_video_path_lists = []
all_video_path_list = []

df = pd.DataFrame(columns=('path','event','id'))
query_df = pd.DataFrame(columns=('path','event','id'))

video_idx = 0
for i, mid_dir_path in enumerate(Path(root_dir).iterdir()):
    if i >= sample_folder_num:
        break
    if mid_dir_path.is_dir():
        path_videos = list(Path(mid_dir_path).iterdir())
        if len(path_videos) < min_sample_num:
            print('len(path_videos) < min_sample_num, continue.')
            continue
        sample_video_path_list = random.sample(path_videos, min_sample_num)
        all_video_path_lists.append(sample_video_path_list)
        all_video_path_list += [str(path) for path in sample_video_path_list]
        for j, path in enumerate(sample_video_path_list):
            video_idx += 1
            if j == 0:
                query_df = query_df.append(pd.DataFrame({'path': [str(path)],'event':[path.parent.stem],'id': [video_idx]}),ignore_index=True)
            df = df.append(pd.DataFrame({'path': [str(path)],'event':[path.parent.stem],'id': [video_idx]}),ignore_index=True)

all_sample_video_dicts = []
for i, sample_video_path_list in enumerate(all_video_path_lists):
    anchor_video = sample_video_path_list[0]
    pos_video_path_list = sample_video_path_list[1:]
    neg_video_path_lists = all_video_path_lists[:i] + all_video_path_lists[i + 1:]
    neg_video_path_list = [neg_video_path_list[0] for neg_video_path_list in neg_video_path_lists]
    all_sample_video_dicts.append({
        'anchor_video': anchor_video,
        'pos_video_path_list': pos_video_path_list,
        'neg_video_path_list': neg_video_path_list
    })

id2event = df.set_index(['id'])['event'].to_dict()
id2path = df.set_index(['id'])['path'].to_dict()

df_csv_path = 'video_info.csv'
query_df_csv_path = 'query_video_info.csv'
df.to_csv(df_csv_path)
query_df.to_csv(query_df_csv_path)
df

以上代码返回了一个 95 x 3 的表格，其中第一列是视频门路（path）、第二列是视频形容的事件（event）、第三列是视频编号（id）：

咱们将数据集中的视频分为基准视频（Anchor video）、指标视频（positive video）、非指标视频（negative video），并以 GIF 的模式展现：

random_video_pair = random.sample(all_sample_video_dicts, 1)[0]
neg_sample_num = min(5, sample_folder_num)
anchor_video = random_video_pair['anchor_video']
anchor_video_event = anchor_video.parent.stem
pos_video_list = random_video_pair['pos_video_path_list']
pos_video_list_events = [path.parent.stem for path in pos_video_list]
neg_video_list = random_video_pair['neg_video_path_list'][:neg_sample_num]
neg_video_list_events = [path.parent.stem for path in neg_video_list]

show_video_list = [str(anchor_video)] + [str(path) for path in pos_video_list] + [str(path) for path in neg_video_list]
# print(show_video_list)
caption_list = ['anchor video:' + anchor_video_event] + ['positive video' + str(i + 1) for i in range(len(pos_video_list))] + ['negative video' + str(i + 1) + ':' + neg_video_list_events[i] for i in range(len(neg_video_list))]
print(caption_list)
tmpdirname = './tmp_gifs'
display_gifs_from_video(show_video_list, caption_list, tmpdirname=tmpdirname)

比方咱们选取电影《援救大兵瑞恩》（‘saving_private_ryan_omaha_beach’）中的片段作为基准视频，数据集中共有 4 个指标视频，5 个非指标视频：

['anchor video: saving_private_ryan_omaha_beach', 
'positive video 1', 'positive video 2', 'positive video 3', 'positive video 4', 
'negative video 1: obama_kicks_door', 
'negative video 2: the_legend_of_1900_magic_waltz', 
'negative video 3: kennedy_assassination_slow_motion', 
'negative video 4: scent_of_woman_tango', 
'negative video 5: bolt_beijing_100m']

（请留神，咱们将仅变动比例的视频也视作反复视频。）

anchor video: saving_private_ryan_omaha_beach

positive videos

negative videos

03

创立汇合

在创立 Milvus 合集之前，请确保你曾经装置并启动了 Milvus。Milvus 是解决非结构化数据的好手，它能在后续的类似度检索和近邻搜寻中施展至关重要的作用。而后，咱们在 Milvus 数据库中创立一个「视频去重」的汇合（Collection），配置如下：

• 数据蕴含 2 列（Fields）：

  • id：主键，惟一且不反复
  • embedding：向量数据
• 创立索引（Index）能够减速检索：基于 embedding 列创立 IVF_FLAT 索引，应用参数"nlist":2048

• 类似度掂量形式（Metric）: L2 欧式间隔，越小示意越相近

  from pymilvus import connections, FieldSchema, CollectionSchema, DataType, Collection, utility
  
  connections.connect(host='127.0.0.1', port='19530')
  
  def create_milvus_collection(collection_name, dim):
    if utility.has_collection(collection_name):
        utility.drop_collection(collection_name)
    
    fields = [FieldSchema(name='id', dtype=DataType.INT64, descrition='ids', is_primary=True, auto_id=False),
    FieldSchema(name='embedding', dtype=DataType.FLOAT_VECTOR, descrition='embedding vectors', dim=dim)
    ]
    schema = CollectionSchema(fields=fields, description='video deduplication')
    collection = Collection(name=collection_name, schema=schema)
  
    # create IVF_FLAT index for collection.
    index_params = {
        'metric_type':'L2', #IP
        'index_type':"IVF_FLAT",
        'params':{"nlist":2048}
    }
    collection.create_index(field_name="embedding", index_params=index_params)
    return collection
    
    collection = create_milvus_collection('video_deduplication', 1024)

  04

  插入向量

在这一环节，咱们会用到预训练好的 DnS（distill and select）student 模型提取视频的特征向量，随后将视频向量插入到当时创立好的 Milvus 汇合中。

DnS 是一套高效检索和筛选视频的计划，首先对细粒度但简单的 teacher 模型进行预训练，而后在 teacher 模型的疏导下，用更多数据对粗粒度但更轻便的 student 模型进行训练。依据原论文的数据，这个办法训练出的 student 模型性能表现出色，不仅在速度上比 teacher 模型快了 20 倍，在内存上也缩小了 240 倍。在这里例子中，咱们仅应用预训练好 student 模型搭建一个视频级别（粗粒度）的「视频去重」零碎，因而每个视频会被转换成一个向量，代表了粗粒度的视频特色。

详情可参考论文原文：https://arxiv.org/abs/2106.13266

咱们利用 Towhee 提供的 DC API 以及算子 distill_and_select 搭建一个流水线，将视频库里的每个视频都转换成一个向量，并存入 Milvus 汇合中：

import os
import towhee
from towhee import dc
device = 'cuda'
# device = 'cpu'

dc = (towhee.read_csv('path_to_csv.csv').unstream()
            .runas_op['id', 'id'](func=lambda x: int(x "'id', 'id'"))
            .video_decode.ffmpeg['path', 'frames'](
                 start_time=0.0,
                 end_time=60.0,
                 sample_type='time_step_sample',
                 args={'time_step': 1})
            .runas_op['frames', 'frames'](func=lambda x: [y for y in x] "'frames', 'frames'")
            .distill_and_select['frames', 'vec'](
                 model_name='cg_student',
                 device=device)
            .to_milvus['id', 'vec'](collection=collection, batch=30 "'id', 'vec'")
      )

咱们在这里对下面的代码做一些阐明：

• towhee.read_csv(df_csv_path)：从 csv 文件中读取数据
• .runas_op['id', 'id'](func=lambda x: int(x "'id', 'id'"))：将 csv 中 id 列的数据类型从 str 转换为int
• .video_decode.ffmpeg: 每隔一秒对视频进行对立的二次采样，失去对应的视频帧列表
• .distill_and_select['frames', 'vec'](model_name='cg_student' "'frames', 'vec'")：应用 DnS 中的粗粒度 student 模型从视频中提取特征向量

• .to_milvus['id', 'vec'](collection=collection, batch=30 "'id', 'vec'")：将视频向量 30 个一批（batch=30）存入到 Milvus 汇合中

  print('Total number of inserted data is {}.'.format(collection.num_entities))

  最初，咱们存储到 Milvus 汇合中的向量共有 95 个：Total number of inserted data is 95.

05

查问与评估

当胜利将视频库中的视频都转换成向量存储到 Milvus 汇合后，咱们就能够进行查问了：输出一个视频，查问视频库中与其反复的视频。

dc = (towhee.read_csv('query_video_csv.csv').unstream()
          .runas_op['event', 'ground_truth_event'](func=lambda x:[x] "'event', 'ground_truth_event'")
          .video_decode.ffmpeg['path', 'frames'](
               start_time=0.0,
               end_time=60.0,
               sample_type='time_step_sample',
               args={'time_step': 1})
          .runas_op['frames', 'frames'](func=lambda x: [y for y in x] "'frames', 'frames'")
          .distill_and_select['frames', 'vec'](
               model_name='cg_student',
               device=device)
          .milvus_search['vec', 'topk_raw_res'](
               collection=collection,
               limit=min_sample_num)
          .runas_op['topk_raw_res', 'topk_events'](func=lambda res: [id2event[x.id] for i, x in enumerate(res)])
          .runas_op['topk_raw_res', 'topk_path'](func=lambda res: [id2path[x.id] for i, x in enumerate(res)])
      )

dc_list = dc.to_list()
# random_idx = random.randint(0, len(dc_list) - 1)
sample_num = 3
sample_idxs = random.sample(range(len(dc_list)), sample_num)
def get_query_and_predict_videos(idx):
    query_video = id2path[int(dc_list[idx].id)]
    print('query_video =', query_video)
    predict_topk_video_list = dc_list[idx].topk_path[1:]
    print('predict_topk_video_list =', predict_topk_video_list)
    return query_video, predict_topk_video_list
dsp_res_list = []
for idx in sample_idxs:
    query_video, predict_topk_video_list = get_query_and_predict_videos(idx)
    show_video_list = [query_video] + predict_topk_video_list
    caption_list = ['query video:' + Path(query_video).parent.stem] + ['result{0} video'.format(i) for i in range(len(predict_topk_video_list))]
    dsp_res_list.append(display_gifs_from_video(show_video_list, caption_list, tmpdirname=tmpdirname))

咱们随机查看三个查问的后果：左侧是咱们查问的视频，右侧则是零碎检测到的反复视频：

咱们先查看第一个视频的后果：第一个查问的视频选取自 NBA 的球员 Tracy McGrady 在 35 秒内狂夺 13 分的片段，咱们的零碎检测出了 4 个反复的视频。

dsp_res_list[0]

query for t-mac_13_points_in_35_seconds

咱们接着再看第二个视频的检测后果：第二个视频选取自奥巴马的一场演讲视频。令人诧异的是，在演讲完结后，他怄气地踹了门。能够发现第四个后果并不蕴含奥巴马这段演讲的片段，而是一段奥巴马在室外的演讲视频。

dsp_res_list[1]

query for obama_kicks_door

最初，咱们看第三个视频的检测后果：第三个视频出自 2004 年的一部名叫 Troy 的电影中 Hector 和 Achilles 的对决片段。

dsp_res_list[2]

（因为平台限度的起因，无奈上传图片，请关注 Zilliz 微信号、CSDN 等其它平台查看图片）

从下面的后果能够发信，咱们的「视频去重」零碎都能精确地查找出类似或反复的视频！找到类似的视频片段后，咱们能够自行决定是否删除，是不是很不便呢？

接下来，咱们将应用 mAP@topk 评估这个「视频去重」零碎。咱们利用指标后果 ground_truth_event 和检测后果 topk_events 对系统进行评估：

benchmark = (dc.with_metrics(['mean_average_precision',])
        .evaluate['ground_truth_event', 'topk_events'](name='map_at_k' "'ground_truth_event', 'topk_events'")
        .report())

在本文选取的数据上，咱们能够失去 mAP@top5 大概是 97 %：map_at_k：0.973977 这表明咱们的「视频去重」零碎获得了很高的分数！

06

总结

在明天的这篇文章中，咱们构建了一个简略的「视频去重」零碎，这个零碎能够帮忙咱们查找到反复的视频片段，缩小存储空间的占用以及在个性化举荐视频时晋升用户的体验。然而这个零碎仅限于粗粒度的去重，无奈实现更精密的辨认和检测。比方视频反复片段占比拟小的状况下，该零碎会因为受到大量的不反复片段烦扰而检测失败。那么如何解决这种状况，实现更精密的视频去重呢？咱们会在下一篇文章中揭晓，敬请期待！