File size: 8,285 Bytes
59022ca e2f048d 59022ca e2f048d 59022ca e2f048d 59022ca e2f048d 59022ca e2f048d 59022ca e2f048d | 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 | """Wiring for a real desktop: preflight checks, then one-call agent construction.
This module exists because "make it work on my computer" is mostly a diagnosis
problem. Rather than failing deep inside an episode, :func:`doctor` reports exactly
which piece is missing - display, capture backend, input backend, OCR engine, model
runtime - and what to do about each.
"""
from __future__ import annotations
import importlib.util
import os
import platform
from dataclasses import dataclass
from typing import Any, Callable, Iterable, Sequence
from tiny_computer_agent.act import DryRunExecutor, Executor, PyAutoGuiExecutor
from tiny_computer_agent.brain import Brain, PlannerBrain
from tiny_computer_agent.loop import AgentLoop, Step
from tiny_computer_agent.observe import GrabScreen, ScreenSource
from tiny_computer_agent.perception import (
Detector, IndexedScreen, detector_available, make_detector,
)
from tiny_computer_agent.gates import Gate
from tiny_computer_agent.safety import Policy
from tiny_computer_agent.skills import SkillBrain, SkillLibrary
DEFAULT_KILL_SWITCH = "/tmp/tiny-computer-agent.STOP"
BRAIN_NAMES = ("planner", "text-planner", "vlm")
@dataclass(frozen=True)
class Check:
"""One preflight result: what was checked, whether it works, and how to fix it."""
name: str
ok: bool
detail: str
fix: str = ""
required: bool = True
def render(self) -> str:
marker = "ok " if self.ok else ("FAIL" if self.required else "warn")
line = f" [{marker}] {self.name}: {self.detail}"
if not self.ok and self.fix:
line += f"\n fix: {self.fix}"
return line
def _installed(module: str) -> bool:
return importlib.util.find_spec(module) is not None
def display_available() -> tuple[bool, str]:
"""Whether a windowing session looks reachable from this process."""
system = platform.system()
if system in ("Darwin", "Windows"):
return True, f"{system} desktop session assumed"
display = os.environ.get("DISPLAY") or os.environ.get("WAYLAND_DISPLAY")
if display:
return True, f"DISPLAY={display}"
return False, "no DISPLAY or WAYLAND_DISPLAY set (headless container or SSH session)"
def run_checks(*, brain: str = "planner", model_id: str | None = None,
detector: str = "ocr", real_input: bool = False,
kill_switch_path: str | None = DEFAULT_KILL_SWITCH) -> list[Check]:
"""Every precondition for driving a real desktop, in one list."""
checks: list[Check] = []
ok, detail = display_available()
checks.append(Check(
"display server", ok, detail,
fix="run this on the desktop session itself (a headless server cannot be driven)",
))
checks.append(Check(
"pillow", _installed("PIL"),
"installed" if _installed("PIL") else "not installed",
fix="pip install 'tiny-computer-agent[desktop]'",
))
has_mss = _installed("mss")
checks.append(Check(
"screen capture", has_mss or _installed("PIL"),
"mss (fast, multi-monitor)" if has_mss else "pillow ImageGrab fallback",
fix="pip install mss for faster multi-monitor capture",
required=False,
))
has_pyautogui = _installed("pyautogui")
checks.append(Check(
"input backend (pyautogui)", has_pyautogui,
"installed" if has_pyautogui else "not installed",
fix="pip install 'tiny-computer-agent[desktop]'",
required=real_input,
))
detector_ok, detector_detail = detector_available(detector)
checks.append(Check(
f"element detector ({detector})", detector_ok, detector_detail,
fix="pip install 'tiny-computer-agent[ocr]'",
))
if brain == "planner":
checks.append(Check(
"model runtime", True,
"not needed: the planner is rule-based (no weights, no download)",
required=False,
))
else:
from tiny_computer_agent.models import (
DEFAULT_TEXT_MODEL, DEFAULT_VISION_MODEL, model_cached, model_dependencies,
)
deps_ok, deps_detail = model_dependencies()
model = model_id or (DEFAULT_TEXT_MODEL if brain == "text-planner" else DEFAULT_VISION_MODEL)
checks.append(Check(
"model runtime", deps_ok, deps_detail,
fix="pip install 'tiny-computer-agent[vision]'",
))
checks.append(Check(
f"model weights ({model})",
model_cached(model),
"cached locally" if model_cached(model) else "not downloaded yet",
fix=f"first run will download {model} from the Hub (one time)",
required=False,
))
checks.append(Check(
"execution mode",
True,
"REAL mouse and keyboard input" if real_input else "dry run (nothing is clicked)",
fix="pass --real-input AND set TINY_COMPUTER_AGENT_ALLOW_REAL_INPUT=1",
required=False,
))
checks.append(Check(
"kill switch", True,
f"create {kill_switch_path} to abort the next action"
if kill_switch_path else "disabled",
required=False,
))
return checks
def doctor(*, brain: str = "planner", model_id: str | None = None,
detector: str = "ocr", real_input: bool = False,
kill_switch_path: str | None = DEFAULT_KILL_SWITCH) -> str:
"""A human-readable preflight report."""
checks = run_checks(brain=brain, model_id=model_id, detector=detector,
real_input=real_input, kill_switch_path=kill_switch_path)
blocking = [c for c in checks if c.required and not c.ok]
lines = [
f"tiny-computer-agent preflight ({platform.system()} {platform.machine()}, python {platform.python_version()})",
"",
*(c.render() for c in checks),
"",
]
if blocking:
lines.append(f"NOT READY: {len(blocking)} blocking problem(s): "
+ ", ".join(c.name for c in blocking))
else:
lines.append("READY: the harness can drive this desktop.")
if not real_input:
lines.append("Dry run is on: nothing will be clicked. Read the plan first.")
return "\n".join(lines)
def build_agent(*, brain: Brain | str | None = None, screen: ScreenSource | None = None,
detector: Detector | str | None = None, dry_run: bool = True,
max_steps: int = 10, kill_switch_path: str | None = DEFAULT_KILL_SWITCH,
allow_grid_clicks: bool = False,
on_step: Callable[[Step], None] | None = None,
detector_options: dict[str, Any] | None = None,
skills: SkillLibrary | None = None,
gate: Gate | None = None) -> AgentLoop:
"""Wire a real desktop into an :class:`AgentLoop` in one call.
Defaults are the safe ones: dry run, a step budget of 10, and a kill-switch file
on disk that aborts the next action if it appears. ``screen``/``detector``/
``brain`` can be injected, which is how the tests drive this without a desktop.
Pass ``skills`` and the brain becomes a :class:`~tiny_computer_agent.skills.SkillBrain`
that answers from the library first and only falls back to the model when no learned
skill applies; pass ``gate`` and the operator stands in front of every action.
"""
if detector is None or isinstance(detector, str):
detector = make_detector(detector or "ocr", **(detector_options or {}))
if screen is None:
screen = IndexedScreen(GrabScreen(), detector)
if brain is None or isinstance(brain, str):
name = brain or "planner"
if name == "planner":
brain = PlannerBrain()
else:
from tiny_computer_agent.models import make_model_brain
brain = make_model_brain(name)
if skills is not None:
brain = SkillBrain(skills, fallback=brain)
executor: Executor = DryRunExecutor() if dry_run else PyAutoGuiExecutor()
policy = Policy(dry_run=dry_run, step_budget=max_steps,
kill_switch_path=kill_switch_path,
allow_grid_clicks=allow_grid_clicks)
return AgentLoop(screen=screen, brain=brain, executor=executor, policy=policy,
on_step=on_step, gate=gate)
|