Buckets:
| """ | |
| Process Reward Evaluator for EduAgent Benchmark. | |
| Verifies that the agent's tool-call sequence satisfies ordering constraints | |
| defined in `process_constraints`. This is a programmatic check — no LLM judge needed. | |
| Process constraints are defined per task as: | |
| "process_constraints": [ | |
| {"before": "get_quiz_submissions", "after": "gs_update_slide_content", | |
| "description": "Must analyze quiz data before updating slides"}, | |
| {"before": "get_quiz_submissions", "after": "create_quiz", | |
| "description": "Must diagnose KC gaps before creating follow-up quiz"}, | |
| {"all_before_any": ["get_student_activity", "get_quiz_submissions"], | |
| "after": "send_canvas_message", | |
| "description": "Must gather all data before sending any messages"}, | |
| ] | |
| Constraint types: | |
| - before/after: tool A must appear before tool B in the trajectory | |
| - all_before_any: ALL tools in the list must appear before ANY call to the after tool | |
| - requires: tool A must appear at least N times | |
| - forbidden: tool A, or any tool in a list, must not appear in the trajectory | |
| - critical: if true, a failed constraint zeros the PROCESS reward | |
| For ordering constraints, optional `before_nonempty_args` / `after_nonempty_args` | |
| lists can require that the matched tool call contains non-empty argument fields. | |
| This is useful for workflow tasks where an empty artifact shell should not count | |
| as the completed artifact, e.g. `create_quiz` before messages only if the | |
| `questions` argument is non-empty. | |
| Set `strictly_before_turn` when the before tool must finish in an earlier | |
| assistant turn than the after tool; this prevents same-batch tool calls from | |
| counting as evidence that the agent observed the prerequisite result. | |
| """ | |
| from typing import Optional | |
| from tau2.data_model.message import AssistantMessage, Message, ToolMessage | |
| from tau2.data_model.simulation import RewardInfo | |
| from tau2.data_model.tasks import RewardType, Task | |
| def _canonical_tool_name(name) -> str: | |
| """Normalize internal helper prefixes to evaluator-visible names. Defensively | |
| accepts dict-shaped constraint items ({"func_name": ...}) so a single malformed | |
| constraint cannot crash the whole task evaluation.""" | |
| if isinstance(name, dict): | |
| name = name.get("func_name") or name.get("name") or "" | |
| return (str(name) if name is not None else "").lstrip("_") | |
| def _extract_tool_call_sequence(trajectory: list[Message]) -> list[dict]: | |
| """Extract ordered list of tool calls from trajectory. | |
| Returns list of {"name": str, "arguments": dict, "index": int} | |
| """ | |
| successful_tool_call_ids = { | |
| msg.id | |
| for msg in trajectory | |
| if isinstance(msg, ToolMessage) and msg.requestor == "assistant" and not msg.error | |
| } | |
| calls = [] | |
| idx = 0 | |
| for message_index, msg in enumerate(trajectory): | |
| if isinstance(msg, AssistantMessage) and msg.is_tool_call(): | |
| for tc in msg.tool_calls: | |
| # Process constraints should reward completed actions/data reads, | |
| # not hallucinated or unavailable tool names. The environment | |
| # records failed tool executions as ToolMessage(error=True), so | |
| # only count assistant tool calls that received a successful | |
| # tool response. | |
| call_id = getattr(tc, "id", "") | |
| if call_id and call_id not in successful_tool_call_ids: | |
| continue | |
| # ToolCall pydantic model: tc.name, tc.arguments | |
| name = getattr(tc, 'name', '') or '' | |
| if not name and hasattr(tc, 'function'): | |
| name = getattr(tc.function, 'name', '') | |
| calls.append({ | |
| "name": _canonical_tool_name(name), | |
| "arguments": getattr(tc, 'arguments', {}), | |
| "index": idx, | |
| "message_index": message_index, | |
| }) | |
| idx += 1 | |
| return calls | |
| def _arg_value(arguments: dict, key_path: str): | |
| """Read a possibly dotted argument path from a tool-call argument dict.""" | |
| value = arguments | |
| for part in str(key_path).split("."): | |
| if not isinstance(value, dict) or part not in value: | |
| return None | |
| value = value[part] | |
| return value | |
| def _arguments_match( | |
| arguments: dict, | |
| expected: Optional[dict], | |
| nonempty_args: Optional[list[str]] = None, | |
| ) -> bool: | |
| """Return whether a tool call's arguments contain the expected key-values.""" | |
| if not isinstance(arguments, dict): | |
| return False | |
| if expected: | |
| for key, expected_value in expected.items(): | |
| actual_value = arguments.get(key) | |
| if isinstance(expected_value, list): | |
| if actual_value not in expected_value: | |
| return False | |
| elif actual_value != expected_value: | |
| return False | |
| for key in nonempty_args or []: | |
| value = _arg_value(arguments, key) | |
| if value in (None, "", [], {}): | |
| return False | |
| return True | |
| def _first_index_of( | |
| calls: list[dict], | |
| tool_name: str, | |
| expected_args: Optional[dict] = None, | |
| nonempty_args: Optional[list[str]] = None, | |
| ) -> Optional[int]: | |
| """Find index of first occurrence of a tool in the call sequence.""" | |
| canonical_tool_name = _canonical_tool_name(tool_name) | |
| for c in calls: | |
| if c["name"] == canonical_tool_name and _arguments_match( | |
| c.get("arguments", {}), | |
| expected_args, | |
| nonempty_args, | |
| ): | |
| return c["index"] | |
| return None | |
| def _first_call_of( | |
| calls: list[dict], | |
| tool_name: str, | |
| expected_args: Optional[dict] = None, | |
| nonempty_args: Optional[list[str]] = None, | |
| ) -> Optional[dict]: | |
| """Find first matching call object, including message index metadata.""" | |
| canonical_tool_name = _canonical_tool_name(tool_name) | |
| for c in calls: | |
| if c["name"] == canonical_tool_name and _arguments_match( | |
| c.get("arguments", {}), expected_args, nonempty_args | |
| ): | |
| return c | |
| return None | |
| def _call_by_index(calls: list[dict], index: Optional[int]) -> Optional[dict]: | |
| if index is None: | |
| return None | |
| for call in calls: | |
| if call.get("index") == index: | |
| return call | |
| return None | |
| def _requires_earlier_turn(constraint: dict) -> bool: | |
| return bool( | |
| constraint.get("strictly_before_turn") | |
| or constraint.get("strictly_before_message") | |
| ) | |
| def _last_index_of( | |
| calls: list[dict], | |
| tool_name: str, | |
| expected_args: Optional[dict] = None, | |
| nonempty_args: Optional[list[str]] = None, | |
| ) -> Optional[int]: | |
| """Find index of last occurrence of a tool in the call sequence.""" | |
| result = None | |
| canonical_tool_name = _canonical_tool_name(tool_name) | |
| for c in calls: | |
| if c["name"] == canonical_tool_name and _arguments_match( | |
| c.get("arguments", {}), | |
| expected_args, | |
| nonempty_args, | |
| ): | |
| result = c["index"] | |
| return result | |
| def _count_of( | |
| calls: list[dict], | |
| tool_name: str, | |
| expected_args: Optional[dict] = None, | |
| nonempty_args: Optional[list[str]] = None, | |
| ) -> int: | |
| """Count occurrences of a tool in the call sequence.""" | |
| canonical_tool_name = _canonical_tool_name(tool_name) | |
| return sum( | |
| 1 | |
| for c in calls | |
| if c["name"] == canonical_tool_name | |
| and _arguments_match(c.get("arguments", {}), expected_args, nonempty_args) | |
| ) | |
| class ProcessRewardEvaluator: | |
| """Evaluates whether the agent's tool-call ordering satisfies process constraints.""" | |
| def calculate_reward( | |
| cls, | |
| task: Task, | |
| full_trajectory: list[Message], | |
| ) -> RewardInfo: | |
| ec = task.evaluation_criteria | |
| if ec is None: | |
| return RewardInfo( | |
| reward=1.0, | |
| info={"note": "No evaluation criteria"}, | |
| reward_breakdown={RewardType.PROCESS: 1.0}, | |
| ) | |
| constraints = getattr(ec, "process_constraints", None) | |
| if not constraints: | |
| return RewardInfo( | |
| reward=1.0, | |
| info={"note": "No process constraints"}, | |
| reward_breakdown={RewardType.PROCESS: 1.0}, | |
| ) | |
| calls = _extract_tool_call_sequence(full_trajectory) | |
| results = cls._check_constraints(calls, constraints) | |
| n_pass = sum(1 for r in results if r["met"]) | |
| n_total = len(results) | |
| reward = n_pass / n_total if n_total > 0 else 1.0 | |
| critical_failed = any(r.get("critical") and not r.get("met") for r in results) | |
| if critical_failed: | |
| reward = 0.0 | |
| return RewardInfo( | |
| reward=reward, | |
| info={ | |
| "process_checks": results, | |
| "tool_call_count": len(calls), | |
| "tool_call_sequence": [c["name"] for c in calls], | |
| # When a `critical` gate fails the PROCESS reward is a hard zero | |
| # that the downstream soft-floor multiplier must NOT lift: a | |
| # skipped required-baseline check is a genuine pass-blocker. | |
| # A non-critical fraction of 0.0 (no critical gate) is recoverable | |
| # and the floor applies normally. | |
| "critical_failed": critical_failed, | |
| }, | |
| reward_breakdown={RewardType.PROCESS: reward}, | |
| ) | |
| def _check_constraints(cls, calls: list[dict], constraints: list[dict]) -> list[dict]: | |
| results = [] | |
| for constraint in constraints: | |
| desc = constraint.get("description", "") | |
| if "before_any" in constraint and "after" in constraint: | |
| # Any of the "before_any" tools must appear before the "after" tool | |
| # "after" can be a list — find earliest occurrence of any of the after tools | |
| before_tools = [_canonical_tool_name(t) for t in constraint["before_any"]] | |
| after_name = constraint["after"] | |
| if isinstance(after_name, list): | |
| after_name = [_canonical_tool_name(t) for t in after_name] | |
| else: | |
| after_name = _canonical_tool_name(after_name) | |
| before_args = constraint.get("before_args") | |
| after_args = constraint.get("after_args") | |
| before_nonempty_args = constraint.get("before_nonempty_args") | |
| after_nonempty_args = constraint.get("after_nonempty_args") | |
| if isinstance(after_name, list): | |
| after_candidates = [ | |
| (t, _first_index_of(calls, t, after_args, after_nonempty_args)) | |
| for t in after_name | |
| ] | |
| called_after = [(t, i) for t, i in after_candidates if i is not None] | |
| if called_after: | |
| found_after_tool, after_idx = min(called_after, key=lambda x: x[1]) | |
| else: | |
| found_after_tool, after_idx = after_name[0], None | |
| after_label = f"any({after_name})" | |
| else: | |
| after_idx = _first_index_of(calls, after_name, after_args, after_nonempty_args) | |
| found_after_tool = after_name | |
| after_label = after_name | |
| if after_idx is None: | |
| # After tool never called — constraint is vacuously satisfied | |
| # (can't violate ordering if the action never happened) | |
| met = True | |
| reason = f"Tool '{after_label}' never called (constraint vacuously satisfied)" | |
| else: | |
| # Check if ANY of the before tools was called before the after tool | |
| earliest_before = None | |
| found_tool = None | |
| for bt in before_tools: | |
| idx = _first_index_of(calls, bt, before_args, before_nonempty_args) | |
| if idx is not None and (earliest_before is None or idx < earliest_before): | |
| earliest_before = idx | |
| found_tool = bt | |
| if earliest_before is not None and earliest_before < after_idx: | |
| if _requires_earlier_turn(constraint): | |
| before_call = _call_by_index(calls, earliest_before) | |
| after_call = _call_by_index(calls, after_idx) | |
| if before_call and after_call and before_call.get("message_index") < after_call.get("message_index"): | |
| met = True | |
| reason = f"'{found_tool}' (idx={earliest_before}, msg={before_call.get('message_index')}) before '{found_after_tool}' (idx={after_idx}, msg={after_call.get('message_index')})" | |
| else: | |
| met = False | |
| reason = f"'{found_tool}' must complete in an earlier assistant turn than '{found_after_tool}'" | |
| else: | |
| met = True | |
| reason = f"'{found_tool}' (idx={earliest_before}) before '{found_after_tool}' (idx={after_idx})" | |
| elif earliest_before is not None: | |
| met = False | |
| reason = f"Data tool '{found_tool}' (idx={earliest_before}) NOT before '{found_after_tool}' (idx={after_idx})" | |
| else: | |
| met = False | |
| reason = f"None of {before_tools} called before '{found_after_tool}'" | |
| results.append({ | |
| "constraint": constraint, | |
| "met": met, | |
| "reason": reason, | |
| "description": desc, | |
| "critical": bool(constraint.get("critical")), | |
| }) | |
| elif "before" in constraint and "after" in constraint: | |
| # Simple ordering: before tool must appear before after tool. | |
| # "before" and "after" can be lists — satisfied if ANY listed | |
| # before tool appears before ANY listed after tool. | |
| before_name = constraint["before"] | |
| if isinstance(before_name, list): | |
| before_name = [_canonical_tool_name(t) for t in before_name] | |
| else: | |
| before_name = _canonical_tool_name(before_name) | |
| after_name = constraint["after"] | |
| if isinstance(after_name, list): | |
| after_name = [_canonical_tool_name(t) for t in after_name] | |
| else: | |
| after_name = _canonical_tool_name(after_name) | |
| before_args = constraint.get("before_args") | |
| after_args = constraint.get("after_args") | |
| before_nonempty_args = constraint.get("before_nonempty_args") | |
| after_nonempty_args = constraint.get("after_nonempty_args") | |
| if isinstance(before_name, list): | |
| before_indices = [ | |
| (t, _first_index_of(calls, t, before_args, before_nonempty_args)) | |
| for t in before_name | |
| ] | |
| called_before = [(t, i) for t, i in before_indices if i is not None] | |
| if called_before: | |
| before_tool, before_idx = min(called_before, key=lambda x: x[1]) | |
| else: | |
| before_tool, before_idx = before_name[0], None | |
| before_label = f"any({before_name})" | |
| else: | |
| before_tool = before_name | |
| before_idx = _first_index_of(calls, before_name, before_args, before_nonempty_args) | |
| before_label = before_name | |
| if isinstance(after_name, list): | |
| # Find earliest occurrence of any after tool | |
| after_indices = [ | |
| (t, _first_index_of(calls, t, after_args, after_nonempty_args)) | |
| for t in after_name | |
| ] | |
| called = [(t, i) for t, i in after_indices if i is not None] | |
| if called: | |
| after_tool, after_idx = min(called, key=lambda x: x[1]) | |
| else: | |
| after_tool, after_idx = after_name[0], None | |
| after_label = f"any({after_name})" | |
| else: | |
| after_tool = after_name | |
| after_idx = _first_index_of(calls, after_name, after_args, after_nonempty_args) | |
| after_label = after_name | |
| if before_idx is None: | |
| met = False | |
| reason = f"Tool '{before_label}' never called" | |
| elif after_idx is None: | |
| met = False | |
| reason = f"Tool '{after_label}' never called" | |
| elif before_idx < after_idx: | |
| if _requires_earlier_turn(constraint): | |
| before_call = _call_by_index(calls, before_idx) | |
| after_call = _call_by_index(calls, after_idx) | |
| if before_call and after_call and before_call.get("message_index") < after_call.get("message_index"): | |
| met = True | |
| reason = f"'{before_tool}' (idx={before_idx}, msg={before_call.get('message_index')}) before '{after_tool}' (idx={after_idx}, msg={after_call.get('message_index')})" | |
| else: | |
| met = False | |
| reason = f"'{before_tool}' must complete in an earlier assistant turn than '{after_tool}'" | |
| else: | |
| met = True | |
| reason = f"'{before_tool}' (idx={before_idx}) before '{after_tool}' (idx={after_idx})" | |
| else: | |
| met = False | |
| reason = f"'{before_tool}' (idx={before_idx}) NOT before '{after_tool}' (idx={after_idx})" | |
| results.append({ | |
| "constraint": constraint, | |
| "met": met, | |
| "reason": reason, | |
| "description": desc, | |
| "critical": bool(constraint.get("critical")), | |
| }) | |
| elif "all_before_any" in constraint: | |
| # All tools in list must complete before any call to after tool | |
| before_tools = [_canonical_tool_name(t) for t in constraint["all_before_any"]] | |
| after_name = constraint["after"] | |
| if isinstance(after_name, list): | |
| after_name = [_canonical_tool_name(t) for t in after_name] | |
| else: | |
| after_name = _canonical_tool_name(after_name) | |
| before_args = constraint.get("before_args") | |
| after_args = constraint.get("after_args") | |
| before_nonempty_args = constraint.get("before_nonempty_args") | |
| after_nonempty_args = constraint.get("after_nonempty_args") | |
| if isinstance(after_name, list): | |
| after_indices = [ | |
| (t, _first_index_of(calls, t, after_args, after_nonempty_args)) | |
| for t in after_name | |
| ] | |
| called_after = [(t, i) for t, i in after_indices if i is not None] | |
| if called_after: | |
| after_tool, after_first = min(called_after, key=lambda x: x[1]) | |
| else: | |
| after_tool, after_first = after_name[0], None | |
| after_label = f"any({after_name})" | |
| else: | |
| after_tool = after_name | |
| after_first = _first_index_of(calls, after_name, after_args, after_nonempty_args) | |
| after_label = after_name | |
| if after_first is None: | |
| met = False | |
| reason = f"Tool '{after_label}' never called" | |
| else: | |
| all_before = True | |
| missing = [] | |
| late = [] | |
| for bt in before_tools: | |
| bt_last = _last_index_of(calls, bt, before_args, before_nonempty_args) | |
| if bt_last is None: | |
| all_before = False | |
| missing.append(bt) | |
| elif bt_last >= after_first: | |
| all_before = False | |
| late.append(f"{bt}(idx={bt_last})") | |
| if missing: | |
| met = False | |
| reason = f"Tools never called: {missing}" | |
| elif late: | |
| met = False | |
| reason = f"Tools called after '{after_tool}'(idx={after_first}): {late}" | |
| else: | |
| met = True | |
| reason = f"All prerequisite tools called before '{after_tool}'" | |
| results.append({ | |
| "constraint": constraint, | |
| "met": met, | |
| "reason": reason, | |
| "description": desc, | |
| "critical": bool(constraint.get("critical")), | |
| }) | |
| elif "requires" in constraint: | |
| # Tool must be called at least N times | |
| tool_name = _canonical_tool_name(constraint["requires"]) | |
| min_count = constraint.get("min_count", 1) | |
| actual = _count_of( | |
| calls, | |
| tool_name, | |
| constraint.get("args"), | |
| constraint.get("nonempty_args"), | |
| ) | |
| met = actual >= min_count | |
| reason = f"'{tool_name}' called {actual} times (min={min_count})" | |
| results.append({ | |
| "constraint": constraint, | |
| "met": met, | |
| "reason": reason, | |
| "description": desc, | |
| "critical": bool(constraint.get("critical")), | |
| }) | |
| elif "requires_any" in constraint: | |
| # At least one of the listed tools must be called | |
| tools = [_canonical_tool_name(t) for t in constraint["requires_any"]] | |
| min_count = constraint.get("min_count", 1) | |
| actual = sum( | |
| _count_of( | |
| calls, | |
| t, | |
| constraint.get("args"), | |
| constraint.get("nonempty_args"), | |
| ) | |
| for t in tools | |
| ) | |
| met = actual >= min_count | |
| reason = f"Any of {tools} called {actual} times (min={min_count})" | |
| results.append({ | |
| "constraint": constraint, | |
| "met": met, | |
| "reason": reason, | |
| "description": desc, | |
| "critical": bool(constraint.get("critical")), | |
| }) | |
| elif "forbidden" in constraint: | |
| # None of the listed tools may appear in the trajectory | |
| forbidden_value = constraint["forbidden"] | |
| if isinstance(forbidden_value, list): | |
| forbidden_tools = [_canonical_tool_name(t) for t in forbidden_value] | |
| else: | |
| forbidden_tools = [_canonical_tool_name(forbidden_value)] | |
| counts = {tool: _count_of(calls, tool) for tool in forbidden_tools} | |
| used = {tool: count for tool, count in counts.items() if count > 0} | |
| met = not used | |
| reason = ( | |
| f"Forbidden tools not called: {forbidden_tools}" | |
| if met | |
| else f"Forbidden tools called: {used}" | |
| ) | |
| results.append({ | |
| "constraint": constraint, | |
| "met": met, | |
| "reason": reason, | |
| "description": desc, | |
| "critical": bool(constraint.get("critical")), | |
| }) | |
| return results | |
Xet Storage Details
- Size:
- 24.4 kB
- Xet hash:
- 3b0357b33db0c7ff8a4c9517ec7a8b08b0fa66b99333a4fd9d256c18cf67911e
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.