Spaces:
Sleeping
Sleeping
File size: 3,175 Bytes
662eb9e | 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 | from __future__ import annotations
from benchmark.agents import MultiAgent
from env.agents.coordinator import CoordinatorAgent
from env.environment import ExecutiveEmailEnv
def _first_action(task_id: str, seed: int = 42, persona: str = "balanced"):
"""Reset the env for a task and ask a task-aware coordinator for its first move."""
env = ExecutiveEmailEnv(task_id=task_id, seed=seed, persona=persona)
observation = env.reset(task_id=task_id, seed=seed, persona=persona)
coordinator = CoordinatorAgent(task_id=task_id)
return coordinator.execute(observation)
def test_coordinator_priority_order_is_task_aware():
# Classification/prioritization bias toward the ClassifierAgent; the hard task
# keeps the original risk-first escalation ordering.
assert CoordinatorAgent(task_id="easy_classification")._get_priority_order()[0] == (
"ClassifierAgent"
)
assert CoordinatorAgent(task_id="medium_prioritization")._get_priority_order()[0] == (
"ClassifierAgent"
)
assert CoordinatorAgent(task_id="hard_full_management")._get_priority_order()[0] == (
"EscalatorAgent"
)
def test_coordinator_defaults_to_hard_behavior():
# Default construction (no task supplied) preserves the legacy risk-first order,
# so existing callers keep their behavior.
assert CoordinatorAgent()._get_priority_order()[0] == "EscalatorAgent"
def test_set_task_updates_priority_order():
coordinator = CoordinatorAgent()
assert coordinator._get_priority_order()[0] == "EscalatorAgent"
coordinator.set_task("easy_classification")
assert coordinator._get_priority_order()[0] == "ClassifierAgent"
def test_easy_classification_emits_classify_action():
# The inbox carries risk-tagged emails, so EscalatorAgent.can_handle is True; a
# task-blind coordinator would escalate. The task-aware coordinator must classify.
action = _first_action("easy_classification")
assert action is not None
assert action.action_type == "classify"
def test_easy_classification_scores_well_above_zero():
metrics = MultiAgent().run(task_id="easy_classification", seed=42, persona="balanced")
# Task-blind coordinator floored at ~0 (epsilon) by escalating instead of
# classifying; task-aware routing must beat that decisively.
assert metrics.score > 0.5
def test_medium_prioritization_emits_prioritize_first():
# None of the content specialists emit a ranking, so the task-aware coordinator
# must open the prioritization task with a `prioritize` action.
action = _first_action("medium_prioritization")
assert action is not None
assert action.action_type == "prioritize"
assert action.priority_order # non-empty ranking
def test_medium_prioritization_scores_well_above_zero():
metrics = MultiAgent().run(task_id="medium_prioritization", seed=42, persona="balanced")
assert metrics.score > 0.5
def test_hard_full_management_still_escalates_risk():
# Genuine legal/security risk must still be escalated on the hard task.
action = _first_action("hard_full_management")
assert action is not None
assert action.action_type == "escalate"
|