モデルファインチューニングの概要

# 使用前にインストールしてください：pip install opencv-python
import math
import os
import logging
import cv2

logger = logging.getLogger(__name__)

FRAME_FACTOR = 2

# Qwen3.6、Qwen3.5、Qwen3-VL、qwen-vl-max-0813、qwen-vl-plus-0815、qwen-vl-plus-0710 などのモデルでは、画像のスケーリング係数は 32 です。
IMAGE_FACTOR = 32

# その他のモデルでは、画像のスケーリング係数は 28 です。
# IMAGE_FACTOR = 28

# ビデオフレームの最大アスペクト比
MAX_RATIO = 200
# ビデオフレームのピクセル下限
VIDEO_MIN_PIXELS = 4 * 32 * 32
# ビデオフレームのピクセル上限。Qwen3-VL-Plus モデルの場合、VIDEO_MAX_PIXELS は 640 * 32 * 32 です。その他のモデルでは 768 * 32 * 32 です。
VIDEO_MAX_PIXELS = 640 * 32 * 32

# ユーザーが FPS パラメーターを渡さない場合、fps にはデフォルト値が使用されます。
FPS = 2.0
# 抽出される最小フレーム数
FPS_MIN_FRAMES = 4
# 抽出される最大フレーム数 (選択したモデルに基づいて設定)
FPS_MAX_FRAMES = 2000

# ビデオ入力の最大ピクセル値。Qwen3-VL-Plus モデルの場合、VIDEO_TOTAL_PIXELS を 131072 * 32 * 32 に設定します。その他のモデルでは 65536 * 32 * 32 に設定します。
VIDEO_TOTAL_PIXELS = int(float(os.environ.get('VIDEO_TOTAL_PIXELS', 131072 * 32 * 32)))

def round_by_factor(number: int, factor: int) -> int:
    """'number' に最も近く、'factor' で割り切れる整数を返します。"""
    return round(number / factor) * factor

def ceil_by_factor(number: int, factor: int) -> int:
    """'number' 以上で、'factor' で割り切れる最小の整数を返します。"""
    return math.ceil(number / factor) * factor

def floor_by_factor(number: int, factor: int) -> int:
    """'number' 以下で、'factor' で割り切れる最大の整数を返します。"""
    return math.floor(number / factor) * factor

def extract_vision_info(conversations):
    vision_infos = []
    if isinstance(conversations[0], dict):
        conversations = [conversations]
    for conversation in conversations:
        for message in conversation:
            if isinstance(message["content"], list):
                for ele in message["content"]:
                    if (
                        "image" in ele
                        or "image_url" in ele
                        or "video" in ele
                        or ele.get("type","") in ("image", "image_url", "video")
                    ):
                        vision_infos.append(ele)
    return vision_infos

def smart_nframes(ele,total_frames,video_fps):
    """抽出されるビデオフレームの数を計算します。

    Args:
        ele (dict): ビデオ設定を含む辞書。
            - fps: モデル用に抽出される入力フレームの数を制御します。
        total_frames (int): ビデオの元の総フレーム数。
        video_fps (int | float): ビデオの元のフレームレート。

    Raises:
        nframes が [FRAME_FACTOR, total_frames] の範囲内にない場合、エラーが報告されます。

    Returns:
        モデル入力用のビデオフレーム数。
    """
    assert not ("fps" in ele and "nframes" in ele), "Only accept either `fps` or `nframes`"
    fps = ele.get("fps", FPS)
    min_frames = ceil_by_factor(ele.get("min_frames", FPS_MIN_FRAMES), FRAME_FACTOR)
    max_frames = floor_by_factor(ele.get("max_frames", min(FPS_MAX_FRAMES, total_frames)), FRAME_FACTOR)
    duration = total_frames / video_fps if video_fps != 0 else 0
    if duration-int(duration)>(1/fps):
        total_frames = math.ceil(duration * video_fps)
    else:
        total_frames = math.ceil(int(duration)*video_fps)
    nframes = total_frames / video_fps * fps
    if nframes > total_frames:
        logger.warning(f"smart_nframes: nframes[{nframes}] > total_frames[{total_frames}]")
    nframes = int(min(min(max(nframes, min_frames), max_frames), total_frames))
    if not (FRAME_FACTOR <= nframes and nframes <= total_frames):
        raise ValueError(f"nframes should in interval [{FRAME_FACTOR}, {total_frames}], but got {nframes}.")

    return nframes

