Skip to content

05 视频打点:AI在媒体工作流中的提效应用

本门课程为精品小课,不标配音频

你好,我是王吕。今天我们一起来了解下 AI 在视频处理中的应用。

在极客时间,有大量的优质视频课和各种直播素材,内容团队每天都会花很多精力处理这些资源。我们在跟业务沟通之后,找到了以下几个场景,可以使用 AI 改进视频处理效率。

  • 上传视频后自动提取字幕

  • 把视频分割成几个节点,然后总结每个节点的内容,形成视频的要点大纲

其实这两个场景是可以合并为一个工作流的。因为想要使用 LLM,首先要做的就是把视频转成文字(自然语言)。这里有两种方法,第一个就是直接提取字幕,得到视频讲话稿,这种方式适用于视频内容中语言信息较为重要的场景,如演讲、访谈、教程等。第二种方式是将视频按帧分割,并使用视觉模型逐帧分析每个画面,适用于需要对视觉内容进行详细分析的场景,如动作检测、物体识别、场景理解等。

通过逐帧处理,视觉模型能够从图像中提取更多的视觉信息,从而增强对视频内容的感知。这种方式通常适合需要深入了解视频场景或动作的复杂任务,如视频监控、自动驾驶或影视内容的视觉分析等。极客时间就非常适合字幕提取的方式。得到字幕之后,再针对文字稿提炼要点,然后进行时间标记,形成视频打点,我画了一幅图来描述一下这个工作流。

图片

确定流程之后,我们来详细说说实现过程。

前期准备

启动开发之前,我们首先要做的是技术选型。从视频中提取音频,选择的是 FFmpeg,FFmpeg 是一个 CPU 密集型应用,选择服务器配置的时候,尽量选择高主频和多核 CPU。其次是音频转录文字,当时我们做这个功能的时候,还只有 OpenAI 的 Whisper 模型支持,所以我们是基于 Whisper 来实现的。

当然,现在通义听悟也可以实现了,并且目前极客时间已经切换到了通义听悟,不过这里我们还是按照 Whisper 讲解。本地运行 Whisper 需要使用大量的 GPU 资源,它本身也是基于 Transformer 架构,为了节省资源,最终我们选择 OpenAI 提供的公共服务。

准备就绪之后,我们开始实现。

提取音频

我们打算把整个流程做成自动处理、自动触发的。所以每当有新视频上传之后,会把视频源地址发送给我们的视频处理应用,应用根据地址下载视频到本地,然后使用 FFmpeg 提取音频,保存音频供下一步处理,做完之后记得清理掉本地的视频。相关代码如下:

import os
import requests
import ffmpeg
from urllib.parse import urlparse
import uuid
import datetime

# 应用级别的常量
TEMP_VIDEO_DIR = './temp_videos'
AUDIO_OUTPUT_DIR = './audio_output'

def download_video(video_url):
    try:
        # 确保临时目录存在
        os.makedirs(TEMP_VIDEO_DIR, exist_ok=True)

        # 生成随机文件名
        video_filename = datetime.datetime.now().strftime('%Y%m%d') + '/' + str(uuid.uuid4()) + '.mp4'
        local_path = os.path.join(TEMP_VIDEO_DIR, video_filename)

        # 确保目录存在
        os.makedirs(os.path.dirname(local_path), exist_ok=True)

        # 下载视频
        response = requests.get(video_url, stream=True)
        response.raise_for_status()

        with open(local_path, 'wb') as file:
            for chunk in response.iter_content(chunk_size=8192):
                file.write(chunk)

        # todo: 打印log
        return local_path
    except requests.RequestException as e:
        # todo: 处理异常
        return None

def extract_audio(video_path):
    try:
        # 确保输出目录存在
        os.makedirs(AUDIO_OUTPUT_DIR, exist_ok=True)

        # 构建音频输出路径,使用随机字符串作为音频名字
        audio_filename = uuid.uuid4().hex + '.mp3'
        output_path = os.path.join(AUDIO_OUTPUT_DIR, audio_filename)

        # 使用ffmpeg提取音频
        stream = ffmpeg.input(video_path)
        stream = ffmpeg.output(stream, output_path, acodec='libmp3lame', ab='128k')
        ffmpeg.run(stream, overwrite_output=True)

        # todo: 打印log
        return output_path
    except ffmpeg.Error as e:
        # todo: 处理异常
        return None

def cleanup_video(video_path):
    try:
        os.remove(video_path)
        # todo: 打印log
    except OSError as e:
        # todo: 处理异常
        pass

def process_video(video_url):
    # 下载视频
    video_path = download_video(video_url)
    if not video_path:
        return None

    # 提取音频
    audio_path = extract_audio(video_path)

    # 清理视频文件
    cleanup_video(video_path)

    # 返回音频文件路径
    return audio_path

