AnyFlow

来自 NVIDIA、新加坡国立大学和麻省理工学院的 AnyFlow: Any-Step Video Diffusion Model with On-Policy Flow Map Distillation，作者为 Yuchao Gu、Guian Fang、Yuxin Jiang、Weijia Mao、Song Han、Han Cai 和 Mike Zheng Shou。

摘要：AnyFlow 是首个基于流映射构建的任意步数视频扩散框架，它使单个模型（双向或因果）能够适应任意的推理预算。

一致性模型极大地推动了少步视频生成的发展。然而，在任意步视频扩散模型中，由于固定点公式的限制，其性能往往会下降。为了解决这一局限性，我们提出了 AnyFlow，这是首个基于流映射构建的任意步视频扩散蒸馏框架。与仅学习映射 z_t → z_0 不同，AnyFlow 学习在任意时间区间上的转换 z_t → z_r，使单个模型能够适应不同的推理预算。我们设计了一种改进的前向流映射训练方案，将预训练的视频扩散模型微调为流映射模型，并引入了 Flow Map Backward Simulation，以实现对流映射模型的在策略蒸馏。在双向和因果架构上，在 1.3B 到 14B 的规模范围内，针对文本到视频和图像到视频任务的大量实验表明，AnyFlow 在保持高保真度和灵活采样的同时，优于基于一致性的基线方法，且能适应不同的步数预算。

AnyFlow 流水线由 AnyFlow 团队贡献。原始代码可在 GitHub 上获取，项目页面位于 nvlabs.github.io/AnyFlow，预训练模型可在 Hugging Face 的 nvidia/anyflow 集合中找到。

可用模型：

检查点	骨干	描述
`nvidia/AnyFlow-Wan2.1-T2V-1.3B-Diffusers`	Wan2.1 1.3B	双向 T2V
`nvidia/AnyFlow-Wan2.1-T2V-14B-Diffusers`	Wan2.1 14B	双向 T2V
`nvidia/AnyFlow-FAR-Wan2.1-1.3B-Diffusers`	FAR + Wan2.1 1.3B	因果 T2V / I2V / V2V
`nvidia/AnyFlow-FAR-Wan2.1-14B-Diffusers`	FAR + Wan2.1 14B	因果 T2V / I2V / V2V

AnyFlowPipeline 专为文本到视频（T2V）生成中的双向扩散模型而设计。AnyFlowFARPipeline 是一种分块因果扩散模型，支持文本到视频（T2V）生成、图像到视频（I2V）生成以及视频续写（V2V）。

使用 AnyFlow（双向 T2V）进行生成

import torch
from diffusers import AnyFlowPipeline
from diffusers.utils import export_to_video

pipe = AnyFlowPipeline.from_pretrained(
    "nvidia/AnyFlow-Wan2.1-T2V-1.3B-Diffusers", dtype=torch.bfloat16
).to("cuda")  # or "mps", "xpu", "cpu"

prompt = (
    "An astronaut runs smoothly and appears almost weightless on the lunar surface, "
    "as seen from a low-angle shot that highlights the vast, desolate background of the moon. "
    "The moon's craters and rocky terrain are clearly visible, creating a stark contrast against "
    "the running astronaut who moves with graceful, fluid motions."
)
video = pipe(prompt, num_inference_steps=4, num_frames=81).frames[0]
export_to_video(video, "anyflow_t2v.mp4", fps=16)

使用 AnyFlow（FAR 因果）进行生成

因果流水线通过 video（或 video_latents）参数在 T2V / I2V / V2V 之间进行选择：两者均省略即为纯文本生成视频；若需基于现有帧进行条件生成，请传入形状为 (B, T, C, H, W)、取值范围在 [0, 1] 且满足 T = 4n + 1 的 video=<tensor>。I2V 使用单帧作为条件，V2V 续写则使用较长的片段。如果已有符合模型布局的预编码潜变量，可通过 video_latents=<tensor> 传入以跳过 VAE 编码。video 与 video_latents 互斥。

