Backup-bdg/OpenHands

1import copy2from typing import Any3 4from openhands.events.event import RecallType5from openhands.events.observation.agent import (6    AgentCondensationObservation,7    AgentStateChangedObservation,8    AgentThinkObservation,9    MicroagentKnowledge,10    RecallObservation,11)12from openhands.events.observation.browse import BrowserOutputObservation13from openhands.events.observation.commands import (14    CmdOutputMetadata,15    CmdOutputObservation,16    IPythonRunCellObservation,17)18from openhands.events.observation.delegate import AgentDelegateObservation19from openhands.events.observation.empty import (20    NullObservation,21)22from openhands.events.observation.error import ErrorObservation23from openhands.events.observation.files import (24    FileEditObservation,25    FileReadObservation,26    FileWriteObservation,27)28from openhands.events.observation.mcp import MCPObservation29from openhands.events.observation.observation import Observation30from openhands.events.observation.reject import UserRejectObservation31from openhands.events.observation.success import SuccessObservation32 33observations = (34    NullObservation,35    CmdOutputObservation,36    IPythonRunCellObservation,37    BrowserOutputObservation,38    FileReadObservation,39    FileWriteObservation,40    FileEditObservation,41    AgentDelegateObservation,42    SuccessObservation,43    ErrorObservation,44    AgentStateChangedObservation,45    UserRejectObservation,46    AgentCondensationObservation,47    AgentThinkObservation,48    RecallObservation,49    MCPObservation,50)51 52OBSERVATION_TYPE_TO_CLASS = {53    observation_class.observation: observation_class  # type: ignore[attr-defined]54    for observation_class in observations55}56 57 58def _update_cmd_output_metadata(59    metadata: dict[str, Any] | CmdOutputMetadata | None, **kwargs: Any60) -> dict[str, Any] | CmdOutputMetadata:61    """Update the metadata of a CmdOutputObservation.62 63    If metadata is None, create a new CmdOutputMetadata instance.64    If metadata is a dict, update the dict.65    If metadata is a CmdOutputMetadata instance, update the instance.66    """67    if metadata is None:68        return CmdOutputMetadata(**kwargs)69 70    if isinstance(metadata, dict):71        metadata.update(**kwargs)72    elif isinstance(metadata, CmdOutputMetadata):73        for key, value in kwargs.items():74            setattr(metadata, key, value)75    return metadata76 77 78def handle_observation_deprecated_extras(extras: dict) -> dict:79    # These are deprecated in https://github.com/All-Hands-AI/OpenHands/pull/488180    if 'exit_code' in extras:81        extras['metadata'] = _update_cmd_output_metadata(82            extras.get('metadata', None), exit_code=extras.pop('exit_code')83        )84    if 'command_id' in extras:85        extras['metadata'] = _update_cmd_output_metadata(86            extras.get('metadata', None), pid=extras.pop('command_id')87        )88 89    # formatted_output_and_error has been deprecated in https://github.com/All-Hands-AI/OpenHands/pull/667190    if 'formatted_output_and_error' in extras:91        extras.pop('formatted_output_and_error')92    return extras93 94 95def observation_from_dict(observation: dict) -> Observation:96    observation = observation.copy()97    if 'observation' not in observation:98        raise KeyError(f"'observation' key is not found in {observation=}")99    observation_class = OBSERVATION_TYPE_TO_CLASS.get(observation['observation'])100    if observation_class is None:101        raise KeyError(102            f"'{observation['observation']=}' is not defined. Available observations: {OBSERVATION_TYPE_TO_CLASS.keys()}"103        )104    observation.pop('observation')105    observation.pop('message', None)106    content = observation.pop('content', '')107    extras = copy.deepcopy(observation.pop('extras', {}))108 109    extras = handle_observation_deprecated_extras(extras)110 111    # convert metadata to CmdOutputMetadata if it is a dict112    if observation_class is CmdOutputObservation:113        if 'metadata' in extras and isinstance(extras['metadata'], dict):114            extras['metadata'] = CmdOutputMetadata(**extras['metadata'])115        elif 'metadata' in extras and isinstance(extras['metadata'], CmdOutputMetadata):116            pass117        else:118            extras['metadata'] = CmdOutputMetadata()119 120    if observation_class is RecallObservation:121        # handle the Enum conversion122        if 'recall_type' in extras:123            extras['recall_type'] = RecallType(extras['recall_type'])124 125        # convert dicts in microagent_knowledge to MicroagentKnowledge objects126        if 'microagent_knowledge' in extras and isinstance(127            extras['microagent_knowledge'], list128        ):129            extras['microagent_knowledge'] = [130                MicroagentKnowledge(**item) if isinstance(item, dict) else item131                for item in extras['microagent_knowledge']132            ]133 134    obs = observation_class(content=content, **extras)135    assert isinstance(obs, Observation)136    return obs137