Cosmos 3

NVIDIA Cosmos 3 هو نموذج أساسي موحّد للعالم (WFM) للذكاء الاصطناعي الفيزيائي — وهو نموذج شامل واحد يجمع بين توليد العالم، والاستدلال الفيزيائي، وتوليد الإجراءات. إنه يحل محل نماذج التنبؤ، والاستدلال، والنقل المنفصلة من إصدارات Cosmos السابقة: سواء كنت تبني حلولاً للروبوتات، أو المركبات ذاتية القيادة، أو المساحات الذكية، فإن Cosmos 3 يمنحك أساسًا واحدًا لمحاكاة العالم الفيزيائي وفهمه.

ما يتم إصداره مع هذا الإصدار:

• النماذج على Hugging Face Hub مع بطاقات النماذج وترخيص
• تكامل Cosmos 3 Diffusers في خطوط أنابيب التوليد (هذه الصفحة)
• سكربتات ما بعد التدريب لضبط Cosmos 3 على بياناتك الخاصة
• مجموعات بيانات مفتوحة لتوليد البيانات الاصطناعية (SDG) للذكاء الاصطناعي الفيزيائي

ما الجديد في Cosmos 3

أبرز تغيير عن إصدارات Cosmos السابقة هو أن Cosmos 3 هو نموذج شامل، مبني على بنية مزيج التحويلات (Mixture-of-Transformers، MoT). سابقًا، كان المطورون يعملون مع نماذج منفصلة لتوليد العالم (Predict)، والتوليد المُتحكَّم به (Transfer)، وفهم المشهد (Reason)، وتوليد سياسات الإجراء. يوحّد Cosmos 3 جميع هذه الوظائف في نموذج واحد يستنتج ويولّد عبر الأنماط المختلفة في تمريرة أمامية واحدة.

من نموذج واحد يمكنك:

• إنشاء عوالم فيديو فيزيائية معقولة من نصوص أو صور أو مدخلات إجراء (تحويل الصورة إلى فيديو، تحويل النص إلى فيديو، توليد فيديو مشروط بالإجراء).
• التفكير في الخصائص الفيزيائية مثل الحركة والسببية والعلاقات المكانية.
• تنبؤ بالتسلسلات المستقبلية للفيديو والإجراءات من الحالة الحالية.
• نقل المشاهد عبر نقاط نظر وظروف مختلفة مع تحكم هيكلي (قريباً).

في الخلفية، يعمل نموذج Cosmos3OmniTransformer واحد على نموذج لغوي بنمط Qwen بالتوازي مع مسار توليد بالانتشار: حيث تمر رموز النص عبر تدفق "فهم" سببي، بينما تمر الكمونات (latents) الخاصة بالفيديو والصوت عبر تدفق "توليد" ذي انتباه ثنائي الاتجاه، وتُدمج معًا بواسطة RoPE متعدد الوسائط ثلاثي الأبعاد. راجع ورقة Cosmos World Foundation Model Platform للاطلاع على الخلفية المعمارية.

نقاط التفتيش المتاحة

تم إصدار نقطتي تفتيش على الـ Hub — nvidia/Cosmos3-Nano (أصغر حجمًا وأسرع) و nvidia/Cosmos3-Super (أكبر حجمًا وجودة أعلى). تدعم فئة خط الأنابيب نفسها تحويل النص إلى صورة، وتحويل النص إلى فيديو، وتحويل الصورة إلى فيديو، و(مع نقطة تفتيش تدعم الصوت) تحويل النص والصورة إلى فيديو مع صوت — اختر مستودعًا واستخدم تبويب النموذج الخاص بكل سير عمل أدناه.

تأكد من الاطلاع على دليل المجدولين لمعرفة كيفية استكشاف التوازن بين سرعة المجدول وجودة النتائج، واطّلع على قسم إعادة استخدام المكونات عبر خطوط الأنابيب لمعرفة كيفية تحميل نفس المكونات بكفاءة في خطوط أنابيب متعددة.

إزالة الضوضاء المختلطة FP8 W8A8/W8A16

