File size: 8,443 Bytes
e187916 59022ca e187916 59022ca 57356d8 e187916 57356d8 59022ca e187916 59022ca e187916 59022ca e187916 59022ca e187916 57356d8 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 | """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.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 has 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) -> 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.max_repeats = max_repeats
def run(self, goal: str, *, max_steps: int | None = None) -> Episode:
budget = max_steps or self.policy.step_budget
episode = Episode(goal=goal, stop_reason=f"step budget of {budget} exhausted")
# A small model that repeats an action against an unchanged screen is stuck,
# not persistent. Tolerate `max_repeats` such repeats, then stop - burning the
# whole budget on the same click teaches the user nothing.
previous: tuple[tuple, str] | None = None
repeats = 0
try:
observation = self.screen.capture()
except Exception as exc:
episode.stop_reason = f"capture failed: {type(exc).__name__}: {exc}"
return episode
for index in range(budget):
if self.policy.should_stop():
episode.stop_reason = "kill switch engaged"
break
try:
action = self.brain.decide(goal, observation)
except GoalUnreachable as exc:
self._record(episode, Step(index, observation, error=f"gave up: {exc}"))
episode.stop_reason = f"brain gave up: {exc}"
break
except Exception as exc:
self._record(episode, Step(index, observation,
error=f"brain failed: {type(exc).__name__}: {exc}"))
episode.stop_reason = "brain error"
break
if action.kind == "done":
self._record(episode, Step(
index, observation, action=action,
result=ActionResult(action=action, ok=True, note="goal reported met"),
))
episode.success = True
episode.stop_reason = f"brain reported done: {action.reason or 'no reason given'}"
break
try:
self.policy.check(action, observation)
except PolicyViolation as exc:
self._record(episode, Step(index, observation, action=action,
error=f"refused: {exc}"))
episode.stop_reason = f"policy refusal: {exc}"
break
signature = (self._fingerprint(observation), action.describe())
repeats = repeats + 1 if signature == previous else 0
previous = signature
if repeats > self.max_repeats:
self._record(episode, Step(
index, observation, action=action,
error=f"no progress: repeating {action.describe()} "
f"against an unchanged screen",
))
episode.stop_reason = (
f"no progress after {repeats} identical actions on an unchanged "
f"screen ({action.kind} {action.label or action.text or ''})".strip()
)
break
try:
result = self.executor.execute(action, observation)
except Exception as exc:
self._record(episode, Step(index, observation, action=action,
error=f"executor failed: {type(exc).__name__}: {exc}"))
episode.stop_reason = "executor error"
break
self._record(episode, Step(index, observation, action=action, result=result))
if not result.ok:
episode.stop_reason = f"action failed: {result.note}"
break
try:
observation = self.screen.capture() # verify: what did that change?
except Exception as exc:
self._record(episode, Step(index + 1, observation,
error=f"capture failed: {type(exc).__name__}: {exc}"))
episode.stop_reason = "perception failed mid-episode"
break
if self.verify is not None and self._verified(observation):
episode.success = True
episode.stop_reason = f"verified after {index + 1} action(s)"
break
return episode
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)
|