التقطيع في الوقت الفعلي (RTC)

يتيح التقطيع في الوقت الفعلي (RTC) لسياسات الروبوتات الكبيرة القائمة على مطابقة التدفق، مثل Pi0 وPi0.5 وSmolVLA، إنتاج حركة سلسة ومتصلة واستجابة سريعة رغم ارتفاع زمن الاستدلال. يوفر LeRobot وضع التوجيه الأصلي أثناء الاستدلال، وللفحوصات المتوافقة مع Pi05، تكييف الإجراء أثناء التدريب مع استدلال رخيص بتبعية صارمة.

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

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

كيف يعمل RTC (بشكل مبسّط)

يتيح RTC للروبوت التفكير مسبقًا أثناء حركته. عندما يكون الروبوت في طور تنفيذ جزء من الأفعال، يبدأ RTC في إنشاء الجزء التالي مبكرًا. ولكن بما أن الروبوت يكون قد تحرك قليلًا بحلول الوقت الذي يصبح فيه الجزء الجديد جاهزًا، يجب على RTC التأكد من أن الجزء الجديد يتوافق بسلاسة مع ما يقوم به الروبوت حاليًا.

للقيام بذلك، يعامل RTC بداية الجزء الجديد كمشكلة إعادة رسم أو “ملء الفجوات”: فهو يعدّل بلطف الجزء الأول من الجزء الجديد ليتكامل بشكل طبيعي مع حركة الروبوت المستمرة. والنتيجة هي عدم وجود توقفات أو قفزات مفاجئة.

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

البدء السريع

التثبيت

يُدمج RTC في LeRobot. ما عليك سوى تثبيت تبعيات السياسة التي تحتاجها:

# For Pi0 or Pi0.5
pip install -e ".[pi]"

# For SmolVLA
pip install -e ".[smolvla]"

استخدام RTC مع Pi0

يمكنك استخدام lerobot-rollout --strategy.type=base --inference.type=rtc لنشر RTC على روبوتات حقيقية. يوفر المقطع أدناه مثالًا زائفًا مبسطًا لكيفية عمل RTC مع Pi0 في خط أنابيبك:

from lerobot.policies.pi0 import PI0Policy, PI0Config
from lerobot.configs import RTCAttentionSchedule
from lerobot.policies.rtc import RTCConfig, ActionQueue

# Load the Pi0 config so RTC can be enabled before the policy is built
policy_cfg = PI0Config.from_pretrained("lerobot/pi0_base")

# Enable RTC
policy_cfg.rtc_config = RTCConfig(
    enabled=True,
    execution_horizon=10,  # How many steps to blend with previous chunk
    max_guidance_weight=10.0,  # How strongly to enforce consistency
    prefix_attention_schedule=RTCAttentionSchedule.EXP,  # Exponential blend
)

# Load the policy
policy = PI0Policy.from_pretrained("lerobot/pi0_base", config=policy_cfg).to("cuda")

# Now use predict_action_chunk with RTC parameters
inference_delay = 4  # How many steps of inference latency, this value should be calculated based on the inference latency of the policy

# Initialize the action queue
action_queue = ActionQueue(policy_cfg.rtc_config)

# Start in a separate thread with the following function
def get_actions():
  while True:
    if should_get_actions:

      prev_actions = action_queue.get_left_over()
      obs = get_robot_observations(robot)

      # Generate actions WITH RTC
      actions = policy.predict_action_chunk(
          obs,
          inference_delay=inference_delay,
          prev_chunk_left_over=prev_actions,
      )

      action_queue.merge(
          actions, actions, inference_delay
      )

for step in range(num_steps):
    action = action_queue.get()

    # Execute the first N actions
    execute_actions(action)

المعلمات الرئيسية

لـ RTCConfig المعاملات التالية التي يمكن ضبطها:

يحدد mode طريقة تكييف بادئة الإجراء:

• guided (الافتراضي) يطبّق التوجيه الأصلي عبر جاكوبيان أثناء إزالة الضوضاء، ويعمل مع نقاط التثبيت العادية لمطابقة التدفق.
• trained يقوم بتطبيق inpainting قوي على بادئة المقطع السابق باستخدام خطوات تدفق لكل إجراء. ويتطلب حاليًا نقطة تحقق Pi05 تم تدريبها باستخدام policy.rtc_training_max_delay > 0 ويتجنب مسار التوجيه العكسي.

