DiffusionGemma

DiffusionGemma هو نموذج لغة مُرمِّز-مُفكِّك يعتمد على الانتشار على مستوى الكتل. يقرأ المُرمِّز السببي الموجه النظيف (وأي كتل تم توليدها سابقًا) إلى ذاكرة KV، بينما يُزيل المُفكِّك الاتجاهي الضوضاء من «لوحة» بحجم ثابت تتكون من canvas_length رمزًا من خلال الانتباه المتقاطع إلى تلك الذاكرة. يتناوب التوليد بين حلقة خارجية ذاتية التراجع على مستوى اللوحات وحلقة داخلية لإزالة الضوضاء، حيث يقوم كل خطوة فيها بسحب رموز مرشحة، والتأكيد على الأكثر ثقة منها عبر BlockRefinementScheduler في وضع التلطيخ المنتظم، وإعادة إضافة الضوضاء إلى البقية. يقع النموذج نفسه في transformers على شكل DiffusionGemmaForBlockDiffusion؛ والنقطة الفاحصة المُصدَّرة هي google/diffusiongemma-26B-A4B-it.

الاستخدام

import torch
from transformers import AutoProcessor, DiffusionGemmaForBlockDiffusion

from diffusers import BlockRefinementScheduler, DiffusionGemmaPipeline

model_id = "google/diffusiongemma-26B-A4B-it"
model = DiffusionGemmaForBlockDiffusion.from_pretrained(model_id, dtype=torch.bfloat16, device_map="auto")
processor = AutoProcessor.from_pretrained(model_id)
scheduler = BlockRefinementScheduler()

pipe = DiffusionGemmaPipeline(model=model, scheduler=scheduler, processor=processor)
pipe.model.model.decoder = torch.compile(pipe.model.model.decoder, mode="reduce-overhead", fullgraph=True)
output = pipe(
    prompt="Why is the sky blue?",
    gen_length=256,
    num_inference_steps=48,
    cache_implementation="static",
)
print(output.texts[0])

num_inference_steps هو عدد خطوات إزالة الضوضاء لكل لوحة (48 يتطابق مع نقطة التفتيش المنشورة)؛ فعدد خطوات أقل يكون أسرع لكن بجودة أدنى. يتيح cache_implementation="static" أن يكون المُفكِّك قابلاً للتجميع باستخدام torch.compile مع cudagraphs (انظر التخزين المؤقت الثابت والتجميع)؛ احذف كليهما للحصول على تشغيل أبسط بتخزين مؤقت ديناميكي.

للإدخالات متعددة الأدوار أو متعددة الوسائط، مرّر محادثة خام messages بدلاً من prompt. وهو قائمة من قواميس {"role", "content"} بالصيغة المعتادة للمحادثات، يمرّ المعالج بها عبر قالب المحادثة الخاص به:

messages = [
    {"role": "user", "content": "Why is the sky blue?"},
]
# or with an image:
messages = [
    {
        "role": "user",
        "content": [
            {"type": "image", "image": image},
            {"type": "text", "text": "Describe this image."},
        ],
    },
]
output = pipe(messages=messages, gen_length=256)

في حالة دور مستخدم واحد، يمكنك تجاوز messages وتقديم image إلى جانب prompt؛ حيث يقوم المعالج بتحويلها تلقائيًا إلى مدخلات الصورة الخاصة بالنموذج.

المجدولون

المجدول هو المُعيّن الذي يُزيل الضوضاء من كل لوحة، وهو قابل للاستبدال: استبدله لتغيير استراتيجية العينة دون المساس بأي شيء آخر. يتوفر ثلاثة مجدول

