Phantom 根據圖片和文字描述，自動生成一段視頻，并且動作、場景等內容會按照文字描述來呈現

flyfish

視頻生成的實踐效果展示

Phantom 視頻生成的實踐
Phantom 視頻生成的流程
Phantom 視頻生成的命令

Wan2.1 圖生視頻 支持批量生成
Wan2.1 文生視頻 支持批量生成、參數化配置和多語言提示詞管理
Wan2.1 加速推理方法
Wan2.1 通過首尾幀生成視頻

AnyText2 在圖片里玩文字而且還是所想即所得
Python 實現從 MP4 視頻文件中平均提取指定數量的幀

配置

{"task": "s2v-1.3B","size": "832*480","frame_num": 81,"ckpt_dir": "./Wan2.1-T2V-1.3B","phantom_ckpt": "./Phantom-Wan-1.3B/Phantom-Wan-1.3B.pth","offload_model": false,"ulysses_size": 1,"ring_size": 1,"t5_fsdp": false,"t5_cpu": false,"dit_fsdp": false,"use_prompt_extend": false,"prompt_extend_method": "local_qwen","prompt_extend_model": null,"prompt_extend_target_lang": "ch","base_seed": 40,"sample_solver": "unipc","sample_steps": null,"sample_shift": null,"sample_guide_scale": 5.0,"sample_guide_scale_img": 5.0,"sample_guide_scale_text": 7.5
}

參數

一、參數作用解析

1. 任務與模型路徑

• task: "s2v-1.3B"

  • 作用：指定任務類型為 文本到視頻生成（Text-to-Video，T2V），1.3B 表示使用的基礎模型（如 Wan2.1-T2V-1.3B）參數規模為 130億。

• ckpt_dir: "./Wan2.1-T2V-1.3B"

  • 作用：指定基礎模型的權重文件路徑。根據你之前提供的文件夾內容，該路徑下包含：
    • models_t5_umt5-xxl-enc-bf16.pth：T5文本編碼器的權重文件（用于處理文本提示）。
    • Wan2.1_VAE.pth：VAE模型的權重（用于視頻的時空壓縮和重建）。
    • google/umt5-xxl文件夾：可能包含T5模型的結構定義或配置文件。

• phantom_ckpt: "./Phantom-Wan-1.3B/Phantom-Wan-1.3B.pth"

  • 作用：指定 Phantom跨模態對齊模型 的權重路徑，用于鎖定參考圖像的主體特征（如顏色、輪廓），確保生成視頻中主體與參考圖像一致。

2. 視頻生成配置

• size: "832*480"

  • 作用：生成視頻的分辨率，格式為 寬度×高度，因此 832是寬度，480是高度。例如，常見的16:9分辨率中，寬度大于高度。

• frame_num: 81

  • 作用：生成視頻的總幀數。假設幀率為24fps，81幀約為3.375秒的視頻（實際時長取決于幀率設置）。

3. 模型性能與資源配置

• offload_model: false

  • 作用：是否將模型參數卸載到CPU或磁盤以節省GPU內存。設為false時，模型全程運行在GPU內存中，速度更快但需更大顯存。

• ulysses_size: 1 和 ring_size: 1

  • 作用：與分布式訓練（如FSDP）相關的參數，用于多卡并行計算。設為1時，表示 單卡運行，不啟用分布式分片。

• t5_fsdp: false 和 t5_cpu: false

  • t5_fsdp：是否對T5文本編碼器使用全分片數據并行（FSDP），false表示單卡加載T5模型。
  • t5_cpu：是否將T5模型放在CPU上運行，false表示運行在GPU上（推薦，速度更快）。

• dit_fsdp: false

  • 作用：是否對擴散Transformer（DIT，Diffusion Transformer）使用FSDP，false表示單卡運行。

4. 提示與生成控制

• use_prompt_extend: false

  • 作用：是否啟用提示擴展功能（增強文本提示的語義豐富度）。設為false時，直接使用輸入的文本提示，不進行擴展。

• prompt_extend_method: "local_qwen" 和 prompt_extend_model: null

  • 作用：提示擴展方法指定為本地Qwen模型，但prompt_extend_model設為null表示未加載該模型，因此擴展功能實際未啟用。

• prompt_extend_target_lang: "ch"

  • 作用：提示擴展的目標語言為中文，若啟用擴展功能，會將中文提示轉換為更復雜的語義表示。