那么经过上述处理,就可以把音频发送给 Whisper 了吗?答案是还不行,OpenAI 的Whisper 支持的音频大小不能超过 25MB,这也是为什么第一步要提取音频,而不是直接传视频,目的就是减小媒体资源的体积,但是这还不够,因为转出来的音频很有可能还是比 25MB 大。所以我们还要切割音频,分段传递给 Whisper,之后再把结果合并起来。我们目前是按照20分钟作为每段时长处理的,其实远远小于 25MB,你也可以根据你的业务需求来选择时长。

from pydub import AudioSegment

AUDIO_SEGMENTS_DIR = './audio_segments'
SEGMENT_LENGTH = 20 * 60 * 1000  # 20分钟,以毫秒为单位

def split_audio(audio_path):
    try:
        # 确保音频片段目录存在
        os.makedirs(AUDIO_SEGMENTS_DIR, exist_ok=True)

        # 加载音频文件
        audio = AudioSegment.from_mp3(audio_path)

        # 计算需要切割的片段数
        num_segments = math.ceil(len(audio) / SEGMENT_LENGTH)

        segments = []
        for i in range(num_segments):
            # 计算当前片段的起始和结束时间
            start = i * SEGMENT_LENGTH
            end = min((i + 1) * SEGMENT_LENGTH, len(audio))

            # 提取片段
            segment = audio[start:end]

            # 生成片段文件名
            base_name = os.path.splitext(os.path.basename(audio_path))[0]
            segment_filename = f"{base_name}_segment_{i+1}.mp3"
            segment_path = os.path.join(AUDIO_SEGMENTS_DIR, segment_filename)

            # 导出片段
            segment.export(segment_path, format="mp3")
            segments.append(segment_path)

        # todo: 打印log
        return segments
    except Exception as e:
        # todo: 处理异常
        return None

上述代码就可以把音频切割成多个片段,每个片段最长不超过 20 分钟,这里有几个需要注意的地方。

  • pydub 的音频切割功能是以毫秒为单位,只能在整数毫秒处切割,有可能会造成声音被异常截断,影响转录效果。

  • 我们分割的方式是按照固定时长,一种极端情况是:在切割点位置,正好有人在说话,这里又分为两种情况,第一种是正好卡在一句话中间,但是没有卡在字或者词上,这样会导致一句完整的话被分割成两部分,在合并的时候要处理下;第二种情况是正好卡在一个字或者词上,比如 “hello” 这个词,刚发出 “Ha” 这个音的时候,就被截断了,那么就会导致在转录阶段这个词被错误处理,这个问题我会在最后改进的时候,提供一些方法。

  • 每个步骤都会产生中间资源,比如视频、音频、切片等,这些资源在使用完成之后,一定要记得清理。

  • 如果自己部署开源的 Whisper 模型处理,那么理论上可以不用切片分段,但是需要注意资源使用情况,防止长时间过载使用,导致异常。

现在音频文件都准备好了,开始使用 Whisper 进行转录。

视频转录

使用 OpenAI 提供的服务进行转录很简单,直接用OpenAI 提供的 SDK 就可以了。

from openai import OpenAI

client = OpenAI()

def transcript_audio(audio_path):
    try:
        # 调用OpenAI API进行音频转写
        audio_file = open(audio_path, "rb")
        transcript = client.audio.transcriptions.create(
            file=audio_file,
            model="whisper-1",
            response_format="srt",
            language="zh-CN",
            timestamp_granularities=["segment"],
            prompt=""
        )

        # todo: 打印log
        return transcript
    except Exception as e:
        # todo: 处理异常
        return None

这里我们摸索了一些经验技巧,可以帮助你提升转录效果。

  1. 因为我们要生成的是字幕文件,直接把 response_format 这个参数的值设置为 srt, 可以直接得到带时间戳的标准 srt 格式文件,这样做也没问题。不过还有一种方式,就是把这个参数设置为 verbose_json,这样会得到一个json,这个json 里包含了每句台词的详细数据,包括但不限于文本、开始时间、结束时间,这里的开始结束时间都是精确到小数点后 15位的,如果追求后续处理的高精度,那么可以选择这个格式,然后自己手动处理 json。

  2. 如果确定音频文件的主要语言,最好设置一下 language 参数,可以提升转录效果。

  3. 细心的你可能发现,我的 prompt 是空的,这里的 prompt 和我们传统的 LLM 使用的 prompt 还不太一样,这里的 prompt 其实是一个词典。你可以把音频中高频出现的专有名词等写在这里,这样在转录的过程就可以识别这些专有名词了,这里推荐做成动态参数,根据音频主题,找一些音频中可能聊到的名词填上去,还需要注意一下格式。这里只支持名词列表,之间用逗号分隔,并且整个 prompt 最大是 255个字符,如果写成其他格式,可能会不生效。

