File size: 11,643 Bytes
e187916 e2f048d e187916 e2f048d 59022ca e187916 59022ca 57356d8 e2f048d 57356d8 e187916 57356d8 e2f048d 59022ca b038669 e187916 b038669 59022ca b038669 59022ca b038669 e187916 b038669 e187916 b038669 e187916 b038669 e187916 b038669 e2f048d b038669 e187916 b038669 59022ca b038669 e187916 b038669 e2f048d b038669 e187916 b038669 e187916 b038669 e187916 b038669 57356d8 b038669 e187916 57356d8 59022ca e187916 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 | """The harness itself: observe -> think -> act -> verify, bounded and traced."""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any, Callable
from tiny_computer_agent.act import ActionResult, DryRunExecutor, Executor, PyAutoGuiExecutor
from tiny_computer_agent.actions import Action
from tiny_computer_agent.brain import Brain, GoalUnreachable
from tiny_computer_agent.observe import Observation, ScreenSource
from tiny_computer_agent.gates import Approval, Gate
from tiny_computer_agent.safety import Policy, PolicyViolation
@dataclass(frozen=True)
class Step:
"""One turn of the loop, with everything needed to audit it afterwards."""
index: int
observation: Observation
action: Action | None = None
result: ActionResult | None = None
error: str | None = None
@property
def ok(self) -> bool:
return self.error is None and (self.result is None or self.result.ok)
def line(self) -> str:
bits = [f"step {self.index}"]
if self.action is not None:
bits.append(self.action.describe())
if self.error:
bits.append(f"ERROR {self.error}")
elif self.result is not None:
bits.append(self.result.summary())
return " | ".join(bits)
@dataclass
class Episode:
"""An append-only record of one run: the unit you replay and score."""
goal: str
steps: list[Step] = field(default_factory=list)
stop_reason: str = ""
success: bool = False
def to_rows(self) -> list[dict[str, Any]]:
"""Rows for a dataframe / notebook table."""
rows = []
for step in self.steps:
rows.append({
"step": step.index,
"elements": len(step.observation.elements),
"action": step.action.describe() if step.action else "",
"outcome": step.error or (step.result.summary() if step.result else ""),
})
return rows
def render(self) -> str:
lines = [
f"goal: {self.goal}",
f"result: {'success' if self.success else 'failure'} - {self.stop_reason}",
]
lines += [f" {step.line()}" for step in self.steps]
return "\n".join(lines)
class AgentLoop:
"""Wires a screen, a brain, an executor and a policy into an agent.
Guarantees, by construction rather than by prompting:
* every action is one of six validated verbs
* nothing executes until policy **and** any gate have passed it
* the episode ends on a step budget, a kill switch, a brain error, or a
failed screen capture (reported as a step, never as a traceback)
* the screen is re-read after every action, so the next decision sees the
consequence of the last one
"""
def __init__(self, *, screen: ScreenSource, brain: Brain,
executor: Executor | None = None, policy: Policy | None = None,
on_step: Callable[[Step], None] | None = None,
max_repeats: int = 1,
verify: Callable[[Observation], bool] | None = None,
gate: Gate | None = None) -> None:
"""``verify`` runs after every action: the harness, not the model, decides
whether the goal was reached. A brain saying "done" is a claim; a verifier
looking at the screen is evidence."""
self.screen = screen
self.brain = brain
self.policy = policy or Policy()
if executor is None:
executor = DryRunExecutor() if self.policy.dry_run else PyAutoGuiExecutor()
self.executor = executor
self.on_step = on_step
self.verify = verify
self.gate = gate
self.max_repeats = max_repeats
# the stepwise state: an episode in progress, and the guards around it
self.episode: Episode | None = None
self.observation: Observation | None = None
self.paused = False
self._done = False
self._budget = self.policy.step_budget
self._turns = 0
self._previous: tuple | None = None
self._repeats = 0
# -- stepwise control -------------------------------------------------- #
def begin(self, goal: str, *, max_steps: int | None = None) -> "AgentLoop":
"""Start a new episode: read the screen once, and reset the guards.
The stepwise half of the API. :meth:`run` is the batch form and is built
from exactly this; a console calls ``begin`` then :meth:`step_once`, so one
button press can advance exactly one turn.
"""
budget = max_steps or self.policy.step_budget
self.episode = Episode(goal=goal, stop_reason=f"step budget of {budget} exhausted")
self._budget = budget
self._turns = 0
self._previous = None
self._repeats = 0
self._done = False
self.paused = False
try:
self.observation = self.screen.capture()
except Exception as exc:
self.observation = None
self._finish(f"capture failed: {type(exc).__name__}: {exc}")
return self
@property
def finished(self) -> bool:
"""True once this episode has ended - success, failure, budget or switch."""
return self._done
def pause(self) -> "AgentLoop":
"""Stop between steps: :meth:`run` returns and the episode stays open."""
self.paused = True
return self
def resume(self) -> "AgentLoop":
self.paused = False
return self
def step_once(self) -> Step | None:
"""One observe -> think -> act -> verify turn.
Returns the step it recorded, or ``None`` once the episode is over, so a
caller can loop on it without knowing any of the stopping rules.
"""
if self.episode is None or self._done or self.paused:
return None
episode, observation = self.episode, self.observation
if self._turns >= self._budget:
self._done = True
return None
if self.policy.should_stop():
self._finish("kill switch engaged")
return None
if observation is None: # a screen that never worked
self._done = True
return None
index, self._turns = self._turns, self._turns + 1
try:
action = self.brain.decide(episode.goal, observation)
except GoalUnreachable as exc:
return self._record_step(Step(index, observation, error=f"gave up: {exc}"),
f"brain gave up: {exc}")
except Exception as exc:
return self._record_step(
Step(index, observation, error=f"brain failed: {type(exc).__name__}: {exc}"),
"brain error")
if action.kind == "done":
episode.success = True
return self._record_step(
Step(index, observation, action=action,
result=ActionResult(action=action, ok=True, note="goal reported met")),
f"brain reported done: {action.reason or 'no reason given'}")
try:
self.policy.check(action, observation)
except PolicyViolation as exc:
return self._record_step(Step(index, observation, action=action,
error=f"refused: {exc}"),
f"policy refusal: {exc}")
if self.gate is not None: # the operator's yes, before anything moves
approval: Approval = self.gate.approve(action, observation)
if not approval.approved:
return self._record_step(
Step(index, observation, action=action,
error=f"not approved: {approval.reason}"),
f"operator refused: {approval.reason}")
signature = (self._fingerprint(observation), action.describe())
self._repeats = self._repeats + 1 if signature == self._previous else 0
self._previous = signature
if self._repeats > self.max_repeats:
return self._record_step(
Step(index, observation, action=action,
error=f"no progress: repeating {action.describe()} "
f"against an unchanged screen"),
f"no progress after {self._repeats} identical actions on an unchanged "
f"screen ({action.kind} {action.label or action.text or ''})".strip())
try:
result = self.executor.execute(action, observation)
except Exception as exc:
return self._record_step(
Step(index, observation, action=action,
error=f"executor failed: {type(exc).__name__}: {exc}"),
"executor error")
step = Step(index, observation, action=action, result=result)
self._append(step)
if not result.ok:
self._finish(f"action failed: {result.note}")
return step
try:
self.observation = self.screen.capture() # verify: what did that change?
except Exception as exc:
self._append(Step(index + 1, observation,
error=f"capture failed: {type(exc).__name__}: {exc}"))
self._finish("perception failed mid-episode")
return step
if self.verify is not None and self._verified(self.observation):
episode.success = True
self._finish(f"verified after {self._turns} action(s)")
return step
def run(self, goal: str | None = None, *, max_steps: int | None = None) -> Episode:
"""Run to the end of the episode - or continue one that was paused.
Guarantees, by construction rather than by prompting:
* every action is one of six validated verbs
* nothing executes until policy **and** any gate have passed it
* the episode ends on a step budget, a kill switch, a brain error, or a
failed screen capture (reported as a step, never as a traceback)
* the screen is re-read after every action, so the next decision sees the
consequence of the last one
"""
if self.episode is None or self._done or (goal is not None and goal != self.episode.goal):
self.begin(goal or "", max_steps=max_steps)
while not self.paused and not self._done:
if self.step_once() is None:
break
return self.episode
def _append(self, step: Step) -> None:
self.episode.steps.append(step)
if self.on_step is not None:
self.on_step(step)
def _record_step(self, step: Step, stop_reason: str) -> Step:
self._append(step)
self._finish(stop_reason)
return step
def _finish(self, stop_reason: str) -> None:
self.episode.stop_reason = stop_reason
self._done = True
def _verified(self, observation: Observation) -> bool:
"""A broken verifier must never fake a success."""
try:
return bool(self.verify(observation))
except Exception:
return False
@staticmethod
def _fingerprint(observation: Observation) -> tuple:
"""What the agent can see: enough to tell "nothing changed" from "it worked"."""
return (tuple(element.label for element in observation.elements),
observation.text[:200])
def _record(self, episode: Episode, step: Step) -> None:
episode.steps.append(step)
if self.on_step is not None:
self.on_step(step)
|