专栏名称: Databri AI
创始人刘强出版过专著「推荐系统:算法、案例与大模型」、「构建企业级推荐系统」等。Databri AI聚焦金融、传统行业的数智化转型,提供咨询、培训、项目实施解决方案。过去3年服务过中国银联、中国移动、中盐、招商银行、广发银行等大客户。
目录
相关文章推荐
51好读  ›  专栏  ›  Databri AI

「大模型智能体」27| 从视频中提取精美宣传海报

Databri AI  · 公众号  ·  · 2025-01-08 13:12

正文

请到「今天看啥」查看全文


你好,我是刘强。

本课程是关于 大模型智能体 的实战课程,包括原理、算法、应用场景、代码实战案例等,下表是本次课程的大纲。本课是第27节,讲解实战案例:从视频中提取精美海报。本课约7700字,阅读时长35min。


以下是本次课程的正文:



我们在前面课程中的案例处理的主要是文件、音频等模态信息,而图片、视频是互联网中数量更多、大家接触更多的媒介,多模态大模型也是大模型领域发展的最重要方向之一。为了让读者可以更好地了解大模型、智能体在多模态上的应用,本课我们聚焦解决一个视觉模态问题——从视频中提取精美宣传海报。

本课除了会用到商业大模型API外,还会使用一些主流的视频、图片处理工具和小模型(比如OpenCV、OCR、图片超分等),大模型+小模型的配合是智能体应用的主流方法,各个领域的垂直小模型读者需要深入了解。这部分知识不是本书的重点,我不会深入讲解,读者需要自己去学习。本课我们从整体业务流程、视频预处理、候选海报生成、最好海报选择、海报美化等5个小节展开讲解。

27.1 整体业务流程

智能海报生成从原始视频开始,需要经过4个步骤来实现。第一步是从原始视频生成视频内容介绍和候选帧,接着利用垂直小模型从候选帧中筛选出topN(本课N=5)最合适的帧做为候选海报,然后利用多模态大模型结合视频介绍文本、候选帧及大模型的世界知识选择最合适做为海报的一张候选帧做为海报,最后对海报进行优化(剔除不必要的文字、提升图片分辨率)生成最终的海报,具体流程见下面图27-1,下面小节我们详细讲解每个步骤的具体原理和代码实现。

图27-1:智能海报流程


27.2 视频预处理

视频预处理是通过OpenCV按照一定间距从视频中抽帧并按照一定的要求( 清晰度 )筛选候选帧做为后面的海报备选。为了后面步骤能够更好地选择海报,我们还需要从视频中抽取音频文字并利用大模型(这里使用的DeepSeek的大模型)对视频内容进行总结(控制在70个字以内),利用一段文字高度概括视频的主要内容。下面是具体的代码实现,代码对应github上的代码仓库中的 llm_agent_abc/video_poster_generate/prepare_data.py,代码中对相关功能和函数已经做了比较细致的说明。

这里提一下,通用的函数我们放到utils里面了(对应github仓库中的llm_agent_abc/video_poster_generate/utils.py,后面不赘述,本课也不提供utils中的代码案例)。

import osimport tempfilefrom moviepy.video.io.VideoFileClip import VideoFileClipimport whisperimport cv2import syssys.path.append('./')from openai import OpenAIfrom utils import save_frames_as_images, write_text_to_filefrom video_poster_generate.configs.model_config import DEEPSEEK_API_KEY, DEEPSEEK_MODEL, DEEPSEEK_BASE_URL
client = OpenAI( api_key=DEEPSEEK_API_KEY, base_url=DEEPSEEK_BASE_URL,)
def generate_summary(content): """ 使用 OpenAI 的 GPT 生成文章的中文长文总结(70字)。
:param content: 文章内容 :return: 中文长文总结 """ prompt = ( "基于提供的中文内容,请你对内容进行总结。希望你的总结全面、概括,不要遗漏重点信息,请控制在70个字以内。\n" f"给你提供的内容是:{content}\n" "\n现在请你撰写总结。" ) response = client.chat.completions.create( model=DEEPSEEK_MODEL, messages=[ {"role": "system", "content": "你是一名专业的文字工作者。"}, {"role": "user", "content": prompt}, ], max_tokens=4096 ) message_content = response.choices[0].message.content.strip() return message_content

