`agents.components.mllm`#

Module Contents#

Classes#

MLLM

This component utilizes multi-modal large language models (e.g. Llava) that can be used to process text and image data.

API#

class agents.components.mllm.MLLM(*, inputs: List[Union[agents.ros.Topic, agents.ros.FixedInput]], outputs: List[agents.ros.Topic], model_client: Optional[agents.clients.model_base.ModelClient] = None, config: Optional[agents.config.MLLMConfig] = None, db_client: Optional[agents.clients.db_base.DBClient] = None, trigger: Union[agents.ros.Topic, List[agents.ros.Topic], float, agents.ros.Event] = 1.0, depth: Optional[agents.ros.Topic] = None, camera_info: Optional[agents.ros.Topic] = None, component_name: str, **kwargs)#

Bases: agents.components.depth_lift.DepthLiftMixin, agents.components.llm.LLM

This component utilizes multi-modal large language models (e.g. Llava) that can be used to process text and image data.

Parameters:

inputs (list[Topic | FixedInput]) – The input topics or fixed inputs for the MLLM component. This should be a list of Topic objects or FixedInput instances, limited to String and Image types.
outputs (list[Topic]) – The output topics for the MLLM component. This should be a list of Topic objects. String, Detections2D and PointsOfInterest2D types is handled automatically. With the “grounding” or “affordance” task, a Detections3D output additionally lifts the grounded boxes into metric space (see the depth and camera_info parameters), labeled with the query that grounded them. Two component actions serve callers such as the Cortex planner: describe answers a question about a camera frame with text, and run_task runs the configured task once on a frame and publishes the result on the output topics above (available only when task is set in the config).
model_client (Optional[ModelClient]) – The model client for the MLLM component. This should be an instance of ModelClient. Optional if enable_local_model is set to True in the config.
config (MLLMConfig) – Optional configuration for the MLLM component. This should be an instance of MLLMConfig. If not provided, defaults to MLLMConfig().
trigger (Union[Topic, list[Topic], float]) – The trigger value or topic for the MLLM component. This can be a single Topic object, a list of Topic objects, or a float value for a timed component. Defaults to 1.
depth (Optional[Topic]) – Depth topic for the camera the VLM grounds on, either a depth Image registered to its pictures or a PointCloud2 from it or from another sensor whose frame TF relates to the camera’s. Used together with camera_info to lift grounded boxes into metric 3D boxes when a Detections3D output is given (requires the “grounding” or “affordance” task). The depth is latched when the picture is captured for inference, so the boxes measure the scene the VLM actually saw.
camera_info (Optional[Topic]) – Camera intrinsics topic of the pictures the VLM grounds on, which registered depth shares and a point cloud is projected with. Required alongside depth for a Detections3D output.
component_name (str) – The name of the MLLM component. This should be a string and defaults to “mllm_component”.

Example usage:

text0 = Topic(name="text0", msg_type="String")
image0 = Topic(name="image0", msg_type="Image")
text0 = Topic(name="text1", msg_type="String")
config = MLLMConfig()
model = TransformersMLLM(name='idefics')
model_client = ModelClient(model=model)
mllm_component = MLLM(inputs=[text0, image0],
                      outputs=[text1],
                      model_client=model_client,
                      config=config,
                      component_name='mllm_component')

Example usage with local model:

text0 = Topic(name="text0", msg_type="String")
image0 = Topic(name="image0", msg_type="Image")
text1 = Topic(name="text1", msg_type="String")
config = MLLMConfig(enable_local_model=True)
mllm_component = MLLM(inputs=[text0, image0],
                      outputs=[text1],
                      config=config,
                      component_name='local_vlm')

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

set_task(task: Literal[general, pointing, affordance, trajectory, grounding]) → None#

Set a task for the MLLM component. This is useful when using a multimodal LLM model that has been trained on specific tasks. This method can be invoked as an action, in response to an event, to change the task at runtime. For an example checkout RoboBrain2.0, available on RoboML.

Parameters:: task – A task that is one of the following “general”, “pointing”, “affordance”, “trajectory”, “grounding”.

describe(topic_name: str, query: str = 'Describe what you see in the image.', timeout: float = 0.5) → agents.ros.ActionReturnType#

Capture a frame from an image topic and describe it.

Grabs the latest frame from the specified image input topic, runs VLM inference with the given query, and publishes the text result to the component’s String output topics.

Parameters:

topic_name (str) – Name of the image input topic to capture from.
query (str) – Question or instruction about the image.
timeout (float) – Seconds to wait for a frame. Defaults to 0.5.

Returns:

Whether a description was produced, with the description or why not

Return type:

ActionReturnType

run_task(query: str, topic_name: Optional[str] = None, timeout: float = 0.5) → agents.ros.ActionReturnType#

Run the configured task once and publish the result.

The on-demand counterpart of the streaming task mode: one frame, one inference with the caller’s query, published on every output topic the task’s result type matches, exactly as the regular execution path does. Only available when task is configured.

Parameters:

query – What to find or do in the image; it names the results
topic_name – Image input topic to capture from. Defaults to the lift camera, else the first image input
timeout – Seconds to wait for a frame. Defaults to 0.5

Returns:

Whether the task ran, with a JSON summary: task, query, the topics published on, the number of results, and the located objects (label, metric center, size, frame) when the 3D lift ran. Fails without a configured task, without an output of the task’s result type to publish on, and when no frame or no model output is available

custom_on_deactivate()#: Destroy model client if it exists

create_all_service_clients()#: Override create all clients in the LLM

