Spaces:
Sleeping
Sleeping
File size: 3,220 Bytes
fe0c391 4b98131 fe0c391 4b98131 fe0c391 4b98131 fe0c391 | 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 | # Copyright (c) Meta Platforms, Inc. and affiliates.
# All rights reserved.
#
# This source code is licensed under the BSD-style license found in the
# LICENSE file in the root directory of this source tree.
"""
Data models for the Dispatch911 Environment.
Action: the agent's structured dispatch decision per incoming 911 call.
Observation: the text-based observation the agent receives each step.
"""
from typing import Dict, List, Literal, Optional
from openenv.core.env_server.types import Action, Observation
from pydantic import Field
# Reroute sub-action
class RerouteAction(Action):
"""Optional reroute block inside a dispatch action."""
vehicle_to_reroute: str = Field(..., description="Unit ID of vehicle to redirect")
from_event_id: str = Field(..., description="EVT-NNNN the vehicle is pulled from")
replacement_vehicle_id: Optional[str] = Field(
None, description="Free unit to cover the abandoned event"
)
# Agent action
class SmartEmergencyAction(Action):
"""
The agent's response to an incoming 911 call.
Three modes:
- action_type='dispatch': handle a new emergency
- action_type='duplicate': flag as repeat of an existing event
- action_type='hold': queue event for a busy vehicle to handle after it frees
"""
action_type: Literal["dispatch", "duplicate", "hold"] = Field(
..., description="'dispatch', 'duplicate', or 'hold'"
)
severity_pred: int = Field(
..., ge=1, le=5, description="Predicted severity 1-5"
)
is_duplicate: bool = Field(
False, description="Whether the agent believes this is a repeat call"
)
duplicate_of_event_id: Optional[str] = Field(
None, description="EVT-NNNN of the event this duplicates (required if is_duplicate)"
)
vehicle_type: Optional[str] = Field(
None, description="'police', 'ambulance', or 'fire' (required if dispatch or hold)"
)
vehicle_id: Optional[str] = Field(
None, description="Unit to dispatch now (dispatch) or busy unit to queue for (hold)"
)
reroute: Optional[RerouteAction] = Field(
None, description="Optional reroute instruction"
)
# Observation
class SmartEmergencyObservation(Observation):
"""
Observation returned to the agent each step.
Contains the full text prompt (transcript + active events + unit status +
city reference + dispatcher notes) and structured metadata for logging.
"""
prompt: str = Field(default="", description="Full text observation for the LLM")
step: int = Field(default=0, description="Current step number")
call_id: str = Field(default="", description="ID of the incoming call")
reward_breakdown: Dict[str, float] = Field(
default_factory=dict, description="Per-component reward breakdown"
)
active_event_ids: List[str] = Field(
default_factory=list, description="Currently active event IDs"
)
fleet_utilisation: float = Field(
default=0.0, description="Fraction of fleet currently busy"
)
ground_truth: Dict = Field(
default_factory=dict,
description="Hidden ground truth for the current call (populated after step)",
)
|