Spaces:
Sleeping
Sleeping
File size: 2,069 Bytes
248269f | 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 | """Task 2 — Conflict Classification (Medium).
The agent observes an infeasible scheduling instance and must identify the
constraint violation type from the closed vocabulary:
resource_overload, deadline_violation, precedence_violation,
availability_conflict, capacity_exceeded
Grading:
1.0 — exact match
0.5 — related category (same constraint family)
0.1 — valid category but wrong family
0.0 — empty or unknown
Max steps per episode: 5.
Expected agent accuracy: ~60%.
"""
from __future__ import annotations
from typing import Any
from environment import INSTANCE_BANK, SchedulingOptEnv
from models import Action
TASK_ID = "conflict_classification"
MAX_STEPS = 5
DIFFICULTY = "medium"
def run_episode(env: SchedulingOptEnv, agent_fn: Any) -> dict[str, Any]:
"""Run a single conflict-classification episode.
Args:
env: An initialized SchedulingOptEnv instance.
agent_fn: Callable receiving an Observation, returning a violation-type string.
Returns:
Episode summary dict.
"""
obs = env.reset(task_id=TASK_ID)
total_reward = 0.0
steps = 0
info: dict[str, Any] = {}
for _ in range(MAX_STEPS):
response = agent_fn(obs)
action = Action(response=response, task_id=TASK_ID)
obs, reward, done, info = env.step(action)
total_reward += reward
steps += 1
if done:
break
return {
"task": TASK_ID,
"difficulty": DIFFICULTY,
"steps": steps,
"total_reward": round(total_reward, 4),
"info": info,
}
def get_infeasible_instances() -> list[dict[str, Any]]:
"""Return only instances that have violations (for classification task)."""
return [
{
"instance": entry["instance"],
"violation_type": entry["violation_type"],
"description": entry["description"],
}
for entry in INSTANCE_BANK
if not entry["is_feasible"]
]
|