# AnyFlow

来自 NVIDIA、新加坡国立大学和麻省理工学院的 [AnyFlow: Any-Step Video Diffusion Model with On-Policy Flow Map Distillation](https://huggingface.co/papers/2605.13724)，作者为 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](https://github.com/NVlabs/AnyFlow) 上获取，项目页面位于 [nvlabs.github.io/AnyFlow](https://nvlabs.github.io/AnyFlow)，预训练模型可在 Hugging Face 的 [nvidia/anyflow](https://huggingface.co/collections/nvidia/anyflow) 集合中找到。

可用模型：

| 检查点 | 骨干 | 描述 |
|---|---|---|
| [`nvidia/AnyFlow-Wan2.1-T2V-1.3B-Diffusers`](https://huggingface.co/nvidia/AnyFlow-Wan2.1-T2V-1.3B-Diffusers) | Wan2.1 1.3B | 双向 T2V |
| [`nvidia/AnyFlow-Wan2.1-T2V-14B-Diffusers`](https://huggingface.co/nvidia/AnyFlow-Wan2.1-T2V-14B-Diffusers) | Wan2.1 14B | 双向 T2V |
| [`nvidia/AnyFlow-FAR-Wan2.1-1.3B-Diffusers`](https://huggingface.co/nvidia/AnyFlow-FAR-Wan2.1-1.3B-Diffusers) | FAR + Wan2.1 1.3B | 因果 T2V / I2V / V2V |
| [`nvidia/AnyFlow-FAR-Wan2.1-14B-Diffusers`](https://huggingface.co/nvidia/AnyFlow-FAR-Wan2.1-14B-Diffusers) | FAR + Wan2.1 14B | 因果 T2V / I2V / V2V |

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

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

```py
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`` 互斥。

> [!IMPORTANT]
> 发布的检查点将 `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`，否则 pipeline 会抛出 `ValueError`。

```py
import torch
from diffusers import AnyFlowFARPipeline
from diffusers.utils import export_to_video

pipe = AnyFlowFARPipeline.from_pretrained(
    "nvidia/AnyFlow-FAR-Wan2.1-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."
)
video = pipe(prompt, num_inference_steps=4, num_frames=81).frames[0]
export_to_video(video, "anyflow_far_t2v.mp4", fps=16)
```

```py
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", dtype=torch.bfloat16
).to("cuda")  # or "mps", "xpu", "cpu"

# Example conditioning image from the AnyFlow repo.
first_frame = load_image(
    "https://raw.githubusercontent.com/NVlabs/AnyFlow/main/assets/evaluation/example/images/1.jpg"
).resize((832, 480))
arr = np.asarray(first_frame).astype("float32") / 255.0  # (480, 832, 3)
context_tensor = torch.from_numpy(arr).permute(2, 0, 1).unsqueeze(0).unsqueeze(1).to("cuda")  # (1, 1, 3, 480, 832)

prompt = (
    "A towering, battle-scarred humanoid robot, reminiscent of a Transformer with powerful, segmented armor "
    "and glowing red optics, walking through the skeletal remains of a city ruin. Twisted metal and shattered "
    "concrete crunch under its heavy steps, as the robot scans the desolate, dust-choked skyline under an dark sky."
)
video = pipe(
    prompt=prompt,
    video=context_tensor,
    num_inference_steps=4,
    num_frames=81,
).frames[0]
export_to_video(video, "anyflow_far_i2v.mp4", fps=16)
```

```py
import numpy as np
import torch
from diffusers import AnyFlowFARPipeline
from diffusers.utils import export_to_video, load_video

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

# Example conditioning clip from the AnyFlow repo — take the first 9 frames (3 latent frames at VAE temporal stride 4).
context_frames = load_video(
    "https://raw.githubusercontent.com/NVlabs/AnyFlow/main/assets/evaluation/example/videos/2.mp4"
)[:9]
arr = np.stack([np.asarray(f.resize((832, 480))) for f in context_frames]).astype("float32") / 255.0
context_tensor = torch.from_numpy(arr).permute(0, 3, 1, 2).unsqueeze(0).to("cuda")  # (1, 9, 3, 480, 832)

prompt = (
    "A focused trail runner's powerful strides through a dense, sun-dappled forest. "
    "The camera tracks alongside, highlighting muscular exertion, sweat, and determined facial expression."
)
video = pipe(
    prompt=prompt,
    video=context_tensor,
    num_inference_steps=4,
    num_frames=81,
    # Override chunk_partition so the first chunk covers exactly the 3 latent context frames.
    chunk_partition=[3, 3, 3, 3, 3, 3, 3],
).frames[0]
export_to_video(video, "anyflow_far_v2v.mp4", fps=16)
```

## 备注

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

## AnyFlowPipeline[[diffusers.AnyFlowPipeline]]

#### diffusers.AnyFlowPipeline[[diffusers.AnyFlowPipeline]]

```python
diffusers.AnyFlowPipeline(tokenizer: AutoTokenizer, text_encoder: UMT5EncoderModel, transformer: AnyFlowTransformer3DModel, vae: AutoencoderKLWan, scheduler: FlowMapEulerDiscreteScheduler)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow.py#L80)

**参数：**

tokenizer ([*AutoTokenizer*])：来自 [google/umt5-xxl](https://huggingface.co/google/umt5-xxl) 的 Tokenizer。

text_encoder ([*UMT5EncoderModel*]) : [google/umt5-xxl](https://huggingface.co/google/umt5-xxl) 文本编码器。

transformer ([*AnyFlowTransformer3DModel*])：双向流图 3D Transformer。

vae ([*AutoencoderKLWan*])：一种将视频编码/解码为潜在表示的 VAE。

scheduler ([*FlowMapEulerDiscreteScheduler*])：Flow-map 采样器。该流水线在每个推理步骤中调用 `scheduler.step(..., timestep, sample, r_timestep)`。

用于 [AnyFlow](https://huggingface.co/papers/2605.13724) 中引入的 flow-map-distilled 检查点的双向文本到视频生成流水线。

AnyFlow learns arbitrary-interval transitions \\(z_t \to z_r\\) rather than the fixed \\(z_t \to z_0\\) mapping of consistency models, so a single distilled checkpoint can be evaluated at 1, 2, 4, 8, 16... NFE without retraining. This pipeline operates over the full video tensor in one bidirectional pass; for chunk-wise autoregressive (causal) generation use `AnyFlowFARPipeline`.

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

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

#### __调用__[[diffusers.AnyFlowPipeline.__call__]]

```python
__call__(prompt: typing.Union[str, typing.List[str]] = None, video: typing.Optional[torch.Tensor] = None, video_latents: typing.Optional[torch.Tensor] = None, negative_prompt: typing.Union[str, typing.List[str]] = None, height: int = 480, width: int = 832, num_frames: int = 81, num_inference_steps: int = 50, sigmas: typing.Optional[typing.List[float]] = None, timesteps: typing.Optional[typing.List[float]] = None, guidance_scale: float = 1.0, num_videos_per_prompt: typing.Optional[int] = 1, generator: typing.Union[torch.Generator, typing.List[torch.Generator], NoneType] = None, latents: typing.Optional[torch.Tensor] = None, prompt_embeds: typing.Optional[torch.Tensor] = None, negative_prompt_embeds: typing.Optional[torch.Tensor] = None, output_type: typing.Optional[str] = 'np', return_dict: bool = True, attention_kwargs: typing.Optional[typing.Dict[str, typing.Any]] = None, callback_on_step_end: typing.Union[typing.Callable[[int, int, typing.Dict], NoneType], diffusers.callbacks.PipelineCallback, diffusers.callbacks.MultiPipelineCallbacks, NoneType] = None, callback_on_step_end_tensor_inputs: typing.List[str] = ['latents'], max_sequence_length: int = 512, use_mean_velocity: bool = True)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow.py#L380)

**参数：**

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`](https://pytorch.org/docs/stable/generated/torch.Generator.html)，用于使生成过程具有确定性。

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](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/attention_processor.py) 中 `self.processor` 下定义的 `AttentionProcessor`。

callback_on_step_end (`Callable`，*可选*)：在每个推理步骤结束时调用的函数或 `PipelineCallback`。有关详细信息，请参阅 [`callbacks`](../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 的调用函数。

示例：
```python
>>> 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[[diffusers.AnyFlowPipeline.encode_prompt]]

```python
encode_prompt(prompt: str | list[str], negative_prompt: str | list[str] | None = None, do_classifier_free_guidance: bool = True, num_videos_per_prompt: int = 1, prompt_embeds: typing.Optional[torch.Tensor] = None, negative_prompt_embeds: typing.Optional[torch.Tensor] = None, max_sequence_length: int = 226, device: typing.Optional[torch.device] = None, dtype: typing.Optional[torch.dtype] = None)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow.py#L180)

**参数：**

prompt (`str` 或 `list[str]`，*可选*)：要编码的提示

negative_prompt (`str` 或 `list[str]`，*可选*)：用于引导图像生成的负面提示词。如果未定义，则必须传入 `negative_prompt_embeds`。当不使用引导时（即 `guidance_scale` 小于 `1` 时），此参数将被忽略。

do_classifier_free_guidance (`bool`，*可选*，默认为 `True`)：是否使用无分类器引导。

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[[diffusers.AnyFlowPipeline.encode_video]]

```python
encode_video(video: Tensor, height: int, width: int)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow.py#L360)

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

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

## AnyFlowFARPipeline[[diffusers.AnyFlowFARPipeline]]

#### diffusers.AnyFlowFARPipeline[[diffusers.AnyFlowFARPipeline]]

```python
diffusers.AnyFlowFARPipeline(tokenizer: AutoTokenizer, text_encoder: UMT5EncoderModel, transformer: AnyFlowFARTransformer3DModel, vae: AutoencoderKLWan, scheduler: FlowMapEulerDiscreteScheduler)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow_far.py#L92)

**参数：**

tokenizer ([*AutoTokenizer*])：来自 [google/umt5-xxl](https://huggingface.co/google/umt5-xxl) 的 Tokenizer。

text_encoder ([*UMT5EncoderModel*]) : [google/umt5-xxl](https://huggingface.co/google/umt5-xxl) 文本编码器。

transformer ([*AnyFlowFARTransformer3DModel*])：FAR 因果流映射 3D Transformer。

vae ([*AutoencoderKLWan*])：一种将视频编码/解码为潜在表示的 VAE。

scheduler （*FlowMapEulerDiscreteScheduler*）：Flow-map 采样器。

Causal 基于 FAR 的文本生成视频 / 图像生成视频 / 视频生成视频流水线，用于 AnyFlow 检查点，由 [AnyFlow](https://huggingface.co/papers/2605.13724) 引入。

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

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

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

FAR 主干网络是 [FAR](https://huggingface.co/papers/2503.19325) 中提出的因果 Wan2.1 变体。推理过程在每个块中采用均值速度形式的普通 Euler 方法，且无需重新加噪。单个蒸馏模型即可支持联合 T2V / I2V / V2V 任务。

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

#### __调用__[[diffusers.AnyFlowFARPipeline.__call__]]

```python
__call__(prompt: typing.Union[str, typing.List[str]] = None, video: typing.Optional[torch.Tensor] = None, video_latents: typing.Optional[torch.Tensor] = None, negative_prompt: typing.Union[str, typing.List[str]] = None, height: int = 480, width: int = 832, num_frames: int = 81, num_inference_steps: int = 50, sigmas: typing.Optional[typing.List[float]] = None, timesteps: typing.Optional[typing.List[float]] = None, guidance_scale: float = 1.0, num_videos_per_prompt: typing.Optional[int] = 1, generator: typing.Union[torch.Generator, typing.List[torch.Generator], NoneType] = None, latents: typing.Optional[torch.Tensor] = None, prompt_embeds: typing.Optional[torch.Tensor] = None, negative_prompt_embeds: typing.Optional[torch.Tensor] = None, output_type: typing.Optional[str] = 'np', return_dict: bool = True, attention_kwargs: typing.Optional[typing.Dict[str, typing.Any]] = None, callback_on_step_end: typing.Union[typing.Callable[[int, int, typing.Dict], NoneType], diffusers.callbacks.PipelineCallback, diffusers.callbacks.MultiPipelineCallbacks, NoneType] = None, callback_on_step_end_tensor_inputs: typing.List[str] = ['latents'], max_sequence_length: int = 512, use_mean_velocity: bool = True, use_kv_cache: bool = True, chunk_partition: typing.Optional[typing.List[int]] = None)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow_far.py#L448)

**参数：**

prompt (`str` 或 `List[str]`，*可选*)：用于引导视频生成的提示词或提示词列表。如果未定义，请改为传入 `prompt_embeds`。

video (`torch.Tensor`, *可选*) : 形状为 `(B, T, C, H, W)` 的预 VAE 条件帧，取值范围为 `[0, 1]`（`T = 4n + 1`）。如果提供，流水线将对它们进行 VAE 编码，并在采样过程中保持对应的潜在前缀固定。与 `video_latents` 互斥。

video_latents (`torch.Tensor`, *可选*) : 采用 FAR 布局 `(B, T_latent, C, H_latent, W_latent)` 的预编码 VAE 潜变量。跳过流水线端的 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](https://github.com/huggingface/diffusers/blob/main/src/diffusers/models/attention_processor.py) 中 `self.processor` 下定义的 `AttentionProcessor`。

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]`, *可选*) : 每个分块的帧数。默认值为 `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 的调用函数。

示例：
```python
>>> 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[[diffusers.AnyFlowFARPipeline.encode_prompt]]

```python
encode_prompt(prompt: str | list[str], negative_prompt: str | list[str] | None = None, do_classifier_free_guidance: bool = True, num_videos_per_prompt: int = 1, prompt_embeds: typing.Optional[torch.Tensor] = None, negative_prompt_embeds: typing.Optional[torch.Tensor] = None, max_sequence_length: int = 226, device: typing.Optional[torch.device] = None, dtype: typing.Optional[torch.dtype] = None)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow_far.py#L197)

**参数：**

prompt (`str` 或 `list[str]`，*可选*)：要编码的提示

negative_prompt (`str` 或 `list[str]`，*可选*)：用于引导图像生成的负面提示词。如果未定义，则必须传入 `negative_prompt_embeds`。当不使用引导时（即 `guidance_scale` 小于 `1` 时），此参数将被忽略。

do_classifier_free_guidance (`bool`，*可选*，默认为 `True`)：是否使用无分类器引导。

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[[diffusers.AnyFlowFARPipeline.encode_video]]

```python
encode_video(video: Tensor, height: int, width: int)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_anyflow_far.py#L380)

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

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

## AnyFlowPipelineOutput[[diffusers.pipelines.anyflow.pipeline_output.AnyFlowPipelineOutput]]

#### diffusers.pipelines.anyflow.pipeline_output.AnyFlowPipelineOutput[[diffusers.pipelines.anyflow.pipeline_output.AnyFlowPipelineOutput]]

```python
diffusers.pipelines.anyflow.pipeline_output.AnyFlowPipelineOutput(frames: Tensor)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/anyflow/pipeline_output.py#L23)

**参数：**

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 流水线的输出类。