File size: 6,164 Bytes
29edd33 c35438f 094509b c35438f 29edd33 c35438f 29edd33 76c31b2 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 76c31b2 c35438f 29edd33 c35438f 29edd33 c35438f 29edd33 c35438f | 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 158 159 160 161 162 163 164 165 166 167 168 | import random
from typing import Dict, Any, Optional
from openenv.models import Observation, Action, Reward, EnvState, StepResult, ResetResult
from openenv.tasks import TASK_REGISTRY
from openenv.reward import compute_reward
# Phase labels for logging and prompting
PHASE_LABELS = {
1: "triage",
2: "route",
3: "resolve",
}
PHASE_INSTRUCTIONS = {
1: (
"PHASE 1 — TRIAGE: Read the ticket carefully and classify it.\n"
"Identify the issue type and include it in the response field.\n"
"Issue types: shipping, billing, technical, returns, safety, cancellation, complaint, orders, sales\n"
"Required action: {\"action_type\": \"classify\", \"response\": \"<issue_type>\"}"
),
2: (
"PHASE 2 — ROUTE: Assign the ticket to the correct specialist team.\n"
"Required action: {\"action_type\": \"assign\", \"team\": \"<team_name>\"}\n"
"Teams: logistics_team | tech_support_team | safety_team | finance_team | orders_team | management_team"
),
3: (
"PHASE 3 — RESOLVE: Take the final resolution action.\n"
"Choose ONE of:\n"
" escalate — angry/manager request/safety emergency (include team + response)\n"
" refund — provably company fault/wrong product (include response)\n"
" respond — customer needs information (write detailed response)"
),
}
class SupportEnv:
def __init__(self, task_name: str = "easy"):
self.task_name = task_name
self.scenarios = TASK_REGISTRY.get(task_name, TASK_REGISTRY["easy"])
self.current_scenario = None
self.step_count = 0
self.phase = 1
self.history = []
self.done = False
self.total_reward = 0.0
self.reset(task_name)
def reset(self, task_name: Optional[str] = None) -> ResetResult:
if task_name:
self.task_name = task_name
self.scenarios = TASK_REGISTRY.get(task_name, TASK_REGISTRY["easy"])
self.current_scenario = random.choice(self.scenarios)
self.step_count = 0
self.phase = 1
self.history = []
self.done = False
self.total_reward = 0.0
obs = self.current_scenario.ticket.copy(deep=True)
obs.history = self.history.copy()
return ResetResult(observation=obs, done=False, info={})
def step(self, action: Action) -> StepResult:
if self.done:
return StepResult(
observation=self.get_current_observation(),
reward=0.0,
done=True,
info={"error": "Episode already done."}
)
self.step_count += 1
gt = self.current_scenario.ground_truth
# Log action to history
act_str = f"[Phase {self.phase}/{PHASE_LABELS[self.phase]}] Agent: {action.action_type}"
if action.team:
act_str += f" → {action.team}"
if action.response:
act_str += f" | \"{action.response[:80]}{'...' if len(action.response) > 80 else ''}\""
self.history.append(act_str)
# Grade this step against its phase
reward = compute_reward(action, gt, self.step_count, phase=self.phase)
self.total_reward += reward.score
# Advance phase and update observation state
if self.phase == 1:
# After triage: reveal the true issue_type so phase 2 routing is informed
self.current_scenario.ticket.issue_type = gt.issue_type
self.history.append(
f"System: Issue classified as '{gt.issue_type}'. Proceed to route the ticket."
)
self.phase = 2
elif self.phase == 2:
# After routing: record which team was assigned; prepare for resolution
assigned = action.team or "(no team)"
self.history.append(
f"System: Ticket routed to '{assigned}'. Proceed to resolve the customer's issue."
)
self.phase = 3
elif self.phase == 3:
# Resolution phase — episode terminates here
if action.action_type == "respond" and reward.score < 0.5:
# Poor response: customer is still unhappy, allow one retry (up to max_steps)
self.history.append(
"Customer: I'm still not satisfied. Could you be more specific?"
)
# Stay in phase 3 for retry
else:
self.done = True
if action.action_type in ("refund", "escalate"):
self.history.append("System: Ticket escalated/refunded and closed.")
else:
self.history.append("Customer: Thank you, that answers my question!")
# Hard cap at 5 steps
if self.step_count >= 5:
self.done = True
obs = self.get_current_observation()
return StepResult(
observation=obs,
reward=reward.score,
done=self.done,
info={"feedback": reward.feedback, "phase": self.phase}
)
def state(self) -> EnvState:
return EnvState(
observation=self.get_current_observation(),
step_count=self.step_count,
done=self.done,
episode_id="local_run",
total_reward=self.total_reward,
task_name=self.task_name,
max_steps=5
)
def get_current_observation(self) -> Observation:
if not self.current_scenario:
return Observation(
ticket_id="None",
issue_type="None",
sentiment="None",
priority="None",
message="No active scenario. Please reset.",
history=self.history.copy()
)
obs = self.current_scenario.ticket.copy(deep=True)
obs.history = self.history.copy()
return obs
def get_phase_instruction(self) -> str:
return PHASE_INSTRUCTIONS.get(self.phase, PHASE_INSTRUCTIONS[3])
def close(self):
pass
@classmethod
def from_docker_image(cls, image_name: str = None):
return cls() |