Screen AI 与视觉 Agent 全景解析：从 Computer Use 到 Omi，AI 终于「看到」了世界

截图（1920×1080 像素）→ 视觉模型 → 结构化描述
  - "当前页面是 Gmail 收件箱"
  - "有一个红色的'删除'按钮在右上角"
  - "输入框为空，光标已聚焦"

import base64
import time
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import pyautogui
import requests


class ActionType(Enum):
    CLICK = "click"
    TYPE = "type"
    SCROLL = "scroll"
    WAIT = "wait"
    HOTKEY = "hotkey"


@dataclass
class ScreenState:
    """屏幕状态的抽象表示"""
    screenshot_b64: str
    description: str = ""
    detected_elements: list = field(default_factory=list)
    timestamp: float = field(default_factory=time.time)

    def age_seconds(self) -> float:
        return time.time() - self.timestamp


@dataclass
class ScreenAction:
    """定义一个屏幕操作"""
    action_type: ActionType
    x: Optional[int] = None
    y: Optional[int] = None
    text: Optional[str] = None
    keys: Optional[list] = None
    scroll_amount: Optional[int] = None

    def execute(self) -> bool:
        """执行操作并返回结果"""
        try:
            if self.action_type == ActionType.CLICK:
                pyautogui.click(self.x, self.y)
            elif self.action_type == ActionType.TYPE:
                pyautogui.typewrite(self.text)
            elif self.action_type == ActionType.SCROLL:
                pyautogui.scroll(self.scroll_amount)
            elif self.action_type == ActionType.HOTKEY:
                pyautogui.hotkey(*self.keys)
            elif self.action_type == ActionType.WAIT:
                time.sleep(1)
            return True
        except Exception as e:
            print(f"操作执行失败: {e}")
            return False


class ScreenAIAgent:
    """Screen AI Agent 核心类"""

    def __init__(self, vision_api_url: str, api_key: str):
        self.vision_api_url = vision_api_url
        self.headers = {"Authorization": f"Bearer {api_key}"}
        self.history: list[ScreenState] = []

    def capture_screen(self) -> ScreenState:
        """截取屏幕并创建状态"""
        screenshot = pyautogui.screenshot()
        buffered = base64.b64encode(
            screenshot.tobytes()
        ).decode('utf-8')
        state = ScreenState(screenshot_b64=buffered)
        self.history.append(state)
        return state

    def analyze(self, state: ScreenState, goal: str) -> dict:
        """调用视觉模型分析屏幕状态"""
        payload = {
            "image": state.screenshot_b64,
            "goal": goal,
            "history": [s.description for s in self.history[-5:]]
        }
        response = requests.post(
            f"{self.vision_api_url}/analyze",
            headers=self.headers,
            json=payload
        )
        return response.json()

    def plan_and_execute(self, goal: str, max_steps: int = 10) -> str:
        """完整的 Screen AI 执行循环"""
        for step in range(max_steps):
            state = self.capture_screen()
            analysis = self.analyze(state, goal)
            state.description = analysis.get("description", "")
            
            if analysis.get("completed", False):
                return f"✅ 任务完成（{step + 1}步）"
            
            action_data = analysis.get("next_action", {})
            action = ScreenAction(
                action_type=ActionType(action_data.get("type", "wait")),
                x=action_data.get("x"),
                y=action_data.get("y"),
                text=action_data.get("text"),
                keys=action_data.get("keys"),
                scroll_amount=action_data.get("scroll")
            )
            
            success = action.execute()
            if not success:
                return f"❌ 第 {step + 1} 步执行失败"
            
            time.sleep(0.5)
        
        return "⏱️ 达到最大步数限制"


# 使用示例
agent = ScreenAIAgent(
    vision_api_url="http://localhost:8080",
    api_key="your-api-key"
)
result = agent.plan_and_execute("在浏览器中搜索 'AI Agent 最新进展'")
print(result)

from dataclasses import dataclass, field
from typing import Optional
import hashlib
import json
import time


@dataclass
class ExecutionStep:
    """执行步骤记录"""
    step_num: int
    screenshot_hash: str
    action_taken: str
    expected_result: str
    actual_result: str
    success: bool
    timestamp: float = field(default_factory=time.time)


@dataclass
class CorrectionStrategy:
    """自我修正策略"""
    max_retries: int = 3
    retry_delay: float = 1.0
    fallback_actions: list = field(default_factory=list)
    learn_from_failures: bool = True


class SelfCorrectingScreenAgent:
    """带自我修正能力的 Screen AI Agent"""

    def __init__(self):
        self.execution_log: list[ExecutionStep] = []
        self.correction = CorrectionStrategy()
        self.failure_patterns: dict = {}

    def execute_with_correction(
        self,
        action: str,
        goal: str,
        verify_fn: callable
    ) -> bool:
        """执行操作，失败时自动修正"""
        for attempt in range(self.correction.max_retries):
            result = self._perform_action(action)
            
            screenshot = self._capture_screenshot()
            screenshot_hash = hashlib.md5(
                screenshot.tobytes()
            ).hexdigest()
            
            expected = f"操作后屏幕应显示: {goal}"
            actual = verify_fn(screenshot)
            success = actual == goal
            
            step = ExecutionStep(
                step_num=attempt,
                screenshot_hash=screenshot_hash,
                action_taken=action,
                expected_result=expected,
                actual_result=actual,
                success=success
            )
            self.execution_log.append(step)
            
            if success:
                return True
            
            if self.correction.learn_from_failures:
                pattern_key = f"{action}|{actual}"
                self.failure_patterns[pattern_key] = (
                    self.failure_patterns.get(pattern_key, 0) + 1
                )
            
            if attempt < self.correction.max_retries - 1:
                corrected_action = self._plan_correction(
                    action, actual, goal
                )
                if corrected_action:
                    action = corrected_action
                    print(f"  → 修正: {action}")
                time.sleep(self.correction.retry_delay)
        
        return False

    def _plan_correction(
        self,
        original_action: str,
        actual_state: str,
        expected_state: str
    ) -> Optional[str]:
        """根据失败情况规划修正操作"""
        if "click" in original_action:
            return f"click(备选坐标) for {original_action}"
        if "not visible" in actual_state:
            return f"scroll({{amount: 500}}) then {original_action}"
        if "loading" in actual_state:
            return f"wait(2s) then {original_action}"
        if "popup" in actual_state or "dialog" in actual_state:
            return "close_popup() then " + original_action
        return None

    def get_failure_report(self) -> dict:
        """生成失败分析报告"""
        total = len(self.execution_log)
        failed = sum(1 for s in self.execution_log if not s.success)
        return {
            "total_steps": total,
            "success_rate": round((total - failed) / total * 100, 1),
            "common_failures": sorted(
                self.failure_patterns.items(),
                key=lambda x: x[1],
                reverse=True
            )[:5],
            "recent_log": [
                {"step": s.step_num, "action": s.action_taken, "success": s.success}
                for s in self.execution_log[-10:]
            ]
        }


# 使用示例
agent = SelfCorrectingScreenAgent()

def verify_search_result(screenshot) -> str:
    return "搜索结果显示"

result = agent.execute_with_correction(
    action="click(搜索按钮)",
    goal="显示搜索结果",
    verify_fn=verify_search_result
)
print(f"执行结果: {'成功' if result else '失败'}")
print(json.dumps(agent.get_failure_report(), indent=2))