#
# 根据 Apache 许可证 2.0 版本（“许可证”）授权
# 您只能在遵守许可证的情况下使用本文件。
# 您可以在以下位置获取许可证副本
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# 除非适用法律要求或双方书面同意，软件
# 根据许可证分发，本软件按“原样”
# 不附带任何明示或暗示的保证或条件。
# 请参阅许可证中关于权限的具体语言
# 许可证下的限制。-->

# Cosmos 3

NVIDIA Cosmos 3 是一款面向物理 AI 的统一世界基础模型（WFM）——一个将世界生成、物理推理和动作生成相结合的单一全模态模型。它取代了早期 Cosmos 版本中独立的 Predict、Reason 和 Transfer 模型：无论您是从事机器人、自动驾驶还是智能空间领域的开发，Cosmos 3 都能为您提供一个用于模拟和理解物理世界的统一基础。

本次发布包含以下内容：

- Hugging Face Hub 上带有模型卡和许可证的模型
- Cosmos 3 Diffusers 集成，用于生成流水线（本页）
- 用于在自有数据上微调 Cosmos 3 的后训练脚本
- 面向物理 AI 的开放合成数据生成（SDG）数据集

## Cosmos 3 的新特性

与之前的 Cosmos 版本相比，最大的变化在于 Cosmos 3 是一个*全模态模型*，基于 Mixture-of-Transformers (MoT) 架构构建。此前，开发者需要分别使用不同的模型来处理世界生成（Predict）、受控生成（Transfer）、场景理解（Reason）以及动作策略生成。Cosmos 3 将所有这些功能统一在一个模型中，能够在单次前向传播中跨模态进行推理和生成。

从一个模型，你可以：

- 根据文本、图像或动作输入（图像转视频、文本转视频、动作条件视频生成）生成物理上合理的视频世界。
- 推理物理属性，例如运动、因果关系和空间关系。
- 根据当前状态预测未来的视频和动作序列。
- 通过结构控制在不同视角和条件下迁移场景 *(即将推出)*。

