File size: 5,057 Bytes
246d201 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 |
from openhands.events.observation import (
CmdOutputMetadata,
CmdOutputObservation,
Observation,
)
from openhands.events.serialization import (
event_from_dict,
event_to_dict,
event_to_memory,
event_to_trajectory,
)
def serialization_deserialization(
original_observation_dict, cls, max_message_chars: int = 10000
):
observation_instance = event_from_dict(original_observation_dict)
assert isinstance(
observation_instance, Observation
), 'The observation instance should be an instance of Action.'
assert isinstance(
observation_instance, cls
), 'The observation instance should be an instance of CmdOutputObservation.'
serialized_observation_dict = event_to_dict(observation_instance)
serialized_observation_trajectory = event_to_trajectory(observation_instance)
serialized_observation_memory = event_to_memory(
observation_instance, max_message_chars
)
assert (
serialized_observation_dict == original_observation_dict
), 'The serialized observation should match the original observation dict.'
assert (
serialized_observation_trajectory == original_observation_dict
), 'The serialized observation trajectory should match the original observation dict.'
original_observation_dict.pop('message', None)
original_observation_dict.pop('id', None)
original_observation_dict.pop('timestamp', None)
assert (
serialized_observation_memory == original_observation_dict
), 'The serialized observation memory should match the original observation dict.'
# Additional tests for various observation subclasses can be included here
def test_observation_event_props_serialization_deserialization():
original_observation_dict = {
'id': 42,
'source': 'agent',
'timestamp': '2021-08-01T12:00:00',
'observation': 'run',
'message': 'Command `ls -l` executed with exit code 0.',
'extras': {
'command': 'ls -l',
'hidden': False,
'metadata': {
'exit_code': 0,
'hostname': None,
'pid': -1,
'prefix': '',
'py_interpreter_path': None,
'suffix': '',
'username': None,
'working_dir': None,
},
},
'content': 'foo.txt',
'success': True,
}
serialization_deserialization(original_observation_dict, CmdOutputObservation)
def test_command_output_observation_serialization_deserialization():
original_observation_dict = {
'observation': 'run',
'extras': {
'command': 'ls -l',
'hidden': False,
'metadata': {
'exit_code': 0,
'hostname': None,
'pid': -1,
'prefix': '',
'py_interpreter_path': None,
'suffix': '',
'username': None,
'working_dir': None,
},
},
'message': 'Command `ls -l` executed with exit code 0.',
'content': 'foo.txt',
'success': True,
}
serialization_deserialization(original_observation_dict, CmdOutputObservation)
def test_success_field_serialization():
# Test success=True
obs = CmdOutputObservation(
content='Command succeeded',
command='ls -l',
metadata=CmdOutputMetadata(
exit_code=0,
),
)
serialized = event_to_dict(obs)
assert serialized['success'] is True
# Test success=False
obs = CmdOutputObservation(
content='No such file or directory',
command='ls -l',
metadata=CmdOutputMetadata(
exit_code=1,
),
)
serialized = event_to_dict(obs)
assert serialized['success'] is False
def test_legacy_serialization():
original_observation_dict = {
'id': 42,
'source': 'agent',
'timestamp': '2021-08-01T12:00:00',
'observation': 'run',
'message': 'Command `ls -l` executed with exit code 0.',
'extras': {
'command': 'ls -l',
'hidden': False,
'exit_code': 0,
'command_id': 3,
},
'content': 'foo.txt',
'success': True,
}
event = event_from_dict(original_observation_dict)
assert isinstance(event, Observation)
assert isinstance(event, CmdOutputObservation)
assert event.metadata.exit_code == 0
assert event.success is True
assert event.command == 'ls -l'
assert event.hidden is False
event_dict = event_to_dict(event)
assert event_dict['success'] is True
assert event_dict['extras']['metadata']['exit_code'] == 0
assert event_dict['extras']['metadata']['pid'] == 3
assert event_dict['extras']['command'] == 'ls -l'
assert event_dict['extras']['hidden'] is False
|