تتوفر نقاط التفتيش الرسمية لنموذج ModelOpt FP8 على Hub في المراجعة fp8 (على سبيل المثال nvidia/Cosmos3-Nano مع revision="fp8").

جميع هذه نقاط التفتيش مُكمّمة بالطريقة نفسها. يستخدم W8A8 أوزانًا بدقة 8 بت وتنشيطات بدقة 8 بت (GEMM المُستعاد من ModelOpt). يعيد W8A16 استخدام تلك الأوزان ذاتها بدقة 8 بت لكنه يتخطى كمّمة التنشيطات: يتم فك كمّمة الوزن FP8 ويُنفَّذ خطي قياسي على تنشيطات BF16/FP16/FP32.

قد يؤدي تشغيل W8A8 في كل خطوة إلى ظهور وميض واضح في توليد الفيديو متعدد الخطوات. لذلك، تُعلن نقاط التفتيش الخاصة بنماذج الفيديو Nano / Super / Super-I2V FP8 عن جدول في transformer/config.json: W8A16 في أول 3 خطوات وآخر 3 خطوات، W8A8 في المنتصف. تقرأ مكتبة Diffusers هذه الأعداد من نقطة التفتيش بدلاً من تثبيتها في الكود. يتم اختيار الدقة مرة واحدة لكل خطوة من خطوات المجدول (scheduler) بحيث تتطابق استدعاءات cond/uncond الخاصة بالإرشاد الحر من التصنيف (classifier-free guidance).

لا تُظهر عمليات توليد الصور ونقاط التفتيش المُقطّرة بعدد خطوات قليلة ذلك الوميض، لذا فإن مستودع Super-T2I ومستودعات FP8 المُقطّرة ذات الخطوات الأربع لا تُعلن عن جدول زمني وتبقى على وضع W8A8 في كل خطوة. كما أن الجدول الزمني خاص بـ ModelOpt FP8 فقط: فواجهات الكمّية الأخرى (مثل TorchAO) تحافظ على عملياتها الأمامية الأصلية.

حمّل المراجعة fp8 باستخدام مسار الاستعادة نفسه كما في دليل ModelOpt (حيث إن revision="fp8" يحمل بالفعل إعدادات الكمّة). ثم يتبع الدقة المختلطة نقطة التفتيش تلقائيًا:

import torch
from diffusers import Cosmos3OmniPipeline

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

خياران في وقت التوليد:

• افتراضي (mixed_precision_format=None): إذا كان ملف الحفظ (checkpoint) يُعلن عن diffusion_step_policy، فسيتم تشغيل W8A16 في الخطوات الأولى/الأخيرة N، وW8A8 الأصلية في الخطوات الوسطى. هذه هي الوصفة المقصودة لـ فيديو متعدد الخطوات FP8 (يقلل من الوميض مقارنةً باستخدام W8A8 بالكامل). أما نماذج FP8 المُقطّرة ذات الخطوات الأربع وSuper-T2I FP8 فتستثني السياسة، لذا فإن الوضع الافتراضي هو W8A8 بالكامل بالفعل.
• mixed_precision_format="none": يُبقي كل خطوة على W8A8 الأصلي. أسرع، لأن W8A16 هو إلغاء تكميم + torch.nn.functional.linear بدلاً من GEMM FP8 المُستعاد، لكن الفيديو متعدد الخطوات قد يرتجف. استخدم هذا لاختبار جدولة A/B أو لمطابقة خط أساس مُكمَّم بالكامل.

على محطة عمل واحدة من Blackwell، كان Cosmos3-Nano @fp8 عند دقة 720×1280 / 35 خطوة أبطأ بنسبة 27% تقريبًا (T2I) وأبطأ بنسبة 13% (T2V بإطار 49) مع جدول المزج الافتراضي مقارنةً باستخدام "none". هذه الأرقام ليست ضمانًا لمعدل الإنتاجية. مرّر "fp8" فقط لفرض جدول أول/آخر N على نقطة تحقق ModelOpt FP8 التي لا تحتوي على سياسة.

