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

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

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

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

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

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

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

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

## البدء السريع

### التثبيت

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

```bash
# 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 في خط أنابيبك:

```python
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 خطوة

```python
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 باستخدام عينات من مجموعة البيانات لتصور كيفية عمله:

```bash
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 الموجَّه.

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

  <img
    src="https://huggingface.co/datasets/huggingface/documentation-images/resolve/main/lerobot/flow_matching.png"
    alt="Denoising steps with and without RTC"
    width="100%"
  />

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

```bash
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 من زمن الاستجابة المقاس مقابل ذلك:

```bash
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 ولا يستخدم أي نقل. تشغيل خادم سياسة وعميل روبوت كعمليتين منفصلتين على نفس الجهاز يستخدم الخلفية البعيدة بدلاً من ذلك.

تستخدم كلا الخوادم الخلفية غير المتزامنة نفس [سلوك الانتظار المحدود والإيقاف](./remote_inference#what-the-robot-does-while-waiting-or-stopping) على الروبوتات المدعومة. راجع [مخصصات RTC المحلية](./inference#inference-backends) للاطلاع على القيم الافتراضية لوقت انتهاء المهلة. لا يؤدي الانتظار إلى تخفيف فحوصات RTC الخاصة بالبداية أو التأخير أو حداثة البيانات.

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

| الجانب           | الاستدلال عن بُعد                                              | RTC                                                   |
| ---------------- | ------------------------------------------------------------- | ----------------------------------------------------- |
| **الغرض**      | تشغيل الاستدلال في عملية أخرى أو على حاسوب آخر       | توقعات الحالة على المسار المتبقي     |
| **التنفيذ**    | التشغيل المخزّن محليًا، مع المحاذاة والدمج الاختياريين      | الاستكمال الموجَّه أو المدرَّب للسياسات المدعومة |
| **العلاقة** | يمكن تقديم مقاطع عادية أو مقاطع RTC                                 | يمكن التنفيذ محليًا أو من خلال الخادم البعيد     |
| **القيود**       | ما تزال زمن الاستجابة، حداثة البيانات، وتغطية المخزن المؤقت عوامل تحدّ من الحركة | يتطلب عقد RTC المُعلن عنه في السياسة           |

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

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

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

```python
# 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 دون اتصال.

## المراجع

- [سياسات روبوتية سلسة كالزبدة](https://alexander-soare.github.io/robotics/2025/08/05/smooth-as-butter-robot-policies.html) - شرح تقني ممتاز مع نتائج روبوتات حقيقية
- [فيزياء الذكاء - التقطيع في الوقت الفعلي](https://www.physicalintelligence.company/research/real_time_chunking) - الورقة البحثية والأبحاث الأصلية
- [تنفيذ Kinetix RTC](https://github.com/Physical-Intelligence/real-time-chunking-kinetix) - التنفيذ المرجعي من Physical Intelligence
- [شرطية الإجراءات أثناء التدريب](https://arxiv.org/abs/2512.05964) - RTC فعّال مع شرطية بادئة نظيفة أثناء التدريب
- [RLDX-1](https://github.com/RLWRLD/RLDX-1) - مرجع PyTorch المستخدم لتكامل RTC أثناء التدريب