Skip to main content

openhands.controller.agent_controller

AgentController Objects​

class AgentController()

__init__​

def __init__(agent: Agent,
             event_stream: EventStream,
             max_iterations: int,
             max_budget_per_task: float | None = None,
             agent_to_llm_config: dict[str, LLMConfig] | None = None,
             agent_configs: dict[str, AgentConfig] | None = None,
             sid: str = 'default',
             confirmation_mode: bool = False,
             initial_state: State | None = None,
             is_delegate: bool = False,
             headless_mode: bool = True,
             status_callback: Callable | None = None)

Initializes a new instance of the AgentController class.

Arguments:

  • agent - The agent instance to control.
  • event_stream - The event stream to publish events to.
  • max_iterations - The maximum number of iterations the agent can run.
  • max_budget_per_task - The maximum budget (in USD) allowed per task, beyond which the agent will stop.
  • agent_to_llm_config - A dictionary mapping agent names to LLM configurations in the case that we delegate to a different agent.
  • agent_configs - A dictionary mapping agent names to agent configurations in the case that we delegate to a different agent.
  • sid - The session ID of the agent.
  • initial_state - The initial state of the controller.
  • is_delegate - Whether this controller is a delegate.
  • headless_mode - Whether the agent is run in headless mode.

close​

async def close()

Closes the agent controller, canceling any ongoing tasks and unsubscribing from the event stream.

Note that it's fairly important that this closes properly, otherwise the state is incomplete.

log​

def log(level: str, message: str, extra: dict | None = None)

Logs a message to the agent controller's logger.

Arguments:

  • message str - The message to log.

start_step_loop​

async def start_step_loop()

The main loop for the agent's step-by-step execution.

on_event​

async def on_event(event: Event)

Callback from the event stream. Notifies the controller of incoming events.

Arguments:

  • event Event - The incoming event to process.

reset_task​

def reset_task()

Resets the agent's task.

set_agent_state_to​

async def set_agent_state_to(new_state: AgentState)

Updates the agent's state and handles side effects. Can emit events to the event stream.

Arguments:

  • new_state AgentState - The new state to set for the agent.

get_agent_state​

def get_agent_state()

Returns the current state of the agent.

Returns:

  • AgentState - The current state of the agent.

start_delegate​

async def start_delegate(action: AgentDelegateAction)

Start a delegate agent to handle a subtask.

OpenHands is a multi-agentic system. A task is a conversation between OpenHands (the whole system) and the user, which might involve one or more inputs from the user. It starts with an initial input (typically a task statement) from the user, and ends with either an AgentFinishAction initiated by the agent, a stop initiated by the user, or an error.

A subtask is a conversation between an agent and the user, or another agent. If a task is conducted by a single agent, then it's also a subtask. Otherwise, a task consists of multiple subtasks, each executed by one agent.

Arguments:

  • action AgentDelegateAction - The action containing information about the delegate agent to start.

get_state​

def get_state()

Returns the current running state object.

Returns:

  • State - The current state object.

set_initial_state​

def set_initial_state(state: State | None,
                      max_iterations: int,
                      confirmation_mode: bool = False)

Sets the initial state for the agent, either from the previous session, or from a parent agent, or by creating a new one.

Arguments:

  • state - The state to initialize with, or None to create a new state.
  • max_iterations - The maximum number of iterations allowed for the task.
  • confirmation_mode - Whether to enable confirmation mode.