from dataclasses import dataclass, field from enum import Enum from typing import Any from openhands.core.schema import ActionType from openhands.events.action.action import Action from openhands.events.event import RecallType @dataclass class ChangeAgentStateAction(Action): """Fake action, just to notify the client that a task state has changed.""" agent_state: str thought: str = '' action: str = ActionType.CHANGE_AGENT_STATE @property def message(self) -> str: return f'Agent state changed to {self.agent_state}' class AgentFinishTaskCompleted(Enum): FALSE = 'false' PARTIAL = 'partial' TRUE = 'true' @dataclass class AgentFinishAction(Action): """An action where the agent finishes the task. Attributes: final_thought (str): The message to send to the user. task_completed (enum): Whether the agent believes the task has been completed. outputs (dict): The other outputs of the agent, for instance "content". thought (str): The agent's explanation of its actions. action (str): The action type, namely ActionType.FINISH. """ final_thought: str = '' task_completed: AgentFinishTaskCompleted | None = None outputs: dict[str, Any] = field(default_factory=dict) thought: str = '' action: str = ActionType.FINISH @property def message(self) -> str: if self.thought != '': return self.thought return "All done! What's next on the agenda?" @dataclass class AgentThinkAction(Action): """An action where the agent logs a thought. Attributes: thought (str): The agent's explanation of its actions. action (str): The action type, namely ActionType.THINK. """ thought: str = '' action: str = ActionType.THINK @property def message(self) -> str: return f'I am thinking...: {self.thought}' @dataclass class AgentRejectAction(Action): outputs: dict = field(default_factory=dict) thought: str = '' action: str = ActionType.REJECT @property def message(self) -> str: msg: str = 'Task is rejected by the agent.' if 'reason' in self.outputs: msg += ' Reason: ' + self.outputs['reason'] return msg @dataclass class AgentDelegateAction(Action): agent: str inputs: dict thought: str = '' action: str = ActionType.DELEGATE @property def message(self) -> str: return f"I'm asking {self.agent} for help with this task." @dataclass class RecallAction(Action): """This action is used for retrieving content, e.g., from the global directory or user workspace.""" recall_type: RecallType query: str = '' thought: str = '' action: str = ActionType.RECALL @property def message(self) -> str: return f'Retrieving content for: {self.query[:50]}' def __str__(self) -> str: ret = '**RecallAction**\n' ret += f'QUERY: {self.query[:50]}' return ret @dataclass class CondensationAction(Action): """This action indicates a condensation of the conversation history is happening. There are two ways to specify the events to be forgotten: 1. By providing a list of event IDs. 2. By providing the start and end IDs of a range of events. In the second case, we assume that event IDs are monotonically increasing, and that _all_ events between the start and end IDs are to be forgotten. Raises: ValueError: If the optional fields are not instantiated in a valid configuration. """ action: str = ActionType.CONDENSATION forgotten_event_ids: list[int] | None = None """The IDs of the events that are being forgotten (removed from the `View` given to the LLM).""" forgotten_events_start_id: int | None = None """The ID of the first event to be forgotten in a range of events.""" forgotten_events_end_id: int | None = None """The ID of the last event to be forgotten in a range of events.""" summary: str | None = None """An optional summary of the events being forgotten.""" summary_offset: int | None = None """An optional offset to the start of the resulting view indicating where the summary should be inserted.""" def _validate_field_polymorphism(self) -> bool: """Check if the optional fields are instantiated in a valid configuration.""" # For the forgotton events, there are only two valid configurations: # 1. We're forgetting events based on the list of provided IDs, or using_event_ids = self.forgotten_event_ids is not None # 2. We're forgetting events based on the range of IDs. using_event_range = ( self.forgotten_events_start_id is not None and self.forgotten_events_end_id is not None ) # Either way, we can only have one of the two valid configurations. forgotten_event_configuration = using_event_ids ^ using_event_range # We also need to check that if the summary is provided, so is the # offset (and vice versa). summary_configuration = ( self.summary is None and self.summary_offset is None ) or (self.summary is not None and self.summary_offset is not None) return forgotten_event_configuration and summary_configuration def __post_init__(self): if not self._validate_field_polymorphism(): raise ValueError('Invalid configuration of the optional fields.') @property def forgotten(self) -> list[int]: """The list of event IDs that should be forgotten.""" # Start by making sure the fields are instantiated in a valid # configuration. We check this whenever the event is initialized, but we # can't make the dataclass immutable so we need to check it again here # to make sure the configuration is still valid. if not self._validate_field_polymorphism(): raise ValueError('Invalid configuration of the optional fields.') if self.forgotten_event_ids is not None: return self.forgotten_event_ids # If we've gotten this far, the start/end IDs are not None. assert self.forgotten_events_start_id is not None assert self.forgotten_events_end_id is not None return list( range(self.forgotten_events_start_id, self.forgotten_events_end_id + 1) ) @property def message(self) -> str: if self.summary: return f'Summary: {self.summary}' return f'Condenser is dropping the events: {self.forgotten}.'