مقدمة إلى المعالجات

في الروبوتات، يوجد تفاوت أساسي بين البيانات التي ينتجها الروبوتات والبشر وما تتوقعه نماذج التعلم الآلي. تُخرج الروبوتات بيانات مستشعرات خام مثل صور الكاميرا ومواضع المفاصل، والتي تحتاج إلى تطبيع، وتجميع، وتحديد الموقع على الجهاز قبل أن تتمكن النماذج من معالجتها. يجب تحويل التعليمات اللغوية من البشر إلى تمثيلات رقمية، كما أن الروبوتات المختلفة تستخدم أنظمة إحداثيات مختلفة تحتاج إلى توحيد.

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

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

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

هذا يعني أنه يمكن استخدام سياسة المفضلة لديك على النحو التالي:

import torch

from lerobot.datasets import LeRobotDataset
from lerobot.policies import make_pre_post_processors
from lerobot.policies.your_policy import YourPolicy
from lerobot.processor import RobotProcessorPipeline, PolicyProcessorPipeline
dataset = LeRobotDataset("hf_user/dataset", episodes=[0])
sample = dataset[10]

model = YourPolicy.from_pretrained(
    "hf_user/model",
)
model.eval()
model.to("cuda")
preprocessor, postprocessor = make_pre_post_processors(model.config, pretrained_path="hf_user/model", dataset_stats=dataset.meta.stats)

preprocessed_sample = preprocessor(sample)
action = model.select_action(preprocessed_sample)
postprocessed_action = postprocessor(action)

ما هي المعالجات؟

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

• مُطَبَّع: مُقَيَّس إلى نطاقات مناسبة لمعالجة الشبكات العصبية
• مجمّع: مُنظَّم بأبعاد مناسبة لمعالجة الدفعات
• مُرمّز: نص تم تحويله إلى تمثيلات رقمية
• مُوضع على الجهاز: تم نقله إلى العتاد المناسب (وحدة المعالجة المركزية/وحدة معالجة الرسوميات)
• تحويل النوع: يُحوَّل إلى أنواع البيانات المناسبة

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

المفاهيم الأساسية

EnvTransition: حاوية البيانات الشاملة

يُعدّ EnvTransition هيكل البيانات الأساسي الذي يمر عبر جميع المعالجات. وهو قاموس مُنمّط (typed dictionary) يُمثّل تفاعلًا كاملًا بين الروبوت والبيئة:

• الملاحظة: جميع بيانات المستشعرات (الصور، الحالات، الإدراك الذاتي)
• ACTION: الإجراء الذي يجب تنفيذه أو الذي تم تنفيذه
• REWARD: إشارة التعلم المعزز
• DONE/TRUNCATED: مؤشرات حدود الحلقة
• معلومات: بيانات وصفية اختيارية
• COMPLEMENTARY_DATA: أوصاف المهام، الفهارس، أعلام الحشو، البيانات بين الخطوات

تحتفظ محولات الدفعات ببيانات وصفية للفهرسة مثل index وepisode_index وframe_index في COMPLEMENTARY_DATA. في رحلة ذهاب وإياب بين batch_to_transition وtransition_to_batch، يحتفظ frame_index بقيمته الأصلية سواء كانت قياسية أو متجهية. إذا لم يتضمن الإدخال frame_index، فإن المحولات لا تضيفه.

ProcessorStep: وحدة البناء

يُعدّ ProcessorStep وحدة تحويل مفردة تعالج الانتقالات. وهو فئة أساسية مجردة تتضمن طريقتين مطلوبتين:

from lerobot.processor import ProcessorStep, EnvTransition

class MyProcessorStep(ProcessorStep):
    """Example processor step - inherit and implement abstract methods."""

    def __call__(self, transition: EnvTransition) -> EnvTransition:
        """Transform the transition - REQUIRED abstract method."""
        # Your processing logic here
        return transition

    def transform_features(self, features):
        """Declare how this step transforms feature shapes/types - REQUIRED abstract method."""
        return features  # Most processors return features unchanged

__call__ هو جوهر خطوة المعالج لديك. يستقبل EnvTransition ويعيد EnvTransition معدّلاً.

يُستخدم transform_features للإعلان عن كيفية تحويل هذا الخطوة لأشكال/أنواع الميزات.

DataProcessorPipeline: المنسّق العام

تربط DataProcessorPipeline[TInput, TOutput] عدة كائنات من ProcessorStep معًا:

from lerobot.processor import RobotProcessorPipeline, PolicyProcessorPipeline

# For robot hardware (unbatched data)
robot_processor = RobotProcessorPipeline[RobotAction, RobotAction](
    steps=[step1, step2, step3],
    name="robot_pipeline"
)