تطبّق عملية النشر المحلية وخادم السياسة نفس قاعدة القبول المتحفظة في وضع التدريب: rtc_training_max_delay <= execution_horizon <= prediction_steps - rtc_training_max_delay. بالنسبة إلى Pi05، يمثل prediction_steps قيمة chunk_size في نقطة الحفظ. يحافظ هذا على الحد المحلي الحالي: فالبادئة المُعدّة تغطي أقصى تأخير مُدرَّب وتترك على الأقل عددًا مماثلًا من الخطوات المتوقعة خارجها. تتطلب عملية النشر المحلية أيضًا inference.queue_threshold >= rtc_training_max_delay قبل توصيل الروبوت؛ بينما يستخدم توقيت الطلبات البعيدة إعدادات إعادة التعبئة الخاصة به.

هنا يُوفَّر execution_horizon كسعة بادئة مُدخلة، وليس كمدة التشغيل بين الطلبات. يُشترط Pi05 بشدة على بادئة inference_delay فقط. قاعدة القبول ليست ضمانًا زمنياً ولا ادعاءً بأن السعات الأكبر يجب أن تفشل وقت التشغيل؛ إذ لا تزال القياسات المتعلقة بالتداخل، والإجراءات المتبقية، وفحوصات حداثة البيانات هي التي تحدد ما إذا كان بإمكان المقطع المُعاد التنفيذ. المقاطع العادية وRTC الموجَّه غير متأثرين بهذه القاعدة المُدرَّبة فقط.

execution_horizon: عدد خطوات الزمن من المقطع السابق التي يجب الحفاظ على الاتساق معها. تشير القيم الأعلى إلى انتقالات أكثر سلاسة، لكنها قد تعني تفاعلية أقل.

القيم النموذجية: 8-12 خطوة

RTCConfig(execution_horizon=10)

max_guidance_weight: مدى قوة فرض الاتساق مع الجزء السابق. هذا معامل فائق يمكن ضبطه لموازنة سلاسة الانتقالات وردّ فعل السياسة. بالنسبة لمطابقة التدفق على 10 خطوات (SmolVLA، Pi0، Pi0.5)، فإن القيمة 10.0 هي القيمة المثلى.

prefix_attention_schedule: كيفية وزن الاتساق عبر منطقة التداخل.

• LINEAR: تضاؤل خطي من inference_delay إلى execution_horizon
• EXP: اضمحلال أُسّي (يُنصح به عند البدء)
• ONES: وزن كامل عبر أفق التنفيذ بالكامل
• ZEROS: ثنائي (الوزن الكامل حتى inference_delay، ثم صفر)

inference_delay: عدد خطوات الاستدلال التي يمثلها زمن تأخير الاستدلال في نظامك. يتم تمرير هذا إلى predict_action_chunk() بدلاً من الإعدادات، لأنه قد يتغير أثناء التشغيل.

اختبار RTC دون اتصال

قبل التشغيل على روبوت حقيقي، اختبر RTC باستخدام عينات من مجموعة البيانات لتصور كيفية عمله:

python examples/rtc/eval_dataset.py \
    --policy.path=lerobot/pi0_libero_finetuned \
    --dataset.repo_id=HuggingFaceVLA/libero \
    --rtc.execution_horizon=10 \
    --rtc.max_guidance_weight=10.0 \
    --device=cuda

أضف --rtc.mode=trained عند تقييم نقطة تفتيش Pi05 متوافقة مع تدريب RTC أثناء التدريب. ترفض السياسات غير المدعومة الوضع المدرّب بدلاً من الرجوع إلى RTC الموجَّه.

يولّد النص البرمجي تمثيلاً مرئياً لعملية إزالة الضوضاء، ويقارن بين التوليد القياسي (يسار) و (يمين). في الرسوم البيانية الخاصة بـ، يمكنك ملاحظة كيف تُوجَّه الخطوات الأولى (الخطوط الزرقاء/البنفسجية) لتتطابق مع مسار الحقيقة المرجعية الأحمر (نهاية المقطع السابق)، مما يضمن انتقالًا سلسًا بين المقاطع.