5. 隨機種子與生成算法

• base_seed: 40
  • 作用：隨機種子，用于復現相同的生成結果。固定種子后，相同提示和參數下生成的視頻內容一致。

二、分辨率寬度確認

• 832*480 中，832是寬度，480是高度。
  • 分辨率的表示規則為 寬度×高度（Width×Height），例如：
    • 1080p是1920×1080（寬1920，高1080），
    • 這里的832×480接近16:9的比例（832÷480≈1.733，接近16:9的1.777）。

采樣參數設置

一、核心參數解析

1. sample_solver: "unipc"

• 作用：指定采樣算法（擴散模型生成視頻的核心求解器）。
  • UniPC（Unified Predictor-Corrector）：一種高效的數值積分方法，適用于擴散模型采樣，兼顧速度與生成質量，支持動態調整步長，在較少步數下可實現較好效果。
  • 對比其他 solver：相比傳統的 DDIM/PLMS 等算法，UniPC 在相同步數下生成細節更豐富，尤其適合視頻生成的時空連貫性優化。

2. sample_steps: 50

• 作用：采樣過程中執行的擴散步數（從噪聲反向生成清晰樣本的迭代次數）。
  • 數值影響：
    • 50步：中等計算量，適合平衡速度與質量。步數不足可能導致細節模糊、動態不連貫；步數過高（如100+）會增加耗時，但收益可能邊際遞減。
    • 建議場景：若追求快速生成，可設為30-50；若需高保真細節（如復雜光影、精細紋理），可嘗試60-80步。

3. sample_shift: 5.0

• 作用控制跨幀生成時的時間步長偏移或運動連貫性約束。
  • 在視頻生成中，相鄰幀的生成需考慮時間序列的連續性，sample_shift 可能用于調整幀間采樣的時間相關性（如抑制突然運動或增強動態平滑度）。
  • 數值較高（如5.0）可能增強幀間約束，減少閃爍或跳躍，但可能限制劇烈動作的表現力；數值較低（如1.0-2.0）允許更自由的動態變化。

4. 引導尺度參數（guide_scale 系列）

引導尺度控制文本提示和參考圖像對生成過程的約束強度，數值越高，生成結果越貼近輸入條件，但可能導致多樣性下降或過擬合。

• sample_guide_scale: 5.0（通用引導尺度）：

  • 全局控制文本+圖像引導的綜合強度，若未單獨設置 img/text 參數，默認使用此值。

• sample_guide_scale_img: 5.0（圖像引導尺度）：

  • 參考圖像對生成的約束強度（適用于圖生視頻 s2v 任務）。
  • 5.0 含義：中等強度，生成內容會保留參考圖像的視覺特征（如顏色、構圖、主體形態），但允許一定程度的變化（如視角調整、動態延伸）。

• sample_guide_scale_text: 7.5（文本引導尺度）：

  • 文本提示對生成的約束強度，數值顯著高于圖像引導（7.5 > 5.0），表明：
    • 優先遵循文本描述：生成內容會嚴格匹配文本語義（如“夕陽下的海灘”“機械恐龍奔跑”），可能犧牲部分圖像參考的細節。
    • 風險與收益：高文本引導可能導致圖像參考的視覺特征（如主體顏色、背景元素）被覆蓋，需確保文本與圖像語義一致（如文本描述需包含圖像中的關鍵視覺元素）。

多組提示詞+多參考圖像輸入

舉例說明

[{"prompt": "內容","image_paths": ["examples/1.jpg","examples/3.jpg"]},{"prompt": "內容","image_paths": ["examples/2.jpg","examples/3.jpg"]},{"prompt": "內容","image_paths": ["examples/3.png","examples/8.jpg"]}
]

一、JSON輸入結構解析

prompt.json包含3組生成任務，每組結構為：

{"prompt": "文本提示詞",       // 描述生成內容的語義（如“貓在草地跳躍”）"image_paths": ["圖1路徑", "圖2路徑"]  // 用于主體對齊的參考圖像列表（支持多圖）
}

關鍵特點：

• 每組任務可包含 1~N張參考圖像（如examples/1.jpg和examples/3.jpg共同定義主體）。
• 多圖輸入時，模型會自動融合多張圖像的特征，適用于需要捕捉主體多角度、多姿態的場景（如生成人物行走視頻時，用正面+側面照片定義體型）。

二、處理多參考圖像

