`agents.components.vla`¶

Module Contents¶

Classes¶

VLA

Vision-Language-Agent (VLA) Component.

API¶

class agents.components.vla.VLA(*, inputs: List[agents.ros.Topic], outputs: List[agents.ros.Topic], model_client: agents.clients.lerobot.LeRobotClient, config: agents.config.VLAConfig, component_name: str, **kwargs)¶

Bases: agents.components.model_component.ModelComponent

Vision-Language-Agent (VLA) Component.

custom_on_activate()¶: Custom activation

custom_on_deactivate()¶: Custom deactivation

set_termination_trigger(mode: Literal[timesteps, pynput.keyboard, event] = 'timesteps', max_timesteps: int = 100, stop_key: str = 'q', stop_event: Optional[agents.ros.Event] = None)¶

Set the condition used to determine when an action is done.

Parameters:

mode – One of ‘timesteps’, ‘keyboard’, ‘event’.
max_timesteps – The number of timesteps after which to stop (used if mode=‘timesteps’ or ‘event’).
stop_key – The key to press to stop the action (used if mode=‘keyboard’).

signal_done()¶: Signals that the action is complete. Can be used as an action for signaled events

set_aggregation_function(agg_fn: Callable[[numpy.ndarray, numpy.ndarray], numpy.ndarray])¶

Set the aggregation function to be used for aggregating generated actions from the robot policy model

Parameters:: agg_fn (Callable[[np.ndarray, np.ndarray], np.ndarray]) – A callable that takes two numpy arrays as input and returns a single numpy array.
Raises:: TypeError – If agg_fn is not a callable or does not match the expected signature.

main_action_callback(goal_handle: agents.ros.VisionLanguageAction.Goal)¶

Callback for the VLA main action server

Parameters:: goal_handle (VisionLanguageAction.Goal) – Incoming action goal
Returns:: Action result
Return type:: VisionLanguageAction.Result

property additional_model_clients: Optional[Dict[str, agents.clients.model_base.ModelClient]]¶

Get the dictionary of additional model clients registered to this component.

Returns:: A dictionary mapping client names (str) to ModelClient instances, or None if not set.
Return type:: Optional[Dict[str, ModelClient]]

change_model_client(model_client_name: str) → bool¶

Hot-swap the active model client at runtime.

This method replaces the component’s current model_client with one from the registered additional_model_clients. It handles the safe de-initialization of the old client and initialization of the new one.

This is commonly used as a target for Actions in the Event system.

Parameters:: model_client_name (str) – The key corresponding to the desired client in additional_model_clients.
Returns:: True if the swap was successful, False otherwise (e.g., if the name was not found or initialization failed).
Return type:: bool
Example:

    from agents.ros import Action

    # Define an action to switch to the 'local_backup' client defined previously
    switch_to_local = Action(
        method=brain.change_model_client,
        args=("local_backup",)
    )

    # Trigger this action if the component fails (e.g. internet loss)
    brain.on_component_fail(action=switch_to_local, max_retries=3)

custom_on_configure()¶: Create model client if provided and initialize model.

property warmup: bool¶: Enable warmup of the model.

create_all_subscribers()¶: Override to handle trigger topics and fixed inputs. Called by parent BaseComponent

activate_all_triggers() → None¶: Activates component triggers by attaching execution step to callbacks

destroy_all_subscribers() → None¶: Destroys all node subscribers

agents.components.vla¶

Module Contents¶

Classes¶

API¶

`agents.components.vla`¶