发布的检查点将 chunk_partition=[1, 3, 3, 3, 3, 3, 3, 2]（总和为 21）固化在 transformer 配置中，以匹配标准的 81 个原始帧（在 VAE 时间步长为 4 的情况下，对应 21 个潜在帧）。当您修改 num_frames 时，请传入一个总和为 (num_frames - 1) // 4 + 1 的相应 chunk_partition，否则流水线将抛出 ValueError。

t2v

i2v

v2v

备注

Classifier-free guidance 已融合到发布的检查点中，因此推理时不会执行第二次带引导的前向传播。除非您自己的检查点有特殊要求，否则请保持默认的 guidance_scale=1.0。
FlowMapEulerDiscreteScheduler 是通用型的。您可以通过 from_pretrained(..., scheduler=FlowMapEulerDiscreteScheduler.from_config(...)) 将其附加到任何 flow-map-distilled 检查点。
AnyFlowPipeline 使用 AnyFlowTransformer3DModel（双向）。AnyFlowFARPipeline 使用 AnyFlowFARTransformer3DModel，该模型增加了压缩帧的 patch embedding 以及 FAR 因果块掩码。
LoRA 加载通过 WanLoraLoaderMixin 支持，该 mixin 与上游 Wan 流水线所使用的相同。
有关训练方案（前向流映射训练和在线策略蒸馏），请参阅原始 AnyFlow 训练框架 NVlabs/AnyFlow；训练不在 diffusers 的范围内。

AnyFlowPipeline

类 diffusers.AnyFlowPipeline

< source >

( tokenizer: AutoTokenizertext_encoder: UMT5EncoderModeltransformer: AnyFlowTransformer3DModelvae: AutoencoderKLWanscheduler: FlowMapEulerDiscreteScheduler )

参数

tokenizer（[AutoTokenizer]）—— 来自 google/umt5-xxl 的 Tokenizer。
text_encoder ([UMT5EncoderModel]) — google/umt5-xxl 文本编码器。
transformer ([AnyFlowTransformer3DModel]) — 双向流图 3D Transformer。
vae ([AutoencoderKLWan]) — 用于将视频编码/解码为潜在表示的 VAE。
scheduler（[FlowMapEulerDiscreteScheduler]）—— Flow-map 采样器。该流水线在每次推理步骤中调用 scheduler.step(..., timestep, sample, r_timestep)。

用于 AnyFlow 中引入的 flow-map-distilled 检查点的双向文本到视频生成流水线。

AnyFlow 学习任意区间的转换 $z_t \to z_r$ 而非一致性模型中固定的 $z_t \to z_0$ 映射。因此，单个蒸馏检查点可以在 1、2、4、8、16… NFE 下进行评估，而无需重新训练。此流水线在单次双向传递中处理完整的视频张量；如需分块自回归（因果）生成，请使用 AnyFlowFARPipeline.

采样采用均值速度形式的普通欧拉方法（z_r = z_t - (t - r) * u），不进行重新加噪。发布的 NVIDIA 检查点已将无分类器引导（classifier-free guidance）融合到模型权重中，因此默认的 guidance_scale=1.0 是推荐设置。

该模型继承自 [DiffusionPipeline]。有关为所有流水线实现的通用方法（下载、保存、在特定设备上运行等），请参阅父类文档。

call

< source >

( prompt: typing.Union[str, typing.List[str]] = Nonevideo: typing.Optional[torch.Tensor] = Nonevideo_latents: typing.Optional[torch.Tensor] = Nonenegative_prompt: typing.Union[str, typing.List[str]] = Noneheight: int = 480width: int = 832num_frames: int = 81num_inference_steps: int = 50sigmas: typing.Optional[typing.List[float]] = Nonetimesteps: typing.Optional[typing.List[float]] = Noneguidance_scale: float = 1.0num_videos_per_prompt: typing.Optional[int] = 1generator: typing.Union[torch.Generator, typing.List[torch.Generator], NoneType] = Nonelatents: typing.Optional[torch.Tensor] = Noneprompt_embeds: typing.Optional[torch.Tensor] = Nonenegative_prompt_embeds: typing.Optional[torch.Tensor] = Noneoutput_type: typing.Optional[str] = 'np'return_dict: bool = Trueattention_kwargs: typing.Optional[typing.Dict[str, typing.Any]] = Nonecallback_on_step_end: typing.Union[typing.Callable[[int, int, typing.Dict], NoneType], diffusers.callbacks.PipelineCallback, diffusers.callbacks.MultiPipelineCallbacks, NoneType] = Nonecallback_on_step_end_tensor_inputs: typing.List[str] = ['latents']max_sequence_length: int = 512use_mean_velocity: bool = True ) → ~AnyFlowPipelineOutput or tuple