在底层实现中，单个 `Cosmos3OmniTransformer` 并行运行 Qwen 风格的语言模型与扩散生成路径：文本 token 流经因果式的“理解”流，而视频和音频潜变量则流经双向注意的“生成”流，两者通过 3D 多模态 RoPE 进行连接。有关架构背景，请参阅 [Cosmos World Foundation Model Platform 论文](https://huggingface.co/papers/2501.03575)。

## 可用的检查点

Hub 上发布了两个检查点 —— [`nvidia/Cosmos3-Nano`](https://huggingface.co/nvidia/Cosmos3-Nano)（更小、更快）和 [`nvidia/Cosmos3-Super`](https://huggingface.co/nvidia/Cosmos3-Super)（更大、质量更高）。同一个 pipeline 类支持文本生成图像、文本生成视频、图像生成视频，以及（使用支持声音的检查点时）文本+图像生成带声音的视频 —— 请选择一个仓库，并在以下每个工作流中使用对应模型的选项卡。

> [!TIP]
> 请务必参阅 [调度器指南](../../using-diffusers/schedulers)，了解如何探索调度器速度与质量之间的权衡，并查看 [跨流水线复用组件](../../using-diffusers/loading#reusing-models-in-multiple-pipelines) 部分，了解如何高效地将相同组件加载到多个流水线中。

## FP8 混合 W8A8/W8A16 去噪

Nano、Super 和 Edge 的官方 ModelOpt FP8 检查点已作为独立的 Hub 仓库发布：[`nvidia/Cosmos3-Nano-FP8`](https://huggingface.co/nvidia/Cosmos3-Nano-FP8)、[`nvidia/Cosmos3-Super-FP8`](https://huggingface.co/nvidia/Cosmos3-Super-FP8) 和 [`nvidia/Cosmos3-Edge-FP8`](https://huggingface.co/nvidia/Cosmos3-Edge-FP8)。请加载 FP8 仓库的 `main` 分支。[`nvidia/Cosmos3-Super-Image2Video`](https://huggingface.co/nvidia/Cosmos3-Super-Image2Video) 没有单独的 FP8 仓库；请使用 `revision="fp8"` 进行加载。

所有这些检查点均采用相同的量化方式。**W8A8** 使用 8 位权重和 8 位激活值（即恢复的 ModelOpt GEMM）。**W8A16** 复用相同的 8 位权重，但跳过激活值量化：FP8 权重会被反量化，并在 BF16/FP16/FP32 激活值上执行标准线性运算。

在每一步都运行 W8A8 可能会导致多步视频生成中出现明显的闪烁。因此，视频 Nano、Super、Super-I2V 和 Edge FP8 检查点在 `transformer/config.json` 中声明了一个调度计划：**前 3 步和后 3 步使用 W8A16**，**中间步骤使用 W8A8**。Diffusers 从检查点读取这些计数，而不是硬编码它们。精度在每个调度器步骤中只选择一次，以确保无分类器引导的 cond/uncond 调用相匹配。

图像生成和少步蒸馏检查点不会出现这种闪烁现象，因此 Super-T2I 和蒸馏 4 步 FP8 仓库不声明调度，并在每一步都保持 W8A8。该调度也仅适用于 **ModelOpt FP8**：其他量化后端（例如 TorchAO）保留其原生前向传播。

加载一个 FP8 仓库，其恢复路径与 [ModelOpt 指南](../../quantization/modelopt) 相同。该仓库已包含量化配置。混合精度随后会自动遵循检查点：

```python
import torch
from diffusers import Cosmos3OmniPipeline

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano-FP8",
    dtype=torch.bfloat16,
    device_map="cuda",
)
result = pipe(prompt="...", num_inference_steps=35)
```

两个生成时的选项：

- **默认**（`mixed_precision_format=None`）：如果检查点声明了 `diffusion_step_policy`，则在前/后 N 步中运行 W8A16，在中间步骤中运行原生 W8A8。这是多步 **视频** FP8 的预期配置（相比全程 W8A8，闪烁更少）。蒸馏 4 步和 Super-T2I FP8 省略了该策略，因此默认即为全程 W8A8。
- **`mixed_precision_format="none"`**：将每一步都保持在原生 W8A8 上。速度更快，因为 W8A16 是反量化加 `torch.nn.functional.linear`，而不是恢复的 FP8 GEMM，但多步视频可能会出现闪烁。使用此设置来 A/B 测试调度，或匹配完全量化的基线。

在一台 Blackwell 工作站上，使用默认混合调度时，Cosmos3-Nano-FP8 在 720×1280 / 35 步设置下的性能比使用 `"none"` 时慢约 **27%**（T2I）和 **13%**（49 帧 T2V）。这些数字不代表吞吐量保证。仅当 ModelOpt FP8 检查点没有策略时，才传入 `"fp8"` 以强制使用首尾 N 步调度。

## 提示词上采样

Cosmos 3 是在长且描述性极强的字幕上训练的。为了获得最佳质量，在将短文本提示传入流水线之前，应将其**上采样为特定的 JSON 结构**。该上采样器位于 [cosmos-framework](https://github.com/NVIDIA/cosmos-framework) 包中。

从一个简短的纯文本提示词开始，并将其保存到 `assets/prompt.txt`。对于下面的文生视频示例，原始提示词为 *"A robotic arm is cleaning a plate in a kitchen"*：

```bash
mkdir -p assets
echo "A robotic arm is cleaning a plate in a kitchen" > assets/prompt.txt
```

然后安装该框架并运行上采样器。以下示例使用 Opus-4.6 对文本到视频进行上采样：

```bash
git clone https://github.com/NVIDIA/cosmos-framework.git packages/cosmos-framework
pip install -e packages/cosmos-framework

export PROMPT_UPSAMPLER_ENDPOINT_URL="https://api.anthropic.com/v1/"
export PROMPT_UPSAMPLER_MODEL_NAME="claude-opus-4-6"
export PROMPT_UPSAMPLER_API_TOKEN="<your_token>"

python -m cosmos_framework.inference.prompt_upsampling \
    --input assets/prompt.txt \
    --output assets/example_t2v_prompt.json \
    --mode text2video \
    --endpoint-url "${PROMPT_UPSAMPLER_ENDPOINT_URL}" \
    --model "${PROMPT_UPSAMPLER_MODEL_NAME}" \
    --api-token "${PROMPT_UPSAMPLER_API_TOKEN}" \
    --resolution 720 \
    --aspect-ratio "16,9"
```

将 `--mode` 切换为与目标工作流匹配的值（`text2image`、`text2video` 或 `image2video`）。该命令会将上采样后的提示词以 JSON 数组的形式写入 `--output` 文件（`--input` 中每个非空行对应一个对象）；如果传入 `.jsonl` 路径，则每行输出一个 JSON 对象。对于 `image2video`，还必须通过 `--image-url`（URL 或本地路径）或 `--image-list`（每个提示词对应一张图像）提供条件图像。

为了方便使用，我们提供了预上采样的正向提示词（`assets/example_t2v_prompt.json`）和负向提示词（`assets/negative_prompt.json`），下方的生成示例将使用这些文件。这些示例会加载这些 JSON 文件，并通过 `json.dumps(...)` 以 JSON 字符串的形式将其传递给流水线。

## 文本到视频

仅以文本为条件进行多帧生成。根据目标时长选择 `num_frames` —— 默认的 `num_frames=189` 在 24 FPS 下生成约 7.9 秒的视频。提示词和负面提示词从 [提示词上采样](#prompt-upsampling) 中描述的 JSON 上采样文件中读取。

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_t2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    num_frames=189,
    height=720,
    width=1280,
    num_inference_steps=35,
    guidance_scale=6.0,
    fps=24.0,
)
# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "cosmos3_t2v.mp4", fps=24, macro_block_size=1)
```

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_t2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Super", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    num_frames=189,
    height=720,
    width=1280,
    num_inference_steps=35,
    guidance_scale=6.0,
    fps=24.0,
)
# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "cosmos3_t2v.mp4", fps=24, macro_block_size=1)
```

## 文本到图像

单帧生成。模型仅以文本提示作为条件；请传入 `num_frames=1`。使用 `--mode text2image` 进行上采样以生成 JSON 提示。

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline

# JSON-upsampled prompt (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_t2i_prompt.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)

result = pipe(prompt=json.dumps(json_prompt), num_frames=1, height=720, width=1280)
result.video[0].save("cosmos3_t2i.jpg", format="JPEG", quality=85)
```

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline

# JSON-upsampled prompt (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_t2i_prompt.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Super", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)

result = pipe(prompt=json.dumps(json_prompt), num_frames=1, height=720, width=1280)
result.video[0].save("cosmos3_t2i.jpg", format="JPEG", quality=85)
```

## 图像转视频

通过 `image=` 传入条件图像。该流水线将第 0 帧锚定到所提供的图像，并对其余帧进行去噪。图像会在保持宽高比的同时进行缩放，中心裁剪至所需的输出尺寸，并使用与 uint8 等效的舍入方式进行归一化。使用 `--mode image2video` 进行上采样以生成 JSON 提示。

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.utils import export_to_video, load_image

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_i2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)

image = load_image(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/releases/download/assets/robot_153.jpg"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    image=image,
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
)
# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "cosmos3_i2v.mp4", fps=24, macro_block_size=1)
```

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.utils import export_to_video, load_image

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_i2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Super", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)

image = load_image(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/releases/download/assets/robot_153.jpg"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    image=image,
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
)
# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "cosmos3_i2v.mp4", fps=24, macro_block_size=1)
```

## 视频到视频

通过 `video=` 传入条件片段（例如来自 `load_video`）。该流水线将 `condition_frame_indexes_vision`（默认值为 `[0, 1]`）指定的前部潜在帧锚定到该片段，并对其余部分进行去噪。使用 `condition_video_keep`（`"first"` 或 `"last"`）可选择从较长源片段的哪一端提取条件帧。与其他模式一样，提示词应遵循 [Prompt upsampling](#prompt-upsampling) 中所述的描述性 JSON 结构。

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video, load_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_v2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/vision/robot_pouring.mp4"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    video=video,
    condition_frame_indexes_vision=[0, 1],
    condition_video_keep="first",
    num_frames=189,
    height=720,
    width=1280,
    num_inference_steps=35,
    guidance_scale=6.0,
    fps=24.0,
)
# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "cosmos3_v2v.mp4", fps=24, macro_block_size=1)
```

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video, load_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_v2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Super", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/vision/robot_pouring.mp4"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    video=video,
    condition_frame_indexes_vision=[0, 1],
    condition_video_keep="first",
    num_frames=189,
    height=720,
    width=1280,
    num_inference_steps=35,
    guidance_scale=6.0,
    fps=24.0,
)
# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "cosmos3_v2v.mp4", fps=24, macro_block_size=1)
```

## 带声音的视频到视频

当检查点包含 `sound_tokenizer` 时，请在视频到视频的调用中添加 `enable_sound=True`，以同步生成音频轨道。波形将与视频一起返回，并可通过 [encode_video()](/docs/diffusers/v0.43.0/zh/api/utilities#diffusers.utils.encode_video) 混入 MP4 中。

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import encode_video, load_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_v2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/vision/robot_pouring.mp4"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    video=video,
    condition_frame_indexes_vision=[0, 1],
    condition_video_keep="first",
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    enable_sound=True,
)

encode_video(
    result.video,
    fps=24,
    audio=result.sound,
    audio_sample_rate=pipe.sound_tokenizer.config.sampling_rate,
    output_path="cosmos3_v2v_with_sound.mp4",
)
```

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import encode_video, load_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_v2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Super", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/vision/robot_pouring.mp4"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    video=video,
    condition_frame_indexes_vision=[0, 1],
    condition_video_keep="first",
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    enable_sound=True,
)

encode_video(
    result.video,
    fps=24,
    audio=result.sound,
    audio_sample_rate=pipe.sound_tokenizer.config.sampling_rate,
    output_path="cosmos3_v2v_with_sound.mp4",
)
```

## 带声音的文本生成视频

当检查点包含 `sound_tokenizer` 时，请传入 `enable_sound=True` 以同步生成音频轨道。波形将与视频一起返回，并可通过 [encode_video()](/docs/diffusers/v0.43.0/zh/api/utilities#diffusers.utils.encode_video) 混流到 MP4 中。

这与上文中的文本生成视频示例中的调用相同，只是额外添加了 `enable_sound=True`：

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.utils import encode_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_t2v_sound_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    enable_sound=True,
)

encode_video(
    result.video,
    fps=24,
    audio=result.sound,
    audio_sample_rate=pipe.sound_tokenizer.config.sampling_rate,
    output_path="cosmos3_with_sound.mp4",
)
```

```python
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.utils import encode_video

# JSON-upsampled positive and negative prompts (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_t2v_sound_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt.json"))

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Super", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)

result = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    enable_sound=True,
)

encode_video(
    result.video,
    fps=24,
    audio=result.sound,
    audio_sample_rate=pipe.sound_tokenizer.config.sampling_rate,
    output_path="cosmos3_with_sound.mp4",
)
```

## 基于动作条件的生成

Action runs 将每个动作特定的输入整合为一个 [CosmosActionCondition](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.CosmosActionCondition)，并通过 `action` 参数传递，而不是通过顶层的 `image` / `video` / `height` / `width` 参数。请将 `resolution_tier`（`256`/`480`/`704`/`720`）设置为接近输入视频原生分辨率的值；它将决定条件画布。Cosmos 3 支持三种动作模式——`policy`、`forward_dynamics` 和 `inverse_dynamics`。`policy` 和 `forward_dynamics` 仅以第一帧作为条件（因此 `image` 或 `video` 均可使用），而 `inverse_dynamics` 需要一个 `video`。动作运行中的条件视频设置在 `action.video`（或 `action.image`）上，而不是管道的顶层 `video` 参数上。

将纯任务描述作为 `prompt` 传入，并通过 `action.view_point` 选择相机视角（默认为 `"ego_view"`；还支持 `"third_person_view"`、`"wrist_view"`、`"concat_view"`）。该流水线会将这些输入转换为模型训练所用的结构化 JSON 描述，因此动作提示不应经过 LLM 上采样。

### 行为策略

Action policy generation 根据首个观测帧、文本提示词以及动作域元数据，预测未来的视频和动作 token。下例使用 Bridge 机器人域，并将预测的动作块写入 JSON，其格式为模型归一化动作空间。

```python
import json

import torch
from diffusers import Cosmos3OmniPipeline, CosmosActionCondition
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video, load_video

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

prompt = "Put the pot to the left of the purple item."
video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/action/bridge_20260501_0.mp4"
)

result = pipe(
    prompt=prompt,
    action=CosmosActionCondition(
        mode="policy",
        chunk_size=16,
        domain_name="bridge_orig_lerobot",
        resolution_tier=480,
        video=video,
        view_point="ego_view",
    ),
    fps=5,
    num_inference_steps=30,
    guidance_scale=1.0,
    use_system_prompt=False,
)

# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "sample.mp4", fps=5, macro_block_size=1)

if result.action is not None:
    with open("sample_action.json", "w") as f:
        json.dump(result.action[0].tolist(), f)
```

```python
import json

import torch
from diffusers import Cosmos3OmniPipeline, CosmosActionCondition
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video, load_video

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Super", dtype=torch.bfloat16, device_map="cuda"  # or "mps", "xpu", "cpu"
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

prompt = "Put the pot to the left of the purple item."
video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/action/bridge_20260501_0.mp4"
)

result = pipe(
    prompt=prompt,
    action=CosmosActionCondition(
        mode="policy",
        chunk_size=16,
        domain_name="bridge_orig_lerobot",
        resolution_tier=480,
        video=video,
        view_point="ego_view",
    ),
    fps=5,
    num_inference_steps=30,
    guidance_scale=1.0,
    use_system_prompt=False,
)

# macro_block_size=1 allows arbitrary frame sizes (Cosmos3 outputs are not always divisible by 16).
export_to_video(result.video, "sample.mp4", fps=5, macro_block_size=1)

if result.action is not None:
    with open("sample_action.json", "w") as f:
        json.dump(result.action[0].tolist(), f)
```

## SeaCache

SeaCache 默认处于禁用状态。Cosmos 3 支持通过 [SeaCacheConfig](/docs/diffusers/v0.43.0/zh/api/cache#diffusers.SeaCacheConfig) 显式启用它。当 Spectral-Evolution-Aware 指标变化缓慢时，SeaCache 会复用 transformer 残差，从而减少完整 transformer 执行次数。它从原始视觉潜变量计算该指标，对于图像到视频生成，还包括干净的 conditioning 帧。请通过来自 pipeline 的 scheduler 元数据回调，在 transformer 上启用它：

```python
import torch
from diffusers import Cosmos3OmniPipeline, SeaCacheConfig

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16, device_map="cuda"
)

