`agents.components.decision`#

Module Contents#

Classes#

DecisionComponent

This component asks a decision model typed questions about its inputs and publishes the answers. A decision model (served through the TypeSafe-compatible /v1/systemone API) answers each question in one forward pass, without generating text. A yes/no probability, a choice among given options, or a score on ordered levels, with a probability for every option. This makes it a fast “System 1” as compared to autoregressive LLM/VLM models.

API#

class agents.components.decision.DecisionComponent(*, inputs: Optional[List[Union[agents.ros.Topic, agents.ros.FixedInput]]] = None, action_states: Optional[Dict[str, Callable]] = None, questions: Optional[Dict[str, Dict]] = None, model_client: agents.clients.model_base.ModelClient, config: Optional[agents.config.DecisionConfig] = None, trigger: Union[agents.ros.Topic, List[agents.ros.Topic], float, agents.ros.Event] = 1.0, component_name: str, **kwargs)#

Bases: agents.components.model_component.ModelComponent

The questions are given as a dictionary keyed by question id, in the format of the API: each question has a type (noul for yes/no, choice or score), instructions, and criteria (the options of a choice, as a dictionary of option to description, or the levels of a score, as a list). On each trigger, all questions are asked about the current inputs in one request: text inputs make up the state, as a JSON object keyed by topic name, and image inputs are given as images (with a model that reads images). The state can also hold what other components report when asked: each of the action_states is a component action that is called before the questions are asked, and its result joins the state under the given name. When one of them fails the questions are not asked on that trigger.

The component does not need to be given output topics. Each question is published on its own topic, named <component_name>/<question_id>, with a Decision message per answer. Use that topic in an event condition or give it to the UI (the topics are also listed in out_topics). Since each answer field is a scalar, thresholds and confidence cutoffs are set in the event condition.

The component also offers ask as an action that an LLM component or Cortex can call as a tool: it answers one question now, about the current inputs or about a state the caller gives.

Parameters:

inputs (Optional[list[Union[Topic, FixedInput]]]) – The input topics the questions are asked about. Text (String, StreamingString), detections (Detections, Detections3D, DetectionsMultiSource) and images (Image, CompressedImage, RGBD). Without inputs, the component is only used as a tool: another model gives ask the state to decide about, and there are no standing questions.
action_states (Optional[dict[str, Callable]]) – Component actions whose results are part of the state, keyed by the name each result has in the state, e.g. {"task": vla.get_current_task}. They are called on every trigger, in whatever process their components run.
questions (Optional[dict[str, dict]]) – The standing questions to ask on every trigger, keyed by id, in the API format. Each gets an output topic. Needs inputs or action states to be asked about.
model_client (ModelClient) – A model client with a decision model: GenericHTTPClient with a GenericDecisionModel, or OllamaClient with the OllamaModel of a decision model tag.
config (Optional[DecisionConfig]) – The configuration for the component. Defaults to DecisionConfig().
trigger (Union[Topic, list[Topic], float, Event]) – The trigger for asking the questions: input topic(s), a rate in Hz, or an event. Defaults to 1.0 Hz.
component_name (str) – The name of the component, which prefixes its question topics.

Example usage:

speech = Topic(name="speech", msg_type="String")
decider = DecisionComponent(
    inputs=[speech],
    questions={
        "stop": {"type": "noul", "instructions": "Is the person telling the robot to stop?"},
        "addressed": {"type": "noul", "instructions": "Is this speech directed at the robot?"},
    },
    model_client=GenericHTTPClient(GenericDecisionModel(name="lev", checkpoint="lev")),
    trigger=speech,
    component_name="speech_decider",
)

# Answers are published on speech_decider/stop and speech_decider/addressed
stop = Topic(name="speech_decider/stop", msg_type="Decision")
stop_event = Event(stop.msg.noul > 0.8, on_change=True)
launcher.add_pkg(components=[decider, vla], events_actions={stop_event: Action(vla.cancel_main_goal)})

inspect_component() → str#: Return component info including its questions and their topics

ask(instructions: str, type: str = 'noul', criteria: Optional[Union[Dict, List]] = None, state: Optional[Any] = None) → agents.ros.ActionReturnType#

Ask one question now, about the current inputs or a given state.

Parameters:

instructions – The question
type – noul (yes/no), choice or score
criteria – The options of a choice or the levels of a score
state – What to ask about instead of the component’s inputs and action states

Returns:

Whether the answer was obtained, with the answer as JSON or why not

Return type:

ActionReturnType

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)

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

custom_on_deactivate()#: Destroy model client if it exists

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

create_all_service_clients()#: Create a client for each component whose actions this component calls

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

Module Contents#

Classes#

API#

`agents.components.decision`#