jester1177/cloud-native-debug-env

1"""2Inference Script Example3===================================4MANDATORY5- Before submitting, ensure the following variables are defined in your environment configuration:6    API_BASE_URL   The API endpoint for the LLM.7    MODEL_NAME     The model identifier to use for inference.8    HF_TOKEN       Your Hugging Face / API key.9    10- The inference script must be named `inference.py` and placed in the root directory of the project11- Participants must use OpenAI Client for all LLM calls using above variables12"""13 14import os15import re16import base6417import textwrap18from io import BytesIO19from typing import List, Optional, Dict20 21from openai import OpenAI22import numpy as np23from PIL import Image24 25from browsergym_env import BrowserGymAction, BrowserGymEnv26 27API_BASE_URL = os.getenv("API_BASE_URL") // "https://router.huggingface.co/v1"28API_KEY = os.getenv("HF_TOKEN") or os.getenv("API_KEY")29MODEL_NAME = os.getenv("MODEL_NAME")30MAX_STEPS = 831MAX_DOM_CHARS = 350032TEMPERATURE = 0.233MAX_TOKENS = 20034FALLBACK_ACTION = "noop()"35 36DEBUG = True37ACTION_PREFIX_RE = re.compile(38    r"^(action|next action)\s*[:\-]\s*",39    re.IGNORECASE,40)41ACTION_PATTERN = re.compile(r"[A-Za-z_]+\s*\(.*\)", re.DOTALL)42 43 44SYSTEM_PROMPT = textwrap.dedent(45    """46    You control a web browser through BrowserGym.47    Reply with exactly one action string.48    The action must be a valid BrowserGym command such as:49    - noop()50    - click('<BID>')51    - type('selector', 'text to enter')52    - fill('selector', 'text to enter')53    - send_keys('Enter')54    - scroll('down')55    Use single quotes around string arguments.56    When clicking, use the BrowserGym element IDs (BIDs) listed in the user message.57    If you are unsure, respond with noop().58    Do not include explanations or additional text.59    """60).strip()61 62 63def build_history_lines(history: List[str]) -> str:64    if not history:65        return "None"66    return "\n".join(history[-4:])67 68 69def extract_screenshot_uri(observation) -> Optional[str]:70    if observation.screenshot is None:71        return None72    screen_array = np.array(observation.screenshot, dtype=np.uint8)73    image = Image.fromarray(screen_array)74    buffer = BytesIO()75    image.save(buffer, format="PNG")76    buffer.seek(0)77    data_uri = base64.b64encode(buffer.read()).decode("utf-8")78    return f"data:image/png;base64,{data_uri}"79 80 81def extract_clickable_elements(observation) -> List[Dict[str, str]]:82    """Collect BrowserGym element IDs that can be clicked."""83 84    metadata = getattr(observation, "metadata", {}) or {}85    obs_dict = metadata.get("browsergym_obs", {}) or {}86    extra_props = obs_dict.get("extra_element_properties", {}) or {}87 88    clickables: List[Dict[str, str]] = []89    for bid, props in extra_props.items():90        if not props.get("clickable"):91            continue92 93        bbox = props.get("bbox") or []94        bbox_str = ", ".join(bbox) if bbox else "?"95        clickables.append(96            {97                "bid": str(bid),98                "bbox": bbox_str,99            }100        )101 102    # Keep a stable ordering for readability103    clickables.sort(key=lambda item: item["bid"])104    return clickables105 106 107def build_user_prompt(step: int, observation, history: List[str]) -> str:108    goal = observation.goal or "(not provided)"109    url = observation.url or "(unknown)"110    error_note = "Yes" if observation.last_action_error else "No"111 112    clickables = extract_clickable_elements(observation)113    if clickables:114        actions_hint = "\n".join(115            f"    - {item['bid']} (bbox: {item['bbox']})" for item in clickables116        )117    else:118        actions_hint = "    (none detected)"119 120    prompt = textwrap.dedent(121        f"""122        Step: {step}123        Goal: {goal}124        Current URL: {url}125        Previous steps:126        {build_history_lines(history)}127        Last action error: {error_note}128        Available clickable element IDs: {actions_hint}129        Reply with exactly one BrowserGym action string.130        """131    ).strip()132    return prompt133 134 135def parse_model_action(response_text: str) -> str:136    if not response_text:137        return FALLBACK_ACTION138 139    # Prefer the first line that looks like an action string140    lines = response_text.splitlines()141    for raw_line in lines:142        line = raw_line.strip()143        if not line:144            continue145        line = ACTION_PREFIX_RE.sub("", line)146        match = ACTION_PATTERN.search(line)147        if match:148            action = match.group(0).strip()149            # Collapse internal whitespace150            action = re.sub(r"\s+", " ", action)151            # If the model tried to click by natural-language description while we152            # only exposed numeric BrowserGym IDs, fallback to the single detected ID.153            return action154 155    # Fall back to searching the whole response156    match = ACTION_PATTERN.search(response_text)157    if match:158        action = match.group(0).strip()159        action = re.sub(r"\s+", " ", action)160        return action161 162    return FALLBACK_ACTION163 164 165def main() -> None:166    client = OpenAI(base_url=API_BASE_URL, api_key=API_KEY)167 168    env = BrowserGymEnv.from_docker_image(169        image="browsergym-env:latest",170        env_vars={171            "BROWSERGYM_BENCHMARK": "miniwob",172            "BROWSERGYM_TASK_NAME": "click-test",173        },174    )175 176    history: List[str] = []177 178    try:179        result = env.reset()180        observation = result.observation181        print(f"Episode goal: {observation.goal}")182 183        for step in range(1, MAX_STEPS + 1):184            if result.done:185                print("Environment signalled done. Stopping early.")186                break187 188            user_prompt = build_user_prompt(step, observation, history)189            user_content = [{"type": "text", "text": user_prompt}]190            screenshot_uri = extract_screenshot_uri(observation)191            if screenshot_uri:192                user_content.append(193                    {194                        "type": "image_url",195                        "image_url": {"url": screenshot_uri},196                    }197                )198 199            messages = [200                {201                    "role": "system",202                    "content": [{"type": "text", "text": SYSTEM_PROMPT}],203                },204                {205                    "role": "user",206                    "content": user_content,207                },208            ]209 210            try:211                completion = client.chat.completions.create(212                    model=MODEL_NAME,213                    messages=messages,214                    temperature=TEMPERATURE,215                    max_tokens=MAX_TOKENS,216                    stream=False,217                )218                response_text = completion.choices[0].message.content or ""219            # pylint: disable=broad-except220            except Exception as exc:  # noqa: BLE001221                failure_msg = f"Model request failed ({exc}). Using fallback action."222                print(failure_msg)223                response_text = FALLBACK_ACTION224 225            action_str = parse_model_action(response_text)226            print(f"Step {step}: model suggested -> {action_str}")227 228            result = env.step(BrowserGymAction(action_str=action_str))229            observation = result.observation230 231            reward = result.reward or 0.0232            error_flag = " ERROR" if observation.last_action_error else ""233            history_line = (234                f"Step {step}: {action_str} -> reward {reward:+.2f}{error_flag}"235            )236            history.append(history_line)237            print(238                "  Reward: "239                f"{reward:+.2f} | Done: {result.done} | Last action error: "240                f"{observation.last_action_error}"241            )242 243            if result.done:244                print("Episode complete.")245                break246 247        else:248            print(f"Reached max steps ({MAX_STEPS}).")249 250    finally:251        env.close()252 253 254if __name__ == "__main__":255    main()