pipe.transformer.enable_cache(SeaCacheConfig(threshold=0.2, max_consecutive_cached=2))
```

相同的模型级 API 同样适用于 [Cosmos3OmniPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniPipeline)、[Cosmos3OmniModularPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniModularPipeline) 和 [Cosmos3DistilledModularPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3DistilledModularPipeline)。SeaCache 是一种近似方法，可能会改变生成的输出。当需要每个去噪步骤都执行完整的 transformer 时，请使用 `pipe.transformer.disable_cache()` 将其禁用。每次 pipeline 调用后，缓存状态都会重置，并且条件引导分支和无条件引导分支会保持独立的历史记录。

Cosmos 3 将 SeaCache 门控置于其重复解码器层之外，因此与区域编译兼容。在启用缓存后，编译这些层：

```python
pipe.transformer.compile_repeated_blocks(fullgraph=True)
```

SeaCache 还支持下文所述的 Cosmos 3 Ulysses 上下文并行以及基于 DTensor 的张量并行辅助功能。缓存决策会在各个 rank 之间进行同步。我们不对完整模型编译或其他模型分片策略作出承诺。

## 上下文并行

对于长视频或高分辨率，单次前向传播可能会超出单个 GPU 的内存和延迟预算。Cosmos 3 支持 **上下文并行（CP）**，将序列维度分片到多个 GPU 上，拆分注意力计算，使每个设备仅持有部分 token。

Cosmos 3 支持 **Ulysses** 上下文并行（all-to-all 序列/头交换）。不支持 Ring attention。

与大多数 diffusers 模型不同，Cosmos 3 **并未**将 CP 接入 transformer 或声明式的 `enable_parallelism()` 路径：其分组查询注意力、独立的理解/生成流（生成流会同时关注两者）以及各流之间不规则的长度，都无法表示为 `_cp_plan`。相反，该模型暴露了小型的空操作 shard/gather 接口，具体实现位于 [`examples/cosmos3/cosmos_parallel.py`](https://github.com/huggingface/diffusers/blob/main/examples/cosmos3/cosmos_parallel.py)——一个可从头到尾阅读并自行适配的自包含模块。它提供了两个正交且可组合的分片维度：

| 辅助 | 分片 | 用于 |
|---|---|---|
| `enable_cosmos3_context_parallel(transformer, cp_mesh)` | 序列（CP / Ulysses） | 在单 GPU 可容纳的模型（`Nano`）上的延迟 |
| `enable_cosmos3_tensor_parallel(transformer, tp_mesh)` | 权重（TP） | 训练无法装入单个 GPU 的模型(`Super`) |

可以单独使用其中一种，也可以将两者结合，应用于二维 `(tp, cp)` 网格（参见 [使用张量并行适配大型模型](#fitting-large-models-with-tensor-parallelism)）。

Cosmos 3 有两个特定要求：

- 使用 `native` 注意力后端。Cosmos 3 采用分组查询注意力（GQA），而原生 SDPA 后端是唯一接受 `enable_gqa` 的后端（cuDNN 和 flash 会拒绝该参数）。辅助函数会将 KV 头扩展至查询头数量，并以 `enable_gqa=False` 调用 SDPA，从而确保其仍会分发至 flash 内核（若回退到数学实现，则会实例化完整的 `[S, S]` 分数矩阵，在长序列上导致内存溢出）。
- CP（Ulysses）度必须能整除查询头数量（`Nano` 为 32，`Super` 为 64）；对于 TP，该度必须能整除 KV 头数量（8）。理解（文本）和生成（视频/音频）流沿序列维度独立分片，且不规则长度会在内部零填充至世界规模的整数倍。

### 运行

完整的 CLI [`examples/cosmos3/inference_cosmos3.py`](https://github.com/huggingface/diffusers/blob/main/examples/cosmos3/inference_cosmos3.py) 使用 [Cosmos3OmniModularPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniModularPipeline) 并复用这些辅助函数，因此**任意模态**（文生图/视频、图生视频、声音、动作模式）都可以通过 `--tp-degree` / `--cp-degree` 在多 GPU 上运行。使用 [torchrun](https://docs.pytorch.org/docs/stable/elastic/run.html) 启动；`--tp-degree * --cp-degree` 必须等于 `--nproc_per_node`。每个 rank 都会产生相同的输出；rank 0 负责写入。

```bash
# CP only — Nano (fits one GPU); CP degree must divide 32 query heads.
torchrun --nproc_per_node=4 examples/cosmos3/inference_cosmos3.py --model nano --cp-degree 4 --prompt "..."

