mirror of
https://github.com/OpenHands/OpenHands.git
synced 2025-12-26 05:48:36 +08:00
55 lines
2.0 KiB
Python
55 lines
2.0 KiB
Python
from openhands.events.observation.agent import AgentStateChangedObservation
|
|
from openhands.events.observation.browse import BrowserOutputObservation
|
|
from openhands.events.observation.commands import (
|
|
CmdOutputObservation,
|
|
IPythonRunCellObservation,
|
|
)
|
|
from openhands.events.observation.delegate import AgentDelegateObservation
|
|
from openhands.events.observation.empty import NullObservation
|
|
from openhands.events.observation.error import ErrorObservation
|
|
from openhands.events.observation.files import (
|
|
FileEditObservation,
|
|
FileReadObservation,
|
|
FileWriteObservation,
|
|
)
|
|
from openhands.events.observation.observation import Observation
|
|
from openhands.events.observation.reject import UserRejectObservation
|
|
from openhands.events.observation.success import SuccessObservation
|
|
|
|
observations = (
|
|
NullObservation,
|
|
CmdOutputObservation,
|
|
IPythonRunCellObservation,
|
|
BrowserOutputObservation,
|
|
FileReadObservation,
|
|
FileWriteObservation,
|
|
FileEditObservation,
|
|
AgentDelegateObservation,
|
|
SuccessObservation,
|
|
ErrorObservation,
|
|
AgentStateChangedObservation,
|
|
UserRejectObservation,
|
|
)
|
|
|
|
OBSERVATION_TYPE_TO_CLASS = {
|
|
observation_class.observation: observation_class # type: ignore[attr-defined]
|
|
for observation_class in observations
|
|
}
|
|
|
|
|
|
def observation_from_dict(observation: dict) -> Observation:
|
|
observation = observation.copy()
|
|
if 'observation' not in observation:
|
|
raise KeyError(f"'observation' key is not found in {observation=}")
|
|
observation_class = OBSERVATION_TYPE_TO_CLASS.get(observation['observation'])
|
|
if observation_class is None:
|
|
raise KeyError(
|
|
f"'{observation['observation']=}' is not defined. Available observations: {OBSERVATION_TYPE_TO_CLASS.keys()}"
|
|
)
|
|
observation.pop('observation')
|
|
observation.pop('message', None)
|
|
content = observation.pop('content', '')
|
|
extras = observation.pop('extras', {})
|
|
|
|
return observation_class(content=content, **extras)
|