`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, 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 – Each planned step is executed sequentially. Before each step, a brief LLM confirmation call decides: EXECUTE, SKIP, or ABORT, based on the original plan and results so far.

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.
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",
)

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. Runs two-phase planning and execution.

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() → bool¶

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:: True if the switch was successful, False otherwise.
Return type:: bool
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) → bool¶

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:: True if the swap was successful, False otherwise (e.g., if the name was not found or initialization failed).
Return type:: bool
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_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

safe_restart()¶: Stop the component, yield for operations, then restart and wait for ACTIVE state.

agents.components.cortex¶

Module Contents¶

Classes¶

API¶

`agents.components.cortex`¶