1. 加載與預處理階段（load_ref_images函數）

• 輸入：image_paths列表（如["examples/1.jpg", "examples/3.jpg"]）。
• 處理邏輯：
  1. 逐張加載圖像，轉換為RGB格式。
  2. 對每張圖像進行保持比例縮放+中心填充，統一為目標尺寸（如832×480）：
     • 若圖像寬高比與目標尺寸不一致，先按比例縮放至最長邊等于目標邊，再用白色填充短邊
  3. 輸出ref_images列表，每張圖像為PIL.Image對象，尺寸均為832×480。

2. 模型生成階段（Phantom_Wan_S2V.generate）

• 輸入：ref_images列表（多圖） + prompt文本。
• 核心邏輯：
  1. 多圖特征融合：
     跨模態模型（Phantom-Wan）會提取每張參考圖像的主體特征（如顏色、輪廓），并計算平均特征向量或動態特征融合（根據圖像順序加權），形成對主體的綜合描述。
  2. 動態對齊：
     在生成視頻的每一幀時，模型會同時參考所有輸入圖像的特征，確保主體在不同視角下的一致性（如正面圖像約束面部特征，側面圖像約束身體比例）。

三、例子

場景1：復雜主體多角度定義

• 需求：生成一個“機器人從左側走向右側”的視頻，需要機器人正面和側面外觀一致。
• 輸入：

  {"prompt": "銀色機器人在灰色地面行走，頭部有藍色燈光","image_paths": ["robot_front.jpg", "robot_side.jpg"]  // 正面+側面圖
  }
  

• 效果：
  • 視頻中機器人正面視角時匹配robot_front.jpg的面部細節。
  • 轉向側面時匹配robot_side.jpg的身體輪廓和機械結構。

場景2：主體特征互補

• 需求：修復單張圖像缺失的細節（如證件照生成生活視頻）。
• 輸入：

  {"prompt": "穿藍色襯衫的人在公園跑步，風吹動頭發","image_paths": ["id_photo.jpg", "hair_reference.jpg"]  // 證件照+發型參考圖
  }
  

• 效果：
  • 主體面部和服裝來自證件照，頭發動態和顏色來自hair_reference.jpg，解決證件照中頭發靜止的問題。

場景3：多主體生成

• 需求：生成“兩個人握手”的視頻，兩人外觀分別來自不同圖像。
• 輸入：

  {"prompt": "穿西裝的男人和穿裙子的女人在會議室握手","image_paths": ["man.jpg", "woman.jpg"]  // 兩人的參考圖像
  }
  

• 效果：
  • 模型自動識別圖像中的兩個主體，分別對齊到視頻中的對應人物，確保兩人外觀與參考圖像一致。

。

四、調用示例

1. 終端命令

python main.py --config_file config.json --prompt_file prompt.json

2. 生成結果示例

假設輸入為：

[{"prompt": "戴帽子的狗在雪地里打滾","image_paths": ["dog_front.jpg", "dog_side.jpg"]}
]

生成的視頻中：

• 狗的頭部特征（如眼睛、鼻子）來自dog_front.jpg。
• 身體姿態和帽子形狀來自dog_side.jpg。
• 雪地、打滾動作由文本提示驅動生成。

多參考圖像輸入是Phantom-Wan實現復雜主體動態生成的核心能力之一，通過融合多張圖像的特征。

完整代碼

