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 论文。

可用的检查点

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

请务必参阅调度器指南，了解如何探索调度器速度与质量之间的权衡，并查看跨流水线复用组件部分，了解如何高效地将相同组件加载到多个流水线中。

FP8 混合 W8A8/W8A16 去噪

Nano、Super 和 Edge 的官方 ModelOpt FP8 检查点已作为独立的 Hub 仓库发布：nvidia/Cosmos3-Nano-FP8、nvidia/Cosmos3-Super-FP8 和 nvidia/Cosmos3-Edge-FP8。请加载 FP8 仓库的 main 分支。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 指南相同。该仓库已包含量化配置。混合精度随后会自动遵循检查点：

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 包中。

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

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

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

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 秒的视频。提示词和负面提示词从提示词上采样中描述的 JSON 上采样文件中读取。

Nano

太棒了

文本到图像

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

Nano

太棒了

图像转视频

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

Nano

太棒了

视频到视频

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

Nano

太棒了

带声音的视频到视频

当检查点包含 sound_tokenizer 时，请在视频到视频的调用中添加 enable_sound=True，以同步生成音频轨道。波形将与视频一起返回，并可通过 encode_video() 混入 MP4 中。

Nano

太棒了

带声音的文本生成视频

当检查点包含 sound_tokenizer 时，请传入 enable_sound=True 以同步生成音频轨道。波形将与视频一起返回，并可通过 encode_video() 混流到 MP4 中。

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

Nano

太棒了

基于动作条件的生成

Action runs 将每个动作特定的输入整合为一个 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，其格式为模型归一化动作空间。

Nano

太棒了

SeaCache

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

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、Cosmos3OmniModularPipeline 和 Cosmos3DistilledModularPipeline。SeaCache 是一种近似方法，可能会改变生成的输出。当需要每个去噪步骤都执行完整的 transformer 时，请使用 pipe.transformer.disable_cache() 将其禁用。缓存状态会在每次 pipeline 调用后重置，且条件引导分支和非条件引导分支各自维护独立的历史记录。

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

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 —— 一个可从头到尾阅读并自行适配的自包含模块。它提供了两个正交且可组合的分片维度：

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

可以单独使用其中一种，也可以将两者结合，应用于二维 (tp, cp) 网格（参见使用张量并行适配大型模型）。

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 使用 Cosmos3OmniModularPipeline 并复用这些辅助函数，因此任意模态（文生图/视频、图生视频、声音、动作模式）都可以通过 --tp-degree / --cp-degree 在多 GPU 上运行。使用 torchrun 启动；--tp-degree * --cp-degree 必须等于 --nproc_per_node。每个 rank 都会产生相同的输出；rank 0 负责写入。

# 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 对 activations 进行分片，但在每个 rank 上复制所有权重，因此不会减少模型的权重占用——如果模型无法装入单个 GPU，仅使用 CP 仍然无法装入。要对 weights 进行分片，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 上，因此完整的模型永远不会在单个设备上完整实例化。

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

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

CLI 标志虽然方便，但你也可以直接使用 Cosmos3OmniModularPipeline 调用辅助函数。在 CPU 上加载流水线配置和组件，在将流水线移至 rank 本地 GPU 之前应用 TP，切换到 native 后端，然后启用 CP。在此流程中请勿使用 device_map：

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。

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__ 禁用其中任意一对：

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 中的 CosmosSafetyChecker，并默认启用。文本护栏会在生成前拒绝不安全的提示词（抛出 ValueError）；视频护栏则对解码后的帧进行处理，要么对检测到的人脸进行像素化，要么拒绝输出。音频输出不受护栏保护。

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

pip install cosmos_guardrail

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

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

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,
)

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

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

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

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

Cosmos3OmniPipeline

class diffusers.Cosmos3OmniPipeline

< source >

( transformer: Cosmos3OmniTransformertext_tokenizer: AutoTokenizervae: AutoencoderKLWanscheduler: UniPCMultistepSchedulersound_tokenizer: diffusers.models.autoencoders.autoencoder_cosmos3_audio.Cosmos3AVAEAudioTokenizer | None = Nonesafety_checker: diffusers.pipelines.cosmos.pipeline_cosmos3_omni.CosmosSafetyChecker | None = Noneenable_safety_checker: bool = Truedefault_use_system_prompt: bool = Trueuse_native_flow_schedule: bool = False )

decode_sound

< source >

( latent: Tensor )

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

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

prepare_latents

< source >