# TP only — Super; TP degree must divide 64 query heads and 8 KV heads.
torchrun --nproc_per_node=4 examples/cosmos3/inference_cosmos3.py --model super --tp-degree 4 --prompt "..."

# TP + CP — Super, with sound (TP=2 x CP=2 across 4 GPUs).
torchrun --nproc_per_node=4 examples/cosmos3/inference_cosmos3.py \
    --model super --tp-degree 2 --cp-degree 2 --enable-sound --prompt "..."
```

`Super` 约 120 GB 的权重无法装入单张 96 GB 的 GPU，因此需要 TP；`Nano` 可以装入单张 GPU，因此对它的 CP 纯粹是延迟优化。（省略这两个标志即可在单 GPU 上运行。）

### 使用张量并行训练大型模型

CP 对 *激活值*进行分片，但在每个 rank 上复制所有权重，因此不会减少模型的权重占用——如果模型无法装入单张 GPU，仅靠 CP 仍然无法装入。要对 **权重**进行分片，`enable_cosmos3_tensor_parallel(transformer, tp_mesh)` 会在第二个正交的 mesh 轴上应用 Megatron 风格的张量并行：

- 注意力投影和 MLP 投影在 TP 组内按列/行进行分片（`to_q/to_k/to_v` + `add_q/k/v` 以及 MLP 的 `gate/up` 采用列并行；`to_out/to_add_out` 以及 MLP 的 `down` 采用行并行并执行 all-reduce）。每个 rank 最终拥有 `query_heads / tp` 个 query 头和 `kv_heads / tp` 个 KV 头。
- TP 与 CP 在二维 `(tp, cp)` 设备网格上组合使用：TP 对头/权重进行持久化拆分，CP 则在此基础上对序列进行分片。约束条件是 `tp` 必须整除 KV 头数（8），且 `tp * cp` 必须整除查询头数（`Nano` 为 32，`Super` 为 64）。
- 权重会加载到 CPU 上，然后逐层分片到各个 GPU 上，因此完整的模型永远不会在单个设备上完整实例化。

> [!TIP]
> TP 会在每个注意力块和 MLP 块上执行 all-reduce，因此带宽开销较大。在没有 NVLink 的主机上，它是主要的性能瓶颈；建议采用能让权重放下的最小 TP 度，并将剩余的 GPU 用于 CP。

### 在您的自定义模块化流水线中使用

CLI 标志虽然方便，但你也可以直接使用 [Cosmos3OmniModularPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniModularPipeline) 调用辅助函数。在 CPU 上加载流水线配置和组件，在将流水线移至 rank 本地 GPU *之前* 应用 TP，切换到 `native` 后端，然后启用 CP。在此流程中请勿使用 `device_map`：

```python
import os
import sys

import torch
import torch.distributed as dist
from diffusers import Cosmos3OmniModularPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from torch.distributed.device_mesh import init_device_mesh

# Make the helper module importable.
sys.path.insert(0, "examples/cosmos3")
from cosmos_parallel import (
    enable_cosmos3_context_parallel,
    enable_cosmos3_flash_attention,
    enable_cosmos3_tensor_parallel,
)

# torchrun sets RANK / WORLD_SIZE / LOCAL_RANK. Pick tp_degree * cp_degree == world size.
local_rank = int(os.environ["LOCAL_RANK"])
torch.cuda.set_device(local_rank)
dist.init_process_group("nccl")
mesh = init_device_mesh("cuda", (tp_degree, cp_degree), mesh_dim_names=("tp", "cp"))

# Load components on CPU first; a TP-sharded model may not fit one GPU.
pipe = Cosmos3OmniModularPipeline.from_pretrained(model_id)
pipe.load_components(dtype=torch.bfloat16)
pipe.enable_safety_checker()

if tp_degree > 1:
    enable_cosmos3_tensor_parallel(pipe.transformer, mesh["tp"])  # shard weights -> GPUs
pipe.to(f"cuda:{local_rank}")                                     # move the replicated remainder
pipe.transformer.set_attention_backend("native")
if cp_degree > 1:
    enable_cosmos3_context_parallel(pipe.transformer, mesh["cp"])  # shard the sequence
