File size: 4,316 Bytes
29edd33 a23653a 29edd33 fa17f30 29edd33 | 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 | """
Typed Pydantic models for the OpenEnv Customer Support Environment.
Defines Observation, Action, Reward, EnvState, StepResult, TicketScenario,
and GroundTruth — the complete data contract for the environment.
"""
from __future__ import annotations
import enum
from typing import Any, Dict, List, Optional
from openai import OpenAI
from pydantic import BaseModel, Field, field_validator
# ---------------------------------------------------------------------------
# Enums for constrained fields
# ---------------------------------------------------------------------------
class ActionType(str, enum.Enum):
CLASSIFY = "classify"
ASSIGN = "assign"
RESPOND = "respond"
REFUND = "refund"
ESCALATE = "escalate"
class Sentiment(str, enum.Enum):
POSITIVE = "positive"
NEUTRAL = "neutral"
NEGATIVE = "negative"
ANGRY = "angry"
class Priority(str, enum.Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
# ---------------------------------------------------------------------------
# Core OpenEnv Models
# ---------------------------------------------------------------------------
class Observation(BaseModel):
"""What the agent sees at each step."""
ticket_id: str = Field(..., description="Unique ticket identifier")
issue_type: str = Field(..., description="Category of the customer issue")
sentiment: str = Field(..., description="Customer sentiment: positive|neutral|negative|angry")
priority: str = Field(..., description="Ticket priority: low|medium|high|critical")
message: str = Field(..., description="The customer's message text")
history: List[str] = Field(default_factory=list, description="Previous interaction history")
class Action(BaseModel):
"""What the agent does at each step."""
action_type: str = Field(
...,
description="Action to take: classify|assign|respond|refund|escalate",
)
team: Optional[str] = Field(
None,
description="Target team for assignment (required for assign action)",
)
response: Optional[str] = Field(
None,
description="Response message to the customer (required for respond action)",
)
@field_validator("action_type")
@classmethod
def validate_action_type(cls, v: str) -> str:
allowed = {e.value for e in ActionType}
if v not in allowed:
raise ValueError(f"action_type must be one of {allowed}, got '{v}'")
return v
class Reward(BaseModel):
"""Reward signal returned after each step."""
score: float = Field(..., ge=0.002, le=0.998, description="Reward score between 0.002 and 0.998")
feedback: str = Field(..., description="Human-readable explanation of the score")
class EnvState(BaseModel):
"""Full environment state for inspection."""
observation: Observation
step_count: int = Field(default=0, ge=0)
done: bool = False
episode_id: str = ""
total_reward: float = 0.0
task_name: str = ""
max_steps: int = 5
class StepResult(BaseModel):
"""Return value of env.step()."""
observation: Observation
reward: float = 0.0
done: bool = False
info: Dict[str, Any] = Field(default_factory=dict)
class ResetResult(BaseModel):
"""Return value of env.reset()."""
observation: Observation
done: bool = False
info: Dict[str, Any] = Field(default_factory=dict)
# ---------------------------------------------------------------------------
# Task / Ground-Truth Models
# ---------------------------------------------------------------------------
class GroundTruth(BaseModel):
"""Expected correct answer for a ticket scenario."""
issue_type: str = Field(..., description="Correct issue classification")
team: Optional[str] = Field(None, description="Correct team assignment")
action_type: str = Field(..., description="Correct action to take")
response_keywords: List[str] = Field(
default_factory=list,
description="Keywords that should appear in a correct response",
)
requires_refund: bool = False
requires_escalation: bool = False
class TicketScenario(BaseModel):
"""A complete ticket scenario: the ticket itself plus the expected answer."""
ticket: Observation
ground_truth: GroundTruth
|