# For model training/inference (batched data)
policy_processor = PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
    steps=[step1, step2, step3],
    name="policy_pipeline"
)

RobotProcessorPipeline مقابل PolicyProcessorPipeline

الفرق الجوهري يكمن في هياكل البيانات التي يتعامل معها:

استخدم RobotProcessorPipeline لواجهات العتاد الخاصة بالروبوتات:

# Robot data structures: dict[str, Any] for observations and actions
robot_obs: dict[str, Any] = {
    "joint_1": 0.5,           # Individual joint values
    "joint_2": -0.3,
    "camera_0": image_array   # Raw camera data
}

robot_action: dict[str, Any] = {
    "joint_1": 0.2,          # Target joint positions
    "joint_2": 0.1,
    "gripper": 0.8
}

استخدم PolicyProcessorPipeline لتدريب النموذج ومعالجة الدفعات:

# Policy data structures: batch dicts and tensors
policy_batch: dict[str, Any] = {
    "observation.state": torch.tensor([[0.5, -0.3]]),      # Batched states
    "observation.images.camera0": torch.tensor(...),        # Batched images
    "action": torch.tensor([[0.2, 0.1, 0.8]])              # Batched actions
}

policy_action: torch.Tensor = torch.tensor([[0.2, 0.1, 0.8]])  # Model output tensor

دوال المحوّل

توفر مكتبة LeRobot دوال تحويل لجسر الفجوة بين صيغ البيانات المختلفة في lerobot.processor.converters. تتولى هذه الدوال الترجمة الأساسية بين هياكل بيانات أجهزة الروبوت، وصيغ نماذج السياسة، وتمثيل EnvTransition الداخلي الذي يمر عبر خطوط معالجة البيانات.

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

أمثلة على المعالج

تُبيّن الأمثلة التالية تكوينات معالجات حقيقية لتدريب السياسات والاستدلال.

فيما يلي مثال على معالج لتدريب السياسة والاستدلال:

# Training data preprocessing (optimized order for GPU performance)
training_preprocessor = PolicyProcessorPipeline[dict[str, Any], dict[str, Any]](
    steps=[
        RenameObservationsProcessorStep(rename_map={}),     # Standardize keys
        AddBatchDimensionProcessorStep(),                   # Add batch dims
        TokenizerProcessorStep(tokenizer_name="...", ...),  # Tokenize language
        DeviceProcessorStep(device="cuda"),                 # Move to GPU first
        NormalizerProcessorStep(features=..., stats=...),   # Normalize on GPU
    ]
)

# Model output postprocessing
training_postprocessor = PolicyProcessorPipeline[torch.Tensor, torch.Tensor](
    steps=[
        DeviceProcessorStep(device="cpu"),                  # Move to CPU
        UnnormalizerProcessorStep(features=..., stats=...), # Denormalize
    ],
    to_transition=policy_action_to_transition,
    to_output=transition_to_policy_action,
)

تفاعل بين روبوت وسياسة مع معالجات

يجمع السيناريو الأكثر شيوعًا في العالم الحقيقي بين كلا نوعَي خطّ الأنابيب: فعتاد الروبوت يولّد ملاحظات تتطلب معالجة بواسطة السياسة، ومخرجات السياسة تتطلب ما بعد المعالجة المتوافقة مع الروبوت:

# Real deployment: Robot sensors → Model → Robot commands
with torch.no_grad():
    while not done:
        raw_obs = robot.get_observation()  # dict[str, Any]

        # Add your robot observation to policy observation processor

        policy_input = policy_preprocessor(raw_obs)  # Batched dict

        policy_output = policy.select_action(policy_input)  # Policy tensor

        policy_action = policy_postprocessor(policy_output)

        # Add your robot action to policy action processor

        robot.send_action(policy_action)

عقود الميزات: تحويل الشكل والنوع

لا تقتصر المعالجات على تحويل البيانات فحسب، بل يمكنها أيضاً تغيير بنية البيانات نفسها. تُعلن طريقة transform_features() عن هذه التغييرات، وهو أمر بالغ الأهمية لتسجيل مجموعات البيانات وإنشاء السياسات.

أهمية عقود الميزات

عند بناء مجموعات البيانات أو السياسات، يحتاج LeRobot إلى معرفة ما يلي:

• حقول البيانات التي ستوجد بعد المعالجة
• الأشكال والأنواع التي سيحملها كل حقل
• كيفية تكوين النماذج لبنية البيانات المتوقعة

