Function run

scripts/eval_step_attack.py:255–320

(args: argparse.Namespace, dataset, use_caption=True)

@beartype
def run(args: argparse.Namespace, dataset, use_caption=True) -> None:
    caption_image_fn = None  # Don't use captioning for the demo, due to extra resources required to run BLIP-2.

    agent = construct_agent(
        args,
        captioning_fn=caption_image_fn if args.observation_type == "accessibility_tree_with_captioner" else None,
    )  # NOTE: captioning_fn here is used for captioning input images.

    all_actions = {}
    for ex_id, example in enumerate(dataset):
        print(f"Running evaluation on Example {example['id']}")
        try:
            intent, images, trajectory, meta_data = load(
                example, attack=args.attack, use_caption=use_caption, args=args
            )
        except Exception as e:
            print(f"Error loading example {example['id']}: {repr(e)}")
            continue
        try:
            # Load task.
            logger.info(f"[Intent]: {intent}")

            try:
                print("=" * 30)
                print("Agent: Thinking...")
                action = agent.next_action(
                    trajectory, intent, images=images, meta_data=meta_data, output_response=True
                )
            except ValueError as e:
                # get the error message
                action = create_stop_action(f"ERROR: {str(e)}")

            print("--- Raw prediction ---")
            print(action["raw_prediction"])

            pattern = r"\```(.+?)```"
            match = re.search(pattern, action["raw_prediction"], re.DOTALL)
            if match:
                content = match.group(1)
                content = content.strip()
            else:
                content = action["raw_prediction"]
            # print(content)
            all_actions[example["id"]] = content

        except openai.OpenAIError as e:
            logger.info(f"[OpenAI Error] {repr(e)}")
        except Exception as e:
            logger.info(f"[Unhandled Error] {repr(e)}]")
            import traceback

            # write to error file
            with open(Path(args.result_dir) / "error.txt", "a") as f:
                f.write(f"[Unhandled Error] {repr(e)}\n")
                f.write(traceback.format_exc())  # write stack trace to file

        if "gemini" in args.model:
            print("Sleeping for 11 seconds...")