import argparse
from datetime import datetime
import logging
import os
import sys
import warnings
import json
import time
from uuid import uuid4  # 新增：用于生成唯一標識符warnings.filterwarnings('ignore')import torch, random
import torch.distributed as dist
from PIL import Image, ImageOpsimport phantom_wan
from phantom_wan.configs import WAN_CONFIGS, SIZE_CONFIGS, MAX_AREA_CONFIGS, SUPPORTED_SIZES
from phantom_wan.utils.prompt_extend import DashScopePromptExpander, QwenPromptExpander
from phantom_wan.utils.utils import cache_video, cache_image, str2booldef _validate_args(args):"""參數驗證函數"""# 基礎檢查assert args.ckpt_dir is not None, "請指定檢查點目錄"assert args.phantom_ckpt is not None, "請指定Phantom-Wan檢查點"assert args.task in WAN_CONFIGS, f"不支持的任務: {args.task}"args.base_seed = args.base_seed if args.base_seed >= 0 else random.randint(0, sys.maxsize)# 尺寸檢查["832*480", "480*832"]assert args.size in SUPPORTED_SIZES[args.task], \f"任務{args.task}不支持尺寸{args.size}，支持尺寸：{', '.join(SUPPORTED_SIZES[args.task])}"def _parse_args():"""參數解析函數"""parser = argparse.ArgumentParser(description="使用Phantom生成視頻")parser.add_argument("--config_file", type=str, default="config.json", help="配置JSON文件路徑")parser.add_argument("--prompt_file", type=str, default="prompt.json", help="提示詞JSON文件路徑")args = parser.parse_args()# 從配置文件加載參數with open(args.config_file, 'r') as f:config = json.load(f)for key, value in config.items():setattr(args, key, value)_validate_args(args)return argsdef _init_logging(rank):"""日志初始化函數"""if rank == 0:logging.basicConfig(level=logging.INFO,format="[%(asctime)s] %(levelname)s: %(message)s",handlers=[logging.StreamHandler(stream=sys.stdout)])else:logging.basicConfig(level=logging.ERROR)def load_ref_images(path, size):"""加載參考圖像并預處理"""h, w = size[1], size[0]  # 尺寸格式轉換ref_images = []for image_path in path:with Image.open(image_path) as img:img = img.convert("RGB")img_ratio = img.width / img.heighttarget_ratio = w / h# 保持比例縮放if img_ratio > target_ratio:new_width = wnew_height = int(new_width / img_ratio)else:new_height = hnew_width = int(new_height * img_ratio)img = img.resize((new_width, new_height), Image.Resampling.LANCZOS)# 中心填充至目標尺寸delta_w = w - img.size[0]delta_h = h - img.size[1]padding = (delta_w//2, delta_h//2, delta_w-delta_w//2, delta_h-delta_h//2)new_img = ImageOps.expand(img, padding, fill=(255, 255, 255))ref_images.append(new_img)return ref_imagesdef generate(args):"""主生成函數"""rank = int(os.getenv("RANK", 0))world_size = int(os.getenv("WORLD_SIZE", 1))local_rank = int(os.getenv("LOCAL_RANK", 0))device = local_rank_init_logging(rank)# 分布式環境配置if world_size > 1:torch.cuda.set_device(local_rank)dist.init_process_group(backend="nccl", init_method="env://", rank=rank, world_size=world_size)# 模型并行配置if args.ulysses_size > 1 or args.ring_size > 1:assert args.ulysses_size * args.ring_size == world_size, "ulysses_size與ring_size乘積需等于總進程數"from xfuser.core.distributed import initialize_model_parallel, init_distributed_environmentinit_distributed_environment(rank=dist.get_rank(), world_size=dist.get_world_size())initialize_model_parallel(sequence_parallel_degree=dist.get_world_size(),ring_degree=args.ring_size,ulysses_degree=args.ulysses_size,)# 提示詞擴展初始化prompt_expander = Noneif args.use_prompt_extend:if args.prompt_extend_method == "dashscope":prompt_expander = DashScopePromptExpander(model_name=args.prompt_extend_model, is_vl="i2v" in args.task)elif args.prompt_extend_method == "local_qwen":prompt_expander = QwenPromptExpander(model_name=args.prompt_extend_model,is_vl="i2v" in args.task,device=rank)else:raise NotImplementedError(f"不支持的提示詞擴展方法: {args.prompt_extend_method}")# 模型初始化（僅加載一次）cfg = WAN_CONFIGS[args.task]logging.info(f"初始化模型，任務類型: {args.task}")if "s2v" in args.task:# 視頻生成（參考圖像輸入）wan = phantom_wan.Phantom_Wan_S2V(config=cfg,checkpoint_dir=args.ckpt_dir,phantom_ckpt=args.phantom_ckpt,device_id=device,rank=rank,t5_fsdp=args.t5_fsdp,dit_fsdp=args.dit_fsdp,use_usp=(args.ulysses_size > 1 or args.ring_size > 1),t5_cpu=args.t5_cpu,)elif "t2v" in args.task or "t2i" in args.task:# 文本生成（圖像/視頻）wan = phantom_wan.WanT2V(config=cfg,checkpoint_dir=args.ckpt_dir,device_id=device,rank=rank,t5_fsdp=args.t5_fsdp,dit_fsdp=args.dit_fsdp,use_usp=(args.ulysses_size > 1 or args.ring_size > 1),t5_cpu=args.t5_cpu,)else:# 圖像生成視頻（i2v）wan = phantom_wan.WanI2V(config=cfg,checkpoint_dir=args.ckpt_dir,device_id=device,rank=rank,t5_fsdp=args.t5_fsdp,dit_fsdp=args.dit_fsdp,use_usp=(args.ulysses_size > 1 or args.ring_size > 1),t5_cpu=args.t5_cpu,)# 加載提示詞列表with open(args.prompt_file, 'r') as f:prompts = json.load(f)total_generation_time = 0generation_counter = 0  # 新增：生成計數器防止文件名重復for prompt_info in prompts:prompt = prompt_info["prompt"]image_paths = prompt_info.get("image_paths", [])  # 處理可能不存在的鍵start_time = time.time()# 分布式環境同步種子if dist.is_initialized():base_seed = [args.base_seed] if rank == 0 else [None]dist.broadcast_object_list(base_seed, src=0)args.base_seed = base_seed[0]# 提示詞擴展處理if args.use_prompt_extend and rank == 0:logging.info("正在擴展提示詞...")if "s2v" in args.task or "i2v" in args.task and image_paths:img = Image.open(image_paths[0]).convert("RGB")prompt_output = prompt_expander(prompt, image=img, seed=args.base_seed)else:prompt_output = prompt_expander(prompt, seed=args.base_seed)if not prompt_output.status:logging.warning(f"提示詞擴展失敗: {prompt_output.message}, 使用原始提示詞")input_prompt = promptelse:input_prompt = prompt_output.prompt# 分布式廣播擴展后的提示詞input_prompt = [input_prompt] if rank == 0 else [None]if dist.is_initialized():dist.broadcast_object_list(input_prompt, src=0)prompt = input_prompt[0]logging.info(f"擴展后提示詞: {prompt}")# 執行生成logging.info(f"開始生成，提示詞: {prompt}")if "s2v" in args.task:ref_images = load_ref_images(image_paths, SIZE_CONFIGS[args.size])video = wan.generate(prompt,ref_images,size=SIZE_CONFIGS[args.size],frame_num=args.frame_num,shift=args.sample_shift,sample_solver=args.sample_solver,sampling_steps=args.sample_steps,guide_scale_img=args.sample_guide_scale_img,guide_scale_text=args.sample_guide_scale_text,seed=args.base_seed,offload_model=args.offload_model)elif "t2v" in args.task or "t2i" in args.task:video = wan.generate(prompt,size=SIZE_CONFIGS[args.size],frame_num=args.frame_num,shift=args.sample_shift,sample_solver=args.sample_solver,sampling_steps=args.sample_steps,guide_scale=args.sample_guide_scale,seed=args.base_seed,offload_model=args.offload_model)else:  # i2v任務img = Image.open(image_paths[0]).convert("RGB")video = wan.generate(prompt,img,max_area=MAX_AREA_CONFIGS[args.size],frame_num=args.frame_num,shift=args.sample_shift,sample_solver=args.sample_solver,sampling_steps=args.sample_steps,guide_scale=args.sample_guide_scale,seed=args.base_seed,offload_model=args.offload_model)# 計算生成時間generation_time = time.time() - start_timetotal_generation_time += generation_timelogging.info(f"生成耗時: {generation_time:.2f}秒")# 主進程保存結果if rank == 0:generation_counter += 1  # 計數器遞增timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")safe_prompt = prompt.replace(" ", "_").replace("/", "_")[:50]  # 安全文件名處理file_uuid = str(uuid4())[:8]  # 新增：添加UUID短標識suffix = '.png' if "t2i" in args.task else '.mp4'# 生成唯一文件名save_file = f"{args.task}_{args.size}_{args.ulysses_size}_{args.ring_size}_" \f"{safe_prompt}_{timestamp}_{generation_counter}_{file_uuid}{suffix}"logging.info(f"保存結果到: {save_file}")if "t2i" in args.task:cache_image(tensor=video.squeeze(1)[None],save_file=save_file,nrow=1,normalize=True,value_range=(-1, 1))else:cache_video(tensor=video[None],save_file=save_file,fps=cfg.sample_fps,nrow=1,normalize=True,value_range=(-1, 1))logging.info(f"總生成耗時: {total_generation_time:.2f}秒")logging.info("生成完成")if __name__ == "__main__":args = _parse_args()generate(args)