Spaces:
Running
Running
File size: 2,964 Bytes
35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd af2ccc5 35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd 250ab26 35ea9cd af2ccc5 35ea9cd 250ab26 35ea9cd af2ccc5 35ea9cd 250ab26 35ea9cd af2ccc5 35ea9cd | 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 | from enum import Enum
from typing import Any, Dict, List, Optional
from pydantic import BaseModel, Field
class TaskType(str, Enum):
TASK1 = "task1"
TASK2 = "task2"
TASK3 = "task3"
class SeverityLevel(str, Enum):
SEV1 = "SEV1"
SEV2 = "SEV2"
SEV3 = "SEV3"
class RootCauseCategory(str, Enum):
DATABASE = "DATABASE"
NETWORK = "NETWORK"
APPLICATION = "APPLICATION"
INFRASTRUCTURE = "INFRASTRUCTURE"
THIRD_PARTY = "THIRD_PARTY"
UNKNOWN = "UNKNOWN"
class RecommendedAction(str, Enum):
ROLLBACK = "ROLLBACK"
SCALE_UP = "SCALE_UP"
RESTART_SERVICE = "RESTART_SERVICE"
FAILOVER = "FAILOVER"
NOTIFY_VENDOR = "NOTIFY_VENDOR"
INVESTIGATE = "INVESTIGATE"
NO_ACTION = "NO_ACTION"
class IncidentObservation(BaseModel):
incident_id: str
task_type: TaskType
difficulty: str
task_description: str
alert_text: str
context: Dict[str, Any]
expected_field: str
allowed_values: List[str] = Field(default_factory=list)
step_count: int = 0
max_steps: int = 1
last_action_summary: Optional[str] = None
last_reward: float = Field(default=0.01, gt=0.0, lt=1.0)
episode_status: str = "awaiting_action"
class IncidentAction(BaseModel):
incident_id: str
task_type: TaskType
severity: Optional[SeverityLevel] = Field(None)
root_cause: Optional[RootCauseCategory] = Field(None)
action: Optional[RecommendedAction] = Field(None)
def populated_fields(self) -> Dict[str, str]:
fields: Dict[str, str] = {}
if self.severity is not None:
fields["severity"] = self.severity.value
if self.root_cause is not None:
fields["root_cause"] = self.root_cause.value
if self.action is not None:
fields["action"] = self.action.value
return fields
def selected_field(self) -> Optional[str]:
populated = self.populated_fields()
if len(populated) != 1:
return None
return next(iter(populated))
def selected_value(self) -> Optional[str]:
selected = self.selected_field()
if selected is None:
return None
return self.populated_fields()[selected]
class IncidentReward(BaseModel):
value: float = Field(..., gt=0.0, lt=1.0)
reason: str
class StepResult(BaseModel):
observation: IncidentObservation
reward: IncidentReward
done: bool
info: Dict[str, Any] = Field(default_factory=dict)
class IncidentState(BaseModel):
episode_id: str
session_id: Optional[str] = None
step_count: int
max_steps: int
total_reward: float = Field(default=0.01, gt=0.0, lt=1.0)
done: bool
incident_id: str
task_type: TaskType
difficulty: str
status: str
last_reward: float = Field(default=0.01, gt=0.0, lt=1.0)
class ResetRequest(BaseModel):
task_type: Optional[TaskType] = None
ticket_id: Optional[str] = None
seed: Optional[int] = None
|