agents.components.vla

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.events.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 additional model clients.

change_model_client(model_client_name: str) → bool#

Change the model client

This method can change the model client that the component is using, at runtime.

It can be invoked as a consequent action in response to an event. For example if one client communicating with a cloud model becomes unresponsive, one can replace it with another client for a locally deployed model.

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

trigger(trigger: Union[agents.ros.Topic, List[agents.ros.Topic], float, agents.ros.Event, None]) → None#

Set component trigger

validate_topics(topics: Sequence[Union[agents.ros.Topic, agents.ros.FixedInput]], allowed_topic_types: Optional[Dict[str, List[Union[Type[agents.ros.SupportedType], List[Type[agents.ros.SupportedType]]]]]] = None, topics_direction: str = 'Topics')#

Verify component specific inputs or outputs using allowed topics if provided

Contents