`agents.components.cortex`#

Module Contents#

Classes#

Cortex

The Cortex component is an LLM-powered task planner and executor that also serves as the system monitor.

API#

class agents.components.cortex.Cortex(*, actions: Optional[List[agents.ros.Action]] = None, output: Optional[agents.ros.Topic] = None, model_client: Optional[agents.clients.model_base.ModelClient] = None, db_client: Optional[agents.clients.db_base.DBClient] = None, routines: Optional[List[agents.ros.Routine]] = None, config: Optional[agents.config.CortexConfig] = None, component_name: str, **kwargs)#

Bases: agents.components.model_component.ModelComponent, agents.ros.Monitor

The Cortex component is an LLM-powered task planner and executor that also serves as the system monitor.

Named after the cerebral cortex, the brain region responsible for higher-order planning, reasoning, and action sequencing, this component takes a high-level task, uses an LLM to decompose it into sub-tasks, and executes them by dispatching Actions registered on other components.

Task execution follows a two-phase approach:

Planning – A single LLM call with all available actions as tools produces a step-by-step plan (returned as multiple tool_calls). Optional RAG context from a vector DB is injected during this phase.
Execution – Consecutive steps whose arguments are all known run as one routine. Before every other step, a brief LLM confirmation call decides to execute it, skip it, abort the plan, or wait for an action that is still running, based on the original plan and results so far. To execute a step that depends on an earlier one, it returns the step’s tool call with the arguments filled in.

The component runs as a ROS2 action server, receiving task goals and providing feedback during execution.

Parameters:

actions (list[Action]) – The action palette – a list of Action objects with descriptions, representing the actions available to the planner.
output (Topic) – Output topic for publishing results for tasks where an action is not required or a plan is not generated.
model_client (Optional[ModelClient]) – The model client for LLM inference. Optional if enable_local_model is set to True in the config.
db_client (Optional[DBClient]) – Optional database client for RAG context during planning.
routines (Optional[List[Routine]]) – Routines to host and offer to the planner as skills, for a recipe where no event triggers them and no UI is enabled. Each needs a description that the planner can read. Routines triggered by events or given to the UI are given to cortex and don’t have to be listed here.
config (Optional[CortexConfig]) – Configuration for the Cortex component.
component_name (str) – The name of this component.

Example usage:

from agents.components import Cortex
from agents.config import CortexConfig
from agents.ros import Action, Topic, Launcher

cortex = Cortex(
    actions=[
        Action(method=nav.go_to, description="Navigate to a location"),
        Action(method=arm.grasp, description="Grasp an object"),
    ],
    model_client=my_client,
    config=CortexConfig(max_planning_steps=10, max_execution_steps=15),
    component_name="cortex",
)

set_robot_description(plugin: Any) → None#

Augment the planning prompt with the attached robot’s identity.

When a robot plugin is attached, Cortex is embodied in a specific physical robot. This builds a compact identity addendum from the plugin’s metadata and prepends it to the planning prompt, so the agent can answer “who/what are you” questions about its own body instead of hallucinating a generic robot.

Parameters:: plugin – A robot.RobotPlugin instance, or None.

set_sensor_descriptions(plugins: List[Any]) → None#

Augment the planning prompt with the sensors attached to the robot.

Each attached sensor is listed under the id its tools are namespaced by, with what its plugin says about it, so the agent knows what each sensor is and can tell two of the same kind apart.

Parameters:: plugins – The attached robot.SensorPlugin instances.

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

custom_on_activate()#: Custom configuration for creating triggers.

custom_on_deactivate()#: Destroy model client if it exists

add_documents(ids: List[str], metadatas: List[Dict], documents: List[str]) → None#: Add documents to vector DB for RAG context during planning.

main_action_callback(goal_handle)#

Action server callback. Iterative plan-execute loop.

Plans a batch of steps, executes them, then feeds the results back into the planning conversation so the LLM can continue with additional steps if needed. The loop ends when the planner returns no more tool calls or an abort/failure occurs.

Parameters:: goal_handle – Incoming action goal
Returns:: Action 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]]

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.

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

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

Module Contents#

Classes#

API#

`agents.components.cortex`#