توسيع عينات المحفّز

تم تدريب 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). تكتب الأوامر الموجهات المُوسَّعة إلى ملف --output كمصفوفة JSON (كائن واحد لكل سطر غير فارغ في --input)؛ مرّر مسار .jsonl بدلاً من ذلك للحصول على كائن JSON واحد لكل سطر. فيما يتعلق بـimage2video، يجب عليك أيضاً تقديم صورة الشرط عبر --image-url (رابط URL أو مسار محلي) أو --image-list (صورة واحدة لكل موجه).

يُوفَّر موجه إيجابي مُوسَّع مسبقًا (assets/example_t2v_prompt.json) وموجه سلبي (assets/negative_prompt.json) لتسهيل الاستخدام، وتُستخدم في أمثلة التوليد أدناه. تقوم الأمثلة بتحميل هذه الملفات بصيغة JSON وتمريرها إلى خط الأنابيب كنصوص JSON عبر json.dumps(...).

النص إلى فيديو

توليد متعدد الإطارات مشروط بالنص فقط. اختر num_frames بناءً على المدة المستهدفة — القيمة الافتراضية num_frames=189 تنتج ≈ 7.9 ثانية عند 24 إطارًا في الثانية. يتم قراءة الموجه والموجه السلبي من ملفات JSON المُحسَّنة الموصوفة في تحسين الموجهات.

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)

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.

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)

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.

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)

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") لاختيار أي طرف من مقطع المصدر الأطول تُؤخذ منه الإطارات الشرطية. وكما هو الحال مع الأوضاع الأخرى، يجب أن يتبع الأمر البصري البنية الوصفية بصيغة JSON الموضحة في توسيع الأوامر البصرية.

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)

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 إلى استدعاء تحويل الفيديو إلى فيديو لتوليد مسار صوتي متزامن بشكل مشترك. يُعاد الموجة الصوتية إلى جانب الفيديو ويمكن دمجها في ملف MP4 باستخدام encode_video().

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

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 لتوليد مسار صوتي متزامن بشكل مشترك. يُعاد الموجة الصوتية إلى جانب الفيديو ويمكن دمجها في ملف MP4 باستخدام encode_video().

هذا هو نفس الاستدعاء الموجود في مثال النص إلى الفيديو أعلاه مع إضافة enable_sound=True:

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

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 كل مدخلات محددة الإجراء في 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.

سياسة الإجراء

تتنبأ توليد سياسة الإجراء بعلامات الفيديو والإجراءات المستقبلية انطلاقًا من إطار الملاحظة الأول، والموجّه النصي، وبيانات الوصف الخاصة بمجال الإجراء. يستخدم المثال أدناه مجال روبوت Bridge ويكتب شريحة الإجراء المتنبأ بها إلى ملف JSON في فضاء الإجراء المعيارية للنموذج.

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)

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. يعيد SeaCache استخدام بقايا المحوّل (transformer) عندما يتغير مؤشر Spectral-Evolution-Aware ببطء، مما يقلل عدد مرات تنفيذ المحوّل بالكامل. يحسب المؤشر من كامنات الرؤية الخام، بما في ذلك إطارات التكييف النظيفة لتوليد الفيديو من الصورة. فعّله على المحوّل باستخدام مستدعات البيانات الوصفية للمجدول من خط الأنابيب:

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 تقريبي وقد يُغيّر المخرجات المولّدة. عطّله باستخدام pipe.transformer.disable_cache() عندما تحتاج إلى تنفيذ كل خطوة إزالة الضوضاء عبر المحوّل الكامل. يُعاد تعيين حالة الكاش بعد كل استدعاء للخطّ، وتحتفظ فروع التوجيه الشرطي وغير الشرطي بسجلات مستقلة.

يحافظ Cosmos 3 على بوابة SeaCache خارج طبقات المُفكِّك المتكررة، مما يجعله متوافقًا مع التجميع الإقليمي. قم بتجميع الطبقات بعد تمكين ذاكرة التخزين المؤقت:

pipe.transformer.compile_repeated_blocks(fullgraph=True)

يدعم SeaCache أيضًا أدوات Ulysses الخاصة بالتوازي السياقي وDTensor الخاصة بالتوازي التّنسوري لنموذج Cosmos 3، كما هو موثّق أدناه. يتم مزامنة قرارات التخزين المؤقت عبر الرتب. ولا يُدَّعى دعم تجميع النموذج الكامل أو استراتيجيات تقسيم النموذج الأخرى.

توازي السياق

في حالة مقاطع الفيديو الطويلة أو الدقة العالية، قد يتجاوز المرور الأمامي الواحد ميزانية الذاكرة وزمن الاستجابة الخاصة بوحدة معالجة رسومية واحدة. يدعم Cosmos 3 التوازي السياقي (CP) لتقسيم بُعد التسلسل عبر وحدات معالجة رسومية متعددة، مما يقسم حساب الانتباه بحيث تحتفظ كل وحدة بشريحة من الرموز فقط.

يدعم Cosmos 3 توازياً سياقياً عبر Ulysses (تبادل تسلسلي/رؤوس بين الجميع). ولا يُدعم الانتباه الحلقي (Ring attention).

Unlike most diffusers models, Cosmos 3 does not wire CP into the transformer or the declarative enable_parallelism() path: its grouped-query attention, separate understanding/generation streams (the generation stream attends to both), and ragged per-stream lengths can’t be expressed as a _cp_plan. Instead, the model exposes small no-op shard/gather seams, and the implementation lives in examples/cosmos3/cosmos_parallel.py — a self-contained module you can read end to end and adapt. It offers two orthogonal, composable sharding axes:

استخدم أيًّا منهما بمفرده أو كلاهما معًا على شبكة ثنائية الأبعاد (tp, cp) (انظر ملاءمة النماذج الكبيرة باستخدام التوازي على مستوى الموترات).

هناك متطلبان محددان لـ Cosmos 3:

• استخدم واجهة native الانتباه. يستخدم Cosmos 3 انتباه الاستعلام المجمّع (GQA)، وواجهة SDPA الأصلية هي الوحيدة التي تقبل enable_gqa (يرفضها cuDNN وflash). تقوم الدوال المساعدة بتوسيع رؤوس KV إلى عدد رؤوس الاستعلام وتستدعي SDPA مع enable_gqa=False حتى تستمر في التوجيه إلى نواة flash (فإن كان المسار البديل الرياضي سيُنتج مصفوفة كاملة من الدرجات [S, S] مما يؤدي إلى نفاد الذاكرة في التسلسلات الطويلة).
• يجب أن يكون درجة CP (Ulysses) قابلاً للقسمة على عدد رؤوس الاستعلام (32 لـ Nano، و64 لـ Super)؛ أما بالنسبة لـ TP، فيجب أن تكون الدرجة قابلة للقسمة على رؤوس KV (8). يتم تقسيم تدفقات الفهم (النص) والتوليد (الفيديو/الصوت) بشكل مستقل على طول التسلسل، وتُملأ الأطوال غير المنتظمة بصفر داخلياً حتى تصبح مضاعفاً لحجم العالم.

شغّلها

يستخدم ملف سطر الأوامر الكامل examples/cosmos3/inference_cosmos3.py Cosmos3OmniModularPipeline ويعيد استخدام هذه الأدوات المساعدة، لذا يمكن تشغيل أي نمط (نص إلى صورة/فيديو، صورة إلى فيديو، صوت، أوضاع الإجراء) على عدة وحدات معالجة رسومية عبر --tp-degree / --cp-degree. ابدأ التشغيل باستخدام torchrun؛ يجب أن يساوي --tp-degree * --cp-degree قيمة --nproc_per_node. تنتج كل رتبة نفس المخرجات؛ وتكتب الرتبة 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 غيغابايت لا تتسع على وحدة معالجة رسومية واحدة بسعة 96 غيغابايت، لذا فهي تتطلب TP؛ بينما تتسع Nano على وحدة معالجة رسومية واحدة، لذا فإن CP لها هي تحسين محض للزمن المستغرق. (تجاهل كلا العلمين لتشغيل وحدة معالجة رسومية واحدة.)