( image: typing.Union[PIL.Image.Image, numpy.ndarray, torch.Tensor, NoneType] = Nonevideo: typing.Union[list[PIL.Image.Image], torch.Tensor, numpy.ndarray, NoneType] = Nonecondition_frame_indexes_vision: collections.abc.Iterable[int] = (0, 1)condition_video_keep: typing.Literal['first', 'last'] = 'first'num_frames: int | None = Noneheight: int | None = Nonewidth: int | None = Nonefps: float = 24.0latents: typing.Optional[torch.Tensor] = Nonesound_latents: typing.Optional[torch.Tensor] = Noneaction_latents: typing.Optional[torch.Tensor] = Nonegenerator: typing.Optional[torch.Generator] = Nonedevice: str = 'cuda'dtype: dtype = torch.bfloat16enable_sound: bool = Falseaction: CosmosActionCondition | None = None )

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

tokenize_prompt

< source >

( prompt: strnegative_prompt: str | None = Nonenum_frames: int = 189height: int = 720width: int = 1280fps: float = 24.0use_system_prompt: bool | None = Noneadd_resolution_template: bool = Trueadd_duration_template: bool = Trueaction_mode: str | None = Noneaction_view_point: str | None = None )

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

此流水线不运行独立的文本编码器：联合的 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 仍然可用；模块化流水线与其共存，并覆盖相同的模式（text2image、text2video、image2video、video2video、动作条件生成，以及 transfer（结构控制），在检查点支持时可选生成声音）。

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 进行加载，并让仓库配置选择 blocks 类：

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()：

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 中单独展示。

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），长片段以 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 中。对于下文的边缘示例，请将其下载到本地的 assets/ 文件夹中：

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

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)

蒸馏（少步）文本到图像和图像到视频

少步蒸馏检查点由 Cosmos3DistilledModularPipeline（blocks: Cosmos3DistilledBlocks）提供；基础的 Cosmos3OmniModularPipeline 和 Cosmos3OmniPipeline 不支持它们。num_inference_steps 固定为 distilled_sigmas 流水线配置的长度（来自检查点的 modular_model_index.json），并且由于 guidance 已烘焙到权重中，guidance_scale 被强制设为 1.0——为这两个参数传入任何其他值都会引发错误，而 negative_prompt 会被警告并忽略。

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

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

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)

class diffusers.Cosmos3OmniModularPipeline

< source >

( blocks: diffusers.modular_pipelines.modular_pipeline.ModularPipelineBlocks | None = Nonepretrained_model_name_or_path: str | os.PathLike | None = Nonecomponents_manager: diffusers.modular_pipelines.components_manager.ComponentsManager | None = Nonecollection: str | None = Noneworkflow: str | None = Nonemodular_config_dict: dict[str, typing.Any] | None = Noneconfig_dict: dict[str, typing.Any] | None = None**kwargs )

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

全部
调用

Cosmos3DistilledModularPipeline

class diffusers.Cosmos3DistilledModularPipeline

< source >

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

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

全部
调用

CosmosActionCondition

class diffusers.CosmosActionCondition

< source >

( mode: typing.Literal['policy', 'forward_dynamics', 'inverse_dynamics']chunk_size: intdomain_name: strresolution_tier: int = 480raw_actions: typing.Optional[torch.Tensor] = Noneimage: typing.Union[PIL.Image.Image, numpy.ndarray, torch.Tensor, NoneType] = Nonevideo: typing.Union[list, numpy.ndarray, torch.Tensor, NoneType] = Noneview_point: str = 'ego_view' )

参数

mode (str) — 动作任务。取值为 "forward_dynamics"（根据首帧和给定的 raw_actions 序列推演未来视频）、"inverse_dynamics"（推断连接条件帧的动作）或 "policy"（从首帧同时推演未来视频和动作）。
chunk_size (int) — 块中的动作转换步数。配对的 conditioning 视频跨越 chunk_size + 1 帧。
domain_name (str) — 具身智能域，用于选择域感知的动作投影权重。必须是已注册的 Cosmos 3 具身智能域之一。它还确定了用于切片预测动作的未填充动作宽度，该宽度在内部根据此名称解析（参见 _EMBODIMENT_TO_RAW_ACTION_DIM）。
resolution_tier (int，默认为 480) — 动作条件分辨率的层级（取值为 256、480、704 或 720 之一）。该层级会选择一个宽高比最接近输入的预定义画布；输入图像会被缩小（绝不放大）并填充到该画布中，用于条件输入。这并非输出帧的尺寸，输出帧尺寸会跟随输入内容。请将层级与输入的原生分辨率相匹配：较低的层级会丢失细节，而较高的层级不会增加分辨率（不进行放大），却会在填充上浪费计算资源，并且会造成训练与推理的不匹配，从而可能损害质量。
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

class diffusers.pipelines.cosmos.pipeline_cosmos3_omni.Cosmos3OmniPipelineOutput

< source >

( video: Anysound: typing.Optional[torch.Tensor] = Noneaction: list[torch.Tensor] | None = None )

参数

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 — 预测的动作 token。除非处于预测动作的动作模式，否则为 None。

Cosmos3OmniPipeline 的输出数据类。

Update on GitHub