芯耀
与非AI
113
转载自公众号：敢敢AUTOHUB
0. 引言
2026年6月1日，NVIDIA正式发布Cosmos 3——一个专为物理AI（Physical AI）设计的开放世界基础模型。与传统多模态模型主要停留在视觉内容理解不同，Cosmos 3把视觉推理、世界生成和动作预测放到同一条工程链路里：模型不仅要看懂画面，还要判断物体之间的物理关系，预测动作之后的未来状态，并在机器人、自动驾驶和工业智能这类真实交互场景中承担世界模型的角色。本文的重点不是复述发布信息，而是沿着“能力边界 -> 架构机制 -> action 建模 -> cosmos / cosmos-framework 工程入口 -> 后训练流程”这条路线，分析哪些代码可以直接参考官方仓库，哪些只能作为应用层伪代码，避免把不存在的类名、配置文件或接口写成确定事实。
1. 物理AI的核心挑战：从识别到闭环控制
1.1 传统AI与物理AI的本质差异
传统人工智能系统处理的是单向数据流：输入数据经模型处理后输出结果。例如，图像分类模型接收图片并输出物体类别与位置；视频理解模型观看视频后输出场景描述或事件摘要。这种范式虽在内容理解、信息检索和决策支持等领域成效显著，但对于需要与物理世界交互的系统而言，单向流程存在根本性局限。物理AI系统必须处理双向交互循环：系统不仅要理解当前环境状态，还要预测自身动作对环境的影响，并根据环境反馈动态调整后续行为。这个循环可形式化为：
其中 表示时刻t的环境状态， 是系统采取的动作， 是环境物理参数， 是任务目标， 是历史观测序列， 是策略函数。这个形式化描述揭示了物理AI的三项核心能力要求：第一是世界状态理解，系统要从视觉、深度、力传感器等多模态输入中提取物体位置、形状、运动状态和空间关系；第二是动作结果预测，系统在执行动作前要估计动作会如何改变环境，例如机械臂从不同角度抓取杯子时是否会滑落、旋转或碰撞；第三是策略优化与执行，系统必须在当前状态、目标和历史观测之间选择动作序列，并同时满足安全约束、控制约束和物理限制。换句话说，物理AI不是“识别以后给一句答案”，而是要把感知、预测和控制放进同一个反馈回路里。
1.2 Cosmos 3在闭环控制中的定位
Cosmos 3并非完整的物理AI系统，而是为闭环控制流程提供基础能力的通用世界模型（World Model）。它的理解能力主要由 Reasoner 承担，用于分析视觉场景中的物理关系、异常事件、任务可行性和潜在风险；生成能力主要由 Generator 承担，用文本、初始图像、视频或动作条件生成未来图像、视频、声音和中间状态；动作建模则把 Action 提升为一级输入输出模态，使模型既能接收动作并预测未来，也能根据观测反推出动作片段。这里的关键是，Cosmos 3不是替代规划器和控制器，而是给闭环系统补上一种可学习的“未来预测器”和“物理语义解释器”。
实际部署中，Cosmos 3通常与传统规划器、控制器和安全系统协同工作。例如，仓库机器人场景中，Cosmos 3负责预测"机器人继续前进是否会与突然出现的工人发生碰撞"，而传统路径规划算法负责生成具体运动轨迹。这种分工使系统既能利用深度学习模型对复杂场景的理解能力，又能保持传统控制算法的可靠性和可解释性。
2. Cosmos 3的架构设计：Mixture-of-Transformers
2.1 双塔架构的设计动机
Cosmos 3采用Mixture-of-Transformers（MoT）架构，这是一种将自回归推理和扩散生成统一于单一模型的创新设计。其核心思想是：不同类型任务需要不同建模方式，但这些任务间存在共享的表示空间。
自回归路径（Reasoner Tower）适合处理因果推理、规划和决策任务。在这些任务中，模型需根据历史信息逐步推导出结论，每个推理步骤都依赖前面的结果。例如，判断"机器人是否应该停止"需要首先识别障碍物，然后评估距离，最后综合安全规则做出决策。这种串行推理过程与自回归生成机制天然匹配。
扩散路径（Generator Tower）适合处理连续数据的生成任务。视频、图像和音频本质上是高维连续分布，直接用离散自回归方式建模会导致效率低下且难以捕捉精细细节。扩散模型通过逐步去噪生成数据，能更好地保持内容的连贯性和物理一致性。
MoT架构的创新之处在于让两条路径共享底层表示学习。具体而言，输入数据（无论文本、图像还是动作）首先被编码为统一向量表示，然后在推理和生成过程中通过交叉注意力机制进行信息交换。这种设计使模型在生成视频时能利用推理模块对场景的语义理解，推理过程也能参考生成模块对未来的预测。
2.2 输入编码与模态融合
Cosmos 3支持文本、图像、视频、音频和动作等模态，但本地 cosmos 仓库并没有暴露一个可直接调用的 CosmosInputEncoder 类。更稳妥的写法，是按官方 README 中公开的输入输出规格来理解模态融合：Reasoner 面向文本输出，Generator 面向图像、视频、声音和动作输出，动作条件由具体 embodiment 决定。
下面这段不是模型内部源码，而是把代码的公开 I/O 规格整理成工程侧配置，方便开发者选择接口和检查请求参数：
COSMOS3_PUBLIC_IO = {
"reasoner": {
"inputs": ["text", "image", "video"],
"outputs": ["text"],
"typical_backends": ["vLLM", "NIM", "Cosmos Framework"],
"api_shape": "OpenAI-compatible /v1/chat/completions",
},
"generator": {
"inputs": ["text", "image", "video", "sound", "action"],
"outputs": ["image", "video", "sound", "action"],
"typical_backends": ["Diffusers", "vLLM-Omni", "Cosmos Framework"],
"video_api_shape": "OpenAI-compatible /v1/videos/sync or /v1/videos",
},
"vision_conditioning": {
"720p": "1280x720",
"480p": "832x480",
"256p": "320x192",
"video_conditioning": "5 frames at the matching resolution",
},
"generation_settings": {
"resolution_tiers": ["256p", "480p", "720p"],
"aspect_ratios": ["16:9", "4:3", "1:1", "3:4", "9:16"],
"fps": [10, 16, 24, 30],
"frame_count": "5 to 300 frames, default 189",
"precision": "BF16 tested",
},
"action_conditioning": {
"camera_motion": "9D",
"autonomous_vehicle": "9D",
"egocentric_motion": "57D",
"single_arm_robot": "10D",
"dual_arm_robot": "20D",
"humanoid_robot": "29D",
},
}上述代码展示了公开 I/O 规格的工程化整理，关键点包括：
公开接口优先：工程接入时不要假设内部 encoder 类名或 token 形状，优先按照 README 和 cookbook 给出的 API 形态构造请求。
视觉条件：官方示例明确给出 720p、480p、256p 的条件分辨率，视频条件通常按匹配分辨率取 5 帧，而不是在文章里自行推导 patch 数。
动作表示：动作维度与 embodiment 绑定，例如 AV 是 9D，DROID/UMI 单臂机器人是 10D。文中后续代码应围绕这些公开动作规格构造请求，而不是自造 joint/cartesian/velocity 分类。
2.3 Reasoner与Generator的协同机制
MoT架构中最关键的设计是推理和生成路径的协同机制。按照 cosmos/README.md 的描述，Reasoner Mode 使用因果自注意力处理语言和视觉理解 token，Generator Mode 对带噪图像、视频、音频和动作 token 做去噪生成；两者共享 Transformer 架构、多模态注意力层和 3D multi-dimensional RoPE 表示。
文章中的代码应该体现“如何选择运行面”，而不是虚构 CosmosModel.forward() 内部实现。下面是工程侧任务路由示例：
def choose_cosmos3_surface(task: str) -> dict:
"""根据任务选择 Cosmos 3 运行面；这是应用层路由，不是模型内部源码。"""
reasoner_tasks = {
"caption",
"temporal_localization",
"physical_plausibility",
"situation_understanding",
"robot_planning",
"2d_grounding",
}
generator_tasks = {
"text_to_image",
"text_to_video",
"image_to_video",
"forward_dynamics",
"inverse_dynamics",
"policy",
}
if task in reasoner_tasks:
return {
"surface": "Reasoner",
"backend": "vLLM or NIM",
"endpoint": "/v1/chat/completions",
"output": "text",
}
if task in generator_tasks:
return {
"surface": "Generator",
"backend": "Diffusers, vLLM-Omni, or Cosmos Framework",
"endpoint": "/v1/videos/sync for audiovisual generation; /v1/videos for action jobs",
"output": "image/video/sound/action",
}
raise ValueError(f"Unknown Cosmos 3 task: {task}")这个流程中的几个关键工程细节：
运行面分离：需要文字解释、物理可行性判断、场景问答时走 Reasoner；需要图像/视频/声音/动作输出时走 Generator。
后端分离：Reasoner 的生产路径常见是 vLLM 或 NIM；Generator 的研究路径常见是 Diffusers，服务路径常见是 vLLM-Omni，原生 PyTorch 路径走 Cosmos Framework。
参数分离：视频生成 cookbook 中常见默认值是 num_frames=189、fps=24、num_inference_steps=35、guidance_scale=6.0、flow_shift=10.0；action cookbook 中常见默认值是 num_inference_steps=30、guidance_scale=1.0、flow_shift=10.0。
3. 动作作为一级模态：从观察到执行的桥梁
3.1 为什么动作需要成为核心能力
传统视觉语言模型的任务止步于理解和描述。即使模型能准确识别"桌上有一个杯子"，也无法回答"如何抓取这个杯子"，因为后者需要输出连续的动作轨迹而非离散的文本描述。Cosmos 3将动作提升为与文本、图像平级的模态，使模型能直接参与物理交互的规划和执行。进一步看，动作建模不是单一任务，而是把“动作作为输入预测未来”和“动作作为输出服务策略”两条路线合在一起，具体体现为正向动力学、逆向动力学和策略建模三类能力。
正向动力学（Forward Dynamics）：给定当前观察 和动作 ，预测下一时刻的观察 ，形式化为：
这种能力可用于评估动作后果。例如，机器人抓取前可预测"从这个角度接近，物体会不会被推倒"。
逆向动力学（Inverse Dynamics）：给定当前观察 和目标观察 ，推断需要执行的动作序列 。这是路径规划的基础。例如，给定"杯子在桌上"和"杯子在机械臂上"两个状态，模型需输出完整的抓取轨迹。
策略建模（Policy Modeling）：结合视觉观察、语言指令和任务上下文，直接输出下一步应执行的动作。这是端到端控制的关键，可表示为：
3.2 动作表示与编码
机器人动作可以用多种方式表示，但 Cosmos 3 action cookbook 中已经给出更具体的定义：动作 token 表示连续视觉状态之间的转移；ego pose 和 end-effector pose 使用 9D pose delta，即 3D 平移加 6D 连续旋转表示；夹爪状态通常是 1D 开合值。因此，文章里的动作代码应该围绕这些公开规格写，而不是泛化成未在 cookbook 中出现的 joint_space、cartesian_space、velocity_control 类。
COSMOS3_ACTION_SPECS = {
"av": {
"representation": "ego pose delta",
"raw_action_dim": 9,
"unit": "meter",
"post_processing": "normalization",
"generation_duration": "60 frames @ 10 FPS",
},
"droid_lerobot": {
"representation": "end-effector pose delta + gripper grasp state",
"raw_action_dim": 10,
"unit": "meter",
"post_processing": "multiview concatenation, to-OpenCV, normalization",
"generation_duration": "16 frames @ 15 FPS",
},
"umi": {
"representation": "end-effector pose delta + gripper grasp state",
"raw_action_dim": 10,
"unit": "meter",
"post_processing": "normalization",
"generation_duration": "16 frames @ 20 FPS",
},
}
def validate_action_chunk(action: list[list[float]], domain_name: str) -> None:
"""校验 action JSON 是否符合 cookbook 中公开的维度约束。"""
spec = COSMOS3_ACTION_SPECS[domain_name]
expected_dim = spec["raw_action_dim"]
if not action:
raise ValueError("action chunk is empty")
if any(len(row) != expected_dim for row in action):
raise ValueError(f"{domain_name} action rows must be {expected_dim}D")Cosmos 3的动作请求需要把动作模式、embodiment 名称、chunk 大小和动作数组一起传入 vLLM-Omni 的 extra_params。下面是对 generator/action/run_fd_with_vllm.ipynb 请求形态的整理：
import json
from pathlib import Path
def build_forward_dynamics_extra_params(record: dict) -> str:
"""构造 vLLM-Omni forward dynamics 请求中的 extra_params。"""
action = json.loads(Path(record["action_path"]).read_text())
validate_action_chunk(action, record["domain_name"])
extra_params = {
"action_mode": "forward_dynamics",
"domain_name": record["domain_name"], # av / droid_lerobot / umi
"action_chunk_size": record["action_chunk_size"],
"image_size": record["image_size"],
"view_point": record["view_point"],
"action": action,
"guardrails": False,
}
return json.dumps(extra_params)3.3 动作条件生成：预测未来世界状态
动作条件生成是Cosmos 3最独特的功能之一。给定当前场景的图像和一个动作序列，模型可以生成执行这些动作后的视频。这种能力可以用于：
动作评估：在真实执行前，预测动作的结果，识别潜在的失败模式。
数据增强：为机器人学习生成大量的"假设场景"，扩充训练数据。
用户交互：向人类操作员展示机器人的意图，获取反馈。
下面是一个使用 vLLM-Omni 做 forward dynamics 的实际请求形态。它对应 cosmos/cookbooks/cosmos3/generator/action/run_fd_with_vllm.ipynb，不是自造的 Diffusers action API：
import json
import mimetypes
from pathlib import Path
import requests
from PIL import Image
VLLM_BASE_URL = "<http://localhost:8001>"
def submit_forward_dynamics(record: dict) -> dict:
"""向 vLLM-Omni 提交 action-conditioned future rollout。"""
vision_path = Path(record["vision_path"])
input_width, input_height = Image.open(vision_path).size
mime_type = mimetypes.guess_type(vision_path.name)[0] or "application/octet-stream"
form = {
"prompt": record.get("prompt", " "),
"num_frames": record["action_chunk_size"] + 1,
"fps": record["fps"],
"size": f"{input_width}x{input_height}",
"num_inference_steps": 30,
"guidance_scale": 1.0,
"flow_shift": 10.0,
"seed": record["seed"],
"extra_params": build_forward_dynamics_extra_params(record),
}
with vision_path.open("rb") as image_file:
response = requests.post(
f"{VLLM_BASE_URL}/v1/videos",
data={key: str(value) for key, value in form.items()},
files={"input_reference": (vision_path.name, image_file, mime_type)},
timeout=120,
)
response.raise_for_status()
return response.json()4. 工程实践：从模型到系统
4.1 仓库结构与核心接口
Cosmos 3的代码不能只看一个仓库。NVIDIA/cosmos 更像面向用户的入口，主要承载 README、cookbook、Diffusers/vLLM-Omni/NIM 接入示例和模型能力说明，适合先跑 demo、理解输入输出格式和部署表面；NVIDIA/cosmos-framework 则是训练和原生 PyTorch 推理的底层框架，包含 cosmos_framework.scripts.inference、cosmos_framework.scripts.train、SFT TOML recipe、数据转换脚本、checkpoint 转换和 RoboLab policy server。如果写博客里的“最小可运行命令”，应优先从这两个仓库各取其实际公开内容：cosmos 负责 cookbook 风格的服务调用，cosmos-framework 负责框架内的推理、数据集、后训练和导出，不要把 NIM、Diffusers、Framework 的参数混成同一套 API。
4.2 与 cosmos-framework 的对应关系
对照 cosmos-framework文档后，本文可以直接参考的内容主要有五类：第一，原生推理入口是python -m cosmos_framework.scripts.inference，单 GPU 走--parallelism-preset=latency，多 GPU 批处理走torchrun --nproc-per-node=N -m cosmos_framework.scripts.inference；第二，推理样本由 JSON/JSONL/YAML 提供，model_mode支持text2image、text2video、image2video、video2video、forward_dynamics、inverse_dynamics和policy，不同模式要求的字段不同；第三，SFT 数据 JSONL 在 framework 文档中有固定结构，例如uuid、duration、width、height、vision_path和t2w_windows，不能直接把项目侧审计字段当成训练 schema；第四，训练入口是cosmos_framework.scripts.train --sft-toml=<recipe>，官方 recipe 放在examples/toml/sft_config/，启动脚本放在examples/launch_sft_*.sh；第五，真实机器人策略服务可以参考cosmos_framework.scripts.action_policy_server_robolab，它服务的是 RoboLab/openpi 风格 WebSocket policy，而不是通用视频生成 HTTP 接口。
COSMOS3_FRAMEWORK_SURFACES = {
"inference": {
"entrypoint": "python -m cosmos_framework.scripts.inference",
"sample_inputs": "inputs/omni/*.json",
"outputs": ["sample_args.json", "sample_outputs.json", "vision.jpg", "vision.mp4"],
"modes": [
"text2image",
"text2video",
"image2video",
"video2video",
"forward_dynamics",
"inverse_dynamics",
"policy",
],
},
"sft_training": {
"entrypoint": "python -m cosmos_framework.scripts.train",
"recipe_dir": "examples/toml/sft_config",
"recommended_launcher": "bash examples/launch_sft_vision_nano.sh",
},
"dataset_tools": {
"caption_video": "python -m cosmos_framework.scripts.caption_from_video",
"build_jsonl": "python -m cosmos_framework.scripts.captions_to_sft_jsonl",
"required_video_jsonl_keys": [
"uuid",
"duration",
"width",
"height",
"vision_path",
"t2w_windows",
],
},
"checkpoint_tools": {
"to_dcp": "python -m cosmos_framework.scripts.convert_model_to_dcp",
"to_vlm_safetensors": "python -m cosmos_framework.scripts.convert_model_to_vlm_safetensors",
"export": "python -m cosmos_framework.scripts.export_model",
},
"policy_server": {
"entrypoint": "python -m cosmos_framework.scripts.action_policy_server_robolab",
"default_checkpoint": "nvidia/Cosmos3-Nano-Policy-DROID",
"default_domain": "droid_lerobot",
},
}4.3 最小推理示例：Framework 原生模式与 Reasoner 模式
如果目标是验证 cosmos-framework 的原生推理链路，最小示例应从 docs/inference.md 中的命令开始，而不是先写 NIM 容器。单 GPU 延迟优先可以直接跑一个输入文件，多 GPU 吞吐优先则用 torchrun 读取一批 inputs/omni/*.json；推理完成后，每个样本目录下会保存 sample_args.json、sample_outputs.json 以及对应的 vision.jpg 或 vision.mp4。这类命令最适合验证 checkpoint、parallelism、guardrails 和 action mode 是否能在本地环境跑通。
# Single GPU: quick latency-oriented inference.
python -m cosmos_framework.scripts.inference
--parallelism-preset=latency
-i "inputs/omni/t2v.json"
-o outputs/omni_nano
--checkpoint-path Cosmos3-Nano
--seed=0
# Multi GPU: throughput-oriented batch inference.
torchrun --nproc-per-node=8 -m cosmos_framework.scripts.inference
--parallelism-preset=throughput
-i "inputs/omni/*.json"
-o outputs/omni_nano
--checkpoint-path Cosmos3-Nano
--seed=0Framework 的输入文件本身也值得参考，因为它把 action 任务的字段写得很清楚。以 forward_dynamics 为例，样本至少需要 model_mode、domain_name、vision_path 和 action_path，其中 action_path 指向动作序列 JSON；policy 和 inverse_dynamics 则会把预测动作写入 sample_outputs.json。这意味着博客中的动作代码不应虚构一个独立的 Diffusers action API，而应围绕 framework 的样本字段或 cosmos action cookbook 的 HTTP 字段来写。
{
"model_mode": "forward_dynamics",
"name": "action_forward_dynamics_robot",
"prompt": "Put the pot to the left of the purple item.",
"domain_name": "bridge_orig_lerobot",
"vision_path": "<https://github.com/nvidia-cosmos/cosmos-dependencies/raw/2b17a2413bd86b2cf9b03823637108851e4ddf2d/inputs/action/bridge_20260501_0.mp4>",
"action_path": "<https://github.com/nvidia-cosmos/cosmos-dependencies/raw/2b17a2413bd86b2cf9b03823637108851e4ddf2d/inputs/action/bridge_20260501_0.json>",
"action_chunk_size": 16,
"fps": 5,
"image_size": 480,
"view_point": "ego_view",
"seed": 0
}Reasoner 模式用于场景理解和物理推理。如果走生产服务，NIM（NVIDIA Inference Microservices）是另一条入口；下面保留 NIM 容器部署示例，但它对应的是 OpenAI-compatible Reasoner API，不是 cosmos-framework 的 SFT 或原生 Generator 推理入口：
# 第一步：启动Reasoner容器
export NGC_API_KEY="your_ngc_api_key"
export CONTAINER_NAME="cosmos3-reasoner"
export IMG_NAME="nvcr.io/nim/nvidia/cosmos3-reasoner:1.7.0"
export LOCAL_CACHE=~/.cache/nim
mkdir -p "$LOCAL_CACHE"
docker run -d --name=$CONTAINER_NAME
--runtime=nvidia
--gpus all
--shm-size=32GB
-e NGC_API_KEY=$NGC_API_KEY
-e NIM_MODEL_SIZE=nano
-v "$LOCAL_CACHE:/opt/nim/.cache"
-p 8000:8000
$IMG_NAME
# 等待服务启动（约2-3分钟）
while ! curl -s <http://localhost:8000/health> > /dev/null; do
echo "Waiting for service to start..."
sleep 5
done
echo "Reasoner service is ready"服务启动后，可以通过OpenAI兼容的接口调用：
import base64
import openai
# 初始化客户端
client = openai.OpenAI(
api_key="not-used", # NIM不需要API key
base_url="<http://127.0.0.1:8000/v1>"
)
def analyze_warehouse_scene(image_path, question):
"""
分析仓库场景并回答物理相关问题
"""
with open(image_path, 'rb') as f:
image_data = f.read()
# 将图像转换为base64
image_base64 = base64.b64encode(image_data).decode('utf-8')
messages = [
{
"role": "system",
"content": (
"你是一个物理AI推理助手。分析场景时，重点关注：n"
"1. 物体的空间位置和相互关系n"
"2. 运动趋势和潜在碰撞n"
"3. 物理约束（重力、摩擦、稳定性）n"
"4. 安全风险和异常情况"
)
},
{
"role": "user",
"content": [
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
},
{
"type": "text",
"text": question
}
]
}
]
response = client.chat.completions.create(
model="nvidia/cosmos3-nano-reasoner",
messages=messages,
max_tokens=512,
temperature=0.2 # 低温度以获得更确定的回答
)
return response.choices[0].message.content
# 实际应用示例
analysis_questions = [
"机器人当前是否可以安全前进？分析画面中的障碍物。",
"如果机器人继续当前速度移动，3秒后会发生什么？",
"画面中的箱子是否稳定？是否有掉落风险？",
"工人进入了哪个危险区域？机器人应该采取什么措施？"
]
for question in analysis_questions:
result = analyze_warehouse_scene(
"warehouse_frame_001.jpg",
question
)
print(f"nQ: {question}")
print(f"A: {result}")
print("-" * 80)这个示例展示了Reasoner的实际应用模式。与通用视觉语言模型不同，Cosmos 3 Reasoner的system prompt强调物理因果关系和安全评估，使其更适合工业场景。
4.4 cosmos-framework 推理使用代码：从输入文件到输出检查
如果读者想真正理解 cosmos-framework 怎么做推理，最应该展示的是一个完整的“输入 JSON -> CLI 推理 -> 输出检查 -> Python pipeline 调用”流程。这里不需要写模型内部 forward，因为 framework 已经把样本读取、默认参数、checkpoint 加载、guardrails 和输出保存封装在 cosmos_framework.scripts.inference 与 OmniInference 里；文章只要告诉读者怎样组织输入、怎样调用入口、怎样确认结果即可。
第一类输入是普通视觉生成样本。下面的 image2video JSON 摘自 framework 输入格式，核心字段是 model_mode、prompt 和 vision_path；prompt 可以是结构化 JSON 字符串，vision_path 可以是本地路径或 URL。推理时，framework 会把完整 sample arguments 保存到输出目录，方便复盘到底用了哪些默认值。
{
"model_mode": "image2video",
"name": "i2v_robot_demo",
"prompt": "{"temporal_caption":"A robotic arm moves a red sphere from the top shelf to the lower shelf while the left arm remains still.","camera":"static high-angle lab view","fps":24,"duration":"0:08"}",
"vision_path": "<https://github.com/nvidia-cosmos/cosmos-dependencies/raw/2b17a2413bd86b2cf9b03823637108851e4ddf2d/inputs/vision/robot_153.jpg>",
"seed": 0
}第二类输入是动作策略样本。policy 模式不只是生成视频，它还会把预测动作写到 sample_outputs.json，同时可输出 future rollout；这正是物理AI文章里最应该展示的代码，因为它能说明 cosmos-framework 如何把视觉观测、任务 prompt 和动作域绑定起来。
{
"model_mode": "policy",
"name": "action_policy_robot",
"prompt": "Put the pot to the left of the purple item.",
"domain_name": "bridge_orig_lerobot",
"vision_path": "<https://github.com/nvidia-cosmos/cosmos-dependencies/raw/2b17a2413bd86b2cf9b03823637108851e4ddf2d/inputs/action/bridge_20260501_0.mp4>",
"action_path": "<https://github.com/nvidia-cosmos/cosmos-dependencies/raw/2b17a2413bd86b2cf9b03823637108851e4ddf2d/inputs/action/bridge_20260501_0.json>",
"action_chunk_size": 16,
"fps": 5,
"image_size": 480,
"view_point": "ego_view",
"seed": 0
}实际项目中可以把样本文件放到 inputs/my_cases/，再用同一个 CLI 批量跑。下面脚本展示了最小的可复现推理流程：写入一个样本、执行推理、检查输出文件是否存在。它不是伪代码，而是围绕 framework 文档里的 -i、-o、--checkpoint-path 和输出目录约定组织出来的使用方式。
mkdir -p inputs/my_cases outputs/checks
cat > inputs/my_cases/policy_robot.json <<'JSON'
{
"model_mode": "policy",
"name": "policy_robot_check",
"prompt": "Put the pot to the left of the purple item.",
"domain_name": "bridge_orig_lerobot",
"vision_path": "<https://github.com/nvidia-cosmos/cosmos-dependencies/raw/2b17a2413bd86b2cf9b03823637108851e4ddf2d/inputs/action/bridge_20260501_0.mp4>",
"action_path": "<https://github.com/nvidia-cosmos/cosmos-dependencies/raw/2b17a2413bd86b2cf9b03823637108851e4ddf2d/inputs/action/bridge_20260501_0.json>",
"action_chunk_size": 16,
"fps": 5,
"image_size": 480,
"view_point": "ego_view",
"seed": 0
}
JSON
python -m cosmos_framework.scripts.inference
--parallelism-preset=latency
-i inputs/my_cases/policy_robot.json
-o outputs/checks
--checkpoint-path Cosmos3-Nano
--seed=0
test -f outputs/checks/policy_robot_check/sample_args.json
test -f outputs/checks/policy_robot_check/sample_outputs.json
test -f outputs/checks/policy_robot_check/vision.mp4如果希望在 Python 里调用 framework，而不是每次都走 CLI，可以参考 examples/inference_pipeline.py 的写法。下面这段保留了官方示例的核心 API：用 OmniSetupOverrides 构造 setup，用 OmniInference.create() 加载模型，用 OmniSampleOverrides 构造样本，再交给 generate_batch()。这类代码适合放在博客里，因为它展示的是 framework 公开的推理管线，而不是随手猜测模型内部结构。
from pathlib import Path
from cosmos_framework.inference.common.init import init_output_dir, init_script
from cosmos_framework.inference.args import (
DEFAULT_CHECKPOINT_NAME,
OmniSampleOverrides,
OmniSetupOverrides,
)
from cosmos_framework.inference.inference import OmniInference, get_sample_data
def run_minimal_pipeline() -> None:
init_script()
output_dir = Path("outputs/blog_pipeline").absolute()
init_output_dir(output_dir)
setup_args = OmniSetupOverrides(
checkpoint_path=DEFAULT_CHECKPOINT_NAME,
output_dir=output_dir,
).build_setup()
pipe = OmniInference.create(setup_args)
sample_args = OmniSampleOverrides(
name="blog_t2i_check",
output_dir=output_dir,
prompt="A modern robotics research laboratory with a clean workbench and one robotic arm.",
num_frames=1,
).build_sample(model_config=pipe.model_config)
data_batch = get_sample_data(sample_args, model=pipe.model)
pipe.generate_batch([sample_args], data_batch)
if __name__ == "__main__":
run_minimal_pipeline()4.5 Generator模式：视频生成与未来预测
Generator模式是Cosmos 3最引人注目的功能。与传统文生视频模型不同，Cosmos 3的生成能力是为物理AI场景设计的，重点不在于视觉效果，而在于物理一致性和动作可控性。下面使用 cosmos/cookbooks/cosmos3/generator/audiovisual/README.md 中的 Diffusers 调用形态，直接读取 cookbook 的结构化 prompt 和 negative prompt：
import json
import torch
from diffusers import Cosmos3OmniPipeline
from diffusers.schedulers.scheduling_unipc_multistep import UniPCMultistepScheduler
from diffusers.utils import export_to_video
prompt = json.load(open("assets/prompts/text2video/robot_kitchen.json"))
negative = json.load(open("assets/negative_prompts/text2video/neg_prompt.json"))
pipe = Cosmos3OmniPipeline.from_pretrained(
"nvidia/Cosmos3-Nano",
torch_dtype=torch.bfloat16,
device_map="cuda",
)
pipe.scheduler = UniPCMultistepScheduler.from_config(
pipe.scheduler.config,
flow_shift=10.0,
)
result = pipe(
prompt=json.dumps(prompt),
negative_prompt=json.dumps(negative),
image=None,
num_frames=189,
height=720,
width=1280,
fps=24,
num_inference_steps=35,
guidance_scale=6.0,
enable_sound=False,
add_resolution_template=False,
add_duration_template=False,
generator=torch.Generator(device="cuda").manual_seed(1234),
)
export_to_video(result.video, "/tmp/cosmos3_t2v_diffusers.mp4", fps=24)这个示例展示了四个工程细节：第一，Cosmos 3 的 audiovisual cookbook 倾向于使用结构化 JSON prompt，而不是只有一句自然语言描述，开发者可以把主体、动作、场景、镜头和音频描述拆成可检查字段；第二，negative_prompt用于压制非物理行为、异常肢体、漂浮物体和错误碰撞，不能把它当成随意的审美补丁；第三，固定torch.Generator(...).manual_seed(1234)对复现生成结果、比较不同 prompt 或不同 checkpoint 非常重要；第四，num_frames=189、fps=24 、num_inference_steps=35、guidance_scale=6.0、flow_shift=10.0这些值来自 cookbook 风格的生成示例，适合说明默认推理形态，但在显存、时长和分辨率变化后仍需重新压测。
5. 训练与微调：从通用模型到领域专家
5.1 后训练的必要性判断
并非所有应用都需要对 Cosmos 3 做后训练。官方模型已经覆盖了大量通用视觉和物理场景，很多早期验证只需要跑通 Reasoner、Generator 或 action inference，就足以判断任务是否值得继续投入；真正需要 SFT 的情况通常集中在四类：第一，视觉域差异很强，例如医疗影像、工业相机、夜间红外或特定机器人视角，公开数据上的纹理、视角和运动模式与真实业务相差太大；第二，动作空间具有专有定义，例如机器人使用非标准关节配置、特殊末端执行器或企业内部控制协议，需要模型学习这些动作维度和后处理方式；第三，物理约束特别严格，例如精密装配、危险工业流程、自动驾驶长尾场景，模型不能只生成“看起来合理”的视频，而要在特定失败模式上更稳；第四，数据隐私或部署环境不允许调用云端服务，只能在本地使用 cosmos-framework 完成数据处理、训练、导出和部署验证。
5.2 数据准备：JSONL格式与标注规范
Cosmos Framework 的完整后训练流程应以 cosmos-framework/docs/training.md 和 docs/dataset_jsonl.md 为准。这里要区分两层数据：第一层是 framework 真正读取的 SFT JSONL，视频 SFT 示例使用 uuid、duration、width、height、vision_path 和 t2w_windows，其中 t2w_windows 记录训练窗口的起止帧、采样间隔和 caption；第二层是项目侧审计清单，用来追踪样本来源、生成 seed、审核结论和 action embodiment 信息。博客里可以同时写这两层，但必须明确前者是训练 schema，后者是工程治理字段，不能把审计字段直接当成 cosmos-framework 的训练输入。
{
"uuid": "episode_000015_clip000",
"duration": 17.4,
"width": 256,
"height": 256,
"vision_path": "videos/episode_000015_clip000.mp4",
"t2w_windows": [
{
"start_frame": 0,
"end_frame": 86,
"temporal_interval": 1,
"caption": "A robotic arm moves with precision over a tabletop workspace, manipulating small objects in a continuous sequence while the camera remains static."
}
]
}如果没有现成 caption，framework 也提供了从视频生成 caption 并转换为 SFT JSONL 的脚本。这个流程先把视频交给本地 vLLM Reasoner server 做 dense narrative caption，再用 captions_to_sft_jsonl 把 captions/ 和 videos/ 对齐成 video_dataset_file.jsonl。这一步的价值在于让数据处理链路可复现：同一个视频、同一个 caption 模板、同一个输出 JSONL 文件可以被记录下来，而不是在训练前临时手写描述。
# Start a local VLM server that can read local video files.
uvx --with nvidia-cuda-runtime-cu12
vllm@0.19.0 serve Qwen/Qwen3-VL-8B-Instruct-FP8
--tensor-parallel-size 1
--allowed-local-media-path /
# Caption one video or a directory of videos.
python -m cosmos_framework.scripts.caption_from_video
--video /path/to/videos/
-o outputs/captions
--server <http://localhost:8000/v1>
# Convert captions + videos into the SFT JSONL expected by the dataset flow.
python -m cosmos_framework.scripts.captions_to_sft_jsonl
--captions-dir outputs/sft_dataset/train/captions
--videos-dir outputs/sft_dataset/train/videos
-o outputs/sft_dataset/train/video_dataset_file.jsonl项目侧仍然建议额外保留一份审计 JSONL，它不一定直接进入训练，但能回答“这个样本从哪里来、用哪个模型生成、是否经过人工审核、对应哪个动作域”这类工程问题。真实视频、生成视频、动作轨迹、prompt、模型版本和人工审核结论必须能一一对应，尤其是合成样本不能未经筛选直接进入训练；下面示例保留的是审计字段，不是 framework 固定 schema：
{"id":"warehouse_001","source":"real","video":"videos/warehouse_forklift_stop.mp4","caption":"A forklift operator sees a worker and stops immediately, maintaining safe distance.","task":"safety_prediction","review_label":"pass"}
{"id":"warehouse_synth_017","source":"cosmos3_generator","video":"generated/worker_blind_spot_seed42.mp4","prompt_file":"prompts/worker_blind_spot.json","negative_prompt_file":"prompts/neg_prompt.json","model":"nvidia/Cosmos3-Nano","seed":42,"task":"long_tail_expansion","review_label":"needs_human_check"}
{"id":"fd_droid_003","source":"cosmos3_forward_dynamics","image":"images/droid_start.png","action":"actions/droid_chunk_003.json","domain_name":"droid_lerobot","raw_action_dim":10,"action_chunk_size":16,"task":"forward_dynamics"}建议保留的字段：
• id: 唯一标识符•
source: 数据来源，如 real、cosmos3_generator、cosmos3_forward_dynamics
video / image: 媒体路径•
caption: 文本描述，应包含物理因果关系•
prompt_file / negative_prompt_file: 生成样本时使用的提示词文件•
model、seed: 生成可复现性信息•
domain_name、raw_action_dim、action_chunk_size: action 任务必须记录的 embodiment 信息•
review_label: 人工审核状态，避免合成样本未经筛选直接进入训练
5.3 训练流程：从数据到模型
Cosmos Framework 的训练入口位于 cosmos_framework.scripts.train，但具体参数、配置文件和数据适配器应以 cosmos-framework/docs/training.md 为准。本文不再编造 configs/cosmos3_nano_finetune.yaml 这类不存在于本地仓库的配置，而是使用 framework 文档里的三段式流程：先准备数据和 VAE/checkpoint 依赖，再把基础 checkpoint 转成 PyTorch Distributed Checkpoint（DCP），最后通过 paired launch shell 或 raw torchrun --sft-toml 启动 SFT。这个流程默认是 8×H100 级别的训练环境，普通单卡机器更适合做推理验证或小规模数据格式检查。
# From the cosmos-framework repo root.
# 1. Install the training dependency group.
uv sync --all-extras --group=cu130-train
# If the node uses CUDA 12.8:
# uv sync --all-extras --group=cu128-train
# 2. Convert the base checkpoint to DCP.
BASE_CHECKPOINT_NAME=Cosmos3-Nano
python -m cosmos_framework.scripts.convert_model_to_dcp
-o examples/checkpoints/$BASE_CHECKPOINT_NAME
--checkpoint-path $BASE_CHECKPOINT_NAME
# 3A. Recommended: run the paired launch shell after data + checkpoint setup.
bash examples/launch_sft_vision_nano.sh
# 3B. Raw torchrun: explicitly export the paths consumed by the recipe.
TOML_FILE="examples/toml/sft_config/vision_sft_nano.toml"
export DATASET_PATH="$PWD/examples/data/bridge-v2-subset-synthetic-captions/sft_dataset_bridge"
export BASE_CHECKPOINT_PATH="$PWD/examples/checkpoints/Cosmos3-Nano"
export WAN_VAE_PATH="$PWD/examples/checkpoints/wan22_vae/Wan2.2_VAE.pth"
IMAGINAIRE_OUTPUT_ROOT=outputs/train PYTHONPATH=.
torchrun --nproc_per_node=8 -m cosmos_framework.scripts.train
--sft-toml=$TOML_FILE训练输出会落在 $IMAGINAIRE_OUTPUT_ROOT/<project>/<group>/<name>/ 下，包含解析后的 config.yaml、launch_info.yaml、job_env.yaml、DCP checkpoint、回调输出和可选 W&B 文件。完成训练后，如果要把 DCP checkpoint 变成 Hugging Face safetensors 形式，再使用 cosmos_framework.scripts.export_model；导出的目录可以在后续 inference 命令里通过 --checkpoint-path $RUN_DIR/model 使用。
RUN_DIR=outputs/train/cosmos3/sft/vision_sft_nano
CHECKPOINT_ITER=$(cat $RUN_DIR/checkpoints/latest_checkpoint.txt)
CHECKPOINT_PATH=$RUN_DIR/checkpoints/$CHECKPOINT_ITER
python -m cosmos_framework.scripts.export_model
--checkpoint-path $CHECKPOINT_PATH
--config-file $RUN_DIR/config.yaml
-o $RUN_DIR/model关键参数不要脱离 recipe 单独解释。vision_sft_nano.toml 中可以看到 [model].precision = "bfloat16"、[model.parallelism] 使用 FSDP 相关拓扑、[model.activation_checkpointing] 控制激活重算、[optimizer] 选择训练参数子集、[trainer] 设置 max_iter 和 grad_accum_iter、[checkpoint].load_path 通过 ${oc.env:BASE_CHECKPOINT_PATH} 读取 DCP 路径。更实用的做法是复制最接近任务的官方 TOML recipe，只调整数据路径、checkpoint 路径、输出目录和少量训练步数；不要在博客中随手写固定 batch size、hidden size、层数或噪声调度。
5.4 最小训练脚本：从 recipe 到验证推理
训练代码也应该写，但写法要和 framework 的训练设计一致：不要把数据加载、模型初始化、分布式并行和 checkpoint 保存全部手写一遍，而是围绕官方 recipe、launch shell 和 cosmos_framework.scripts.train 组织。下面这个脚本适合放在博客里作为“最小 SFT 使用模板”：它检查数据 JSONL、转换基础 checkpoint、导出必要环境变量、启动 paired launch shell，并在训练后导出 safetensors。真实项目只需要替换 DATASET_PATH、BASE_CHECKPOINT_NAME、WAN_VAE_PATH 和 OUTPUT_ROOT，不要改动模型内部源码。
#!/usr/bin/env bash
set -euo pipefail
# Run from the cosmos-framework repo root.
# Hardware note: the official SFT path is documented for 8x H100 80GB.
BASE_CHECKPOINT_NAME="${BASE_CHECKPOINT_NAME:-Cosmos3-Nano}"
DATASET_PATH="${DATASET_PATH:-$PWD/examples/data/bridge-v2-subset-synthetic-captions/sft_dataset_bridge}"
WAN_VAE_PATH="${WAN_VAE_PATH:-$PWD/examples/checkpoints/wan22_vae/Wan2.2_VAE.pth}"
BASE_CHECKPOINT_PATH="${BASE_CHECKPOINT_PATH:-$PWD/examples/checkpoints/$BASE_CHECKPOINT_NAME}"
OUTPUT_ROOT="${OUTPUT_ROOT:-outputs/train}"
test -f "$DATASET_PATH/train/video_dataset_file.jsonl"
test -f "$WAN_VAE_PATH"
python -m cosmos_framework.scripts.convert_model_to_dcp
-o "$BASE_CHECKPOINT_PATH"
--checkpoint-path "$BASE_CHECKPOINT_NAME"
export DATASET_PATH
export WAN_VAE_PATH
export BASE_CHECKPOINT_PATH
export OUTPUT_ROOT
bash examples/launch_sft_vision_nano.sh
RUN_DIR="$OUTPUT_ROOT/cosmos3/sft/vision_sft_nano"
CHECKPOINT_ITER="$(cat "$RUN_DIR/checkpoints/latest_checkpoint.txt")"
CHECKPOINT_PATH="$RUN_DIR/checkpoints/$CHECKPOINT_ITER"
python -m cosmos_framework.scripts.export_model
--checkpoint-path "$CHECKPOINT_PATH"
--config-file "$RUN_DIR/config.yaml"
-o "$RUN_DIR/model"训练不是跑完就结束，还要用导出的模型做一次推理回归。下面命令把 --checkpoint-path 指向导出的 $RUN_DIR/model，用同一组 inputs/omni/*.json 或自己的验证样本生成结果；如果输出目录里没有 sample_outputs.json、vision.mp4 或质量明显退化，就说明训练、导出或数据配置中至少有一环需要回查。
RUN_DIR=outputs/train/cosmos3/sft/vision_sft_nano
torchrun --nproc-per-node=8 -m cosmos_framework.scripts.inference
--parallelism-preset=throughput
-i "inputs/omni/*.json"
-o outputs/after_sft_eval
--checkpoint-path "$RUN_DIR/model"
--seed=0
find outputs/after_sft_eval -name sample_outputs.json -o -name vision.mp4如果不想使用 paired launch shell，也可以直接调用 raw torchrun。这段代码适合解释“训练入口到底在哪里”：真正运行的是 cosmos_framework.scripts.train --sft-toml=<recipe>，DATASET_PATH、BASE_CHECKPOINT_PATH、WAN_VAE_PATH 通过环境变量传入，TOML 只负责声明 recipe 可调字段。
TOML_FILE="examples/toml/sft_config/vision_sft_nano.toml"
export DATASET_PATH="$PWD/examples/data/bridge-v2-subset-synthetic-captions/sft_dataset_bridge"
export BASE_CHECKPOINT_PATH="$PWD/examples/checkpoints/Cosmos3-Nano"
export WAN_VAE_PATH="$PWD/examples/checkpoints/wan22_vae/Wan2.2_VAE.pth"
IMAGINAIRE_OUTPUT_ROOT=outputs/train PYTHONPATH=.
torchrun --nproc_per_node=8 -m cosmos_framework.scripts.train
--sft-toml="$TOML_FILE"5.5 配置文件原则
训练配置不要在没有框架文档确认的情况下手写模型层数、hidden size 或噪声调度。cosmos-framework/docs/sft_config.md 明确说明，SFT recipe 位于 examples/toml/sft_config/<recipe>.toml，加载时会通过 SFTExperimentConfig 做严格 pydantic 校验，未知字段会在训练前触发 ValidationError；常见顶层 section 包括 [job]、[model]、[optimizer]、[scheduler]、[trainer]、[checkpoint] 和 [dataloader_train]。因此，最可靠的写法是从官方 recipe 复制一个最接近任务的配置，只调整数据路径、checkpoint 路径、输出目录、训练步数和少量 recipe 暴露的调参项；动作任务还要确认 model_mode、domain_name、action_chunk_size、image_size、view_point 和动作 JSON 的 raw action dimension 与数据一致。
5.6 RoboLab Policy Server：动作策略的服务入口
如果目标不是生成一段视频，而是把 Cosmos 3 policy 接到 RoboLab/openpi 风格客户端，cosmos-framework 提供了action_policy_server_robolab。这个服务接收 msgpack 编码的 observation dictionary，返回包含action的字典；如果开启--decode-video，也可以返回 rollout video。默认开源流程使用nvidia/Cosmos3-Nano-Policy-DROID，domain 是droid_lerobot，分辨率480，conditioning FPS 为15，输入图像540x640，action chunk size 为32，joint_pos动作维度为8，guidance3.0，denoising steps4，shift5.0。这组参数来自 framework 文档，适合写成默认服务形态；如果换成自己的 checkpoint，就应同时对齐 serving config 中的 resolution、action chunk、action dim 和 seed 策略。
# Install the training + policy-server dependency groups once.
uv sync --all-extras --group=cu130-train --group=policy-server
source .venv/bin/activate
# Serve the released DROID policy checkpoint.
python -m cosmos_framework.scripts.action_policy_server_robolab --port 8000
# Or serve an exported local safetensors checkpoint.
python -m cosmos_framework.scripts.action_policy_server_robolab
--checkpoint-path /path/to/consolidated/model
--port 8000
# Health check.
curl <http://localhost:8000/healthz>6. 结论
NVIDIA Cosmos 3的价值，不在于它把“看图说话”换成了更大的多模态模型，而在于它把视觉推理、世界生成和动作建模放进了同一个物理AI问题里。技术上，它将动作提升为一级模态，使模型可以显式建模“动作 -> 结果”的因果关系；通过 Mixture-of-Transformers 让 Reasoner 和 Generator 共享多模态表示，减少多模型编排带来的语义断裂；同时又通过 cosmos-framework 提供原生推理、SFT、数据转换、checkpoint 转换、导出和 policy server 等工程入口，让开发者不必只停留在 demo 层面。这里最有工程价值的部分，是模型能力、数据闭环和部署表面被放在了一套可追踪流程里。
参考资料
- 1. NVIDIA 新闻稿：NVIDIA Launches Cosmos 3, the Open Frontier Foundation Model for Physical AI (https://www.globenewswire.com/news-release/2026/06/01/3303987/0/en/NVIDIA-Launches-Cosmos-3-the-Open-Frontier-Foundation-Model-for-Physical-AI.html)2. NVIDIA 技术博客：借助 NVIDIA Cosmos 3 开发物理 AI 推理模型、世界模型和动作模型 (https://developer.nvidia.cn/blog/develop-physical-ai-reasoning-world-and-action-models-with-nvidia-cosmos-3/)3. NVIDIA Cosmos 官方页面：Physical AI with World Foundation Models (https://www.nvidia.com/en-us/ai/cosmos/)4. Hugging Face 文章：Welcome NVIDIA Cosmos 3: The First Open Omni-model for Physical AI Reasoning and Action (https://huggingface.co/blog/nvidia/cosmos-3-for-physical-ai)5. arXiv 技术报告：Cosmos 3: Omnimodal World Models for Physical AI (https://arxiv.org/abs/2606.02800)6. GitHub：NVIDIA/cosmos (https://github.com/NVIDIA/cosmos)7. GitHub：NVIDIA/cosmos-framework (https://github.com/NVIDIA/cosmos-framework)8. Hugging Face 模型卡：nvidia/Cosmos3-Nano (https://huggingface.co/nvidia/Cosmos3-Nano)9. NVIDIA NIM 文档：Query the Cosmos 3 Reasoner API (https://docs.nvidia.com/nim/vision-language-models/1.7.0/examples/cosmos-reason3/api.html)10. vLLM-Omni 文档：Cosmos3OmniDiffusersPipeline (https://docs.vllm.ai/projects/vllm-omni/en/latest/api/vllm_omni/diffusion/models/cosmos3/)