def extract_audio_as_text(video_path): """ 提取视频的音频并使用 Whisper 进行 ASR 转换。 :param video_path: :return: """ # 提取音频 video_clip = VideoFileClip(video_path) temp_audio = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) video_clip.audio.write_audiofile(temp_audio.name)
# 使用 Whisper 进行 ASR 转换 model = whisper.load_model("large") result = model.transcribe(temp_audio.name) os.remove(temp_audio.name) return result["text"]

def extract_high_quality_frames(video_path, frame_interval=30, quality_threshold=100.0): """ 从视频中提取质量较高的帧作为候选帧。 视频通过帧间隔采样,跳过不需要处理的帧。 使用 Laplacian 方差法 计算帧的清晰度。 清晰度超过指定阈值的帧会被提取并存储。 :param video_path: :param frame_interval: :param quality_threshold: :return: """ cap = cv2.VideoCapture(video_path) frames = [] frame_quality = [] frame_count = 0
while True: ret, frame = cap.read() if not ret: break # 仅处理间隔帧 if frame_count % frame_interval == 0: # 计算清晰度(Laplacian 方差法) gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY) quality = cv2.Laplacian(gray, cv2.CV_64F).var() if quality >= quality_threshold: # 转换颜色通道顺序为 RGB frame_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB) frames.append(frame_rgb) frame_quality.append(quality)
frame_count += 1
cap.release() return frames, frame_quality

if __name__ == "__main__": video_path = "./video_poster_generate/data/1.mp4" # 替换为你的视频路径 candidate_poster_dir = "./video_poster_generate/output/candidate_frames" os.makedirs(candidate_poster_dir, exist_ok=True)    summary_text_dir = "./video_poster_generate/output/summary_text.txt"
text = extract_audio_as_text(video_path) summary_text = generate_summary(text) # 对文本进行摘要,控制字数(clip模型要求控制字数)同时突出重点。    print(f"summary_text: {summary_text}")
    write_text_to_file(summary_text_dir, summary_text)
    frames, _ = extract_high_quality_frames(video_path, frame_interval=30, quality_threshold=100.0)
save_frames_as_images(frames, candidate_poster_dir)

我们的代码仓库data目录下提供的是一个关于脑机接口的视频,运行上面代码会在output/ candidate_frames下生成46个候选帧,并且生成对视频的文字总结(如下)。

人们已经是半机械人,因为与手机和电脑等数字扩展的通信速率较慢,信息流动有限。提高数据速率,建立高带宽接口,是实现人机长期共生的关键,最终人们可能选择保留生物自我。

27.3 候选海报生成
第一步生成的帧会比较多(如果视频长的话会更多),不方便我们利用大模型进行处理(很多大模型对输入的图片数量有限制),因此本小节会基于一些初步的条件从候选帧中筛选出topN(N=5)个最合适的候选帧。

本步我们利用2个衡量帧的质量的指标(利用CLIP模型计算候选帧跟视频内容的相似度、视频本身的质量)对候选帧的质量求调和平均数,然后降序排列选择最好的5个候选帧做为候选海报。这里用到的CLIP模型和视频质量模型都是垂直小模型,读者自行学习。下面是完整的代码实现(对应github仓库中的 llm_agent_abc/video_poster_generate/generate_candidate_poster.py )。

import sysimport torchimport numpy as npsys.path.append('./')from transformers import CLIPModel, CLIPProcessor, CLIPTokenizerfrom utils import load_images_as_rgb, save_frames_to_jsonimport torchvision.transforms as transformsfrom video_poster_generate.configs.model_config import BEST_POSTER_NUM

