← back to Jev Ultrafast
jev_ultrafast/agent.py
175 lines
"""The complete agent loop. Typed choices, observable state, bounded execution."""
import base64
import time
from pathlib import Path
from .browser import Browser, StalePage
from .model import action_space, choose, field_context, field_text
from .questions import MAX_STEPS
class Agent:
def __init__(self, url, goals, *, record_dir=None, screenshots=False):
task = goals.strip() if isinstance(goals, str) else "\n".join(goals).strip()
if not task:
raise ValueError("Supply a task")
plan = [task]
self.pending_text = None
self.browser = Browser(url)
self.record_dir = Path(record_dir) if record_dir else None
self.screenshots = screenshots or bool(record_dir)
try:
page = self.browser.observe(screenshot=self.screenshots)
except Exception:
self.browser.close()
raise
self.state = dict(
browser=self.browser,
goal="\n".join(plan),
page=page,
decision=None,
history=[],
status="ready",
plan=plan,
plan_index=0,
decisions=[],
text_calls=[],
elapsed_ms=0,
started_at=None,
record=bool(self.record_dir),
)
if self.record_dir:
self.record_dir.mkdir(parents=True, exist_ok=True)
(self.record_dir / "000000.jpg").write_bytes(base64.b64decode(page["screenshot"]))
def snapshot(self):
return {
**{k: v for k, v in self.state.items() if k != "browser"},
"elements": action_space(self.state["page"]["actions"])[0],
}
def command(self, name, body=None):
body = body or {}
state = self.state
if name == "tick":
try:
self.command("predict", {})
return self.command("act", {"fingerprint": state["page"]["fingerprint"]})
except StalePage:
state["decision"] = None
state["status"] = "ready"
state["page"] = state["browser"].observe(screenshot=self.screenshots)
state["elapsed_ms"] = round((time.perf_counter() - state["started_at"]) * 1000)
return self.snapshot()
elif name == "predict":
if not state["browser"]:
raise ValueError("Start a demo first")
if state["started_at"] is None:
state["started_at"] = time.perf_counter()
if not state["browser"].fresh(state["page"]):
state["page"] = state["browser"].observe(screenshot=self.screenshots)
state["decision"] = None
if state["status"] in {"done", "blocked"}:
raise ValueError("This run has stopped. Start a fresh demo.")
if len(state["decisions"]) >= MAX_STEPS * 2:
raise ValueError("Reached the demo's model-call budget")
state["decision"] = choose(state["page"], state["goal"], state["history"])
state["decisions"].append(
{
**state["decision"],
"fingerprint": state["page"]["fingerprint"],
"elapsed_ms": round((time.perf_counter() - state["started_at"]) * 1000),
}
)
state["status"] = "predicted"
elif name == "act":
decision, page = state["decision"], state["page"]
if not decision or body.get("fingerprint") != page["fingerprint"]:
raise ValueError("Observe and choose before acting")
# Consume once, before any mutation or model call. A retry cannot double-click.
state["decision"] = None
selected = decision["choice"]
if selected in {"DONE", "BLOCKED"}:
if not state["browser"].fresh(page):
state["status"] = "ready"
raise StalePage("Page changed since the decision. Choose again.")
state["status"] = "done" if selected == "DONE" else "blocked"
state["plan_index"] = int(selected == "DONE")
state["elapsed_ms"] = round((time.perf_counter() - state["started_at"]) * 1000)
return self.snapshot()
action = next(a for a in page["actions"] if a["id"] == selected)
if len(state["history"]) >= MAX_STEPS:
state["status"] = "blocked"
raise ValueError(f"Stopped at the {MAX_STEPS}-action demo budget")
text, helper = None, None
if action["kind"] == "fill":
if not state["browser"].fresh(page):
raise StalePage("Page changed before text generation. Choose again.")
context = field_context(state["goal"], action, page, state["history"])
if self.pending_text and self.pending_text[0] == context:
_, text, helper = self.pending_text
else:
text, helper = field_text(context)
self.pending_text = (context, text, helper)
state["text_calls"].append({**helper, "field": action["label"], "value": text})
# Browser.act checks freshness immediately before input, including after text generation.
state["browser"].act(action, page, text=text)
self.pending_text = None
state["elapsed_ms"] = round((time.perf_counter() - state["started_at"]) * 1000)
# Record execution before observing. A stale post-action observation must not erase the action.
state["history"].append(
{
"step": len(state["history"]) + 1,
"action": action["label"],
"kind": action["kind"],
"choice": selected,
"probability": decision["probabilities"][selected],
"confidence": decision["confidence"],
"latency_ms": decision["latency_ms"],
"text": text,
"text_helper": helper["model"] if helper else None,
"text_latency_ms": helper["latency_ms"] if helper else 0,
"operation": decision["operation"],
"target": decision["target"],
"page_changed": None,
"url": page["url"],
"usage": decision["usage"],
"executed_ms": round((time.perf_counter() - state["started_at"]) * 1000),
"elapsed_ms": state["elapsed_ms"],
}
)
state["page"] = state["browser"].observe(screenshot=self.screenshots)
state["elapsed_ms"] = round((time.perf_counter() - state["started_at"]) * 1000)
state["history"][-1].update(
page_changed=state["page"]["fingerprint"] != page["fingerprint"],
url=state["page"]["url"],
elapsed_ms=state["elapsed_ms"],
)
if state["record"]:
(self.record_dir / f"{state['elapsed_ms']:06d}.jpg").write_bytes(
base64.b64decode(state["page"]["screenshot"])
)
repeated = state["history"][-3:]
state["status"] = (
"blocked"
if len(repeated) == 3 and all(h["page_changed"] is False and h["kind"] != "wait" for h in repeated)
else "ready"
)
else:
raise ValueError("Unknown command")
return self.snapshot()
def run(self):
while self.state["status"] not in {"done", "blocked"}:
yield self.command("tick")
def close(self):
self.browser.close()
def __enter__(self):
return self
def __exit__(self, *_args):
self.close()