add_documents(ids: List[str], metadatas: List[Dict], documents: List[str]) → None#

Add documents to vector DB for Retrieval Augmented Generation (RAG).

Important

Documents can be provided after parsing them using a document parser. Checkout various document parsers, available in packages like langchain_community

Parameters:

ids (list[str]) – List of unique string ids for each document
metadatas (list[dict]) – List of metadata dicts for each document
documents (list[str]) – List of documents which are to be store in the vector DB

Return type:

None

reset_history() → agents.ros.ActionReturnType#

Forget the conversation, keeping the system prompt. The same as the reset phrase, for events, recipes and the planner.

Returns:: Whether the history was reset, and a message
Return type:: ActionReturnType

get_history() → agents.ros.ActionReturnType#

The conversation as kept, as text.

Returns:: Whether there is a conversation, and its text
Return type:: ActionReturnType

set_topic_prompt(input_topic: agents.ros.Topic, template: Union[str, pathlib.Path]) → None#

Set prompt template on any input topic of type string.

Parameters:

input_topic (Topic) – Name of the input topic on which the prompt template is to be applied
template (Union[str, Path]) – Template in the form of a valid jinja2 string or a path to a file containing the jinja2 string.

Return type:

None

Example usage:

llm_component = LLM(inputs=[text0],
                    outputs=[text1],
                    model_client=model_client,
                    config=config,
                    component_name='llama_component')
llm_component.set_topic_prompt(text0, template="Please answer the following: {{ text0 }}")

set_component_prompt(template: Union[str, pathlib.Path]) → None#

Set component level prompt template which can use multiple input topics.

Parameters:: template (Union[str, Path]) – Template in the form of a valid jinja2 string or a path to a file containing the jinja2 string.
Return type:: None

Example usage:

llm_component = LLM(inputs=[text0],
                    outputs=[text1],
                    model_client=model_client,
                    config=config,
                    component_name='llama_component')
llm_component.set_component_prompt(template="You can see the following items: {{ detections }}. Please answer the following: {{ text0 }}")

set_system_prompt(prompt: str) → None#

Set system prompt for the model, which defines the models ‘personality’.

Parameters:: prompt – string or a path to a file containing the string.
Return type:: None

Example usage:

llm_component = LLM(inputs=[text0],
                    outputs=[text1],
                    model_client=model_client,
                    config=config,
                    component_name='llama_component')
llm_component.set_system_prompt(prompt="You are an amazing and funny robot. You answer all questions with short and concise answers.")

register_tool(tool: Callable, tool_description: Dict, send_tool_response_to_model: bool = False) → None#

Register a tool with the component which can be called by the model. If the send_tool_response_to_model flag is set to True than the output of the tool is sent back to the model and final output of the model is sent to component publishers (i.e. the model “uses” the tool to give a more accurate response.). If the flag is set to False than the output of the tool is sent to publishers of the component.

Parameters:

tool (Callable) – An arbitrary function that needs to be called. The model response will describe a call to this function.
tool_description (dict) – A dictionary describing the function. This dictionary needs to be made in the format shown here. Also see usage example.
send_tool_response_to_model – Whether the model should be called with the tool response. If set to false the tool response will be sent to component publishers. If set to true, the response will be sent back to the model and the final response from the model will be sent to the publishers. Default is False.
send_tool_response_to_model – bool

Return type:

None

Example usage:

def my_arbitrary_function(first_param: str, second_param: int) -> str:
    return f"{first_param}, {second_param}"

my_func_description = {
          'type': 'function',
          'function': {
            'name': 'my_arbitrary_function',
            'description': 'Description of my arbitrary function',
            'parameters': {
              'type': 'object',
              'properties': {
                'first_param': {
                  'type': 'string',
                  'description': 'Description of the first param',
                },
                'second_param': {
                  'type': 'int',
                  'description': 'Description of the second param',
                },
              },
              'required': ['first_param', 'second_param'],
            },
          },
        }

my_component.register_tool(tool=my_arbitrary_function, tool_description=my_func_description, send_tool_response_to_model=False)

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]]

fallback_to_local() → agents.ros.ActionReturnType#

Switch from remote model_client to the built-in local model at runtime.

The local model is deployed on first call (lazy initialization) to avoid consuming GPU memory until actually needed. If enable_local_model is not already set in config, it is enabled automatically.

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

Returns:: Whether the switch happened, with a confirmation message or why the local model could not be deployed.
Return type:: ActionReturnType
Example:

    from agents.ros import Action

    # Define an action to switch to the 'local model' available in each component
    switch_to_local = Action(
        method=brain.fallback_to_local,
    )

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

change_model_client(model_client_name: str) → agents.ros.ActionReturnType#

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:: Whether the swap happened, with a confirmation message or why not: no additional clients are registered, the requested client name is not found, or its initialization failed.
Return type:: ActionReturnType
Example:

    from agents.ros import Action

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

    # Trigger this action if the component fails (e.g. server down)
    brain.on_component_fail(action=switch_to_backup, max_retries=3)

inspect_component() → str#: Return component info including additional model clients.

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

destroy_all_service_clients()#: Destroy the clients for other components’ actions with the rest

custom_on_activate()#: Custom configuration for creating triggers.

init_variables()#

Partition trigger inputs out of the callbacks dict, at activation.

This runs AFTER a robot plugin adapted the callbacks, so the objects moved into trig_callbacks are the adapted ones. Idempotent across reactivations.

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.mllm#

Module Contents#

Classes#

API#

`agents.components.mllm`#