Backup-bdg/OpenHands

1from dataclasses import dataclass, field2from enum import Enum3from typing import Any4 5from openhands.core.schema import ActionType6from openhands.events.action.action import Action7from openhands.events.event import RecallType8 9 10@dataclass11class ChangeAgentStateAction(Action):12    """Fake action, just to notify the client that a task state has changed."""13 14    agent_state: str15    thought: str = ''16    action: str = ActionType.CHANGE_AGENT_STATE17 18    @property19    def message(self) -> str:20        return f'Agent state changed to {self.agent_state}'21 22 23class AgentFinishTaskCompleted(Enum):24    FALSE = 'false'25    PARTIAL = 'partial'26    TRUE = 'true'27 28 29@dataclass30class AgentFinishAction(Action):31    """An action where the agent finishes the task.32 33    Attributes:34        final_thought (str): The message to send to the user.35        task_completed (enum): Whether the agent believes the task has been completed.36        outputs (dict): The other outputs of the agent, for instance "content".37        thought (str): The agent's explanation of its actions.38        action (str): The action type, namely ActionType.FINISH.39    """40 41    final_thought: str = ''42    task_completed: AgentFinishTaskCompleted | None = None43    outputs: dict[str, Any] = field(default_factory=dict)44    thought: str = ''45    action: str = ActionType.FINISH46 47    @property48    def message(self) -> str:49        if self.thought != '':50            return self.thought51        return "All done! What's next on the agenda?"52 53 54@dataclass55class AgentThinkAction(Action):56    """An action where the agent logs a thought.57 58    Attributes:59        thought (str): The agent's explanation of its actions.60        action (str): The action type, namely ActionType.THINK.61    """62 63    thought: str = ''64    action: str = ActionType.THINK65 66    @property67    def message(self) -> str:68        return f'I am thinking...: {self.thought}'69 70 71@dataclass72class AgentRejectAction(Action):73    outputs: dict = field(default_factory=dict)74    thought: str = ''75    action: str = ActionType.REJECT76 77    @property78    def message(self) -> str:79        msg: str = 'Task is rejected by the agent.'80        if 'reason' in self.outputs:81            msg += ' Reason: ' + self.outputs['reason']82        return msg83 84 85@dataclass86class AgentDelegateAction(Action):87    agent: str88    inputs: dict89    thought: str = ''90    action: str = ActionType.DELEGATE91 92    @property93    def message(self) -> str:94        return f"I'm asking {self.agent} for help with this task."95 96 97@dataclass98class RecallAction(Action):99    """This action is used for retrieving content, e.g., from the global directory or user workspace."""100 101    recall_type: RecallType102    query: str = ''103    thought: str = ''104    action: str = ActionType.RECALL105 106    @property107    def message(self) -> str:108        return f'Retrieving content for: {self.query[:50]}'109 110    def __str__(self) -> str:111        ret = '**RecallAction**\n'112        ret += f'QUERY: {self.query[:50]}'113        return ret114 115 116@dataclass117class CondensationAction(Action):118    """This action indicates a condensation of the conversation history is happening.119 120    There are two ways to specify the events to be forgotten:121    1. By providing a list of event IDs.122    2. By providing the start and end IDs of a range of events.123 124    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.125 126    Raises:127        ValueError: If the optional fields are not instantiated in a valid configuration.128    """129 130    action: str = ActionType.CONDENSATION131 132    forgotten_event_ids: list[int] | None = None133    """The IDs of the events that are being forgotten (removed from the `View` given to the LLM)."""134 135    forgotten_events_start_id: int | None = None136    """The ID of the first event to be forgotten in a range of events."""137 138    forgotten_events_end_id: int | None = None139    """The ID of the last event to be forgotten in a range of events."""140 141    summary: str | None = None142    """An optional summary of the events being forgotten."""143 144    summary_offset: int | None = None145    """An optional offset to the start of the resulting view indicating where the summary should be inserted."""146 147    def _validate_field_polymorphism(self) -> bool:148        """Check if the optional fields are instantiated in a valid configuration."""149        # For the forgotton events, there are only two valid configurations:150        # 1. We're forgetting events based on the list of provided IDs, or151        using_event_ids = self.forgotten_event_ids is not None152        # 2. We're forgetting events based on the range of IDs.153        using_event_range = (154            self.forgotten_events_start_id is not None155            and self.forgotten_events_end_id is not None156        )157 158        # Either way, we can only have one of the two valid configurations.159        forgotten_event_configuration = using_event_ids ^ using_event_range160 161        # We also need to check that if the summary is provided, so is the162        # offset (and vice versa).163        summary_configuration = (164            self.summary is None and self.summary_offset is None165        ) or (self.summary is not None and self.summary_offset is not None)166 167        return forgotten_event_configuration and summary_configuration168 169    def __post_init__(self):170        if not self._validate_field_polymorphism():171            raise ValueError('Invalid configuration of the optional fields.')172 173    @property174    def forgotten(self) -> list[int]:175        """The list of event IDs that should be forgotten."""176        # Start by making sure the fields are instantiated in a valid177        # configuration. We check this whenever the event is initialized, but we178        # can't make the dataclass immutable so we need to check it again here179        # to make sure the configuration is still valid.180        if not self._validate_field_polymorphism():181            raise ValueError('Invalid configuration of the optional fields.')182 183        if self.forgotten_event_ids is not None:184            return self.forgotten_event_ids185 186        # If we've gotten this far, the start/end IDs are not None.187        assert self.forgotten_events_start_id is not None188        assert self.forgotten_events_end_id is not None189        return list(190            range(self.forgotten_events_start_id, self.forgotten_events_end_id + 1)191        )192 193    @property194    def message(self) -> str:195        if self.summary:196            return f'Summary: {self.summary}'197        return f'Condenser is dropping the events: {self.forgotten}.'198