elif tp_degree > 1:
    enable_cosmos3_flash_attention(pipe.transformer)               # GQA-safe dense attention

# Modular pipelines replace components through update_components().
scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)
pipe.update_components(scheduler=scheduler)

# A single output name returns that value; a list returns a dictionary.
outputs = pipe(
    prompt='{"scene":"A robot arm in a kitchen"}',
    num_frames=189,
    height=720,
    width=1280,
    output=["videos", "sound", "sampling_rate", "action"],
)
videos = outputs["videos"]
sound = outputs["sound"]  # None unless sound generation was requested.
action = outputs["action"]  # None unless an action workflow produced actions.
```

仅针对 CP（无权重分片），请使用 1-D 网格：`init_device_mesh("cuda", (world_size,), mesh_dim_names=("cp",))`，并仅调用 `enable_cosmos3_context_parallel`。

`enable_safety_checker()` 用于加载并启用默认检查器；`disable_safety_checker()` 用于显式禁用它。请改用这些 pipeline 方法，而不是任务 pipeline 的 `enable_safety_checker=` 构造参数或 `enable_safety_check=` 调用参数。模块化 pipeline 也不会返回 `Cosmos3OmniPipelineOutput`：如果只需要帧，请使用 `output="videos"`；或者如上所示，使用输出列表及其返回的字典，而不是 `result.video`、`result.sound` 或 `result.action`。

> [!TIP]
> On some multi-GPU topologies the first NCCL all-to-all can hang. If a CP run stalls at the start of the first denoising step, set `NCCL_P2P_DISABLE=1` in the environment before launching `torchrun`.

CP 和 TP 可与上述所有工作流（文本生成视频、图像生成视频、带声音的文本生成视频以及动作条件生成）以及 `Nano` 和 `Super` 检查点组合使用——只需更改流水线构建和并行设置相关的代码行即可。

## 元数据模板

`tokenize_prompt` 会在用户消息中追加简短的元数据句子，以便 LLM 看到模型训练时所使用的条件信息。正向提示词会包含诸如 *"该视频时长为 7.9 秒，帧率为 24 FPS。"* 和 *"该视频分辨率为 720x1280。"* 这样的句子；负向提示词则包含其反向表述（*"… 不是 …"*）。

两者默认均为开启状态。可通过 `__call__` 禁用其中任意一对：

```python
result = pipe(
    prompt=prompt,
    negative_prompt=negative_prompt,
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    add_duration_template=False,    # skip the duration sentence on both prompts
    add_resolution_template=False,  # skip the resolution sentence on both prompts
)
```

当 `num_frames == 1`（图像模式）时，`add_duration_template` 不生效；在这种情况下，仅会追加分辨率相关的句子。

## 安全检查器

Cosmos3 集成了 [`cosmos_guardrail`](https://pypi.org/project/cosmos-guardrail/) 中的 `CosmosSafetyChecker`，并**默认启用**。文本护栏会在生成前拒绝不安全的提示词（抛出 `ValueError`）；视频护栏则对解码后的帧进行处理，要么对检测到的人脸进行像素化，要么拒绝输出。音频输出不受护栏保护。

安装可选依赖项以启用默认检查器：

```
pip install cosmos_guardrail
```

检查器在 NVIDIA 开放模型许可协议下是强制要求的。以下两个标志位用于测试和开发工作流，在这些场景中该防护机制是多余的（例如，输入已被清除，或者您有意在边缘输入上运行流水线）。

**在构造时禁用**（不会实例化任何检查器，因此不会下载或加载任何护栏模型到内存中）：

```python
import torch
from diffusers import Cosmos3OmniPipeline

pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano",
    dtype=torch.bfloat16,
    device_map="cuda",  # or "mps", "xpu", "cpu"
    enable_safety_checker=False,
)
```

**针对单次调用禁用**（检查器保持加载状态——适用于在保持后续调用默认启用的情况下进行一次性绕过）：

```python
result = pipe(
    prompt=prompt,
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    enable_safety_check=False,
)
```

要提供自定义检查器（例如，用于快速测试的空操作子类），请将其作为 `safety_checker=` 传入：

```python
pipe = Cosmos3OmniPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano",
    dtype=torch.bfloat16,
    device_map="cuda",  # or "mps", "xpu", "cpu"
    safety_checker=MyCustomSafetyChecker(),
)
```

## Cosmos3OmniPipeline[[diffusers.Cosmos3OmniPipeline]]

#### diffusers.Cosmos3OmniPipeline[[diffusers.Cosmos3OmniPipeline]]

```python
diffusers.Cosmos3OmniPipeline(transformer: Cosmos3OmniTransformer, text_tokenizer: AutoTokenizer, vae: AutoencoderKLWan, scheduler: UniPCMultistepScheduler, sound_tokenizer: diffusers.models.autoencoders.autoencoder_cosmos3_audio.Cosmos3AVAEAudioTokenizer | None = None, safety_checker: diffusers.pipelines.cosmos.pipeline_cosmos3_omni.CosmosSafetyChecker | None = None, enable_safety_checker: bool = True, default_use_system_prompt: bool = True, use_native_flow_schedule: bool = False)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/cosmos/pipeline_cosmos3_omni.py#L413)

#### decode_sound[[diffusers.Cosmos3OmniPipeline.decode_sound]]

```python
decode_sound(latent: Tensor)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/cosmos/pipeline_cosmos3_omni.py#L517)

将声音潜在表示 `[C, T]` 解码为波形 `[audio_ch, N]`。

添加/移除声音分词器解码器所期望的批次维度。

#### prepare_latents[[diffusers.Cosmos3OmniPipeline.prepare_latents]]

```python
prepare_latents(image: typing.Union[PIL.Image.Image, numpy.ndarray, torch.Tensor, NoneType] = None, video: typing.Union[list[PIL.Image.Image], torch.Tensor, numpy.ndarray, NoneType] = None, condition_frame_indexes_vision: collections.abc.Iterable[int] = (0, 1), condition_video_keep: typing.Literal['first', 'last'] = 'first', num_frames: int | None = None, height: int | None = None, width: int | None = None, fps: float = 24.0, latents: typing.Optional[torch.Tensor] = None, sound_latents: typing.Optional[torch.Tensor] = None, action_latents: typing.Optional[torch.Tensor] = None, generator: typing.Optional[torch.Generator] = None, device: str = 'cuda', dtype: dtype = torch.bfloat16, enable_sound: bool = False, action: CosmosActionCondition | None = None)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/cosmos/pipeline_cosmos3_omni.py#L763)

**返回值：**

初始噪声张量，以及用于视觉、声音和可选动作模态的条件掩码/元数据。

为单个样本构建条件 + 初始噪声。

#### tokenize_prompt[[diffusers.Cosmos3OmniPipeline.tokenize_prompt]]