اختبار RTC مع روبوت حقيقي

lerobot-rollout \
    --strategy.type=base \
    --policy.path=${HF_USERNAME}/policy_repo_id \
    --inference.type=rtc \
    --inference.rtc.mode=guided \
    --inference.rtc.execution_horizon=10 \
    --inference.rtc.max_guidance_weight=10.0 \
    --robot.type=so100_follower \
    --robot.port=/dev/tty.usbmodem58FA0834591 \
    --robot.cameras="{ gripper: {type: opencv, index_or_path: 1, width: 640, height: 480, fps: 30}, front: {type: opencv, index_or_path: 0, width: 640, height: 480, fps: 30}}" \
    --task="Move green small object into the purple platform" \
    --duration=120 \
    --device=cuda

لنقطة تفتيش RTC Pi05 وقت التدريب، غيّر الوضع إلى trained. تسجّل نقطة التفتيش أقصى تأخير مدعوم، ويتحقق rollout من زمن الاستجابة المقاس مقابل ذلك:

lerobot-rollout \
    --strategy.type=base \
    --policy.path=${HF_USERNAME}/pi05_training_rtc \
    --inference.type=rtc \
    --inference.rtc.mode=trained \
    --inference.rtc.execution_horizon=10 \
    --robot.type=so100_follower \
    --robot.port=/dev/tty.usbmodem58FA0834591 \
    --task="Move green small object into the purple platform" \
    --duration=120 \
    --device=cuda

كيف يتعلق RTC بالاستدلال عن بُعد

--inference.type=rtc هو RTC محلي: يتم تنفيذ الاستدلال في خيط خلفية داخل عملية rollout ولا يستخدم أي نقل. تشغيل خادم سياسة وعميل روبوت كعمليتين منفصلتين على نفس الجهاز يستخدم الخلفية البعيدة بدلاً من ذلك.

تستخدم كلا الخوادم الخلفية غير المتزامنة نفس سلوك الانتظار المحدود والإيقاف على الروبوتات المدعومة. راجع مخصصات RTC المحلية للاطلاع على القيم الافتراضية لوقت انتهاء المهلة. لا يؤدي الانتظار إلى تخفيف فحوصات RTC الخاصة بالبداية أو التأخير أو حداثة البيانات.

تحسّن كلٌّ من RTC والاستدلال عن بُعد التحكم في الروبوتات في الوقت الفعلي، لكنهما يحلان مشكلتين مختلفتين.

يمكن للاستدلال عن بُعد استخدام RTC عندما يدعم نقطة الحفظ (checkpoint) وضع RTC المتفاوض عليه. كما تدعم المقاطع العادية محاذاة المؤشر الاختيارية والدمج؛ راجد دليل الاستدلال عن بُعد لمعرفة الأوضاع والتوافق وضبط الأداء.

متقدم: تتبع التصحيح

يتضمن RTC تتبعًا مدمجًا للأخطاء يساعدك على فهم ما يحدث أثناء الاستدلال:

# Enable debug tracking
policy_cfg.rtc_config.debug = True
policy_cfg.rtc_config.debug_maxlen = 100
policy.init_rtc_processor()  # re-create the RTC processor so the new debug settings take effect

# After inference, access debug data
debug_data = policy.rtc_processor.get_all_debug_steps()

# Visualize denoising steps, corrections, etc.
from lerobot.policies.rtc.debug_visualizer import RTCDebugVisualizer
visualizer = RTCDebugVisualizer()
# ... create plots

راجع examples/rtc/eval_dataset.py للحصول على مثال كامل لتصور RTC دون اتصال.

المراجع

• سياسات روبوتية سلسة كالزبدة - شرح تقني ممتاز مع نتائج روبوتات حقيقية
• فيزياء الذكاء - التقطيع في الوقت الفعلي - الورقة البحثية والأبحاث الأصلية
• تنفيذ Kinetix RTC - التنفيذ المرجعي من Physical Intelligence
• شرطية الإجراءات أثناء التدريب - RTC فعّال مع شرطية بادئة نظيفة أثناء التدريب
• RLDX-1 - مرجع PyTorch المستخدم لتكامل RTC أثناء التدريب