"""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 # eq. 17: at most 2^6 outcomes per action; extra effects fixed at their mode @dataclass class FGAOConfig: """Controller hyperparameters. Defaults are the evaluated v2 RAG configuration.""" nu: float = 4.0 # prior strength: rho = nu / (nu + n) lam: float = 0.1 # cost weight, in goal units per whole budget gamma: float = 0.9 # continuation weight lookahead: str = "unlock" # "unlock" | "eq16" | "eq17" | "none" lookahead_depth: int = 1 # unlock form only; 0 disables lookahead corrective: bool = True # unlock form: credit repair actions after predicted harm similarity: str = "task" # feedback pooling: "task" (declared groups) | "exact" | "type" estimator: str = "rate" # "rate" (state-normalised gain/harm rates) | "raw" feedback: bool = True # False: priors only (rho = 1) objective: str = "additive" # "additive" | "ratio" (ranking only) beta: float = 0.0 # optional exploration bonus weight optional_mass: Optional[float] = None # weight share of optional goals (None: declared weights) cost_scale: Optional[float] = None # B0; default: the initial budget 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: # actions sharing feedback: one update changes only their estimates 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]]] = {} # ================================================================ step API 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: # eq. (20) 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 # type: ignore[return-value] 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 # ======================================================== parallel calls 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 # ======================================================= checkpointing 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 # ============================================================ v2 algorithm 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 # -------------------------------------------------------------- estimates 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)) # keep v + delta in [0, 1] 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 # ----------------------------------------------- predicted effects of a call 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 # ----------------------------------------- continuation: eq. (15)-(17) 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: # descending 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: # keep the most uncertain, fix the rest at their mode 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 # unlock-form (earlier specification, eq. 7) 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 # -------------------------------------------------------------- scoring 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 # S_t(a), eq. (14) 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) # eq. (16)/(18) 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}