```python
tokenize_prompt(prompt: str, negative_prompt: str | None = None, num_frames: int = 189, height: int = 720, width: int = 1280, fps: float = 24.0, use_system_prompt: bool | None = None, add_resolution_template: bool = True, add_duration_template: bool = True, action_mode: str | None = None, action_view_point: str | None = None)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/cosmos/pipeline_cosmos3_omni.py#L1132)

**返回值：**

`(cond_input_ids, uncond_input_ids)` — 该样本的 token-id 列表。

应用提示增强模板，并通过已配置的聊天模板对条件/无条件提示进行分词。

此流水线不运行独立的文本编码器：联合的 Cosmos3 变换器会同时接收原始 token ID 以及视觉（以及可选的音频）token。

当 `negative_prompt` 为 `None` 时，将使用空字符串；Cosmos3 文档页面记录了用于 text2video / image2video 的推荐质量控制负面提示词，建议显式传入。启用时，时长和分辨率模板会追加到提示词中，而反向模板会追加到负面提示词中。

当 `action_mode` 被设置时，提示词会被转换为模型训练所使用的结构化动作 JSON 字幕（参见 `_build_action_json_prompt`），并使用 `action_view_point` 作为 framing 字段；由于 JSON 中已包含 duration/fps/resolution/aspect_ratio，因此会跳过扁平化的元数据模板。

- 全部
- __调用__

## Cosmos3OmniModularPipeline

Cosmos 3 也可作为 Modular Diffusers 流水线使用。基于任务的 [Cosmos3OmniPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniPipeline) 仍然可用；模块化流水线与其共存，并覆盖相同的模式（`text2image`、`text2video`、`image2video`、`video2video`、动作条件生成，以及 `transfer`（结构控制），在检查点支持时可选生成声音）。

```python
import torch
from diffusers import Cosmos3OmniModularPipeline

pipe = Cosmos3OmniModularPipeline.from_pretrained(
    "nvidia/Cosmos3-Nano", dtype=torch.bfloat16
)
pipe.load_components(dtype=torch.bfloat16)
pipe.enable_safety_checker()

videos = pipe(
    prompt='{"scene":"A robot arm in a kitchen"}',
    num_frames=1,
    height=720,
    width=1280,
    output="videos",
)

# Modular pipelines expose declared outputs directly instead of using the task pipeline's
# `return_dict`/`Cosmos3OmniPipelineOutput` API.
image = videos[0]
```

你还可以通过 [ModularPipeline](/docs/diffusers/v0.43.0/zh/api/modular_diffusers/pipeline#diffusers.ModularPipeline) 进行加载，并让仓库配置选择 blocks 类：

```python
import torch
from diffusers import ModularPipeline

pipe = ModularPipeline.from_pretrained("nvidia/Cosmos3-Nano", dtype=torch.bfloat16)
pipe.load_components(dtype=torch.bfloat16)
pipe.enable_safety_checker()
videos = pipe(
    prompt='{"scene":"A robot arm in a kitchen"}', num_frames=1, height=720, width=1280, output="videos"
)
```

若要检查或自定义特定的 Cosmos 模块化工作流，请使用 `available_workflows` + `get_workflow()`：

```python
available = pipe.blocks.available_workflows
image2video_blocks = pipe.blocks.get_workflow("image2video")
```

### 所有现有工作流的模块化示例

模块化流水线支持与任务流水线相同的调用签名。以下代码片段对应上文展示的所有生成示例（`text2video`、`text2image`、`image2video`、`video2video`、`video2video_sound`、`text2video_sound` 以及 `action_policy`）。Transfer（结构控制）拥有独立的输入，将在下文 [Modular transfer](#modular-transfer-structural-control) 中单独展示。

```python
import json
import torch
from diffusers import Cosmos3OmniModularPipeline, CosmosActionCondition
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import encode_video, export_to_video, load_image, load_video

pipe = Cosmos3OmniModularPipeline.from_pretrained("nvidia/Cosmos3-Nano", dtype=torch.bfloat16)
pipe.load_components(dtype=torch.bfloat16)
pipe.enable_safety_checker()
pipe.to("cuda")  # or "mps", "xpu", "cpu"
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

# text2video
json_prompt = json.load(open("assets/example_t2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt.json"))
videos = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    num_frames=189,
    height=720,
    width=1280,
    num_inference_steps=35,
    guidance_scale=6.0,
    fps=24.0,
    output="videos",
)
export_to_video(videos, "cosmos3_modular_t2v.mp4", fps=24, macro_block_size=1)

# text2image
json_prompt = json.load(open("assets/example_t2i_prompt.json"))
videos = pipe(prompt=json.dumps(json_prompt), num_frames=1, height=720, width=1280, output="videos")
videos[0].save("cosmos3_modular_t2i.jpg", format="JPEG", quality=85)

# image2video
json_prompt = json.load(open("assets/example_i2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))
image = load_image("https://github.com/nvidia-cosmos/cosmos-dependencies/releases/download/assets/robot_153.jpg")
videos = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    image=image,
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    output="videos",
)
export_to_video(videos, "cosmos3_modular_i2v.mp4", fps=24, macro_block_size=1)

# video2video
json_prompt = json.load(open("assets/example_v2v_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt_i2v.json"))
video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/vision/robot_pouring.mp4"
)
videos = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    video=video,
    condition_frame_indexes_vision=[0, 1],
    condition_video_keep="first",
    num_frames=189,
    height=720,
    width=1280,
    num_inference_steps=35,
    guidance_scale=6.0,
    fps=24.0,
    output="videos",
)
export_to_video(videos, "cosmos3_modular_v2v.mp4", fps=24, macro_block_size=1)

# video2video_sound
outputs = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    video=video,
    condition_frame_indexes_vision=[0, 1],
    condition_video_keep="first",
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    enable_sound=True,
    output=["videos", "sound", "sampling_rate"],
)
encode_video(
    outputs["videos"],
    fps=24,
    audio=outputs["sound"],
    audio_sample_rate=outputs["sampling_rate"],
    output_path="cosmos3_modular_v2v_with_sound.mp4",
)

# text2video_sound
json_prompt = json.load(open("assets/example_t2v_sound_prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt.json"))
outputs = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    num_frames=189,
    height=720,
    width=1280,
    fps=24.0,
    enable_sound=True,
    output=["videos", "sound", "sampling_rate"],
)
encode_video(
    outputs["videos"],
    fps=24,
    audio=outputs["sound"],
    audio_sample_rate=outputs["sampling_rate"],
    output_path="cosmos3_modular_t2v_with_sound.mp4",
)

