| | """Reward/evaluation rubrics for HR Onboarding/Offboarding tasks. |
| | |
| | Each task has a set of rubric criteria. This module evaluates agent action logs |
| | against those criteria to compute rewards. |
| | """ |
| |
|
| | import re |
| | from typing import Any |
| | try: |
| | from .tasks import Task |
| | except ImportError: |
| | from tasks import Task |
| |
|
| |
|
| | class RubricEvaluator: |
| | """Evaluates agent performance against task rubric criteria.""" |
| |
|
| | def __init__(self): |
| | self._checkers = { |
| | "tool_used": self._check_tool_used, |
| | "tool_not_used": self._check_tool_not_used, |
| | "tool_used_any": self._check_tool_used_any, |
| | "param_value": self._check_param_value, |
| | "param_contains": self._check_param_contains, |
| | "tool_order": self._check_tool_order, |
| | "tool_count": self._check_tool_count, |
| | "result_contains": self._check_result_contains, |
| | } |
| |
|
| | def evaluate(self, task: Task, action_log: list[dict]) -> dict: |
| | """Evaluate action log against task rubric criteria. |
| | |
| | Returns: |
| | { |
| | "task_id": str, |
| | "criteria_results": list of {name, passed, description}, |
| | "score": float (0.0-1.0), |
| | "passed": bool (all criteria satisfied), |
| | } |
| | """ |
| | criteria_results = [] |
| | for criterion in task.rubric_criteria: |
| | check_str = criterion["check"] |
| | passed = self._evaluate_criterion(check_str, action_log) |
| | criteria_results.append({ |
| | "name": criterion["name"], |
| | "description": criterion["description"], |
| | "passed": passed, |
| | }) |
| |
|
| | total = len(criteria_results) |
| | passed_count = sum(1 for c in criteria_results if c["passed"]) |
| | score = passed_count / total if total > 0 else 0.0 |
| |
|
| | return { |
| | "task_id": task.task_id, |
| | "criteria_results": criteria_results, |
| | "score": score, |
| | "passed": all(c["passed"] for c in criteria_results), |
| | "passed_count": passed_count, |
| | "total_criteria": total, |
| | } |
| |
|
| | def _evaluate_criterion(self, check_str: str, action_log: list[dict]) -> bool: |
| | """Parse and evaluate a single criterion check string.""" |
| | |
| | parts = check_str.split(":", 1) |
| | if len(parts) != 2: |
| | return False |
| |
|
| | check_type = parts[0] |
| | check_args = parts[1] |
| |
|
| | checker = self._checkers.get(check_type) |
| | if not checker: |
| | return False |
| |
|
| | return checker(check_args, action_log) |
| |
|
| | def _check_tool_used(self, tool_name: str, action_log: list[dict]) -> bool: |
| | """Check if a specific tool was used at least once.""" |
| | return any(a["tool"] == tool_name for a in action_log) |
| |
|
| | def _check_tool_not_used(self, tool_name: str, action_log: list[dict]) -> bool: |
| | """Check that a specific tool was NOT used.""" |
| | return not any(a["tool"] == tool_name for a in action_log) |
| |
|
| | def _check_tool_used_any(self, tools_csv: str, action_log: list[dict]) -> bool: |
| | """Check if any of the comma-separated tools were used.""" |
| | tool_names = [t.strip() for t in tools_csv.split(",")] |
| | return any(a["tool"] in tool_names for a in action_log) |
| |
|
| | def _check_param_value(self, spec: str, action_log: list[dict]) -> bool: |
| | """Check if a tool was called with a specific parameter value. |
| | Format: tool_name.param_name=expected_value |
| | """ |
| | match = re.match(r"(\w+)\.(\w+)=(.+)", spec) |
| | if not match: |
| | return False |
| | tool_name, param_name, expected_value = match.groups() |
| |
|
| | for action in action_log: |
| | if action["tool"] == tool_name: |
| | actual = action["params"].get(param_name) |
| | if actual is not None and str(actual) == expected_value: |
| | return True |
| | |
| | updates = action["params"].get("updates", {}) |
| | if param_name in updates and str(updates[param_name]) == expected_value: |
| | return True |
| | return False |
| |
|
| | def _check_param_contains(self, spec: str, action_log: list[dict]) -> bool: |
| | """Check if a tool parameter contains a substring. |
| | Format: tool_name.param_name=substring |
| | """ |
| | match = re.match(r"(\w+)\.(\w+)=(.+)", spec) |
| | if not match: |
| | return False |
| | tool_name, param_name, substring = match.groups() |
| |
|
| | for action in action_log: |
| | if action["tool"] == tool_name: |
| | actual = action["params"].get(param_name, "") |
| | if substring.lower() in str(actual).lower(): |
| | return True |
| | return False |
| |
|
| | def _check_tool_order(self, spec: str, action_log: list[dict]) -> bool: |
| | """Check that tool A was called before tool B. |
| | Format: tool_a<tool_b |
| | """ |
| | parts = spec.split("<") |
| | if len(parts) != 2: |
| | return False |
| | tool_a, tool_b = parts |
| |
|
| | idx_a = None |
| | idx_b = None |
| | for i, action in enumerate(action_log): |
| | if action["tool"] == tool_a and idx_a is None: |
| | idx_a = i |
| | if action["tool"] == tool_b and idx_b is None: |
| | idx_b = i |
| |
|
| | if idx_a is None or idx_b is None: |
| | return False |
| | return idx_a < idx_b |
| |
|
| | def _check_tool_count(self, spec: str, action_log: list[dict]) -> bool: |
| | """Check that a tool was called at least N times. |
| | Format: tool_name>=N |
| | """ |
| | match = re.match(r"(\w+)>=(\d+)", spec) |
| | if not match: |
| | return False |
| | tool_name, min_count = match.groups() |
| | min_count = int(min_count) |
| |
|
| | count = sum(1 for a in action_log if a["tool"] == tool_name) |
| | return count >= min_count |
| |
|
| | def _check_result_contains(self, substring: str, action_log: list[dict]) -> bool: |
| | """Check if any tool result contains a substring.""" |
| | for action in action_log: |
| | result_str = str(action.get("result", "")) |
| | if substring.lower() in result_str.lower(): |
| | return True |
| | return False |
| |
|
| |
|
| | def compute_reward(task: Task, action_log: list[dict]) -> float: |
| | """Convenience function to compute reward for a task given action log.""" |
| | evaluator = RubricEvaluator() |
| | result = evaluator.evaluate(task, action_log) |
| | return result["score"] |
| |
|
