File size: 3,329 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 |
from dataclasses import dataclass
from datetime import datetime
from enum import Enum
from openhands.events.tool import ToolCallMetadata
from openhands.llm.metrics import Metrics
class EventSource(str, Enum):
AGENT = 'agent'
USER = 'user'
ENVIRONMENT = 'environment'
class FileEditSource(str, Enum):
LLM_BASED_EDIT = 'llm_based_edit'
OH_ACI = 'oh_aci' # openhands-aci
class FileReadSource(str, Enum):
OH_ACI = 'oh_aci' # openhands-aci
DEFAULT = 'default'
@dataclass
class Event:
INVALID_ID = -1
@property
def message(self) -> str | None:
if hasattr(self, '_message'):
return self._message # type: ignore[attr-defined]
return ''
@property
def id(self) -> int:
if hasattr(self, '_id'):
return self._id # type: ignore[attr-defined]
return Event.INVALID_ID
@property
def timestamp(self):
if hasattr(self, '_timestamp') and isinstance(self._timestamp, str):
return self._timestamp
@timestamp.setter
def timestamp(self, value: datetime) -> None:
if isinstance(value, datetime):
self._timestamp = value.isoformat()
@property
def source(self) -> EventSource | None:
if hasattr(self, '_source'):
return self._source # type: ignore[attr-defined]
return None
@property
def cause(self) -> int | None:
if hasattr(self, '_cause'):
return self._cause # type: ignore[attr-defined]
return None
@property
def timeout(self) -> int | None:
if hasattr(self, '_timeout'):
return self._timeout # type: ignore[attr-defined]
return None
def set_hard_timeout(self, value: int | None, blocking: bool = True) -> None:
"""Set the timeout for the event.
NOTE, this is a hard timeout, meaning that the event will be blocked
until the timeout is reached.
"""
self._timeout = value
if value is not None and value > 600:
from openhands.core.logger import openhands_logger as logger
logger.warning(
'Timeout greater than 600 seconds may not be supported by '
'the runtime. Consider setting a lower timeout.'
)
# Check if .blocking is an attribute of the event
if hasattr(self, 'blocking'):
# .blocking needs to be set to True if .timeout is set
self.blocking = blocking
# optional metadata, LLM call cost of the edit
@property
def llm_metrics(self) -> Metrics | None:
if hasattr(self, '_llm_metrics'):
return self._llm_metrics # type: ignore[attr-defined]
return None
@llm_metrics.setter
def llm_metrics(self, value: Metrics) -> None:
self._llm_metrics = value
# optional field
@property
def tool_call_metadata(self) -> ToolCallMetadata | None:
if hasattr(self, '_tool_call_metadata'):
return self._tool_call_metadata # type: ignore[attr-defined]
return None
@tool_call_metadata.setter
def tool_call_metadata(self, value: ToolCallMetadata) -> None:
self._tool_call_metadata = value
|