def get_video(video_path):
    # ビデオ情報を取得
    cap = cv2.VideoCapture(video_path)

    frame_width = int(cap.get(cv2.CAP_PROP_FRAME_WIDTH))
    # ビデオの高さを取得
    frame_height = int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT))
    total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT))

    video_fps = cap.get(cv2.CAP_PROP_FPS)
    return frame_height, frame_width, total_frames, video_fps

def smart_resize(ele, path, factor=IMAGE_FACTOR):
    # ビデオの元の幅と高さを取得
    height, width, total_frames, video_fps = get_video(path)
    # ビデオフレームのトークン下限
    min_pixels = VIDEO_MIN_PIXELS
    total_pixels = VIDEO_TOTAL_PIXELS
    # 抽出されるビデオフレーム数
    nframes = smart_nframes(ele, total_frames, video_fps)
    max_pixels = max(min(VIDEO_MAX_PIXELS, total_pixels / nframes * FRAME_FACTOR),int(min_pixels * 1.05))

    # ビデオのアスペクト比は 200:1 または 1:200 を超えてはなりません。
    if max(height, width) / min(height, width) > MAX_RATIO:
        raise ValueError(
            f"absolute aspect ratio must be smaller than {MAX_RATIO}, got {max(height, width) / min(height, width)}"
        )

    h_bar = max(factor, round_by_factor(height, factor))
    w_bar = max(factor, round_by_factor(width, factor))
    if h_bar * w_bar > max_pixels:
        beta = math.sqrt((height * width) / max_pixels)
        h_bar = floor_by_factor(height / beta, factor)
        w_bar = floor_by_factor(width / beta, factor)
    elif h_bar * w_bar < min_pixels:
        beta = math.sqrt(min_pixels / (height * width))
        h_bar = ceil_by_factor(height * beta, factor)
        w_bar = ceil_by_factor(width * beta, factor)
    return h_bar, w_bar

def token_calculate(video_path, fps):
    # ビデオパスと fps フレーム抽出パラメーターを渡します。
    messages = [{"content": [{"video": video_path, "fps": fps}]}]
    vision_infos = extract_vision_info(messages)[0]

    resized_height, resized_width = smart_resize(vision_infos, video_path)

    height, width, total_frames, video_fps = get_video(video_path)
    num_frames = smart_nframes(vision_infos, total_frames, video_fps)
    print(f"元のビデオ寸法：{height}*{width}, モデル入力寸法：{resized_height}*{resized_width}, ビデオ総フレーム数：{total_frames}, fps が {fps} の場合に抽出される総フレーム数：{num_frames}", end=", ")
    video_token = int(math.ceil(num_frames / 2) * resized_height / 32 * resized_width / 32)
    video_token += 2   # システムは自動的に <|vision_bos|> と <|vision_eos|> の視覚マーカー (各 1 トークン) を追加します。
    return video_token

video_token = token_calculate("xxx/test.mp4", 1)
print("ビデオトークン：", video_token)

# 使用前にインストールしてください：pip install Pillow
import math
import os
import logging
from typing import Tuple
from PIL import Image

logger = logging.getLogger(__name__)

# ==================== 定数定義 ====================
FRAME_FACTOR = 2
# Qwen3-VL、qwen-vl-max-0813、qwen-vl-plus-0815、qwen-vl-plus-0710 などのモデルでは、スケーリング係数は 32 です。
IMAGE_FACTOR = 32

# その他のモデルでは、スケーリング係数は 28 です。
# IMAGE_FACTOR = 28

# トークン計算用の定数
TOKEN_DIVISOR = 32  # トークン計算の除数
VISION_SPECIAL_TOKENS = 2  # <|vision_bos|> および <|vision_eos|> マーカー

# ビデオフレームの最大アスペクト比
MAX_RATIO = 200
# ビデオフレームのピクセル下限
VIDEO_MIN_PIXELS = 4 * 32 * 32
# ビデオフレームのピクセル上限。Qwen3-VL-Plus モデルの場合、VIDEO_MAX_PIXELS は 640 * 32 * 32 です。その他のモデルでは 768 * 32 * 32 です。
VIDEO_MAX_PIXELS = 640 * 32 * 32

# ビデオ入力の最大ピクセル値。Qwen3-VL-Plus モデルの場合、VIDEO_TOTAL_PIXELS を 131072 * 32 * 32 に設定します。その他のモデルでは 65536 * 32 * 32 に設定します。
VIDEO_TOTAL_PIXELS = int(float(os.environ.get('VIDEO_TOTAL_PIXELS', 131072 * 32 * 32)))

