diff --git a/automation/actions.py b/automation/actions.py new file mode 100644 index 0000000..509c347 --- /dev/null +++ b/automation/actions.py @@ -0,0 +1,181 @@ +from __future__ import annotations + +from enum import Enum +from typing import Annotated, Literal, Union + +from pydantic import BaseModel, ConfigDict, Field, TypeAdapter + + +class ActionType(str, Enum): + """Actions the model is allowed to request.""" + + CLICK = "click" + DOUBLE_CLICK = "double_click" + RIGHT_CLICK = "right_click" + MOVE = "move" + TYPE = "type" + PRESS = "press" + HOTKEY = "hotkey" + SCROLL = "scroll" + WAIT = "wait" + FINISH = "finish" + FAIL = "fail" + + +class BaseAction(BaseModel): + """Shared validation for every model action.""" + + model_config = ConfigDict(extra="forbid") + + action: ActionType + reason: str = Field( + min_length=1, + max_length=300, + description="Short explanation of why this action is needed.", + ) + + +class CoordinateAction(BaseAction): + """Base class for actions that target a screen coordinate.""" + + x: int = Field(ge=0) + y: int = Field(ge=0) + + +class ClickAction(CoordinateAction): + action: Literal[ActionType.CLICK] = ActionType.CLICK + + +class DoubleClickAction(CoordinateAction): + action: Literal[ActionType.DOUBLE_CLICK] = ActionType.DOUBLE_CLICK + + +class RightClickAction(CoordinateAction): + action: Literal[ActionType.RIGHT_CLICK] = ActionType.RIGHT_CLICK + + +class MoveAction(CoordinateAction): + action: Literal[ActionType.MOVE] = ActionType.MOVE + duration: float = Field( + default=0.2, + ge=0, + le=2, + description="Seconds used to move the cursor.", + ) + + +class TypeAction(BaseAction): + action: Literal[ActionType.TYPE] = ActionType.TYPE + text: str = Field( + min_length=1, + max_length=5000, + description="Text to type into the focused application.", + ) + interval: float = Field( + default=0.01, + ge=0, + le=0.25, + description="Delay between keystrokes.", + ) + + +class PressAction(BaseAction): + action: Literal[ActionType.PRESS] = ActionType.PRESS + key: str = Field( + min_length=1, + max_length=30, + description="One keyboard key, such as enter, tab, or esc.", + ) + presses: int = Field(default=1, ge=1, le=20) + interval: float = Field(default=0.05, ge=0, le=1) + + +class HotkeyAction(BaseAction): + action: Literal[ActionType.HOTKEY] = ActionType.HOTKEY + keys: list[str] = Field( + min_length=2, + max_length=5, + description="Keys pressed together, such as ['ctrl', 'l'].", + ) + + +class ScrollAction(BaseAction): + action: Literal[ActionType.SCROLL] = ActionType.SCROLL + amount: int = Field( + ge=-20, + le=20, + description="Positive scrolls up and negative scrolls down.", + ) + x: int | None = Field( + default=None, + ge=0, + description="Optional horizontal position before scrolling.", + ) + y: int | None = Field( + default=None, + ge=0, + description="Optional vertical position before scrolling.", + ) + + +class WaitAction(BaseAction): + action: Literal[ActionType.WAIT] = ActionType.WAIT + seconds: float = Field( + ge=0.1, + le=10, + description="How long to wait for the interface to update.", + ) + + +class FinishAction(BaseAction): + action: Literal[ActionType.FINISH] = ActionType.FINISH + summary: str = Field( + min_length=1, + max_length=500, + description="What was completed.", + ) + + +class FailAction(BaseAction): + action: Literal[ActionType.FAIL] = ActionType.FAIL + error: str = Field( + min_length=1, + max_length=500, + description="Why the task cannot continue.", + ) + + +# The action field tells Pydantic which schema to use. +ComputerAction = Annotated[ + Union[ + ClickAction, + DoubleClickAction, + RightClickAction, + MoveAction, + TypeAction, + PressAction, + HotkeyAction, + ScrollAction, + WaitAction, + FinishAction, + FailAction, + ], + Field(discriminator="action"), +] + +ACTION_ADAPTER = TypeAdapter(ComputerAction) + + +def parse_action(data: str | bytes | dict) -> ComputerAction: + """Validate a model response and return a typed action.""" + + if isinstance(data, dict): + return ACTION_ADAPTER.validate_python(data) + + return ACTION_ADAPTER.validate_json(data) + + +def action_json_schema() -> dict: + """Return the schema sent to the local model.""" + + return ACTION_ADAPTER.json_schema() \ No newline at end of file diff --git a/automation/agent.py b/automation/agent.py index 509c347..76a3b8b 100644 --- a/automation/agent.py +++ b/automation/agent.py @@ -1,181 +1,108 @@ from __future__ import annotations -from enum import Enum -from typing import Annotated, Literal, Union +from pathlib import Path +from typing import Any -from pydantic import BaseModel, ConfigDict, Field, TypeAdapter +from actions import FailAction, FinishAction +from computer import Computer, ComputerError +from model import LocalModel, ModelError -class ActionType(str, Enum): - """Actions the model is allowed to request.""" +class AgentError(Exception): + """Base error for agent failures.""" - CLICK = "click" - DOUBLE_CLICK = "double_click" - RIGHT_CLICK = "right_click" - MOVE = "move" - TYPE = "type" - PRESS = "press" - HOTKEY = "hotkey" - SCROLL = "scroll" - WAIT = "wait" - FINISH = "finish" - FAIL = "fail" - - -class BaseAction(BaseModel): - """Shared validation for every model action.""" - - model_config = ConfigDict(extra="forbid") - action: ActionType - reason: str = Field( - min_length=1, - max_length=300, - description="Short explanation of why this action is needed.", - ) - - -class CoordinateAction(BaseAction): - """Base class for actions that target a screen coordinate.""" - - x: int = Field(ge=0) - y: int = Field(ge=0) - - -class ClickAction(CoordinateAction): - action: Literal[ActionType.CLICK] = ActionType.CLICK - - -class DoubleClickAction(CoordinateAction): - action: Literal[ActionType.DOUBLE_CLICK] = ActionType.DOUBLE_CLICK - - -class RightClickAction(CoordinateAction): - action: Literal[ActionType.RIGHT_CLICK] = ActionType.RIGHT_CLICK - - -class MoveAction(CoordinateAction): - action: Literal[ActionType.MOVE] = ActionType.MOVE - duration: float = Field( - default=0.2, - ge=0, - le=2, - description="Seconds used to move the cursor.", - ) - - -class TypeAction(BaseAction): - action: Literal[ActionType.TYPE] = ActionType.TYPE - text: str = Field( - min_length=1, - max_length=5000, - description="Text to type into the focused application.", - ) - interval: float = Field( - default=0.01, - ge=0, - le=0.25, - description="Delay between keystrokes.", - ) - - -class PressAction(BaseAction): - action: Literal[ActionType.PRESS] = ActionType.PRESS - key: str = Field( - min_length=1, - max_length=30, - description="One keyboard key, such as enter, tab, or esc.", - ) - presses: int = Field(default=1, ge=1, le=20) - interval: float = Field(default=0.05, ge=0, le=1) - - -class HotkeyAction(BaseAction): - action: Literal[ActionType.HOTKEY] = ActionType.HOTKEY - keys: list[str] = Field( - min_length=2, - max_length=5, - description="Keys pressed together, such as ['ctrl', 'l'].", - ) - - -class ScrollAction(BaseAction): - action: Literal[ActionType.SCROLL] = ActionType.SCROLL - amount: int = Field( - ge=-20, - le=20, - description="Positive scrolls up and negative scrolls down.", - ) - x: int | None = Field( - default=None, - ge=0, - description="Optional horizontal position before scrolling.", - ) - y: int | None = Field( - default=None, - ge=0, - description="Optional vertical position before scrolling.", - ) - - -class WaitAction(BaseAction): - action: Literal[ActionType.WAIT] = ActionType.WAIT - seconds: float = Field( - ge=0.1, - le=10, - description="How long to wait for the interface to update.", - ) - - -class FinishAction(BaseAction): - action: Literal[ActionType.FINISH] = ActionType.FINISH - summary: str = Field( - min_length=1, - max_length=500, - description="What was completed.", - ) - - -class FailAction(BaseAction): - action: Literal[ActionType.FAIL] = ActionType.FAIL - error: str = Field( - min_length=1, - max_length=500, - description="Why the task cannot continue.", - ) - - -# The action field tells Pydantic which schema to use. -ComputerAction = Annotated[ - Union[ - ClickAction, - DoubleClickAction, - RightClickAction, - MoveAction, - TypeAction, - PressAction, - HotkeyAction, - ScrollAction, - WaitAction, - FinishAction, - FailAction, - ], - Field(discriminator="action"), -] - -ACTION_ADAPTER = TypeAdapter(ComputerAction) - - -def parse_action(data: str | bytes | dict) -> ComputerAction: - """Validate a model response and return a typed action.""" - - if isinstance(data, dict): - return ACTION_ADAPTER.validate_python(data) - - return ACTION_ADAPTER.validate_json(data) - - -def action_json_schema() -> dict: - """Return the schema sent to the local model.""" - - return ACTION_ADAPTER.json_schema() \ No newline at end of file +class StepLimitError(AgentError): + """Raised when the agent reaches its maximum step count.""" + + +class ComputerAgent: + """Coordinates the model and computer-control loop.""" + + def __init__( + self, + model: LocalModel | None = None, + computer: Computer | None = None, + max_steps: int = 50, + ) -> None: + if max_steps < 1: + raise ValueError("max_steps must be at least 1.") + + self.model = model or LocalModel() + self.computer = computer or Computer() + self.max_steps = max_steps + + def run(self, task: str) -> str: + """Run the agent until the task succeeds, fails, or times out.""" + + cleaned_task = task.strip() + + if not cleaned_task: + raise ValueError("Task cannot be empty.") + + action_history: list[dict[str, Any]] = [] + + for step_number in range(1, self.max_steps + 1): + screenshot_path = self._capture_step_screenshot(step_number) + screen_width, screen_height = self.computer.screen_size() + + try: + action = self.model.generate_action( + task=cleaned_task, + screenshot_path=screenshot_path, + screen_width=screen_width, + screen_height=screen_height, + action_history=action_history, + ) + except ModelError as error: + raise AgentError( + f"Model failed during step {step_number}: {error}" + ) from error + + action_record = action.model_dump(mode="json") + action_record["step"] = step_number + action_history.append(action_record) + + print( + f"[Step {step_number}] " + f"{action.action.value}: {action.reason}" + ) + + if isinstance(action, FinishAction): + return action.summary + + if isinstance(action, FailAction): + raise AgentError(action.error) + + try: + self.computer.execute(action) + except ComputerError as error: + raise AgentError( + f"Computer action failed during step {step_number}: {error}" + ) from error + + except Exception as error: + raise AgentError( + f"Unexpected execution error during step {step_number}: " + f"{error}" + ) from error + + raise StepLimitError( + f"The task did not finish within {self.max_steps} steps." + ) + + def _capture_step_screenshot(self, step_number: int) -> Path: + """Capture the screen before requesting the next action.""" + + filename = f"step_{step_number:03d}.png" + + try: + _, screenshot_path = self.computer.take_screenshot( + filename=filename + ) + except Exception as error: + raise AgentError( + f"Could not capture screenshot for step {step_number}: {error}" + ) from error + + return screenshot_path diff --git a/automation/computer.py b/automation/computer.py index adcd009..6080918 100644 --- a/automation/computer.py +++ b/automation/computer.py @@ -1,59 +1,254 @@ -""" -computer.py +from __future__ import annotations -Handles all interaction with the local computer. -""" - -import subprocess +import time +from datetime import datetime from pathlib import Path -import mss +import pyautogui from PIL import Image +from actions import ( + ClickAction, + ComputerAction, + DoubleClickAction, + FailAction, + FinishAction, + HotkeyAction, + MoveAction, + PressAction, + RightClickAction, + ScrollAction, + TypeAction, + WaitAction, +) + + +class ComputerError(Exception): + """Base error for computer-control failures.""" + + +class InvalidCoordinateError(ComputerError): + """Raised when a coordinate is outside the screen.""" + class Computer: - def __init__(self): - self.generated_file = Path("generated.py") - self.screenshot_file = Path("screenshot.png") + """Controls the mouse, keyboard, and screen.""" + + def __init__( + self, + screenshot_dir: str | Path = "screenshots", + action_pause: float = 0.2, + ) -> None: + self.screenshot_dir = Path(screenshot_dir) + self.screenshot_dir.mkdir(parents=True, exist_ok=True) + + self.action_pause = action_pause - def take_screenshot(self) -> Path: - """ - Captures the current screen. - Returns the path to the screenshot. - """ + # Moving the mouse to a screen corner stops PyAutoGUI. + pyautogui.FAILSAFE = True + pyautogui.PAUSE = action_pause - with mss.mss() as sct: - monitor = sct.monitors[1] - screenshot = sct.grab(monitor) + def screen_size(self) -> tuple[int, int]: + """Return the current screen width and height.""" + + size = pyautogui.size() + return size.width, size.height + + def mouse_position(self) -> tuple[int, int]: + """Return the current mouse position.""" + + position = pyautogui.position() + return position.x, position.y + + def validate_coordinates(self, x: int, y: int) -> None: + """Confirm that a coordinate is inside the current screen.""" + + width, height = self.screen_size() + + if not 0 <= x < width: + raise InvalidCoordinateError( + f"x coordinate {x} is outside the screen width {width}." + ) - image = Image.frombytes( - "RGB", - screenshot.size, - screenshot.rgb, + if not 0 <= y < height: + raise InvalidCoordinateError( + f"y coordinate {y} is outside the screen height {height}." ) - image.save(self.screenshot_file) + def take_screenshot( + self, + filename: str | None = None, + ) -> tuple[Image.Image, Path]: + """Capture the screen and save it to the screenshot directory.""" - return self.screenshot_file + if filename is None: + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + filename = f"screenshot_{timestamp}.png" - def execute(self, code: str) -> dict: - """ - Saves and executes the Python code generated by the AI. - """ + path = self.screenshot_dir / filename + image = pyautogui.screenshot() + image.save(path) - self.generated_file.write_text( - code, - encoding="utf-8", + return image, path + + def move_mouse( + self, + x: int, + y: int, + duration: float = 0.2, + ) -> None: + """Move the cursor to a screen coordinate.""" + + self.validate_coordinates(x, y) + pyautogui.moveTo(x, y, duration=duration) + + def click(self, x: int, y: int) -> None: + """Perform a left click.""" + + self.validate_coordinates(x, y) + pyautogui.click(x=x, y=y, button="left") + + def double_click(self, x: int, y: int) -> None: + """Perform a left double click.""" + + self.validate_coordinates(x, y) + pyautogui.doubleClick( + x=x, + y=y, + interval=0.1, + button="left", ) - result = subprocess.run( - ["python", str(self.generated_file)], - capture_output=True, - text=True, + def right_click(self, x: int, y: int) -> None: + """Perform a right click.""" + + self.validate_coordinates(x, y) + pyautogui.click(x=x, y=y, button="right") + + def type_text( + self, + text: str, + interval: float = 0.01, + ) -> None: + """Type text into the currently focused application.""" + + pyautogui.write(text, interval=interval) + + def press_key( + self, + key: str, + presses: int = 1, + interval: float = 0.05, + ) -> None: + """Press one keyboard key one or more times.""" + + normalized_key = key.strip().lower() + + if normalized_key not in pyautogui.KEYBOARD_KEYS: + raise ComputerError(f"Unsupported keyboard key: {key}") + + pyautogui.press( + normalized_key, + presses=presses, + interval=interval, ) - return { - "stdout": result.stdout, - "stderr": result.stderr, - "returncode": result.returncode, - } \ No newline at end of file + def press_hotkey(self, keys: list[str]) -> None: + """Press a keyboard shortcut.""" + + normalized_keys = [key.strip().lower() for key in keys] + + invalid_keys = [ + key + for key in normalized_keys + if key not in pyautogui.KEYBOARD_KEYS + ] + + if invalid_keys: + raise ComputerError( + f"Unsupported hotkey keys: {', '.join(invalid_keys)}" + ) + + pyautogui.hotkey(*normalized_keys) + + def scroll( + self, + amount: int, + x: int | None = None, + y: int | None = None, + ) -> None: + """Scroll at the current or requested mouse position.""" + + if (x is None) != (y is None): + raise ComputerError( + "Scroll coordinates must include both x and y." + ) + + if x is not None and y is not None: + self.validate_coordinates(x, y) + pyautogui.moveTo(x, y, duration=0.2) + + pyautogui.scroll(amount) + + def wait(self, seconds: float) -> None: + """Pause while the interface updates.""" + + time.sleep(seconds) + + def execute(self, action: ComputerAction) -> str | None: + """Execute one validated computer action.""" + + if isinstance(action, ClickAction): + self.click(action.x, action.y) + + elif isinstance(action, DoubleClickAction): + self.double_click(action.x, action.y) + + elif isinstance(action, RightClickAction): + self.right_click(action.x, action.y) + + elif isinstance(action, MoveAction): + self.move_mouse( + action.x, + action.y, + duration=action.duration, + ) + + elif isinstance(action, TypeAction): + self.type_text( + action.text, + interval=action.interval, + ) + + elif isinstance(action, PressAction): + self.press_key( + action.key, + presses=action.presses, + interval=action.interval, + ) + + elif isinstance(action, HotkeyAction): + self.press_hotkey(action.keys) + + elif isinstance(action, ScrollAction): + self.scroll( + action.amount, + x=action.x, + y=action.y, + ) + + elif isinstance(action, WaitAction): + self.wait(action.seconds) + + elif isinstance(action, FinishAction): + return action.summary + + elif isinstance(action, FailAction): + raise ComputerError(action.error) + + else: + raise ComputerError( + f"Unsupported action object: {type(action).__name__}" + ) + + return None \ No newline at end of file diff --git a/automation/main.py b/automation/main.py index fac48da..dfae790 100644 --- a/automation/main.py +++ b/automation/main.py @@ -1,59 +1,42 @@ -""" -main.py +from __future__ import annotations -Friday's main loop. -""" +import sys -from computer import Computer -from model import Model -from prompt import SYSTEM_PROMPT +from agent import AgentError, ComputerAgent, StepLimitError -def main(): - computer = Computer() - model = Model() +def main() -> int: + """Start the computer-use agent.""" - goal = input("Open word and write me a paragraph on How AI is good.\n> ") + task = " ".join(sys.argv[1:]).strip() - while True: - # Take the latest screenshot - screenshot = computer.take_screenshot() + if not task: + task = input("What should Friday do? ").strip() - # Build the user prompt - user_prompt = f""" -Current Goal: -{goal} + if not task: + print("A task is required.") + return 1 -This is the latest screenshot of the computer. + agent = ComputerAgent() -Figure out the next step. + try: + result = agent.run(task) -If the task is complete, simply output: + except StepLimitError as error: + print(f"Agent stopped: {error}") + return 2 -TASK_COMPLETE + except AgentError as error: + print(f"Agent failed: {error}") + return 1 -Otherwise, output ONLY executable Python code. -""" + except KeyboardInterrupt: + print("\nAgent stopped by user.") + return 130 - # Ask Gemini what to do next - response = model.send( - SYSTEM_PROMPT, - user_prompt, - screenshot, - ) - - # Stop if the task is finished - if response.strip() == "TASK_COMPLETE": - print("Task completed.") - break - - # Execute the generated Python - result = computer.execute(response) - - # Optional: print any errors for debugging - if result["stderr"]: - print(result["stderr"]) + print(f"Task completed: {result}") + return 0 if __name__ == "__main__": - main() \ No newline at end of file + raise SystemExit(main()) \ No newline at end of file diff --git a/automation/model.py b/automation/model.py index 7096902..aee3615 100644 --- a/automation/model.py +++ b/automation/model.py @@ -1,66 +1,111 @@ -""" -model.py +from __future__ import annotations -Maintains a continuous conversation with Gemini. +from pathlib import Path +from typing import Any -Its only job is to: -1. Send the goal and latest screenshot to Gemini. -2. Keep the conversation alive. -3. Return the Python code Gemini generates. -""" +from ollama import Client, ResponseError -import os +from actions import ComputerAction, action_json_schema, parse_action +from prompt import build_prompt -from google import genai +class ModelError(Exception): + """Base error for local model failures.""" -class Model: - def __init__(self): - api_key = os.getenv("GEMINI_API_KEY") - if not api_key: - raise ValueError("GEMINI_API_KEY not found.") +class ModelConnectionError(ModelError): + """Raised when the Ollama server cannot be reached.""" - self.client = genai.Client(api_key=api_key) - self.model = "gemini-2.5-flash" - # Stores the conversation history so Gemini remembers - # the goal and previous screenshots. - self.history = [] +class ModelResponseError(ModelError): + """Raised when the model returns an invalid response.""" - def send( + +class LocalModel: + """Connects the computer-use agent to a local Ollama model.""" + + def __init__( self, - system_prompt: str, - user_prompt: str, - screenshot, - ) -> str: - """ - Sends the latest computer state to Gemini and returns - the Python code it generates. - """ - - self.history.append( - { - "role": "user", - "parts": [ - {"text": f"{system_prompt}\n\n{user_prompt}"}, - screenshot, - ], - } - ) + model_name: str = "qwen2.5vl:7b", + host: str = "http://localhost:11434", + temperature: float = 0.0, + ) -> None: + self.model_name = model_name + self.temperature = temperature + self.client = Client(host=host) + + def generate_action( + self, + task: str, + screenshot_path: str | Path, + screen_width: int, + screen_height: int, + action_history: list[dict[str, Any]] | None = None, + ) -> ComputerAction: + """Generate and validate the next computer action.""" + + screenshot = Path(screenshot_path) + + if not screenshot.exists(): + raise ModelError( + f"Screenshot does not exist: {screenshot}" + ) - response = self.client.models.generate_content( - model=self.model, - contents=self.history, + if not screenshot.is_file(): + raise ModelError( + f"Screenshot path is not a file: {screenshot}" + ) + + prompt = build_prompt( + task=task, + screen_width=screen_width, + screen_height=screen_height, + action_history=action_history, ) - self.history.append( - { - "role": "model", - "parts": [ - {"text": response.text}, + try: + response = self.client.chat( + model=self.model_name, + messages=[ + { + "role": "user", + "content": prompt, + "images": [screenshot], + } ], - } - ) + format=action_json_schema(), + options={ + "temperature": self.temperature, + }, + stream=False, + ) + + except ResponseError as error: + raise ModelResponseError( + f"Ollama rejected the request: {error}" + ) from error + + except ConnectionError as error: + raise ModelConnectionError( + "Could not connect to Ollama. Make sure Ollama is running." + ) from error + + except Exception as error: + raise ModelError( + f"Unexpected local model error: {error}" + ) from error + + content = response.message.content + + if not content or not content.strip(): + raise ModelResponseError( + "The local model returned an empty response." + ) + + try: + return parse_action(content) - return response.text \ No newline at end of file + except Exception as error: + raise ModelResponseError( + f"The local model returned an invalid action: {content}" + ) from error \ No newline at end of file diff --git a/automation/prompt.py b/automation/prompt.py index adc4353..f03beed 100644 --- a/automation/prompt.py +++ b/automation/prompt.py @@ -1,28 +1,85 @@ +from __future__ import annotations + +import json + +from actions import action_json_schema + + SYSTEM_PROMPT = """ -You are Friday. +You are a local computer-use agent. + +Your job is to complete the user's task by looking at the current screenshot and +choosing exactly one computer action at a time. + +You control the computer through a fixed set of actions. You do not directly +control the mouse or keyboard yourself. Another part of the program executes +the action you return. + +Rules: + +1. Return exactly one valid JSON object. +2. Do not return markdown, code fences, or extra text. +3. Use only the actions defined in the provided JSON schema. +4. Choose the smallest useful next action. +5. Do not return multiple actions at once. +6. Use the latest screenshot as the source of truth. +7. Review the previous actions before repeating an action. +8. If the interface is still loading, use the wait action. +9. If the task is visibly complete, use the finish action. +10. If the task cannot continue safely, use the fail action. +11. Do not claim success unless the screenshot confirms the result. +12. Do not guess coordinates outside the visible screenshot. +13. Keep the reason short and specific. +14. Avoid repeating an action that produced no visible progress. +15. Do not type passwords, payment information, private keys, or security codes. +16. Do not approve purchases, delete files, submit forms, or send messages + without explicit permission in the user's task. -You are controlling a Windows computer. +Coordinate system: -The user will give you a goal. +- The top-left corner is x=0, y=0. +- x increases from left to right. +- y increases from top to bottom. +- Coordinates must target the visible center of the intended interface element. -You will continuously receive screenshots of the computer. +Action guidance: -Your job is to accomplish the goal by writing Python code. +- Use click for buttons, icons, text fields, and menu items. +- Use double_click only when the interface normally requires it. +- Use right_click only when a context menu is needed. +- Use move when hovering is required. +- Use type only after the correct field has focus. +- Use press for one key such as enter, tab, esc, or backspace. +- Use hotkey for shortcuts such as ctrl+l or alt+tab. +- Use scroll when the needed content is outside the visible area. +- Use wait when the computer needs time to update. +- Use finish only after verifying the requested outcome. +- Use fail when the task cannot safely or reasonably continue. +""".strip() -The code you write will be executed immediately. -After execution, you will receive another screenshot. +def build_prompt( + task: str, + screen_width: int, + screen_height: int, + action_history: list[dict] | None = None, +) -> str: + """Build the full instruction prompt for one agent step.""" -Based on the new screenshot, continue writing Python until the task is complete. + history = action_history or [] -Guidelines: + prompt_data = { + "task": task, + "screen": { + "width": screen_width, + "height": screen_height, + }, + "previous_actions": history, + "required_output_schema": action_json_schema(), + } -- Use Python. -- Use any installed Python libraries. -- If another library is required, install it. -- Think step by step. -- Only output executable Python code. -- Do not explain your reasoning. -- Do not wrap your code in Markdown. -- Do not output anything except Python. -""" \ No newline at end of file + return ( + f"{SYSTEM_PROMPT}\n\n" + "Current task data:\n" + f"{json.dumps(prompt_data, indent=2)}" + ) \ No newline at end of file