• BlockRefinementScheduler (الافتراضي): يثبّت الرموز الأكثر ثقة في كل خطوة (تلك التي تتجاوز threshold، بالإضافة إلى حصة متساوية لكل خطوة) ويعيد إضافة الضوضاء إلى البقية. كما يتيح editing_threshold إعادة تحرير الرموز المثبتة مسبقًا.
• DiscreteDDIMScheduler: يعيّن كل موضع من التوزيع الخلفي المنفصل الدقيق لعملية التلوث المنتظمة (D3PM). وهو خالٍ من المعلمات، والخطوة النهائية تلتزم بشكل حتمي بالرموز المتوقعة.
• EntropyBoundScheduler: يُثبّت المواضع ذات الإنتروبيا الأدنى التي تبقى إنتروبياها المشتركة دون entropy_bound، بحيث تُقبل الرموز شبه المستقلة معًا. وهو يُبرّد درجة حرارة العينة من t_max (0.8) في الخطوة الأولى إلى t_min (0.4) في الخطوة الأخيرة، بما يتوافق مع مُعيّن نقاط التفتيش المُصدر.

from diffusers import DiscreteDDIMScheduler, EntropyBoundScheduler

pipe.scheduler = DiscreteDDIMScheduler()
# or: pipe.scheduler = EntropyBoundScheduler(entropy_bound=0.1)
output = pipe(prompt="Why is the sky blue?", gen_length=256, num_inference_steps=48)
print(output.texts[0])

تُضبط خيارات العينة الخاصة بالجدول الزمني (مثل threshold/top_k لتحسين الكتلة، وحد الإنتروبيا، ...) على إعدادات الجدول الزمني:

from diffusers import BlockRefinementScheduler

pipe.scheduler = BlockRefinementScheduler.from_config(pipe.scheduler.config, threshold=0.9)

يُخفّض EntropyBoundScheduler درجة حرارة العينة (t_max/t_min) داخلياً على مدى خطوات إزالة الضوضاء؛ بينما يستخدم DiscreteDDIMScheduler وBlockRefinementScheduler قيمة temperature الثابتة المُمرَّرة إلى خط الأنابيب (0.0 للاستخدام الجشع).

عينة التنبؤ والتصحيح

DiscreteDDIMScheduler يدعم مُصحِّح التنبؤ-التصحيح مع استبعاد عنصر واحد (leave-one-out predictor-corrector) في Uniform Diffusion Models Revisited: Leave-One-Out Denoiser and Absorbing State Reformulation. فهو يُحسّن اللوحة عبر corrector_steps من عمليات مسح غيبس (Gibbs sweeps) تعيد أخذ عينات من المواضع الأقل ثقة من الشرطي أحادي الإحداثي (one-coordinate conditional) للتوزيع الحدي الضوضائي (noisy marginal)، مما يُبقي ذلك التوزيع الحدي دون تغيير ويحسّن التوليد دون أي تكلفة تدريب إضافية. ويعمل مباشرة على نقطة التفتيش (checkpoint) المنشورة: ففي الانتشار المنتظم (uniform diffusion) يمكن استبدال مُزيل الضوضاء (denoiser) والخلفي (posterior) الخاص باستبعاد عنصر واحد بشكل مكافئ بصيغة مغلقة، لذا يستعيد المُصحِّح كميات استبعاد عنصر واحد التي يحتاجها دون أي إعادة تدريب.

تُدمج عمليات المسح المُصحِّحة ضمن ميزانية num_inference_steps بدلاً من إضافتها فوقها: حيث يُنفّذ خط الأنابيب عدداً أقل من خطوات المُتنبّئ ويخصص التقدّمات المحرَّرة للمُصحِّحات، وبالتالي يبقى إجمالي عدد تقدّمات النموذج num_inference_steps وتكون تكلفة المُتنبّئ-المُصحِّح مماثلة لتكلفة العيّنات الأصلية البسيطة.

from diffusers import DiscreteDDIMScheduler

pipe.scheduler = DiscreteDDIMScheduler(corrector_steps=2, corrector_k=12)
output = pipe(prompt="Why is the sky blue?", gen_length=256, num_inference_steps=48)
print(output.texts[0])

مُكيّفات PEFT

يعتمد المُزيل للضوضاء على نموذج 🤗 Transformers، لذا يتم تحميل المُكيّفات (adapters) عبر تكامل PEFT الأصلي الخاص به بدلاً من واجهة load_lora_weights في مكتبة diffusers. وبما أن هذا التكامل لا يعتمد على نوع المُكيّف، فإن نفس الاستدعاءات تُحمّل LoRA أو DoRA أو أي مُكيّف PEFT آخر (مثل ناتج SFTTrainer من TRL). قم بإدارة المُكيّفات مباشرة على مكوّن النموذج:

pipe.model.load_adapter("path/to/adapter", adapter_name="sft")  # LoRA, DoRA, ...
pipe.model.set_adapter("sft")
output = pipe(prompt="Why is the sky blue?", gen_length=256)

pipe.model.disable_adapters()  # run the base model
pipe.model.delete_adapter("sft")

تبقى المُكيّفات نشطة وغير مدمجة: يربط DiffusionGemma أوزان الأساس للمُشفِّر والمُفكِّك، لذا فإن دمج المُكيّف فيهما سيُفسد كلا الفرعين.

التخزين المؤقت الثابت والتجميع

يُعبّئ خطّ الأنابيب المُشفِّر مرة واحدة لكل كتلة في ذاكرة مؤقتة قابلة لإعادة الاستخدام (من نوع DynamicCache افتراضيًا). عند تمرير cache_implementation="static"، يتم استخدام StaticCache ذو شكل ثابت بدلاً من ذلك، مما يتيح لك torch.compile المُفكِّك باستخدام cudagraphs لتحقيق تسريع إضافي (حيث يحدد خطّ الأنابيب كل خطوة وينسخ logits حتى لا يتم تجاوز ذاكرة cudagraph)؛ وهذا هو الإعداد المعروض في الاستخدام. احذف كلا من استدعاء torch.compile وcache_implementation="static" للحصول على تشغيل أبسط بذاكرة مؤقتة ديناميكية.

الإيقاف التكيفي

عادةً ما يتقارب الكتل قبل استنفاد جميع num_inference_steps، لذا يقوم خط الأنابيب افتراضيًا بتجميد كل مثال في الدفعة بمجرد أن يصبح تنبؤ argmax الخاص به مستقرًا لعدد stability_threshold من الخطوات وأن ينخفض متوسط إنتروبيا كل رمز إلى ما دون confidence_threshold (0.005، القيمة المستخدمة في نقطة التفتيش المنشورة). ينتهي حلقة إزالة الضوضاء بمجرد تجميد جميع الأمثلة. يقلل هذا تقريبًا من عدد عمليات تمرير المُفكِّك إلى النصف عند جودة مطابقة، وهو أكبر رافعة منفردة لتحسين الإنتاجية. مرر confidence_threshold=None لتشغيل num_inference_steps الكاملة دائمًا:

output = pipe(prompt="Why is the sky blue?", gen_length=256, confidence_threshold=None)  # disable adaptive stopping

المكالمات العكسية

تُنفَّذ استدعاءات المراجعة (Callbacks) بعد كل خطوة إزالة ضوضاء. مرّر callback_on_step_end_tensor_inputs لاختيار المصفوفات (tensors) التي تُدرَج في callback_kwargs؛ حيث يتوفر canvas (رموز الكتل الحالية) وlogits. أعد {"canvas": ...} من استدعاء المراجعة لاستبدال الـ canvas.

def on_step_end(pipe, step, timestep, callback_kwargs):
    canvas = callback_kwargs["canvas"]
    # Inspect or modify `canvas` here.
    return {"canvas": canvas}


out = pipe(
    prompt="Why is the sky blue?",
    callback_on_step_end=on_step_end,
    callback_on_step_end_tensor_inputs=["canvas"],
)

DiffusionGemmaPipeline

>>> import torch
>>> from transformers import AutoProcessor, DiffusionGemmaForBlockDiffusion
>>> from diffusers import BlockRefinementScheduler, DiffusionGemmaPipeline

>>> model_id = "google/diffusiongemma-26B-A4B-it"
>>> model = DiffusionGemmaForBlockDiffusion.from_pretrained(model_id, dtype=torch.bfloat16, device_map="auto")
>>> processor = AutoProcessor.from_pretrained(model_id)
>>> scheduler = BlockRefinementScheduler()

>>> pipe = DiffusionGemmaPipeline(model=model, scheduler=scheduler, processor=processor)
>>> output = pipe(prompt="Why is the sky blue?", gen_length=256)
>>> print(output.texts[0])

DiffusionGemmaPipelineOutput