From 972201bb4d88934d4a320b9a57853bc0c84ee63b Mon Sep 17 00:00:00 2001 From: somew1nd Date: Tue, 1 Sep 2026 21:39:17 -0700 Subject: [PATCH] refactor(explorer): split explorer into facade and phase modules --- artemis/agents/explorer/_facade.py | 34 + artemis/agents/explorer/explorer.py | 2106 +----------------- artemis/agents/explorer/native_runner.py | 796 +++++++ artemis/agents/explorer/perception_tools.py | 664 ++++++ artemis/agents/explorer/run_setup.py | 384 ++++ artemis/agents/explorer/tool_declarations.py | 395 ++++ artemis/agents/explorer/universal_runner.py | 261 +++ 7 files changed, 2589 insertions(+), 2051 deletions(-) create mode 100644 artemis/agents/explorer/_facade.py create mode 100644 artemis/agents/explorer/native_runner.py create mode 100644 artemis/agents/explorer/perception_tools.py create mode 100644 artemis/agents/explorer/run_setup.py create mode 100644 artemis/agents/explorer/tool_declarations.py create mode 100644 artemis/agents/explorer/universal_runner.py diff --git a/artemis/agents/explorer/_facade.py b/artemis/agents/explorer/_facade.py new file mode 100644 index 0000000..21518f6 --- /dev/null +++ b/artemis/agents/explorer/_facade.py @@ -0,0 +1,34 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Late-bound access to the ``artemis.agents.explorer.explorer`` facade module. + +The Explorer implementation is split across several modules, but external +callers (and ``unittest.mock.patch`` targets in the test suite) address +collaborators such as ``settings``, ``StorageManager``, ``draw_dots``, +``search_ui_func``, ``get_llm`` etc. through the historical module path +``artemis.agents.explorer.explorer``. Split-out code therefore resolves those +names at call time via :func:`facade` instead of importing them directly, so +patches applied to the facade module keep affecting the split-out code. + +The import happens inside the function (not at module top level) to stay safe +regardless of which module of the package is imported first. +""" + + +def facade(): + """Return the ``artemis.agents.explorer.explorer`` module object.""" + import artemis.agents.explorer.explorer as explorer_module + + return explorer_module diff --git a/artemis/agents/explorer/explorer.py b/artemis/agents/explorer/explorer.py index 1c48427..1c1a552 100644 --- a/artemis/agents/explorer/explorer.py +++ b/artemis/agents/explorer/explorer.py @@ -12,6 +12,25 @@ # See the License for the specific language governing permissions and # limitations under the License. +"""Explorer agent facade. + +This module is the stable entry point (and ``mock.patch`` target namespace) +for the Explorer agent. The implementation is split across sibling modules: + +- ``tool_declarations``: static tool schemas (universal + native). +- ``perception_tools``: the ``exec_*`` perception/vision tool method group. +- ``universal_runner``: the LangChain-based ``_run_universal`` loop. +- ``run_setup``: flash mode and the pre-loop setup phases of ``run``. +- ``native_runner``: the native Gemini reasoning loop of ``run``. + +All historical module-level names (``settings``, ``StorageManager``, +``genai``, ``search_ui_func``, ``search_by_coordinates_func``, +``_run_object_detection``, ``draw_dots``, ``perform_ocr``, +``is_ocr_configured``, ``get_llm``, ``UNIVERSAL_EXPLORER_TOOLS``, ...) remain +importable and patchable here; split-out code resolves them through this +module at call time (see ``artemis.agents.explorer._facade``). +""" + import asyncio import base64 import glob @@ -52,6 +71,14 @@ from artemis.services.llm import _record_llm_event, _record_llm_retry, get_llm # Import diagnostic functions directly from artemis.agents.explorer.constants import EXPLORE_DESCRIPTIONS +from artemis.agents.explorer.native_runner import NativeRunnerMixin +from artemis.agents.explorer.perception_tools import PerceptionToolsMixin +from artemis.agents.explorer.run_setup import RunSetupMixin +from artemis.agents.explorer.tool_declarations import ( + NATIVE_EXPLORER_TOOL_DECLARATIONS, + UNIVERSAL_EXPLORER_TOOLS, +) +from artemis.agents.explorer.universal_runner import UniversalRunnerMixin from artemis.mcp.xml_search_server import ( search_by_coordinates as search_by_coordinates_func, search_ui as search_ui_func, @@ -141,380 +168,11 @@ async def _generate_content_with_reliability(operation, *, label: str = "Explore await asyncio.sleep(delay) -UNIVERSAL_EXPLORER_TOOLS = [ - { - "type": "function", - "function": { - "name": "ask_perception_tool", - "description": ( - "[Perception] Concurrently executes and awaits three sub-tasks in parallel," - " including database search, coordinate search, and visual object detection." - ), - "parameters": { - "type": "object", - "properties": { - "search_query": { - "type": "string", - "description": "Required. Text query to search for in UI elements.", - }, - "nx": { - "type": "integer", - "description": "Required. Normalized X coordinate (0-1000 range).", - }, - "ny": { - "type": "integer", - "description": "Required. Normalized Y coordinate (0-1000 range).", - }, - "detect_queries": { - "type": "array", - "items": {"type": "string"}, - "description": "Required. List of query strings to visually detect.", - }, - }, - "required": ["search_query", "nx", "ny", "detect_queries"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "detect_objects", - "description": ( - "[Perception] Locates elements, shapes and more using powerful VLM." - " Returns their normalized coordinates (0-1000 scale) with label dots." - ), - "parameters": { - "type": "object", - "properties": { - "target_image_id": { - "type": "string", - "description": "The ID of the image to process.", - }, - "queries": { - "type": "array", - "items": {"type": "string"}, - "description": "A list of single query strings to detect on the screen.", - }, - }, - "required": ["queries", "target_image_id"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "get_ocr_list", - "description": ( - "[Perception] Retrieves all text elements detected on the screen via OCR." - ), - "parameters": {"type": "object", "properties": {}}, - }, - }, - { - "type": "function", - "function": { - "name": "ask_image_processor", - "description": ("[Transformation] A scripting tool for pixel-level image processing."), - "parameters": { - "type": "object", - "properties": { - "target_image_id": { - "type": "string", - "description": "The ID of the image in the Image Pool to start from.", - }, - "instruction": { - "type": "string", - "description": "1-2 precise commands describing a simple task.", - }, - }, - "required": ["instruction", "target_image_id"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "inspect_region", - "description": ( - "[Perception] Crops and resizes a bounding box region of the original screenshot." - ), - "parameters": { - "type": "object", - "properties": { - "x_min": {"type": "integer", "description": "Left X coordinate (0-1000)."}, - "y_min": {"type": "integer", "description": "Top Y coordinate (0-1000)."}, - "x_max": {"type": "integer", "description": "Right X coordinate (0-1000)."}, - "y_max": {"type": "integer", "description": "Bottom Y coordinate (0-1000)."}, - "zoom_factor": { - "type": "number", - "description": "Zoom factor to upscale the cropped region (1.0 to 4.0).", - }, - }, - "required": ["x_min", "y_min", "x_max", "y_max", "zoom_factor"], - }, - }, - }, - { - "type": "function", - "function": { - "name": "submit_answer", - "description": ( - "[Finalization] Submits the final list of ranked candidate UI elements matching" - " the search query, or a fallback message if no elements were found." - ), - "parameters": { - "type": "object", - "properties": { - "candidates": { - "type": "array", - "items": { - "type": "object", - "properties": { - "label": { - "type": "string", - "description": "Reference label (e.g. S1, D2, T3)", - }, - "description": { - "type": "string", - "description": "Brief description of this candidate element", - }, - "coords": { - "type": "array", - "items": {"type": "integer"}, - "description": ( - "[nx, ny] coordinates in normalized 0-1000 scale" - ), - }, - }, - "required": ["label", "coords"], - }, - "description": "A ranked list of candidate UI elements matching the query.", - }, - "fallback_message": { - "type": "string", - "description": ( - "Optional fallback explanation if target elements were not found." - ), - }, - }, - "required": ["candidates"], - }, - }, - }, -] - - -class Explorer: +class Explorer(PerceptionToolsMixin, UniversalRunnerMixin, RunSetupMixin, NativeRunnerMixin): DENYLIST_TEMPLATE = ( "\n# TOOL DENYLIST\n- The following tools are denylisted and cannot be used: {tools}\n" ) - TOOLS = [ - types.FunctionDeclaration( - name="ask_perception_tool", - description=( - "[Perception] Concurrently executes and awaits three sub-tasks" - " in parallel, including database search, coordinate search," - " and visual object detection on the screen.\n\nOutput" - " Format:\nReturns:'text' (str): A single string consolidating" - " results with 0-1000 normalized coordinates, including UI" - " search matches labeled [X]," - " coordinate search matches labeled [X], and visually detected" - " object targets labeled [D].\n(Annotated screenshots are also" - " added to the context containing corresponding colored label" - " dots)." - ), - parameters=types.Schema( - type=types.Type.OBJECT, - properties={ - "search_query": types.Schema( - type=types.Type.STRING, - description=("Required. Text query to search for in UI elements."), - ), - "nx": types.Schema( - type=types.Type.INTEGER, - description=( - "Required. Normalized X coordinate (0-1000 range)" - " to audit for overlapping elements." - ), - ), - "ny": types.Schema( - type=types.Type.INTEGER, - description=( - "Required. Normalized Y coordinate (0-1000 range)" - " to audit for overlapping elements." - ), - ), - "detect_queries": types.Schema( - type=types.Type.ARRAY, - items=types.Schema(type=types.Type.STRING), - description=( - "Required. List of query strings to visually detect on the screen." - ), - ), - }, - required=["search_query", "nx", "ny", "detect_queries"], - ), - ), - types.FunctionDeclaration( - name="detect_objects", - description=( - "[Perception] Locates elements, shapes and more using powerful" - " VLM. Returns their normalized coordinates (in a [0, 1000]" - " scale) labeled with [D1], [D2] etc. in the text output, along" - " with an annotated image containing corresponding visual label" - " dots." - ), - parameters=types.Schema( - type=types.Type.OBJECT, - properties={ - "target_image_id": types.Schema( - type=types.Type.STRING, - description="The ID of the image to process.", - ), - "queries": types.Schema( - type=types.Type.ARRAY, - items=types.Schema(type=types.Type.STRING), - description=("A list of single query strings to detect on the screen."), - ), - }, - required=["queries", "target_image_id"], - ), - ), - types.FunctionDeclaration( - name="get_ocr_list", - description=( - "[Perception] Retrieves all text elements detected on the" - " original, unprocessed screen via OCR. Returns their" - " normalized coordinates (in a [0, 1000] scale) labeled with" - " [T1], [T2] etc. in the text output, along with an annotated" - " image containing corresponding visual label dots." - ), - parameters=types.Schema( - type=types.Type.OBJECT, - properties={}, - ), - ), - types.FunctionDeclaration( - name="ask_image_processor", - description=( - "[Transformation] A scripting tool for pixel-level image" - " processing. CAUTION: This tool is slow and expensive. Returns" - " a summary of the operations performed and new image IDs," - " along with new labels and screen coordinates." - ), - parameters=types.Schema( - type=types.Type.OBJECT, - properties={ - "target_image_id": types.Schema( - type=types.Type.STRING, - description=( - "The ID of the image in the Image Pool to start" - " from (e.g. 'img_0', 'img_1')." - ), - ), - "instruction": types.Schema( - type=types.Type.STRING, - description=( - "1-2 precise commands describing a simple task." - " Include all necessary coordinates and other" - " parameters.\n" - ), - ), - }, - required=["instruction", "target_image_id"], - ), - ), - types.FunctionDeclaration( - name="submit_answer", - description=( - "[Finalization] Submits the final list of ranked candidate UI" - " elements matching the search query, or a fallback message if" - " no elements were found, to complete the search task. Fails" - " validation if: (1) candidates is empty and no" - " fallback_message is provided, (2) coordinates are not exactly" - " 2 integers [nx, ny], or (3) coordinates lie outside" - " normalized [0, 1000] range." - ), - parameters=types.Schema( - type=types.Type.OBJECT, - properties={ - "candidates": types.Schema( - type=types.Type.ARRAY, - items=types.Schema( - type=types.Type.OBJECT, - properties={ - "label": types.Schema( - type=types.Type.STRING, - description=( - "Reference label (e.g., S1, D2, T3, or number like 3)" - ), - ), - "coords": types.Schema( - type=types.Type.ARRAY, - items=types.Schema(type=types.Type.INTEGER), - description=("[nx, ny] coordinates in normalized 0-1000 scale"), - ), - "description": types.Schema( - type=types.Type.STRING, - description=("Brief description of this candidate element"), - ), - }, - required=["label", "coords"], - ), - description=( - "A ranked list of candidate UI elements that match" - " the query, sorted from highest to lowest" - " confidence (maximum 10 candidates). Leave empty" - " if no matching elements were found on the screen." - ), - ), - "fallback_message": types.Schema( - type=types.Type.STRING, - description=( - "Optional fallback explanation if target elements" - " were not found or if there are special" - " observations." - ), - ), - }, - required=["candidates"], - ), - ), - types.FunctionDeclaration( - name="inspect_region", - description=( - "[Perception] Crops and resizes a bounding box region of the" - " original screenshot. Returns the cropped and resized image" - " directly, without any image ID or coordinate transform." - ), - parameters=types.Schema( - type=types.Type.OBJECT, - properties={ - "x_min": types.Schema( - type=types.Type.INTEGER, - description="Left X coordinate (0-1000).", - ), - "y_min": types.Schema( - type=types.Type.INTEGER, - description="Top Y coordinate (0-1000).", - ), - "x_max": types.Schema( - type=types.Type.INTEGER, - description="Right X coordinate (0-1000).", - ), - "y_max": types.Schema( - type=types.Type.INTEGER, - description="Bottom Y coordinate (0-1000).", - ), - "zoom_factor": types.Schema( - type=types.Type.NUMBER, - description=("Zoom factor to upscale the cropped region (1.0 to 4.0)."), - ), - }, - required=["x_min", "y_min", "x_max", "y_max", "zoom_factor"], - ), - ), - ] + TOOLS = NATIVE_EXPLORER_TOOL_DECLARATIONS def __init__(self, ctx: ArtemisContext): self.ctx = ctx @@ -605,805 +263,6 @@ class Explorer: tools = [tool for tool in tools if tool.name == "submit_answer"] return tools - async def _search_ui_helper(self, query: str, prefix: str = "S", color: str = "red") -> dict: - if not self.image_name: - return { - "text": ( - "Error: UI data for the current screen is not found in Data" - " Engine. It might not be synced yet." - ), - "image_path": None, - } - - # 1. Try with high precision threshold - raw_res = search_ui_func(self.image_name, query, 0.7) - if "error" in raw_res: - return {"text": raw_res["error"], "image_path": None} - - matches = raw_res.get("matches", []) - - # 2. If fewer than 3 matches, try with a lower threshold and merge - if len(matches) < 3: - raw_res_low = search_ui_func(self.image_name, query, 0.4) - if "error" not in raw_res_low: - low_matches = raw_res_low.get("matches", []) - existing_keys = { - ( - m.get("matched_text"), - tuple(m.get("bounds")) if m.get("bounds") else None, - ) - for m in matches - } - for m in low_matches: - key = ( - m.get("matched_text"), - tuple(m.get("bounds")) if m.get("bounds") else None, - ) - if key not in existing_keys: - matches.append(m) - existing_keys.add(key) - - if not matches: - return {"text": "No matches found.", "image_path": None} - - points = [] - labels = [] - text_output = [] - - for m in matches: - m_type = m.get("type", "xml") - m_prefix = "O" if m_type == "ocr" else prefix - label_id = f"{m_prefix}{self.global_label_idx}" - self.global_label_idx += 1 - bounds = m.get("bounds") - matched_text = m.get("matched_text", "") - - if bounds and len(bounds) == 4: - left, top, right, bottom = bounds - cx_pixel = (left + right) // 2 - cy_pixel = (top + bottom) // 2 - - points.append([cx_pixel, cy_pixel]) - labels.append(label_id) - - cx_norm = int(max(0, min(1000, cx_pixel * 1000 / self.width))) - cy_norm = int(max(0, min(1000, cy_pixel * 1000 / self.height))) - - text_output.append(f"[{label_id}] '{matched_text}' at [{cx_norm},{cy_norm}]") - else: - text_output.append(f"[{label_id}] '{matched_text}' at unknown") - - if not points: - return {"text": " | ".join(text_output), "image_path": None} - - base_dir = settings.TRACES_PATH - images_dir = base_dir / "images" - search_ui_dir = images_dir / "search_ui" - search_ui_dir.mkdir(parents=True, exist_ok=True) - - existing_files = glob.glob(str(search_ui_dir / f"{self.image_name}_*.jpg")) - max_seq = 0 - for f in existing_files: - match = re.search(r"_(\d+)\.jpg$", f) - if match: - max_seq = max(max_seq, int(match.group(1))) - seq = max_seq + 1 - output_path = search_ui_dir / f"{self.image_name}_{seq}.jpg" - - try: - draw_dots( - self.screenshot_path, - points, - labels, - str(output_path), - color=color, - ) - image_path = str(output_path) - except Exception as e: - logger.warning(f"Failed to draw dots: {e}") - image_path = None - - return {"text": " | ".join(text_output), "image_path": image_path} - - async def _search_by_coords_helper( - self, nx: int, ny: int, prefix: str = "X", color: str = "blue" - ) -> dict: - if not self.image_name: - return { - "text": ( - "Error: UI data for the current screen is not found in Data" - " Engine. It might not be synced yet." - ), - "image_path": None, - } - - x = int(max(0, min(self.width, nx * self.width / 1000))) - y = int(max(0, min(self.height, ny * self.height / 1000))) - raw_res = search_by_coordinates_func(self.image_name, x, y) - - label_id = f"{prefix}{self.global_label_idx}" - self.global_label_idx += 1 - - # Draw a dot at the query coordinates - base_dir = settings.TRACES_PATH - images_dir = base_dir / "images" - coords_audit_dir = images_dir / "coords_audit" - coords_audit_dir.mkdir(parents=True, exist_ok=True) - - existing_files = glob.glob(str(coords_audit_dir / f"{self.image_name}_*.jpg")) - max_seq = 0 - for f in existing_files: - match = re.search(r"_(\d+)\.jpg$", f) - if match: - max_seq = max(max_seq, int(match.group(1))) - seq = max_seq + 1 - output_path = coords_audit_dir / f"{self.image_name}_{seq}.jpg" - - try: - draw_dots( - self.screenshot_path, - [[x, y]], - [label_id], - str(output_path), - color=color, - ) - image_path = str(output_path) - except Exception as e: - logger.warning(f"Failed to draw blue dots: {e}") - image_path = None - - raw_res_str = str(raw_res).replace("\n", " | ").replace(" | | ", " | ") - return { - "text": (f"Matched element at [{nx},{ny}]: [{label_id}] | {raw_res_str}"), - "image_path": image_path, - } - - @trace(type="tool", name="detect_objects") - async def exec_detect_objects( - self, - queries: list[str], - target_image_id: str = "img_0", - prefix: str = "D", - color: str = "red", - ) -> dict: - try: - target_info = self.image_pool.get(target_image_id) - if not target_info: - return { - "text": (f"Error: {target_image_id} not found in Image Pool."), - "image_path": None, - } - - target_path = target_info["path"] - transform = target_info["transform"] - - # Load object detector templates - detector_prompt_path = ( - Path(__file__).parent.parent / "object_detector" / "object_detector.json" - ) - global_timeout = None - try: - with open(detector_prompt_path, encoding="utf-8") as f: - detector_config = json.load(f) - detector_templates = detector_config.get("templates", []) - detector_instructions = detector_config.get("instructions", "") - global_timeout = detector_config.get("global_timeout", None) - except Exception as e: - logger.warning(f"Failed to load detector prompt config: {e}") - detector_templates = [ - "Point to the following objects in the provided image: {labels_str}." - ] - detector_instructions = "" - - templates = [f"{t}\n\n{detector_instructions}" for t in detector_templates] - - result = await _run_object_detection( - self.ctx, - target_path, - queries, - templates, - global_timeout=global_timeout, - ) - try: - detected_items = result.get("detected", []) - failed_queries = result.get("failed", []) - - if not detected_items: - text_output = ["No objects detected."] - if failed_queries: - text_output.append( - "\n[NOTE: Failed to detect the following queries:" - f" {', '.join(failed_queries)}]" - ) - return { - "text": "".join(text_output), - "image_path": str(target_path), - } - - base_dir = settings.TRACES_PATH - images_dir = base_dir / "images" - object_detect_dir = images_dir / "object_detect" - object_detect_dir.mkdir(parents=True, exist_ok=True) - - image_name_safe = self.image_name or "temp_image" - existing_files = glob.glob(str(object_detect_dir / f"{image_name_safe}_*.jpg")) - max_seq = 0 - for f in existing_files: - match = re.search(r"_(\d+)\.jpg$", f) - if match: - max_seq = max(max_seq, int(match.group(1))) - seq = max_seq + 1 - output_path = object_detect_dir / f"{image_name_safe}_{seq}.jpg" - - points = [] - d_labels = [] - text_output = [] - - if target_image_id != "img_0": - text_output.append( - "[NOTE: Coordinates are automatically mapped back to" - " the original 1080x2400 screen space from" - f" {target_image_id}]" - ) - - for item in detected_items: - label_id = f"{prefix}{self.global_label_idx}" - self.global_label_idx += 1 - pos = item.get("point") - if pos and isinstance(pos, list) and len(pos) == 2: - x_norm, y_norm = pos - - t_img = cv2.imread(target_path) - t_h, t_w = t_img.shape[:2] - - # Pixels in target image - x_target_pixel = (x_norm / 1000) * t_w - y_target_pixel = (y_norm / 1000) * t_h - - # Pixels in original image - x_orig_pixel = (x_target_pixel / transform["scale_x"]) + transform[ - "offset_x" - ] - y_orig_pixel = (y_target_pixel / transform["scale_y"]) + transform[ - "offset_y" - ] - - points.append([int(x_orig_pixel), int(y_orig_pixel)]) - d_labels.append(label_id) - - # Original image norm - x_orig_norm = int(max(0, min(1000, x_orig_pixel * 1000 / self.width))) - y_orig_norm = int(max(0, min(1000, y_orig_pixel * 1000 / self.height))) - - text_output.append( - f'[{label_id}] "{item.get("label")}" at [{x_orig_norm},{y_orig_norm}]' - ) - - draw_dots( - self.screenshot_path, - points, - d_labels, - str(output_path), - color=color, - ) - - if failed_queries: - text_output.append( - " | [NOTE: Failed to detect the following queries:" - f" {', '.join(failed_queries)}]" - ) - - return { - "text": " | ".join(text_output), - "image_path": str(output_path), - } - - except Exception as e: - logger.warning(f"Failed to process detection output for annotation: {e}") - return {"text": str(result), "image_path": None} - except Exception as e: - return { - "text": f"Error: Object detection failed: {e}", - "image_path": None, - } - - @trace(type="tool", name="ask_perception_tool") - async def exec_ask_perception_tool( - self, search_query: str, nx: int, ny: int, detect_queries: list[str] - ) -> dict: - if nx is None or ny is None or not (0 <= nx <= 1000) or not (0 <= ny <= 1000): - coords_invalid = True - else: - coords_invalid = False - - # Run tasks concurrently in parallel! - tasks = [ - self._search_ui_helper(search_query, prefix="X", color="green"), - self._search_by_coords_helper(nx, ny, prefix="X", color="blue") - if not coords_invalid - else asyncio.sleep( - 0, - result={ - "text": "Error: Invalid coordinates format", - "image_path": None, - }, - ), - self.exec_detect_objects( - detect_queries, target_image_id="img_0", prefix="D", color="red" - ), - ] - - results = await asyncio.gather(*tasks) - - # Consolidate text results - text_parts = [] - - xml_text = results[0].get("text", "No matches found.") - if xml_text and xml_text != "No matches found.": - text_parts.append(f"Text Search Results are: {xml_text}") - - coord_text = results[1].get("text", "No elements found.") - if coord_text and coord_text != "No elements found.": - text_parts.append(f"Coordinate Search Results are: {coord_text}") - - obj_text = results[2].get("text", "No objects detected.") - if obj_text: - if obj_text.startswith("No objects detected."): - note_part = obj_text.replace("No objects detected.", "").strip() - if note_part: - text_parts.append(f"Object Detection Results are: {note_part}") - else: - text_parts.append(f"Object Detection Results are: {obj_text}") - - if not text_parts: - unified_text = "No matches or objects found across any perception method." - else: - unified_text = ". ".join(text_parts) + "." - - # Collect image paths - image_paths = [] - for r in results: - if isinstance(r, dict) and r.get("image_path"): - image_paths.append(r["image_path"]) - - return { - "text": unified_text, - "image_paths": image_paths, - } - - @trace(type="tool", name="ask_image_processor") - async def exec_ask_image_processor( - self, instruction: str, target_image_id: str = "img_0" - ) -> dict: - coder = ImageProcessor(self.ctx) - target_info = self.image_pool.get(target_image_id) - if not target_info: - return { - "text": f"Error: {target_image_id} not found in Image Pool.", - "image_path": None, - } - - target_path = target_info["path"] - result = await coder.run(instruction, target_path) - - if not isinstance(result, dict) or "error" in result: - err_msg = result.get("error") if isinstance(result, dict) else "Unknown error" - return { - "text": f"Image Processor failed: {err_msg}", - "image_paths": [], - } - - outputs = result.get("outputs", []) - summary = result.get("summary", "") - - parent_transform = target_info["transform"] - registered_ids = [] - annotations_sections = [] - image_paths = [] - - for entry in outputs: - new_id = f"img_{self.next_img_id}" - self.next_img_id += 1 - - new_path = entry["path"] - image_paths.append(new_path) - transform = entry["transform"] - - final_scale_x = parent_transform["scale_x"] * transform["scale_x"] - final_scale_y = parent_transform["scale_y"] * transform["scale_y"] - final_offset_x = ( - transform["offset_x"] / parent_transform["scale_x"] - ) + parent_transform["offset_x"] - final_offset_y = ( - transform["offset_y"] / parent_transform["scale_y"] - ) + parent_transform["offset_y"] - - self.image_pool[new_id] = { - "path": new_path, - "transform": { - "offset_x": final_offset_x, - "offset_y": final_offset_y, - "scale_x": final_scale_x, - "scale_y": final_scale_y, - }, - "description": (f"Generated by coder from {target_image_id}: {summary}"), - } - registered_ids.append(new_id) - - # Parse and translate annotations - annotations = entry.get("annotations", {}) - if annotations: - lines = [f"{new_id}:"] - for label, coord in sorted(annotations.items(), key=lambda item: item[0]): - px, py = coord - sx = int(round((px / final_scale_x) + final_offset_x)) - sy = int(round((py / final_scale_y) + final_offset_y)) - lines.append(f" - [{label}]: [{sx}, {sy}]") - annotations_sections.append("\n".join(lines)) - - if registered_ids: - ids_str = ", ".join(registered_ids) - text = ( - "Image Processor completed successfully. New image(s) saved as" - f" {ids_str} in Image Pool. Summary: {summary}" - ) - if annotations_sections: - text += "\n\nAnnotations (wrt original screenshot):\n" + "\n".join( - annotations_sections - ) - return { - "text": text, - "image_paths": image_paths, - } - else: - return { - "text": f"ImageProcessor error: {err_msg}", - "image_paths": [], - } - - output_path = result.get("output_path") - trans = result.get("transform") - new_image_id = f"img_{len(self.image_pool)}" - if output_path and trans: - self.image_pool[new_image_id] = { - "path": output_path, - "transform": trans, - } - - text_msg = ( - f"Image processed successfully. New image saved as" - f" '{new_image_id}'. You can now use '{new_image_id}' as" - " 'target_image_id' in subsequent tool calls." - ) - return { - "text": text_msg, - "image_path": output_path, - "image_id": new_image_id, - } - - @trace(type="tool", name="get_ocr_list") - async def exec_get_ocr_list(self, prefix: str = "O", color: str = "green") -> dict: - if not self.image_name: - return { - "text": ("Error: UI data for the current screen is not found in Data Engine."), - "image_path": None, - } - - try: - db_path = settings.DATA_ENGINE_DB_PATH - base_dir = settings.TRACES_PATH - storage = StorageManager(db_path=str(db_path), base_dir=base_dir) - - record = storage.get_ui_record(self.image_name) - if not record or not record.ocr_result: - return { - "text": "No text elements detected on the screen.", - "image_path": None, - } - - points = [] - labels = [] - text_output = [] - - for item in record.ocr_result: - text = item.get("text", "").strip() - bounds = item.get("bounds", []) - if text and len(bounds) == 4: - left, top, right, bottom = bounds - cx = (left + right) // 2 - cy = (top + bottom) // 2 - if 0 <= cx <= self.width and 0 <= cy <= self.height: - label_id = f"{prefix}{self.global_label_idx}" - self.global_label_idx += 1 - points.append([cx, cy]) - labels.append(label_id) - - cx_norm = int(max(0, min(1000, cx * 1000 / self.width))) - cy_norm = int(max(0, min(1000, cy * 1000 / self.height))) - - text_output.append(f"[{label_id}] '{text}' coords: [{cx_norm},{cy_norm}]") - - if not points: - return { - "text": "No valid OCR results found.", - "image_path": None, - } - - images_dir = base_dir / "images" - ocr_dir = images_dir / "ocr" - ocr_dir.mkdir(parents=True, exist_ok=True) - - output_path = ocr_dir / f"{self.image_name}_{self.global_label_idx}.jpg" - draw_dots(self.screenshot_path, points, labels, str(output_path), color=color) - - return { - "text": "\n".join(text_output), - "image_path": str(output_path), - } - - except Exception as e: - return { - "text": f"Error retrieving or processing OCR results: {e}", - "image_path": None, - } - - @trace(type="tool", name="inspect_region") - async def exec_inspect_region( - self, - x_min: int, - y_min: int, - x_max: int, - y_max: int, - zoom_factor: float = 2.0, - ) -> dict: - try: - px_start = int(max(0, min(self.width, x_min * self.width / 1000.0))) - py_start = int(max(0, min(self.height, y_min * self.height / 1000.0))) - px_end = int(max(0, min(self.width, x_max * self.width / 1000.0))) - py_end = int(max(0, min(self.height, y_max * self.height / 1000.0))) - - if px_end <= px_start or py_end <= py_start: - return { - "text": ( - f"Error: Invalid crop coordinates. x_min={x_min}," - f" x_max={x_max}, y_min={y_min}, y_max={y_max}." - " Bounding box dimensions must be strictly positive." - ), - "image_path": None, - } - - img = cv2.imread(self.screenshot_path) - if img is None: - return { - "text": ( - f"Error: Failed to read original screenshot at {self.screenshot_path}" - ), - "image_path": None, - } - - cropped = img[py_start:py_end, px_start:px_end] - - if zoom_factor is None: - zoom_factor = 2.0 - zoom_factor = max(1.0, min(4.0, float(zoom_factor))) - - c_h, c_w = cropped.shape[:2] - if c_h > 0 and c_w > 0: - new_w = int(round(c_w * zoom_factor)) - new_h = int(round(c_h * zoom_factor)) - if new_w > 0 and new_h > 0: - cropped = cv2.resize(cropped, (new_w, new_h)) - - base_dir = settings.TRACES_PATH - images_dir = base_dir / "images" - inspect_region_dir = images_dir / "inspect_region" - inspect_region_dir.mkdir(parents=True, exist_ok=True) - - image_name_safe = self.image_name or "temp_image" - output_path = inspect_region_dir / f"{image_name_safe}_{self.global_label_idx}.jpg" - cv2.imwrite(str(output_path), cropped) - - return { - "text": (f"Inspected region coordinates: [{x_min}, {y_min}] to [{x_max}, {y_max}]"), - "image_path": str(output_path), - } - - except Exception as e: - return { - "text": f"Error: Region inspection failed: {e}", - "image_path": None, - } - - async def _run_universal( - self, - query: str, - context_feedback: str, - screenshot_path: str, - state: State, - minimal_list: str, - version: str, - prompt_template: str, - max_iterations: int, - ) -> str: - """Executes Explorer reasoning loop via Universal LangChain ChatModel.""" - llm = get_llm(self.ctx, name="explorer") - - # Filter universal tools based on denylist and OCR configuration - denylisted = set(self.denylisted_tools) - if not is_ocr_configured(): - denylisted.add("get_ocr_list") - exposed_tools = [ - t for t in UNIVERSAL_EXPLORER_TOOLS if t["function"]["name"] not in denylisted - ] - bound_llm = llm.bind_tools(exposed_tools) - - with open(screenshot_path, "rb") as f: - img_b64 = base64.b64encode(f.read()).decode("utf-8") - - user_content: list[dict[str, Any]] = [ - { - "type": "text", - "text": ( - f"Operator Request:\n- Query: {query}\n" - f"- Context Feedback: {context_feedback}\n\n" - "Initial marked UI elements list:\n" - f"{minimal_list}\n" - ), - }, - { - "type": "image_url", - "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}, - }, - ] - - messages: list[BaseMessage] = [ - SystemMessage(content=prompt_template), - HumanMessage(content=user_content), - ] - - iterations = 0 - agent_outcome = "" - - while iterations < max_iterations: - iterations += 1 - is_final_turn = iterations == max_iterations - - if is_final_turn: - messages.append( - HumanMessage( - content=( - "[WARNING] This is your final iteration. You MUST" - " call 'submit_answer' to submit your final result." - ) - ) - ) - submit_only_tools = [ - t for t in exposed_tools if t["function"]["name"] == "submit_answer" - ] - current_llm = llm.bind_tools(submit_only_tools) if submit_only_tools else bound_llm - else: - current_llm = bound_llm - - response = await asyncio.wait_for(current_llm.ainvoke(messages), timeout=180) - messages.append(response) - - tool_calls = getattr(response, "tool_calls", None) or [] - if not tool_calls: - logger.warning( - f"Universal Explorer iteration {iterations}: No tool calls generated." - ) - if is_final_turn: - return json.dumps( - { - "candidates": [], - "fallback_message": str(response.content) or "No candidates found.", - }, - ensure_ascii=False, - ) - continue - - # Check for submit_answer - submit_call = next((tc for tc in tool_calls if tc.get("name") == "submit_answer"), None) - if submit_call: - args = submit_call.get("args", {}) - candidates = args.get("candidates", []) - fallback_message = args.get("fallback_message", "") - - # Validate coordinates - valid_candidates = [] - for cand in candidates: - if isinstance(cand, dict) and "coords" in cand: - coords = cand["coords"] - if isinstance(coords, list) and len(coords) == 2: - try: - nx, ny = int(coords[0]), int(coords[1]) - if 0 <= nx <= 1000 and 0 <= ny <= 1000: - valid_candidates.append(cand) - except (ValueError, TypeError): - pass - - agent_outcome = json.dumps( - { - "candidates": valid_candidates, - "fallback_message": fallback_message, - }, - ensure_ascii=False, - ) - return agent_outcome - - # Execute non-submit tools - for tc in tool_calls: - name = tc.get("name") - args = tc.get("args", {}) - call_id = tc.get("id", f"call_{iterations}_{name}") - - if name in self.denylisted_tools: - messages.append( - ToolMessage( - tool_call_id=call_id, - name=name, - content=f"Tool '{name}' is denylisted and unavailable.", - ) - ) - continue - - try: - if name == "ask_perception_tool": - res = await self.exec_ask_perception_tool( - search_query=args.get("search_query"), - nx=args.get("nx"), - ny=args.get("ny"), - detect_queries=args.get("detect_queries"), - ) - elif name == "detect_objects": - res = await self.exec_detect_objects( - queries=args.get("queries"), - target_image_id=args.get("target_image_id", "img_0"), - ) - elif name == "get_ocr_list": - res = await self.exec_get_ocr_list() - elif name == "ask_image_processor": - res = await self.exec_ask_image_processor( - instruction=args.get("instruction"), - target_image_id=args.get("target_image_id", "img_0"), - ) - elif name == "inspect_region": - res = await self.exec_inspect_region( - x_min=args.get("x_min"), - y_min=args.get("y_min"), - x_max=args.get("x_max"), - y_max=args.get("y_max"), - zoom_factor=args.get("zoom_factor", 2.0), - ) - else: - res = {"text": f"Error: Tool '{name}' is not recognized."} - - res_text = res.get("text") or res.get("result") or str(res) - messages.append(ToolMessage(tool_call_id=call_id, name=name, content=res_text)) - - except Exception as tool_err: - logger.error(f"Error executing tool '{name}': {tool_err}") - messages.append( - ToolMessage( - tool_call_id=call_id, - name=name, - content=f"Error executing tool '{name}': {tool_err}", - ) - ) - - if not agent_outcome: - agent_outcome = json.dumps( - { - "candidates": [], - "fallback_message": ( - "Explorer reached max iterations without finding candidates." - ), - }, - ensure_ascii=False, - ) - - return agent_outcome - @trace(type="agent", name="explorer") async def run( self, @@ -1415,316 +274,39 @@ class Explorer: enable_caching: bool = False, version: Literal["flash", "pro", "ultra"] = "pro", ) -> str: - ctx = self.ctx if version == "flash": - detector_prompt_path = ( - Path(__file__).parent.parent / "object_detector" / "object_detector.json" - ) - global_timeout = None - try: - with open(detector_prompt_path, encoding="utf-8") as f: - detector_config = json.load(f) - detector_templates = detector_config.get("templates", []) - detector_instructions = detector_config.get("instructions", "") - global_timeout = detector_config.get("global_timeout", None) - except Exception as e: - logger.warning(f"Failed to load detector prompt config: {e}") - detector_templates = [ - "Point to the following objects in the provided image: {labels_str}." - ] - detector_instructions = "" - - templates = [f"{t}\n\n{detector_instructions}" for t in detector_templates] - - self.screenshot_path = screenshot_path - self.image_name = None - - queries = [q.strip() for q in query.split("|") if q.strip()] - - try: - result = await _run_object_detection( - self.ctx, - screenshot_path, - queries, - templates, - global_timeout=global_timeout, - ) - detected_items = result.get("detected", []) - candidates = [] - for idx, item in enumerate(detected_items): - pos = item.get("point") - if pos and isinstance(pos, list) and len(pos) == 2: - candidates.append( - { - "label": f"D{idx + 1}", - "coords": pos, - "description": item.get("label", query), - } - ) - fallback_message = "" if candidates else f"Failed to detect: {query}" - return json.dumps( - { - "candidates": candidates, - "fallback_message": fallback_message, - }, - ensure_ascii=False, - ) - except Exception as e: - logger.error(f"Flash mode object detection failed: {e}") - return json.dumps( - { - "candidates": [], - "fallback_message": f"Flash mode detection error: {e}", - }, - ensure_ascii=False, - ) + return await self._run_flash(query, screenshot_path) self.http_client = httpx.AsyncClient() # 1. Compute current image hash and verify with Data Engine - image_name = None - record = None - try: - sha256_hash = hashlib.sha256() - with open(screenshot_path, "rb") as f: - for byte_block in iter(lambda: f.read(4096), b""): - sha256_hash.update(byte_block) - computed_hash = sha256_hash.hexdigest() - logger.info(f"Computed screenshot hash: {computed_hash}") - - # Check if it exists in DB - db_path = settings.DATA_ENGINE_DB_PATH - base_dir = settings.TRACES_PATH - storage = StorageManager(db_path, base_dir) - record = storage.get_image(computed_hash) - - if record: - image_name = computed_hash - else: - logger.warning( - f"Image hash {computed_hash} not found in Data Engine DB." - " Data Engine might not be synced yet." - ) - - except Exception as e: - logger.warning(f"Failed to compute hash or check DB: {e}") - + image_name, record = self._resolve_image_record(screenshot_path) self.image_name = image_name self.screenshot_path = screenshot_path # Resolve parameters - mode = version.capitalize() - version_denylist = set() - if version == "ultra": - max_iterations = 8 - elif version == "pro": - max_iterations = 3 - version_denylist = { - "ask_image_processor", - "get_ocr_list", - "inspect_region", - "detect_objects", - } - else: - max_iterations = 3 - version_denylist = { - "ask_image_processor", - "get_ocr_list", - "inspect_region", - } - - if not is_ocr_configured(): - version_denylist.add("get_ocr_list") - - try: - denylisted_config = ( - self.ctx.agent_config.denylisted_tools.get("explorer", []) - if self.ctx.agent_config - else [] - ) - self.denylisted_tools = version_denylist.union(set(denylisted_config)) - except (TypeError, AttributeError): - self.denylisted_tools = version_denylist + mode, max_iterations = self._resolve_version_limits(version) # 2. Prepare Native Tools declarations - self.width = 1080 - self.height = 2400 - operator_raw_data = getattr(state, "operator_raw_data", {}) or {} - w_raw = operator_raw_data.get("width") - h_raw = operator_raw_data.get("height") - if isinstance(w_raw, int) and isinstance(h_raw, int): - self.width = w_raw - self.height = h_raw - else: - if ctx.device and getattr(ctx.device, "device_width", None): - self.width = ctx.device.device_width - if ctx.device and getattr(ctx.device, "device_height", None): - self.height = ctx.device.device_height + self._resolve_screen_dimensions(state) # Initialize Image Pool - self.image_pool = { - "img_0": { - "path": screenshot_path, - "transform": { - "offset_x": 0.0, - "offset_y": 0.0, - "scale_x": 1.0, - "scale_y": 1.0, - }, - "description": "Original complete screenshot", - } - } - self.next_img_id = 1 + self._init_image_pool(screenshot_path) # 4. Initialize or reuse dynamic context-level GenAI client for connection pooling - client = getattr(ctx, "_genai_client", None) - if client is None: - logger.info( - "Initializing new GenAI client on context for connection pooling (Explorer)..." - ) - client = genai.Client( - api_key=settings.GOOGLE_API_KEY.get_secret_value() - if settings.GOOGLE_API_KEY - else None - ) - ctx._genai_client = client + client = self._ensure_genai_client() - llm_cfg = ctx.llm_config.explorer - temperature = 0.1 - fallback_model = None - thinking_level = None - if llm_cfg: - model_name = llm_cfg.model - if "/" in model_name: - model_name = model_name.split("/")[-1] - if getattr(llm_cfg, "temperature", None) is not None: - temperature = llm_cfg.temperature - if getattr(llm_cfg, "thinking_level", None) is not None: - thinking_level = llm_cfg.thinking_level - if getattr(llm_cfg, "fallback", None): - fallback_model = llm_cfg.fallback.model - if "/" in fallback_model: - fallback_model = fallback_model.split("/")[-1] - else: - model_name = "gemini-3.7-flash" + model_name, temperature, thinking_level, fallback_model = self._resolve_model_params() # 5. Construct Prompt & Initial Message List - prompt_path = Path(__file__).parent / "explorer.json" - if not prompt_path.exists(): - return "Error: Explorer prompt template not found." - - try: - content = prompt_path.read_text(encoding="utf-8") - content_no_comments = re.sub(r"(? types.Part: - if use_file_api: - file_ref = client.files.upload(file=file_path) - uploaded_files.append(file_ref) - return types.Part( - file_data=types.FileData( - file_uri=file_ref.uri, - mime_type=file_ref.mime_type or "image/jpeg", - ) - ) - else: - with open(file_path, "rb") as f: - img_bytes = f.read() - return types.Part.from_bytes(data=img_bytes, mime_type="image/jpeg") - - cached_content = None - try: - # Upload initial screenshot - initial_part = get_image_part(img_to_read) - - contents = [ - types.Content( - role="user", - parts=[ - types.Part.from_text( - text=f"""Operator Request: -- Query: {query} -- Context Feedback: {context_feedback} - -Initial marked UI elements list (corresponding to numbers â‘ , â‘¡... in the image): -{minimal_list} -""" - ), - initial_part, - ], - ) - ] - - # Caching initialization - if enable_caching is None: - env_cache = os.getenv("ARTEMIS_EXPLORER_CACHING", "").lower() - if env_cache in ["true", "false"]: - enable_caching = env_cache == "true" - else: - enable_caching = getattr(settings, "EXPLORER_CACHING", True) - - cached_content = None - if enable_caching: - try: - logger.info("Creating cache resource for Explorer...") - cached_content = client.caches.create( - model=model_name, - config=types.CreateCachedContentConfig( - contents=[contents[0]], - system_instruction=prompt_template, - tools=[types.Tool(function_declarations=self.get_exposed_tools())], - tool_config=types.ToolConfig( - function_calling_config=types.FunctionCallingConfig( - mode=types.FunctionCallingConfigMode.ANY - ), - include_server_side_tool_invocations=True, - ), - ttl="300s", - ), - ) - logger.info(f"Cache resource created successfully: {cached_content.name}") - except Exception as cache_err: - logger.error(f"Failed to create cache resource: {cache_err}") - cached_content = None - - # 6. Execution Loop (Native SDK Tools Dispatching) - iterations = 0 - agent_outcome = "" - self.turn_latencies = [] - self.turn_cached_tokens = [] - self.trace_history = [] - - while iterations < max_iterations: - iterations += 1 - logger.info(f"Iteration {iterations}: Invoking Native Gemini SDK for Explorer...") - - self._prune_historical_images(contents, keep_last=1) - - is_final_turn = iterations == max_iterations - if is_final_turn: - warning_msg = ( - "\n[WARNING] This is your final iteration. You MUST" - " call 'submit_answer' to submit your final result. No" - " other tools are available." - ) - contents.append( - types.Content( - role="user", - parts=[types.Part.from_text(text=warning_msg)], - ) - ) - - start_turn = time.perf_counter() - with TraceSpan(name="gemini_explorer_call", ctx=ctx) as span: - if cached_content and not is_final_turn: - generate_config = types.GenerateContentConfig( - cached_content=cached_content.name, - temperature=temperature, - safety_settings=SAFETY_SETTINGS_BLOCK_NONE, - **( - { - "thinking_config": types.ThinkingConfig( - thinking_level=thinking_level - ) - } - if thinking_level - else {} - ), - ) - else: - generate_config = types.GenerateContentConfig( - system_instruction=prompt_template, - temperature=temperature, - safety_settings=SAFETY_SETTINGS_BLOCK_NONE, - tools=[ - types.Tool( - function_declarations=self.get_exposed_tools( - only_submit=is_final_turn - ) - ) - ], - tool_config=types.ToolConfig( - function_calling_config=types.FunctionCallingConfig( - mode=types.FunctionCallingConfigMode.ANY - ), - include_server_side_tool_invocations=True, - ), - **( - { - "thinking_config": types.ThinkingConfig( - thinking_level=thinking_level - ) - } - if thinking_level - else {} - ), - ) - - response = await _generate_content_with_reliability( - lambda: asyncio.wait_for( - client.aio.models.generate_content( - model=model_name, - contents=contents, - config=generate_config, - ), - timeout=180, - ), - ) - end_turn = time.perf_counter() - turn_dur = end_turn - start_turn - self.turn_latencies.append(turn_dur) - - cached_tokens = 0 - if getattr(response, "usage_metadata", None) and getattr( - response.usage_metadata, - "cached_content_token_count", - None, - ): - cached_tokens = response.usage_metadata.cached_content_token_count - self.turn_cached_tokens.append(cached_tokens) - span.result = ( - f"Function calls: {len(response.function_calls)}" - if response.function_calls - else "Final answer" - ) - # Extract and log model thoughts and text if present - thinking_parts = [] - text_parts = [] - candidates = getattr(response, "candidates", None) - if not candidates and isinstance(response, dict): - candidates = response.get("candidates") - - if candidates and len(candidates) > 0: - candidate = candidates[0] - content = getattr(candidate, "content", None) - if not content and isinstance(candidate, dict): - content = candidate.get("content") - - parts = getattr(content, "parts", None) - if not parts and isinstance(content, dict): - parts = content.get("parts") - - if parts: - for part in parts: - is_thought = False - part_text = None - if isinstance(part, dict): - is_thought = part.get("thought", False) - part_text = part.get("text", None) - else: - is_thought = getattr(part, "thought", False) - part_text = getattr(part, "text", None) - - if is_thought and part_text: - thinking_parts.append(part_text) - elif part_text: - text_parts.append(part_text) - if thinking_parts: - span.payload["explorer_thought"] = "\n".join(thinking_parts) - if text_parts: - span.payload["explorer_text"] = "\n".join(text_parts) - - turn_record = { - "iteration": iterations, - "thoughts": ("\n".join(thinking_parts) if thinking_parts else ""), - "tool_calls": [], - } - - function_calls = response.function_calls - if not function_calls: - logger.warning( - f"Explorer iteration {iterations}: Model hallucinated" - " plain text. Forcing retry." - ) - turn_record["tool_calls"].append( - { - "name": "hallucinated_plain_text", - "args": {"text": "\n".join(text_parts) if text_parts else ""}, - "response": { - "error": ( - "Forced retry because model failed to call" - " submit_answer or any other tool" - ) - }, - } - ) - self.trace_history.append(turn_record) - contents.append(response.candidates[0].content) - contents.append( - types.Content( - role="user", - parts=[ - types.Part.from_text( - text=( - "You have not called 'submit_answer'" - " to submit your final answer yet. If" - " you are not certain about the final" - " answer, you can continue exploring" - f" {max_iterations - iterations} more" - " time(s)." - ) - ) - ], - ) - ) - continue - - # Record model's function call request in history - if response.candidates and response.candidates[0].content: - contents.append(response.candidates[0].content) - else: - contents.append( - types.Content( - role="model", - parts=[types.Part(function_call=fc) for fc in function_calls], - ) - ) - - # Execute function calls - tool_response_parts = [] - - # 6.1 Intercept and validate submit_answer to manage task lifecycle - submit_call = next( - ( - fc - for fc in function_calls - if (fc.name.split(":")[-1] if ":" in fc.name else fc.name) - == "submit_answer" - ), - None, - ) - if submit_call: - args = submit_call.args or {} - candidates = args.get("candidates", []) - errors = [] - - if len(function_calls) > 1: - errors.append( - "The tool 'submit_answer' MUST be called" - " individually in your final turn without any other" - " parallel tool calls. Please remove other tool" - " calls and re-submit." - ) - - fallback_message = args.get("fallback_message", "") - if not candidates and not fallback_message: - errors.append( - "The candidates list is empty. You must provide at" - " least one candidate matching the query, or" - " explain observations via fallback_message." - ) - - for i, cand in enumerate(candidates): - label = cand.get("label") - coords = cand.get("coords") - if not label: - errors.append(f"Candidate at index {i} is missing a 'label'.") - if not coords or not isinstance(coords, list) or len(coords) != 2: - errors.append( - f"Candidate '{label or i}' must have 'coords'" - " as an array of exactly 2 integers `[nx," - " ny]`." - ) - else: - try: - nx, ny = int(coords[0]), int(coords[1]) - if not (0 <= nx <= 1000) or not (0 <= ny <= 1000): - errors.append( - f"Candidate '{label or i}' coordinates" - f" `[{nx}, {ny}]` must strictly be in" - " the `[0-1000]` normalized scale" - " range inclusive." - ) - except (ValueError, TypeError, IndexError): - errors.append( - f"Candidate '{label or i}' coordinates are invalid integers." - ) - - if errors: - logger.warning(f"Explorer submit_answer validation failed: {errors}") - turn_record["tool_calls"].append( - { - "name": "submit_answer", - "args": args, - "response": {"error": ("Validation Failed:\n" + "\n".join(errors))}, - } - ) - tool_response_parts.append( - types.Part.from_function_response( - name="submit_answer", - response={ - "error": ( - "Validation Failed:\n" - + "\n".join(errors) - + "\nPlease correct these formatting" - " errors and re-submit using" - " submit_answer." - ) - }, - ) - ) - # Allow ReAct loop to continue so LLM can self-correct and re-submit - else: - logger.info("Explorer submit_answer validation passed successfully!") - agent_outcome = json.dumps(args, ensure_ascii=False) - turn_record["tool_calls"].append( - { - "name": "submit_answer", - "args": args, - "response": {"result": "success"}, - } - ) - self.trace_history.append(turn_record) - break - - for fc in function_calls: - name = fc.name.split(":")[-1] if ":" in fc.name else fc.name - if name == "submit_answer": - continue - args = fc.args or {} - tool_call_trace = { - "name": name, - "args": args, - } - if name in self.denylisted_tools: - logger.warning( - "Explorer attempted to call denylisted tool" - f" '{name}'. Blocking execution." - ) - tool_call_trace["response"] = { - "error": (f"Tool '{name}' is denylisted and unavailable.") - } - turn_record["tool_calls"].append(tool_call_trace) - tool_response_parts.append( - types.Part.from_function_response( - name=fc.name, - response={ - "error": (f"Tool '{name}' is denylisted and unavailable.") - }, - ) - ) - continue - - logger.info(f"Explorer executing tool '{name}' sequentially...") - - try: - if name == "ask_perception_tool": - res = await self.exec_ask_perception_tool( - search_query=args.get("search_query"), - nx=args.get("nx"), - ny=args.get("ny"), - detect_queries=args.get("detect_queries"), - ) - tool_call_trace["response"] = res - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={ - "text": res.get("text"), - }, - ) - ) - if res.get("image_paths"): - for img_p in res["image_paths"]: - try: - tool_response_parts.append(get_image_part(img_p)) - except Exception as e: - logger.warning( - f"Failed to load image response part for {img_p}: {e}" - ) - - elif name == "detect_objects": - res = await self.exec_detect_objects( - queries=args.get("queries"), - target_image_id=args.get("target_image_id", "img_0"), - ) - tool_call_trace["response"] = res - - if res.get("image_path"): - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={ - "result": res.get("text"), - }, - ) - ) - tool_response_parts.append(get_image_part(res["image_path"])) - else: - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={"result": res.get("text")}, - ) - ) - - elif name == "ask_image_processor": - res = await self.exec_ask_image_processor( - instruction=args.get("instruction"), - target_image_id=args.get("target_image_id", "img_0"), - ) - tool_call_trace["response"] = res - - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={"result": res.get("text")}, - ) - ) - if res.get("image_paths"): - for img_p in res["image_paths"]: - try: - tool_response_parts.append(get_image_part(img_p)) - except Exception as e: - logger.warning( - "Failed to upload image path" - f" {img_p} for ask_image_processor" - f" tool response: {e}" - ) - - elif name == "get_ocr_list": - res = await self.exec_get_ocr_list() - tool_call_trace["response"] = res - - if res.get("image_path"): - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={ - "result": res.get("text"), - }, - ) - ) - tool_response_parts.append(get_image_part(res["image_path"])) - else: - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={"result": res.get("text")}, - ) - ) - - elif name == "inspect_region": - res = await self.exec_inspect_region( - x_min=args.get("x_min"), - y_min=args.get("y_min"), - x_max=args.get("x_max"), - y_max=args.get("y_max"), - zoom_factor=args.get("zoom_factor"), - ) - tool_call_trace["response"] = res - - if res.get("image_path"): - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={ - "result": res.get("text"), - }, - ) - ) - tool_response_parts.append(get_image_part(res["image_path"])) - else: - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={"result": res.get("text")}, - ) - ) - else: - tool_call_trace["response"] = {"error": f"Tool {name} not found"} - tool_response_parts.append( - types.Part.from_function_response( - name=name, - response={"error": f"Tool {name} not found"}, - ) - ) - except Exception as e: - logger.error(f"Explorer tool {name} execution failed: {e}") - tool_call_trace["response"] = {"error": str(e)} - tool_response_parts.append( - types.Part.from_function_response(name=name, response={"error": str(e)}) - ) - finally: - turn_record["tool_calls"].append(tool_call_trace) - - if tool_response_parts: - # Native Gemini SDK allows 'tool' role to return mixed parts - contents.append(types.Content(role="user", parts=tool_response_parts)) - - self.trace_history.append(turn_record) - - if iterations >= max_iterations and not agent_outcome: - agent_outcome = ( - "Error: Explorer reached maximum iterations without a conclusive answer." - ) - - except Exception as e: - logger.error(f"Explorer execution loop failed: {e}") - - err_msg = str(e) - if "preempted" in err_msg.lower(): - clean_msg = ( - "Model request was preempted by the server due to high" - " demand. Please try again later." - ) - elif "overloaded" in err_msg.lower() or "503" in err_msg: - clean_msg = "Model service is currently overloaded. Please try again later." - elif "quota" in err_msg.lower() or "429" in err_msg: - clean_msg = "Model API quota exceeded or rate limited." - elif "thinkingconfig" in err_msg.lower(): - clean_msg = ( - "Model configuration error: ThinkingConfig is not" - f" supported for model {model_name}." - ) - else: - clean_msg = f"Explorer agent execution failed: {err_msg}" - - agent_outcome = json.dumps( - {"candidates": [], "fallback_message": clean_msg}, - ensure_ascii=False, - ) - finally: - if self.http_client: - try: - await self.http_client.aclose() - logger.info("Closed Explorer HTTP client.") - except Exception as close_err: - logger.warning(f"Failed to close Explorer HTTP client: {close_err}") - if cached_content: - try: - logger.info(f"Deleting cache resource: {cached_content.name}") - client.caches.delete(name=cached_content.name) - except Exception as cleanup_cache_err: - logger.warning(f"Failed to delete cache resource: {cleanup_cache_err}") - for file_ref in uploaded_files: - try: - client.files.delete(name=file_ref.name) - except Exception as cleanup_err: - logger.warning(f"Failed to delete uploaded file {file_ref.name}: {cleanup_err}") - logger.info(f"Explorer turn latencies: {self.turn_latencies}") - - return agent_outcome + return await self._run_native( + client=client, + query=query, + context_feedback=context_feedback, + minimal_list=minimal_list, + img_to_read=img_to_read, + enable_caching=enable_caching, + prompt_template=prompt_template, + model_name=model_name, + temperature=temperature, + thinking_level=thinking_level, + max_iterations=max_iterations, + ) diff --git a/artemis/agents/explorer/native_runner.py b/artemis/agents/explorer/native_runner.py new file mode 100644 index 0000000..60d32d9 --- /dev/null +++ b/artemis/agents/explorer/native_runner.py @@ -0,0 +1,796 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Native Gemini execution loop of ``Explorer.run``. + +Split out of ``artemis.agents.explorer.explorer``: the native-SDK reasoning +loop and its named phases (model invocation, submit_answer interception and +validation, tool dispatching, failure mapping, resource cleanup), packaged as +a mixin consumed by ``Explorer``. Patched collaborators (``settings``, +``logger``, ``_generate_content_with_reliability``) are resolved through the +facade module at call time; see ``artemis.agents.explorer._facade``. +""" + +import asyncio +import json +import os +import time + +from google.genai import types + +from artemis.agents.explorer._facade import facade +from artemis.constants import SAFETY_SETTINGS_BLOCK_NONE +from artemis.data_engine.trace import TraceSpan + + +class NativeRunnerMixin: + """Native Gemini SDK reasoning loop of :class:`Explorer`.""" + + def _make_image_part_getter(self, client, use_file_api: bool, uploaded_files: list): + """Builds the image-part loader used for screenshots and tool images.""" + + def get_image_part(file_path: str) -> types.Part: + if use_file_api: + file_ref = client.files.upload(file=file_path) + uploaded_files.append(file_ref) + return types.Part( + file_data=types.FileData( + file_uri=file_ref.uri, + mime_type=file_ref.mime_type or "image/jpeg", + ) + ) + else: + with open(file_path, "rb") as f: + img_bytes = f.read() + return types.Part.from_bytes(data=img_bytes, mime_type="image/jpeg") + + return get_image_part + + def _build_initial_contents( + self, query: str, context_feedback: str, minimal_list: str, initial_part: types.Part + ) -> list[types.Content]: + """Builds the initial user content with the operator request and screenshot.""" + return [ + types.Content( + role="user", + parts=[ + types.Part.from_text( + text=f"""Operator Request: +- Query: {query} +- Context Feedback: {context_feedback} + +Initial marked UI elements list (corresponding to numbers â‘ , â‘¡... in the image): +{minimal_list} +""" + ), + initial_part, + ], + ) + ] + + def _resolve_caching_flag(self, enable_caching): + """Resolves the effective caching flag from env or settings when unset.""" + _ex = facade() + if enable_caching is None: + env_cache = os.getenv("ARTEMIS_EXPLORER_CACHING", "").lower() + if env_cache in ["true", "false"]: + enable_caching = env_cache == "true" + else: + enable_caching = getattr(_ex.settings, "EXPLORER_CACHING", True) + return enable_caching + + def _create_cache_resource( + self, client, model_name: str, prompt_template: str, contents: list + ): + """Creates the explicit cache resource; returns None on failure.""" + _ex = facade() + cached_content = None + try: + _ex.logger.info("Creating cache resource for Explorer...") + cached_content = client.caches.create( + model=model_name, + config=types.CreateCachedContentConfig( + contents=[contents[0]], + system_instruction=prompt_template, + tools=[types.Tool(function_declarations=self.get_exposed_tools())], + tool_config=types.ToolConfig( + function_calling_config=types.FunctionCallingConfig( + mode=types.FunctionCallingConfigMode.ANY + ), + include_server_side_tool_invocations=True, + ), + ttl="300s", + ), + ) + _ex.logger.info(f"Cache resource created successfully: {cached_content.name}") + except Exception as cache_err: + _ex.logger.error(f"Failed to create cache resource: {cache_err}") + cached_content = None + return cached_content + + async def _run_native( + self, + *, + client, + query: str, + context_feedback: str, + minimal_list: str, + img_to_read: str, + enable_caching, + prompt_template: str, + model_name: str, + temperature, + thinking_level, + max_iterations: int, + ) -> str: + """Runs the native Gemini path: cache setup, reasoning loop, cleanup.""" + use_file_api = os.getenv("ARTEMIS_USE_FILE_API", "false").lower() == "true" + uploaded_files = [] + get_image_part = self._make_image_part_getter(client, use_file_api, uploaded_files) + + cached_content = None + try: + # Upload initial screenshot + initial_part = get_image_part(img_to_read) + contents = self._build_initial_contents( + query, context_feedback, minimal_list, initial_part + ) + + # Caching initialization + enable_caching = self._resolve_caching_flag(enable_caching) + + cached_content = None + if enable_caching: + cached_content = self._create_cache_resource( + client, model_name, prompt_template, contents + ) + + # 6. Execution Loop (Native SDK Tools Dispatching) + agent_outcome = await self._native_loop( + client=client, + contents=contents, + get_image_part=get_image_part, + cached_content=cached_content, + prompt_template=prompt_template, + model_name=model_name, + temperature=temperature, + thinking_level=thinking_level, + max_iterations=max_iterations, + ) + + except Exception as e: + agent_outcome = self._format_native_failure(e, model_name) + finally: + await self._cleanup_native_resources(client, cached_content, uploaded_files) + + return agent_outcome + + async def _native_loop( + self, + *, + client, + contents: list, + get_image_part, + cached_content, + prompt_template: str, + model_name: str, + temperature, + thinking_level, + max_iterations: int, + ) -> str: + """Main native reasoning loop: one model turn per iteration.""" + _ex = facade() + iterations = 0 + agent_outcome = "" + self.turn_latencies = [] + self.turn_cached_tokens = [] + self.trace_history = [] + + while iterations < max_iterations: + iterations += 1 + _ex.logger.info(f"Iteration {iterations}: Invoking Native Gemini SDK for Explorer...") + + self._prune_historical_images(contents, keep_last=1) + + is_final_turn = iterations == max_iterations + if is_final_turn: + warning_msg = ( + "\n[WARNING] This is your final iteration. You MUST" + " call 'submit_answer' to submit your final result. No" + " other tools are available." + ) + contents.append( + types.Content( + role="user", + parts=[types.Part.from_text(text=warning_msg)], + ) + ) + + response, thinking_parts, text_parts = await self._invoke_native_model( + client=client, + contents=contents, + cached_content=cached_content, + is_final_turn=is_final_turn, + prompt_template=prompt_template, + model_name=model_name, + temperature=temperature, + thinking_level=thinking_level, + ) + + turn_record = { + "iteration": iterations, + "thoughts": ("\n".join(thinking_parts) if thinking_parts else ""), + "tool_calls": [], + } + + function_calls = response.function_calls + if not function_calls: + self._handle_missing_tool_calls( + contents, response, turn_record, text_parts, max_iterations, iterations + ) + continue + + # Record model's function call request in history + if response.candidates and response.candidates[0].content: + contents.append(response.candidates[0].content) + else: + contents.append( + types.Content( + role="model", + parts=[types.Part(function_call=fc) for fc in function_calls], + ) + ) + + # Execute function calls + tool_response_parts = [] + + # 6.1 Intercept and validate submit_answer to manage task lifecycle + submit_outcome = self._process_submit_answer( + function_calls, turn_record, tool_response_parts + ) + if submit_outcome is not None: + agent_outcome = submit_outcome + break + + await self._dispatch_tool_calls( + function_calls, get_image_part, turn_record, tool_response_parts + ) + + if tool_response_parts: + # Native Gemini SDK allows 'tool' role to return mixed parts + contents.append(types.Content(role="user", parts=tool_response_parts)) + + self.trace_history.append(turn_record) + + if iterations >= max_iterations and not agent_outcome: + agent_outcome = ( + "Error: Explorer reached maximum iterations without a conclusive answer." + ) + + return agent_outcome + + def _build_generate_config( + self, + *, + cached_content, + is_final_turn: bool, + prompt_template: str, + temperature, + thinking_level, + ) -> types.GenerateContentConfig: + """Builds the per-turn generation config (cached vs full tool config).""" + thinking_kwargs = ( + {"thinking_config": types.ThinkingConfig(thinking_level=thinking_level)} + if thinking_level + else {} + ) + if cached_content and not is_final_turn: + return types.GenerateContentConfig( + cached_content=cached_content.name, + temperature=temperature, + safety_settings=SAFETY_SETTINGS_BLOCK_NONE, + **thinking_kwargs, + ) + else: + return types.GenerateContentConfig( + system_instruction=prompt_template, + temperature=temperature, + safety_settings=SAFETY_SETTINGS_BLOCK_NONE, + tools=[ + types.Tool( + function_declarations=self.get_exposed_tools(only_submit=is_final_turn) + ) + ], + tool_config=types.ToolConfig( + function_calling_config=types.FunctionCallingConfig( + mode=types.FunctionCallingConfigMode.ANY + ), + include_server_side_tool_invocations=True, + ), + **thinking_kwargs, + ) + + @staticmethod + def _extract_response_texts(response) -> tuple[list[str], list[str]]: + """Extracts model thoughts and plain text parts from a response.""" + thinking_parts = [] + text_parts = [] + candidates = getattr(response, "candidates", None) + if not candidates and isinstance(response, dict): + candidates = response.get("candidates") + + if candidates and len(candidates) > 0: + candidate = candidates[0] + content = getattr(candidate, "content", None) + if not content and isinstance(candidate, dict): + content = candidate.get("content") + + parts = getattr(content, "parts", None) + if not parts and isinstance(content, dict): + parts = content.get("parts") + + if parts: + for part in parts: + is_thought = False + part_text = None + if isinstance(part, dict): + is_thought = part.get("thought", False) + part_text = part.get("text", None) + else: + is_thought = getattr(part, "thought", False) + part_text = getattr(part, "text", None) + + if is_thought and part_text: + thinking_parts.append(part_text) + elif part_text: + text_parts.append(part_text) + return thinking_parts, text_parts + + async def _invoke_native_model( + self, + *, + client, + contents: list, + cached_content, + is_final_turn: bool, + prompt_template: str, + model_name: str, + temperature, + thinking_level, + ): + """Invokes the model for one turn under a trace span; returns response and texts.""" + _ex = facade() + ctx = self.ctx + start_turn = time.perf_counter() + with TraceSpan(name="gemini_explorer_call", ctx=ctx) as span: + generate_config = self._build_generate_config( + cached_content=cached_content, + is_final_turn=is_final_turn, + prompt_template=prompt_template, + temperature=temperature, + thinking_level=thinking_level, + ) + + response = await _ex._generate_content_with_reliability( + lambda: asyncio.wait_for( + client.aio.models.generate_content( + model=model_name, + contents=contents, + config=generate_config, + ), + timeout=180, + ), + ) + end_turn = time.perf_counter() + turn_dur = end_turn - start_turn + self.turn_latencies.append(turn_dur) + + cached_tokens = 0 + if getattr(response, "usage_metadata", None) and getattr( + response.usage_metadata, + "cached_content_token_count", + None, + ): + cached_tokens = response.usage_metadata.cached_content_token_count + self.turn_cached_tokens.append(cached_tokens) + span.result = ( + f"Function calls: {len(response.function_calls)}" + if response.function_calls + else "Final answer" + ) + # Extract and log model thoughts and text if present + thinking_parts, text_parts = self._extract_response_texts(response) + if thinking_parts: + span.payload["explorer_thought"] = "\n".join(thinking_parts) + if text_parts: + span.payload["explorer_text"] = "\n".join(text_parts) + + return response, thinking_parts, text_parts + + def _handle_missing_tool_calls( + self, + contents: list, + response, + turn_record: dict, + text_parts: list[str], + max_iterations: int, + iterations: int, + ) -> None: + """Handles a turn where the model produced plain text instead of tool calls.""" + _ex = facade() + _ex.logger.warning( + f"Explorer iteration {iterations}: Model hallucinated" + " plain text. Forcing retry." + ) + turn_record["tool_calls"].append( + { + "name": "hallucinated_plain_text", + "args": {"text": "\n".join(text_parts) if text_parts else ""}, + "response": { + "error": ( + "Forced retry because model failed to call" + " submit_answer or any other tool" + ) + }, + } + ) + self.trace_history.append(turn_record) + contents.append(response.candidates[0].content) + contents.append( + types.Content( + role="user", + parts=[ + types.Part.from_text( + text=( + "You have not called 'submit_answer'" + " to submit your final answer yet. If" + " you are not certain about the final" + " answer, you can continue exploring" + f" {max_iterations - iterations} more" + " time(s)." + ) + ) + ], + ) + ) + + @staticmethod + def _validate_submit_args(function_calls: list, args: dict, candidates: list) -> list[str]: + """Validates a submit_answer call; returns the list of validation errors.""" + errors = [] + + if len(function_calls) > 1: + errors.append( + "The tool 'submit_answer' MUST be called" + " individually in your final turn without any other" + " parallel tool calls. Please remove other tool" + " calls and re-submit." + ) + + fallback_message = args.get("fallback_message", "") + if not candidates and not fallback_message: + errors.append( + "The candidates list is empty. You must provide at" + " least one candidate matching the query, or" + " explain observations via fallback_message." + ) + + for i, cand in enumerate(candidates): + label = cand.get("label") + coords = cand.get("coords") + if not label: + errors.append(f"Candidate at index {i} is missing a 'label'.") + if not coords or not isinstance(coords, list) or len(coords) != 2: + errors.append( + f"Candidate '{label or i}' must have 'coords'" + " as an array of exactly 2 integers `[nx," + " ny]`." + ) + else: + try: + nx, ny = int(coords[0]), int(coords[1]) + if not (0 <= nx <= 1000) or not (0 <= ny <= 1000): + errors.append( + f"Candidate '{label or i}' coordinates" + f" `[{nx}, {ny}]` must strictly be in" + " the `[0-1000]` normalized scale" + " range inclusive." + ) + except (ValueError, TypeError, IndexError): + errors.append( + f"Candidate '{label or i}' coordinates are invalid integers." + ) + + return errors + + def _process_submit_answer( + self, function_calls: list, turn_record: dict, tool_response_parts: list + ) -> str | None: + """Intercepts submit_answer; returns the final outcome JSON when valid. + + On validation failure the error is fed back as a tool response and + None is returned so the ReAct loop can continue and self-correct. + """ + _ex = facade() + submit_call = next( + ( + fc + for fc in function_calls + if (fc.name.split(":")[-1] if ":" in fc.name else fc.name) + == "submit_answer" + ), + None, + ) + if not submit_call: + return None + + args = submit_call.args or {} + candidates = args.get("candidates", []) + errors = self._validate_submit_args(function_calls, args, candidates) + + if errors: + _ex.logger.warning(f"Explorer submit_answer validation failed: {errors}") + turn_record["tool_calls"].append( + { + "name": "submit_answer", + "args": args, + "response": {"error": ("Validation Failed:\n" + "\n".join(errors))}, + } + ) + tool_response_parts.append( + types.Part.from_function_response( + name="submit_answer", + response={ + "error": ( + "Validation Failed:\n" + + "\n".join(errors) + + "\nPlease correct these formatting" + " errors and re-submit using" + " submit_answer." + ) + }, + ) + ) + # Allow ReAct loop to continue so LLM can self-correct and re-submit + return None + else: + _ex.logger.info("Explorer submit_answer validation passed successfully!") + agent_outcome = json.dumps(args, ensure_ascii=False) + turn_record["tool_calls"].append( + { + "name": "submit_answer", + "args": args, + "response": {"result": "success"}, + } + ) + self.trace_history.append(turn_record) + return agent_outcome + + @staticmethod + def _append_tool_response_with_image( + name: str, res: dict, get_image_part, tool_response_parts: list + ) -> None: + """Appends a tool result part plus its annotated image when present.""" + if res.get("image_path"): + tool_response_parts.append( + types.Part.from_function_response( + name=name, + response={ + "result": res.get("text"), + }, + ) + ) + tool_response_parts.append(get_image_part(res["image_path"])) + else: + tool_response_parts.append( + types.Part.from_function_response( + name=name, + response={"result": res.get("text")}, + ) + ) + + async def _execute_native_tool( + self, + name: str, + args: dict, + get_image_part, + tool_call_trace: dict, + tool_response_parts: list, + ) -> None: + """Executes one non-submit tool call and appends its response parts.""" + _ex = facade() + if name == "ask_perception_tool": + res = await self.exec_ask_perception_tool( + search_query=args.get("search_query"), + nx=args.get("nx"), + ny=args.get("ny"), + detect_queries=args.get("detect_queries"), + ) + tool_call_trace["response"] = res + tool_response_parts.append( + types.Part.from_function_response( + name=name, + response={ + "text": res.get("text"), + }, + ) + ) + if res.get("image_paths"): + for img_p in res["image_paths"]: + try: + tool_response_parts.append(get_image_part(img_p)) + except Exception as e: + _ex.logger.warning( + f"Failed to load image response part for {img_p}: {e}" + ) + + elif name == "detect_objects": + res = await self.exec_detect_objects( + queries=args.get("queries"), + target_image_id=args.get("target_image_id", "img_0"), + ) + tool_call_trace["response"] = res + self._append_tool_response_with_image(name, res, get_image_part, tool_response_parts) + + elif name == "ask_image_processor": + res = await self.exec_ask_image_processor( + instruction=args.get("instruction"), + target_image_id=args.get("target_image_id", "img_0"), + ) + tool_call_trace["response"] = res + + tool_response_parts.append( + types.Part.from_function_response( + name=name, + response={"result": res.get("text")}, + ) + ) + if res.get("image_paths"): + for img_p in res["image_paths"]: + try: + tool_response_parts.append(get_image_part(img_p)) + except Exception as e: + _ex.logger.warning( + "Failed to upload image path" + f" {img_p} for ask_image_processor" + f" tool response: {e}" + ) + + elif name == "get_ocr_list": + res = await self.exec_get_ocr_list() + tool_call_trace["response"] = res + self._append_tool_response_with_image(name, res, get_image_part, tool_response_parts) + + elif name == "inspect_region": + res = await self.exec_inspect_region( + x_min=args.get("x_min"), + y_min=args.get("y_min"), + x_max=args.get("x_max"), + y_max=args.get("y_max"), + zoom_factor=args.get("zoom_factor"), + ) + tool_call_trace["response"] = res + self._append_tool_response_with_image(name, res, get_image_part, tool_response_parts) + else: + tool_call_trace["response"] = {"error": f"Tool {name} not found"} + tool_response_parts.append( + types.Part.from_function_response( + name=name, + response={"error": f"Tool {name} not found"}, + ) + ) + + async def _dispatch_tool_calls( + self, + function_calls: list, + get_image_part, + turn_record: dict, + tool_response_parts: list, + ) -> None: + """Dispatches all non-submit tool calls sequentially with tracing.""" + _ex = facade() + for fc in function_calls: + name = fc.name.split(":")[-1] if ":" in fc.name else fc.name + if name == "submit_answer": + continue + args = fc.args or {} + tool_call_trace = { + "name": name, + "args": args, + } + if name in self.denylisted_tools: + _ex.logger.warning( + "Explorer attempted to call denylisted tool" + f" '{name}'. Blocking execution." + ) + tool_call_trace["response"] = { + "error": (f"Tool '{name}' is denylisted and unavailable.") + } + turn_record["tool_calls"].append(tool_call_trace) + tool_response_parts.append( + types.Part.from_function_response( + name=fc.name, + response={ + "error": (f"Tool '{name}' is denylisted and unavailable.") + }, + ) + ) + continue + + _ex.logger.info(f"Explorer executing tool '{name}' sequentially...") + + try: + await self._execute_native_tool( + name, args, get_image_part, tool_call_trace, tool_response_parts + ) + except Exception as e: + _ex.logger.error(f"Explorer tool {name} execution failed: {e}") + tool_call_trace["response"] = {"error": str(e)} + tool_response_parts.append( + types.Part.from_function_response(name=name, response={"error": str(e)}) + ) + finally: + turn_record["tool_calls"].append(tool_call_trace) + + def _format_native_failure(self, e: Exception, model_name: str) -> str: + """Maps a loop failure to the user-facing fallback outcome JSON.""" + _ex = facade() + _ex.logger.error(f"Explorer execution loop failed: {e}") + + err_msg = str(e) + if "preempted" in err_msg.lower(): + clean_msg = ( + "Model request was preempted by the server due to high" + " demand. Please try again later." + ) + elif "overloaded" in err_msg.lower() or "503" in err_msg: + clean_msg = "Model service is currently overloaded. Please try again later." + elif "quota" in err_msg.lower() or "429" in err_msg: + clean_msg = "Model API quota exceeded or rate limited." + elif "thinkingconfig" in err_msg.lower(): + clean_msg = ( + "Model configuration error: ThinkingConfig is not" + f" supported for model {model_name}." + ) + else: + clean_msg = f"Explorer agent execution failed: {err_msg}" + + return json.dumps( + {"candidates": [], "fallback_message": clean_msg}, + ensure_ascii=False, + ) + + async def _cleanup_native_resources( + self, client, cached_content, uploaded_files: list + ) -> None: + """Closes the HTTP client and deletes cache/file resources.""" + _ex = facade() + if self.http_client: + try: + await self.http_client.aclose() + _ex.logger.info("Closed Explorer HTTP client.") + except Exception as close_err: + _ex.logger.warning(f"Failed to close Explorer HTTP client: {close_err}") + if cached_content: + try: + _ex.logger.info(f"Deleting cache resource: {cached_content.name}") + client.caches.delete(name=cached_content.name) + except Exception as cleanup_cache_err: + _ex.logger.warning(f"Failed to delete cache resource: {cleanup_cache_err}") + for file_ref in uploaded_files: + try: + client.files.delete(name=file_ref.name) + except Exception as cleanup_err: + _ex.logger.warning(f"Failed to delete uploaded file {file_ref.name}: {cleanup_err}") + _ex.logger.info(f"Explorer turn latencies: {self.turn_latencies}") diff --git a/artemis/agents/explorer/perception_tools.py b/artemis/agents/explorer/perception_tools.py new file mode 100644 index 0000000..f502017 --- /dev/null +++ b/artemis/agents/explorer/perception_tools.py @@ -0,0 +1,664 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Perception and vision tool implementations for the Explorer agent. + +Split out of ``artemis.agents.explorer.explorer``: the ``exec_*`` tool method +group plus its search helpers, packaged as a mixin consumed by ``Explorer``. +Patched collaborators (``settings``, ``draw_dots``, ``search_ui_func``, +``search_by_coordinates_func``, ``_run_object_detection``, ``StorageManager``, +``logger``) are resolved through the facade module at call time; see +``artemis.agents.explorer._facade``. +""" + +import asyncio +import glob +import json +from pathlib import Path +import re + +import cv2 + +from artemis.agents.explorer._facade import facade +from artemis.agents.image_processor.image_processor import ImageProcessor +from artemis.data_engine.trace import trace + + +def load_detector_templates() -> tuple[list, object]: + """Load object-detector prompt templates and the optional global timeout. + + Shared by ``exec_detect_objects`` and the flash-mode ``run`` path; the two + call sites historically carried an identical inline copy of this block. + """ + _ex = facade() + detector_prompt_path = ( + Path(__file__).parent.parent / "object_detector" / "object_detector.json" + ) + global_timeout = None + try: + with open(detector_prompt_path, encoding="utf-8") as f: + detector_config = json.load(f) + detector_templates = detector_config.get("templates", []) + detector_instructions = detector_config.get("instructions", "") + global_timeout = detector_config.get("global_timeout", None) + except Exception as e: + _ex.logger.warning(f"Failed to load detector prompt config: {e}") + detector_templates = [ + "Point to the following objects in the provided image: {labels_str}." + ] + detector_instructions = "" + + templates = [f"{t}\n\n{detector_instructions}" for t in detector_templates] + return templates, global_timeout + + +class PerceptionToolsMixin: + """Perception/vision tool method group of :class:`Explorer`.""" + + async def _search_ui_helper(self, query: str, prefix: str = "S", color: str = "red") -> dict: + _ex = facade() + if not self.image_name: + return { + "text": ( + "Error: UI data for the current screen is not found in Data" + " Engine. It might not be synced yet." + ), + "image_path": None, + } + + # 1. Try with high precision threshold + raw_res = _ex.search_ui_func(self.image_name, query, 0.7) + if "error" in raw_res: + return {"text": raw_res["error"], "image_path": None} + + matches = raw_res.get("matches", []) + + # 2. If fewer than 3 matches, try with a lower threshold and merge + if len(matches) < 3: + raw_res_low = _ex.search_ui_func(self.image_name, query, 0.4) + if "error" not in raw_res_low: + low_matches = raw_res_low.get("matches", []) + existing_keys = { + ( + m.get("matched_text"), + tuple(m.get("bounds")) if m.get("bounds") else None, + ) + for m in matches + } + for m in low_matches: + key = ( + m.get("matched_text"), + tuple(m.get("bounds")) if m.get("bounds") else None, + ) + if key not in existing_keys: + matches.append(m) + existing_keys.add(key) + + if not matches: + return {"text": "No matches found.", "image_path": None} + + points = [] + labels = [] + text_output = [] + + for m in matches: + m_type = m.get("type", "xml") + m_prefix = "O" if m_type == "ocr" else prefix + label_id = f"{m_prefix}{self.global_label_idx}" + self.global_label_idx += 1 + bounds = m.get("bounds") + matched_text = m.get("matched_text", "") + + if bounds and len(bounds) == 4: + left, top, right, bottom = bounds + cx_pixel = (left + right) // 2 + cy_pixel = (top + bottom) // 2 + + points.append([cx_pixel, cy_pixel]) + labels.append(label_id) + + cx_norm = int(max(0, min(1000, cx_pixel * 1000 / self.width))) + cy_norm = int(max(0, min(1000, cy_pixel * 1000 / self.height))) + + text_output.append(f"[{label_id}] '{matched_text}' at [{cx_norm},{cy_norm}]") + else: + text_output.append(f"[{label_id}] '{matched_text}' at unknown") + + if not points: + return {"text": " | ".join(text_output), "image_path": None} + + base_dir = _ex.settings.TRACES_PATH + images_dir = base_dir / "images" + search_ui_dir = images_dir / "search_ui" + search_ui_dir.mkdir(parents=True, exist_ok=True) + + existing_files = glob.glob(str(search_ui_dir / f"{self.image_name}_*.jpg")) + max_seq = 0 + for f in existing_files: + match = re.search(r"_(\d+)\.jpg$", f) + if match: + max_seq = max(max_seq, int(match.group(1))) + seq = max_seq + 1 + output_path = search_ui_dir / f"{self.image_name}_{seq}.jpg" + + try: + _ex.draw_dots( + self.screenshot_path, + points, + labels, + str(output_path), + color=color, + ) + image_path = str(output_path) + except Exception as e: + _ex.logger.warning(f"Failed to draw dots: {e}") + image_path = None + + return {"text": " | ".join(text_output), "image_path": image_path} + + async def _search_by_coords_helper( + self, nx: int, ny: int, prefix: str = "X", color: str = "blue" + ) -> dict: + _ex = facade() + if not self.image_name: + return { + "text": ( + "Error: UI data for the current screen is not found in Data" + " Engine. It might not be synced yet." + ), + "image_path": None, + } + + x = int(max(0, min(self.width, nx * self.width / 1000))) + y = int(max(0, min(self.height, ny * self.height / 1000))) + raw_res = _ex.search_by_coordinates_func(self.image_name, x, y) + + label_id = f"{prefix}{self.global_label_idx}" + self.global_label_idx += 1 + + # Draw a dot at the query coordinates + base_dir = _ex.settings.TRACES_PATH + images_dir = base_dir / "images" + coords_audit_dir = images_dir / "coords_audit" + coords_audit_dir.mkdir(parents=True, exist_ok=True) + + existing_files = glob.glob(str(coords_audit_dir / f"{self.image_name}_*.jpg")) + max_seq = 0 + for f in existing_files: + match = re.search(r"_(\d+)\.jpg$", f) + if match: + max_seq = max(max_seq, int(match.group(1))) + seq = max_seq + 1 + output_path = coords_audit_dir / f"{self.image_name}_{seq}.jpg" + + try: + _ex.draw_dots( + self.screenshot_path, + [[x, y]], + [label_id], + str(output_path), + color=color, + ) + image_path = str(output_path) + except Exception as e: + _ex.logger.warning(f"Failed to draw blue dots: {e}") + image_path = None + + raw_res_str = str(raw_res).replace("\n", " | ").replace(" | | ", " | ") + return { + "text": (f"Matched element at [{nx},{ny}]: [{label_id}] | {raw_res_str}"), + "image_path": image_path, + } + + @trace(type="tool", name="detect_objects") + async def exec_detect_objects( + self, + queries: list[str], + target_image_id: str = "img_0", + prefix: str = "D", + color: str = "red", + ) -> dict: + _ex = facade() + try: + target_info = self.image_pool.get(target_image_id) + if not target_info: + return { + "text": (f"Error: {target_image_id} not found in Image Pool."), + "image_path": None, + } + + target_path = target_info["path"] + transform = target_info["transform"] + + # Load object detector templates + templates, global_timeout = load_detector_templates() + + result = await _ex._run_object_detection( + self.ctx, + target_path, + queries, + templates, + global_timeout=global_timeout, + ) + try: + detected_items = result.get("detected", []) + failed_queries = result.get("failed", []) + + if not detected_items: + text_output = ["No objects detected."] + if failed_queries: + text_output.append( + "\n[NOTE: Failed to detect the following queries:" + f" {', '.join(failed_queries)}]" + ) + return { + "text": "".join(text_output), + "image_path": str(target_path), + } + + base_dir = _ex.settings.TRACES_PATH + images_dir = base_dir / "images" + object_detect_dir = images_dir / "object_detect" + object_detect_dir.mkdir(parents=True, exist_ok=True) + + image_name_safe = self.image_name or "temp_image" + existing_files = glob.glob(str(object_detect_dir / f"{image_name_safe}_*.jpg")) + max_seq = 0 + for f in existing_files: + match = re.search(r"_(\d+)\.jpg$", f) + if match: + max_seq = max(max_seq, int(match.group(1))) + seq = max_seq + 1 + output_path = object_detect_dir / f"{image_name_safe}_{seq}.jpg" + + points = [] + d_labels = [] + text_output = [] + + if target_image_id != "img_0": + text_output.append( + "[NOTE: Coordinates are automatically mapped back to" + " the original 1080x2400 screen space from" + f" {target_image_id}]" + ) + + for item in detected_items: + label_id = f"{prefix}{self.global_label_idx}" + self.global_label_idx += 1 + pos = item.get("point") + if pos and isinstance(pos, list) and len(pos) == 2: + x_norm, y_norm = pos + + t_img = cv2.imread(target_path) + t_h, t_w = t_img.shape[:2] + + # Pixels in target image + x_target_pixel = (x_norm / 1000) * t_w + y_target_pixel = (y_norm / 1000) * t_h + + # Pixels in original image + x_orig_pixel = (x_target_pixel / transform["scale_x"]) + transform[ + "offset_x" + ] + y_orig_pixel = (y_target_pixel / transform["scale_y"]) + transform[ + "offset_y" + ] + + points.append([int(x_orig_pixel), int(y_orig_pixel)]) + d_labels.append(label_id) + + # Original image norm + x_orig_norm = int(max(0, min(1000, x_orig_pixel * 1000 / self.width))) + y_orig_norm = int(max(0, min(1000, y_orig_pixel * 1000 / self.height))) + + text_output.append( + f'[{label_id}] "{item.get("label")}" at [{x_orig_norm},{y_orig_norm}]' + ) + + _ex.draw_dots( + self.screenshot_path, + points, + d_labels, + str(output_path), + color=color, + ) + + if failed_queries: + text_output.append( + " | [NOTE: Failed to detect the following queries:" + f" {', '.join(failed_queries)}]" + ) + + return { + "text": " | ".join(text_output), + "image_path": str(output_path), + } + + except Exception as e: + _ex.logger.warning(f"Failed to process detection output for annotation: {e}") + return {"text": str(result), "image_path": None} + except Exception as e: + return { + "text": f"Error: Object detection failed: {e}", + "image_path": None, + } + + @trace(type="tool", name="ask_perception_tool") + async def exec_ask_perception_tool( + self, search_query: str, nx: int, ny: int, detect_queries: list[str] + ) -> dict: + if nx is None or ny is None or not (0 <= nx <= 1000) or not (0 <= ny <= 1000): + coords_invalid = True + else: + coords_invalid = False + + # Run tasks concurrently in parallel! + tasks = [ + self._search_ui_helper(search_query, prefix="X", color="green"), + self._search_by_coords_helper(nx, ny, prefix="X", color="blue") + if not coords_invalid + else asyncio.sleep( + 0, + result={ + "text": "Error: Invalid coordinates format", + "image_path": None, + }, + ), + self.exec_detect_objects( + detect_queries, target_image_id="img_0", prefix="D", color="red" + ), + ] + + results = await asyncio.gather(*tasks) + + # Consolidate text results + text_parts = [] + + xml_text = results[0].get("text", "No matches found.") + if xml_text and xml_text != "No matches found.": + text_parts.append(f"Text Search Results are: {xml_text}") + + coord_text = results[1].get("text", "No elements found.") + if coord_text and coord_text != "No elements found.": + text_parts.append(f"Coordinate Search Results are: {coord_text}") + + obj_text = results[2].get("text", "No objects detected.") + if obj_text: + if obj_text.startswith("No objects detected."): + note_part = obj_text.replace("No objects detected.", "").strip() + if note_part: + text_parts.append(f"Object Detection Results are: {note_part}") + else: + text_parts.append(f"Object Detection Results are: {obj_text}") + + if not text_parts: + unified_text = "No matches or objects found across any perception method." + else: + unified_text = ". ".join(text_parts) + "." + + # Collect image paths + image_paths = [] + for r in results: + if isinstance(r, dict) and r.get("image_path"): + image_paths.append(r["image_path"]) + + return { + "text": unified_text, + "image_paths": image_paths, + } + + @trace(type="tool", name="ask_image_processor") + async def exec_ask_image_processor( + self, instruction: str, target_image_id: str = "img_0" + ) -> dict: + coder = ImageProcessor(self.ctx) + target_info = self.image_pool.get(target_image_id) + if not target_info: + return { + "text": f"Error: {target_image_id} not found in Image Pool.", + "image_path": None, + } + + target_path = target_info["path"] + result = await coder.run(instruction, target_path) + + if not isinstance(result, dict) or "error" in result: + err_msg = result.get("error") if isinstance(result, dict) else "Unknown error" + return { + "text": f"Image Processor failed: {err_msg}", + "image_paths": [], + } + + outputs = result.get("outputs", []) + summary = result.get("summary", "") + + parent_transform = target_info["transform"] + registered_ids = [] + annotations_sections = [] + image_paths = [] + + for entry in outputs: + new_id = f"img_{self.next_img_id}" + self.next_img_id += 1 + + new_path = entry["path"] + image_paths.append(new_path) + transform = entry["transform"] + + final_scale_x = parent_transform["scale_x"] * transform["scale_x"] + final_scale_y = parent_transform["scale_y"] * transform["scale_y"] + final_offset_x = ( + transform["offset_x"] / parent_transform["scale_x"] + ) + parent_transform["offset_x"] + final_offset_y = ( + transform["offset_y"] / parent_transform["scale_y"] + ) + parent_transform["offset_y"] + + self.image_pool[new_id] = { + "path": new_path, + "transform": { + "offset_x": final_offset_x, + "offset_y": final_offset_y, + "scale_x": final_scale_x, + "scale_y": final_scale_y, + }, + "description": (f"Generated by coder from {target_image_id}: {summary}"), + } + registered_ids.append(new_id) + + # Parse and translate annotations + annotations = entry.get("annotations", {}) + if annotations: + lines = [f"{new_id}:"] + for label, coord in sorted(annotations.items(), key=lambda item: item[0]): + px, py = coord + sx = int(round((px / final_scale_x) + final_offset_x)) + sy = int(round((py / final_scale_y) + final_offset_y)) + lines.append(f" - [{label}]: [{sx}, {sy}]") + annotations_sections.append("\n".join(lines)) + + if registered_ids: + ids_str = ", ".join(registered_ids) + text = ( + "Image Processor completed successfully. New image(s) saved as" + f" {ids_str} in Image Pool. Summary: {summary}" + ) + if annotations_sections: + text += "\n\nAnnotations (wrt original screenshot):\n" + "\n".join( + annotations_sections + ) + return { + "text": text, + "image_paths": image_paths, + } + else: + return { + "text": f"ImageProcessor error: {err_msg}", + "image_paths": [], + } + + output_path = result.get("output_path") + trans = result.get("transform") + new_image_id = f"img_{len(self.image_pool)}" + if output_path and trans: + self.image_pool[new_image_id] = { + "path": output_path, + "transform": trans, + } + + text_msg = ( + f"Image processed successfully. New image saved as" + f" '{new_image_id}'. You can now use '{new_image_id}' as" + " 'target_image_id' in subsequent tool calls." + ) + return { + "text": text_msg, + "image_path": output_path, + "image_id": new_image_id, + } + + @trace(type="tool", name="get_ocr_list") + async def exec_get_ocr_list(self, prefix: str = "O", color: str = "green") -> dict: + _ex = facade() + if not self.image_name: + return { + "text": ("Error: UI data for the current screen is not found in Data Engine."), + "image_path": None, + } + + try: + db_path = _ex.settings.DATA_ENGINE_DB_PATH + base_dir = _ex.settings.TRACES_PATH + storage = _ex.StorageManager(db_path=str(db_path), base_dir=base_dir) + + record = storage.get_ui_record(self.image_name) + if not record or not record.ocr_result: + return { + "text": "No text elements detected on the screen.", + "image_path": None, + } + + points = [] + labels = [] + text_output = [] + + for item in record.ocr_result: + text = item.get("text", "").strip() + bounds = item.get("bounds", []) + if text and len(bounds) == 4: + left, top, right, bottom = bounds + cx = (left + right) // 2 + cy = (top + bottom) // 2 + if 0 <= cx <= self.width and 0 <= cy <= self.height: + label_id = f"{prefix}{self.global_label_idx}" + self.global_label_idx += 1 + points.append([cx, cy]) + labels.append(label_id) + + cx_norm = int(max(0, min(1000, cx * 1000 / self.width))) + cy_norm = int(max(0, min(1000, cy * 1000 / self.height))) + + text_output.append(f"[{label_id}] '{text}' coords: [{cx_norm},{cy_norm}]") + + if not points: + return { + "text": "No valid OCR results found.", + "image_path": None, + } + + images_dir = base_dir / "images" + ocr_dir = images_dir / "ocr" + ocr_dir.mkdir(parents=True, exist_ok=True) + + output_path = ocr_dir / f"{self.image_name}_{self.global_label_idx}.jpg" + _ex.draw_dots(self.screenshot_path, points, labels, str(output_path), color=color) + + return { + "text": "\n".join(text_output), + "image_path": str(output_path), + } + + except Exception as e: + return { + "text": f"Error retrieving or processing OCR results: {e}", + "image_path": None, + } + + @trace(type="tool", name="inspect_region") + async def exec_inspect_region( + self, + x_min: int, + y_min: int, + x_max: int, + y_max: int, + zoom_factor: float = 2.0, + ) -> dict: + _ex = facade() + try: + px_start = int(max(0, min(self.width, x_min * self.width / 1000.0))) + py_start = int(max(0, min(self.height, y_min * self.height / 1000.0))) + px_end = int(max(0, min(self.width, x_max * self.width / 1000.0))) + py_end = int(max(0, min(self.height, y_max * self.height / 1000.0))) + + if px_end <= px_start or py_end <= py_start: + return { + "text": ( + f"Error: Invalid crop coordinates. x_min={x_min}," + f" x_max={x_max}, y_min={y_min}, y_max={y_max}." + " Bounding box dimensions must be strictly positive." + ), + "image_path": None, + } + + img = cv2.imread(self.screenshot_path) + if img is None: + return { + "text": ( + f"Error: Failed to read original screenshot at {self.screenshot_path}" + ), + "image_path": None, + } + + cropped = img[py_start:py_end, px_start:px_end] + + if zoom_factor is None: + zoom_factor = 2.0 + zoom_factor = max(1.0, min(4.0, float(zoom_factor))) + + c_h, c_w = cropped.shape[:2] + if c_h > 0 and c_w > 0: + new_w = int(round(c_w * zoom_factor)) + new_h = int(round(c_h * zoom_factor)) + if new_w > 0 and new_h > 0: + cropped = cv2.resize(cropped, (new_w, new_h)) + + base_dir = _ex.settings.TRACES_PATH + images_dir = base_dir / "images" + inspect_region_dir = images_dir / "inspect_region" + inspect_region_dir.mkdir(parents=True, exist_ok=True) + + image_name_safe = self.image_name or "temp_image" + output_path = inspect_region_dir / f"{image_name_safe}_{self.global_label_idx}.jpg" + cv2.imwrite(str(output_path), cropped) + + return { + "text": (f"Inspected region coordinates: [{x_min}, {y_min}] to [{x_max}, {y_max}]"), + "image_path": str(output_path), + } + + except Exception as e: + return { + "text": f"Error: Region inspection failed: {e}", + "image_path": None, + } diff --git a/artemis/agents/explorer/run_setup.py b/artemis/agents/explorer/run_setup.py new file mode 100644 index 0000000..c85d626 --- /dev/null +++ b/artemis/agents/explorer/run_setup.py @@ -0,0 +1,384 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Setup phases of ``Explorer.run``. + +Split out of ``artemis.agents.explorer.explorer``: the flash-mode fast path +and the named preparation phases executed before the reasoning loop (image +record resolution, version limits, screen dimensions, image pool, GenAI +client, model parameters, prompt construction, and initial screenshot +annotation), packaged as a mixin consumed by ``Explorer``. Patched +collaborators (``settings``, ``StorageManager``, ``genai``, +``is_ocr_configured``, ``perform_ocr``, ``draw_dots``, +``_run_object_detection``, ``logger``) are resolved through the facade module +at call time; see ``artemis.agents.explorer._facade``. +""" + +import base64 +import glob +import hashlib +import json +import os +from pathlib import Path +import re + +from artemis.agents.explorer._facade import facade +from artemis.agents.explorer.constants import EXPLORE_DESCRIPTIONS +from artemis.agents.explorer.perception_tools import load_detector_templates +from artemis.graph.state import State +from artemis.utils.ocr_xml_fusion import fuse_ocr_with_xml +from artemis.utils.visualization import format_minimal_list_with_points + + +class RunSetupMixin: + """Flash mode and pre-loop setup phases of :class:`Explorer`.""" + + async def _run_flash(self, query: str, screenshot_path: str) -> str: + """Flash mode: single-shot object detection without a reasoning loop.""" + _ex = facade() + templates, global_timeout = load_detector_templates() + + self.screenshot_path = screenshot_path + self.image_name = None + + queries = [q.strip() for q in query.split("|") if q.strip()] + + try: + result = await _ex._run_object_detection( + self.ctx, + screenshot_path, + queries, + templates, + global_timeout=global_timeout, + ) + detected_items = result.get("detected", []) + candidates = [] + for idx, item in enumerate(detected_items): + pos = item.get("point") + if pos and isinstance(pos, list) and len(pos) == 2: + candidates.append( + { + "label": f"D{idx + 1}", + "coords": pos, + "description": item.get("label", query), + } + ) + fallback_message = "" if candidates else f"Failed to detect: {query}" + return json.dumps( + { + "candidates": candidates, + "fallback_message": fallback_message, + }, + ensure_ascii=False, + ) + except Exception as e: + _ex.logger.error(f"Flash mode object detection failed: {e}") + return json.dumps( + { + "candidates": [], + "fallback_message": f"Flash mode detection error: {e}", + }, + ensure_ascii=False, + ) + + def _resolve_image_record(self, screenshot_path: str) -> tuple: + """Computes the screenshot hash and looks it up in the Data Engine DB.""" + _ex = facade() + image_name = None + record = None + try: + sha256_hash = hashlib.sha256() + with open(screenshot_path, "rb") as f: + for byte_block in iter(lambda: f.read(4096), b""): + sha256_hash.update(byte_block) + computed_hash = sha256_hash.hexdigest() + _ex.logger.info(f"Computed screenshot hash: {computed_hash}") + + # Check if it exists in DB + db_path = _ex.settings.DATA_ENGINE_DB_PATH + base_dir = _ex.settings.TRACES_PATH + storage = _ex.StorageManager(db_path, base_dir) + record = storage.get_image(computed_hash) + + if record: + image_name = computed_hash + else: + _ex.logger.warning( + f"Image hash {computed_hash} not found in Data Engine DB." + " Data Engine might not be synced yet." + ) + + except Exception as e: + _ex.logger.warning(f"Failed to compute hash or check DB: {e}") + + return image_name, record + + def _resolve_version_limits(self, version: str) -> tuple[str, int]: + """Resolves mode label, iteration budget, and the effective tool denylist.""" + _ex = facade() + mode = version.capitalize() + version_denylist = set() + if version == "ultra": + max_iterations = 8 + elif version == "pro": + max_iterations = 3 + version_denylist = { + "ask_image_processor", + "get_ocr_list", + "inspect_region", + "detect_objects", + } + else: + max_iterations = 3 + version_denylist = { + "ask_image_processor", + "get_ocr_list", + "inspect_region", + } + + if not _ex.is_ocr_configured(): + version_denylist.add("get_ocr_list") + + try: + denylisted_config = ( + self.ctx.agent_config.denylisted_tools.get("explorer", []) + if self.ctx.agent_config + else [] + ) + self.denylisted_tools = version_denylist.union(set(denylisted_config)) + except (TypeError, AttributeError): + self.denylisted_tools = version_denylist + + return mode, max_iterations + + def _resolve_screen_dimensions(self, state: State) -> None: + """Resolves device screen dimensions from operator data or the device.""" + ctx = self.ctx + self.width = 1080 + self.height = 2400 + operator_raw_data = getattr(state, "operator_raw_data", {}) or {} + w_raw = operator_raw_data.get("width") + h_raw = operator_raw_data.get("height") + if isinstance(w_raw, int) and isinstance(h_raw, int): + self.width = w_raw + self.height = h_raw + else: + if ctx.device and getattr(ctx.device, "device_width", None): + self.width = ctx.device.device_width + if ctx.device and getattr(ctx.device, "device_height", None): + self.height = ctx.device.device_height + + def _init_image_pool(self, screenshot_path: str) -> None: + """Seeds the Image Pool with the original screenshot as img_0.""" + self.image_pool = { + "img_0": { + "path": screenshot_path, + "transform": { + "offset_x": 0.0, + "offset_y": 0.0, + "scale_x": 1.0, + "scale_y": 1.0, + }, + "description": "Original complete screenshot", + } + } + self.next_img_id = 1 + + def _ensure_genai_client(self): + """Initializes or reuses the context-level GenAI client for connection pooling.""" + _ex = facade() + ctx = self.ctx + client = getattr(ctx, "_genai_client", None) + if client is None: + _ex.logger.info( + "Initializing new GenAI client on context for connection pooling (Explorer)..." + ) + client = _ex.genai.Client( + api_key=_ex.settings.GOOGLE_API_KEY.get_secret_value() + if _ex.settings.GOOGLE_API_KEY + else None + ) + ctx._genai_client = client + return client + + def _resolve_model_params(self) -> tuple: + """Resolves model name, temperature, thinking level, and fallback model.""" + llm_cfg = self.ctx.llm_config.explorer + temperature = 0.1 + fallback_model = None + thinking_level = None + if llm_cfg: + model_name = llm_cfg.model + if "/" in model_name: + model_name = model_name.split("/")[-1] + if getattr(llm_cfg, "temperature", None) is not None: + temperature = llm_cfg.temperature + if getattr(llm_cfg, "thinking_level", None) is not None: + thinking_level = llm_cfg.thinking_level + if getattr(llm_cfg, "fallback", None): + fallback_model = llm_cfg.fallback.model + if "/" in fallback_model: + fallback_model = fallback_model.split("/")[-1] + else: + model_name = "gemini-3.7-flash" + + return model_name, temperature, thinking_level, fallback_model + + def _build_prompt_template( + self, version: str, mode: str, max_iterations: int + ) -> tuple[str | None, str | None]: + """Builds the system prompt template; returns (prompt, error_message).""" + prompt_path = Path(__file__).parent / "explorer.json" + if not prompt_path.exists(): + return None, "Error: Explorer prompt template not found." + + try: + content = prompt_path.read_text(encoding="utf-8") + content_no_comments = re.sub(r"(? tuple[str, object]: + """Draws labeled dots for the fused hierarchy; returns (minimal_list, marked_path).""" + _ex = facade() + ctx = self.ctx + marked_path = None + try: + _ex.logger.info( + "Explorer self-annotating initial screenshot using latest_ui_hierarchy..." + ) + formatted_list, points, labels = format_minimal_list_with_points( + fused_xml, self.width, self.height + ) + minimal_list = formatted_list + self.global_label_idx = len(points) + 1 + + base_dir = ( + Path(ctx.data_engine.base_dir) + if ctx.data_engine and getattr(ctx.data_engine, "base_dir", None) + else None + ) + if not base_dir: + db_path = _ex.settings.DATA_ENGINE_DB_PATH + base_dir = _ex.settings.TRACES_PATH + images_dir = base_dir / "images" + initial_marked_dir = images_dir / "initial_marked" + initial_marked_dir.mkdir(parents=True, exist_ok=True) + + existing_files = glob.glob( + str(initial_marked_dir / f"{image_name or 'temp_image'}_*.jpg") + ) + max_seq = 0 + for f in existing_files: + match = re.search(r"_(\d+)\.jpg$", f) + if match: + max_seq = max(max_seq, int(match.group(1))) + seq = max_seq + 1 + marked_path = initial_marked_dir / f"{image_name or 'temp_image'}_{seq}.jpg" + + # Draw dots on the raw screenshot + _ex.draw_dots(screenshot_path, points, labels, str(marked_path)) + _ex.logger.info( + f"Successfully drew {len(points)} dots and saved marked" + f" image to {marked_path}" + ) + except Exception as e: + _ex.logger.error(f"Failed to self-annotate initial screenshot: {e}") + return minimal_list, marked_path + + async def _prepare_initial_annotation( + self, record, state: State, screenshot_path: str, minimal_list: str, image_name + ) -> tuple[str, str]: + """Generates initial visual annotations when no minimal list is provided. + + Returns the (possibly updated) minimal list and the image path the + model should read (marked screenshot when available, raw otherwise). + """ + marked_path = None + if not minimal_list: + fused_xml = await self._build_fused_hierarchy(record, state, screenshot_path) + if fused_xml: + minimal_list, marked_path = self._annotate_initial_screenshot( + fused_xml, screenshot_path, image_name, minimal_list + ) + + if marked_path and os.path.exists(str(marked_path)): + img_to_read = str(marked_path) + else: + img_to_read = screenshot_path + return minimal_list, img_to_read diff --git a/artemis/agents/explorer/tool_declarations.py b/artemis/agents/explorer/tool_declarations.py new file mode 100644 index 0000000..fee1e1e --- /dev/null +++ b/artemis/agents/explorer/tool_declarations.py @@ -0,0 +1,395 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Static tool declarations for the Explorer agent. + +Split out of ``artemis.agents.explorer.explorer`` as pure data: +``UNIVERSAL_EXPLORER_TOOLS`` (OpenAI-style dict schemas for the universal +LangChain path) and ``NATIVE_EXPLORER_TOOL_DECLARATIONS`` (google-genai +``FunctionDeclaration`` objects for the native Gemini path). Both are +re-exported by the ``explorer`` facade module. +""" + +from google.genai import types + +UNIVERSAL_EXPLORER_TOOLS = [ + { + "type": "function", + "function": { + "name": "ask_perception_tool", + "description": ( + "[Perception] Concurrently executes and awaits three sub-tasks in parallel," + " including database search, coordinate search, and visual object detection." + ), + "parameters": { + "type": "object", + "properties": { + "search_query": { + "type": "string", + "description": "Required. Text query to search for in UI elements.", + }, + "nx": { + "type": "integer", + "description": "Required. Normalized X coordinate (0-1000 range).", + }, + "ny": { + "type": "integer", + "description": "Required. Normalized Y coordinate (0-1000 range).", + }, + "detect_queries": { + "type": "array", + "items": {"type": "string"}, + "description": "Required. List of query strings to visually detect.", + }, + }, + "required": ["search_query", "nx", "ny", "detect_queries"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "detect_objects", + "description": ( + "[Perception] Locates elements, shapes and more using powerful VLM." + " Returns their normalized coordinates (0-1000 scale) with label dots." + ), + "parameters": { + "type": "object", + "properties": { + "target_image_id": { + "type": "string", + "description": "The ID of the image to process.", + }, + "queries": { + "type": "array", + "items": {"type": "string"}, + "description": "A list of single query strings to detect on the screen.", + }, + }, + "required": ["queries", "target_image_id"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "get_ocr_list", + "description": ( + "[Perception] Retrieves all text elements detected on the screen via OCR." + ), + "parameters": {"type": "object", "properties": {}}, + }, + }, + { + "type": "function", + "function": { + "name": "ask_image_processor", + "description": ("[Transformation] A scripting tool for pixel-level image processing."), + "parameters": { + "type": "object", + "properties": { + "target_image_id": { + "type": "string", + "description": "The ID of the image in the Image Pool to start from.", + }, + "instruction": { + "type": "string", + "description": "1-2 precise commands describing a simple task.", + }, + }, + "required": ["instruction", "target_image_id"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "inspect_region", + "description": ( + "[Perception] Crops and resizes a bounding box region of the original screenshot." + ), + "parameters": { + "type": "object", + "properties": { + "x_min": {"type": "integer", "description": "Left X coordinate (0-1000)."}, + "y_min": {"type": "integer", "description": "Top Y coordinate (0-1000)."}, + "x_max": {"type": "integer", "description": "Right X coordinate (0-1000)."}, + "y_max": {"type": "integer", "description": "Bottom Y coordinate (0-1000)."}, + "zoom_factor": { + "type": "number", + "description": "Zoom factor to upscale the cropped region (1.0 to 4.0).", + }, + }, + "required": ["x_min", "y_min", "x_max", "y_max", "zoom_factor"], + }, + }, + }, + { + "type": "function", + "function": { + "name": "submit_answer", + "description": ( + "[Finalization] Submits the final list of ranked candidate UI elements matching" + " the search query, or a fallback message if no elements were found." + ), + "parameters": { + "type": "object", + "properties": { + "candidates": { + "type": "array", + "items": { + "type": "object", + "properties": { + "label": { + "type": "string", + "description": "Reference label (e.g. S1, D2, T3)", + }, + "description": { + "type": "string", + "description": "Brief description of this candidate element", + }, + "coords": { + "type": "array", + "items": {"type": "integer"}, + "description": ( + "[nx, ny] coordinates in normalized 0-1000 scale" + ), + }, + }, + "required": ["label", "coords"], + }, + "description": "A ranked list of candidate UI elements matching the query.", + }, + "fallback_message": { + "type": "string", + "description": ( + "Optional fallback explanation if target elements were not found." + ), + }, + }, + "required": ["candidates"], + }, + }, + }, +] + + +NATIVE_EXPLORER_TOOL_DECLARATIONS = [ + types.FunctionDeclaration( + name="ask_perception_tool", + description=( + "[Perception] Concurrently executes and awaits three sub-tasks" + " in parallel, including database search, coordinate search," + " and visual object detection on the screen.\n\nOutput" + " Format:\nReturns:'text' (str): A single string consolidating" + " results with 0-1000 normalized coordinates, including UI" + " search matches labeled [X]," + " coordinate search matches labeled [X], and visually detected" + " object targets labeled [D].\n(Annotated screenshots are also" + " added to the context containing corresponding colored label" + " dots)." + ), + parameters=types.Schema( + type=types.Type.OBJECT, + properties={ + "search_query": types.Schema( + type=types.Type.STRING, + description=("Required. Text query to search for in UI elements."), + ), + "nx": types.Schema( + type=types.Type.INTEGER, + description=( + "Required. Normalized X coordinate (0-1000 range)" + " to audit for overlapping elements." + ), + ), + "ny": types.Schema( + type=types.Type.INTEGER, + description=( + "Required. Normalized Y coordinate (0-1000 range)" + " to audit for overlapping elements." + ), + ), + "detect_queries": types.Schema( + type=types.Type.ARRAY, + items=types.Schema(type=types.Type.STRING), + description=( + "Required. List of query strings to visually detect on the screen." + ), + ), + }, + required=["search_query", "nx", "ny", "detect_queries"], + ), + ), + types.FunctionDeclaration( + name="detect_objects", + description=( + "[Perception] Locates elements, shapes and more using powerful" + " VLM. Returns their normalized coordinates (in a [0, 1000]" + " scale) labeled with [D1], [D2] etc. in the text output, along" + " with an annotated image containing corresponding visual label" + " dots." + ), + parameters=types.Schema( + type=types.Type.OBJECT, + properties={ + "target_image_id": types.Schema( + type=types.Type.STRING, + description="The ID of the image to process.", + ), + "queries": types.Schema( + type=types.Type.ARRAY, + items=types.Schema(type=types.Type.STRING), + description=("A list of single query strings to detect on the screen."), + ), + }, + required=["queries", "target_image_id"], + ), + ), + types.FunctionDeclaration( + name="get_ocr_list", + description=( + "[Perception] Retrieves all text elements detected on the" + " original, unprocessed screen via OCR. Returns their" + " normalized coordinates (in a [0, 1000] scale) labeled with" + " [T1], [T2] etc. in the text output, along with an annotated" + " image containing corresponding visual label dots." + ), + parameters=types.Schema( + type=types.Type.OBJECT, + properties={}, + ), + ), + types.FunctionDeclaration( + name="ask_image_processor", + description=( + "[Transformation] A scripting tool for pixel-level image" + " processing. CAUTION: This tool is slow and expensive. Returns" + " a summary of the operations performed and new image IDs," + " along with new labels and screen coordinates." + ), + parameters=types.Schema( + type=types.Type.OBJECT, + properties={ + "target_image_id": types.Schema( + type=types.Type.STRING, + description=( + "The ID of the image in the Image Pool to start" + " from (e.g. 'img_0', 'img_1')." + ), + ), + "instruction": types.Schema( + type=types.Type.STRING, + description=( + "1-2 precise commands describing a simple task." + " Include all necessary coordinates and other" + " parameters.\n" + ), + ), + }, + required=["instruction", "target_image_id"], + ), + ), + types.FunctionDeclaration( + name="submit_answer", + description=( + "[Finalization] Submits the final list of ranked candidate UI" + " elements matching the search query, or a fallback message if" + " no elements were found, to complete the search task. Fails" + " validation if: (1) candidates is empty and no" + " fallback_message is provided, (2) coordinates are not exactly" + " 2 integers [nx, ny], or (3) coordinates lie outside" + " normalized [0, 1000] range." + ), + parameters=types.Schema( + type=types.Type.OBJECT, + properties={ + "candidates": types.Schema( + type=types.Type.ARRAY, + items=types.Schema( + type=types.Type.OBJECT, + properties={ + "label": types.Schema( + type=types.Type.STRING, + description=( + "Reference label (e.g., S1, D2, T3, or number like 3)" + ), + ), + "coords": types.Schema( + type=types.Type.ARRAY, + items=types.Schema(type=types.Type.INTEGER), + description=("[nx, ny] coordinates in normalized 0-1000 scale"), + ), + "description": types.Schema( + type=types.Type.STRING, + description=("Brief description of this candidate element"), + ), + }, + required=["label", "coords"], + ), + description=( + "A ranked list of candidate UI elements that match" + " the query, sorted from highest to lowest" + " confidence (maximum 10 candidates). Leave empty" + " if no matching elements were found on the screen." + ), + ), + "fallback_message": types.Schema( + type=types.Type.STRING, + description=( + "Optional fallback explanation if target elements" + " were not found or if there are special" + " observations." + ), + ), + }, + required=["candidates"], + ), + ), + types.FunctionDeclaration( + name="inspect_region", + description=( + "[Perception] Crops and resizes a bounding box region of the" + " original screenshot. Returns the cropped and resized image" + " directly, without any image ID or coordinate transform." + ), + parameters=types.Schema( + type=types.Type.OBJECT, + properties={ + "x_min": types.Schema( + type=types.Type.INTEGER, + description="Left X coordinate (0-1000).", + ), + "y_min": types.Schema( + type=types.Type.INTEGER, + description="Top Y coordinate (0-1000).", + ), + "x_max": types.Schema( + type=types.Type.INTEGER, + description="Right X coordinate (0-1000).", + ), + "y_max": types.Schema( + type=types.Type.INTEGER, + description="Bottom Y coordinate (0-1000).", + ), + "zoom_factor": types.Schema( + type=types.Type.NUMBER, + description=("Zoom factor to upscale the cropped region (1.0 to 4.0)."), + ), + }, + required=["x_min", "y_min", "x_max", "y_max", "zoom_factor"], + ), + ), +] diff --git a/artemis/agents/explorer/universal_runner.py b/artemis/agents/explorer/universal_runner.py new file mode 100644 index 0000000..6eee3d0 --- /dev/null +++ b/artemis/agents/explorer/universal_runner.py @@ -0,0 +1,261 @@ +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Universal (LangChain ChatModel) execution loop for the Explorer agent. + +Split out of ``artemis.agents.explorer.explorer``: the ``_run_universal`` +reasoning loop, packaged as a mixin consumed by ``Explorer``. Patched +collaborators (``get_llm``, ``is_ocr_configured``, ``logger``, +``UNIVERSAL_EXPLORER_TOOLS``) are resolved through the facade module at call +time; see ``artemis.agents.explorer._facade``. +""" + +import asyncio +import base64 +import json +from typing import Any + +from langchain_core.messages import ( + BaseMessage, + HumanMessage, + SystemMessage, + ToolMessage, +) + +from artemis.agents.explorer._facade import facade +from artemis.graph.state import State + + +class UniversalRunnerMixin: + """Universal-engine reasoning loop of :class:`Explorer`.""" + + async def _run_universal( + self, + query: str, + context_feedback: str, + screenshot_path: str, + state: State, + minimal_list: str, + version: str, + prompt_template: str, + max_iterations: int, + ) -> str: + """Executes Explorer reasoning loop via Universal LangChain ChatModel.""" + _ex = facade() + llm = _ex.get_llm(self.ctx, name="explorer") + + # Filter universal tools based on denylist and OCR configuration + denylisted = set(self.denylisted_tools) + if not _ex.is_ocr_configured(): + denylisted.add("get_ocr_list") + exposed_tools = [ + t for t in _ex.UNIVERSAL_EXPLORER_TOOLS if t["function"]["name"] not in denylisted + ] + bound_llm = llm.bind_tools(exposed_tools) + + messages = self._build_universal_messages( + query, context_feedback, screenshot_path, minimal_list, prompt_template + ) + + iterations = 0 + agent_outcome = "" + + while iterations < max_iterations: + iterations += 1 + is_final_turn = iterations == max_iterations + + if is_final_turn: + messages.append( + HumanMessage( + content=( + "[WARNING] This is your final iteration. You MUST" + " call 'submit_answer' to submit your final result." + ) + ) + ) + submit_only_tools = [ + t for t in exposed_tools if t["function"]["name"] == "submit_answer" + ] + current_llm = llm.bind_tools(submit_only_tools) if submit_only_tools else bound_llm + else: + current_llm = bound_llm + + response = await asyncio.wait_for(current_llm.ainvoke(messages), timeout=180) + messages.append(response) + + tool_calls = getattr(response, "tool_calls", None) or [] + if not tool_calls: + _ex.logger.warning( + f"Universal Explorer iteration {iterations}: No tool calls generated." + ) + if is_final_turn: + return json.dumps( + { + "candidates": [], + "fallback_message": str(response.content) or "No candidates found.", + }, + ensure_ascii=False, + ) + continue + + # Check for submit_answer + submit_outcome = self._universal_submit_outcome(tool_calls) + if submit_outcome is not None: + agent_outcome = submit_outcome + return agent_outcome + + # Execute non-submit tools + await self._universal_dispatch_tools(tool_calls, messages, iterations) + + if not agent_outcome: + agent_outcome = json.dumps( + { + "candidates": [], + "fallback_message": ( + "Explorer reached max iterations without finding candidates." + ), + }, + ensure_ascii=False, + ) + + return agent_outcome + + def _build_universal_messages( + self, + query: str, + context_feedback: str, + screenshot_path: str, + minimal_list: str, + prompt_template: str, + ) -> list[BaseMessage]: + """Builds the initial system/user message pair for the universal loop.""" + with open(screenshot_path, "rb") as f: + img_b64 = base64.b64encode(f.read()).decode("utf-8") + + user_content: list[dict[str, Any]] = [ + { + "type": "text", + "text": ( + f"Operator Request:\n- Query: {query}\n" + f"- Context Feedback: {context_feedback}\n\n" + "Initial marked UI elements list:\n" + f"{minimal_list}\n" + ), + }, + { + "type": "image_url", + "image_url": {"url": f"data:image/jpeg;base64,{img_b64}"}, + }, + ] + + messages: list[BaseMessage] = [ + SystemMessage(content=prompt_template), + HumanMessage(content=user_content), + ] + return messages + + def _universal_submit_outcome(self, tool_calls: list) -> str | None: + """Returns the validated submit_answer outcome JSON, or None if absent.""" + submit_call = next((tc for tc in tool_calls if tc.get("name") == "submit_answer"), None) + if not submit_call: + return None + + args = submit_call.get("args", {}) + candidates = args.get("candidates", []) + fallback_message = args.get("fallback_message", "") + + # Validate coordinates + valid_candidates = [] + for cand in candidates: + if isinstance(cand, dict) and "coords" in cand: + coords = cand["coords"] + if isinstance(coords, list) and len(coords) == 2: + try: + nx, ny = int(coords[0]), int(coords[1]) + if 0 <= nx <= 1000 and 0 <= ny <= 1000: + valid_candidates.append(cand) + except (ValueError, TypeError): + pass + + return json.dumps( + { + "candidates": valid_candidates, + "fallback_message": fallback_message, + }, + ensure_ascii=False, + ) + + async def _universal_dispatch_tools( + self, tool_calls: list, messages: list[BaseMessage], iterations: int + ) -> None: + """Executes non-submit tool calls and appends their ToolMessages.""" + _ex = facade() + for tc in tool_calls: + name = tc.get("name") + args = tc.get("args", {}) + call_id = tc.get("id", f"call_{iterations}_{name}") + + if name in self.denylisted_tools: + messages.append( + ToolMessage( + tool_call_id=call_id, + name=name, + content=f"Tool '{name}' is denylisted and unavailable.", + ) + ) + continue + + try: + if name == "ask_perception_tool": + res = await self.exec_ask_perception_tool( + search_query=args.get("search_query"), + nx=args.get("nx"), + ny=args.get("ny"), + detect_queries=args.get("detect_queries"), + ) + elif name == "detect_objects": + res = await self.exec_detect_objects( + queries=args.get("queries"), + target_image_id=args.get("target_image_id", "img_0"), + ) + elif name == "get_ocr_list": + res = await self.exec_get_ocr_list() + elif name == "ask_image_processor": + res = await self.exec_ask_image_processor( + instruction=args.get("instruction"), + target_image_id=args.get("target_image_id", "img_0"), + ) + elif name == "inspect_region": + res = await self.exec_inspect_region( + x_min=args.get("x_min"), + y_min=args.get("y_min"), + x_max=args.get("x_max"), + y_max=args.get("y_max"), + zoom_factor=args.get("zoom_factor", 2.0), + ) + else: + res = {"text": f"Error: Tool '{name}' is not recognized."} + + res_text = res.get("text") or res.get("result") or str(res) + messages.append(ToolMessage(tool_call_id=call_id, name=name, content=res_text)) + + except Exception as tool_err: + _ex.logger.error(f"Error executing tool '{name}': {tool_err}") + messages.append( + ToolMessage( + tool_call_id=call_id, + name=name, + content=f"Error executing tool '{name}': {tool_err}", + ) + )