`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

agents.components.decision#

Module Contents#

Classes#

API#

`agents.components.decision`#