Backup-bdg/OpenHands

1from __future__ import annotations2 3from openhands.core.logger import openhands_logger as logger4from openhands.events.action.action import Action5from openhands.events.action.message import MessageAction6from openhands.events.event import Event, EventSource7from openhands.events.observation.empty import NullObservation8from openhands.events.serialization.event import event_from_dict9 10 11class ReplayManager:12    """ReplayManager manages the lifecycle of a replay session of a given trajectory.13 14    Replay manager keeps track of a list of events, replays actions, and ignore15    messages and observations.16 17    Note that unexpected or even errorneous results could happen if18    1) any action is non-deterministic, OR19    2) if the initial state before the replay session is different from the20    initial state of the trajectory.21    """22 23    def __init__(self, events: list[Event] | None):24        replay_events = []25        for event in events or []:26            if event.source == EventSource.ENVIRONMENT:27                # ignore ENVIRONMENT events as they are not issued by28                # the user or agent, and should not be replayed29                continue30            if isinstance(event, NullObservation):31                # ignore NullObservation32                continue33            replay_events.append(event)34 35        if replay_events:36            logger.info(f'Replay events loaded, events length = {len(replay_events)}')37            for index in range(len(replay_events) - 1):38                event = replay_events[index]39                if isinstance(event, MessageAction) and event.wait_for_response:40                    # For any message waiting for response that is not the last41                    # event, we override wait_for_response to False, as a response42                    # would have been included in the next event, and we don't43                    # want the user to interfere with the replay process44                    logger.info(45                        'Replay events contains wait_for_response message action, ignoring wait_for_response'46                    )47                    event.wait_for_response = False48        self.replay_events = replay_events49        self.replay_mode = bool(replay_events)50        self.replay_index = 051 52    def _replayable(self) -> bool:53        return (54            self.replay_events is not None55            and self.replay_index < len(self.replay_events)56            and isinstance(self.replay_events[self.replay_index], Action)57        )58 59    def should_replay(self) -> bool:60        """61        Whether the controller is in trajectory replay mode, and the replay62        hasn't finished. Note: after the replay is finished, the user and63        the agent could continue to message/act.64 65        This method also moves "replay_index" to the next action, if applicable.66        """67        if not self.replay_mode:68            return False69 70        assert self.replay_events is not None71        while self.replay_index < len(self.replay_events) and not self._replayable():72            self.replay_index += 173 74        return self._replayable()75 76    def step(self) -> Action:77        assert self.replay_events is not None78        event = self.replay_events[self.replay_index]79        assert isinstance(event, Action)80        self.replay_index += 181        return event82 83    @staticmethod84    def get_replay_events(trajectory: list[dict]) -> list[Event]:85        if not isinstance(trajectory, list):86            raise ValueError(87                f'Expected a list in {trajectory}, got {type(trajectory).__name__}'88            )89        replay_events = []90        for item in trajectory:91            event = event_from_dict(item)92            if event.source == EventSource.ENVIRONMENT:93                # ignore ENVIRONMENT events as they are not issued by94                # the user or agent, and should not be replayed95                continue96            # cannot add an event with _id to event stream97            event._id = None  # type: ignore[attr-defined]98            replay_events.append(event)99        return replay_events100