至此,我们得到了每个音频切片对应的字幕文件,这个时候,可以写一个方法把这些切片字幕合并到一起,合并的时候要注意两个问题:

  1. 一定要保证顺序正确,并且切片的字幕也能一一对应上,一旦中途缺少对应的字幕文件,整个转换就算失败了,需要重新转录对应的片段。

  2. 合并的过程,要按照顺序,修改每个切片字幕中的时间戳,比如,第二个字幕的开始时间戳是 20分 0 秒,而不是 0分0秒,第三个字幕的开始时间是 40分0秒,以此类推。字幕文件中所有涉及到的时间戳都要改写,并且增加上对应的开始时间。

合并完成之后,我们就得到了完整的字幕文件,交给业务系统做后续处理就可以了。

视频打点

现在我们有了视频对应的字幕文件,开始进行打点总结。考虑字幕的长度,总结也是采用 refine 的方式,很快你就会发现这里的处理流程跟上节课思维导图那里使用的方法是一样的,到这里,我们又回到了压缩信息这个问题上,我们可以直接复用上一节的代码,只需要调整一下对应的 prompt。

你是一位精通视频内容分析的AI助手你的任务是分析给定的视频字幕文本提取关键要点并生成一份JSON格式的要点列表请遵循以下指南

1. 仔细阅读提供的字幕文本识别主要话题和关键信息
2. 提取3-5个最重要的要点每个要点不超过10个字
3. 每个要点必须附带SRT格式的开始和结束时间戳00:00:00,000)。
4. 如果提供了前段总结请在此基础上整合提炼和补充避免重复
5. 使用以下JSON格式输出结果

{
  "summary": [
    {
      "point": "要点1",
      "start_time": "00:00:00,000",
      "end_time": "00:00:00,000"
    },
    {
      "point": "要点2",
      "start_time": "00:00:00,000",
      "end_time": "00:00:00,000"
    }
  ]
}

前段总结
<|previous_summary|>

字幕文本
<|subtitle_text|>

请基于以上信息生成或refine视频要点列表并以指定的JSON格式输出

可能需要你去反复调试的部分,也是跟上节课不一样的部分,就是时间戳那一块,采用好的模型可以提高准确度,包括 prompt 也是需要反复调整的。

到这里,我们就完成了针对视频的字幕提取、视频打点等工作,我们可以把打点内容推给搜索数据库,这样用户在搜索的时候,就可以用文字搜索到视频了,并且可以跳转到对应的播放位置,对用户来说,大大降低了在视频中找内容的时间成本,也提高了平台视频内容的利用率。

进一步改进

在上述工作流中,还有几个地方我们可以做进一步优化,以提升最终的内容质量,下面我提几个思路。

  1. 人声检测:在切割音频阶段,我们也可以使用一些技术,检测切割点是否有人声,如果检测到人声的话,把切割点往前减少1秒,再次尝试切割,可以考虑使用 Python 的 webrtcvad 这个库,这样可以减少切割人声导致的问题。不过要注意的是,一旦不按照固定长度切割,就需要使用 verbose_json 的格式,手动把上一个切片的结束时间加到下一个切片的开始时间上,在时间处理上,需要格外小心。

  2. LLM 重写字幕:在把字幕合并完之后,在切割点可能会存在不准确的情况,甚至转录也存在个别错误。这时候可以把转录的结果使用 LLM 重写一下,把 temperature 降到最低,只让 LLM 修复其中的文字错误。合并错误切割的语句,可以使用一些超长 token 的模型,比如 qwen-long,也可以提高转录质量。

  3. 打点时间戳:如果直接针对字幕文件进行总结,里边包含了大量的时间戳文本,对于某些模型来说,可能会干扰结果的生成。这里可以处理一下,把字幕中的时间戳替换成占位符,比如 [0001]、[0002] 这样子,同时也减少了字幕文件的文本数量,等到总结完之后,再替换回来。

小结

这节课我们按照视频处理工作流顺序,一起实现了一遍字幕提取和视频打点功能,中间分享了一些我在实践过程中的经验,最后又发散了一些优化方法。

可以看到的是,大模型在内容处理上是可以大大提升企业内的生产效率的,这里的一个实践原则就是: 把所有的内容都变成文本,然后再用对口的 LLM 统一处理

我们把整个流程展开看,在每一个步骤中都可以再去寻找优化点,经过一层一层的处理,最终得到我们想要的效果。

最后我还提了几个优化思路,你还能想到哪些优化方法?欢迎一起交流讨论,我们共同成长。