تدريب نماذج كبيرة باستخدام توازي الموترات

يقوم تقطيع CP بتقسيم التفعيلات لكنه يكرر كل الأوزان على كل رتبة، لذا لا يقلل من حجم الأوزان للنموذج — فأي نموذج لا يتسع على وحدة معالجة رسومية واحدة لن يتسع أيضاً باستخدام CP وحده. لتقسيم الأوزان، يطبّق enable_cosmos3_tensor_parallel(transformer, tp_mesh) توازياً على مستوى المصفوفات بأسلوب Megatron على محور شبكة ثانٍ متعامد:

• تُقسَّم إسقاطات الانتباه ووحدة التحويل متعددة الطبقات (MLP) عمودياً/صفوفياً عبر مجموعة TP (حيث تكون to_q/to_k/to_v وadd_q/k/v وإسقاطات gate/up في وحدات التحويل متعددة الطبقات متوازية عمودياً؛ بينما تكون to_out/to_add_out وإسقاط down في وحدات التحويل متعددة الطبقات متوازية صفوفياً مع إجراء all-reduce). في النهاية، يمتلك كل رتبة query_heads / tp من رؤوس الاستعلام وkv_heads / tp من رؤوس KV.
• يتكامل TP مع CP على شبكة أجهزة ثنائية الأبعاد (tp, cp): حيث يقسم TP الرؤوس/الأوزان بشكل دائم، بينما يقسم CP التسلسل فوق ذلك. القيود هي أن tp يقسم رؤوس KV (8)، وأن tp * cp يقسم رؤوس الاستعلام (32 لـ Nano، و64 لـ Super).
• تُحمَّل الأوزان إلى وحدة المعالجة المركزية (CPU) ثم تُقسَّم على وحدات معالجة الرسوميات (GPUs) طبقةً طبقة، بحيث لا يتم إنشاء النموذج الكامل على جهاز واحد.

تصدر TP عملية all-reduce في كل كتلة انتباه وكتلة MLP، لذا فهي كثيفة النطاق الترددي. على الأجهزة التي لا تدعم NVLink، تُعدّ هذه العملية التكلفة المهيمنة؛ يُفضَّل استخدام درجة TP الأصغر التي تجعل الأوزان تتسع، وإدراج وحدات المعالجة الرسومية المتبقية في CP.

استخدمه في خط أنابيبك الخاص ذي البنية المعيارية

تُعدّ أعلام سطر الأوامر مريحة، لكن يمكنك استدعاء أدوات المساعدة مباشرةً باستخدام Cosmos3OmniModularPipeline. حمّل إعدادات خط الأنابيب ومكوناته على المعالج المركزي (CPU)، وطبّق قبل نقل خط الأنابيب إلى وحدة معالجة الرسوميات المحلية للرتبة (rank-local GPU)، ثم بدّل إلى الخلفية 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 فقط (بدون تقسيم الأوزان)، استخدم شبكة أحادية البعد: init_device_mesh("cuda", (world_size,), mesh_dim_names=("cp",)) وفعّل enable_cosmos3_context_parallel فقط.

enable_safety_checker() يحمّل ويفعّل الفاحص الافتراضي؛ بينما disable_safety_checker() يعطّله صراحةً. استخدم تلك الطرق الخاصة بالخطوط الأنابيب بدلاً من معامل الإنشاء enable_safety_checker= الخاص بخط الأنابيب المهامي أو معامل الاستدعاء enable_safety_check=. كما لا تُرجع الخطوط الأنابيب المعيارية Cosmos3OmniPipelineOutput: استخدم output="videos" للإطارات وحدها، أو قائمة مخرجات وديكتمونيها المُرجَع كما هو موضح أعلاه بدلاً من result.video أو result.sound أو result.action.