def get_text_similarity_scores(frames_with_paths, text, batch_size=50): """    使用 CLIP 模型对帧和文本进行匹配,并用 Softmax 对分值进行归一化。
Args: - frames_with_paths (list): 每个元素为字典,格式: [ {'image_dir': "图片的相对路径", "frame": 图片的 RGB 格式}, ... ] - text (str): 要匹配的文本。    - batch_size (int): 每批处理的帧数。
Returns: - result (list): 包含每帧路径、RGB 图像和归一化文本相似度分数的列表,格式: [ {'image_dir': "图片的相对路径", "frame": 图片的 RGB 格式, "text_similar_score": 归一化相似度分数}, ... ] """ # 加载 CLIP 模型和处理器 model = CLIPModel.from_pretrained("openai/clip-vit-large-patch14")    processor = CLIPProcessor.from_pretrained("openai/clip-vit-large-patch14")
# 获取模型支持的最大文本长度 max_text_length = model.config.text_config.max_position_embeddings
# 截断文本以适应最大长度 if len(text) > max_text_length:        text = text[:max_text_length]
# 准备数据 pil_frames = [frame_info["frame"] for frame_info in frames_with_paths] image_dirs = [frame_info["image_dir"] for frame_info in frames_with_paths] # 分批处理帧 logits_per_image = [] for i in range(0, len(pil_frames), batch_size): batch_frames = pil_frames[i:i + batch_size] inputs = processor(text=[text], images=batch_frames, return_tensors="pt", padding=True, truncation=True) # 模型推理 with torch.no_grad(): outputs = model(**inputs) logits_per_image_batch = outputs.logits_per_image.detach().numpy() logits_per_image.extend(logits_per_image_batch.squeeze().tolist()) # 累积结果 # Min-Max 归一化分值 logits_per_image = np.array(logits_per_image) min_score = np.min(logits_per_image) max_score = np.max(logits_per_image) # 防止分母为0的情况 if max_score - min_score == 0: normalized_scores = np.ones_like(logits_per_image) # 如果所有值相同,归一化为1 else:        normalized_scores = (logits_per_image - min_score) / (max_score - min_score)
# 生成结果 result = [ { "image_dir": image_dirs[i], "frame": pil_frames[i], "text_similar_score": round(normalized_scores[i], 4) } for i in range(len(pil_frames)) ]
return result

def get_quality_scores(frames_with_info): """ 为一组帧计算质量分数,并返回包含质量分数的帧信息。
Args: - frames_with_info (list): 每个元素为字典,格式: [ {'image_dir': 'frame1.jpg', 'frame': , 'text_similar_score': 0.85}, {'image_dir': 'frame2.jpg', 'frame': , 'text_similar_score': 0.78}, {'image_dir': 'frame3.jpg', 'frame': , 'text_similar_score': 0.90} ]
Returns: - results (list): 包含质量分数的帧信息列表,格式: [ {'image_dir': 'frame1.jpg', 'frame': , 'text_similar_score': 0.85, 'quality_score': 0.8}, {'image_dir': 'frame2.jpg', 'frame': , 'text_similar_score': 0.78, 'quality_score': 0.75}, {'image_dir': 'frame3.jpg', 'frame': , 'text_similar_score': 0.90, 'quality_score': 0.9} ] """ # 检查是否有可用的 GPU device = torch.device("cuda") if torch.cuda.is_available() else "mps" # 加载模型 model = torch.hub.load( repo_or_dir="miccunifi/ARNIQA", source="github", model="ARNIQA", regressor_dataset="kadid10k", # 可选其他数据集 ) model.eval().to(device) # 定义预处理流水线 preprocess = transforms.Compose([ transforms.ToTensor(), transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]), ]) # 初始化结果存储 results = [] # 遍历所有帧并计算分数 for frame_info in frames_with_info: image_dir = frame_info['image_dir'] frame = frame_info['frame'] text_similar_score = frame_info['text_similar_score'] # 获取帧的半分辨率版本        img_ds = transforms.Resize((frame.size[1] // 2, frame.size[0] // 2))(frame)
# 预处理 img = preprocess(frame).unsqueeze(0).to(device) img_ds = preprocess(img_ds).unsqueeze(0).to(device) # 计算质量得分 with torch.no_grad(), torch.cuda.amp.autocast(): quality_score = model(img, img_ds, return_embedding=False, scale_score=True).item() # 添加结果 results.append({ 'image_dir': image_dir, 'frame': frame, 'text_similar_score': text_similar_score, 'quality_score': round(quality_score, 4) })
return results

