Spaces:
Sleeping
Sleeping
| from typing import Dict, Any, List, Optional, Literal | |
| from pydantic import BaseModel, Field, field_validator | |
| ActionType = Literal[ | |
| "classify_email", "route_to", "draft_reply", | |
| "escalate", "mark_spam", "request_more_info", "noop" | |
| ] | |
| # ── Strict bounds: scores never touch 0.0 or 1.0 ────────────────── | |
| SCORE_MIN = 0.05 # well above 0.0 — satisfies strict > 0 requirement | |
| SCORE_MAX = 0.95 # well below 1.0 — satisfies strict < 1 requirement | |
| _EPSILON = 1e-6 # extra guard against floating-point edge cases | |
| def clamp_score(v: float) -> float: | |
| """Clamp score to strictly between 0 and 1 (exclusive). | |
| Guaranteed range: [SCORE_MIN, SCORE_MAX] == [0.05, 0.95]. | |
| Values are additionally kept strictly inside (0, 1) via epsilon guard. | |
| """ | |
| try: | |
| val = float(v) | |
| except (TypeError, ValueError): | |
| return 0.5 | |
| clamped = max(SCORE_MIN, min(SCORE_MAX, round(val, 4))) | |
| # Epsilon guard: ensure result is strictly inside (0.0, 1.0) | |
| clamped = max(_EPSILON, min(1.0 - _EPSILON, clamped)) | |
| return float(clamped) | |
| def clamp_breakdown(breakdown: Dict[str, float]) -> Dict[str, float]: | |
| """Clamp every value in a breakdown dict to [SCORE_MIN, SCORE_MAX]. | |
| This prevents any metadata/breakdown field from leaking 0.0 or 1.0. | |
| """ | |
| return {k: clamp_score(v) for k, v in breakdown.items()} | |
| class Observation(BaseModel): | |
| email_id: str | |
| subject: str | |
| sender_type: str | |
| email_body: str | |
| urgency_score: float = Field(default=0.5) | |
| def clamp_urgency(cls, v: Any) -> float: | |
| return clamp_score(v) | |
| sentiment: str | |
| thread_history: List[str] = Field(default_factory=list) | |
| current_status: str | |
| previous_actions: List[str] = Field(default_factory=list) | |
| class Action(BaseModel): | |
| action_type: ActionType | |
| arguments: Dict[str, Any] = Field(default_factory=dict) | |
| class Reward(BaseModel): | |
| score: float = Field(default=0.5) | |
| breakdown: Dict[str, float] = Field(default_factory=dict) | |
| def clamp_score_range(cls, v: Any) -> float: | |
| """Ensure score is always strictly between 0 and 1.""" | |
| try: | |
| v = float(v) | |
| except (TypeError, ValueError): | |
| return 0.5 | |
| return clamp_score(v) | |
| def clamp_breakdown_range(cls, v: Any) -> Dict[str, float]: | |
| """Ensure every breakdown value is strictly between 0 and 1.""" | |
| if isinstance(v, dict): | |
| return clamp_breakdown(v) | |
| return v | |
| class StepResponse(BaseModel): | |
| observation: Observation | |
| reward: Reward | |
| done: bool | |
| info: Dict[str, Any] = Field(default_factory=dict) | |
| class ResetRequest(BaseModel): | |
| task_id: Optional[str] = "task_1" | |