参数

prompt (str 或 List[str]，可选) — 用于引导视频生成的提示词。如果未定义，请改为传入 prompt_embeds。
video (torch.Tensor, 可选) — 形状为 (B, T, C, H, W)、取值范围在 [0, 1] 内的预 VAE 条件帧。如果提供，流水线将对它们进行 VAE 编码，并在采样过程中保持对应的潜在前缀固定。与 video_latents 互斥。
video_latents (torch.Tensor, 可选) — AnyFlow 布局下的预编码 VAE 潜变量 (B, T_latent, C, H_latent, W_latent)。跳过流水线端的 VAE 编码。与 video 互斥。
negative_prompt (str 或 List[str]，可选) — 视频生成过程中需要避免的提示词。当不使用引导（guidance_scale < 1）时将被忽略。
height (int，默认为 480) — 生成视频的高度（以像素为单位）。
width (int，默认为 832) — 生成视频的宽度（以像素为单位）。
num_frames (int，默认为 81) — 生成视频的帧数。必须满足 (num_frames - 1) % vae_scale_factor_temporal == 0。
num_inference_steps (int，默认为 50) — 去噪步数。蒸馏后的 AnyFlow 检查点支持任意步数采样，因此 1、2、4 或 8 等较低的值也很常见。如果提供了 sigmas 或 timesteps，则此参数将被忽略。
sigmas (List[float]，可选) — 用于任意步数采样的自定义 sigma 调度，取值范围为 [0, 1]，并按从含噪到干净排序。其长度决定了有效的 num_inference_steps；调度器会追加终端 0 sigma。
timesteps (List[float]，可选) — 用于任意步数采样的自定义时间步调度，单位与 self.scheduler.timesteps 相同（即按 num_train_timesteps 进行缩放）。与 sigmas 互斥。
guidance_scale (float，默认为 1.0) — 无分类器引导（Classifier-free guidance）的缩放系数。发布的 AnyFlow 检查点在训练期间已将 CFG 融合到权重中；除非您明确知道您的检查点需要不同的设置，否则请保持为 1.0。
num_videos_per_prompt (int，可选，默认为 1) — 每个提示词生成的视频数量。
generator (torch.Generator 或 List[torch.Generator]，可选) — 用于使生成过程确定性的 torch.Generator。
latents (torch.Tensor, 可选) — 预先生成的含噪潜变量，用作输入。如果未提供，则从给定的 generator 中采样潜变量。
prompt_embeds (torch.Tensor, 可选) — 预生成的文本嵌入。可用于调整文本输入（例如提示词加权）。如果未提供，则从 prompt 生成嵌入。
negative_prompt_embeds (torch.Tensor，可选) — 预生成的负向文本嵌入。
output_type (str，可选，默认为 "np") — 输出格式。取值为 "pil"、"np"、"pt" 或 "latent" 之一。
return_dict (bool，可选，默认为 True) — 是否返回 AnyFlowPipelineOutput 而不是普通元组。
attention_kwargs (dict, 可选) — 一个 kwargs 字典，如果指定，将传递给 diffusers.models.attention_processor 中定义的 AttentionProcessor，该处理器在 self.processor 下定义。
callback_on_step_end (Callable，可选) — 在每个推理步骤结束时调用的函数或 PipelineCallback。有关详细信息，请参阅 callbacks。
callback_on_step_end_tensor_inputs (List[str]，可选，默认为 ["latents"]) — 转发给回调函数的张量输入。必须是 self._callback_tensor_inputs 的子集。
max_sequence_length (int，默认为 512) — 文本编码器的最大序列长度。较长的提示词将被截断。
use_mean_velocity (bool，默认为 True) — 当 True 时，flow-map 模型会同时以源时间步 t 和目标时间步 r 作为条件来预测平均速度，从而与训练时的行为保持一致。将其禁用可模拟原始 Euler 步进（r = t）。

