Backup-bdg/OpenHands

1from dataclasses import asdict2from datetime import datetime3from enum import Enum4from typing import Any5 6from pydantic import BaseModel7 8from openhands.events import Event, EventSource9from openhands.events.serialization.action import action_from_dict10from openhands.events.serialization.observation import observation_from_dict11from openhands.events.serialization.utils import remove_fields12from openhands.events.tool import ToolCallMetadata13from openhands.llm.metrics import Cost, Metrics, ResponseLatency, TokenUsage14 15# TODO: move `content` into `extras`16TOP_KEYS = [17    'id',18    'timestamp',19    'source',20    'message',21    'cause',22    'action',23    'observation',24    'tool_call_metadata',25    'llm_metrics',26]27UNDERSCORE_KEYS = [28    'id',29    'timestamp',30    'source',31    'cause',32    'tool_call_metadata',33    'llm_metrics',34]35 36DELETE_FROM_TRAJECTORY_EXTRAS = {37    'dom_object',38    'axtree_object',39    'active_page_index',40    'last_browser_action',41    'last_browser_action_error',42    'focused_element_bid',43    'extra_element_properties',44}45 46DELETE_FROM_TRAJECTORY_EXTRAS_AND_SCREENSHOTS = DELETE_FROM_TRAJECTORY_EXTRAS | {47    'screenshot',48    'set_of_marks',49}50 51 52def event_from_dict(data: dict[str, Any]) -> 'Event':53    evt: Event54    if 'action' in data:55        evt = action_from_dict(data)56    elif 'observation' in data:57        evt = observation_from_dict(data)58    else:59        raise ValueError(f'Unknown event type: {data}')60    for key in UNDERSCORE_KEYS:61        if key in data:62            value = data[key]63            if key == 'timestamp' and isinstance(value, datetime):64                value = value.isoformat()65            if key == 'source':66                value = EventSource(value)67            if key == 'tool_call_metadata':68                value = ToolCallMetadata(**value)69            if key == 'llm_metrics':70                metrics = Metrics()71                if isinstance(value, dict):72                    metrics.accumulated_cost = value.get('accumulated_cost', 0.0)73                    for cost in value.get('costs', []):74                        metrics._costs.append(Cost(**cost))75                    metrics.response_latencies = [76                        ResponseLatency(**latency)77                        for latency in value.get('response_latencies', [])78                    ]79                    metrics.token_usages = [80                        TokenUsage(**usage) for usage in value.get('token_usages', [])81                    ]82                    # Set accumulated token usage if available83                    if 'accumulated_token_usage' in value:84                        metrics._accumulated_token_usage = TokenUsage(85                            **value.get('accumulated_token_usage', {})86                        )87                value = metrics88            setattr(evt, '_' + key, value)89    return evt90 91 92def _convert_pydantic_to_dict(obj: BaseModel | dict) -> dict:93    if isinstance(obj, BaseModel):94        return obj.model_dump()95    return obj96 97 98def event_to_dict(event: 'Event') -> dict:99    props = asdict(event)100    d = {}101    for key in TOP_KEYS:102        if hasattr(event, key) and getattr(event, key) is not None:103            d[key] = getattr(event, key)104        elif hasattr(event, f'_{key}') and getattr(event, f'_{key}') is not None:105            d[key] = getattr(event, f'_{key}')106        if key == 'id' and d.get('id') == -1:107            d.pop('id', None)108        if key == 'timestamp' and 'timestamp' in d:109            if isinstance(d['timestamp'], datetime):110                d['timestamp'] = d['timestamp'].isoformat()111        if key == 'source' and 'source' in d:112            d['source'] = d['source'].value113        if key == 'recall_type' and 'recall_type' in d:114            d['recall_type'] = d['recall_type'].value115        if key == 'tool_call_metadata' and 'tool_call_metadata' in d:116            d['tool_call_metadata'] = d['tool_call_metadata'].model_dump()117        if key == 'llm_metrics' and 'llm_metrics' in d:118            d['llm_metrics'] = d['llm_metrics'].get()119        props.pop(key, None)120    if 'security_risk' in props and props['security_risk'] is None:121        props.pop('security_risk')122    if 'action' in d:123        d['args'] = props124        if event.timeout is not None:125            d['timeout'] = event.timeout126    elif 'observation' in d:127        d['content'] = props.pop('content', '')128 129        # props is a dict whose values can include a complex object like an instance of a BaseModel subclass130        # such as CmdOutputMetadata131        # we serialize it along with the rest132        # we also handle the Enum conversion for RecallObservation133        d['extras'] = {134            k: (v.value if isinstance(v, Enum) else _convert_pydantic_to_dict(v))135            for k, v in props.items()136        }137        # Include success field for CmdOutputObservation138        if hasattr(event, 'success'):139            d['success'] = event.success140    else:141        raise ValueError(f'Event must be either action or observation. has: {event}')142    return d143 144 145def event_to_trajectory(event: 'Event', include_screenshots: bool = False) -> dict:146    d = event_to_dict(event)147    if 'extras' in d:148        remove_fields(149            d['extras'],150            DELETE_FROM_TRAJECTORY_EXTRAS151            if include_screenshots152            else DELETE_FROM_TRAJECTORY_EXTRAS_AND_SCREENSHOTS,153        )154    return d155 156 157def truncate_content(content: str, max_chars: int | None = None) -> str:158    """Truncate the middle of the observation content if it is too long."""159    if max_chars is None or len(content) <= max_chars or max_chars < 0:160        return content161 162    # truncate the middle and include a message to the LLM about it163    half = max_chars // 2164    return (165        content[:half]166        + '\n[... Observation truncated due to length ...]\n'167        + content[-half:]168    )169