Spaces:
Sleeping
Sleeping
| """ | |
| agents/senior_reviewer.py — SeniorEngineerReviewer | |
| Rule-based (NOT an LLM). Fast, deterministic. | |
| Episode-varying qualification constraints = Snorkel bonus track. | |
| API matches env.py exactly: | |
| constraints = sample_episode_constraints(active_params, rng) | |
| reviewer = SeniorEngineerReviewer(constraints) | |
| result = reviewer.review(params_dict) ← plain dict, not FabAction | |
| """ | |
| import numpy as np | |
| from typing import Dict, Optional, Tuple | |
| from environment.rsm_simulator import PARAM_RANGES | |
| def sample_episode_constraints( | |
| active_params: list, | |
| rng=None, | |
| tightness_lo: float = 0.50, | |
| tightness_hi: float = 0.70, | |
| ) -> Dict[str, Tuple[float, float]]: | |
| """ | |
| Sample episode-varying qualified ranges for each active parameter. | |
| Returns a dict of param → (q_lo, q_hi). | |
| Called once per episode in env.reset(). Changes every episode (Snorkel bonus). | |
| The agent doesn't see these values until it submits a recipe. | |
| """ | |
| if rng is None: | |
| rng = np.random.default_rng() | |
| constraints = {} | |
| for param in active_params: | |
| lo, hi = PARAM_RANGES[param] | |
| span = hi - lo | |
| window = span * rng.uniform(tightness_lo, tightness_hi) | |
| offset = rng.uniform(0, span - window) | |
| constraints[param] = ( | |
| round(lo + offset, 4), | |
| round(lo + offset + window, 4), | |
| ) | |
| return constraints | |
| class SeniorEngineerReviewer: | |
| """ | |
| Reviews submitted process recipes against episode-varying constraints. | |
| Multi-agent bonus track: | |
| - Snorkel AI: constraints change every episode | |
| - Fleet AI: one agent monitors/evaluates another agent's outputs | |
| Called from env.step() when action.submit=True. | |
| review() takes a plain dict of {param: value} — same as env._clamp_params output. | |
| """ | |
| def __init__(self, episode_constraints: Dict[str, Tuple[float, float]]): | |
| self.qualified_ranges = episode_constraints | |
| self.revision_budget = 2 | |
| self.reviews_done = 0 | |
| def review(self, recipe: Dict[str, float]) -> Dict: | |
| """ | |
| Check each param in the recipe against its qualified range. | |
| Args: | |
| recipe: plain dict {param_name: float_value} | |
| — exactly what env._clamp_params() returns | |
| Returns: | |
| dict with keys: approved, feedback, violations, revision_budget_remaining | |
| """ | |
| self.reviews_done += 1 | |
| violations = [] | |
| for param, value in recipe.items(): | |
| if param not in self.qualified_ranges: | |
| continue | |
| q_lo, q_hi = self.qualified_ranges[param] | |
| if not (q_lo <= value <= q_hi): | |
| violations.append({ | |
| "param": param, | |
| "submitted": round(value, 4), | |
| "qualified_range": [round(q_lo, 4), round(q_hi, 4)], | |
| "delta": round(min(abs(value - q_lo), abs(value - q_hi)), 4), | |
| }) | |
| remaining = max(0, self.revision_budget - self.reviews_done) | |
| if violations: | |
| parts = [ | |
| f"{v['param']}={v['submitted']} outside " | |
| f"[{v['qualified_range'][0]}, {v['qualified_range'][1]}]" | |
| for v in violations | |
| ] | |
| return { | |
| "approved": False, | |
| "feedback": ( | |
| f"Recipe REJECTED ({len(violations)} violation(s)). " | |
| f"Revise: {'; '.join(parts)}. " | |
| f"Revision budget remaining: {remaining} experiments." | |
| ), | |
| "violations": violations, | |
| "revision_budget_remaining": remaining, | |
| } | |
| return { | |
| "approved": True, | |
| "feedback": "Recipe APPROVED. Qualified for production.", | |
| "violations": [], | |
| "revision_budget_remaining": remaining, | |
| } | |
| def get_constraint_hint(self) -> str: | |
| """ | |
| Vague hint shown in agent prompt after step 8. | |
| Forces agent to hedge toward center-of-range values. | |
| """ | |
| return ( | |
| "NOTE: Your final recipe will be reviewed against production " | |
| "qualification constraints. Parameters far from their nominal " | |
| "process window are at risk of rejection." | |
| ) |