while True:
    screenshot = 截屏()
    action = 大模型看圖分析(screenshot, "用戶的指令")
    執(zhí)行操作(action)  # 點(diǎn)擊、輸入、滾動(dòng)...
    if action == "done":
        break

pip install pyautogui openai pillow

import pyautogui
import base64
import io
import json
from openai import OpenAI

# 初始化客戶端（兼容 OpenAI 協(xié)議的接口都行）
client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.ofox.ai/v1"  # 支持 50+ 模型的聚合接口
)

SYSTEM_PROMPT = """你是一個(gè)桌面操作助手。用戶會給你一張屏幕截圖和一個(gè)任務(wù)。
你需要分析截圖，返回下一步操作。

返回 JSON 格式：
{"action": "click", "x": 100, "y": 200, "reason": "點(diǎn)擊搜索框"}
{"action": "type", "text": "hello world", "reason": "輸入搜索詞"}
{"action": "scroll", "direction": "down", "reason": "向下滾動(dòng)查看更多"}
{"action": "hotkey", "keys": ["command", "c"], "reason": "復(fù)制選中內(nèi)容"}
{"action": "done", "reason": "任務(wù)完成"}

只返回 JSON，不要其他內(nèi)容。"""

def screenshot_to_base64():
    """截屏并轉(zhuǎn)為 base64"""
    img = pyautogui.screenshot()
    buffer = io.BytesIO()
    img.save(buffer, format="PNG")
    return base64.b64encode(buffer.getvalue()).decode()

def ask_model(image_b64, task, history=None):
    """讓大模型看圖決策"""
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    if history:
        messages.extend(history)
    messages.append({
        "role": "user",
        "content": [
            {"type": "text", "text": f"任務(wù)：{task}\n請分析截圖，返回下一步操作。"},
            {"type": "image_url", "image_url": {
                "url": f"data:image/png;base64,{image_b64}"
            }}
        ]
    })

    resp = client.chat.completions.create(
        model="anthropic/claude-sonnet-4-6",  # 也可以換 gpt-5.4、gemini-3-flash 等
        messages=messages,
        max_tokens=300,
        temperature=0
    )
    return json.loads(resp.choices[0].message.content)

def execute_action(action):
    """執(zhí)行大模型返回的操作"""
    act = action["action"]
    if act == "click":
        pyautogui.click(action["x"], action["y"])
    elif act == "type":
        pyautogui.write(action["text"], interval=0.05)
    elif act == "scroll":
        pyautogui.scroll(-3 if action["direction"] == "down" else 3)
    elif act == "hotkey":
        pyautogui.hotkey(*action["keys"])
    elif act == "done":
        return False
    return True

def run_agent(task, max_steps=10):
    """主循環(huán)"""
    print(f"?? 開始執(zhí)行: {task}")
    history = []

    for step in range(max_steps):
        img_b64 = screenshot_to_base64()
        action = ask_model(img_b64, task, history)
        print(f"  Step {step+1}: {action.get('reason', '...')}")

        history.append({"role": "assistant", "content": json.dumps(action)})

        if not execute_action(action):
            print("? 任務(wù)完成!")
            return True

        pyautogui.sleep(1)  # 等 UI 響應(yīng)

    print("?? 達(dá)到最大步數(shù)，停止執(zhí)行")
    return False

# 用法
if __name__ == "__main__":
    run_agent("打開瀏覽器，搜索今天的天氣")

run_agent("打開 Chrome，在百度搜索 Python 3.13 新特性")

run_agent("把桌面上所有 .tmp 文件拖到廢紙簍")

run_agent("打開公司的 OA 系統(tǒng)，填寫今天的日報(bào)，內(nèi)容寫'完成了桌面 Agent 的技術(shù)調(diào)研'")

# 修復(fù)：截圖時(shí)縮放到邏輯分辨率
def screenshot_to_base64():
    img = pyautogui.screenshot()
    # macOS Retina 需要縮放
    logical_size = (img.width // 2, img.height // 2)
    img = img.resize(logical_size)
    buffer = io.BytesIO()
    img.save(buffer, format="PNG")
    return base64.b64encode(buffer.getvalue()).decode()

import re

def parse_action(text):
    """容錯(cuò)解析 JSON"""
    # 去掉可能的 markdown 代碼塊標(biāo)記
    text = re.sub(r'```json?\s*', '', text)
    text = re.sub(r'```\s*$', '', text)
    return json.loads(text.strip())

pyautogui.FAILSAFE = True  # 默認(rèn)就是 True，別改成 False！

def execute_action(action):
    act = action["action"]
    # 危險(xiǎn)操作確認(rèn)
    if act == "hotkey" and "delete" in str(action.get("keys", [])):
        confirm = input(f"?? 即將執(zhí)行刪除操作: {action['reason']}，確認(rèn)? (y/n) ")
        if confirm != 'y':
            return True  # 跳過這步
    # ... 其余執(zhí)行邏輯

def ask_model(image_b64, task, history=None):
    messages = [{"role": "system", "content": SYSTEM_PROMPT}]
    if history:
        # 只保留最近 5 輪，避免 token 爆炸
        messages.extend(history[-10:])
    # ... 其余邏輯不變

def run_agent_with_reflection(task, max_steps=10):
    prev_img = None
    for step in range(max_steps):
        curr_img = screenshot_to_base64()
        if prev_img and curr_img == prev_img:
            print("?? 屏幕沒變化，可能操作沒生效，重試...")
        # ...
        prev_img = curr_img