视频处理器
Video Processor 是一个实用工具,负责为视频模型准备输入特征,并处理其输出的后处理。它提供诸如调整大小、归一化以及转换为 PyTorch 等变换。除了变换之外,VideoProcessor 类还负责从本地路径或 URL 解码视频(需要 torchcodec),并根据特定于模型的策略进行帧采样。
视频处理器扩展了图像处理器的功能,使视觉大语言模型(VLM)能够处理视频,并且与图像相比,视频使用一组不同的参数。它充当原始视频数据与模型之间的桥梁,确保输入特征针对 VLM 进行了优化。
视频处理器的设置保存在 video_preprocessor_config.json 中。较旧的检查点可能将它们存储在 preprocessor_config.json 中。重新加载时自定义后端的注册,请参阅 保存和重新加载。
使用示例
以下是使用 llava-hf/llava-onevision-qwen2-0.5b-ov-hf 模型加载视频处理器的示例:
from transformers import AutoVideoProcessor
processor = AutoVideoProcessor.from_pretrained("llava-hf/llava-onevision-qwen2-0.5b-ov-hf")预处理在输入张量所在的设备上运行,对于其他输入则在 CPU 上运行。设置 device 可在预处理前将视频移至其他设备;这不会更改后端。使用加速器时,你还可以使用 torch.compile 编译 torchvision 预处理。
import torch
from transformers.video_utils import load_video
from transformers import AutoVideoProcessor
device = torch.accelerator.current_accelerator().type if torch.accelerator.is_available() else "cpu"
video = load_video("video.mp4")
processor = AutoVideoProcessor.from_pretrained("llava-hf/llava-onevision-qwen2-0.5b-ov-hf", device=device)
processor = torch.compile(processor)
processed_video = processor(video, return_tensors="pt")视频处理器后端
内置视频处理器使用 torchvision 进行批量调整大小、重新缩放和归一化。要使用它们,请安装 PyTorch 和 torchvision。与图像处理器不同,内置视频处理器不提供 PIL 后端。
AutoVideoProcessor 默认选择 torchvision 后端。您也可以显式指定该后端,并检查已加载处理器的 backend 属性。
from transformers import AutoVideoProcessor
model_id = "llava-hf/llava-onevision-qwen2-0.5b-ov-hf"
video_processor = AutoVideoProcessor.from_pretrained(model_id, backend="torchvision")
print(video_processor.backend)
# torchvisionbackend 参数用于选择预处理实现,而非视频解码器。有关解码和帧采样的详细信息,请参阅 采样行为。
自定义后端
BaseVideoProcessor 提供基于 torchvision 的预处理。对于非 torchvision 的实现,您可以继承 VideoProcessorMixin 并重写 preprocess,使其返回一个包含模型所需输入的 BatchFeature。
使用 AutoVideoProcessor.register() 注册自定义类,并使用 video_processor_classes 将后端名称映射到相应的类。以下示例假设 NewCustomVideoProcessor 已定义,并包含 LLaVA-OneVision 检查点所需的预处理。
from transformers import LlavaOnevisionConfig
AutoVideoProcessor.register(
LlavaOnevisionConfig,
video_processor_classes={"custom": NewCustomVideoProcessor},
exist_ok=True,
)
custom_processor = AutoVideoProcessor.from_pretrained(model_id, backend="custom")使用 exist_ok=True,因为 LlavaOnevisionConfig 已经注册。提供的条目会合并到其后端映射中,匹配的条目会被替换,而未提供的条目(例如 torchvision)会被保留。
保存和重新加载
save_pretrained() 保存的是预处理配置,而不是通过 AutoVideoProcessor.register() 添加的后端映射。在同一个 Python 进程中,使用 backend="custom" 重新加载已保存的 processor:
custom_processor.save_pretrained("./saved-video-processor")
custom_processor = AutoVideoProcessor.from_pretrained("./saved-video-processor", backend="custom")自定义注册仅适用于当前 Python 进程。在新进程中,需要先定义或导入自定义类并重新注册,然后再进行加载。
采样行为
视频处理器还可以使用最适合给定模型的技术来采样视频帧。采样行为由 do_sample_frames 参数控制,并可通过模型特定参数(例如 num_frames 或 fps,即视频的采样速率)进行配置。如果输入视频以本地路径或 URL(str)的形式提供,处理器将自动对其进行解码。要获取有关已解码视频的元数据(例如采样帧索引、原始尺寸、时长和 fps),请将 return_metadata=True 传递给处理器。
指定
num_frames并不保证输出恰好包含该数量的帧。根据模型的不同,采样器可能会强制应用最小或最大帧数限制。默认解码器为
torchcodec,必须安装该组件。
import torch
from transformers import AutoVideoProcessor
device = torch.accelerator.current_accelerator().type if torch.accelerator.is_available() else "cpu"
processor = AutoVideoProcessor.from_pretrained("llava-hf/llava-onevision-qwen2-0.5b-ov-hf", device=device)
processed_video_inputs = processor(videos=["video_path.mp4"], return_metadata=True, do_sample_frames=True, return_tensors="pt")
video_metadata = processed_video_inputs["video_metadata"]
# See how many frames the original video had and what was the original FPS
print(video_metadata.total_num_frames, video_metadata.fps)如果您传入的是已解码的视频数组,但仍希望启用模型特定的帧采样,强烈建议提供 video_metadata。这能让采样器获知原始视频的时长和 FPS。您可以将元数据作为 VideoMetadata 对象或普通 dict 传入。
import torch
from transformers import AutoVideoProcessor
from transformers.video_utils import VideoMetadata
device = torch.accelerator.current_accelerator().type if torch.accelerator.is_available() else "cpu"
processor = AutoVideoProcessor.from_pretrained("llava-hf/llava-onevision-qwen2-0.5b-ov-hf", device=device)
decoded_video = torch.randint(0, 255, size=(100, 3, 1280, 1280)) # short video of 100 frames
video_metadata = VideoMetadata(
total_num_frames=100,
fps=24,
duration=4.1, # in seconds
)
processed_video_inputs = processor(videos=[decoded_video], video_metadata=video_metadata, do_sample_frames=True, num_frames=10, return_tensors="pt")
print(processed_video_inputs.pixel_values_videos.shape)
[10, 3, 384, 384]VideoProcessorMixin
class transformers.VideoProcessorMixin
< source >( **kwargs: *<class 'transformers.processing_utils.VideosKwargs'> )
fetch_videos
< source >( video_url_or_urls: str | list[str] | list[list[str]]sample_indices_fn = None )
将单个 URL 或 URL 列表转换为对应的 np.array 对象。
如果传入单个 url,返回值将是一个对象。如果传入一个列表,则返回一个对象列表。
from_dict
< source >( video_processor_dict: dict**kwargs ) → ~video_processing_utils.VideoProcessorBase
从 Python 参数字典实例化 ~video_processing_utils.VideoProcessorBase 的某个类型。
from_json_file
< source >( json_file: str | os.PathLike ) → A video processor of type ~video_processing_utils.VideoProcessorBase
从参数 JSON 文件的路径实例化类型为 ~video_processing_utils.VideoProcessorBase 的视频处理器。
from_pretrained
< source >( pretrained_model_name_or_path: str | os.PathLikecache_dir: str | os.PathLike | None = Noneforce_download: bool = Falselocal_files_only: bool = Falsetoken: str | bool | None = Nonerevision: str = 'main'**kwargs )
参数
- pretrained_model_name_or_path (
str或os.PathLike) — 可以是以下之一:- 一个字符串,即托管在 huggingface.co 上的模型仓库中预训练视频的 model id。
- 指向一个目录的路径,该目录包含使用
~video_processing_utils.VideoProcessorBase.save_pretrained方法保存的视频处理器文件,例如./my_model_directory/。 - 指向已保存的视频处理器 JSON 文件的路径,例如
./my_model_directory/video_preprocessor_config.json。
- cache_dir (
str或os.PathLike,可选) — 如果不应使用标准缓存,则下载预训练模型视频处理器时应缓存到的目录路径。 - force_download (
bool,可选,默认为False) — 是否强制(重新)下载视频处理器文件,并在缓存版本已存在时覆盖它们。 - proxies (
dict[str, str], 可选) — 按协议或端点使用的代理服务器字典,例如{'http': 'foo.bar:3128', 'http://hostname': 'foo.bar:4012'}.每次请求都会使用这些代理。 - token (
str或bool,可选) — 用于远程文件的 HTTP bearer 授权令牌。如果为True或未指定,则使用运行hf auth login时生成的令牌(存储在~/.huggingface中)。 - revision (
str,可选,默认为"main") — 要使用的特定模型版本。由于我们在 huggingface.co 上使用基于 git 的系统来存储模型和其他工件,因此它可以是分支名称、标签名称或提交 id,revision可以是 git 允许的任何标识符。
从视频处理器实例化一个 ~video_processing_utils.VideoProcessorBase 类型。
示例:
# We can't instantiate directly the base class *VideoProcessorBase* so let's show the examples on a
# derived class: *LlavaOnevisionVideoProcessor*
video_processor = LlavaOnevisionVideoProcessor.from_pretrained(
"llava-hf/llava-onevision-qwen2-0.5b-ov-hf"
) # Download video_processing_config from huggingface.co and cache.
video_processor = LlavaOnevisionVideoProcessor.from_pretrained(
"./test/saved_model/"
) # E.g. video processor (or model) was saved using *save_pretrained('./test/saved_model/')*
video_processor = LlavaOnevisionVideoProcessor.from_pretrained("./test/saved_model/video_preprocessor_config.json")
video_processor = LlavaOnevisionVideoProcessor.from_pretrained(
"llava-hf/llava-onevision-qwen2-0.5b-ov-hf", do_normalize=False, foo=False
)
assert video_processor.do_normalize is False
video_processor, unused_kwargs = LlavaOnevisionVideoProcessor.from_pretrained(
"llava-hf/llava-onevision-qwen2-0.5b-ov-hf", do_normalize=False, foo=False, return_unused_kwargs=True
)
assert video_processor.do_normalize is False
assert unused_kwargs == {"foo": False}get_video_processor_dict
< source >( pretrained_model_name_or_path: str | os.PathLike**kwargs ) → tuple[Dict, Dict]
从 pretrained_model_name_or_path 解析为参数字典,用于使用 from_dict 实例化类型为 ~video_processing_utils.VideoProcessorBase 的视频处理器。
预处理
< source >( videos: typing.Union[list['PIL.Image.Image'], numpy.ndarray, ForwardRef('torch.Tensor'), list[numpy.ndarray], list['torch.Tensor'], list[list['PIL.Image.Image']], list[list[numpy.ndarray]], list[list['torch.Tensor']], transformers.video_utils.URL, list[transformers.video_utils.URL], list[list[transformers.video_utils.URL]], transformers.video_utils.Path, list[transformers.video_utils.Path], list[list[transformers.video_utils.Path]]]**kwargs: *<class 'transformers.processing_utils.VideosKwargs'> ) → ~image_processing_base.BatchFeature
参数
- videos (
Union[list[PIL.Image.Image], numpy.ndarray, torch.Tensor, list[numpy.ndarray], list[torch.Tensor], list[list[PIL.Image.Image]], list[list[numpy.ndarray]], list[list[torch.Tensor]], ~video_utils.URL, list[~video_utils.URL], list[list[~video_utils.URL]], ~video_utils.Path, list[~video_utils.Path], list[list[~video_utils.Path]]]) — 要预处理的视频。期望输入单个视频或一批视频,其像素值范围为 0 到 255。如果传入的像素值在 0 到 1 之间,请设置do_rescale=False。 - do_convert_rgb (
bool,kwargs,可选) — 是否将图像转换为 RGB。 - do_resize (
bool,kwargs,可选) — 是否调整图像大小。 - size (
Annotated[int | list[int] | tuple[int, ...] | dict[str, int] | None, None], kwargs) — 描述模型的最大输入维度。 - default_to_square (
bool, kwargs, 可选) — 当 size 为整数时,是否在调整大小时默认使用正方形图像。 - resample (
Annotated[Union[int, PILImageResampling, NoneType], None], kwargs) — 如果调整图像大小,则使用的重采样滤波器。可以是枚举PILImageResampling中的一个值。仅当do_resize设置为True时生效。 - do_rescale (
bool,kwargs,可选) — 是否对图像进行重新缩放。 - rescale_factor (
float, kwargs, 可选) — 如果do_rescale设置为True,则用于对图像进行缩放的缩放因子。 - do_normalize (
bool,kwargs,可选) — 是否对图像进行归一化。 - image_mean (
Union[float, list[float], tuple[float, ...]], kwargs, 可选) — 用于归一化的图像均值。仅当do_normalize设置为True时生效。 - image_std (
Union[float, list[float], tuple[float, ...]], kwargs, 可选) — 用于归一化的图像标准差。仅当do_normalize设置为True时生效。 - do_center_crop (
bool,kwargs,可选) — 是否对图像进行中心裁剪。 - do_pad (
bool, kwargs, 可选) — 是否对图像进行填充。填充操作会将图像填充至批次中的最大尺寸,或填充至每张图像的固定正方形尺寸。具体的填充策略取决于模型。 - crop_size (
Annotated[int | list[int] | tuple[int, ...] | dict[str, int] | None, None], kwargs) — 应用center_crop后输出图像的尺寸。 - data_format (
Union[str, ~image_utils.ChannelDimension], kwargs, 可选) — 仅支持ChannelDimension.FIRST。为兼容旧版处理器而添加。 - input_data_format (
Union[str, ~image_utils.ChannelDimension], kwargs, 可选) — 输入图像的通道维度格式。如果未设置,则从输入图像中推断通道维度格式。可以是以下之一:"channels_first"或ChannelDimension.FIRST:图像采用 (num_channels, height, width) 格式。"channels_last"或ChannelDimension.LAST:图像采用 (height, width, num_channels) 格式。"none"或ChannelDimension.NONE:图像为 (height, width) 格式。
- device (
Annotated[Union[str, torch.device, NoneType], None], kwargs) — 用于处理视频的设备。如果未设置,则从输入视频中推断设备。 - do_sample_frames (
bool, kwargs, 可选) — 是否在处理前对视频进行帧采样,还是处理整个视频。 - video_metadata (
Annotated[~video_utils.VideoMetadata | dict | list[dict | ~video_utils.VideoMetadata] | list[list[dict | ~video_utils.VideoMetadata]] | None, None], kwargs) — 视频的元数据,包含总时长、帧率(fps)和总帧数等信息。该元数据将用于从视频中采样帧或计算时间戳。除非您打算在预处理之前手动解码视频,否则请勿传入任何元数据。 - fps (
Annotated[int | float | None, None], kwargs) — 当do_sample_frames=True时,每秒要采样的目标帧数。 - num_frames (
Annotated[int | None, None], kwargs) — 当do_sample_frames=True时,要采样的最大帧数。 - return_metadata (
bool, kwargs, 可选) — 是否返回视频元数据。视频元数据是一个包含视频时长、帧率、解码后端等信息的对象。 - return_tensors (
Annotated[str | ~utils.generic.TensorType | None, None], kwargs) — 如果设置为'pt',则返回堆叠的张量,否则返回张量列表。
返回值
~image_processing_base.BatchFeature
- data (
dict) — 由 call 方法返回的列表/数组/张量字典(如 ‘pixel_values’ 等)。 - tensor_type (
Union[None, str, TensorType], 可选) — 您可以在此处指定 tensor_type,以便在初始化时将 PyTorch/Numpy 张量中的整数列表进行转换。
register_for_auto_class
< source >( auto_class = 'AutoVideoProcessor' )
将此注册到指定的 auto 类。此操作仅适用于自定义视频处理器,因为库中的视频处理器已经与 AutoVideoProcessor 进行了映射。
此 API 目前处于实验阶段,在后续版本中可能会出现一些轻微的破坏性变更。
save_pretrained
< source >( save_directory: str | os.PathLikepush_to_hub: bool = False**kwargs )
参数
- save_directory (
str或os.PathLike) — 保存视频处理器 JSON 文件的目录(如果该目录不存在,则会自动创建)。 - push_to_hub (
bool,可选,默认为False) — 是否在保存模型后将其推送到 Hugging Face 模型库。您可以使用repo_id指定要推送到的仓库(默认使用您命名空间中save_directory的名称)。 - kwargs (
dict[str, Any],可选) — 传递给 push_to_hub() 方法的额外关键字参数。
将视频处理器对象保存到目录 save_directory,以便之后可以使用 ~video_processing_utils.VideoProcessorBase.from_pretrained 类方法重新加载。
将此实例序列化为 Python 字典。
BaseVideoProcessor
class transformers.BaseVideoProcessor
< source >( **kwargs: *<class 'transformers.processing_utils.VideosKwargs'> )
convert_to_rgb
< source >( video: torch.Tensor ) → torch.Tensor
将视频转换为 RGB 格式。
sample_frames
< source >( metadata: VideoMetadatanum_frames: int | None = Nonefps: int | float | None = None**kwargs ) → np.ndarray
默认采样函数,在 0 到总帧数之间均匀采样指定数量的帧。
如果同时传入 fps 和元数据,则按 fps 帧/秒均匀采样。参数 num_frames 和 fps 互斥。