Spaces:
Sleeping
Sleeping
File size: 5,435 Bytes
7bdbe90 | 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 150 151 152 153 154 155 156 157 | # 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 Meeting Scheduling RL Environment.
Defines the Action, Observation, and State Pydantic models used by the
scheduling environment to coordinate meeting proposals, rescheduling,
and conflict resolution across multiple attendees.
"""
from __future__ import annotations
from typing import Any, Dict, List, Literal, Optional
from pydantic import Field
from openenv.core.env_server.types import Action, Observation, State
class SchedulingAction(Action):
"""Action the agent can take in the scheduling environment."""
action_type: Literal["propose_slot", "reschedule_meeting", "finalize", "reject"] = Field(
default="propose_slot",
description="Type of scheduling action to perform.",
)
proposed_start: Optional[str] = Field(
default=None,
description="ISO8601 datetime string for the proposed meeting start (used with propose_slot).",
)
proposed_duration: Optional[int] = Field(
default=None,
description="Duration in minutes for the proposed meeting (used with propose_slot).",
)
meeting_id_to_move: Optional[str] = Field(
default=None,
description="Identifier of an existing meeting to reschedule (used with reschedule_meeting).",
)
new_start_time: Optional[str] = Field(
default=None,
description="ISO8601 datetime string for the new start time of a rescheduled meeting.",
)
class SchedulingObservation(Observation):
"""Observation returned to the agent after each step."""
requested_duration: int = Field(
default=0,
description="Requested meeting duration in minutes.",
)
requested_priority: int = Field(
default=3,
description="Priority of the meeting request (1=highest, 5=lowest).",
)
attendee_ids: List[str] = Field(
default_factory=list,
description="List of attendee user IDs required for the meeting.",
)
busy_slots: List[Dict[str, Any]] = Field(
default_factory=list,
description="Busy time slots: [{start, end, priority, summary, attendee}].",
)
collective_work_hours: Dict[str, int] = Field(
default_factory=dict,
description="Shared working hours window: {min_start_hour, max_end_hour}.",
)
preference_constraints: Dict[str, Any] = Field(
default_factory=dict,
description="Attendee preference constraints (e.g. preferred times, avoid windows).",
)
current_proposal: Optional[Dict[str, str]] = Field(
default=None,
description="Currently proposed slot: {start, end} as ISO8601 strings.",
)
conflicts: List[Dict[str, Any]] = Field(
default_factory=list,
description="List of conflicts with the current proposal.",
)
preference_penalty: float = Field(
default=0.0,
description="Accumulated penalty from violating attendee preferences.",
)
num_rescheduled: int = Field(
default=0,
description="Number of existing meetings rescheduled so far.",
)
steps_taken: int = Field(
default=0,
description="Number of steps taken in the current episode.",
)
max_steps: int = Field(
default=20,
description="Maximum number of steps allowed in the episode.",
)
success: bool = Field(
default=False,
description="Whether the meeting was successfully scheduled.",
)
error_message: Optional[str] = Field(
default=None,
description="Error message if the last action was invalid.",
)
class SchedulingState(State):
"""Internal environment state tracking the full scheduling episode."""
task_id: str = Field(
default="",
description="Unique identifier for the current task.",
)
scenario_name: str = Field(
default="",
description="Human-readable name of the scheduling scenario.",
)
meeting_request: Dict[str, Any] = Field(
default_factory=dict,
description="The incoming meeting request details.",
)
calendars: Dict[str, List[Any]] = Field(
default_factory=dict,
description="Per-user calendars: {user_id: [[start, end, priority, summary], ...]}.",
)
participant_preferences: Dict[str, Dict[str, Any]] = Field(
default_factory=dict,
description="Per-participant scheduling preferences.",
)
proposed_slot: Optional[List[str]] = Field(
default=None,
description="Currently proposed slot as [start_iso, end_iso].",
)
rescheduled_meetings: List[Dict[str, Any]] = Field(
default_factory=list,
description="List of meetings that have been rescheduled during this episode.",
)
total_preference_penalty: float = Field(
default=0.0,
description="Cumulative penalty from preference violations.",
)
total_steps: int = Field(
default=0,
description="Total steps taken so far in the episode.",
)
final_reward: float = Field(
default=0.0,
description="Final computed reward for the episode.",
)
completed: bool = Field(
default=False,
description="Whether the episode has ended.",
)
|