قد تتعطل أول عملية all-to-all في NCCL على بعض التكوينات متعددة وحدات معالجة الرسوميات. إذا توقفت عملية CP في بداية خطوة إزالة الضوضاء الأولى، فاضبط NCCL_P2P_DISABLE=1 في البيئة قبل تشغيل torchrun.

يتكامل CP وTP مع جميع سير العمل المذكورة أعلاه (النص إلى فيديو، الصورة إلى فيديو، النص إلى فيديو مع صوت، والتوليد المشروط بالإجراء) ومع كلا نقطتي التفتيش Nano وSuper — حيث تتغير فقط أسطر بناء خط الأنابيب وإعداد التوازي.

قوالب البيانات الوصفية

tokenize_prompt appends short metadata sentences inside the user message so the LLM sees the conditioning the model was trained with. The positive prompt gets sentences like “The video is 7.9 seconds long and is of 24 FPS.” and “This video is of 720x1280 resolution.”; the negative prompt gets the inverse (”… is not …”).

كلاهما مفعّل افتراضيًا. يمكنك تعطيل أيٍّ من الزوجين عبر __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
)

لا يؤثر add_duration_template عندما يكون num_frames == 1 (وضع الصورة)؛ في هذه الحالة يُضاف جملة الدقة فقط.

فاحص الأمان

يربط 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

• الكل
• call

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 ودع إعداد المستودع اختيار فئة الكتل:

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). ولأن النقل (التحكم الهيكلي) له مدخلاته الخاصة، فقد تم عرضه بشكل منفصل في النقل المعياري أدناه.

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)

النقل المعياري (التحكم الهيكلي)

يتبع النقل فيديو تحكم محسوب مسبقاً (حواف، ضبابية، عمق، تجزئة، أو خريطة سيناريو عالمي) يُمرر عبر control_videos= على شكل {hint: video}. وهو مخصص للفيديو فقط (لا يحتوي على image / video / action / enable_sound)، والموجّه عبارة عن وصف JSON مُوسَّع مسبقاً (انظر توسيع الموجّه)، وتُولَّد المقاطع الطويلة ذاتياً في كتل من num_video_frames_per_chunk وتُدمج تلقائياً. guidance_scale هو CFG النصي المعتاد؛ بينما control_guidance (!= 1.0) يُعزّز إشارة التحكم بشكل إضافي. القيم الابتدائية الموصى بها لكل تلميح:

لا يتضمن Diffusers أصول التحكم. الأصول الجاهزة (فيديو تحكم + ملف prompt.json مطابق لكل تلميح، بالإضافة إلى ملف negative_prompt.json مشترك) متوفرة في دليل Cosmos. بالنسبة لمثال الحافة أدناه، قم بتنزيلها إلى مجلد محلي باسم 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 (الكتل: Cosmos3DistilledBlocks)؛ ولا يدعم خطا الأنابيب الأساسيان Cosmos3OmniModularPipeline وCosmos3OmniPipeline هذه النقاط. يُثبَّت num_inference_steps على طول إعداد خط الأنابيب distilled_sigmas (من ملف modular_model_index.json الخاص بنقطة التفتيش)، ويُجبر guidance_scale على القيمة 1.0 لأن التوجيه مدمج في الأوزان — فتمرير أي قيمة أخرى لأيٍّ منهما يثير خطأً، بينما يُصدر تحذير تجاه negative_prompt ويُتجاهل.

لا تُعلن نقاط التفتيش المُقطّرة بدقة FP8 (revision="fp8") عن سياسة دقة مختلطة، لذا تبقى كل خطوة بدقة W8A8 الأصلية.

تتبع الأوامر (Prompts) نفس بنية JSON الوصفية المستخدمة في النماذج غير المُصغَّرة (non-distilled)، لذا يجب رفع دقة النص القصير أولاً — استخدم --mode text2image (T2I) أو --mode image2video (I2V) كما هو موضح في رفع دقة الأوامر، ثم مرر JSON عبر json.dumps(...).

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)

• الكل
• call

Cosmos3DistilledModularPipeline

• الكل
• call

CosmosActionCondition

Cosmos3OmniPipelineOutput