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)
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.
report_error
async def report_error(message: str, exception: Exception | None = None)
Reports an error to the user and sends the exception to the LLM next step, in the hope it can self-correct.
This method should be called for a particular type of errors, which have:
- a user-friendly message, which will be shown in the chat box. This should not be a raw exception message.
- an ErrorObservation that can be sent to the LLM by the agent, with the exception message, so it can self-correct next time.
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.