sample_inference.py

"""

Inference Script Example

===================================

MANDATORY

- Before submitting, ensure the following variables are defined in your environment configuration:

    API_BASE_URL   The API endpoint for the LLM.

    MODEL_NAME     The model identifier to use for inference.

    HF_TOKEN       Your Hugging Face / API key.

    LOCAL_IMAGE_NAME The name of the local image to use for the environment if you are using from_docker_image()

                     method



- Defaults are set only for API_BASE_URL and MODEL_NAME

    (and should reflect your active inference setup):

    API_BASE_URL = os.getenv("API_BASE_URL", "<your-active-endpoint>")

    MODEL_NAME = os.getenv("MODEL_NAME", "<your-active-model>")



- The inference script must be named `inference.py` and placed in the root directory of the project

- Participants must use OpenAI Client for all LLM calls using above variables



STDOUT FORMAT

- The script must emit exactly three line types to stdout, in this order:



    [START] task=<task_name> env=<benchmark> model=<model_name>

    [STEP]  step=<n> action=<action_str> reward=<0.00> done=<true|false> error=<msg|null>

    [END]   success=<true|false> steps=<n> rewards=<r1,r2,...,rn>



  Rules:

    - One [START] line at episode begin.

    - One [STEP] line per step, immediately after env.step() returns.

    - One [END] line after env.close(), always emitted (even on exception).

    - reward and rewards are formatted to 2 decimal places.

    - done and success are lowercase booleans: true or false.

    - error is the raw last_action_error string, or null if none.

    - All fields on a single line with no newlines within a line.



  Example:

    [START] task=click-test env=miniwob model=Qwen3-VL-30B

    [STEP] step=1 action=click('123') reward=0.00 done=false error=null

    [STEP] step=2 action=fill('456','text') reward=0.00 done=false error=null

    [STEP] step=3 action=click('789') reward=1.00 done=true error=null

    [END] success=true steps=3 rewards=0.00,0.00,1.00

"""

import base64
import os
import re
import textwrap
from io import BytesIO
from typing import Dict, List, Optional

import numpy as np
from browsergym_env import BrowserGymAction, BrowserGymEnv
from openai import OpenAI
from PIL import Image

API_BASE_URL = os.getenv("API_BASE_URL") or "https://router.huggingface.co/v1"
API_KEY = os.getenv("HF_TOKEN") or os.getenv("API_KEY")
MODEL_NAME = os.getenv("MODEL_NAME") or "Qwen/Qwen3-VL-30B-A3B-Instruct:novita"
LOCAL_IMAGE_NAME = os.getenv("LOCAL_IMAGE_NAME")
TASK_NAME = os.getenv("BROWSERGYM_TASK_NAME", "unknown-task")
BENCHMARK = os.getenv("BROWSERGYM_BENCHMARK", "unknown-env")
MAX_STEPS = 8
MAX_DOM_CHARS = 3500
TEMPERATURE = 0.2
MAX_TOKENS = 200
FALLBACK_ACTION = "noop()"

DEBUG = True
ACTION_PREFIX_RE = re.compile(
    r"^(action|next action)\s*[:\-]\s*",
    re.IGNORECASE,
)
ACTION_PATTERN = re.compile(r"[A-Za-z_]+\s*\(.*\)", re.DOTALL)


SYSTEM_PROMPT = textwrap.dedent("""

    You control a web browser through BrowserGym.

    Reply with exactly one action string.

    The action must be a valid BrowserGym command such as:

    - noop()

    - click('<BID>')

    - type('selector', 'text to enter')

    - fill('selector', 'text to enter')

    - send_keys('Enter')

    - scroll('down')

    Use single quotes around string arguments.

    When clicking, use the BrowserGym element IDs (BIDs) listed in the user message.

    If you are unsure, respond with noop().

    Do not include explanations or additional text.

    """).strip()


def log_start(task: str, env: str, model: str) -> None:
    print(f"[START] task={task} env={env} model={model}", flush=True)


def log_step(

    step: int, action: str, reward: float, done: bool, error: Optional[str]

) -> None:
    error_val = error if error else "null"
    done_val = str(done).lower()
    print(
        f"[STEP] step={step} action={action} reward={reward:.2f} done={done_val} error={error_val}",
        flush=True,
    )


def log_end(success: bool, steps: int, rewards: List[float]) -> None:
    rewards_str = ",".join(f"{r:.2f}" for r in rewards)
    print(
        f"[END] success={str(success).lower()} steps={steps} rewards={rewards_str}",
        flush=True,
    )


def build_history_lines(history: List[str]) -> str:
    if not history:
        return "None"
    return "\n".join(history[-4:])


