Aller directement au contenu principal

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,
             replay_events: list[Event] | 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.
  • confirmation_mode - Whether to enable confirmation mode for agent actions.
  • initial_state - The initial state of the controller.
  • is_delegate - Whether this controller is a delegate.
  • event_stream0 - Whether the agent is run in headless mode.
  • event_stream1 - Optional callback function to handle status updates.
  • event_stream2 - A list of logs to replay.

close​

async def close() -> None

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) -> None

Logs a message to the agent controller's logger.

Arguments:

  • level str - The logging level to use (e.g., 'info', 'debug', 'error').
  • message str - The message to log.
  • extra dict | None, optional - Additional fields to include in the log. Defaults to None.

should_step​

def should_step(event: Event) -> bool

Whether the agent should take a step based on an event. In general, the agent should take a step if it receives a message from the user, or observes something in the environment (after acting).

on_event​

def on_event(event: Event) -> None

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

Arguments:

  • event Event - The incoming event to process.

set_agent_state_to​

async def set_agent_state_to(new_state: AgentState) -> None

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() -> AgentState

Returns the current state of the agent.

Returns:

  • AgentState - The current state of the agent.

start_delegate​

async def start_delegate(action: AgentDelegateAction) -> None

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.

end_delegate​

def end_delegate() -> None

Ends the currently active delegate (e.g., if it is finished or errored) so that this controller can resume normal operation.

get_state​

def get_state() -> 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) -> None

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.