Spaces:
Build error
Build error
File size: 4,406 Bytes
0387a1c | 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 | from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Dict, List, Literal, Optional
from pydantic import BaseModel, Field
from openenv.core.env_server.types import Action as OpenEnvAction
from openenv.core.env_server.types import Observation as OpenEnvObservation
from openenv.core.env_server.types import State as OpenEnvState
from app.schemas import EmailMessage, Intent, Tone
ActionType = Literal["classify", "prioritize", "reply", "send", "skip"]
EmailAssistantActionType = Literal[
"categorize",
"respond",
"schedule",
"create_folder",
"move_email",
"tag_email",
"archive",
"delete",
]
class Action(OpenEnvAction):
type: ActionType
payload: Dict[str, Any] = Field(default_factory=dict)
class InboxSummaryItem(BaseModel):
message_id: str
from_email: str
subject: str
deadline_minutes: Optional[int] = Field(default=None, description="Minutes remaining until SLA deadline, if any")
urgency_score: float = Field(default=0.0, ge=0.0, le=1.0)
predicted_intent: Optional[Intent] = None
handled: bool = False
class ActionRecord(BaseModel):
step: int
action: Action
ok: bool = True
notes: str = ""
tool_output: Dict[str, Any] = Field(default_factory=dict)
class Observation(OpenEnvObservation):
current_email: EmailMessage
inbox_summary: List[InboxSummaryItem]
previous_actions: List[ActionRecord]
step_count: int
time_elapsed_minutes: float = Field(
default=0.0,
ge=0.0,
description="Simulated minutes elapsed since reset; used for SLA/deadline rewards.",
)
class RewardComponents(BaseModel):
classification_score: float = 0.0
priority_score: float = 0.0
reply_score: float = 0.0
penalties: Dict[str, float] = Field(default_factory=dict)
class Reward(BaseModel):
value: float
components: RewardComponents
class StepInfo(BaseModel):
task_id: str
done_reason: Optional[str] = None
episode_score: Optional[float] = None
scores: Dict[str, float] = Field(default_factory=dict)
state: Dict[str, Any] = Field(default_factory=dict)
class StepResult(BaseModel):
observation: Observation
reward: Reward
done: bool
info: StepInfo
class ReplyDraft(BaseModel):
reply_subject: str
reply_body: str
tone: Tone = "neutral"
class EmailEnvState(OpenEnvState):
task: Dict[str, Any]
time_elapsed_minutes: float = 0.0
current_message_id: str = ""
selected_priority_message_id: Optional[str] = None
escalated: bool = False
escalation_reason: str = ""
priority_analysis: Dict[str, Any] = Field(default_factory=dict)
handled: Dict[str, bool] = Field(default_factory=dict)
handled_at_minutes: Dict[str, float] = Field(default_factory=dict)
predicted_intent: Dict[str, Intent] = Field(default_factory=dict)
drafts: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
sent: Dict[str, bool] = Field(default_factory=dict)
sent_at_minutes: Dict[str, float] = Field(default_factory=dict)
history: List[Dict[str, Any]] = Field(default_factory=list)
@dataclass
class EmailAssistantState:
"""
Internal state tracker for the email assistant environment.
This is intentionally a dataclass for lightweight in-memory state updates.
"""
current_email_index: int = 0
processed_emails: List[str] = field(default_factory=list)
folders_created: List[str] = field(default_factory=list)
emails_organized: Dict[str, str] = field(default_factory=dict)
drafted_replies: Dict[str, str] = field(default_factory=dict)
sent_emails: List[str] = field(default_factory=list)
skipped_emails: List[str] = field(default_factory=list)
escalated_emails: List[str] = field(default_factory=list)
tags_applied: Dict[str, List[str]] = field(default_factory=dict)
notes: Dict[str, str] = field(default_factory=dict)
@dataclass
class EmailAssistantAction:
"""
Agent-facing action schema for interacting with the email environment.
"""
action_type: EmailAssistantActionType
email_id: int | None = None
target_folder: str | None = None
response_text: str | None = None
schedule_time: str | None = None
category: str | None = None
tags: List[str] = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
|