def round_by_factor(number: int, factor: int) -> int:
    """'number' に最も近く、'factor' で割り切れる整数を返します。"""
    return round(number / factor) * factor

def ceil_by_factor(number: int, factor: int) -> int:
    """'number' 以上で、'factor' で割り切れる最小の整数を返します。"""
    return math.ceil(number / factor) * factor

def floor_by_factor(number: int, factor: int) -> int:
    """'number' 以下で、'factor' で割り切れる最大の整数を返します。"""
    return math.floor(number / factor) * factor

def get_image_size(image_path: str) -> Tuple[int, int]:
    if not os.path.exists(image_path):
        raise FileNotFoundError(f"画像ファイルが見つかりません：{image_path}")

    try:
        image = Image.open(image_path)
        height = image.height
        width = image.width
        image.close()  # ファイルを速やかに閉じます
        return height, width
    except Exception as e:
        raise ValueError(f"画像ファイル {image_path} を読み取れません：{str(e)}")

def smart_resize(height: int, width: int, nframes: int, factor: int = IMAGE_FACTOR) -> Tuple[int, int]:
    """
    画像のスケーリング後の寸法を計算します

    Args:
        height: 元の画像の高さ
        width: 元の画像の幅
        nframes: ビデオフレーム数
        factor: スケーリング係数、デフォルトは IMAGE_FACTOR

    Returns:
        (resized_height, resized_width) スケーリング後の高さと幅

    Raises:
        ValueError: アスペクト比が制限を超えた場合
    """
    # ビデオフレームのトークン下限
    min_pixels = VIDEO_MIN_PIXELS
    total_pixels = VIDEO_TOTAL_PIXELS
    # 抽出されるビデオフレーム数
    max_pixels = max(min(VIDEO_MAX_PIXELS, total_pixels / nframes * FRAME_FACTOR), int(min_pixels * 1.05))

    # ビデオのアスペクト比は 200:1 または 1:200 を超えてはなりません。
    aspect_ratio = max(height, width) / min(height, width)
    if aspect_ratio > MAX_RATIO:
        raise ValueError(
            f"画像のアスペクト比は {MAX_RATIO}:1 未満でなければなりませんが、現在は {aspect_ratio:.2f}:1 です"
        )

    h_bar = max(factor, round_by_factor(height, factor))
    w_bar = max(factor, round_by_factor(width, factor))
    if h_bar * w_bar > max_pixels:
        beta = math.sqrt((height * width) / max_pixels)
        h_bar = floor_by_factor(height / beta, factor)
        w_bar = floor_by_factor(width / beta, factor)
    elif h_bar * w_bar < min_pixels:
        beta = math.sqrt(min_pixels / (height * width))
        h_bar = ceil_by_factor(height * beta, factor)
        w_bar = ceil_by_factor(width * beta, factor)
    return h_bar, w_bar

def calculate_video_tokens(image_path: str, nframes: int = 1, factor: int = IMAGE_FACTOR, verbose: bool = True) -> int:
    """

    Args:
        image_path: ビデオフレームファイルへのパス
        nframes: ビデオフレーム数
        factor: スケーリング係数、デフォルトは IMAGE_FACTOR
        verbose: 詳細情報を表示するかどうか

    Returns:
        消費されるトークン数

    Raises:
        FileNotFoundError: ファイルが存在しない場合
        ValueError: ファイル形式が無効、またはアスペクト比が制限を超えている場合
    """
    # 元の画像寸法を取得 (一度だけ読み取り)
    height, width = get_image_size(image_path)

    # スケーリング後の寸法を計算
    resized_height, resized_width = smart_resize(height, width, nframes, factor)

    # トークン数を計算
    # 計算式：ceil(nframes/2) * (height/TOKEN_DIVISOR) * (width/TOKEN_DIVISOR) + VISION_SPECIAL_TOKENS
    video_token = int(
        math.ceil(nframes / 2) *
        (resized_height / TOKEN_DIVISOR) *
        (resized_width / TOKEN_DIVISOR)
    )
    # 視覚マーカートークン (<|vision_bos|> と <|vision_eos|>) を追加
    video_token += VISION_SPECIAL_TOKENS

    if verbose:
        print(f"元のビデオフレーム寸法：{height}x{width}, モデル入力寸法：{resized_height}x{resized_width}, ", end="")

    return video_token

if __name__ == "__main__":
    try:
        video_token = calculate_video_tokens("xxx/test.jpg", nframes=30)
        print(f"ビデオトークン：{video_token}\n")
    except Exception as e:
        print(f"エラー：{str(e)}\n")