返回值

~AnyFlowPipelineOutput 或 tuple

如果 return_dict 为 True，则返回 AnyFlowPipelineOutput，否则返回一个 tuple，其第一个元素为生成的视频。

用于生成的 pipeline 的调用函数。

示例：

>>> import torch
>>> from diffusers import AnyFlowPipeline
>>> from diffusers.utils import export_to_video

>>> pipe = AnyFlowPipeline.from_pretrained(
...     "nvidia/AnyFlow-Wan2.1-T2V-14B-Diffusers", torch_dtype=torch.bfloat16
... ).to("cuda")

>>> prompt = "A red panda eating bamboo in a forest, cinematic lighting"
>>> video = pipe(prompt, num_inference_steps=4, num_frames=33).frames[0]
>>> export_to_video(video, "anyflow_t2v.mp4", fps=16)

encode_prompt

< source >

( prompt: str | list[str]negative_prompt: str | list[str] | None = Nonedo_classifier_free_guidance: bool = Truenum_videos_per_prompt: int = 1prompt_embeds: typing.Optional[torch.Tensor] = Nonenegative_prompt_embeds: typing.Optional[torch.Tensor] = Nonemax_sequence_length: int = 226device: typing.Optional[torch.device] = Nonedtype: typing.Optional[torch.dtype] = None )

参数

prompt (str 或 list[str]，可选) — 要编码的提示
negative_prompt (str 或 list[str]，可选) — 用于不引导图像生成的提示词。如果未定义，则必须传入 negative_prompt_embeds。在未使用引导时会被忽略（即当 guidance_scale 小于 1 时忽略）。
do_classifier_free_guidance（bool，可选，默认为 True）— 是否使用 classifier free guidance。
num_videos_per_prompt (int，可选，默认为 1) — 每个提示词应生成的视频数量。用于放置结果嵌入的 torch 设备
prompt_embeds (torch.Tensor, 可选) — 预生成的文本嵌入。可用于轻松调整文本输入，例如提示词加权。如果未提供，文本嵌入将从 prompt 输入参数生成。
negative_prompt_embeds (torch.Tensor, 可选) — 预生成的负向文本嵌入。可用于轻松调整文本输入，例如提示词加权。如果未提供，negative_prompt_embeds 将从 negative_prompt 输入参数生成。
device — (torch.device，可选)： torch 设备
dtype — (torch.dtype，可选)： torch 数据类型

将提示词编码为文本编码器的隐藏状态。

encode_video

< source >

( video: Tensorheight: intwidth: int )

将像素空间视频编码为 AnyFlow 的潜在布局。

遵循其他 diffusers 流水线中单一辅助函数的约定（参见 WanImageToVideoPipeline.encode_image）：将预处理、VAE 编码和潜在变量归一化封装为一次调用。输出布局为 (B, T_latent, C, H, W)，这正是 AnyFlow transformer 对条件帧所期望的格式。

AnyFlowFARPipeline

class diffusers.AnyFlowFARPipeline

< source >

( tokenizer: AutoTokenizertext_encoder: UMT5EncoderModeltransformer: AnyFlowFARTransformer3DModelvae: AutoencoderKLWanscheduler: FlowMapEulerDiscreteScheduler )

参数

tokenizer（[AutoTokenizer]）—— 来自 google/umt5-xxl 的 Tokenizer。
text_encoder ([UMT5EncoderModel]) — google/umt5-xxl 文本编码器。
transformer ([AnyFlowFARTransformer3DModel]) — FAR 因果流映射 3D Transformer。
vae ([AutoencoderKLWan]) — 用于将视频编码/解码为潜在表示的 VAE。
scheduler（[FlowMapEulerDiscreteScheduler]）—— Flow-map 采样器。

用于 AnyFlow 检查点的因果（基于 FAR）文本生成视频 / 图像生成视频 / 视频生成视频流水线，在 AnyFlow 中引入。

该流水线驱动一个按块进行的自回归采样循环：每个块通过 flow-map 步骤进行去噪，同时仅通过块稀疏因果注意力关注过去的块，并且中间 KV 缓存在块之间复用。