# action_policy
prompt = "Put the pot to the left of the purple item."
action_video = load_video(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/action/bridge_20260501_0.mp4"
)
outputs = pipe(
    prompt=prompt,
    action=CosmosActionCondition(
        mode="policy",
        chunk_size=16,
        domain_name="bridge_orig_lerobot",
        resolution_tier=480,
        video=action_video,
        view_point="ego_view",
    ),
    fps=5,
    num_inference_steps=30,
    guidance_scale=1.0,
    use_system_prompt=False,
    output=["videos", "action"],
)
export_to_video(outputs["videos"], "cosmos3_modular_action_policy.mp4", fps=5, macro_block_size=1)
if outputs["action"] is not None:
    with open("cosmos3_modular_action_policy.json", "w") as f:
        json.dump(outputs["action"][0].tolist(), f)
```

### 模块化传输（结构控制）

Transfer 遵循一个**预计算的控制视频**（边缘、模糊、深度、分割或世界场景图），通过 `control_videos=` 以 `{hint: video}` 映射的形式传入。它仅支持视频（没有 `image` / `video` / `action` / `enable_sound`），提示词是一个预上采样的 JSON 字幕（参见 [Prompt upsampling](#prompt-upsampling)），长片段以 `num_video_frames_per_chunk` 为块大小自回归生成，并自动拼接。`guidance_scale` 是常规文本 CFG；`control_guidance`（`!= 1.0`）会额外放大控制信号。各提示类型的推荐起始值：

| 提示 | `guidance_scale` | `control_guidance` | `flow_shift` | 几何 |
| --- | --- | --- | --- | --- |
| 边缘 / 模糊 / 深度 | 3.0 | 1.5 | 10.0 | 121 帧 @ 30 FPS |
| 分割 | 3.0 | 2.0 | 10.0 | 121 帧 @ 30 FPS |
| 世界场景（WSM） | 1.0 | 3.0 | 10.0 | 101 帧 @ 10 FPS |

Diffusers 不附带控制资产。现成的资产（每个提示对应一个控制视频和匹配的 `prompt.json`，外加一个共享的 `negative_prompt.json`）位于 [Cosmos cookbook](https://github.com/NVIDIA/cosmos/tree/main/cookbooks/cosmos3/generator/transfer/assets) 中。对于下文的边缘示例，请将其下载到本地的 `assets/` 文件夹中：

```bash
base=https://github.com/NVIDIA/cosmos/raw/refs/heads/main/cookbooks/cosmos3/generator/transfer/assets
mkdir -p assets/edge
curl -sL "$base/edge/control_edge.mp4" -o assets/edge/control_edge.mp4
curl -sL "$base/edge/prompt.json"      -o assets/edge/prompt.json
curl -sL "$base/negative_prompt.json"  -o assets/negative_prompt.json
```

```python
import json
import torch
from diffusers import Cosmos3OmniModularPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video, load_video

pipe = Cosmos3OmniModularPipeline.from_pretrained("nvidia/Cosmos3-Nano", dtype=torch.bfloat16)
pipe.load_components(dtype=torch.bfloat16)
pipe.to("cuda")  # or "mps", "xpu", "cpu"
pipe.scheduler = UniPCMultistepScheduler.from_config(
    pipe.scheduler.config, flow_shift=10.0, use_karras_sigmas=False
)

# Downloaded into assets/ from the Cosmos cookbook (see the curl snippet above).
json_prompt = json.load(open("assets/edge/prompt.json"))
negative_prompt = json.load(open("assets/negative_prompt.json"))
control_edge = load_video("assets/edge/control_edge.mp4")

videos = pipe(
    prompt=json.dumps(json_prompt),
    negative_prompt=json.dumps(negative_prompt),
    control_videos={"edge": control_edge},
    num_frames=121,
    height=720,
    width=1280,
    fps=30.0,
    num_inference_steps=35,
    guidance_scale=3.0,
    control_guidance=1.5,
    output="videos",
)
export_to_video(videos, "cosmos3_modular_transfer_edge.mp4", fps=30, macro_block_size=1)
```

### Distilled （少步）文本到图像和图像到视频[[diffusers.Cosmos3OmniModularPipeline]]

少步蒸馏检查点由 [Cosmos3DistilledModularPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3DistilledModularPipeline)（blocks: `Cosmos3DistilledBlocks`）提供；基础的 [Cosmos3OmniModularPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniModularPipeline) 和 [Cosmos3OmniPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniPipeline) 不支持它们。`num_inference_steps` 固定为 `distilled_sigmas` 流水线配置的长度（来自检查点的 `modular_model_index.json`），并且由于 guidance 已烘焙到权重中，`guidance_scale` 被强制设为 1.0 —— 为其中任意一个传入其他值都会引发错误，而 `negative_prompt` 会被警告并忽略。

FP8 蒸馏检查点（`revision="fp8"`）未声明混合精度策略，因此每个步骤均保持原生 W8A8。

提示词遵循与非蒸馏模型相同的描述性 JSON 结构，因此短文本必须先进行上采样——按照[提示词上采样](#prompt-upsampling)中的说明，使用`--mode text2image`（T2I）或`--mode image2video`（I2V），然后通过`json.dumps(...)`传入 JSON。

```python
import json
import torch
from diffusers import Cosmos3DistilledModularPipeline
from diffusers.utils import export_to_video, load_image

# JSON-upsampled prompt (see "Prompt upsampling" above).
json_prompt = json.load(open("assets/example_t2i_prompt.json"))

repo = "nvidia/Cosmos3-Super-Text2Image-4Step"
pipe = Cosmos3DistilledModularPipeline.from_pretrained(repo, dtype=torch.bfloat16)
pipe.load_components(dtype=torch.bfloat16)
pipe.to("cuda")  # or "mps", "xpu", "cpu"

# text-to-image (distilled)
videos = pipe(
    prompt=json.dumps(json_prompt),
    num_frames=1,
    height=720,
    width=1280,
    output="videos",
)
videos[0].save("cosmos3_distilled_t2i.jpg", format="JPEG", quality=85)

# image-to-video (distilled) — load the I2V repo instead
# JSON-upsampled prompt (see "Prompt upsampling" above); upsampled from the source prompt
# "The right robotic hand picks up the red sphere on the shelf."
json_prompt_i2v = json.load(open("assets/example_i2v_prompt.json"))

repo_i2v = "nvidia/Cosmos3-Super-Image2Video-4Step"
pipe = Cosmos3DistilledModularPipeline.from_pretrained(repo_i2v, dtype=torch.bfloat16)
pipe.load_components(dtype=torch.bfloat16)
pipe.to("cuda")

image = load_image(
    "https://github.com/nvidia-cosmos/cosmos-dependencies/raw/refs/heads/assets/cosmos3/inputs/vision/robot_153.jpg"
)
videos = pipe(
    prompt=json.dumps(json_prompt_i2v),
    image=image,
    num_frames=189,
    height=720,
    width=1280,
    output="videos",
)
export_to_video(videos, "cosmos3_distilled_i2v.mp4", fps=24, macro_block_size=1)
```

#### diffusers.Cosmos3OmniModularPipeline[[diffusers.Cosmos3OmniModularPipeline]]

```python
diffusers.Cosmos3OmniModularPipeline(blocks: diffusers.modular_pipelines.modular_pipeline.ModularPipelineBlocks | None = None, pretrained_model_name_or_path: str | os.PathLike | None = None, components_manager: diffusers.modular_pipelines.components_manager.ComponentsManager | None = None, collection: str | None = None, workflow: str | None = None, modular_config_dict: dict[str, typing.Any] | None = None, config_dict: dict[str, typing.Any] | None = None, **kwargs)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/modular_pipelines/cosmos/modular_pipeline.py#L7)