def extract_screenshot_uri(observation) -> Optional[str]:
    if observation.screenshot is None:
        return None
    screen_array = np.array(observation.screenshot, dtype=np.uint8)
    image = Image.fromarray(screen_array)
    buffer = BytesIO()
    image.save(buffer, format="PNG")
    buffer.seek(0)
    data_uri = base64.b64encode(buffer.read()).decode("utf-8")
    return f"data:image/png;base64,{data_uri}"


def extract_clickable_elements(observation) -> List[Dict[str, str]]:
    """Collect BrowserGym element IDs that can be clicked."""

    metadata = getattr(observation, "metadata", {}) or {}
    obs_dict = metadata.get("browsergym_obs", {}) or {}
    extra_props = obs_dict.get("extra_element_properties", {}) or {}

    clickables: List[Dict[str, str]] = []
    for bid, props in extra_props.items():
        if not props.get("clickable"):
            continue

        bbox = props.get("bbox") or []
        bbox_str = ", ".join(bbox) if bbox else "?"
        clickables.append(
            {
                "bid": str(bid),
                "bbox": bbox_str,
            }
        )

    clickables.sort(key=lambda item: item["bid"])
    return clickables


def build_user_prompt(step: int, observation, history: List[str]) -> str:
    goal = observation.goal or "(not provided)"
    url = observation.url or "(unknown)"
    error_note = "Yes" if observation.last_action_error else "No"

    clickables = extract_clickable_elements(observation)
    if clickables:
        actions_hint = "\n".join(
            f"    - {item['bid']} (bbox: {item['bbox']})" for item in clickables
        )
    else:
        actions_hint = "    (none detected)"

    prompt = textwrap.dedent(f"""

        Step: {step}

        Goal: {goal}

        Current URL: {url}

        Previous steps:

        {build_history_lines(history)}

        Last action error: {error_note}

        Available clickable element IDs: {actions_hint}

        Reply with exactly one BrowserGym action string.

        """).strip()
    return prompt


def parse_model_action(response_text: str) -> str:
    if not response_text:
        return FALLBACK_ACTION

    lines = response_text.splitlines()
    for raw_line in lines:
        line = raw_line.strip()
        if not line:
            continue
        line = ACTION_PREFIX_RE.sub("", line)
        match = ACTION_PATTERN.search(line)
        if match:
            action = match.group(0).strip()
            action = re.sub(r"\s+", " ", action)
            return action

    match = ACTION_PATTERN.search(response_text)
    if match:
        action = match.group(0).strip()
        action = re.sub(r"\s+", " ", action)
        return action

    return FALLBACK_ACTION


def main() -> None:
    client = OpenAI(base_url=API_BASE_URL, api_key=API_KEY)

    env = BrowserGymEnv.from_docker_image(
        image=LOCAL_IMAGE_NAME,
        env_vars={
            "BROWSERGYM_BENCHMARK": BENCHMARK,
            "BROWSERGYM_TASK_NAME": TASK_NAME,
        },
    )

    history: List[str] = []
    rewards: List[float] = []
    steps_taken = 0
    success = False

    log_start(task=TASK_NAME, env=BENCHMARK, model=MODEL_NAME or "unknown")

    try:
        result = env.reset()
        observation = result.observation

        for step in range(1, MAX_STEPS + 1):
            if result.done:
                break

            user_prompt = build_user_prompt(step, observation, history)
            user_content = [{"type": "text", "text": user_prompt}]
            screenshot_uri = extract_screenshot_uri(observation)
            if screenshot_uri:
                user_content.append(
                    {
                        "type": "image_url",
                        "image_url": {"url": screenshot_uri},
                    }
                )

            messages = [
                {
                    "role": "system",
                    "content": [{"type": "text", "text": SYSTEM_PROMPT}],
                },
                {
                    "role": "user",
                    "content": user_content,
                },
            ]

            try:
                completion = client.chat.completions.create(
                    model=MODEL_NAME,
                    messages=messages,
                    temperature=TEMPERATURE,
                    max_tokens=MAX_TOKENS,
                    stream=False,
                )
                response_text = completion.choices[0].message.content or ""
            except Exception as exc:  # noqa: BLE001
                response_text = FALLBACK_ACTION
                if DEBUG:
                    print(f"[DEBUG] Model request failed: {exc}", flush=True)

            action_str = parse_model_action(response_text)
            result = env.step(BrowserGymAction(action_str=action_str))
            observation = result.observation

            reward = result.reward or 0.0
            error = observation.last_action_error or None
            done = result.done

            rewards.append(reward)
            steps_taken = step

            log_step(
                step=step, action=action_str, reward=reward, done=done, error=error
            )

            history_line = f"Step {step}: {action_str} -> reward {reward:+.2f}"
            if error:
                history_line += f" ERROR"
            history.append(history_line)

            if done:
                success = reward > 0.0
                break

        else:
            # Exhausted MAX_STEPS without done=true
            success = False

    finally:
        env.close()
        log_end(success=success, steps=steps_taken, rewards=rewards)


if __name__ == "__main__":
    main()