Spaces:
Running
Running
"""UI-TARS-1.5 and MolmoWeb as released, each in its own agent loop, on MiniWeb human tasks (EXPERIMENT_PLAN, RQ1;
2026-10-03). The counterpart of src/miniweb/evaluation/fara_runner.py for the two other own-loop baselines.
uitars UI-TARS-1.5-7B (ByteDance-Seed, Apache-2.0). Prompt and action parser are the released ones (pip `ui-tars`:
ui_tars.prompt.COMPUTER_USE_DOUBAO, ui_tars.action_parser.parse_action_to_structure_output). The loop is
OSWorld's UI-TARS-1.5 agent (mm_agents/uitars15_v1.py, run_multienv_uitars15_v1.py): system "You are a
helpful assistant.", the prompt, then the full response history with the last 5 screenshots, temperature 0,
top_p 0.9, max_tokens 500 (OSWorld's frequency_penalty 1 is off by default: see --frequency-penalty).
A failed parse is retried up to 3 times at temperature 1,
then the episode ends. Coordinates are in the Qwen2.5-VL smart_resize space of the screenshot and are mapped
back as parsing_response_to_pyautogui_code does. The pyautogui calls become Playwright mouse and keyboard
calls: type -> keyboard.type, plus Enter for a trailing \\n; scroll -> 5 wheel notches of 100 px at the
point; wait -> 5 s. The answer is finished(content=...).
molmoweb MolmoWeb-4B / -8B (AllenAI, Apache-2.0): the released agent code itself (github.com/allenai/molmoweb:
agent.multimodal_agent.MultimodalAgent, utils.envs SimpleEnv and its action executor,
utils.eval_utils.episode.Episode) with its benchmark settings: "molmo_web_think" prompts, the last 10 steps
as text, no past images, temperature 0.7, top_p 0.8, 1024 new tokens, a 1280x720 viewport. Only the
predictor is swapped: their HF predictor's single user turn (prompt text + screenshot), sent to a vLLM
OpenAI endpoint (the repo leaves vLLM to users). Serve it with causal attention on image tokens, as their
HF predictor runs it (vLLM's Molmo2 default is a bidirectional prefix LM): --hf-overrides
{"is_mm_prefix_lm":false}, which granite's ownloop.sh passes. The answer is read as their
benchmarks.evaluate does: the last "[ANSWER] ..." message, else the last other message, else the last
thought.
fara7b Fara-7B (Microsoft, MIT): the released previous-generation agent itself (github.com/microsoft/fara src/,
fara.fara_7b: FaraAgent, its BrowserBB browser manager and PlaywrightController), as `fara-cli --fara-7b`
and webeval's WebSurferSystem run it: 1440x900, the last 3 screenshots, temperature 0, single-tab mode,
their page script. The browser manager is subclassed only to launch our Chromium with our context, in
place of Firefox with an Edge user agent. The answer is theirs: the terminate step's thoughts. A parse or
action error ends the episode, as in the released loop, and the task is re-run up to 5 times, as webeval
(the harness of their published numbers) does by default.
opencua OpenCUA-7B (XLANG, MIT): OSWorld's OpenCUAAgent itself (xlang-ai/OSWorld mm_agents/opencua) with the
OpenCUA-7B command of run_multienv_opencua.py (--coordinate_type qwen25 --use_old_sys_prompt; l2 CoT,
action history, 3 images, temperature 0, top_p 0.9, max_tokens 2048) on a 1920x1080 page, and the loop of
lib_run_single.run_single_example_opencua. Only call_llm is swapped (a vLLM endpoint), and the VM's
`python -c` is a pyautogui-to-Playwright translation (run_pyautogui, literal calls only). The V1 prompt has
no answer field: the answer is the terminate step's answer= if given, else its Thought + Action text.
Serve with --trust-remote-code and a tokenizer dir whose tokenizer_config.json drops unk_token/pad_token
(the remote TikTokenV3 code rejects them under transformers 5; it derives the same [UNK]/[PAD] itself).
Both run in the MiniWeb env (Playwright sync API, DATAGEN_CHROME) and start their own MiniWeb servers on --ports, as
miniwebagent.evaluate does. The released code goes on sys.path through --path (default $OWNLOOP_PATH): a directory holding
`pip install --no-deps --target <dir> ui-tars==0.5.1`, and a clone of allenai/molmoweb. Setup per task is
miniwebagent.evaluate.prepare_human's: /_admin/session-flags (logged out for a login task), then the start page, clipboard
allowed, with the task wrapper of miniwebagent.harness.instruction. Tasks come from `python -m miniweb.evaluation.fara_rq1 prep`
({BASE} placeholders); grading is `python -m miniweb.evaluation.fara_rq1 grade --arm <name>`, from what this saves per task:
<out>/<task_id>/session.json {"base", "answer", "status", "rounds", "record": [...], "log": [...], "error", ...}
<out>/<task_id>/steps.jsonl one line per model call: raw output, parsed action, url, errors
<out>/<task_id>/screens/ step_NN.png: the screenshot each step saw, plus the final one
python -m miniweb.evaluation.ownloop_runner uitars --tasks <out>/tasks.json --out <out> --ports 8312,8313 \\
--endpoint http://127.0.0.1:8694/v1 --model uitars15-7b --path $W/ownloop_pkgs
python -m miniweb.evaluation.ownloop_runner molmoweb --tasks <out>/tasks.json --out <out> --ports 8314,8315 \\
--endpoint http://127.0.0.1:8695/v1 --model molmoweb-4b --path $W/ownloop_pkgs:$W/molmoweb
python -m miniweb.evaluation.ownloop_runner fara7b --tasks <out>/tasks.json --out <out> --ports 8312,8313 \\
--endpoint http://127.0.0.1:8696/v1 --model fara-7b --path $W/fara7b_pkgs:$W/fara/src
python -m miniweb.evaluation.ownloop_runner opencua --tasks <out>/tasks.json --out <out> --ports 8314,8315 \\
--endpoint http://127.0.0.1:8697/v1 --model opencua-7b --path $W/opencua_pkgs:$W/osworld
"""
import argparse
import base64
import io
import json
import math
import os
import re
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
LAUNCH_ARGS = ["--disable-dev-shm-usage", "--no-first-run", "--disable-blink-features=AutomationControlled"]
def instruction(task_text, start_url):
"""Same wrapper as miniwebagent.harness.instruction (our agents' task text) and src/miniweb/evaluation/fara_runner.py."""
return f"You are interacting with a web application at {start_url}. Your task: {task_text}"
def launch_browser(pw):
from miniwebgen import config
exe = config.CHROME if os.path.exists(config.CHROME) else None
return pw.chromium.launch(headless=True, executable_path=exe, args=LAUNCH_ARGS)
def new_context(browser, width, height):
"""miniwebgen.browser.new_context (clipboard granted, focus emulated, en-US / UTC) at this agent's viewport."""
from miniwebgen.browser import new_context as nc
return nc(browser, viewport=(width, height))
def admin_entries(ctx, base, path):
"""A privileged /_admin endpoint, read as this browser's session (the context's cookies)."""
r = ctx.request.get(base + path, timeout=30000)
return (r.json() or {}).get("entries", []) or []
def png_b64(png):
return "data:image/png;base64," + base64.b64encode(png).decode()
class EndpointError(RuntimeError):
"""The model server failed (not the model): the task is left unsaved, for a rerun."""
def chat(client, model, messages, **kw):
"""One chat completion; transport errors are retried (harness-level, not the agent's own retry)."""
for attempt in range(3):
try:
r = client.chat.completions.create(model=model, messages=messages, **kw)
return (r.choices[0].message.content or "").strip()
except Exception as e:
if attempt == 2:
raise EndpointError(f"{type(e).__name__}: {str(e)[:200]}") from e
time.sleep(5)
class Rec:
"""Per-task output: steps.jsonl lines and screenshots."""
def __init__(self, ep):
self.ep = ep
(ep / "screens").mkdir(parents=True, exist_ok=True)
self.f = open(ep / "steps.jsonl", "w")
def step(self, **row):
self.f.write(json.dumps(row, default=str) + "\n")
self.f.flush()
def shot(self, i, png):
(self.ep / "screens" / f"step_{i:02d}.png").write_bytes(png)
def close(self):
self.f.close()
# ── UI-TARS-1.5 ───────────────────────────────────────────────────────────────────────────────────────────────────────
PW_KEYS = {"ctrl": "Control", "control": "Control", "shift": "Shift", "alt": "Alt", "option": "Alt", "cmd": "Meta",
"command": "Meta", "win": "Meta", "meta": "Meta", "super": "Meta", "enter": "Enter", "return": "Enter",
"esc": "Escape", "escape": "Escape", "tab": "Tab", "backspace": "Backspace", "delete": "Delete",
"del": "Delete", "space": "Space", "up": "ArrowUp", "down": "ArrowDown", "left": "ArrowLeft",
"right": "ArrowRight", "arrowup": "ArrowUp", "arrowdown": "ArrowDown", "arrowleft": "ArrowLeft",
"arrowright": "ArrowRight", "pageup": "PageUp", "pagedown": "PageDown", "home": "Home", "end": "End",
"insert": "Insert", "capslock": "CapsLock"}
def pw_key(k):
k = k.strip()
if k.lower() in PW_KEYS:
return PW_KEYS[k.lower()]
if re.fullmatch(r"[fF]\d{1,2}", k):
return k.upper()
return k
class UITars:
"""OSWorld mm_agents/uitars15_v1.py UITARSAgent.predict (infer_mode qwen25vl_normal), with the repo's computer-use
prompt and an OpenAI-compatible endpoint."""
MAX_PIXELS, MIN_PIXELS = 16384 * 28 * 28, 100 * 28 * 28
def __init__(self, client, model, task, history_n=5, max_tokens=500, temperature=0.0, top_p=0.9,
frequency_penalty=0.0, language="English"):
from ui_tars.prompt import COMPUTER_USE_DOUBAO
self.client, self.model = client, model
self.prompt = COMPUTER_USE_DOUBAO.format(language=language, instruction=task)
self.history_n, self.max_tokens = history_n, max_tokens
self.temperature, self.top_p, self.frequency_penalty = temperature, top_p, frequency_penalty
self.images, self.responses = [], []
def _image(self, png):
"""The agent's own resize to [min_pixels, max_pixels] (a no-op at our viewports) -> (png, height, width)."""
from PIL import Image
im = Image.open(io.BytesIO(png))
if im.width * im.height > self.MAX_PIXELS:
f = math.sqrt(self.MAX_PIXELS / (im.width * im.height))
im = im.resize((int(im.width * f), int(im.height * f)))
if im.width * im.height < self.MIN_PIXELS:
f = math.sqrt(self.MIN_PIXELS / (im.width * im.height))
im = im.resize((math.ceil(im.width * f), math.ceil(im.height * f)))
if im.mode != "RGB":
im = im.convert("RGB")
buf = io.BytesIO()
im.save(buf, format="PNG")
return buf.getvalue(), im.height, im.width
def predict(self, png):
"""-> (response text or None, parsed actions or None, [failed attempts])."""
from ui_tars.action_parser import add_box_token, parse_action_to_structure_output
self.images = (self.images + [png])[-self.history_n:]
imgs = [self._image(b) for b in self.images]
messages = [{"role": "system", "content": [{"type": "text", "text": "You are a helpful assistant."}]},
{"role": "user", "content": [{"type": "text", "text": self.prompt}]}]
n = 0
for i, resp in enumerate(self.responses):
if i + self.history_n > len(self.responses): # at most history_n screenshots in all
messages.append({"role": "user", "content": [{"type": "image_url", "image_url": {"url": png_b64(imgs[n][0])}}]})
n += 1
messages.append({"role": "assistant", "content": [{"type": "text", "text": add_box_token(resp)}]})
messages.append({"role": "user", "content": [{"type": "image_url", "image_url": {"url": png_b64(imgs[n][0])}}]})
_, h, w = imgs[-1]
temperature, fails = self.temperature, []
for _ in range(3):
text = chat(self.client, self.model, messages, max_tokens=self.max_tokens, temperature=temperature,
top_p=self.top_p, frequency_penalty=self.frequency_penalty)
try:
parsed = parse_action_to_structure_output(text, 1000, h, w, "qwen25vl", self.MAX_PIXELS, self.MIN_PIXELS)
except Exception as e: # as OSWorld: resample at temperature 1
fails.append({"raw": text, "parse_error": f"{type(e).__name__}: {str(e)[:200]}"})
temperature = 1.0
continue
self.responses.append(text)
return text, parsed, fails
return None, None, fails
def _box_center(box, w, h):
x1, y1, x2, y2 = eval(box) if isinstance(box, str) else box # the parser's own "[x1, y1, x2, y2]" fractions
return round((x1 + x2) / 2 * w, 3), round((y1 + y2) / 2 * h, 3)
def uitars_execute(page, act, w, h):
"""One parsed UI-TARS action on the page, as parsing_response_to_pyautogui_code would run it. -> (stop, answer, note)."""
t, a = act.get("action_type"), act.get("action_inputs") or {}
if t == "finished":
return True, a.get("content", ""), ""
if t == "call_user":
return True, None, "call_user"
if t in ("click", "left_single", "left_double", "right_single", "hover"):
if not a.get("start_box"):
return False, None, "no start_box"
x, y = _box_center(a["start_box"], w, h)
if t == "left_double":
page.mouse.dblclick(x, y)
elif t == "right_single":
page.mouse.click(x, y, button="right")
elif t == "hover":
page.mouse.move(x, y)
else:
page.mouse.click(x, y)
elif t in ("drag", "select"):
if not (a.get("start_box") and a.get("end_box")):
return False, None, "drag without both boxes"
sx, sy = _box_center(a["start_box"], w, h)
ex, ey = _box_center(a["end_box"], w, h)
page.mouse.move(sx, sy)
page.mouse.down()
page.mouse.move(ex, ey, steps=20) # pyautogui.dragTo(duration=1.0)
page.mouse.up()
elif t == "hotkey":
keys = (a.get("key") or a.get("hotkey") or "").split()
if keys:
page.keyboard.press("+".join(pw_key(k) for k in keys))
elif t in ("press", "keydown", "keyup"):
k = pw_key(a.get("key") or a.get("press") or "")
if k:
(page.keyboard.up if t == "keyup" else page.keyboard.down)(k)
if t == "press":
page.keyboard.up(k)
elif t == "type":
content = a.get("content", "")
enter = content.endswith("\n") or content.endswith("\\n")
text = content.rstrip("\\n").rstrip("\n") if enter else content
if text:
page.keyboard.type(text)
if enter:
page.keyboard.press("Enter")
elif t == "scroll":
d = (a.get("direction") or "").lower()
if a.get("start_box"):
page.mouse.move(*_box_center(a["start_box"], w, h))
dx = -500 if "left" in d else 500 if "right" in d else 0
dy = -500 if "up" in d else 500 if "down" in d else 0
page.mouse.wheel(dx, dy)
elif t == "wait":
time.sleep(5)
else:
return False, None, f"unrecognized action {t}"
return False, None, ""
def settle(ctx, page):
"""-> the active page (a click may open a tab: follow the newest) after its DOM is loaded, plus a short buffer."""
pages = [p for p in ctx.pages if not p.is_closed()]
page = pages[-1] if pages else page
try:
page.wait_for_load_state("domcontentloaded", timeout=5000)
except Exception:
pass
time.sleep(1.0)
return page
def run_uitars(t, base, args, ep):
from openai import OpenAI
from playwright.sync_api import sync_playwright
start = t["start"].replace("{BASE}", base)
prompt = t["prompt"].replace("{BASE}", base)
flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "")
res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None,
"record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0}
t0, rec = time.time(), Rec(ep)
client = OpenAI(base_url=args.endpoint, api_key="not-needed", timeout=300)
agent = UITars(client, args.model, instruction(prompt, start), frequency_penalty=args.frequency_penalty)
with sync_playwright() as pw:
browser = launch_browser(pw)
ctx = new_context(browser, args.width, args.height)
try:
page = ctx.new_page()
page.goto(flags, wait_until="domcontentloaded", timeout=60000)
page.goto(start, wait_until="domcontentloaded", timeout=60000)
time.sleep(1.0)
for i in range(1, args.max_steps + 1):
if time.time() - t0 > args.timeout:
res["status"] = "timeout"
break
png = page.screenshot(type="png")
rec.shot(i, png)
text, parsed, fails = agent.predict(png)
res["rounds"] = i
res["parse_fails"] += sum("parse_error" in f for f in fails)
if parsed is None:
rec.step(step=i, url=page.url, raw=None, fails=fails)
res["status"] = "parse_error" # OSWorld: no parsable response in 3 tries ends it
break
stop, notes = False, []
for act in parsed:
try:
stop, answer, note = uitars_execute(page, act, args.width, args.height)
except Exception as e:
stop, answer, note = False, None, f"{type(e).__name__}: {str(e)[:200]}"
res["action_errors"] += bool(note) and note != "call_user"
notes.append(note)
if stop:
res["answer"], res["status"] = answer, "finished" if note != "call_user" else "call_user"
break
rec.step(step=i, url=page.url, raw=text, actions=[{k: v for k, v in a.items() if k != "text"} for a in parsed],
notes=notes, fails=fails)
if stop:
break
page = settle(ctx, page)
else:
res["status"] = "max_steps"
try:
rec.shot(res["rounds"] + 1, settle(ctx, page).screenshot(type="png"))
except Exception:
pass
except Exception as e: # a harness error still keeps whatever the session did
res["error"] = f"{type(e).__name__}: {str(e)[:300]}"
if isinstance(e, EndpointError):
res["status"] = "endpoint_error"
try:
res["record"] = admin_entries(ctx, base, "/_admin/record")
res["log"] = admin_entries(ctx, base, "/_admin/log")
except Exception as e:
res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}"
browser.close()
rec.close()
res["seconds"] = round(time.time() - t0, 1)
return res
# ── MolmoWeb ──────────────────────────────────────────────────────────────────────────────────────────────────────────
class VllmPredictor:
"""Their HFActionPredictor's request (one user turn: prompt text, then the screenshot(s); sampling with
temperature / top_p; max_new_tokens 1024), sent to an OpenAI-compatible endpoint. Stops the episode at the task
deadline: an exception here is the Episode's ACTION_PREDICTION_ERROR, which ends it."""
def __init__(self, client, model, temperature, top_p, deadline, max_tokens=1024):
self.client, self.model, self.deadline = client, model, deadline
self.temperature, self.top_p, self.max_tokens = temperature, top_p, max_tokens
self.failed = False
def predict(self, prompt, image_np, past_actions=None, **kw):
from PIL import Image
if time.time() > self.deadline:
raise TimeoutError("task timeout")
content = [{"type": "text", "text": prompt}]
for im in (image_np if isinstance(image_np, list) else [image_np]):
buf = io.BytesIO()
Image.fromarray(im.astype("uint8")).convert("RGB").save(buf, format="PNG")
content.append({"type": "image_url", "image_url": {"url": png_b64(buf.getvalue())}})
try:
text = chat(self.client, self.model, [{"role": "user", "content": content}], max_tokens=self.max_tokens,
temperature=self.temperature, top_p=self.top_p)
except EndpointError:
self.failed = True
raise
return text or None
def molmoweb_answer(interactions):
"""benchmarks.evaluate.read_trajectory's answer: the last "[ANSWER] ..." message, else the last message that is
not [EXIT], else the last thought."""
last_answer = last_msg = last_thought = None
for it in interactions:
if not it.action:
continue
act = it.action["action_output"].action
msg = getattr(act, "msg", None)
if msg is not None:
if "[ANSWER]" in msg:
last_answer = msg[9:]
elif "[EXIT]" not in msg:
last_msg = msg
if it.action.get("thought"):
last_thought = it.action["thought"]
return last_answer if last_answer is not None else last_msg if last_msg is not None else last_thought
def run_molmoweb(t, base, args, ep):
import numpy as np
from openai import OpenAI
from PIL import Image
from agent.actions import SendMsgToUser
from agent.multimodal_agent import MultimodalAgent
from utils.envs.browser_env import SimpleEnv, _start_playwright
from utils.eval_utils.episode import Episode
start = t["start"].replace("{BASE}", base)
prompt = t["prompt"].replace("{BASE}", base)
flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "")
class MiniWebEnv(SimpleEnv):
"""Their local-Chromium env with our Chromium build and context, opening the session flags first."""
def _launch(self):
self.playwright = _start_playwright()
self.browser = launch_browser(self.playwright)
self.context = new_context(self.browser, self.viewport_width, self.viewport_height)
self.page = self.context.new_page()
def _navigate_to_start(self, url, goal):
self.page.goto(flags, wait_until="domcontentloaded", timeout=60000)
self.page.goto(url, wait_until="domcontentloaded", timeout=60000)
time.sleep(1.0)
return goal
res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None,
"record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0}
t0, rec = time.time(), Rec(ep)
env = MiniWebEnv(start_url=start, goal=instruction(prompt, start), viewport_width=args.width,
viewport_height=args.height)
agent = MultimodalAgent(endpoint_or_checkpoint=args.endpoint, inference_mode="fastapi", system_message="molmo_web_think",
max_past_steps=args.max_past_steps, max_past_images=0,
sampling_temperature=args.temperature, sampling_top_p=args.top_p)
agent.predictor = VllmPredictor(OpenAI(base_url=args.endpoint, api_key="not-needed", timeout=300), args.model,
args.temperature, args.top_p, t0 + args.timeout)
episode = Episode(env=env, agent=agent, eps_name=t["task_id"])
try:
inters, _ = episode.run_episode(max_steps=args.max_steps)
except Exception as e:
inters = episode.interactions
res["error"] = f"{type(e).__name__}: {str(e)[:300]}"
def png(arr):
buf = io.BytesIO()
Image.fromarray(np.asarray(arr).astype("uint8")).save(buf, format="PNG")
return buf.getvalue()
for i, it in enumerate(inters, 1):
obs = it.state.get("obs") or {}
if obs.get("screenshot") is not None:
rec.shot(i, png(obs["screenshot"]))
a = it.action or {}
thought = a.get("thought") or ""
res["parse_fails"] += thought.startswith("Could not parse predicted action")
nxt = (it.next_state or {}).get("obs") or {}
res["action_errors"] += bool(nxt.get("last_action_error"))
rec.step(step=i, url=obs.get("url"), raw=it.raw_output, action=a.get("action"), action_str=a.get("action_str"),
thought=thought, error=it.error, action_error=nxt.get("last_action_error"))
if inters and (inters[-1].next_state or {}).get("obs", {}).get("screenshot") is not None:
rec.shot(len(inters) + 1, png(inters[-1].next_state["obs"]["screenshot"]))
res["rounds"] = len(inters)
res["answer"] = molmoweb_answer(inters)
last = inters[-1] if inters else None
if last is None:
res["status"] = "no_steps"
elif last.error:
err = json.dumps(last.error)
res["status"] = "timeout" if "task timeout" in err else "endpoint_error" if "EndpointError" in err or \
agent.predictor.failed else "error"
res["error"] = res["error"] or json.dumps(last.error)[:300]
elif last.action and isinstance(last.action["action_output"].action, SendMsgToUser) \
and (last.action["action_output"].action.msg or "").startswith("[EXIT]"):
res["status"] = "exit"
else:
res["status"] = "max_steps"
try:
if env.context is not None:
res["record"] = admin_entries(env.context, base, "/_admin/record")
res["log"] = admin_entries(env.context, base, "/_admin/log")
except Exception as e:
res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}"
try:
env.close()
except Exception:
pass
rec.close()
res["seconds"] = round(time.time() - t0, 1)
return res
# ── Fara-7B (added 2026-10-03) ────────────────────────────────────────────────────────────────────────────────────────
_IMPORT_LOCK = threading.Lock()
def _fara_package():
"""Register the `fara` package of the --path clone (github.com/microsoft/fara src/) without running its __init__.py,
which imports the Fara1.5 agent and environment stacks and their extra dependencies. fara.fara_7b and
fara.qwen_helpers are then imported unchanged."""
import importlib.machinery
import importlib.util
with _IMPORT_LOCK:
if "fara" in sys.modules:
return
for p in sys.path:
d = Path(p or ".") / "fara"
if (d / "fara_7b" / "fara_agent.py").is_file():
spec = importlib.machinery.ModuleSpec("fara", None, is_package=True)
spec.submodule_search_locations = [str(d)]
sys.modules["fara"] = importlib.util.module_from_spec(spec)
return
raise ImportError("no fara/fara_7b on sys.path: put the microsoft/fara src/ dir on --path")
class _ThreadStdout:
"""sys.stdout that sends a worker thread's writes to that thread's task file when one is set. The Fara-7B agent
print()s every thought, action and observation. Other writes go to the real stdout."""
def __init__(self, base):
self.base, self.local = base, threading.local()
def _f(self):
return getattr(self.local, "f", None) or self.base
def write(self, s):
return self._f().write(s)
def flush(self):
return self._f().flush()
def __getattr__(self, k):
return getattr(self.base, k)
def _thread_stdout():
with _IMPORT_LOCK:
if not isinstance(sys.stdout, _ThreadStdout):
sys.stdout = _ThreadStdout(sys.stdout)
return sys.stdout
def fara7b_parse(agent, text):
"""-> (ok, tool-call dict): their _parse_thoughts_and_action, plus the keys their generate_model_call and run() read
next (a response without them ends the released loop with an exception, as a parse failure does)."""
try:
_, action = agent._parse_thoughts_and_action(text)
return isinstance(action.get("arguments"), dict) and "action" in action["arguments"], action
except Exception:
return False, None
def run_fara7b(t, base, args, ep):
"""One task, with webeval's error retries. Their eval loop (webeval core.run_eval_single_example) re-runs a task
whose episode raised, which leaves no evaluable trajectory, up to --max_error_task_retries times (default 5). The
same here: an attempt that ended with status error / parse_error is moved to <ep>/error_attempt_N/, and the task
runs again in a fresh browser context (a fresh MiniWeb session). The last attempt is the one graded."""
import asyncio
import shutil
t0, failed = time.time(), []
while True:
res = asyncio.run(_fara7b(t, base, args, ep))
if res["status"] not in ("error", "parse_error") or len(failed) >= args.fara7b_error_retries:
break
keep = ep / f"error_attempt_{len(failed) + 1}"
keep.mkdir()
for p in list(ep.iterdir()):
if not p.name.startswith("error_attempt_"):
shutil.move(str(p), str(keep / p.name))
(keep / "attempt.json").write_text(json.dumps(res, default=str))
failed.append({k: res[k] for k in ("status", "rounds", "error", "seconds", "parse_fails", "action_errors")})
res["error_attempts"] = failed
res["parse_fails"] += sum(f["parse_fails"] for f in failed)
res["action_errors"] += sum(f["action_errors"] for f in failed)
res["seconds"] = round(time.time() - t0, 1)
return res
async def _fara7b(t, base, args, ep):
"""Fara-7B's own agent (fara.fara_7b.FaraAgent) and browser manager (BrowserBB, PlaywrightController) as
`fara-cli --fara-7b` and webeval's WebSurferSystem run it: 1440x900, the last 3 screenshots, temperature 0. Their
browser manager is subclassed only to launch our Chromium build with our context (clipboard granted, focus emulated,
en-US/UTC), in place of Firefox with an Edge user agent."""
import asyncio
import logging
import openai
from miniwebgen import config
_fara_package()
from fara.fara_7b.browser.browser_bb import BrowserBB
from fara.fara_7b.fara_agent import FaraAgent
from fara.fara_7b.fara_types import AssistantMessage
if (args.width, args.height) != (1440, 900):
raise ValueError("FaraAgent maps coordinates to its fixed 1440x900 viewport")
start = t["start"].replace("{BASE}", base)
prompt = t["prompt"].replace("{BASE}", base)
flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "")
res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None,
"record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0}
t0, rec = time.time(), Rec(ep)
fdir = ep / "fara7b" # their downloads folder: screenshot{n}.png, web_surfer.log, prints
fdir.mkdir(parents=True, exist_ok=True)
log = logging.getLogger(f"ownloop.fara7b.{t['task_id']}")
log.setLevel(logging.DEBUG) # webeval's level: their Thought/Action/Observation traces are DEBUG
log.propagate = False
handler = logging.FileHandler(fdir / "web_surfer.log")
log.addHandler(handler)
out = _thread_stdout()
out.local.f = open(fdir / "stdout.log", "w")
class MiniWebBB(BrowserBB):
async def _init_regular_browser(self, channel="chromium"):
exe = config.CHROME if os.path.exists(config.CHROME) else None
self.browser = await self._playwright.chromium.launch(headless=True, executable_path=exe, args=LAUNCH_ARGS)
ctx = self._context = await self.browser.new_context(
viewport={"width": self._viewport_width, "height": self._viewport_height}, device_scale_factor=1,
locale="en-US", timezone_id="UTC", accept_downloads=True)
try:
await ctx.grant_permissions(["clipboard-read", "clipboard-write"])
except Exception:
pass
async def focus(page): # miniwebgen.browser._focus
try:
s = await ctx.new_cdp_session(page)
await s.send("Emulation.setFocusEmulationEnabled", {"enabled": True})
except Exception:
pass
ctx.on("page", focus)
self._page = await ctx.new_page()
bm = MiniWebBB(viewport_height=args.height, viewport_width=args.width, headless=True, page_script_path=None,
browser_channel="chromium", browser_data_dir=None, downloads_folder=str(fdir), to_resize_viewport=True,
single_tab_mode=True, animate_actions=False, use_browser_base=False, logger=log)
agent = FaraAgent(browser_manager=bm, client_config={"model": args.model, "base_url": args.endpoint,
"api_key": "not-needed"},
start_page=flags, downloads_folder=str(fdir), save_screenshots=True, max_rounds=args.max_steps,
logger=log)
answer = None
try:
await agent.initialize() # browser up at the session-flags page (their start_page)
await agent._page.goto(start, wait_until="domcontentloaded", timeout=60000)
await asyncio.sleep(1.0)
answer, _, _ = await asyncio.wait_for(agent.run(instruction(prompt, start)),
timeout=max(60.0, args.timeout - (time.time() - t0)))
except (asyncio.TimeoutError, TimeoutError) as e:
res["status"], res["error"] = "timeout", f"{type(e).__name__}: {str(e)[:200]}"
except (openai.APIConnectionError, openai.InternalServerError, openai.NotFoundError) as e:
res["status"], res["error"] = "endpoint_error", f"{type(e).__name__}: {str(e)[:300]}"
except Exception as e: # the released loop has no recovery: a parse or action error ends it
res["status"], res["error"] = "error", f"{type(e).__name__}: {str(e)[:300]}"
responses = [m.content for m in agent._chat_history if isinstance(m, AssistantMessage)]
res["rounds"] = len(responses)
parsed = [fara7b_parse(agent, r) for r in responses]
for i, (text, (ok, action)) in enumerate(zip(responses, parsed), 1):
rec.step(step=i, raw=text, action=(action or {}).get("arguments") if ok else None, parse_error=not ok)
last_ok, last = parsed[-1] if parsed else (True, None)
if res["status"] is None:
if last_ok and last and last["arguments"]["action"] in ("terminate", "stop"):
res["status"], res["answer"] = "terminate", answer # their final answer: the terminate step's thoughts
else:
res["status"] = "max_steps"
elif res["status"] == "error" and responses and not last_ok:
res["status"], res["parse_fails"] = "parse_error", 1
elif res["status"] == "error":
res["action_errors"] = 1
for png in fdir.glob("screenshot*.png"): # screenshot{n}: after n actions = what step n+1 saw
m = re.fullmatch(r"screenshot(\d+)\.png", png.name)
if m:
os.replace(png, ep / "screens" / f"step_{int(m.group(1)) + 1:02d}.png")
try:
if bm.context is not None:
for key, path in (("record", "/_admin/record"), ("log", "/_admin/log")):
r = await bm.context.request.get(base + path, timeout=30000)
res[key] = ((await r.json()) or {}).get("entries", []) or []
except Exception as e:
res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}"
for close in (agent.close, getattr(bm.browser, "close", None), # their client is left open: closed here so
getattr(agent._openai_client, "close", None)): # GC does not close it after the loop ends
try:
if close is not None:
await close()
except Exception:
pass
out.local.f.close()
out.local.f = None
log.removeHandler(handler)
handler.close()
rec.close()
res["seconds"] = round(time.time() - t0, 1)
return res
# ── OpenCUA (added 2026-10-03) ────────────────────────────────────────────────────────────────────────────────────────
PY_KEYS = {**PW_KEYS, "ctrlleft": "Control", "ctrlright": "Control", "shiftleft": "Shift", "shiftright": "Shift",
"altleft": "Alt", "altright": "Alt", "winleft": "Meta", "winright": "Meta", "pgup": "PageUp",
"pgdn": "PageDown", "prtsc": "PrintScreen", "prtscr": "PrintScreen", "printscreen": "PrintScreen",
"\n": "Enter", "\r": "Enter", "\t": "Tab", " ": "Space"}
def py_key(k):
"""A pyautogui key name -> a Playwright key."""
k = str(k)
if k.lower() in PY_KEYS:
return PY_KEYS[k.lower()]
if re.fullmatch(r"[fF]\d{1,2}", k):
return k.upper()
return k
class PyAutoGUI:
"""The pyautogui functions OSWorld's VM runs (`python -c "import pyautogui; ...; <code>"`), on a Playwright page of
the same screen size, with the mouse position tracked as pyautogui tracks it. One scroll click is 100 px, as in the
UI-TARS translation above."""
SCROLL_PX = 100
API = {"click", "doubleClick", "tripleClick", "rightClick", "middleClick", "moveTo", "moveRel", "move", "dragTo",
"dragRel", "drag", "mouseDown", "mouseUp", "scroll", "vscroll", "hscroll", "write", "typewrite", "press",
"hotkey", "keyDown", "keyUp", "sleep"}
def __init__(self, page):
self.page, self.pos = page, (0.0, 0.0) # Playwright's mouse starts at (0, 0)
def _xy(self, x, y):
if isinstance(x, (tuple, list)):
x, y = x[0], x[1]
if x is None or y is None:
return self.pos
return float(x), float(y)
@staticmethod
def _button(b):
return {"right": "right", "secondary": "right", "middle": "middle"}.get(str(b).lower(), "left")
def moveTo(self, x=None, y=None, duration=0.0, *a, **kw):
self.pos = self._xy(x, y)
self.page.mouse.move(*self.pos, steps=max(1, int(float(duration or 0) * 20)))
def moveRel(self, xOffset=0, yOffset=0, *a, **kw):
self.moveTo(self.pos[0] + float(xOffset or 0), self.pos[1] + float(yOffset or 0))
move = moveRel
def click(self, x=None, y=None, clicks=1, interval=0.0, button="left", *a, **kw):
self.moveTo(x, y)
self.page.mouse.click(*self.pos, button=self._button(button), click_count=max(1, int(clicks)))
def doubleClick(self, x=None, y=None, interval=0.0, button="left", *a, **kw):
self.click(x, y, 2, button=button)
def tripleClick(self, x=None, y=None, interval=0.0, button="left", *a, **kw):
self.click(x, y, 3, button=button)
def rightClick(self, x=None, y=None, *a, **kw):
self.click(x, y, button="right")
def middleClick(self, x=None, y=None, *a, **kw):
self.click(x, y, button="middle")
def mouseDown(self, x=None, y=None, button="left", *a, **kw):
if x is not None:
self.moveTo(x, y)
self.page.mouse.down(button=self._button(button))
def mouseUp(self, x=None, y=None, button="left", *a, **kw):
if x is not None:
self.moveTo(x, y)
self.page.mouse.up(button=self._button(button))
def dragTo(self, x=None, y=None, duration=0.0, button="left", *a, **kw):
self.page.mouse.down(button=self._button(button))
self.pos = self._xy(x, y)
self.page.mouse.move(*self.pos, steps=20)
self.page.mouse.up(button=self._button(button))
def dragRel(self, xOffset=0, yOffset=0, duration=0.0, button="left", *a, **kw):
self.dragTo(self.pos[0] + float(xOffset or 0), self.pos[1] + float(yOffset or 0), button=button)
drag = dragRel
def scroll(self, clicks, x=None, y=None, *a, **kw): # positive clicks scroll up
if x is not None and y is not None:
self.moveTo(x, y)
self.page.mouse.wheel(0, -float(clicks) * self.SCROLL_PX)
vscroll = scroll
def hscroll(self, clicks, x=None, y=None, *a, **kw): # positive clicks scroll right
if x is not None and y is not None:
self.moveTo(x, y)
self.page.mouse.wheel(float(clicks) * self.SCROLL_PX, 0)
def write(self, message, interval=0.0, *a, **kw):
if isinstance(message, (list, tuple)): # typewrite(["a", "enter"]): key names
for k in message:
self.press(k)
return
parts = str(message).split("\n") # pyautogui types "\n" as Enter
for i, part in enumerate(parts):
if part:
self.page.keyboard.type(part)
if i < len(parts) - 1:
self.page.keyboard.press("Enter")
typewrite = write
def press(self, keys, presses=1, interval=0.0, *a, **kw):
for _ in range(max(1, int(presses))):
for k in (keys if isinstance(keys, (list, tuple)) else [keys]):
self.page.keyboard.press(py_key(k))
def hotkey(self, *keys, **kw):
keys = [py_key(k) for k in (keys[0] if len(keys) == 1 and isinstance(keys[0], (list, tuple)) else keys)]
for k in keys:
self.page.keyboard.down(k)
for k in reversed(keys):
self.page.keyboard.up(k)
def keyDown(self, key, *a, **kw):
self.page.keyboard.down(py_key(key))
def keyUp(self, key, *a, **kw):
self.page.keyboard.up(py_key(key))
def sleep(self, seconds=0, *a, **kw):
time.sleep(min(max(float(seconds), 0.0), 60.0))
class UnsupportedCode(Exception):
"""A statement that the `pyautogui` + `time` namespace of OSWorld's VM would not run (e.g. computer.triple_click:
their prompt lists it, but their loop passes it to the VM's python, where it is a NameError)."""
def run_pyautogui(code, gui):
"""OpenCUA's code, statement by statement, as `python -c` runs it: the first failing statement stops the rest.
Model output is never exec'd: only literal-argument calls of pyautogui.* and time.sleep, `import pyautogui|time`,
`pass`, and `for _ in range(<int>):` loops over those statements run."""
import ast
def call(c):
f = c.func
if not (isinstance(f, ast.Attribute) and isinstance(f.value, ast.Name)):
raise UnsupportedCode(f"unsupported call: {ast.unparse(c)[:120]}")
mod, name = f.value.id, f.attr
args = [ast.literal_eval(a) for a in c.args]
kwargs = {k.arg: ast.literal_eval(k.value) for k in c.keywords if k.arg}
if mod == "time" and name == "sleep":
return gui.sleep(*args, **kwargs)
if mod == "pyautogui" and name in gui.API:
return getattr(gui, name)(*args, **kwargs)
raise UnsupportedCode(f"{mod}.{name} is not defined")
def stmt(node):
if isinstance(node, ast.Import) and {a.name for a in node.names} <= {"pyautogui", "time"}:
return
if isinstance(node, ast.Pass):
return
if isinstance(node, ast.Expr) and isinstance(node.value, ast.Call):
return call(node.value)
if (isinstance(node, ast.For) and not node.orelse and isinstance(node.iter, ast.Call)
and isinstance(node.iter.func, ast.Name) and node.iter.func.id == "range"):
n = range(*[int(ast.literal_eval(a)) for a in node.iter.args])
for _ in range(min(len(n), 100)):
for s in node.body:
stmt(s)
return
raise UnsupportedCode(f"unsupported statement: {ast.unparse(node)[:120]}")
for node in ast.parse(code).body:
stmt(node)
def chat_full(client, model, messages, **kw):
"""chat() that also returns the finish reason; a 400 (e.g. a prompt over the context) is the request's own error,
not an endpoint failure."""
import openai
for attempt in range(3):
try:
r = client.chat.completions.create(model=model, messages=messages, **kw)
return r.choices[0].message.content or "", r.choices[0].finish_reason
except openai.BadRequestError:
raise
except Exception as e:
if attempt == 2:
raise EndpointError(f"{type(e).__name__}: {str(e)[:200]}") from e
time.sleep(5)
_OPENCUA = {}
def opencua_agent_class():
"""Their OpenCUAAgent (xlang-ai/OSWorld mm_agents/opencua) with call_llm pointed at a vLLM endpoint. Everything else
is theirs: the prompt, the history, the retries, the parsing and the coordinate projection."""
with _IMPORT_LOCK:
if "cls" in _OPENCUA:
return _OPENCUA["cls"]
from loguru import logger
logger.remove() # their module logs every prompt and response through loguru
from mm_agents.opencua.opencua_agent import OpenCUAAgent
class VllmOpenCUAAgent(OpenCUAAgent):
"""Their call_llm posts to their hosted endpoint and re-posts the same payload (up to 20 times) until
finish_reason == "stop". At temperature 0 a re-post repeats the same output, so here a non-"stop" finish
returns None at once; their predict() then retries (5 calls in all, temperature >= 0.2 after the first)."""
def __init__(self, client, deadline, **kw):
super().__init__(**kw)
self.client, self.deadline = client, deadline
self.calls, self.endpoint_failed, self.timed_out = [], None, False
def call_llm(self, payload, model):
if self.endpoint_failed:
raise EndpointError(self.endpoint_failed)
if time.time() > self.deadline:
self.timed_out = True
raise TimeoutError("task timeout")
try:
text, finish = chat_full(self.client, **payload)
except EndpointError as e:
self.endpoint_failed = str(e)
raise
except Exception as e:
self.calls.append({"error": f"{type(e).__name__}: {str(e)[:300]}",
"temperature": payload.get("temperature")})
raise
self.calls.append({"raw": text, "finish_reason": finish, "temperature": payload.get("temperature")})
return text if finish == "stop" else None
_OPENCUA["cls"] = VllmOpenCUAAgent
return VllmOpenCUAAgent
def opencua_answer(cot):
"""The answer of a terminate step: its answer= argument (their V2 prompt's terminate has one) if given, else the
step's Thought and Action text. OpenCUA-7B's V1 prompt has no answer field, and OSWorld grades no text answers."""
import ast
code = cot.get("original_code") or ""
try:
for node in ast.walk(ast.parse(code)):
if isinstance(node, ast.Call) and getattr(node.func, "attr", None) == "terminate":
for k in node.keywords:
if k.arg == "answer":
return str(ast.literal_eval(k.value))
except Exception:
m = re.search(r"answer\s*=\s*('''|\"\"\"|'|\")(.*?)\1", code, re.S)
if m:
return m.group(2)
return "\n".join(p for p in (cot.get("thought"), cot.get("action")) if p) or None
def front_page(ctx, page):
"""The page on screen: a click may open a tab, which a desktop browser brings to the front."""
pages = [p for p in ctx.pages if not p.is_closed()]
page = pages[-1] if pages else page
try:
page.wait_for_load_state("domcontentloaded", timeout=5000)
except Exception:
pass
return page
def run_opencua(t, base, args, ep):
"""OSWorld's OpenCUA-7B command (run_multienv_opencua.py --coordinate_type qwen25 --use_old_sys_prompt, the
defaults: cot_level l2, action_history, 3 images, temperature 0, top_p 0.9, max_tokens 2048) and loop
(lib_run_single.run_single_example_opencua) on a 1920x1080 page: predict, then each returned action through
env.step semantics (DONE / FAIL end the episode, WAIT sleeps, pyautogui code runs), each step followed by
sleep_after_execution (5 s)."""
from openai import OpenAI
from playwright.sync_api import sync_playwright
start = t["start"].replace("{BASE}", base)
prompt = t["prompt"].replace("{BASE}", base)
flags = base + "/_admin/session-flags" + ("?logout=1" if t["needs_login"] else "")
res = {"task_id": t["task_id"], "base": base, "answer": None, "status": None, "rounds": 0, "error": None,
"record": [], "log": [], "seconds": None, "parse_fails": 0, "action_errors": 0}
t0, rec = time.time(), Rec(ep)
agent = opencua_agent_class()(
OpenAI(base_url=args.endpoint, api_key="not-needed", timeout=300), t0 + args.timeout,
model=args.model, history_type="action_history", max_steps=args.max_steps, max_image_history_length=3,
platform="ubuntu", max_tokens=args.opencua_max_tokens, top_p=0.9, temperature=0.0, action_space="pyautogui",
observation_type="screenshot", cot_level="l2", screen_size=(args.width, args.height), coordinate_type="qwen25",
use_old_sys_prompt=True)
task = instruction(prompt, start)
pause = args.opencua_pause
with sync_playwright() as pw:
browser = launch_browser(pw)
ctx = new_context(browser, args.width, args.height)
try:
page = ctx.new_page()
page.goto(flags, wait_until="domcontentloaded", timeout=60000)
page.goto(start, wait_until="domcontentloaded", timeout=60000)
time.sleep(1.0)
gui = PyAutoGUI(page)
for i in range(1, args.max_steps + 1):
if time.time() - t0 > args.timeout:
res["status"] = "timeout"
break
png = page.screenshot(type="png")
rec.shot(i, png)
n_actions, n_calls = len(agent.actions), len(agent.calls)
response, actions, cot = agent.predict(task, {"screenshot": png})
calls, accepted = agent.calls[n_calls:], len(agent.actions) > n_actions
res["rounds"] = i
res["parse_fails"] += len(calls) - accepted
row = {"step": i, "url": page.url, "raw": response if accepted else None,
"attempts": calls if len(calls) != 1 or not accepted else None, "action": cot.get("action"),
"code": cot.get("original_code"), "executed": actions}
if agent.endpoint_failed or agent.timed_out or not accepted:
rec.step(**row, error=response)
res["status"] = "endpoint_error" if agent.endpoint_failed else "timeout" if agent.timed_out \
else "parse_error" # their predict: 5 failed calls -> FAIL, the episode ends
res["error"] = agent.endpoint_failed or str(response)[:300]
break
if not actions or actions[0] == "" or actions[0].lower().startswith("error"):
rec.step(**row)
res["status"] = "no_action"
break
stop, notes = False, []
for a in actions:
note = ""
if a in ("DONE", "FAIL"):
stop = True
elif a == "WAIT":
time.sleep(pause) # env.step sleeps `pause` for WAIT, then once more
else:
try:
run_pyautogui(a, gui)
except Exception as e:
note = f"{type(e).__name__}: {str(e)[:200]}"
res["action_errors"] += 1
notes.append(note)
if stop:
break
time.sleep(pause)
gui.page = page = front_page(ctx, page)
rec.step(**row, notes=notes)
if stop:
code = (cot.get("original_code") or "").lower()
if "computer.terminate" in code: # theirs: "fail" in the code -> FAIL, else "success"
res["status"] = "fail" if "fail" in code else "done"
res["answer"] = opencua_answer(cot)
else: # predict() forces FAIL at the step budget
res["status"] = "max_steps"
break
else:
res["status"] = "max_steps"
try:
rec.shot(res["rounds"] + 1, front_page(ctx, page).screenshot(type="png"))
except Exception:
pass
except Exception as e: # a harness error still keeps whatever the session did
res["error"] = f"{type(e).__name__}: {str(e)[:300]}"
res["status"] = res["status"] or "error"
try:
res["record"] = admin_entries(ctx, base, "/_admin/record")
res["log"] = admin_entries(ctx, base, "/_admin/log")
except Exception as e:
res["error"] = (res["error"] or "") + f" | backend read: {type(e).__name__}: {str(e)[:200]}"
browser.close()
rec.close()
res["seconds"] = round(time.time() - t0, 1)
return res
# ── driver ────────────────────────────────────────────────────────────────────────────────────────────────────────────
RUNNERS = {"uitars": (run_uitars, (1280, 800)), "molmoweb": (run_molmoweb, (1280, 720)),
"fara7b": (run_fara7b, (1440, 900)), "opencua": (run_opencua, (1920, 1080))}
def main(argv=None):
ap = argparse.ArgumentParser()
ap.add_argument("agent", choices=sorted(RUNNERS))
ap.add_argument("--tasks", required=True, help="tasks.json of `python -m miniweb.evaluation.fara_rq1 prep`")
ap.add_argument("--out", required=True)
ap.add_argument("--ports", required=True, help="MiniWeb server ports to start (8300-8320), e.g. 8312,8313")
ap.add_argument("--endpoint", required=True, help="vLLM OpenAI endpoint, e.g. http://127.0.0.1:8694/v1")
ap.add_argument("--model", required=True, help="served model name")
ap.add_argument("--path", default=os.environ.get("OWNLOOP_PATH", ""),
help="colon list prepended to sys.path: the ui-tars --target dir, the molmoweb clone")
ap.add_argument("--workers", type=int, default=8)
ap.add_argument("--max-steps", type=int, default=40, help="model actions per task (our MiniWeb budget: 40)")
ap.add_argument("--timeout", type=int, default=1800)
ap.add_argument("--width", type=int, default=0, help="viewport; default uitars 1280x800, molmoweb 1280x720")
ap.add_argument("--height", type=int, default=0)
ap.add_argument("--frequency-penalty", type=float, default=0.0,
help="uitars: 0 as the UI-TARS README's deployment example; OSWorld's request sends 1, which turned "
"answers that repeat the thought's words into Chinese in the smoke test")
ap.add_argument("--temperature", type=float, default=0.7, help="molmoweb sampling (their default)")
ap.add_argument("--top-p", type=float, default=0.8)
ap.add_argument("--max-past-steps", type=int, default=10, help="molmoweb text history (their benchmark default)")
ap.add_argument("--fara7b-error-retries", type=int, default=5,
help="fara7b: webeval's --max_error_task_retries (an episode that raised is re-run)")
ap.add_argument("--opencua-max-tokens", type=int, default=2048,
help="opencua: run_multienv_opencua.py's --max_tokens (the agent class's own default is 1500)")
ap.add_argument("--opencua-pause", type=float, default=5.0,
help="opencua: run_multienv_opencua.py's --sleep_after_execution, slept after every action")
ap.add_argument("--limit", type=int, default=0)
ap.add_argument("--task-ids", default="")
args = ap.parse_args(argv)
for p in reversed([p for p in args.path.split(":") if p]):
sys.path.insert(0, p)
run, (w, h) = RUNNERS[args.agent]
args.width, args.height = args.width or w, args.height or h
out = Path(args.out)
tasks = json.loads(Path(args.tasks).read_text())
if args.task_ids:
keep = set(args.task_ids.split(","))
tasks = [t for t in tasks if t["task_id"] in keep]
done = {p.parent.name for p in out.glob("*/session.json")}
tasks = [t for t in tasks if t["task_id"] not in done] # resume
if args.limit:
tasks = tasks[:args.limit]
from miniwebgen import browser as B
lock, queue = threading.Lock(), list(tasks)
ports = [int(p) for p in args.ports.split(",")]
with B.Servers(n=len(ports), ports=ports) as srv:
bases = srv.bases
print(f"{args.agent}: {len(tasks)} tasks, {args.workers} workers on {bases}", flush=True)
def worker(i):
while True:
with lock:
if not queue:
return
t = queue.pop(0)
ep = out / t["task_id"]
ep.mkdir(parents=True, exist_ok=True)
try:
r = run(t, bases[i % len(bases)], args, ep)
except Exception as e:
r = {"task_id": t["task_id"], "base": bases[i % len(bases)], "answer": None, "status": "crash",
"rounds": 0, "error": f"{type(e).__name__}: {str(e)[:300]}", "record": [], "log": []}
# a harness or model-server failure is kept aside, unsaved, so a rerun (resume) repeats the task
name = "session_failed.json" if r["status"] in ("crash", "endpoint_error") else "session.json"
(ep / name).write_text(json.dumps(r, default=str))
print(f"{t['task_id']}: status={r['status']} rounds={r['rounds']} {r.get('seconds')}s "
f"parse_fails={r.get('parse_fails')} action_errors={r.get('action_errors')} err={r['error']} "
f"answer={str(r['answer'])[:80]!r}", flush=True)
with ThreadPoolExecutor(args.workers) as ex:
list(ex.map(worker, range(args.workers)))
if __name__ == "__main__":
main()
"""UI-TARS-1.5 and MolmoWeb as released, each in its own agent loop, on MiniWeb human tasks (EXPERIMENT_PLAN, RQ1;
2026-10-03). The counterpart of src/miniweb/evaluation/fara_runner.py for the two other own-loop baselines.
uitars UI-TARS-1.5-7B (ByteDance-Seed, Apache-2.0). Prompt and action parser are the released ones (pip `ui-tars`:
ui_tars.prompt.COMPUTER_USE_DOUBAO, ui_tars.action_parser.parse_action_to_structure_output). The loop is
OSWorld's UI-TARS-1.5 agent (mm_agents/uitars15_v1.py, run_multienv_uitars15_v1.py): system "You are a
helpful assistant.", the prompt, then the full response history with the last 5 screenshots, temperature 0,
top_p 0.9, max_tokens 500 (OSWorld's frequency_penalty 1 is off by default: see --frequency-penalty).
A failed parse is retried up to 3 times at temperature 1,
then the episode ends. Coordinates are in the Qwen2.5-VL smart_resize space of the screenshot and are mapped
back as parsing_response_to_pyautogui_code does. The pyautogui calls become Playwright mouse and keyboard
calls: type -> keyboard.type, plus Enter for a trailing \\n; scroll -> 5 wheel notches of 100 px at the
point; wait -> 5 s. The answer is finished(content=...).
molmoweb MolmoWeb-4B / -8B (AllenAI, Apache-2.0): the released agent code itself (github.com/allenai/molmoweb:
agent.multimodal_agent.MultimodalAgent, utils.envs SimpleEnv and its action executor,
utils.eval_utils.episode.Episode) with its benchmark settings: "molmo_web_think" prompts, the last 10 steps
as text, no past images, temperature 0.7, top_p 0.8, 1024 new tokens, a 1280x720 viewport. Only the
predictor is swapped: their HF predictor's single user turn (prompt text + screenshot), sent to a vLLM
OpenAI endpoint (the repo leaves vLLM to users). Serve it with causal attention on image tokens, as their
HF predictor runs it (vLLM's Molmo2 default is a bidirectional prefix LM): --hf-overrides
{"is_mm_prefix_lm":false}, which granite's ownloop.sh passes. The answer is read as their
benchmarks.evaluate does: the last "[ANSWER] ..." message, else the last other message, else the last
thought.
fara7b Fara-7B (Microsoft, MIT): the released previous-generation agent itself (github.com/microsoft/fara src/,
fara.fara_7b: FaraAgent, its BrowserBB browser manager and PlaywrightController), as `fara-cli --fara-7b`
and webeval's WebSurferSystem run it: 1440x900, the last 3 screenshots, temperature 0, single-tab mode,
their page script. The browser manager is subclassed only to launch our Chromium with our context, in
place of Firefox with an Edge user agent. The answer is theirs: the terminate step's thoughts. A parse or
action error ends the episode, as in the released loop, and the task is re-run up to 5 times, as webeval
(the harness of their published numbers) does by default.
opencua OpenCUA-7B (XLANG, MIT): OSWorld's OpenCUAAgent itself (xlang-ai/OSWorld mm_agents/opencua) with the
OpenCUA-7B command of run_multienv_opencua.py (--coordinate_type qwen25 --use_old_sys_prompt; l2 CoT,
action history, 3 images, temperature 0, top_p 0.9, max_tokens 2048) on a 1920x1080 page, and the loop of
lib_run_single.run_single_example_opencua. Only call_llm is swapped (a vLLM endpoint), and the VM's
`python -c` is a pyautogui-to-Playwright translation (run_pyautogui, literal calls only). The V1 prompt has
no answer field: the answer is the terminate step's answer= if given, else its Thought + Action text.
Serve with --trust-remote-code and a tokenizer dir whose tokenizer_config.json drops unk_token/pad_token
(the remote TikTokenV3 code rejects them under transformers 5; it derives the same [UNK]/[PAD] itself).
Both run in the MiniWeb env (Playwright sync API, DATAGEN_CHROME) and start their own MiniWeb servers on --ports, as
miniwebagent.evaluate does. The released code goes on sys.path through --path (default $OWNLOOP_PATH): a directory holding
`pip install --no-deps --target <dir> ui-tars==0.5.1`, and a clone of allenai/molmoweb. Setup per task is
miniwebagent.evaluate.prepare_human's: /_admin/session-flags (logged out for a login task), then the start page, clipboard
allowed, with the task wrapper of miniwebagent.harness.instruction. Tasks come from `python -m miniweb.evaluation.fara_rq1 prep`
({BASE} placeholders); grading is `python -m miniweb.evaluation.fara_rq1 grade --arm <name>`, from what this saves per task:
<out>/<task_id>/session.json {"base", "answer", "status", "rounds", "record": [...], "log": [...], "error", ...}
<out>/<task_id>/steps.jsonl one line per model call: raw output, parsed action, url, errors
<out>/<task_id>/screens/ step_NN.png: the screenshot each step saw, plus the final one
python -m miniweb.evaluation.ownloop_runner uitars --tasks <out>/tasks.json --out <out> --ports 8312,8313 \\
--endpoint http://127.0.0.1:8694/v1 --model uitars15-7b --path $W/ownloop_pkgs
python -m miniweb.evaluation.ownloop_runner molmoweb --tasks <out>/tasks.json --out <out> --ports 8314,8315 \\
--endpoint http://127.0.0.1:8695/v1 --model molmoweb-4b --path $W/ownloop_pkgs:$W/molmoweb
python -m miniweb.evaluation.ownloop_runner fara7b --tasks <out>/tasks.json --out <out> --ports 8312,8313 \\
--endpoint http://127.0.0.1:8696/v1 --model fara-7b --path $W/fara7b_pkgs:$W/fara/src
python -m miniweb.evaluation.ownloop_runner opencua --tasks <out>/tasks.json --out <out> --ports 8314,8315 \\
--endpoint http://127.0.0.1:8697/v1 --model opencua-7b --path $W/opencua_pkgs:$W/osworld
"""
import argparse
import base64
import io
import json
import math
import os
import re
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
LAUNCH_ARGS = ["--disable-dev-shm-usage", "--no-first-run", "--disable-blink-features=AutomationControlled"]
def instruction(task_text, start_url):
"""Same wrapper as miniwebagent.harness.instruction (our agents' task text) and src/miniweb/evaluation/fara_runner.py."""
return f"You are interacting with a web application at {start_url}. Your task: {task_text}"
def launch_browser(pw):
from miniwebgen import config
exe = config.CHROME if os.path.exists(config.CHROME) else None
return pw.chromium.launch(headless=True, executable_path=exe, args=LAUNCH_ARGS)
def new_context(browser, width, height):
"""miniwebgen.browser.new_context (clipboard granted, focus emulated, en-US / UTC) at this agent's viewport."""
from miniwebgen.browser import new_context as nc
return nc(browser, viewport=(width, height))
def admin_entries(ctx, base, path):
"""A privileged /_admin endpoint, read as this browser's session (the context's cookies)."""
r = ctx.request.get(base + path, timeout=30000)
return (r.json() or {}).get("entries", []) or []
def png_b64(png):
return "data:image/png;base64," + base64.b64encode(png).decode()