| """FGAO v2 controller (Feedback-Greedy Agentic Orchestration). |
| |
| A training-free, model-free numerical controller. At every step it picks |
| |
| a* = argmax_a Delta_hat(a) - lam * c_hat(a) / B0 + gamma * H_hat(a) |
| |
| where Delta_hat is the predicted weighted progress on the verifiable goals |
| (declared prior blended with online feedback, rho = nu / (nu + n)), c_hat the |
| estimated cost, and H_hat the one-step continuation value (lookahead). It stops |
| when every mandatory goal is verified, or when no admissible, affordable action |
| has positive value (eq. 20), or when the budget is spent. |
| |
| The controller performs no I/O. It is driven through three calls:: |
| |
| ctrl = FGAOController(conditions, actions, budget=16) |
| ctrl.reset(observation) |
| while ctrl.should_stop() is None: |
| decision = ctrl.select_action() |
| observation, cost = execute(decision.action) # your code |
| ctrl.update(decision.action, observation, cost) |
| |
| The numerical algorithm (estimator, scoring, lookahead forms, tie-breaking, |
| stopping rule) is the tested v2 implementation, unchanged. Defaults are the |
| configuration used for RAG: unlock lookahead, lam = 0.1, nu = 4, gamma = 0.9, |
| task-declared comparability groups. |
| """ |
| from __future__ import annotations |
|
|
| import itertools |
| import math |
| import random |
| import time |
| from dataclasses import dataclass |
| from typing import Callable, Dict, Hashable, Iterable, List, Mapping, Optional, Sequence, Set, Tuple, Union |
|
|
| from .estimator import EffectStatistics, blend |
| from .types import ( |
| SAT_THRESHOLD, |
| Action, |
| ActionSpec, |
| Condition, |
| Decision, |
| Observation, |
| ObsState, |
| Scores, |
| Semantics, |
| Status, |
| all_mandatory_verified, |
| deficiency, |
| ) |
|
|
| MAX_ENUM_EFFECTS = 6 |
|
|
|
|
| @dataclass |
| class FGAOConfig: |
| """Controller hyperparameters. Defaults are the evaluated v2 RAG configuration.""" |
|
|
| nu: float = 4.0 |
| lam: float = 0.1 |
| gamma: float = 0.9 |
| lookahead: str = "unlock" |
| lookahead_depth: int = 1 |
| corrective: bool = True |
| similarity: str = "task" |
| estimator: str = "rate" |
| feedback: bool = True |
| objective: str = "additive" |
| beta: float = 0.0 |
| optional_mass: Optional[float] = None |
| cost_scale: Optional[float] = None |
| max_steps: int = 10_000 |
|
|
| def __post_init__(self) -> None: |
| if self.lookahead not in ("unlock", "eq16", "eq17", "none"): |
| raise ValueError("lookahead must be unlock | eq16 | eq17 | none") |
| if self.estimator not in ("rate", "raw"): |
| raise ValueError("estimator must be rate | raw") |
| if self.objective not in ("additive", "ratio"): |
| raise ValueError("objective must be additive | ratio") |
| if self.similarity not in ("task", "exact", "type"): |
| raise ValueError("similarity must be task | exact | type") |
|
|
|
|
| class ControllerStopped(RuntimeError): |
| """Raised by ``select_action`` once the controller has stopped.""" |
|
|
|
|
| ActionsArg = Union[Sequence[ActionSpec], Mapping[Action, Semantics]] |
|
|
|
|
| class FGAOController: |
| """Step-wise FGAO v2 controller for one episode (one query / one task run). |
| |
| Parameters |
| ---------- |
| conditions : goal conditions with weights; mandatory ones define success. |
| actions : ``ActionSpec`` list (or ``{Action: Semantics}``) in a fixed |
| order. The order matters only for exact ties, which are broken |
| by a seeded RNG. |
| config : ``FGAOConfig``. |
| budget : execution budget B0 (``inf`` allowed). |
| seed : seeds the tie-breaking RNG. |
| comparability_key : optional ``Action -> Hashable``; overrides ``ActionSpec.group``. |
| prior : optional ``Action -> Semantics`` used for the effect prior only |
| (preconditions and declared costs still come from ``actions``). |
| """ |
|
|
| def __init__( |
| self, |
| conditions: Sequence[Condition], |
| actions: ActionsArg, |
| config: Optional[FGAOConfig] = None, |
| *, |
| budget: float = float("inf"), |
| seed: int = 0, |
| comparability_key: Optional[Callable[[Action], Hashable]] = None, |
| prior: Optional[Callable[[Action], Semantics]] = None, |
| ) -> None: |
| self.cfg = config or FGAOConfig() |
| if isinstance(actions, Mapping): |
| specs = [ActionSpec(a, s) for a, s in actions.items()] |
| else: |
| specs = list(actions) |
| if len({s.action for s in specs}) != len(specs): |
| raise ValueError("duplicate action in the action set") |
| self.conditions: List[Condition] = list(conditions) |
| names = {c.name for c in self.conditions} |
| if len(names) != len(self.conditions): |
| raise ValueError("duplicate condition name") |
| self.weights = self.assign_weights(self.conditions) |
| self.candidates: List[Action] = [s.action for s in specs] |
| self._req: Dict[Action, Semantics] = {s.action: s.semantics for s in specs} |
| self._sem: Dict[Action, Semantics] = {a: (prior(a) if prior else self._req[a]) for a in self.candidates} |
| self._needs_pred: Dict[str, List[Action]] = {} |
| self._needs_cond: Dict[str, List[Action]] = {} |
| for a in self.candidates: |
| for p in self._req[a].requires_predicates: |
| self._needs_pred.setdefault(p, []).append(a) |
| for c in self._req[a].requires_conditions: |
| self._needs_cond.setdefault(c, []).append(a) |
| self._init_gainers() |
| groups = {s.action: s.group for s in specs if s.group is not None} |
| key_fn = None |
| if self.cfg.similarity == "task": |
| key_fn = comparability_key or (lambda a: groups.get(a, a.signature)) |
| self._key_fn = key_fn |
| self.stats = EffectStatistics(self.cfg.similarity, key_fn) |
| self._by_key: Dict[Hashable, List[Action]] = {} |
| for a in self.candidates: |
| self._by_key.setdefault(self.stats.key(a), []).append(a) |
| self.n_effect_evals = 0 |
| self.budget0 = float(budget) |
| self._cost_scale = self.cfg.cost_scale or (self.budget0 if math.isfinite(self.budget0) else 1.0) |
| self._rng = random.Random(seed * 7919 + 17) |
| self.seed = seed |
| self._started = False |
| self._invalidate() |
|
|
| def _init_gainers(self) -> None: |
| self._gainers: Dict[str, set] = {} |
| for a in self.candidates: |
| for c, g in self._sem[a].gain.items(): |
| if g > 0: |
| self._gainers.setdefault(c, set()).add(a) |
|
|
| def _invalidate_key(self, key: Hashable) -> None: |
| for a in self._by_key.get(key, ()): |
| self._cost_memo.pop(a, None) |
| self._support.pop(a, None) |
| self._rate_memo.pop(a, None) |
|
|
| def _invalidate(self) -> None: |
| """Drop memoised estimates. They depend only on the statistics, which change only in ``update``.""" |
| self._rate_memo: Dict[Action, Dict[Tuple[str, str], float]] = {} |
| self._cost_memo: Dict[Action, float] = {} |
| self._support: Dict[Action, List[Tuple[str, float, float, float]]] = {} |
|
|
| |
| def reset(self, observation: Observation) -> "FGAOController": |
| """Start a fresh episode from the initial observation (online statistics start empty, as in v2).""" |
| self.stats = EffectStatistics(self.cfg.similarity, self._key_fn) |
| self._init_gainers() |
| self.n_effect_evals = 0 |
| self._rng = random.Random(self.seed * 7919 + 17) |
| self.v: Dict[str, float] = dict(observation.values) |
| self.preds: Dict[str, bool] = dict(observation.predicates) |
| self.budget = self.budget0 |
| self.spent = 0.0 |
| self.t = 0 |
| self.history: List[Dict] = [] |
| self.status: str = Status.RUNNING |
| self._pending: Optional[Decision] = None |
| self._rng_at_decide = None |
| self.blocked: Set[Action] = set() |
| self._started = True |
| self._invalidate() |
| return self |
|
|
| def block(self, action: Action) -> None: |
| """Exclude ``action`` for the rest of the episode (e.g. a human rejected it).""" |
| if action not in self._req: |
| raise KeyError(f"unknown action {action}") |
| self.blocked.add(action) |
| if self._pending is not None and self._pending.action == action: |
| self._discard_pending() |
|
|
| def _discard_pending(self) -> None: |
| if self._rng_at_decide is not None: |
| self._rng.setstate(self._rng_at_decide) |
| self._pending = None |
| self._rng_at_decide = None |
|
|
| @property |
| def state(self) -> ObsState: |
| return ObsState(dict(self.v), dict(self.preds), self.budget, self.t) |
|
|
| @property |
| def success(self) -> bool: |
| return all_mandatory_verified(self.conditions, self.v) |
|
|
| @property |
| def deficiency(self) -> float: |
| return deficiency(self.conditions, self.v) |
|
|
| def admissible_actions(self) -> List[Action]: |
| st = self.state |
| return [a for a in self.candidates if self._admissible(a, st)] |
|
|
| def _decide(self) -> None: |
| if self.status != Status.RUNNING or self._pending is not None: |
| return |
| state = ObsState(self.v, self.preds, self.budget, self.t) |
| if all_mandatory_verified(self.conditions, self.v): |
| self.status = Status.SUCCESS |
| return |
| if self.t >= self.cfg.max_steps: |
| self.status = Status.STEP_LIMIT |
| return |
| if self.budget <= 0: |
| self.status = Status.BUDGET_EXHAUSTED |
| return |
| t_sel = time.perf_counter() |
| self._rng_at_decide = self._rng.getstate() |
| admissible = [a for a in self.candidates if self._admissible(a, state) and a not in self.blocked] |
| if not admissible: |
| self.status = Status.BLOCKED |
| return |
| affordable = [a for a in admissible if self.estimated_cost(a) <= self.budget] |
| if not affordable: |
| self.status = Status.BUDGET_EXHAUSTED |
| return |
| self._begin_step(affordable, state) |
| scored = [(a, self.score(a, state)) for a in affordable] |
| best_exploit = max(s["exploit"] for _, s in scored) |
| if best_exploit <= 0: |
| self.status = Status.STALLED |
| return |
| top = max(s["rank"] for _, s in scored) |
| ties = [(a, s) for a, s in scored if s["rank"] >= top - 1e-12] |
| action, sc = ties[self._rng.randrange(len(ties))] if len(ties) > 1 else ties[0] |
| self._pending = Decision( |
| step=self.t, |
| action=action, |
| scores=Scores(**sc), |
| candidates=[(a, Scores(**s)) for a, s in scored], |
| n_admissible=len(admissible), |
| select_time=time.perf_counter() - t_sel, |
| ) |
|
|
| def should_stop(self) -> Optional[str]: |
| """``None`` while running; otherwise the termination status. |
| |
| Calling it computes (and caches) the next decision, so it is cheap to |
| follow with ``select_action``. |
| """ |
| if not self._started: |
| raise RuntimeError("call reset(observation) first") |
| self._decide() |
| return None if self.status == Status.RUNNING else self.status |
|
|
| def select_action(self) -> Decision: |
| """The action to execute next, with its scores and every alternative.""" |
| if self.should_stop() is not None: |
| raise ControllerStopped(self.status) |
| return self._pending |
|
|
| def update( |
| self, |
| action: Action, |
| observation: Observation, |
| cost: Optional[float] = None, |
| error: Optional[str] = None, |
| ) -> Dict: |
| """Record the outcome of executing ``action``. |
| |
| ``observation`` holds the goal values and predicates *after* the |
| execution, ``cost`` the measured cost (default: the declared cost). |
| Updates budget, history and the online effect estimator. |
| """ |
| if not self._started: |
| raise RuntimeError("call reset(observation) first") |
| if self.status != Status.RUNNING: |
| raise ControllerStopped(self.status) |
| if action not in self._req: |
| raise KeyError(f"unknown action {action}") |
| dec = self._pending if (self._pending is not None and self._pending.action == action) else None |
| cost = self._req[action].cost if cost is None else float(cost) |
| v_before, p_before = dict(self.v), dict(self.preds) |
| self.v = dict(observation.values) |
| self.preds = dict(observation.predicates) |
| self.budget -= cost |
| self.spent += cost |
| if self.cfg.feedback: |
| self.stats.update(action, v_before, self.v, p_before, self.preds, cost) |
| self._invalidate_key(self.stats.key(action)) |
| for c in self.weights: |
| if self.v.get(c, 0.0) > v_before.get(c, 0.0): |
| self._gainers.setdefault(c, set()).add(action) |
| observed = sum(self.weights[c] * (self.v.get(c, 0.0) - v_before.get(c, 0.0)) for c in self.weights) |
| rec = { |
| "step": self.t, |
| "action": str(action), |
| "Q": dec.scores.Q if dec else None, |
| "predicted_gain": dec.scores.gain if dec else None, |
| "continuation": dec.scores.continuation if dec else None, |
| "estimated_cost": dec.scores.cost if dec else None, |
| "observed_effect": observed, |
| "cost": cost, |
| "budget_left": self.budget, |
| "deficiency": deficiency(self.conditions, self.v), |
| "values": dict(self.v), |
| "error": error, |
| "n_actions": len(dec.candidates) if dec else None, |
| "select_time": dec.select_time if dec else None, |
| } |
| self.history.append(rec) |
| self.t += 1 |
| self._pending = None |
| self._rng_at_decide = None |
| return rec |
|
|
| |
| def select_batch(self, max_parallel: int, |
| conflicts: Optional[Callable[[Action, Action], bool]] = None, |
| independent: bool = True) -> List[Decision]: |
| """Up to ``max_parallel`` actions to execute concurrently. |
| |
| The first is exactly ``select_action()`` (the v2 choice). Further slots are |
| filled greedily with actions that are admissible and affordable now, do not |
| conflict with the batch, and still have positive net value |
| S(b | s_hat) = Delta_hat(b | s_hat) - lam c_hat(b)/B0 in the *expected* state s_hat |
| after the actions already chosen (each chosen action moves every goal to |
| v + delta_hat). Actions are never repeated within a batch. With |
| ``max_parallel=1`` this is ``[select_action()]``. |
| |
| independent=True (default) admits only calls whose effects are disjoint from |
| the batch: they may not gain, harm or produce anything an earlier member |
| gains, harms or produces. Calls on the same goal are then never run at the |
| same time (no hedged duplicates, no conflicting writers), and every observed |
| change has at most one declared owner, so feedback attribution stays exact. |
| independent=False is the plain expected-state rule. |
| """ |
| first = self.select_action() |
| batch = [first] |
| if max_parallel <= 1: |
| return batch |
| real = ObsState(self.v, self.preds, self.budget, self.t) |
| s_hat = self._expect(first.action, real.copy()) |
| left = self.budget - self.estimated_cost(first.action) |
| chosen = {first.action} |
| touched = set(self._attribution(first.action)) if independent else set() |
| t_sel = time.perf_counter() |
| while len(batch) < max_parallel: |
| best, best_val = None, 0.0 |
| for a in self.candidates: |
| if a in chosen or a in self.blocked or not self._admissible(a, real): |
| continue |
| c = self.estimated_cost(a) |
| if c > left or (conflicts is not None and any(conflicts(a, b.action) for b in batch)): |
| continue |
| if independent and self._attribution(a) & touched: |
| continue |
| val = self._net(a, s_hat) |
| if val > best_val + 1e-12: |
| best, best_val = a, val |
| if best is None: |
| break |
| g = self.predicted_gain(best, s_hat) |
| c = self.estimated_cost(best) |
| sc = Scores(Q=best_val, rank=best_val, exploit=best_val, gain=g, cost=c, bonus=0.0, continuation=0.0) |
| batch.append(Decision(step=self.t + len(batch), action=best, scores=sc, candidates=[], n_admissible=0, |
| select_time=0.0)) |
| chosen.add(best) |
| if independent: |
| touched |= self._attribution(best) |
| left -= c |
| s_hat = self._expect(best, s_hat) |
| if len(batch) > 1: |
| dt = time.perf_counter() - t_sel |
| for d in batch[1:]: |
| d.select_time = dt / (len(batch) - 1) |
| self._batch_decs = {d.action: d for d in batch} |
| return batch |
|
|
| def _expect(self, a: Action, s: ObsState) -> ObsState: |
| """Expected state after ``a``: v_i <- v_i + delta_hat_i(a | s).""" |
| d = self.predicted_effect(a, s) |
| for c, x in d.items(): |
| s.v[c] = min(1.0, max(0.0, s.v.get(c, 0.0) + x)) |
| return s |
|
|
| def _attribution(self, a: Action) -> Set[str]: |
| sem = self._sem[a] |
| out = {c for c, x in sem.gain.items() if x != 0} | {c for c, x in sem.harm.items() if x != 0} |
| out |= set(sem.produces) |
| out |= set(self.stats.observed_targets(a, "gain")) | set(self.stats.observed_targets(a, "harm")) |
| out |= set(self.stats.observed_targets(a, "produce")) |
| return out |
|
|
| def update_batch(self, actions: Sequence[Action], observation: Observation, |
| costs: Optional[Sequence[Optional[float]]] = None, |
| errors: Optional[Sequence[Optional[str]]] = None) -> List[Dict]: |
| """Record a batch executed concurrently, observed once afterwards. |
| |
| Feedback attribution: a goal or predicate that changed is credited to the batch |
| members that declare it or have observed it before; a change no member |
| targets is credited to all of them. A single-action batch is exactly ``update``. |
| """ |
| actions = list(actions) |
| costs = list(costs) if costs is not None else [None] * len(actions) |
| errors = list(errors) if errors is not None else [None] * len(actions) |
| if len(actions) == 1: |
| return [self.update(actions[0], observation, costs[0], errors[0])] |
| if not self._started: |
| raise RuntimeError("call reset(observation) first") |
| if self.status != Status.RUNNING: |
| raise ControllerStopped(self.status) |
| if len(set(actions)) != len(actions): |
| raise ValueError("an action appears twice in the batch") |
| for a in actions: |
| if a not in self._req: |
| raise KeyError(f"unknown action {a}") |
| decs = dict(getattr(self, "_batch_decs", {}) or {}) |
| if self._pending is not None: |
| decs[self._pending.action] = self._pending |
| v0, p0 = dict(self.v), dict(self.preds) |
| v1, p1 = dict(observation.values), dict(observation.predicates) |
| changed = {c for c in set(v0) | set(v1) if v1.get(c, 0.0) != v0.get(c, 0.0)} |
| changed |= {q for q in set(p0) | set(p1) if bool(p1.get(q, False)) != bool(p0.get(q, False))} |
| attr = {a: self._attribution(a) for a in actions} |
| orphan = {k for k in changed if not any(k in attr[a] for a in actions)} |
| recs = [] |
| rnd = self.t |
| for a, cost, err in zip(actions, costs, errors): |
| mine = (attr[a] & changed) | orphan |
| va = {c: (v1.get(c, 0.0) if c in mine else v0.get(c, 0.0)) for c in set(v0) | set(v1)} |
| pa = {q: (p1.get(q, False) if q in mine else p0.get(q, False)) for q in set(p0) | set(p1)} |
| cost = self._req[a].cost if cost is None else float(cost) |
| if self.cfg.feedback: |
| self.stats.update(a, v0, va, p0, pa, cost) |
| self._invalidate_key(self.stats.key(a)) |
| for c in self.weights: |
| if va.get(c, 0.0) > v0.get(c, 0.0): |
| self._gainers.setdefault(c, set()).add(a) |
| self.budget -= cost |
| self.spent += cost |
| dec = decs.get(a) |
| recs.append({ |
| "step": self.t, "round": rnd, "action": str(a), "Q": dec.scores.Q if dec else None, |
| "predicted_gain": dec.scores.gain if dec else None, |
| "continuation": dec.scores.continuation if dec else None, |
| "estimated_cost": dec.scores.cost if dec else None, |
| "observed_effect": sum(self.weights[c] * (va.get(c, 0.0) - v0.get(c, 0.0)) for c in self.weights), |
| "cost": cost, "budget_left": self.budget, "deficiency": deficiency(self.conditions, v1), |
| "values": dict(v1), "error": err, "n_actions": None, "select_time": None, |
| }) |
| self.t += 1 |
| self.v, self.preds = v1, p1 |
| self.history.extend(recs) |
| self._pending = None |
| self._rng_at_decide = None |
| self._batch_decs = {} |
| return recs |
|
|
| |
| def snapshot(self) -> Dict: |
| """JSON-serialisable episode state: goal values, predicates, budget, history, |
| online statistics, tie-breaking RNG and blocked actions. |
| |
| ``restore`` on a controller built with the same conditions, actions and |
| configuration continues with exactly the decisions the original would |
| have made. |
| """ |
| if not self._started: |
| raise RuntimeError("call reset(observation) first") |
| index = {a: i for i, a in enumerate(self.candidates)} |
| rep: Dict[Hashable, int] = {} |
| for a in self.candidates: |
| rep.setdefault(self.stats.key(a), index[a]) |
| stats = [] |
| for key, ks in self.stats._stats.items(): |
| if key not in rep: |
| continue |
| stats.append({ |
| "action": rep[key], "executions": ks.executions, "cost": [ks.cost.n, ks.cost.total], |
| **{kind: {t: [r.n, r.total] for t, r in getattr(ks, kind).items()} |
| for kind in ("gain", "harm", "produce", "raw")}, |
| }) |
| rng = self._rng_at_decide if self._rng_at_decide is not None else self._rng.getstate() |
| return { |
| "format": 1, "n_actions": len(self.candidates), "conditions": [c.name for c in self.conditions], |
| "v": dict(self.v), "predicates": dict(self.preds), "budget": self.budget, "budget0": self.budget0, |
| "spent": self.spent, "t": self.t, "status": self.status, "history": list(self.history), |
| "stats": stats, "gainers": {c: sorted(index[a] for a in acts) for c, acts in self._gainers.items()}, |
| "rng": [rng[0], list(rng[1]), rng[2]], "blocked": sorted(index[a] for a in self.blocked), |
| "n_effect_evals": self.n_effect_evals, |
| } |
|
|
| def restore(self, snap: Mapping) -> "FGAOController": |
| """Load a ``snapshot`` (see there).""" |
| from .estimator import _KeyStats, _Running |
|
|
| if snap.get("format") != 1: |
| raise ValueError("unknown snapshot format") |
| if snap["n_actions"] != len(self.candidates) or snap["conditions"] != [c.name for c in self.conditions]: |
| raise ValueError("snapshot belongs to a different action set or goal set") |
| self.reset(Observation(dict(snap["v"]), dict(snap["predicates"]))) |
| self.budget, self.budget0, self.spent = snap["budget"], snap["budget0"], snap["spent"] |
| self.t, self.status, self.history = snap["t"], snap["status"], list(snap["history"]) |
| self.stats._stats.clear() |
| for e in snap["stats"]: |
| ks = _KeyStats() |
| ks.executions = e["executions"] |
| ks.cost = _Running(*e["cost"]) |
| for kind in ("gain", "harm", "produce", "raw"): |
| table = getattr(ks, kind) |
| for t, (n, total) in e[kind].items(): |
| table[t] = _Running(n, total) |
| self.stats._stats[self.stats.key(self.candidates[e["action"]])] = ks |
| self._gainers = {c: {self.candidates[i] for i in idx} for c, idx in snap["gainers"].items()} |
| r = snap["rng"] |
| self._rng.setstate((r[0], tuple(r[1]), r[2])) |
| self.blocked = {self.candidates[i] for i in snap["blocked"]} |
| self.n_effect_evals = snap.get("n_effect_evals", 0) |
| self._invalidate() |
| return self |
|
|
| |
| def assign_weights(self, conditions) -> Dict[str, float]: |
| """assign_weights(conditions). |
| |
| Default: the task's weights. With ``optional_mass = eps``, mandatory |
| conditions share 1 - eps (in proportion to their task weights) and |
| optional conditions share eps. |
| """ |
| w = {c.name: c.weight for c in conditions} |
| eps = self.cfg.optional_mass |
| if eps is None: |
| return w |
| mand = [c.name for c in conditions if c.mandatory] |
| opt = [c.name for c in conditions if not c.mandatory] |
| if not opt: |
| return w |
| sm = sum(w[c] for c in mand) or 1.0 |
| so = sum(w[c] for c in opt) or 1.0 |
| out = {c: (1.0 - eps) * w[c] / sm for c in mand} |
| out.update({c: eps * w[c] / so for c in opt}) |
| return out |
|
|
| def _admissible(self, a: Action, state: ObsState) -> bool: |
| req = self._req[a] |
| for p in req.requires_predicates: |
| if not state.predicates.get(p, False): |
| return False |
| for c in req.requires_conditions: |
| if state.v.get(c, 0.0) < SAT_THRESHOLD: |
| return False |
| return True |
|
|
| |
| def _nu(self) -> float: |
| return self.cfg.nu if self.cfg.feedback else float("inf") |
|
|
| def _rate(self, a: Action, kind: str, target: str) -> float: |
| memo = self._rate_memo.get(a) |
| if memo is None: |
| memo = self._rate_memo[a] = {} |
| key = (kind, target) |
| r = memo.get(key) |
| if r is None: |
| r = memo[key] = self._rate_uncached(a, kind, target) |
| return r |
|
|
| def _rate_uncached(self, a: Action, kind: str, target: str) -> float: |
| sem = self._sem[a] |
| prior = {"gain": sem.gain, "harm": sem.harm, "produce": sem.produces}[kind].get(target, 0.0) |
| if not self.cfg.feedback: |
| return prior |
| n, emp = self.stats.rate(a, kind, target) |
| return blend(prior, emp, n, self._nu()) |
|
|
| def predicted_effect(self, a: Action, state: ObsState) -> Dict[str, float]: |
| """delta_hat_i(a) for every condition (eqs. 10-12).""" |
| self.n_effect_evals += 1 |
| out = {} |
| if self.cfg.estimator == "raw": |
| sem = self._sem[a] |
| n = self.stats.executions(a) if self.cfg.feedback else 0 |
| for c in self.conditions: |
| vi = state.v.get(c.name, 0.0) |
| prior = (1.0 - vi) * sem.gain.get(c.name, 0.0) - vi * sem.harm.get(c.name, 0.0) |
| emp = self.stats.rate(a, "raw", c.name)[1] if n else 0.0 |
| d = blend(prior, emp, n, self._nu()) |
| out[c.name] = max(-vi, min(1.0 - vi, d)) |
| return out |
| for c in self.conditions: |
| vi = state.v.get(c.name, 0.0) |
| g = self._rate(a, "gain", c.name) |
| h = self._rate(a, "harm", c.name) |
| out[c.name] = (1.0 - vi) * g - vi * h |
| return out |
|
|
| def predicted_gain(self, a: Action, state: ObsState) -> float: |
| """Delta_hat(a | s) = sum_i w_i delta_hat_i(a), eq. (13). |
| |
| For the rate estimator, a goal contributes an exact 0.0 unless the action |
| declares it or has a positive observation for it, so only that support is |
| summed, in condition order; the result is bit-identical to the full sum. |
| """ |
| if self.cfg.estimator != "rate": |
| d = self.predicted_effect(a, state) |
| return sum(self.weights[c] * d[c] for c in d) |
| self.n_effect_evals += 1 |
| sup = self._support.get(a) |
| if sup is None: |
| sup = self._support[a] = self._build_support(a) |
| v = state.v |
| total = 0.0 |
| for c, w, g, h in sup: |
| vi = v.get(c, 0.0) |
| total += w * ((1.0 - vi) * g - vi * h) |
| return total |
|
|
| def _build_support(self, a: Action) -> List[Tuple[str, float, float, float]]: |
| sem = self._sem[a] |
| keys = {c for c, x in sem.gain.items() if x != 0} | {c for c, x in sem.harm.items() if x != 0} |
| if self.cfg.feedback: |
| keys |= set(self.stats.observed_targets(a, "gain")) | set(self.stats.observed_targets(a, "harm")) |
| out = [] |
| for cond in self.conditions: |
| c = cond.name |
| if c in keys: |
| out.append((c, self.weights[c], self._rate(a, "gain", c), self._rate(a, "harm", c))) |
| return out |
|
|
| def estimated_cost(self, a: Action) -> float: |
| r = self._cost_memo.get(a) |
| if r is None: |
| r = self._cost_memo[a] = self._estimated_cost_uncached(a) |
| return r |
|
|
| def _estimated_cost_uncached(self, a: Action) -> float: |
| prior = self._sem[a].cost |
| if not self.cfg.feedback: |
| return prior |
| n, emp = self.stats.mean_cost(a) |
| return blend(prior, emp, n, self._nu()) |
|
|
| def uncertainty(self, a: Action, t: int) -> float: |
| n = self.stats.executions(a) |
| return math.sqrt(2.0 * math.log(t + 2) / (n + 1)) |
|
|
| def _net(self, b: Action, s: ObsState) -> float: |
| """S(b | s) = Delta_hat(b | s) - lam c_hat(b) (eq. 14).""" |
| return self.predicted_gain(b, s) - self.cfg.lam * self.estimated_cost(b) / self._cost_scale |
|
|
| |
| def _targets(self, a: Action, kind: str) -> Set[str]: |
| sem = self._sem[a] |
| decl = {"gain": sem.gain, "harm": sem.harm, "produce": sem.produces}[kind] |
| out = {k for k, x in decl.items() if x > 0} |
| if self.cfg.feedback: |
| out |= set(self.stats.observed_targets(a, kind)) |
| if self.cfg.estimator == "raw" and kind in ("gain", "harm"): |
| for c in self.stats.observed_targets(a, "raw"): |
| m = self.stats.rate(a, "raw", c)[1] |
| if (m > 0) == (kind == "gain"): |
| out.add(c) |
| return out |
|
|
| def _effect_probs(self, a: Action, state: ObsState): |
| """Independent predicted effects of ``a``: list of (kind, key, prob).""" |
| effects = [] |
| gains = set() |
| for c in self._targets(a, "gain"): |
| vi = state.v.get(c, 0.0) |
| if vi < SAT_THRESHOLD: |
| if self.cfg.estimator == "raw": |
| d = self.predicted_effect_single(a, state, c) |
| g = max(0.0, d) / (1.0 - vi) |
| else: |
| g = self._rate(a, "gain", c) |
| if g > 0: |
| effects.append(("gain", c, min(1.0, g))) |
| gains.add(c) |
| for p in self._targets(a, "produce"): |
| if not state.predicates.get(p, False): |
| q = self._rate(a, "produce", p) |
| if q > 0: |
| effects.append(("produce", p, min(1.0, q))) |
| for c in self._targets(a, "harm"): |
| vi = state.v.get(c, 0.0) |
| if vi > 0 and c not in gains: |
| if self.cfg.estimator == "raw": |
| d = self.predicted_effect_single(a, state, c) |
| h = max(0.0, -d) / vi |
| else: |
| h = self._rate(a, "harm", c) |
| if h > 0: |
| effects.append(("harm", c, min(1.0, h))) |
| return effects |
|
|
| def predicted_effect_single(self, a: Action, state: ObsState, c: str) -> float: |
| return self.predicted_effect(a, state)[c] |
|
|
| @staticmethod |
| def _apply(state: ObsState, effects, realised, harm_expected: bool = False) -> ObsState: |
| s2 = state.copy() |
| for (kind, key, prob), r in zip(effects, realised): |
| if kind == "gain" and r: |
| s2.v[key] = 1.0 |
| elif kind == "produce" and r: |
| s2.predicates[key] = True |
| elif kind == "harm": |
| if harm_expected: |
| s2.v[key] = state.v.get(key, 0.0) * (1.0 - prob) |
| elif r: |
| s2.v[key] = 0.0 |
| return s2 |
|
|
| |
| def _dependents(self, changed_conds: Iterable[str], changed_preds: Iterable[str]) -> Set[Action]: |
| """Calls whose admissibility or predicted value can change with these keys.""" |
| out: Set[Action] = set() |
| for c in changed_conds: |
| out.update(self._needs_cond.get(c, ())) |
| out.update(self._deps.get(c, ())) |
| for p in changed_preds: |
| out.update(self._needs_pred.get(p, ())) |
| return out |
|
|
| def _best_after(self, s2: ObsState, changed_conds, changed_preds) -> float: |
| """max{0, max_{b in A(s2), affordable} S(b | s2)}, reusing S(b | z_t) for unaffected b.""" |
| affected = self._dependents(changed_conds, changed_preds) |
| best = 0.0 |
| for val, b in self._S_sorted: |
| if val <= best: |
| break |
| if b in affected: |
| continue |
| if self._cost[b] <= s2.budget: |
| best = val |
| break |
| for b in affected: |
| if self._cost.get(b, self.estimated_cost(b)) > s2.budget or not self._admissible(b, s2): |
| continue |
| val = self._net(b, s2) |
| if val > best: |
| best = val |
| return best |
|
|
| def _cont_eq16(self, a: Action, state: ObsState) -> float: |
| effects = self._effect_probs(a, state) |
| s2 = self._apply(state, effects, [True] * len(effects), harm_expected=True) |
| s2.budget = state.budget - self._cost[a] |
| changed_c = [k for kind, k, _ in effects if kind in ("gain", "harm")] |
| changed_p = [k for kind, k, _ in effects if kind == "produce"] |
| return self._best_after(s2, changed_c, changed_p) |
|
|
| def _cont_eq17(self, a: Action, state: ObsState) -> float: |
| effects = self._effect_probs(a, state) |
| if len(effects) > MAX_ENUM_EFFECTS: |
| effects.sort(key=lambda e: abs(e[2] - 0.5)) |
| fixed, effects = effects[MAX_ENUM_EFFECTS:], effects[:MAX_ENUM_EFFECTS] |
| state = self._apply(state, fixed, [e[2] >= 0.5 for e in fixed]) |
| budget2 = state.budget - self._cost[a] |
| total = 0.0 |
| for realised in itertools.product((True, False), repeat=len(effects)): |
| prob = 1.0 |
| for (kind, key, q), r in zip(effects, realised): |
| prob *= q if r else (1.0 - q) |
| if prob <= 0: |
| continue |
| s2 = self._apply(state, effects, realised) |
| s2.budget = budget2 |
| changed_c = [k for (kind, k, _), r in zip(effects, realised) if r and kind in ("gain", "harm")] |
| changed_p = [k for (kind, k, _), r in zip(effects, realised) if r and kind == "produce"] |
| total += prob * self._best_after(s2, changed_c, changed_p) |
| return total |
|
|
| |
| def _success_state(self, a: Action, state: ObsState): |
| s2 = state.copy() |
| p_pred: Dict[str, float] = {} |
| p_cond: Dict[str, float] = {} |
| harmed: List[str] = [] |
| for kind, key, q in self._effect_probs(a, state): |
| if kind == "produce": |
| s2.predicates[key] = True |
| p_pred[key] = q |
| elif kind == "gain": |
| s2.v[key] = 1.0 |
| p_cond[key] = q |
| elif kind == "harm" and self.cfg.corrective: |
| s2.v[key] = state.v.get(key, 0.0) * (1.0 - q) |
| harmed.append(key) |
| s2.budget = state.budget - self.estimated_cost(a) |
| s2.step = state.step + 1 |
| return s2, p_pred, p_cond, harmed |
|
|
| def continuation(self, a: Action, state: ObsState, depth: int) -> float: |
| """Unlock form: max_{b in Next(a)} P(unlock b | a) [S(b | a)]_+ (+ repairs).""" |
| if depth <= 0: |
| return 0.0 |
| s2, p_pred, p_cond, harmed = self._success_state(a, state) |
| if not p_pred and not p_cond and not harmed: |
| return 0.0 |
| unlock = set() |
| for p in p_pred: |
| unlock.update(self._needs_pred.get(p, ())) |
| for c in p_cond: |
| unlock.update(self._needs_cond.get(c, ())) |
| best = 0.0 |
| for b in unlock: |
| if b == a or self._admissible(b, state) or not self._admissible(b, s2): |
| continue |
| if self.estimated_cost(b) > s2.budget: |
| continue |
| req = self._req[b] |
| prob = 1.0 |
| for p in req.requires_predicates: |
| if p in p_pred: |
| prob *= p_pred[p] |
| for c in req.requires_conditions: |
| if c in p_cond: |
| prob *= p_cond[c] |
| val = self._net(b, s2) |
| if depth > 1: |
| val += self.cfg.gamma * max(0.0, self.continuation(b, s2, depth - 1)) |
| best = max(best, prob * max(0.0, val)) |
| if harmed: |
| repairs = set() |
| for c in harmed: |
| repairs.update(self._gainers.get(c, ())) |
| for b in repairs: |
| if b == a or not self._admissible(b, s2) or self.estimated_cost(b) > s2.budget: |
| continue |
| now = max(0.0, self._net(b, state)) if self._admissible(b, state) else 0.0 |
| best = max(best, max(0.0, self._net(b, s2)) - now) |
| return best |
|
|
| |
| def _begin_step(self, actions: List[Action], state: ObsState) -> None: |
| """Per-step caches: costs, S(b | z_t), and which conditions each b depends on.""" |
| self._cost = {a: self.estimated_cost(a) for a in self.candidates} |
| self._S = {a: self._net(a, state) for a in actions} |
| self._S_sorted = sorted(((v, a) for a, v in self._S.items()), key=lambda x: -x[0]) |
| deps: Dict[str, Set[Action]] = {} |
| if self.cfg.lookahead in ("eq16", "eq17"): |
| for b in self.candidates: |
| for c in self._targets(b, "gain") | self._targets(b, "harm"): |
| deps.setdefault(c, set()).add(b) |
| if self.cfg.estimator == "raw": |
| sem = self._sem[b] |
| for c in set(sem.gain) | set(sem.harm): |
| deps.setdefault(c, set()).add(b) |
| self._deps = deps |
|
|
| def score(self, a: Action, state: ObsState) -> Dict[str, float]: |
| gain = self.predicted_gain(a, state) |
| cost = self._cost[a] if hasattr(self, "_cost") and a in self._cost else self.estimated_cost(a) |
| immediate = gain - self.cfg.lam * cost / self._cost_scale |
| la = self.cfg.lookahead |
| if la == "none" or (la == "unlock" and self.cfg.lookahead_depth <= 0): |
| cont = 0.0 |
| elif la == "unlock": |
| cont = self.continuation(a, state, self.cfg.lookahead_depth) |
| elif la == "eq16": |
| cont = self._cont_eq16(a, state) |
| else: |
| cont = self._cont_eq17(a, state) |
| exploit = immediate + self.cfg.gamma * max(0.0, cont) |
| unc = self.uncertainty(a, state.step) if self.cfg.beta > 0 else 0.0 |
| q = exploit + self.cfg.beta * unc |
| if self.cfg.objective == "ratio": |
| rank = (gain + self.cfg.gamma * max(0.0, cont)) / max(cost, 1e-9) |
| else: |
| rank = q |
| return {"Q": q, "rank": rank, "exploit": exploit, "gain": gain, "cost": cost, |
| "bonus": unc, "continuation": cont} |
|
|
|
|