def filter_and_sort_frames(frames, best_poster_num=5): """ 根据指定条件筛选并排序满足条件的帧。
Args: - frames (list): 输入的帧列表,每个帧为包含以下键的字典: [ {'image_dir': 'frame1.jpg', 'frame': , 'text_similar_score': 0.85, 'quality_score': 0.8}, {'image_dir': 'frame2.jpg', 'frame': , 'text_similar_score': 0.78, 'quality_score': 0.75}, {'image_dir': 'frame3.jpg', 'frame': , 'text_similar_score': 0.90, 'quality_score': 0.9} ] - min_similarity_score (float): 最小文本相似度分数。 - min_quality_score (float): 最小质量分数。 - best_poster_num (int): 返回的最大帧数量。
Returns: - filtered_frames (list): 筛选后的帧列表,每个元素为原始字典格式。 """ # 1. 按照调和平均数排序 sorted_frames = sorted( frames, key=lambda x: ( 2 * x['text_similar_score'] * x['quality_score'] / (x['text_similar_score'] + x['quality_score'] if x['text_similar_score'] + x[ 'quality_score'] > 0 else 1) ), reverse=True ) # 2. 限制返回帧的数量不超过 `best_poster_num` return sorted_frames[:best_poster_num]

if __name__ == "__main__": candidate_poster_dir = "./video_poster_generate/output/candidate_frames" summary_text_dir = "./video_poster_generate/output/summary_text.txt" # 读取文本 with open(summary_text_dir, "r", encoding="utf-8") as f: text = f.read()
    frames_with_paths = load_images_as_rgb(candidate_poster_dir)
frames_with_info = get_text_similarity_scores(frames_with_paths, text)
frames_with_scores = get_quality_scores(frames_with_info)
frames_dir = "./video_poster_generate/output/frames_with_scores.json" save_frames_to_json(frames_with_scores, frames_dir)
filtered_frames = filter_and_sort_frames(frames_with_scores, BEST_POSTER_NUM)
filtered_frames_dir = "./video_poster_generate/output/filtered_frames.json" save_frames_to_json(filtered_frames, filtered_frames_dir)

运行上面代码后,我们会生成2个JSON文件: frames_with_scores.json和filtered_frames.json,前者是所有候选帧计算完前面2个质量指标后的文件,后者是选择了topN后的文件,文件中会包含视频地址、2类指令指标分值,具体结果参考下面JSON。

[    {        "image_dir": "video_poster_generate/output/candidate_frames/poster_1.jpg",        "text_similar_score": 0.567,        "quality_score": 0.3802    },    {        "image_dir": "video_poster_generate/output/candidate_frames/poster_46.jpg",        "text_similar_score": 0.7309,        "quality_score": 0.3081    },    {        "image_dir": "video_poster_generate/output/candidate_frames/poster_44.jpg",        "text_similar_score": 0.8549,        "quality_score": 0.2803    },    {        "image_dir": "video_poster_generate/output/candidate_frames/poster_45.jpg",        "text_similar_score": 0.759,        "quality_score": 0.2904    },    {        "image_dir": "video_poster_generate/output/candidate_frames/poster_33.jpg",        "text_similar_score": 0.8385,        "quality_score": 0.2687    }]

27.4 最好海报选择

其实上面一步也可以作为选择最好海报的方法(即取排在第一位的帧),但这个方法的问题是只利用了相似性和图片本身质量,有可能会误判(比如可能排第二个的效果会更好),所以在这一步我们利用能力更强的多模态大模型来帮我们从最好的5个候选海报中进行选择。大模型可以结合图片、视频描述及大模型中压缩的世界知识进行更加综合的权衡。







请到「今天看啥」查看全文