任务模式（T2V / I2V / V2V）由传递给 __call__ 的条件参数决定：

video=None 和 video_latents=None 均为空——纯文本生成视频。
video=<tensor of shape (B, T, C, H, W) in [0, 1] with T = 4n + 1> — 预 VAE 条件帧；流水线会对它们进行 VAE 编码。对于 I2V，请传入单帧视频；对于 V2V，请传入多帧片段。
video_latents=<latent tensor of shape (B, T_latent, C, H_latent, W_latent)> — 已编码的潜在张量，采用 FAR 布局（跳过 VAE 编码步骤）。

FAR 主干网络是 FAR 引入的因果 Wan2.1 变体。推理过程在每个块（chunk）中采用均值速度形式的普通 Euler 方法，且无需重新加噪。单个蒸馏模型即可支持联合 T2V / I2V / V2V 任务。

该模型继承自 [DiffusionPipeline]。有关为所有流水线实现的通用方法（下载、保存、在特定设备上运行等），请参阅父类文档。

call

< source >

参数

prompt (str 或 List[str]，可选) — 用于引导视频生成的提示词。如果未定义，请改为传入 prompt_embeds。
video (torch.Tensor, 可选) — 形状为 (B, T, C, H, W)、取值范围在 [0, 1] 内的预 VAE 条件帧（T = 4n + 1）。如果提供，流水线将对它们进行 VAE 编码，并在采样过程中保持对应的潜在前缀固定。与 video_latents 互斥。
video_latents (torch.Tensor, 可选) — 预编码的 VAE 潜变量，采用 FAR 布局 (B, T_latent, C, H_latent, W_latent)。跳过流水线侧的 VAE 编码。与 video 互斥。
negative_prompt (str 或 List[str]，可选) — 视频生成过程中需要避免的提示词。当不使用引导（guidance_scale < 1）时将被忽略。
height (int，默认为 480) — 生成视频的高度（以像素为单位）。
width (int，默认为 832) — 生成视频的宽度（以像素为单位）。
num_frames (int，默认为 81) — 生成视频的帧数。必须满足 (num_frames - 1) % vae_scale_factor_temporal == 0。
num_inference_steps (int，默认为 50) — 每个 chunk 的去噪步数。蒸馏后的 AnyFlow-FAR 检查点支持任意步数采样（1、2、4、8、……）。如果提供了 sigmas 或 timesteps，则忽略此参数。
sigmas (List[float]，可选) — 用于任意步数采样的自定义 sigma 调度，取值范围为 [0, 1]，并按从含噪到干净排序。其长度决定了有效的 num_inference_steps；调度器会追加终端 0 sigma。
timesteps (List[float]，可选) — 用于任意步数采样的自定义时间步调度，单位与 self.scheduler.timesteps 相同（即按 num_train_timesteps 进行缩放）。与 sigmas 互斥。
guidance_scale (float，默认为 1.0) — 无分类器引导（Classifier-free guidance）的缩放系数。发布的 AnyFlow 检查点在训练期间已将 CFG 融合到权重中；除非检查点另有要求，否则请保持为 1.0。
num_videos_per_prompt (int，可选，默认为 1) — 每个提示词生成的视频数量。
generator (torch.Generator 或 List[torch.Generator]，可选) — 用于为采样设置种子的生成器。
latents (torch.Tensor, 可选) — 预生成的含噪潜变量。如果未提供，则从给定的 generator 中采样潜变量。
prompt_embeds (torch.Tensor, 可选) — 预生成的文本嵌入。如果未提供，则从 prompt 生成嵌入。
negative_prompt_embeds (torch.Tensor，可选) — 预生成的负向文本嵌入。
output_type (str，可选，默认为 "np") — 输出格式。取值为 "pil"、"np"、"pt" 或 "latent" 之一。
return_dict (bool，可选，默认为 True) — 是否返回 AnyFlowPipelineOutput 而不是普通元组。
attention_kwargs (dict, 可选) — 一个 kwargs 字典，如果指定，将传递给 diffusers.models.attention_processor 中定义的 AttentionProcessor，该处理器在 self.processor 下定义。
callback_on_step_end (Callable，可选) — 在每个推理步骤结束时调用的函数或 PipelineCallback。
callback_on_step_end_tensor_inputs (List[str]，可选，默认为 ["latents"]) — 转发给回调函数的张量输入。必须是 self._callback_tensor_inputs 的子集。
max_sequence_length (int，默认为 512) — 文本编码器的最大序列长度。
use_mean_velocity (bool，默认为 True) — 当 True 时，flow-map 模型会同时以源时间步 t 和目标时间步 r 作为条件，来预测平均速度。将其禁用可模拟原始 Euler 步进。
use_kv_cache (bool，默认为 True) — 在因果块之间复用 FAR 注意力的 KV 缓存。仅在调试时禁用。
chunk_partition (List[int], 可选) — 每个 chunk 的帧数。默认值为 self.transformer.config.chunk_partition（与发布的 81 帧检查点相匹配）。当你更改 num_frames 时，请提供一个总和为 (num_frames - 1) // vae_scale_factor_temporal + 1 的 chunk_partition。

