Spaces:
Sleeping
Sleeping
File size: 6,625 Bytes
9c1645f | 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 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | """
Policy tracker for adaptive adversary system.
Tracks agent behavior across episodes to identify patterns and weaknesses.
"""
from collections import defaultdict, deque
from typing import Any
from env.state import ActionRecord
class PolicyTracker:
"""
Tracks agent behavior across episodes.
Maintains a sliding window of last N episodes for analysis.
"""
def __init__(self, window_size: int = 20):
"""
Initialize policy tracker.
Args:
window_size: Number of recent episodes to track
"""
self.window_size = window_size
self.episodes = deque(maxlen=window_size)
self.action_counts = defaultdict(int)
self.total_actions = 0
self.root_cause_scores = defaultdict(list)
self.escalation_steps = []
self.first_actions = []
def record_episode(
self,
episode_id: str,
task_level: str,
root_cause_category: str,
root_cause_id: str,
actions: list[ActionRecord],
score: float
) -> None:
"""
Record completed episode into tracker.
Args:
episode_id: Unique episode identifier
task_level: Difficulty level
root_cause_category: Category of root cause
root_cause_id: Specific root cause ID
actions: List of actions taken
score: Episode score (0-1)
"""
episode_data = {
"episode_id": episode_id,
"task_level": task_level,
"root_cause_category": root_cause_category,
"root_cause_id": root_cause_id,
"actions": actions,
"score": score,
"num_actions": len(actions)
}
self.episodes.append(episode_data)
# Update action counts
for action in actions:
self.action_counts[action.action_name] += 1
self.total_actions += 1
# Track root cause performance
self.root_cause_scores[root_cause_category].append(score)
# Track escalation patterns
for i, action in enumerate(actions):
if action.action_name == "escalate_ticket":
self.escalation_steps.append(i + 1) # 1-indexed
break
# Track first action
if actions:
self.first_actions.append(actions[0].action_name)
def get_action_frequencies(self, task_level: str | None = None) -> dict[str, float]:
"""
Get normalized frequency of each action type.
Args:
task_level: Optional filter by task level
Returns:
Dictionary mapping action names to frequencies (0-1)
"""
if not self.total_actions:
return {}
# Filter episodes if task level specified
if task_level:
filtered_actions = defaultdict(int)
total = 0
for ep in self.episodes:
if ep["task_level"] == task_level:
for action in ep["actions"]:
filtered_actions[action.action_name] += 1
total += 1
if total == 0:
return {}
return {
action: count / total
for action, count in filtered_actions.items()
}
# Return overall frequencies
return {
action: count / self.total_actions
for action, count in self.action_counts.items()
}
def get_first_action_distribution(self) -> dict[str, float]:
"""
Analyze what agent does first.
Returns:
Dictionary mapping action names to probability of being first
"""
if not self.first_actions:
return {}
counts = defaultdict(int)
for action in self.first_actions:
counts[action] += 1
total = len(self.first_actions)
return {
action: count / total
for action, count in counts.items()
}
def get_escalation_step_distribution(self) -> list[int]:
"""
Get list of steps at which agent escalated.
Returns:
List of step numbers (useful for statistical analysis)
"""
return self.escalation_steps.copy()
def get_score_by_root_cause_category(self) -> dict[str, float]:
"""
Get average score per root cause category.
Returns:
Dictionary mapping categories to average scores
"""
return {
category: sum(scores) / len(scores) if scores else 0.0
for category, scores in self.root_cause_scores.items()
}
def get_average_score(self, task_level: str | None = None) -> float:
"""
Get average score across episodes.
Args:
task_level: Optional filter by task level
Returns:
Average score (0-1)
"""
if task_level:
scores = [
ep["score"] for ep in self.episodes
if ep["task_level"] == task_level
]
else:
scores = [ep["score"] for ep in self.episodes]
return sum(scores) / len(scores) if scores else 0.0
def get_episode_count(self) -> int:
"""Get number of recorded episodes."""
return len(self.episodes)
def get_weakness_categories(self, threshold: float = 0.40) -> list[str]:
"""
Identify root cause categories where agent performs poorly.
Args:
threshold: Score threshold below which category is "weak"
Returns:
List of category names with average score < threshold
"""
scores_by_category = self.get_score_by_root_cause_category()
return [
category for category, score in scores_by_category.items()
if score < threshold
]
def uses_action_frequently(self, action_name: str, threshold: float = 0.40) -> bool:
"""
Check if agent uses a specific action frequently.
Args:
action_name: Name of action to check
threshold: Frequency threshold (0-1)
Returns:
True if action frequency >= threshold
"""
frequencies = self.get_action_frequencies()
return frequencies.get(action_name, 0.0) >= threshold
|