# Example: A processor that adds velocity to observations
class VelocityProcessor(ObservationProcessorStep):
    def observation(self, obs):
        new_obs = obs.copy()
        if "observation.state" in obs:
            # concatenate computed velocity field to the state
            new_obs["observation.state"] = self._compute_velocity(obs["observation.state"])
        return new_obs

    def transform_features(self, features):
        """Declare the new velocity field we're adding."""
        state_feature = features[PipelineFeatureType.OBSERVATION].get("observation.state")
        if state_feature:
            double_shape = (state_feature.shape[0] * 2,) if state_feature.shape else (2,)
            features[PipelineFeatureType.OBSERVATION]["observation.state"] = PolicyFeature(
                type=FeatureType.STATE, shape=double_shape
            )
        return features

دوال مواصفات الميزات

تحل الدالتان create_initial_features() وaggregate_pipeline_dataset_features() مشكلة حاسمة في إنشاء مجموعات البيانات: تحديد بنية البيانات النهائية الدقيقة قبل معالجة أي بيانات. بما أن خطوط معالجة البيانات (processor pipelines) يمكن أن تضيف ميزات جديدة (مثل حقول السرعة)، أو تغيّر أشكال المصفوفات (مثل قصّ الصور)، أو تعيد تسمية المفاتيح، فإن مجموعات البيانات تحتاج إلى معرفة المواصفات الكاملة للمخرجات مسبقًا لتخصيص مساحة تخزين مناسبة وتعريف المخططات (schemas). تعمل هاتان الدالتان معًا بالبدء بمواصفات العتاد الميكانيكي للروبوت (create_initial_features()) ثم محاكاة تحويل خط المعالجة بالكامل (aggregate_pipeline_dataset_features()) لحساب قاموس الميزات النهائي الذي يُمرَّر إلى LeRobotDataset.create()، مما يضمن توافقًا تامًا بين ما تخرجه وحدات المعالجة وما تتوقع مجموعات البيانات تخزينه.

from lerobot.datasets import aggregate_pipeline_dataset_features

# Start with robot's raw features
initial_features = create_initial_features(
    observation=robot.observation_features,  # {"joint_1.pos": float, "camera_0": (480,640,3)}
    action=robot.action_features            # {"joint_1.pos": float, "gripper.pos": float}
)

# Apply processor pipeline to compute final features
final_features = aggregate_pipeline_dataset_features(
    pipeline=my_processor_pipeline,
    initial_features=initial_features,
    use_videos=True
)

# Use for dataset creation
dataset = LeRobotDataset.create(
    repo_id="my_dataset",
    features=final_features,  # Knows exactly what data to expect
    ...
)

خطوات المعالج الشائعة

يوفر LeRobot العديد من خطوات المعالج المسجلة. فيما يلي المعالجات الأساسية الأكثر استخدامًا:

المعالجات الأساسية

• normalizer_processor: تطبيع الملاحظات/الإجراءات باستخدام إحصاءات مجموعة البيانات (المتوسط/الانحراف المعياري أو الحد الأدنى/الحد الأقصى)
• device_processor: نقل المصفوفات إلى وحدة المعالجة المركزية/وحدة معالجة الرسوميات مع إمكانية تحويل نوع البيانات
• to_batch_processor: إضافة أبعاد الدفعة إلى الانتقالات لضمان توافق النموذج
• rename_observations_processor: إعادة تسمية مفاتيح الملاحظات باستخدام قواميس التحويل
• tokenizer_processor: تحويل أوصاف المهام اللغوية الطبيعية إلى رموز وأقنعة انتباه

الخطوات التالية

• تنفيذ معالج خاص بك - إنشاء خطوات معالج مخصصة
• تصحيح أخطاء خط الأنابيب - استكشاف الأخطاء وإصلاحها وتحسين خطوط الأنابيب
• المعالجات للروبوتات والمشغّلين عن بُعد - أنماط التكامل في العالم الحقيقي

ملخص

تُحلّ مشكلة ترجمة البيانات في الروبوتات من خلال المعالجات التي توفر:

• التحويلات المعيارية: خطوات معالجة قابلة للتركيب وإعادة الاستخدام
• سلامة النوع: خطوط أنابيب عامة مع فحص وقت الترجمة
• تحسين الأداء: عمليات مُسرَّعة بواسطة وحدة معالجة الرسوميات
• التمييز بين الروبوت والسياسة: خطوط أنابيب منفصلة لبنى بيانات مختلفة
• نظام بيئي شامل: أكثر من 30 معالجًا مسجّلًا للمهام الشائعة

الدور المحوري: يتولى RobotProcessorPipeline معالجة بيانات أجهزة الروبوت غير المجمّعة، بينما يتولى PolicyProcessorPipeline معالجة بيانات النموذج المجمّعة. اختر الأداة المناسبة لبنية بياناتك!