返回值

~AnyFlowPipelineOutput 或 tuple

如果 return_dict 为 True，则返回 AnyFlowPipelineOutput，否则返回一个 tuple，其第一个元素为生成的视频。

用于生成的 pipeline 的调用函数。

示例：

>>> import numpy as np
>>> import torch
>>> from diffusers import AnyFlowFARPipeline
>>> from diffusers.utils import export_to_video, load_image

>>> pipe = AnyFlowFARPipeline.from_pretrained(
...     "nvidia/AnyFlow-FAR-Wan2.1-1.3B-Diffusers", torch_dtype=torch.bfloat16
... ).to("cuda")

>>> # Single-frame I2V: wrap the conditioning image as a (1, 1, 3, H, W) tensor in [0, 1].
>>> first_frame = load_image("path/to/first_frame.png").resize((832, 480))
>>> arr = np.asarray(first_frame).astype("float32") / 255.0
>>> context = torch.from_numpy(arr).permute(2, 0, 1).unsqueeze(0).unsqueeze(1).to("cuda")

>>> video = pipe(
...     prompt="a cat walks across a sunlit lawn",
...     video=context,
...     num_inference_steps=4,
...     num_frames=81,
... ).frames[0]
>>> export_to_video(video, "anyflow_far.mp4", fps=16)

encode_prompt

< source >

参数

prompt (str 或 list[str]，可选) — 要编码的提示
negative_prompt (str 或 list[str]，可选) — 用于不引导图像生成的提示词。如果未定义，则必须传入 negative_prompt_embeds。在未使用引导时会被忽略（即当 guidance_scale 小于 1 时忽略）。
do_classifier_free_guidance（bool，可选，默认为 True）— 是否使用 classifier free guidance。
num_videos_per_prompt (int，可选，默认为 1) — 每个提示词应生成的视频数量。用于放置结果嵌入的 torch 设备
prompt_embeds (torch.Tensor, 可选) — 预生成的文本嵌入。可用于轻松调整文本输入，例如提示词加权。如果未提供，文本嵌入将从 prompt 输入参数生成。
negative_prompt_embeds (torch.Tensor, 可选) — 预生成的负向文本嵌入。可用于轻松调整文本输入，例如提示词加权。如果未提供，negative_prompt_embeds 将从 negative_prompt 输入参数生成。
device — (torch.device，可选)： torch 设备
dtype — (torch.dtype，可选)： torch 数据类型

将提示词编码为文本编码器的隐藏状态。

encode_video

< source >

( video: Tensorheight: intwidth: int )

将像素空间视频编码为 AnyFlow 的潜在布局。

AnyFlowPipelineOutput

class diffusers.pipelines.anyflow.pipeline_output.AnyFlowPipelineOutput

< source >

( frames: Tensor )

参数

frames (torch.Tensor、np.ndarray 或 list[list[PIL.Image.Image]]) — 视频输出列表 - 它可以是一个长度为 batch_size, 的嵌套列表，其中每个子列表包含长度为 num_frames. 的去噪 PIL 图像序列。它也可以是一个形状为 (batch_size, num_frames, channels, height, width) 的 NumPy 数组或 Torch 张量。

AnyFlow 流水线的输出类。

Update on GitHub