用于 Cosmos 3 全模态生成的 ModularPipeline。

- 全部
- __调用__

## Cosmos3DistilledModularPipeline[[diffusers.Cosmos3DistilledModularPipeline]]

#### diffusers.Cosmos3DistilledModularPipeline[[diffusers.Cosmos3DistilledModularPipeline]]

```python
diffusers.Cosmos3DistilledModularPipeline(blocks: diffusers.modular_pipelines.modular_pipeline.ModularPipelineBlocks | None = None, pretrained_model_name_or_path: str | os.PathLike | None = None, components_manager: diffusers.modular_pipelines.components_manager.ComponentsManager | None = None, collection: str | None = None, workflow: str | None = None, modular_config_dict: dict[str, typing.Any] | None = None, config_dict: dict[str, typing.Any] | None = None, **kwargs)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/modular_pipelines/cosmos/modular_pipeline.py#L121)

用于蒸馏（少步）Cosmos 3 全模态生成的 ModularPipeline。

蒸馏检查点将无分类器引导（classifier-free guidance）固化到权重中，并按照从管道 `distilled_sigmas` 配置（由 `modular_model_index.json` 填充）读取的固定调度进行采样，因此 `guidance_scale` 和 `num_inference_steps` 是固定的，且不支持 `negative_prompt`。

- 全部
- __调用__

## CosmosActionCondition[[diffusers.CosmosActionCondition]]

#### diffusers.CosmosActionCondition[[diffusers.CosmosActionCondition]]

```python
diffusers.CosmosActionCondition(mode: typing.Literal['policy', 'forward_dynamics', 'inverse_dynamics'], chunk_size: int, domain_name: str, resolution_tier: int = 480, raw_actions: typing.Optional[torch.Tensor] = None, image: typing.Union[PIL.Image.Image, numpy.ndarray, torch.Tensor, NoneType] = None, video: typing.Union[list, numpy.ndarray, torch.Tensor, NoneType] = None, view_point: str = 'ego_view')
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/cosmos/pipeline_cosmos3_omni.py#L302)

**参数：**

mode (`str`) : 动作任务。取值为 `"forward_dynamics"`（根据首帧和给定的 `raw_actions` 序列展开未来视频）、`"inverse_dynamics"`（推断连接条件帧的动作）或 `"policy"`（从首帧同时展开未来视频和动作）。

chunk_size (`int`) : 块中的动作转换步数。配对的条件视频跨越 `chunk_size + 1` 帧。

domain_name (`str`) : 具身域，用于选择域感知的动作投影权重。必须是已注册的 Cosmos 3 具身域之一。它还确定了用于切片预测动作的未填充动作宽度，该宽度在内部根据此名称解析（参见 `_EMBODIMENT_TO_RAW_ACTION_DIM`）。

resolution_tier (`int`，默认值为 `480`)：动作条件分辨率的 *tier*（取值之一为 `256`、`480`、`704`、`720`）。该 tier 会选择一个宽高比最接近输入的预定义画布；输入会被缩小（绝不放大）并填充到该画布中用于条件化。这并非输出帧尺寸，输出帧尺寸会跟随输入内容。请将 tier 与输入的原生分辨率相匹配：较低的 tier 会丢失细节，而较高的 tier 不会增加分辨率（不进行放大），会在填充上浪费计算资源，并且会造成训练/推理不匹配，从而可能损害质量。

raw_actions (`torch.Tensor`, *可选*) : 形状为 `[T, raw_action_dim]` 的原始域动作向量，用于驱动 `"forward_dynamics"`。长度小于 `chunk_size` 的序列会重复最后一个动作；更长的序列则会被截断。超出模型 `action_dim` 的通道会被拒绝，而较窄的输入会进行零填充，直至达到 `action_dim`。

image (`PIL.Image.Image`、`np.ndarray` 或 `torch.Tensor`，*可选*)：`"policy"` / `"forward_dynamics"` 的条件帧。与 `video` 互斥。

video (`list`、`np.ndarray` 或 `torch.Tensor`，*可选*)：条件视频，使用 `"inverse_dynamics"` 时必需。对于 `"policy"` / `"forward_dynamics"`，仅使用其第一帧。与 `image` 互斥。

view_point (`str`，默认为 `"ego_view"`)：用于填充动作字幕中 `cinematography.framing` 字段的相机视角标签。取值之一为 `"ego_view"`、`"third_person_view"`、`"wrist_view"` 或 `"concat_view"`。动作模型是在包含此视角句子的结构化 JSON 字幕上训练的；如果标签无法识别，framing 字段将被丢弃（并伴随警告）。

汇总了 Cosmos 3 动作条件生成任务所需的所有输入。

请通过 `action` 参数将此传递给 `Cosmos3OmniPipeline.__call__()`，而不是使用顶层的 `image` / `height` / `width` 参数，这些参数专用于 t2v 和 i2v 运行。

## Cosmos3OmniPipelineOutput[[diffusers.pipelines.cosmos.pipeline_cosmos3_omni.Cosmos3OmniPipelineOutput]]

#### diffusers.pipelines.cosmos.pipeline_cosmos3_omni.Cosmos3OmniPipelineOutput[[diffusers.pipelines.cosmos.pipeline_cosmos3_omni.Cosmos3OmniPipelineOutput]]

```python
diffusers.pipelines.cosmos.pipeline_cosmos3_omni.Cosmos3OmniPipelineOutput(video: Any, sound: typing.Optional[torch.Tensor] = None, action: list[torch.Tensor] | None = None)
```

[源代码](https://github.com/huggingface/diffusers/blob/v0.43.0/src/diffusers/pipelines/cosmos/pipeline_cosmos3_omni.py#L283)

**参数：**

video : 生成的视频。具体类型取决于传递给 pipeline 的 `output_type`：对于 `"pil"`（默认值），为 PIL 帧列表；对于 `"np"`，为形状 `[T, H, W, C]` 的 `np.ndarray`；对于 `"pt"`，为形状 `[T, C, H, W]` 的 `torch.Tensor`；当 `output_type="latent"` 时，为原始潜在张量。

sound : 解码后的音频波形，形状为 `[C, N]`。当 `enable_sound=False` 时为 `None`。

action : 预测的动作标记。除非处于预测动作的动作模式，否则为 `None`。

[Cosmos3OmniPipeline](/docs/diffusers/v0.43.0/zh/api/pipelines/cosmos3#diffusers.Cosmos3OmniPipeline) 的输出数据类。