Download showcase_catalog.py from hkblockchaintechnology/tiny-computer-agent: direct link, hf CLI and curl.
- Browser
- Download file 77.5 kB
-
https://huggingface.co/hkblockchaintechnology/tiny-computer-agent/resolve/main/showcase_catalog.py
- Command line
-
hf download hf://hkblockchaintechnology/tiny-computer-agent/showcase_catalog.py
-
curl -L -o showcase_catalog.py https://huggingface.co/hkblockchaintechnology/tiny-computer-agent/resolve/main/showcase_catalog.py
77.5 kB
| """The capability catalogue: what the harness can do, and the proof of each one. | |
| Every entry names the public callables it demonstrates, so the gallery in the model | |
| card can promise that nothing in the API is left without a picture. | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| from showcase_core import ( | |
| CATALOG, PY, ROOT, SCENES, Shot, annotate, close_scene, crop, grab, open_scene, | |
| overlay_lines, side_by_side, | |
| ) | |
| import os | |
| import subprocess | |
| import time | |
| from pathlib import Path | |
| def term(sid, category, title, shows, note="", argv=None, code=None, env=None, | |
| prepare=None): | |
| CATALOG.append(Shot(id=sid, category=category, title=title, shows=shows, note=note, | |
| kind="terminal", argv=argv, code=code, env=env, prepare=prepare)) | |
| def desk(sid, category, title, shows, note="", scene_name=None, make=None): | |
| CATALOG.append(Shot(id=sid, category=category, title=title, shows=shows, note=note, | |
| kind="desktop", scene=scene_name, make=make)) | |
| def cli(*flags, goal=None): | |
| argv = [PY, "-m", "tiny_computer_agent"] | |
| if goal: | |
| argv.append(goal) | |
| return argv + list(flags) | |
| # =========================================================================== # | |
| # 1. the closed action vocabulary | |
| # =========================================================================== # | |
| term("vocab-verbs", "Action vocabulary", | |
| "The six verbs, and only six", | |
| ["Action", "parse_action", "ACTION_KINDS", "ACTION_MENU", "Action.describe"], | |
| note="Every action is validated on construction, before anything touches the machine.", | |
| code=""" | |
| from tiny_computer_agent import ACTION_KINDS, parse_action | |
| print("the whole vocabulary:", ", ".join(ACTION_KINDS)) | |
| print() | |
| for payload in [ | |
| {"kind": "click", "label": "Save"}, | |
| {"kind": "click", "x": 640, "y": 400}, | |
| {"kind": "type", "text": "report"}, | |
| {"kind": "key", "keys": "ctrl+s"}, | |
| {"kind": "scroll", "amount": -3}, | |
| {"kind": "wait", "seconds": 1.5}, | |
| {"kind": "done", "reason": "the file is saved"}, | |
| ]: | |
| action = parse_action(dict(payload, reason="demonstration")) | |
| print(f" parsed -> {action.describe()}") | |
| """) | |
| term("vocab-refusals", "Action vocabulary", | |
| "Malformed actions are refused, not guessed at", | |
| ["ActionError", "Action", "parse_action"], | |
| note="An unknown verb or a missing target raises before the machine is touched.", | |
| code=""" | |
| from tiny_computer_agent import ActionError, parse_action | |
| print("every one of these is refused, with the reason why:") | |
| for payload in [ | |
| {"kind": "format_disk"}, # not a verb we have | |
| {"kind": "click"}, # click with no target | |
| {"kind": "click", "label": "Save", "x": -5, "y": 2}, # negative pixel | |
| {"kind": "type"}, # type with no text | |
| {"kind": "key"}, # key with no chord | |
| {"kind": "scroll", "amount": 0}, # scroll that scrolls nowhere | |
| {"kind": "wait", "seconds": 0}, # wait that does not wait | |
| ]: | |
| try: | |
| parse_action(payload) | |
| print(f" ACCEPTED (bug!): {payload}") | |
| except ActionError as exc: | |
| print(f" {str(payload)[:38]:40} -> {exc}") | |
| """) | |
| term("vocab-menu", "Action vocabulary", | |
| "The contract handed to a model", | |
| ["ACTION_MENU"], | |
| note="The prompt-side menu: one JSON object, six shapes, no invented verbs.", | |
| code=""" | |
| from tiny_computer_agent import ACTION_MENU | |
| print(ACTION_MENU) | |
| """) | |
| # =========================================================================== # | |
| # 2. menus and lenient parsing | |
| # =========================================================================== # | |
| term("menu-numbered", "Decisions", | |
| "A short numbered menu, in reading order", | |
| ["DecisionMenu", "DecisionMenu.render", "MenuItem", "MenuItem.label"], | |
| note="A 256M model cannot emit coordinates; it can pick a number from twelve.", | |
| code=""" | |
| from tiny_computer_agent import DecisionMenu, Element, Observation | |
| frame = Observation(elements=( | |
| Element("Dialog", x=40, y=20, width=90, height=24), | |
| Element("File name", x=200, y=95, width=120, height=22), | |
| Element("Cancel", x=400, y=270, width=110, height=30), | |
| Element("Save", x=200, y=270, width=110, height=30), | |
| ), text="Untitled document - unsaved changes") | |
| menu = DecisionMenu(frame) | |
| print(f"{len(menu)} numbered choices, sorted top-to-bottom then left-to-right:") | |
| print(menu.render()) | |
| print() | |
| print("capped at 3 for a small context window:") | |
| print(DecisionMenu(frame, max_items=3).render()) | |
| """) | |
| term("menu-lookup", "Decisions", | |
| "Menu lookup tolerates what models actually send", | |
| ["DecisionMenu.item", "DecisionMenu.find", "DecisionMenu.mentioned_in"], | |
| note="1, \"2\", 2.0 and \"Save.\" all resolve; a stray word resolves to nothing.", | |
| code=""" | |
| from tiny_computer_agent import DecisionMenu, Element, Observation | |
| frame = Observation(elements=( | |
| Element("Files", x=10, y=6, width=60, height=20), | |
| Element("Save .", x=200, y=270, width=90, height=30), # OCR kept the full stop | |
| Element("Cancel", x=400, y=270, width=90, height=30), | |
| )) | |
| menu = DecisionMenu(frame) | |
| for probe in [1, "2", 2.0, "2 ", 0, 99, "two"]: | |
| item = menu.item(probe) | |
| print(f" item({probe!r:6}) -> {item.label if item else 'no such item'}") | |
| print() | |
| for label in ["save", "SAVE", "Save .", "nothing here"]: | |
| found = menu.find(label) | |
| print(f" find({label!r:15}) -> {found.label if found else 'not on screen'}") | |
| print() | |
| print("mentioned_in('click Save or Cancel'):", | |
| [i.label for i in menu.mentioned_in("click Save or Cancel")]) | |
| """) | |
| term("decision-json", "Decisions", | |
| "Reading a decision out of a small model's reply", | |
| ["parse_decision", "extract_json_object", "extract_json_objects", "DecisionMenu"], | |
| note="The LAST JSON object wins, because a model that echoes its prompt answers last.", | |
| code=""" | |
| from tiny_computer_agent import DecisionMenu, Element, Observation, extract_json_objects, parse_decision | |
| frame = Observation(elements=( | |
| Element("Files", x=10, y=6, width=60, height=20), | |
| Element("Save", x=200, y=270, width=90, height=30), | |
| Element("Cancel", x=400, y=270, width=90, height=30), | |
| )) | |
| reply = 'Menu: 1. Files 2. Save 3. Cancel\nAnswer: {"kind": "click", "index": 2}' | |
| print("reply:", reply.replace(chr(10), " | "), chr(10)) | |
| print("every JSON object found, in order:", extract_json_objects(reply)) | |
| print("the decision taken:", parse_decision(reply, frame).describe()) | |
| print() | |
| for reply in ["2", "Click on \"Save\" in the dialogue.", "The goal is already done."]: | |
| print(f" {reply!r:42} -> {parse_decision(reply, frame).describe()}") | |
| """) | |
| term("decision-echo-guard", "Decisions", | |
| "An echoed prompt is refused; a right answer in the wrong field is recovered", | |
| ["parse_decision", "BrainError", "DecisionMenu.mentioned_in"], | |
| note="Real 256M/0.5B failure modes, both caught: no blind click either way.", | |
| code=""" | |
| from tiny_computer_agent import BrainError, DecisionMenu, Element, Observation, parse_decision | |
| frame = Observation(elements=( | |
| Element("Files", x=10, y=6, width=60, height=20), | |
| Element("Save", x=200, y=270, width=90, height=30), | |
| Element("Cancel", x=400, y=270, width=90, height=30), | |
| )) | |
| echoed = DecisionMenu(frame).render() # the model read the menu back to us | |
| try: | |
| parse_decision(echoed, frame) | |
| print("echo ACCEPTED (bug!)") | |
| except BrainError as exc: | |
| print("echo refused:", exc) | |
| print() | |
| print("a 0.5B model really did answer like this - correct answer in the wrong field:") | |
| salvaged = parse_decision('{"kind": "click", "index": 0, "reason": "Save"}', frame) | |
| print(" recovered ->", salvaged.describe()) | |
| print() | |
| try: | |
| parse_decision('{"kind": "click", "index": 42}', frame) | |
| except BrainError as exc: | |
| print("an index outside the menu is refused:", exc) | |
| try: | |
| parse_decision('{"kind": "click", "label": "Format disk"}', frame) | |
| except BrainError as exc: | |
| print("a label that is not on screen is refused:", exc) | |
| """) | |
| term("decision-garbled", "Decisions", | |
| "A garbled verb with a valid label is still usable", | |
| ["parse_decision", "BrainError"], | |
| note="A 256M model really does emit {\"kind\": \"file\", \"label\": \"Save\"}.", | |
| code=""" | |
| from tiny_computer_agent import Element, Observation, parse_decision | |
| frame = Observation(elements=(Element("Save", x=200, y=270, width=90, height=30),)) | |
| print('reply: {"kind": "file", "label": "Save"} <- not a verb we have') | |
| print(" ->", parse_decision('{"kind": "file", "label": "Save"}', frame).describe()) | |
| """) | |
| # =========================================================================== # | |
| # 3. execution: the only two places that touch the machine | |
| # =========================================================================== # | |
| term("exec-dry-run", "Execution", | |
| "Dry run resolves every click to pixels - and touches nothing", | |
| ["DryRunExecutor", "DryRunExecutor.execute", "resolve_point", "ActionResult"], | |
| note="The default everywhere: read a whole episode before anything clicks.", | |
| code=""" | |
| from tiny_computer_agent import Action, DryRunExecutor, Element, Observation | |
| frame = Observation(elements=(Element("Save", x=200, y=270, width=110, height=30),)) | |
| executor = DryRunExecutor() | |
| for action in [Action("click", label="Save"), | |
| Action("click", label="Nowhere"), | |
| Action("type", text="report.txt"), | |
| Action("key", keys="ctrl+s"), | |
| Action("scroll", amount=-3), | |
| Action("wait", seconds=0.5), | |
| Action("done", reason="finished")]: | |
| result = executor.execute(action, frame) | |
| print(f" {result.summary()}") | |
| print() | |
| print("actions the executor *performed* on the machine:", executor.performed) | |
| print("the log it kept instead:") | |
| for line in executor.log: | |
| print(" ", line) | |
| """) | |
| term("exec-refuses-unresolved", "Execution", | |
| "Real input refuses to guess a pixel for a label it cannot find", | |
| ["PyAutoGuiExecutor", "PyAutoGuiExecutor.execute", "ActionResult.summary"], | |
| note="This is the gate between a hallucinated label and 'Delete account'.", | |
| code=""" | |
| from tiny_computer_agent import Action, Element, Observation, PyAutoGuiExecutor | |
| frame = Observation(elements=(Element("Save", x=200, y=270, width=110, height=30),)) | |
| executor = PyAutoGuiExecutor() | |
| result = executor.execute(Action("click", label="Delete account"), frame) | |
| print("a label that is not on screen:") | |
| print(" ", result.summary()) | |
| print(" ok =", result.ok, " - nothing was clicked") | |
| print() | |
| resolved = executor.execute(Action("done", reason="nothing to do"), frame) | |
| print("a verb that needs no target passes straight through:") | |
| print(" ", resolved.summary()) | |
| """) | |
| # =========================================================================== # | |
| # 4. the loop | |
| # =========================================================================== # | |
| term("loop-end-to-end", "The loop", | |
| "observe -> think -> act -> verify, traced end to end", | |
| ["AgentLoop", "AgentLoop.run", "Episode", "Episode.render", "Episode.to_rows", "Step", "Step.line", "Step.ok"], | |
| note="One call runs a whole episode; every step is kept for audit.", | |
| code=""" | |
| from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain | |
| screen = FakeScreen([ | |
| (Element("Terminal", x=0, y=0, width=70, height=24), Element("Files", x=80, y=0, width=70, height=24)), | |
| (Element("Name", x=100, y=120, width=200, height=24), Element("Save", x=320, y=120, width=80, height=24)), | |
| (Element("Save", x=320, y=120, width=80, height=24),), | |
| ], text="Untitled document - unsaved changes") | |
| episode = AgentLoop( | |
| screen=screen, | |
| brain=MockBrain([ | |
| {"kind": "click", "label": "Save"}, | |
| {"kind": "type", "text": "report"}, | |
| {"kind": "done", "reason": "the document is saved"}, | |
| ]), | |
| ).run("save the report") | |
| print(episode.render()) | |
| print() | |
| if episode.success: | |
| print("the same episode as rows for a table:") | |
| for row in episode.to_rows(): | |
| print(" ", row) | |
| """) | |
| term("loop-budget", "The loop", | |
| "A hard step budget, so a run cannot wander", | |
| ["Policy", "Policy.step_budget", "AgentLoop.run"], | |
| note="The budget ends the episode and says so; nothing executes past it.", | |
| code=""" | |
| from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain, Policy | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]), | |
| brain=MockBrain([{"kind": "wait", "seconds": 0.2 + i} for i in range(10)]), | |
| policy=Policy(step_budget=3), | |
| ).run("click Save") | |
| print(episode.render()) | |
| print() | |
| print("steps taken:", len(episode.steps), "| success:", episode.success) | |
| """) | |
| term("loop-no-progress-guard", "The loop", | |
| "Repeating one action against an unchanged screen is stopped", | |
| ["AgentLoop.run", "Step", "Step.error"], | |
| note="A stuck agent is not a persistent one - it is stopped and told why.", | |
| code=""" | |
| from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain, Policy | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]), | |
| brain=MockBrain([{"kind": "click", "label": "Save"}] * 10), | |
| policy=Policy(step_budget=10), | |
| ).run("click Save") | |
| print(episode.render()) | |
| print() | |
| print("stopped after", len(episode.steps), "steps instead of burning all 10:") | |
| print(" ", episode.steps[-1].error) | |
| """) | |
| term("loop-kill-switch", "The loop", | |
| "A kill-switch file aborts before the next action", | |
| ["Policy.kill_switch_path", "Policy.should_stop", "AgentLoop.run"], env={"TCA_DEMO": "1"}, | |
| note="Checked before every single action: create the file, the run stops.", | |
| code=""" | |
| import os, tempfile | |
| from pathlib import Path | |
| from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain, Policy | |
| marker = Path(tempfile.mkdtemp()) / "tiny-computer-agent.STOP" | |
| marker.write_text("stop") | |
| print("kill switch file:", marker, "- present:", marker.exists()) | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]), | |
| brain=MockBrain([{"kind": "click", "label": "Save"}]), | |
| policy=Policy(kill_switch_path=str(marker)), | |
| ).run("click Save") | |
| print() | |
| print("steps taken:", episode.steps, "(none)") | |
| print("stop reason:", episode.stop_reason) | |
| """) | |
| term("loop-survives-failures", "The loop", | |
| "A dead screen or a broken brain is an episode, not a traceback", | |
| ["AgentLoop.run", "Episode.stop_reason", "GoalUnreachable", "BrainError"], | |
| note="Every failure path is recorded as a failed step with a reason.", | |
| code=""" | |
| from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain, PlannerBrain, Policy, GoalUnreachable | |
| class Blind: | |
| def capture(self): | |
| raise RuntimeError("X connection failed") | |
| class Exploding: | |
| def decide(self, goal, observation): | |
| raise RuntimeError("the model died") | |
| print("a capture that fails:") | |
| print(" ", AgentLoop(screen=Blind(), brain=MockBrain([{"kind": "done"}])).run("anything").stop_reason) | |
| print("a brain that crashes:") | |
| print(" ", AgentLoop(screen=FakeScreen([(Element("Save"),)]), brain=Exploding()).run("save").stop_reason) | |
| print("a brain that gives up (a failure, never a success):") | |
| episode = AgentLoop(screen=FakeScreen([(Element("Files"),)]), brain=PlannerBrain(max_waits=0)).run("click 'Save'") | |
| print(" success:", episode.success, "|", episode.stop_reason) | |
| """) | |
| term("loop-verifier", "The loop", | |
| "Success is evidence, not the model's word", | |
| ["AgentLoop.run", "Episode.success", "Observation.text"], | |
| note="Without a verifier, a successful action that removes its own target looks like a failure.", | |
| code=""" | |
| from tiny_computer_agent import AgentLoop, Element, MockBrain, Observation, Policy | |
| class ScriptedTextScreen: | |
| def __init__(self, texts): self.texts, self.n = list(texts), 0 | |
| def capture(self) -> Observation: | |
| text = self.texts[min(self.n, len(self.texts) - 1)]; self.n += 1 | |
| return Observation(index=self.n - 1, elements=(Element("Save"),), text=text) | |
| waits = [{"kind": "wait", "seconds": 0.1}] * 4 | |
| screen = lambda: ScriptedTextScreen(["Unsaved changes", "before", "Saved"]) | |
| without = AgentLoop(screen=screen(), brain=MockBrain(waits), policy=Policy(step_budget=3)).run("save") | |
| print("no verifier ->", without.success, "|", without.stop_reason) | |
| with_verifier = AgentLoop( | |
| screen=screen(), brain=MockBrain(waits), policy=Policy(step_budget=4), | |
| verify=lambda observation: "saved" in observation.text.lower(), | |
| ).run("save") | |
| print("with verifier ->", with_verifier.success, "|", with_verifier.stop_reason) | |
| """) | |
| term("loop-on-step", "The loop", | |
| "Every step can be streamed out as it happens", | |
| ["AgentLoop.run", "Step.line", "on_step"], | |
| note="This is the hook a live console, a progress bar or a websocket uses.", | |
| code=""" | |
| from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain | |
| seen = [] | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]), | |
| brain=MockBrain([{"kind": "click", "label": "Save"}, {"kind": "done", "reason": "saved"}]), | |
| on_step=lambda step: seen.append(step.line()), | |
| ).run("save") | |
| print("streamed live, one line per step:") | |
| for line in seen: | |
| print(" ", line) | |
| print() | |
| print("and the finished episode has", len(episode.steps), "steps") | |
| """) | |
| # =========================================================================== # | |
| # 5. the safety gates | |
| # =========================================================================== # | |
| term("safety-deny-text", "Safety", | |
| "Typing from the deny list never reaches the keyboard", | |
| ["Policy", "Policy.check", "PolicyViolation", "DEFAULT_DENY_TEXT"], | |
| note="Checked before execution; a refusal ends the episode and clicks nothing.", | |
| code=""" | |
| from tiny_computer_agent import Action, DryRunExecutor, AgentLoop, Element, FakeScreen, MockBrain, Policy | |
| for text in ["report.txt", "sudo rm -rf /", "my api_key is here", "correct horse battery staple"]: | |
| episode = AgentLoop( | |
| screen=FakeScreen([(Element("Name", kind="input", x=100, y=120, width=200, height=24),)]), | |
| brain=MockBrain([{"kind": "type", "text": text}]), | |
| ).run("fill the form") | |
| verdict = "typed" if episode.steps[-1].error is None else "REFUSED" | |
| print(f" type({text!r:32}) -> {verdict}") | |
| if episode.steps[-1].error: | |
| print(f" {episode.steps[-1].error}") | |
| """) | |
| term("safety-deny-label", "Safety", | |
| "Clicking a denied widget is refused, however the label is spelled", | |
| ["Policy.deny_labels", "Policy.check", "Observation.find"], | |
| note="Terminal, keychain, wallet, password manager, bank - matched on the label.", | |
| code=""" | |
| from tiny_computer_agent import Action, Element, Observation, Policy, PolicyViolation | |
| frame = Observation(elements=( | |
| Element("Terminal", x=0, y=0, width=80, height=24), | |
| Element("Password Manager", x=100, y=0, width=160, height=24), | |
| Element("Save", x=320, y=120, width=80, height=24), | |
| )) | |
| policy = Policy() | |
| for label in ["Terminal", "Password Manager", "T E R M I N A L", "Save"]: | |
| try: | |
| policy.check(Action("click", label=label), frame) | |
| print(f" click({label!r:20}) -> allowed") | |
| except PolicyViolation as exc: | |
| print(f" click({label!r:20}) -> refused: {exc}") | |
| """) | |
| term("safety-deny-keys", "Safety", | |
| "Key chords on the deny list are never pressed", | |
| ["Policy.deny_keys", "Policy.check", "PolicyViolation"], | |
| note="ctrl+alt+t, ctrl+alt+delete and ctrl+shift+escape are refused by default.", | |
| code=""" | |
| from tiny_computer_agent import Action, Policy, PolicyViolation | |
| policy = Policy() | |
| for chord in ["ctrl+s", "ctrl+alt+t", "CTRL+SHIFT+ESCAPE", "ctrl+alt+delete"]: | |
| try: | |
| policy.check(Action("key", keys=chord)) | |
| print(f" key({chord!r:20}) -> allowed") | |
| except PolicyViolation as exc: | |
| print(f" key({chord!r:20}) -> refused: {exc}") | |
| """) | |
| term("safety-blind-grid", "Safety", | |
| "A numbered grid cell is a blind click: allowed to plan, refused for real", | |
| ["Policy.allow_grid_clicks", "Policy.check", "GridDetector"], | |
| note="With real input the blind guess is refused unless you pass --allow-grid.", | |
| code=""" | |
| from tiny_computer_agent import Action, GridDetector, Policy, PolicyViolation | |
| from tiny_computer_agent.demo import fake_desktop | |
| elements = GridDetector(rows=2, cols=3).detect(fake_desktop()) | |
| from tiny_computer_agent import Observation | |
| frame = Observation(elements=elements) | |
| print("a screen with no text degrades to", len(elements), "numbered cells:", | |
| ", ".join(e.label for e in elements)) | |
| print() | |
| for policy, name in [(Policy(dry_run=True), "dry run (planning)"), | |
| (Policy(dry_run=False), "real input, default"), | |
| (Policy(dry_run=False, allow_grid_clicks=True), "real input, --allow-grid")]: | |
| try: | |
| policy.check(Action("click", label=elements[4].label), frame) | |
| print(f" {name:32} -> allowed") | |
| except PolicyViolation as exc: | |
| print(f" {name:32} -> refused: {exc}") | |
| """) | |
| # =========================================================================== # | |
| # 6. screens: where the pixels come from | |
| # =========================================================================== # | |
| term("screen-fake", "Screens", | |
| "A scripted screen, for testing the loop with no display", | |
| ["FakeScreen", "FakeScreen.capture", "Observation"], | |
| note="The final frame repeats forever, so a runaway agent cannot exhaust the script.", | |
| code=""" | |
| from tiny_computer_agent import Element, FakeScreen | |
| screen = FakeScreen([(Element("Files", x=10, y=6, width=60, height=20),), | |
| (Element("Save", x=320, y=120, width=80, height=24),)], text="Untitled document") | |
| for i in range(4): | |
| observation = screen.capture() | |
| print(f" capture {i} -> index={observation.index} elements={[e.label for e in observation.elements]}") | |
| """) | |
| term("screen-image", "Screens", | |
| "A still image (or a sequence) replayed as a screen", | |
| ["ImageScreen", "ImageScreen.capture", "fake_desktop"], | |
| note="Lets the whole pipeline be exercised with no display, and a failure be replayed.", | |
| code=""" | |
| from tiny_computer_agent import ImageScreen, fake_desktop | |
| screen = ImageScreen(fake_desktop()) | |
| for i in range(3): | |
| observation = screen.capture() | |
| print(f" capture {i} -> index={observation.index} frame={type(observation.frame).__name__} {observation.frame.size}") | |
| print() | |
| print("captures beyond the scripted frames keep returning the last one") | |
| """) | |
| term("capture-options", "Screens", | |
| "Real capture: monitors, downscaling, and HiDPI mapping", | |
| ["GrabScreen", "GrabScreen.capture", "GrabScreen.map_to_screen", "ScreenUnavailable"], | |
| note="Frame pixel -> screen pixel is preserved, so a click cannot drift.", | |
| code=""" | |
| from tiny_computer_agent import Element, GrabScreen | |
| for name, screen in [("monitor 1 (native)", GrabScreen(monitor=1)), | |
| ("max_width=640", GrabScreen(monitor=1, max_width=640)), | |
| ("scale=2.0 (HiDPI)", GrabScreen(monitor=1, scale=2.0))]: | |
| observation = screen.capture() | |
| print(f" {name:20} frame={observation.frame.width}x{observation.frame.height}px " | |
| f"origin={screen.origin} frame_scale={screen.frame_scale:.3f}") | |
| print() | |
| scaled = GrabScreen(monitor=1, max_width=640) | |
| scaled.capture() | |
| label = Element("Save", x=100, y=50, width=80, height=24) | |
| print("a label found at (100,50) in the DOWNSCALED frame maps back to", scaled.map_to_screen(label).center) | |
| print("on the real screen - which is why the downscale cannot break a click.") | |
| """) | |
| term("element-matching", "Screens", | |
| "Labels are compared on their alphanumeric core", | |
| ["Element", "Element.center", "Element.matches", "Observation.find", "normalise_label"], | |
| note="OCR loses spaces, models add punctuation - so nothing matches by exact string.", | |
| code=""" | |
| from tiny_computer_agent import Element, Observation, normalise_label | |
| save = Element("Save", x=200, y=270, width=110, height=30) | |
| print("Element('Save', x=200, y=270, width=110, height=30)") | |
| print(" .center:", save.center, "| .matches('save'):", save.matches("save")) | |
| print() | |
| print("normalise_label keeps only letters and digits:") | |
| for raw in ["Save", "Save .", "S A V E", "save!", "Untitled document - unsaved changes"]: | |
| print(f" {raw!r:36} -> {normalise_label(raw)!r}") | |
| print() | |
| frame = Observation(elements=(Element("Save .", x=10, y=10, width=80, height=24), | |
| Element("Save As", x=110, y=10, width=80, height=24))) | |
| for probe in ["Save", "save", "Save As", "Nothing"]: | |
| hit = frame.find(probe) | |
| print(f" find({probe!r:10}) -> {hit.label if hit else 'not on screen'!r}") | |
| """) | |
| term("observation-summary", "Screens", | |
| "The compact text a model is shown", | |
| ["Observation.summary"], | |
| note="This is the whole world as the brain sees it - capped, and honest when empty.", | |
| code=""" | |
| from tiny_computer_agent import Element, Observation | |
| frame = Observation(index=3, text="Untitled document - unsaved changes", elements=( | |
| Element("Files", x=10, y=6, width=60, height=20), | |
| Element("Save", x=200, y=270, width=110, height=30), | |
| Element("Cancel", x=400, y=270, width=110, height=30), | |
| )) | |
| print(frame.summary()) | |
| print() | |
| print("an empty index says so rather than pretending:") | |
| print(Observation(index=4).summary()) | |
| """) | |
| # =========================================================================== # | |
| # 7. perception: pixels -> a labelled element index | |
| # =========================================================================== # | |
| term("detector-available", "Perception", | |
| "Probing for a detector, and building one by name", | |
| ["detector_available", "make_detector", "DetectorUnavailable", "Detector"], | |
| note="The doctor and the CLI both ask this before promising anything.", | |
| code=""" | |
| from tiny_computer_agent import DetectorUnavailable, detector_available, make_detector | |
| for name in ["auto", "ocr", "grid", "telepathy"]: | |
| ok, detail = detector_available(name) | |
| print(f" detector_available({name!r:11}) -> {ok!s:5} {detail}") | |
| print() | |
| for name in ["auto", "grid"]: | |
| print(f" make_detector({name!r:6}) -> {type(make_detector(name)).__name__}") | |
| try: | |
| make_detector("telepathy") | |
| except DetectorUnavailable as exc: | |
| print(" make_detector('telepathy') -> DetectorUnavailable:", exc) | |
| """) | |
| term("ocr-labels", "Perception", | |
| "OCR turns a screenshot into labelled, clickable elements", | |
| ["OcrDetector", "OcrDetector.detect", "Element"], | |
| note="The model never has to guess where anything is: this is the grounding.", | |
| code=""" | |
| from tiny_computer_agent import OcrDetector, fake_desktop | |
| elements = OcrDetector().detect(fake_desktop()) | |
| print(f"OCR read {len(elements)} labelled elements off one screenshot:") | |
| print() | |
| for element in elements: | |
| print(f" {element.label!r:34} kind={element.kind:6} centre={str(element.center):12} box={element.width}x{element.height}") | |
| """) | |
| term("ocr-filters", "Perception", | |
| "Three knobs keep dialog prose out of the menu", | |
| ["OcrDetector.min_score", "OcrDetector.max_label_chars", "OcrDetector.min_label_chars"], | |
| note="Every extra line of noise is one more chance for a small model to click prose.", | |
| code=""" | |
| from tiny_computer_agent import OcrDetector, fake_desktop | |
| image = fake_desktop() | |
| for label, detector in [ | |
| ("default", OcrDetector()), | |
| ("min_score=0.99", OcrDetector(min_score=0.99)), | |
| ("max_label_chars=6", OcrDetector(max_label_chars=6)), | |
| ("min_label_chars=8", OcrDetector(min_label_chars=8)), | |
| ]: | |
| print(f" {label:18} -> {[e.label for e in detector.detect(image)]}") | |
| """) | |
| term("grid-detector", "Perception", | |
| "A numbered grid, for screens with no text at all", | |
| ["GridDetector", "GridDetector.detect"], | |
| note="An agent with an empty index has no legal moves; a grid always has moves.", | |
| code=""" | |
| from tiny_computer_agent import GridDetector, fake_desktop | |
| for rows, cols in [(2, 3), (8, 12)]: | |
| cells = GridDetector(rows=rows, cols=cols).detect(fake_desktop()) | |
| print(f" {rows}x{cols} grid -> {len(cells)} cells: " | |
| f"first={cells[0].label!r} at {cells[0].center}, last={cells[-1].label!r} at {cells[-1].center}") | |
| """) | |
| term("auto-detector", "Perception", | |
| "OCR when there is text, a grid when there is none", | |
| ["AutoDetector", "AutoDetector.detect", "AutoDetector.coordinates_of_last_detect"], | |
| note="The default: useful with nothing installed, better the moment OCR appears.", | |
| code=""" | |
| from PIL import Image | |
| from tiny_computer_agent import AutoDetector, fake_desktop | |
| detector = AutoDetector() | |
| with_text = detector.detect(fake_desktop()) | |
| blank = detector.detect(Image.new("RGB", (720, 460), (18, 18, 22))) | |
| print(" a screenshot with text ->", len(with_text), "elements:", [e.label for e in with_text][:6]) | |
| print(" a screen with no text ->", len(blank), "elements:", [e.label for e in blank][:6]) | |
| """) | |
| term("indexed-screen", "Perception", | |
| "IndexedScreen: capture + detect, composed", | |
| ["IndexedScreen", "IndexedScreen.capture", "Observation.summary"], | |
| note="One capture() call gives the loop an index it can act on.", | |
| code=""" | |
| from tiny_computer_agent import ImageScreen, IndexedScreen, OcrDetector, fake_desktop | |
| screen = IndexedScreen(ImageScreen(fake_desktop()), OcrDetector()) | |
| observation = screen.capture() | |
| print(f"capture() composed pixels -> {len(observation.elements)} labelled elements:") | |
| print() | |
| print(observation.summary()) | |
| """) | |
| # =========================================================================== # | |
| # 8. preflight: what can this machine do? | |
| # =========================================================================== # | |
| term("display-available", "Preflight", | |
| "Is there a desktop session to drive at all?", | |
| ["display_available"], | |
| note="The first question the CLI and the doctor both ask.", | |
| code=""" | |
| from tiny_computer_agent import display_available | |
| ok, detail = display_available() | |
| print("display_available() ->", ok) | |
| print(" detail:", detail) | |
| """) | |
| term("run-checks", "Preflight", | |
| "Every precondition, one list, each with the fix", | |
| ["run_checks", "Check", "Check.render"], | |
| note="Eight checks; a required one that fails is a blocking problem.", | |
| code=""" | |
| from tiny_computer_agent import run_checks | |
| checks = run_checks(brain="planner", detector="ocr", real_input=False) | |
| for check in checks: | |
| print(check.render()) | |
| blocking = [c for c in checks if c.required and not c.ok] | |
| print() | |
| print(f"{len(checks)} checks, {len(blocking)} blocking: {[c.name for c in blocking]}") | |
| """) | |
| term("cli-doctor", "Preflight", | |
| "python -m tiny_computer_agent --doctor", | |
| ["doctor", "display_available", "detector_available"], | |
| note="The first thing to run on a new machine. Exits non-zero when not ready.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--doctor"]) | |
| term("build-agent", "Preflight", | |
| "One call wires screen + detector + brain + executor + policy", | |
| ["build_agent", "Policy", "AgentLoop"], | |
| note="Defaults are the safe ones: dry run, a step budget, and a kill-switch file.", | |
| code=""" | |
| from tiny_computer_agent import Element, FakeScreen, PlannerBrain, build_agent | |
| loop = build_agent( | |
| brain=PlannerBrain(), | |
| screen=FakeScreen([(Element("Save", x=320, y=120, width=80, height=24),)]), | |
| dry_run=True, max_steps=4, | |
| ) | |
| print("build_agent(...) ->", type(loop).__name__) | |
| print(" executor :", type(loop.executor).__name__, "| dry_run:", loop.executor.dry_run) | |
| print(" step budget :", loop.policy.step_budget) | |
| print(" kill switch :", loop.policy.kill_switch_path) | |
| print(" grid clicks :", loop.policy.allow_grid_clicks) | |
| print() | |
| print(loop.run("click 'Save'").render()) | |
| """) | |
| # =========================================================================== # | |
| # 9. the model backends | |
| # =========================================================================== # | |
| term("models-deps", "Models", | |
| "Which model runtime is present, and are the weights cached?", | |
| ["model_dependencies", "model_cached", "MODEL_BRAINS", "make_model_brain", | |
| "DEFAULT_TEXT_MODEL", "DEFAULT_VISION_MODEL", "BrainError"], | |
| note="Weights are loaded lazily, so asking is free.", | |
| code=""" | |
| from tiny_computer_agent import (DEFAULT_TEXT_MODEL, DEFAULT_VISION_MODEL, MODEL_BRAINS, | |
| BrainError, make_model_brain, model_cached, model_dependencies) | |
| ok, detail = model_dependencies() | |
| print("model_dependencies() ->", ok, "|", detail) | |
| print("default text model :", DEFAULT_TEXT_MODEL) | |
| print(" weights cached :", model_cached(DEFAULT_TEXT_MODEL)) | |
| print("default vision model :", DEFAULT_VISION_MODEL) | |
| print(" weights cached :", model_cached(DEFAULT_VISION_MODEL)) | |
| print() | |
| print("brains by name, and an honest refusal for the rest:") | |
| for name in MODEL_BRAINS: | |
| print(f" make_model_brain({name!r:13}) -> {type(make_model_brain(name)).__name__}") | |
| try: | |
| make_model_brain("gpt-9") | |
| except BrainError as exc: | |
| print(" make_model_brain('gpt-9') -> BrainError:", exc) | |
| """) | |
| term("models-device", "Models", | |
| "Device and dtype selection", | |
| ["resolve_device", "pick_dtype"], | |
| note="CUDA if it is there, CPU otherwise; the harness never needs a GPU.", | |
| code=""" | |
| from tiny_computer_agent import pick_dtype, resolve_device | |
| print("resolve_device() ->", resolve_device()) | |
| print("resolve_device('cpu') ->", resolve_device("cpu")) | |
| print("pick_dtype('cpu') ->", pick_dtype("cpu")) | |
| print("pick_dtype('cuda') ->", pick_dtype("cuda")) | |
| """) | |
| term("models-prompt", "Models", | |
| "The prompt a model actually receives - and how it learns from a dead end", | |
| ["ModelBrain", "ModelBrain.build_prompt", "ModelBrain.decide"], | |
| note="Placeholders, not worked examples: a weak model echoes the prompt, and an echo fails validation.", | |
| code=""" | |
| from tiny_computer_agent import Element, ModelBrain, Observation | |
| class Scripted(ModelBrain): | |
| name = "scripted" | |
| def _generate(self, prompt, observation): | |
| self.prompt_seen = prompt | |
| return '{"kind": "click", "index": 2, "reason": "saves the document"}' | |
| brain = Scripted("example/model") | |
| frame = Observation(text="Untitled document - unsaved changes", elements=( | |
| Element("Files", x=10, y=6, width=60, height=20), | |
| Element("Save", x=200, y=270, width=110, height=30), | |
| Element("Cancel", x=400, y=270, width=110, height=30), | |
| )) | |
| action = brain.decide("save the document", frame) | |
| print("the model chose:", action.describe()) | |
| print() | |
| print("the prompt it was given:") | |
| print(brain.prompt_seen) | |
| """) | |
| term("models-prompt-history", "Models", | |
| "A second decision is told what already failed", | |
| ["ModelBrain", "ModelBrain.build_prompt", "ModelBrain.history"], | |
| note="'you already tried this and the screen did NOT change' is the most useful line a small model gets.", | |
| code=""" | |
| from tiny_computer_agent import Element, ModelBrain, Observation | |
| class Scripted(ModelBrain): | |
| name = "scripted" | |
| def _generate(self, prompt, observation): | |
| self.prompt_seen = prompt | |
| return '{"kind": "click", "index": 2, "reason": "again"}' | |
| brain = Scripted("example/model") | |
| frame = Observation(text="Untitled document", elements=( | |
| Element("Files", x=10, y=6, width=60, height=20), | |
| Element("Save", x=200, y=270, width=110, height=30), | |
| )) | |
| brain.decide("save", frame) # first attempt | |
| brain.decide("save", frame) # same screen: nothing changed | |
| print("history the brain is carrying:", brain.history) | |
| print() | |
| print("the tail of the second prompt:") | |
| print(brain.prompt_seen[-420:]) | |
| """) | |
| # =========================================================================== # | |
| # 10. traces and scoring | |
| # =========================================================================== # | |
| term("scenario", "Traces & scoring", | |
| "A task with one right answer, and the frames to show", | |
| ["Scenario", "Scenario.screen", "run_scenario", "Scenario.label"], | |
| note="Ground truth lives with the task, so a run can be scored at all.", | |
| code=""" | |
| from tiny_computer_agent import Element, PlannerBrain, Scenario, run_scenario | |
| def item(label, x): return Element(label, x=x, y=6, width=80, height=20) | |
| def button(label, x): return Element(label, x=x, y=270, width=110, height=30) | |
| scenario = Scenario( | |
| "click 'Save' to save the document", | |
| frames=[(item("Files", 10), item("Edit", 100)), | |
| (Element("Name", x=200, y=195, width=200, height=26), button("Save", 200)), | |
| (item("Files", 10), button("Save", 200))], | |
| expected="Save", name="save the document", | |
| ) | |
| print("Scenario:", scenario.label, "| frames:", len(scenario.frames), "| the right answer:", scenario.expected) | |
| print() | |
| trace = run_scenario(scenario, PlannerBrain()) | |
| print(trace.render()) | |
| """) | |
| term("trace-record", "Traces & scoring", | |
| "An episode becomes a portable, pixel-free trace", | |
| ["Trace", "Trace.record", "Trace.save", "Trace.load", "Trace.to_json", | |
| "Trace.to_dict", "Trace.choice", "Trace.decided", "Trace.frames"], | |
| note="JSON, so it diffs in review and replays on a machine with nothing installed.", | |
| code=""" | |
| import json, tempfile | |
| from pathlib import Path | |
| from tiny_computer_agent import Element, PlannerBrain, Scenario, Trace, run_scenario | |
| scenario = Scenario("click 'Save'", frames=[(Element("Save", x=320, y=120, width=80, height=24),)], | |
| expected="Save", name="save") | |
| trace = run_scenario(scenario, PlannerBrain()) | |
| print("frames:", trace.frames, "| choice:", trace.choice, | |
| "| decided:", trace.decided, "| blocked:", trace.blocked) | |
| path = Path(tempfile.mkdtemp()) / "save.json" | |
| trace.save(path) | |
| print("saved:", path.name, f"({path.stat().st_size} bytes) | reloaded identical:", | |
| Trace.load(path).to_dict() == trace.to_dict()) | |
| print() | |
| print(json.dumps(trace.to_dict(), indent=2)[:560] + " ...") | |
| """) | |
| term("trace-replay", "Traces & scoring", | |
| "Replay a recorded trace against a different brain", | |
| ["Trace.replay", "Trace.observations", "score_trace"], | |
| note="The frames are the fixture; the brain is the only variable. Nothing is captured or clicked.", | |
| code=""" | |
| from tiny_computer_agent import Action, Element, PlannerBrain, Scenario, run_scenario, score_trace | |
| scenario = Scenario("click 'Save'", frames=[ | |
| (Element("Files", x=10, y=6, width=60, height=20),), | |
| (Element("Save", x=320, y=120, width=80, height=24),)], expected="Save", name="save") | |
| recorded = run_scenario(scenario, PlannerBrain()) | |
| print("recorded by the planner :", recorded.choice, "->", score_trace(recorded).correct) | |
| class AlwaysFirst: | |
| name = "first-item" | |
| def decide(self, goal, observation): | |
| return Action("click", label=observation.elements[0].label, reason="first item") | |
| replayed = recorded.replay(AlwaysFirst()) | |
| print("the same frames, another brain:", replayed.choice, "->", score_trace(replayed).correct) | |
| print() | |
| print("replayed frames carried over:", replayed.frames, "| ground truth carried over:", replayed.expected) | |
| """) | |
| term("score-trace", "Traces & scoring", | |
| "Counting how often the choice was right", | |
| ["score_trace", "score", "score_replay", "Verdict", "Verdict.line"], | |
| note="Scored on normalised text, so OCR spacing is forgiven - the same bargain the index makes.", | |
| code=""" | |
| from tiny_computer_agent import (Action, Element, PlannerBrain, Scenario, Trace, TraceStep, | |
| score, score_replay, score_trace) | |
| scenario = Scenario("click 'Save'", frames=[(Element("Save", x=320, y=120, width=80, height=24),)], | |
| expected="Save", name="save") | |
| print("score(scenario, PlannerBrain()) ->", score(scenario, PlannerBrain()).line().strip()) | |
| print() | |
| print("a choice the OCR rendered as 'Save .' is still credited for 'Save':") | |
| print(" ", score_trace(Trace(goal="g", expected="Save", | |
| steps=[TraceStep(0, action=Action("click", label="Save ."))])).line().strip()) | |
| print() | |
| print("a trace with no ground truth says so instead of inventing a verdict:") | |
| print(" ", score_trace(Trace(goal="g", | |
| steps=[TraceStep(0, action=Action("click", label="Save"))])).line().strip()) | |
| print() | |
| print("score_replay(trace, another_brain) is the single-trace form of the roadmap step:") | |
| print(" ", score_replay(Trace(goal="g", expected="Save", | |
| steps=[TraceStep(0, action=Action("click", label="Cancel"))]), PlannerBrain()).line().strip()) | |
| """) | |
| term("scoreboard", "Traces & scoring", | |
| "Verdicts collected into an accuracy", | |
| ["Scoreboard", "Scoreboard.render", "Scoreboard.accuracy", "Scoreboard.by_brain", | |
| "Scoreboard.to_rows", "Scoreboard.confusion", "Scoreboard.blocked"], | |
| note="The number the whole module exists for: which brain is right, and how often.", | |
| code=""" | |
| from tiny_computer_agent import Action, Element, MockBrain, PlannerBrain, Scenario, benchmark | |
| class FirstItem: | |
| name = "first-item" | |
| def decide(self, goal, observation): | |
| return Action("click", label=observation.elements[0].label, reason="first item") | |
| def scenario(goal, expected, frames): | |
| return Scenario(goal, frames=frames, expected=expected, name=expected.lower()) | |
| suite = [ | |
| scenario("click 'Save' to save the document", "Save", | |
| [(Element("Files", x=10, y=6, width=60, height=20),), | |
| (Element("Save", x=320, y=120, width=80, height=24),)]), | |
| scenario("click 'Cancel' to discard the change", "Cancel", | |
| [(Element("Save", x=200, y=270, width=110, height=30), | |
| Element("Cancel", x=400, y=270, width=110, height=30),)]), | |
| ] | |
| board = benchmark(suite, { | |
| "planner": lambda s: PlannerBrain(), | |
| "first-item": lambda s: FirstItem(), | |
| "oracle": lambda s: MockBrain([{"kind": "click", "label": s.expected}]), | |
| }) | |
| print(board.render()) | |
| print() | |
| print("accuracy:", f"{board.accuracy:.0%}", "| by brain:", board.by_brain()) | |
| print("as rows :", board.to_rows()) | |
| print("confusion (expected, chose) -> count:", board.confusion()) | |
| """) | |
| term("benchmark", "Traces & scoring", | |
| "Every scenario through every brain, in one table", | |
| ["benchmark", "Scoreboard.to_rows"], | |
| note="A fresh brain per task, so state like 'already clicked this' cannot leak between them.", | |
| code=""" | |
| from tiny_computer_agent import Action, Element, PlannerBrain, Scenario, benchmark | |
| class FirstItem: | |
| name = "first-item" | |
| def decide(self, goal, observation): | |
| return Action("click", label=observation.elements[0].label, reason="first item") | |
| suite = [Scenario(f"click {label!r}", frames=[(Element("Files", x=10, y=6, width=60, height=20), | |
| Element(label, x=200, y=270, width=110, height=30),)], | |
| expected=label, name=label.lower()) | |
| for label in ("Save", "Cancel")] | |
| board = benchmark(suite, {"planner": lambda s: PlannerBrain(), "first-item": lambda s: FirstItem()}) | |
| for row in board.to_rows(): | |
| print(f" {row['brain']:12} {row['scenario']:8} expected={row['expected']:7} " | |
| f"chose={row['chose']:7} correct={row['correct']}") | |
| """) | |
| term("benchmark-traces", "Traces & scoring", | |
| "Replay a whole recorded suite against one brain", | |
| ["benchmark_traces", "load_traces", "Trace.save"], | |
| note="The literal roadmap step: the frames are fixed, nothing new is captured.", | |
| code=""" | |
| import tempfile | |
| from pathlib import Path | |
| from tiny_computer_agent import (Element, PlannerBrain, Scenario, benchmark_traces, | |
| load_traces, run_scenario) | |
| def scenario(goal, expected, frames): | |
| return Scenario(goal, frames=frames, expected=expected, name=expected.lower()) | |
| suite = [ | |
| scenario("click 'Save'", "Save", [(Element("Files", x=10, y=6, width=60, height=20),), | |
| (Element("Save", x=320, y=120, width=80, height=24),)]), | |
| scenario("click 'Cancel'", "Cancel", [(Element("Save", x=200, y=270, width=110, height=30), | |
| Element("Cancel", x=400, y=270, width=110, height=30),)]), | |
| ] | |
| folder = Path(tempfile.mkdtemp()) | |
| for item in suite: | |
| run_scenario(item, PlannerBrain()).save(folder / f"{item.name}.json") | |
| print("recorded suite on disk:", [p.name for p in sorted(folder.glob("*.json"))]) | |
| traces = load_traces(folder) | |
| print("load_traces ->", [t.name for t in traces]) | |
| print() | |
| board = benchmark_traces(traces, PlannerBrain(), name="planner (replayed)") | |
| print(board.render()) | |
| """) | |
| # =========================================================================== # | |
| # 11. the command line | |
| # =========================================================================== # | |
| term("cli-image", "Command line", | |
| "--image example (replay a screenshot, dry run)", | |
| ["main", "build_parser", "ImageScreen", "detector_available"], | |
| note="No display needed: the whole pipeline over a still image.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--image", "example"]) | |
| term("cli-script", "Command line", | |
| "--script (replay a fixed action list instead of a brain)", | |
| ["MockBrain", "parse_action"], | |
| note="The script is validated exactly like a model's answer: nothing skips the gates.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--script", | |
| '[{"kind": "click", "label": "Save"}, {"kind": "done", "reason": "finished"}]']) | |
| term("cli-verify-text", "Command line", | |
| "--verify-text (the harness decides success, not the brain)", | |
| ["AgentLoop.run", "Episode.success"], | |
| note="Without it, an action that removes its own target looks like a failure.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--script", | |
| '[{"kind": "wait", "seconds": 0.1}]', "--verify-text", "Unsaved", "--max-steps", "2"]) | |
| term("cli-trace", "Command line", | |
| "--trace / --expected (record the run, then score it)", | |
| ["Trace.record", "Trace.save", "score_trace"], | |
| note="Writes JSON next to the code, and prints the score straight away.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--image", "example", | |
| "--trace", "/tmp/gallery-save.json", "--expected", "Save"]) | |
| term("cli-detector", "Command line", | |
| "--no-ocr (index a numbered grid instead)", | |
| ["GridDetector", "make_detector", "detector_available"], | |
| note="Useful on a machine with no OCR engine, or for a canvas app.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--image", "example", "--no-ocr"]) | |
| term("cli-exit-codes", "Command line", | |
| "Exit codes: 0 success, 1 failure, 2 bad usage, 3 no screen", | |
| ["main"], | |
| note="A shell can tell 'it worked' from 'it gave up' without parsing any text.", | |
| code=""" | |
| import subprocess, sys | |
| cases = [ | |
| (["--script", '[{"kind": "done", "reason": "nothing to do"}]'], "a brain-reported success"), | |
| (["click 'Nonexistent button'"], "a goal that cannot be reached"), | |
| (["--real-input"], "real input without the allow env"), | |
| (["--brain", "vlm"], "the vlm brain with no pixels"), | |
| (["--image", "/nonexistent/shot.png"], "a screenshot that is not there"), | |
| ] | |
| for flags, what in cases: | |
| code = subprocess.run([sys.executable, "-m", "tiny_computer_agent", *flags], | |
| capture_output=True, text=True).returncode | |
| print(f" exit {code} {what:38} ({' '.join(flags)[:46]})") | |
| """, env={"TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT": ""}) | |
| term("cli-refusals", "Command line", | |
| "The two deliberate acts before the mouse can move", | |
| ["main", "ALLOW_REAL_INPUT_ENV"], | |
| note="--real-input without the env var, and --brain vlm with no pixels: both refuse.", | |
| code=""" | |
| import subprocess, sys | |
| for flags in (["--real-input"], ["--brain", "vlm"]): | |
| print(f"$ python -m tiny_computer_agent {' '.join(flags)}") | |
| subprocess.run([sys.executable, "-m", "tiny_computer_agent", *flags]) | |
| print() | |
| """, env={"TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT": ""}) | |
| term("cli-live-headless", "Command line", | |
| "--live with no display is refused, with the fix", | |
| ["display_available", "main"], | |
| note="It tells you what to do instead of dying inside an episode.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--live"], | |
| env={"DISPLAY": "", "WAYLAND_DISPLAY": ""}) | |
| term("cli-live-dry-run", "Command line", | |
| "--live (your real screen, dry run)", | |
| ["GrabScreen", "IndexedScreen", "AutoDetector", "main"], | |
| note="Real capture, real OCR, a real element index - and not one click.", | |
| argv=["{py}", "-m", "tiny_computer_agent", "--live", "--max-steps", "2"], | |
| env={"DISPLAY": ":99", "TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT": ""}) | |
| # =========================================================================== # | |
| # 12. on a real desktop: real capture, real OCR, a real click | |
| # =========================================================================== # | |
| SCENES["dialog"] = (ROOT / "showcase_scenes" / "dialog.py").read_text() | |
| SCENES["canvas"] = (ROOT / "showcase_scenes" / "canvas.py").read_text() | |
| SCENES["toolbar"] = (ROOT / "showcase_scenes" / "toolbar.py").read_text() | |
| DIALOG_BOX = (100, 100, 600, 380) | |
| TOOLBAR_BOX = (140, 140, 820, 480) | |
| def make_capture(tmp): | |
| proc = open_scene("dialog", tmp) | |
| try: | |
| return annotate(grab(), [], title="GrabScreen(monitor=1).capture() - a real 1280x800 desktop") | |
| finally: | |
| close_scene(proc) | |
| def make_ocr_index(tmp): | |
| from tiny_computer_agent import OcrDetector | |
| proc = open_scene("dialog", tmp) | |
| try: | |
| frame = grab() | |
| elements = OcrDetector().detect(frame) | |
| return annotate(frame, elements, title="OcrDetector().detect(frame) -> " | |
| + str(len(elements)) + " labelled elements with real coordinates") | |
| finally: | |
| close_scene(proc) | |
| def make_click(tmp): | |
| from tiny_computer_agent import (AgentLoop, GrabScreen, IndexedScreen, OcrDetector, | |
| PlannerBrain, Policy, PyAutoGuiExecutor) | |
| proc = open_scene("dialog", tmp) | |
| try: | |
| screen = IndexedScreen(GrabScreen(monitor=1), OcrDetector()) | |
| before_observation = screen.capture() | |
| before = crop(annotate(grab(), before_observation.elements), DIALOG_BOX) | |
| episode = AgentLoop( | |
| screen=screen, brain=PlannerBrain(), executor=PyAutoGuiExecutor(), | |
| policy=Policy(dry_run=False, step_budget=3), | |
| verify=lambda observation: "saved" in observation.text.lower(), | |
| ).run("click 'Save' to save the document") | |
| target = before_observation.find("Save") | |
| after = crop(annotate(grab(), [], click=target.center if target else None), DIALOG_BOX) | |
| lines = ["result: " + ("success" if episode.success else "failure") | |
| + " - " + episode.stop_reason] | |
| lines += [" " + step.line() for step in episode.steps][:2] | |
| return overlay_lines(side_by_side( | |
| before, after, title="PyAutoGuiExecutor: a real click lands on a real button"), lines) | |
| finally: | |
| close_scene(proc) | |
| def make_verify(tmp): | |
| from tiny_computer_agent import (AgentLoop, GrabScreen, IndexedScreen, OcrDetector, | |
| PlannerBrain, Policy, PyAutoGuiExecutor) | |
| proc = open_scene("dialog", tmp) | |
| try: | |
| episode = AgentLoop( | |
| screen=IndexedScreen(GrabScreen(monitor=1), OcrDetector()), | |
| brain=PlannerBrain(), executor=PyAutoGuiExecutor(), | |
| policy=Policy(dry_run=False, step_budget=3), | |
| verify=lambda observation: "saved" in observation.text.lower(), | |
| ).run("click 'Save' to save the document") | |
| marker = tmp / "dialog.marker" | |
| frame = crop(annotate(grab(), []), DIALOG_BOX) | |
| return overlay_lines(frame, [ | |
| "the window changed its own title to 'Saved to disk',", | |
| "and wrote " + repr(marker.read_text()) + " to " + marker.name, | |
| "verify(...) saw it -> episode.success = " + str(episode.success), | |
| "stop reason: " + episode.stop_reason, | |
| ]) | |
| finally: | |
| close_scene(proc) | |
| def make_grid_fallback(tmp): | |
| from tiny_computer_agent import AutoDetector, OcrDetector | |
| proc = open_scene("canvas", tmp) | |
| try: | |
| frame = grab() | |
| ocr = OcrDetector().detect(frame) | |
| elements = AutoDetector().detect(frame) | |
| return overlay_lines(annotate( | |
| frame, elements, title="a canvas app has no text: the grid fallback"), [ | |
| "OcrDetector found " + str(len(ocr)) + " labels on these pixels", | |
| "AutoDetector fell back to a " + str(len(elements)) | |
| + "-cell numbered grid, so the agent still has moves", | |
| "with real input a blind grid click is refused unless --allow-grid is passed", | |
| ]) | |
| finally: | |
| close_scene(proc) | |
| def make_deny(tmp): | |
| from tiny_computer_agent import Action, Observation, OcrDetector, Policy, PolicyViolation | |
| proc = open_scene("toolbar", tmp) | |
| try: | |
| frame = grab() | |
| elements = OcrDetector().detect(frame) | |
| target = next((e for e in elements if e.matches("Terminal")), None) | |
| message = "no Terminal button on screen" | |
| if target is not None: | |
| try: | |
| Policy(dry_run=False).check( | |
| Action("click", label=target.label), Observation(elements=tuple(elements))) | |
| message = "allowed (unexpected)" | |
| except PolicyViolation as exc: | |
| message = "refused: " + str(exc) | |
| return overlay_lines(crop(annotate( | |
| frame, elements, | |
| title="the harness can see a Terminal button - and will not press it"), TOOLBAR_BOX), | |
| ["Policy(dry_run=False).check(click 'Terminal')", message]) | |
| finally: | |
| close_scene(proc) | |
| def make_cli_real_input(tmp): | |
| from tiny_computer_agent import OcrDetector | |
| proc = open_scene("dialog", tmp) | |
| try: | |
| elements = OcrDetector().detect(grab()) | |
| before = crop(annotate(grab(), elements), DIALOG_BOX) | |
| result = subprocess.run( | |
| [PY, "-m", "tiny_computer_agent", "--live", "--real-input", "--verify-text", "Saved", | |
| "click 'Save' to save the document"], | |
| capture_output=True, text=True, | |
| env=dict(os.environ, TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT="1")) | |
| after = crop(grab(), DIALOG_BOX) | |
| lines = [line for line in result.stdout.splitlines() if line.strip()][:3] | |
| return overlay_lines(side_by_side( | |
| before, after, | |
| title="the CLI driving the real mouse: --live --real-input --verify-text Saved"), lines) | |
| finally: | |
| close_scene(proc) | |
| desk("desk-capture", "Real desktop", | |
| "GrabScreen: capture a real desktop", | |
| ["GrabScreen", "GrabScreen.capture", "ScreenUnavailable"], | |
| note="1280x800 of a real X server. mss first, Pillow's ImageGrab as the fallback.", | |
| scene_name="dialog", make=make_capture) | |
| desk("desk-ocr-index", "Real desktop", | |
| "The element index, drawn on the pixels it came from", | |
| ["OcrDetector", "OcrDetector.detect", "IndexedScreen", "Element.center"], | |
| note="Every box is a real OCR word with a real screen coordinate: the model picks a label, not a pixel.", | |
| scene_name="dialog", make=make_ocr_index) | |
| desk("desk-click", "Real desktop", | |
| "A real click, and the window's own reaction", | |
| ["PyAutoGuiExecutor", "AgentLoop.run", "Policy.dry_run", "resolve_point"], | |
| note="Pixels in, a label chosen, a pixel resolved, a button pressed - and the dialog closed itself.", | |
| scene_name="dialog", make=make_click) | |
| desk("desk-verify", "Real desktop", | |
| "Success only when the harness can see it", | |
| ["AgentLoop.run", "Episode.success", "Episode.stop_reason"], | |
| note="The app changed its own title and wrote a marker file; the verifier saw the change.", | |
| scene_name="dialog", make=make_verify) | |
| desk("desk-grid-fallback", "Real desktop", | |
| "A screen with no text still has legal moves", | |
| ["AutoDetector", "GridDetector", "OcrDetector"], | |
| note="OCR finds nothing on a canvas, so the index degrades to numbered cells - and says so.", | |
| scene_name="canvas", make=make_grid_fallback) | |
| desk("desk-policy-refusal", "Real desktop", | |
| "The safety gate, on a real button, in a real window", | |
| ["Policy.check", "Policy.deny_labels", "PolicyViolation", "Observation.find"], | |
| note="The Terminal button is visible and labelled - and the policy refuses to press it.", | |
| scene_name="toolbar", make=make_deny) | |
| desk("desk-cli-real-input", "Real desktop", | |
| "The same click, driven end to end from the CLI", | |
| ["main", "PyAutoGuiExecutor", "AgentLoop.run"], | |
| note="--live --real-input --verify-text Saved: capture, OCR, click, verify, exit code.", | |
| scene_name="dialog", make=make_cli_real_input) | |
| # =========================================================================== # | |
| # 13. the pieces a user swaps out | |
| # =========================================================================== # | |
| term("protocols", "Extensibility", | |
| "Every stage is a Protocol: bring your own screen, brain, detector, executor", | |
| ["Brain", "Brain.decide", "ScreenSource", "ScreenSource.capture", "Detector", | |
| "Detector.detect", "Executor", "Executor.execute", "ActionResult"], | |
| note="The loop depends on four one-method contracts, not on the shipped implementations.", | |
| code=""" | |
| from tiny_computer_agent import ActionResult, Action, AgentLoop, Element, Observation | |
| class MyScreen: # ScreenSource: capture() -> Observation | |
| def capture(self): | |
| return Observation(elements=(Element("Deploy", x=10, y=10, width=90, height=30),), | |
| text="release ready") | |
| class MyBrain: # Brain: decide(goal, observation) -> Action | |
| def __init__(self): | |
| self.turns = 0 | |
| def decide(self, goal, observation): | |
| self.turns += 1 | |
| if self.turns > 1: | |
| return Action("done", reason="the release is deployed") | |
| return Action("click", label="Deploy", reason="the goal names it") | |
| class MyDetector: # Detector: detect(image) -> tuple[Element, ...] | |
| def detect(self, image): | |
| return (Element("Deploy", x=10, y=10, width=90, height=30),) | |
| class MyExecutor: # Executor: execute(action, observation) -> ActionResult | |
| def __init__(self): | |
| self.saw = [] | |
| def execute(self, action, observation=None): | |
| self.saw.append(action.describe()) | |
| return ActionResult(action=action, ok=True, note="my own executor ran it") | |
| executor = MyExecutor() | |
| episode = AgentLoop(screen=MyScreen(), brain=MyBrain(), executor=executor).run("deploy the release") | |
| print(episode.render()) | |
| print() | |
| print("and what my executor was handed:", executor.saw) | |
| """) | |
| term("brain-mock", "Decisions", | |
| "MockBrain: replay a fixed script, exactly once", | |
| ["MockBrain", "MockBrain.decide", "MockBrain.remaining"], | |
| note="Deterministic and offline - the workhorse for tests, demos and replays.", | |
| code=""" | |
| from tiny_computer_agent import Element, MockBrain, Observation | |
| brain = MockBrain([ | |
| {"kind": "click", "label": "Save"}, | |
| {"kind": "key", "keys": "ctrl+s"}, | |
| {"kind": "done", "reason": "the document is saved"}, | |
| ]) | |
| frame = Observation(elements=(Element("Save", x=320, y=120, width=80, height=24),)) | |
| print("scripted actions:", 3) | |
| while brain.remaining: | |
| action = brain.decide("save the document", frame) | |
| print(f" remaining={brain.remaining} -> {action.describe()}") | |
| print() | |
| print("with the script exhausted it stops claiming work it has not planned:") | |
| print(" ", brain.decide("save the document", frame).describe()) | |
| """) | |
| term("brain-planner", "Decisions", | |
| "PlannerBrain: no weights, no download, and it gives up honestly", | |
| ["PlannerBrain", "PlannerBrain.decide", "GoalUnreachable"], | |
| note="The proof the harness works end to end, and a real fallback when no model is wanted.", | |
| code=""" | |
| from tiny_computer_agent import Element, FakeScreen, GoalUnreachable, PlannerBrain | |
| brain = PlannerBrain() | |
| goal = "click 'Save' to save the document" | |
| frames = FakeScreen([(Element("Files", x=10, y=6, width=60, height=20),), | |
| (Element("Save", x=320, y=120, width=80, height=24),)]) | |
| for step in range(3): | |
| observation = frames.capture() | |
| chosen = brain.decide(goal, observation) | |
| print(f" step {step}: {chosen.describe()}") | |
| print() | |
| print("when the target never appears it gives up rather than flailing:") | |
| try: | |
| PlannerBrain(max_waits=0).decide(goal, FakeScreen([(Element("Files"),)]).capture()) | |
| except GoalUnreachable as exc: | |
| print(" GoalUnreachable:", exc) | |
| """) | |
| term("models-swap", "Models", | |
| "The two real model backends, and lazy loading", | |
| ["TextPlannerBrain", "VLMBrain", "TextPlannerBrain.unload", "VLMBrain.unload", | |
| "ModelBrain.loaded", "make_model_brain"], | |
| note="Constructing a brain costs nothing; the weights arrive on the first decide().", | |
| code=""" | |
| from tiny_computer_agent import make_model_brain | |
| for name in ["text-planner", "vlm"]: | |
| brain = make_model_brain(name) | |
| print(f"{name:13} -> {type(brain).__name__}") | |
| print(f" model id : {brain.model_id}") | |
| print(f" weights loaded : {brain.loaded}") | |
| print(f" needs an index : {brain.requires_index}") | |
| print(f" menu size cap : {brain.max_menu_items} items") | |
| brain.unload() | |
| print(f" after unload() : loaded={brain.loaded}") | |
| print() | |
| print("neither download happens until decide(), which is what makes --doctor cheap") | |
| """) | |
| term("fake-desktop", "Screens", | |
| "The built-in synthetic desktop", | |
| ["fake_desktop", "write_fake_desktop", "OcrDetector"], | |
| note="Draws a screenshot-like desktop offline - the same image --image example replays.", | |
| code=""" | |
| import tempfile | |
| from pathlib import Path | |
| from tiny_computer_agent import OcrDetector, fake_desktop, write_fake_desktop | |
| image = fake_desktop() | |
| print("fake_desktop() ->", type(image).__name__, image.size, image.mode) | |
| path = Path(tempfile.mkdtemp()) / "example.png" | |
| print("write_fake_desktop(path) ->", write_fake_desktop(path).name, f"({path.stat().st_size} bytes)") | |
| print() | |
| print("and OCR reads it like any other screenshot:") | |
| for element in OcrDetector().detect(image): | |
| print(f" {element.label!r:34} at {element.center}") | |
| """) | |
| # =========================================================================== # | |
| # 14. closing the last gaps in the picture map | |
| # =========================================================================== # | |
| # names the existing pictures already exercise but did not claim | |
| _AUGMENT = { | |
| "vocab-verbs": ["Action.to_dict"], | |
| "exec-refuses-unresolved": ["InputUnavailable"], | |
| "scenario": ["Trace.render"], | |
| "trace-record": ["Trace.name", "Trace.decision_step", "Trace.blocked", "TraceStep", | |
| "TraceStep.observation", "TraceStep.to_dict"], | |
| "scoreboard": ["Scoreboard.add", "Scoreboard.total", "Scoreboard.earned", | |
| "Scoreboard.accuracy_for", "Scoreboard.brains"], | |
| "cli-script": ["demo_screen"], | |
| "cli-doctor": ["report"], | |
| } | |
| for _shot in CATALOG: | |
| _extra = _AUGMENT.get(_shot.id) | |
| if _extra: | |
| _shot.shows = sorted(set(_shot.shows) | set(_extra)) | |
| # and the two demos whose output should actually print what they claim | |
| def _edit(sid, old, new): | |
| for _shot in CATALOG: | |
| if _shot.id == sid: | |
| assert old in _shot.code, f"{sid}: anchor not found" | |
| _shot.code = _shot.code.replace(old, new) | |
| return | |
| raise AssertionError(sid) | |
| _edit("vocab-verbs", | |
| ' print(f" parsed -> {action.describe()}")', | |
| ' print(f" parsed -> {action.describe()}")\n' | |
| ' if payload["kind"] == "click":\n' | |
| ' print(f" as data: {action.to_dict()}")') | |
| _edit("trace-record", | |
| 'print("frames:", trace.frames, "| choice:", trace.choice,', | |
| 'print("name:", trace.name, "| decision_step:", trace.decision_step().action.describe())\n' | |
| 'step = trace.steps[0]\n' | |
| 'labels = [e.label for e in step.observation().elements]\n' | |
| 'print("first step as data:", step.to_dict()["action"], "| labels:", labels)\n' | |
| 'print("frames:", trace.frames, "| choice:", trace.choice,') | |
| _edit("scoreboard", | |
| 'print("accuracy:", f"{board.accuracy:.0%}", "| by brain:", board.by_brain())', | |
| 'print("total:", board.total, "| earned:", board.earned, "| accuracy:", f"{board.accuracy:.0%}")\n' | |
| 'print("brains:", board.brains(), "| accuracy_for(\'planner\'):", f"{board.accuracy_for(\'planner\'):.0%}")') | |
| _edit("exec-refuses-unresolved", | |
| 'print()\nresolved = executor.execute(Action("done", reason="nothing to do"), frame)', | |
| 'print()\nfrom tiny_computer_agent import InputUnavailable\n' | |
| 'print("and when pyautogui itself is missing, it says so:")\n' | |
| 'print(" ", InputUnavailable("real input needs pyautogui and a live display: ' | |
| 'pip install \'tiny-computer-agent[desktop]\'"))\nprint()\n' | |
| 'resolved = executor.execute(Action("done", reason="nothing to do"), frame)') | |
| # =========================================================================== # | |
| # 15. the stepwise loop, the masked decoder, and the accessibility tree | |
| # =========================================================================== # | |
| term("stepwise", "The loop", | |
| "The same loop, one turn at a time, with pause and resume", | |
| ["AgentLoop", "AgentLoop.begin", "AgentLoop.step_once", "AgentLoop.pause", | |
| "AgentLoop.resume", "AgentLoop.finished", "AgentLoop.run", "Step.line"], | |
| note="run() is batch; a console needs the caller to own the loop. These are the same rules.", | |
| code=""" | |
| from tiny_computer_agent import AgentLoop, Element, FakeScreen, MockBrain, Policy | |
| screen = FakeScreen([ | |
| (Element("Files", x=10, y=6, width=60, height=20),), | |
| (Element("Save", x=320, y=120, width=80, height=24),), | |
| ]) | |
| loop = AgentLoop( | |
| screen=screen, | |
| brain=MockBrain([{"kind": "wait", "seconds": 0.1}, | |
| {"kind": "click", "label": "Save"}, | |
| {"kind": "done", "reason": "saved"}]), | |
| policy=Policy(step_budget=6), | |
| ) | |
| loop.begin("save the report") | |
| print("after begin(): finished =", loop.finished, "| steps =", len(loop.episode.steps)) | |
| print("step_once() -> ", loop.step_once().line()) | |
| print(" finished =", loop.finished, "| steps =", len(loop.episode.steps)) | |
| loop.pause() | |
| loop.run() # returns at once: the caller said pause | |
| print("after pause(): paused =", loop.paused, "| steps =", len(loop.episode.steps)) | |
| loop.resume() | |
| episode = loop.run() | |
| print("after resume(): finished =", loop.finished, "| success =", episode.success) | |
| print() | |
| print(episode.render()) | |
| """) | |
| term("decoding", "Models", | |
| "Constrained decoding: make an echo impossible, not merely detected", | |
| ["MenuMask", "digit_token_ids", "choice_token_ids", "pair_token_ids", "read_choice", | |
| "read_verdict", "brevity_ladder", "majority_choice", "DIGITS"], | |
| note="The 256M model's failure was answering with the prompt. A mask removes that option.", | |
| code=""" | |
| from tiny_computer_agent.decoding import (DIGITS, MenuMask, brevity_ladder, | |
| choice_token_ids, digit_token_ids, | |
| majority_choice, pair_token_ids, read_choice, | |
| read_verdict) | |
| from tiny_computer_agent.models import DEFAULT_VISION_MODEL | |
| try: | |
| from transformers import AutoTokenizer | |
| tokenizer = AutoTokenizer.from_pretrained(DEFAULT_VISION_MODEL) | |
| source = DEFAULT_VISION_MODEL | |
| except Exception as exc: # pragma: no cover - env | |
| class _StandIn: | |
| table = {str(d): 100 + d for d in range(10)} | |
| def encode(self, text, add_special_tokens=False): | |
| return [self.table[text]] if text in self.table else [7, 7] | |
| tokenizer, source = _StandIn(), f"a stand-in ({type(exc).__name__})" | |
| print("tokenizer:", source) | |
| print("digits we allow :", DIGITS) | |
| print("digit token ids :", digit_token_ids(tokenizer)[:10], "...") | |
| print("ids for a 4-item menu:", choice_token_ids(tokenizer, 4)) | |
| yes_ids, no_ids = pair_token_ids(tokenizer) | |
| print("yes / no token ids :", yes_ids[:3], "/", no_ids[:3]) | |
| mask = MenuMask(choice_token_ids(tokenizer, 4), eos_token_id=2) | |
| print("the mask allows :", mask.allowed_ids, "(everything else is -inf)") | |
| print() | |
| print("reading an answer, and refusing anything that is not one:") | |
| for reply in ["3", " 3 ", "1000", "The goal is to click Save", "3 or 4", ""]: | |
| print(f" read_choice({reply[:26]!r:30}) -> {read_choice(reply, 4)}") | |
| for reply in ["Yes", " no", "maybe", ""]: | |
| print(f" read_verdict({reply!r:9}) -> {read_verdict(reply)!r}") | |
| print() | |
| print("the retry ladder (echoing grows with prompt length):", brevity_ladder(3)) | |
| print("a vote over samples:", majority_choice([2, 3, 2]), "| ties go to the first seen:", | |
| majority_choice([3, 2])) | |
| """) | |
| # --- the accessibility half needs a real GTK window and a session bus -------- | |
| _GTK = {"process": None} | |
| GTK_BOX = (140, 140, 660, 420) | |
| def open_gtk_scene(): | |
| """A session bus plus the GTK window, so AT-SPI has something to read.""" | |
| if _GTK["process"] is not None: | |
| return | |
| address = subprocess.run(["dbus-daemon", "--session", "--fork", "--print-address=1"], | |
| capture_output=True, text=True).stdout.strip() | |
| os.environ["DBUS_SESSION_BUS_ADDRESS"] = address | |
| os.environ["NO_AT_BRIDGE"] = "0" | |
| ready = Path("/tmp/gallery-gtk.ready") | |
| ready.unlink(missing_ok=True) | |
| _GTK["process"] = subprocess.Popen( | |
| [PY, str(ROOT / "showcase_scenes" / "gtk_dialog.py"), | |
| "/tmp/gallery-gtk.marker", str(ready)], | |
| env=dict(os.environ), stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) | |
| for _ in range(60): | |
| if ready.exists(): | |
| break | |
| time.sleep(0.25) | |
| time.sleep(2.5) | |
| def close_gtk_scene(): | |
| process, _GTK["process"] = _GTK["process"], None | |
| if process is not None: | |
| process.terminate() | |
| try: | |
| process.wait(timeout=10) | |
| except subprocess.TimeoutExpired: # pragma: no cover | |
| process.kill() | |
| time.sleep(0.5) | |
| def _with_gtk(tmp=None): | |
| open_gtk_scene() | |
| return close_gtk_scene | |
| term("atspi-index", "Perception", | |
| "The accessibility tree: elements the OS already knows about", | |
| ["AtspiDetector", "AtspiDetector.detect", "AtspiDetector.text", | |
| "AtspiDetector.coordinates_of_last_detect", "AtspiReader", | |
| "AtspiReader.read", "AccessibleNode", "TreeReader", "TreeReader.read", | |
| "accessibility_available", | |
| "INTERACTIVE_ROLES", "TEXT_ROLES"], | |
| note="No pixels are read: GTK publishes every button's name and screen extents.", | |
| prepare=_with_gtk, | |
| code=""" | |
| from tiny_computer_agent import (INTERACTIVE_ROLES, TEXT_ROLES, AtspiDetector, AtspiReader, | |
| accessibility_available) | |
| available, detail = accessibility_available("atspi") | |
| print("accessibility_available('atspi') ->", available) | |
| print(" ", detail) | |
| print() | |
| print("interactive roles offered to a model:") | |
| print(" ", ", ".join(sorted(INTERACTIVE_ROLES))) | |
| print("text roles - readable, but not clickable:") | |
| print(" ", ", ".join(sorted(TEXT_ROLES))) | |
| print() | |
| reader = AtspiReader() | |
| nodes = reader.read() | |
| print(f"AtspiReader().read() -> {len(nodes)} AccessibleNode(s) on the live desktop") | |
| print() | |
| detector = AtspiDetector(reader=reader) | |
| elements = detector.detect() | |
| print(f"AtspiDetector().detect() -> {len(elements)} clickable elements, with no OCR at all:") | |
| for element in elements: | |
| print(f" {element.label!r:12} kind={element.kind:8} " | |
| f"centre={str(element.center):12} box={element.width}x{element.height}") | |
| print() | |
| print("the words on screen, straight from the tree:") | |
| print(" ", detector.text()[:150]) | |
| """) | |
| term("models-decision-modes", "Models", | |
| "Four ways to ask a tiny model, and what each one costs", | |
| ["ModelBrain", "ModelBrain.loaded", "TextPlannerBrain", "VLMBrain", "make_model_brain"], | |
| note="json=free text, choice=one masked number, scan=yes/no per item, rank=log-odds order.", | |
| code=""" | |
| from tiny_computer_agent import ModelBrain, make_model_brain | |
| from tiny_computer_agent.models import DEFAULT_TEXT_MODEL, DEFAULT_VISION_MODEL | |
| print("decision_mode decides how the model is asked:") | |
| print(" json - reply with JSON, parsed leniently (the original behaviour)") | |
| print(" choice - one generation, masked to a menu number") | |
| print(" scan - one masked yes/no question per item, first yes wins") | |
| print(" rank - the same questions, ranked by the yes/no log-odds margin") | |
| print() | |
| for name in ("text-planner", "vlm"): | |
| brain = make_model_brain(name, decision_mode="scan") | |
| print(f"{name:13} -> {type(brain).__name__}") | |
| print(f" model : {brain.model_id}") | |
| print(f" supports masking : {brain.supports_constrained}") | |
| print(f" decision_mode : {brain.decision_mode} | retries {brain.retries} | votes {brain.votes}") | |
| print(f" weights loaded : {brain.loaded}") | |
| brain.unload() | |
| print() | |
| print("defaults are unchanged:", make_model_brain("vlm").decision_mode) | |
| """) | |
| # --- the accessibility tree, on a real desktop, with a real click ------------ | |
| def make_atspi_click(tmp): | |
| from tiny_computer_agent import (AgentLoop, AtspiDetector, GrabScreen, IndexedScreen, | |
| PlannerBrain, Policy, PyAutoGuiExecutor) | |
| open_gtk_scene() | |
| try: | |
| elements = AtspiDetector().detect() | |
| before = crop(annotate(grab(), elements), GTK_BOX) | |
| episode = AgentLoop( | |
| screen=IndexedScreen(GrabScreen(monitor=1), AtspiDetector()), | |
| brain=PlannerBrain(), executor=PyAutoGuiExecutor(), | |
| policy=Policy(dry_run=False, step_budget=3), | |
| ).run("click 'Save' to save the document") | |
| target = next((e for e in elements if e.matches("Save")), None) | |
| after = crop(annotate(grab(), [], click=target.center if target else None), GTK_BOX) | |
| marker = Path("/tmp/gallery-gtk.marker") | |
| lines = [ | |
| "the window's own status label now reads 'last action: Save',", | |
| "and it wrote " + (repr(marker.read_text()) if marker.exists() else "nothing") | |
| + " to " + marker.name, | |
| "result: " + ("success" if episode.success else "failure") | |
| + " - " + episode.stop_reason, | |
| ] | |
| return overlay_lines(side_by_side( | |
| before, after, | |
| title="AT-SPI index -> a real click: not one pixel was read to find the button"), lines) | |
| finally: | |
| close_gtk_scene() | |
| desk("desk-atspi", "Real desktop", | |
| "Indexed by the accessibility tree, acted on with the real mouse", | |
| ["AtspiDetector", "AtspiDetector.detect", "AtspiReader", "IndexedScreen", | |
| "PyAutoGuiExecutor", "PlannerBrain"], | |
| note="Roles and extents come from GTK itself; the click is a real one and the app reacts.", | |
| scene_name="gtk", make=make_atspi_click) | |
| term("uia-mapping", "Perception", | |
| "The Windows backend: the same index, from UI Automation", | |
| ["UiaReader", "UiaReader.read", "UiaDetector", "UiaDetector.detect", "UiaDetector.text", | |
| "UiaDetector.coordinates_of_last_detect"], | |
| note="Windows-only, so the mapping is exercised here against a synthetic control tree.", | |
| code=""" | |
| from tiny_computer_agent import UiaDetector, UiaReader, accessibility_available | |
| print("on this machine:", accessibility_available("uia")) | |
| print() | |
| class FakeBox: | |
| def __init__(self, left, top, right, bottom): | |
| self.left, self.top, self.right, self.bottom = left, top, right, bottom | |
| class FakeControl: | |
| \"\"\"The slice of a UI Automation element the reader actually uses.\"\"\" | |
| def __init__(self, name, control_type, box=None, children=()): | |
| self.Name = name | |
| self.ControlTypeName = control_type | |
| self.BoundingRectangle = box | |
| self._children = list(children) | |
| def GetChildren(self): | |
| return self._children | |
| def tree(): | |
| return FakeControl("Accessible desktop", "WindowControl", FakeBox(160, 160, 620, 380), [ | |
| FakeControl("", "PaneControl", FakeBox(174, 174, 606, 366), [ | |
| FakeControl("Files", "ButtonControl", FakeBox(174, 201, 274, 237)), | |
| FakeControl("Terminal", "ButtonControl", FakeBox(282, 201, 382, 237)), | |
| FakeControl("Save", "ButtonControl", FakeBox(390, 201, 490, 237)), | |
| FakeControl("Cancel", "ButtonControl", FakeBox(498, 201, 598, 237)), | |
| FakeControl("Choose an action", "TextControl", FakeBox(174, 245, 598, 264)), | |
| ]), | |
| ]) | |
| reader = UiaReader(provider=tree) | |
| print(f"UiaReader().read() -> {len(reader.read())} AccessibleNode(s), " | |
| f"control types mapped to roles") | |
| print() | |
| detector = UiaDetector(reader=reader) | |
| print("UiaDetector().detect() -> only the things worth clicking:") | |
| for element in detector.detect(): | |
| print(f" {element.label!r:12} kind={element.kind:8} centre={str(element.center):12} " | |
| f"box={element.width}x{element.height}") | |
| print() | |
| print("the window and the (unnamed) pane are read, and deliberately not offered:") | |
| print(" ", [node.role for node in reader.read()]) | |
| print() | |
| print("UiaDetector().text() ->", detector.text()[:80]) | |
| """) | |