AI Agent 桌面集成：从浏览器到操作系统的全栈架构

# 混合感知方案：DOM API + 截图视觉模型回退
import base64
import time
from playwright.sync_api import sync_playwright
from openai import OpenAI

class DesktopAgentPerception:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
        self.last_perception = None
        self.perception_timestamp = 0

    def perceive_via_dom(self, page) -> dict:
        """通过 DOM 获取结构化信息（高精度，低覆盖）"""
        elements = page.evaluate("""
            () => {
                const interactive = document.querySelectorAll(
                    'button, input, a, select, textarea, [role="button"], [role="link"]'
                );
                return Array.from(interactive).map(el => ({
                    tag: el.tagName.toLowerCase(),
                    type: el.type || el.getAttribute('role'),
                    text: el.textContent?.trim().substring(0, 100),
                    id: el.id,
                    name: el.name,
                    ariaLabel: el.getAttribute('aria-label'),
                    visible: el.offsetParent !== null,
                    rect: el.getBoundingClientRect()
                }));
            }
        """)
        return {"source": "dom", "elements": elements}

    def perceive_via_screenshot(self, page) -> dict:
        """通过截图+视觉模型获取信息（低精度，全覆盖）"""
        screenshot = page.screenshot(type="jpeg", quality=80)
        b64 = base64.b64encode(screenshot).decode()

        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{
                "role": "user",
                "content": [
                    {"type": "text", "text": "描述这个页面上的所有交互元素，包括它们的位置、类型和功能。以JSON格式返回。"},
                    {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}}
                ]
            }]
        )
        return {"source": "vision", "analysis": response.choices[0].message.content}

    def perceive(self, page, prefer_dom: bool = True) -> dict:
        """混合感知：优先 DOM，失败则回退到视觉模型"""
        try:
            if prefer_dom:
                result = self.perceive_via_dom(page)
                if result["elements"] and len(result["elements"]) > 0:
                    self.last_perception = result
                    self.perception_timestamp = time.time()
                    return result
        except Exception as e:
            print(f"DOM 感知失败: {e}，回退到视觉模型")

        # 回退方案
        result = self.perceive_via_screenshot(page)
        self.last_perception = result
        self.perception_timestamp = time.time()
        return result

// macOS Accessibility API 感知方案（Swift）
import ApplicationServices

struct AccessibilityElement {
    let role: String
    let title: String
    let value: String
    let position: CGPoint
    let size: CGSize
    let children: [AccessibilityElement]
}

func perceiveApplication(pid: pid_t) -> [AccessibilityElement] {
    guard let appRef = AXUIElementCreateApplication(pid) else {
        return []
    }

    var windows: CFTypeRef?
    let result = AXUIElementCopyAttributeValue(
        appRef,
        kAXWindowsAttribute as CFString,
        &windows
    )

    guard result == .success, let windowArray = windows as? [AXUIElement] else {
        return []
    }

    var elements: [AccessibilityElement] = []
    for window in windowArray {
        if let root = traverseAXElement(window) {
            elements.append(root)
        }
    }
    return elements
}

func traverseAXElement(_ element: AXUIElement) -> AccessibilityElement? {
    var role: CFTypeRef?
    AXUIElementCopyAttributeValue(element, kAXRoleAttribute as CFString, &role)

    var title: CFTypeRef?
    AXUIElementCopyAttributeValue(element, kAXTitleAttribute as CFString, &title)

    var position: CGPoint = .zero
    var positionRef: CFTypeRef?
    if AXUIElementCopyAttributeValue(element, kAXPositionAttribute as CFString, &positionRef) == .success {
        AXValueGetValue(positionRef as! AXValue, .cgPoint, &position)
    }

    // 递归获取子元素
    var children: [AccessibilityElement] = []
    var childrenRef: CFTypeRef?
    if AXUIElementCopyAttributeValue(element, kAXChildrenAttribute as CFString, &childrenRef) == .success {
        if let childArray = childrenRef as? [AXUIElement] {
            for child in childArray {
                if let childEl = traverseAXElement(child) {
                    children.append(childEl)
                }
            }
        }
    }

    return AccessibilityElement(
        role: role as? String ?? "",
        title: title as? String ?? "",
        value: "",
        position: position,
        size: .zero,
        children: children
    )
}

// 桌面 Agent 执行层：统一的行动抽象 + 条件等待
interface AgentAction {
  type: "click" | "type" | "navigate" | "read" | "wait" | "screenshot";
  selector?: string;
  text?: string;
  url?: string;
  condition?: (ctx: ExecutionContext) => Promise<boolean>;
  timeoutMs?: number;
}

class DesktopAgentExecutor {
  private browser: Browser;
  private maxRetries = 3;

  async execute(action: AgentAction, ctx: ExecutionContext): Promise<ExecutionResult> {
    switch (action.type) {
      case "click":
        return await this.clickWithWait(action, ctx);
      case "type":
        return await this.typeWithWait(action, ctx);
      case "navigate":
        return await this.navigateAndWait(action);
      case "wait":
        return await this.waitForCondition(action, ctx);
      case "screenshot":
        return await this.captureScreenshot();
      default:
        throw new Error(`未知的行动类型: ${action.type}`);
    }
  }

  private async clickWithWait(action: AgentAction, ctx: ExecutionContext): Promise<ExecutionResult> {
    const page = ctx.currentPage;
    const element = await page.waitForSelector(action.selector!, {
      timeout: action.timeoutMs ?? 5000
    });

    // 点击前截图（用于审计和调试）
    const before = await page.screenshot();

    await element.click();

    // 等待条件满足或超时
    if (action.condition) {
      await this.waitForCondition({
        type: "wait",
        condition: action.condition,
        timeoutMs: action.timeoutMs ?? 10000
      }, ctx);
    }

    return { success: true, before, after: await page.screenshot() };
  }

  private async waitForCondition(
    action: AgentAction,
    ctx: ExecutionContext
  ): Promise<ExecutionResult> {
    const start = Date.now();
    const timeout = action.timeoutMs ?? 10000;
    const pollInterval = 200; // 每 200ms 检查一次

    while (Date.now() - start < timeout) {
      try {
        if (action.condition && await action.condition(ctx)) {
          return { success: true, waitTime: Date.now() - start };
        }
      } catch {
        // 条件检查失败，继续等待
      }
      await new Promise(resolve => setTimeout(resolve, pollInterval));
    }

    return { success: false, error: `条件等待超时 (${timeout}ms)` };
  }

  // 行动链：将多个行动组合为一个工作流
  async executeChain(actions: AgentAction[], ctx: ExecutionContext): Promise<ExecutionResult[]> {
    const results: ExecutionResult[] = [];
    for (const action of actions) {
      const result = await this.execute(action, ctx);
      results.push(result);
      if (!result.success) {
        // 行动失败，记录审计日志并停止
        ctx.logAudit({ action, result, timestamp: Date.now() });
        break;
      }
    }
    return results;
  }
}

// 桌面 Agent 核心循环：Perception-Reasoning-Execution (PRE) Loop
import { Browser, Page } from "playwright";
import OpenAI from "openai";

interface PRELoopConfig {
  maxCycles: number;           // 最大循环次数
  cycleDelayMs: number;        // 循环间隔
  userInstruction: string;     // 用户指令
  dangerousActionsRequireConfirmation: boolean; // 危险操作需要确认
}

interface AgentState {
  history: PerceptionRecord[]; // 历史感知记录
  currentCycle: number;        // 当前循环计数
  taskCompleted: boolean;      // 任务是否完成
  taskFailed: boolean;         // 任务是否失败
  failureReason?: string;      // 失败原因
}

interface PerceptionRecord {
  cycle: number;
  timestamp: number;
  perception: Record<string, any>;
  reasoning: string;
  action: AgentAction;
  result: ExecutionResult;
}

class DesktopAgent {
  private openai: OpenAI;
  private browser: Browser;
  private state: AgentState;

  constructor(config: { openaiApiKey: string; browser: Browser }) {
    this.openai = new OpenAI({ apiKey: config.openaiApiKey });
    this.browser = config.browser;
    this.state = {
      history: [],
      currentCycle: 0,
      taskCompleted: false,
      taskFailed: false,
    };
  }

  async run(instruction: string, config: PRELoopConfig): Promise<AgentState> {
    console.log(`[Agent] 开始执行任务: "${instruction}"`);

    while (!this.state.taskCompleted && !this.state.taskFailed) {
      this.state.currentCycle++;

      if (this.state.currentCycle > config.maxCycles) {
        this.state.taskFailed = true;
        this.state.failureReason = `达到最大循环次数 (${config.maxCycles})`;
        break;
      }

      // 步骤 1: 感知 (Perception)
      const perception = await this.perceive();
      console.log(`[Agent] Cycle ${this.state.currentCycle}: 感知完成`);

      // 步骤 2: 推理 (Reasoning)
      const reasoning = await this.reason(
        instruction,
        perception,
        this.state.history
      );
      console.log(`[Agent] Cycle ${this.state.currentCycle}: 推理完成`);

      // 步骤 3: 执行 (Execution)
      const result = await this.execute(reasoning.action, config);

      // 记录完整的历史
      this.state.history.push({
        cycle: this.state.currentCycle,
        timestamp: Date.now(),
        perception,
        reasoning: reasoning.summary,
        action: reasoning.action,
        result,
      });

      // 判断任务状态
      if (reasoning.taskComplete) {
        this.state.taskCompleted = true;
        console.log(`[Agent] 任务完成（${this.state.currentCycle} 次循环）`);
      } else if (result.error && !reasoning.canRecover) {
        this.state.taskFailed = true;
        this.state.failureReason = result.error;
        console.log(`[Agent] 任务失败: ${result.error}`);
      }

      // 循环间隔
      await new Promise(r => setTimeout(r, config.cycleDelayMs));
    }

    return this.state;
  }

  private async perceive(): Promise<Record<string, any>> {
    // 简化版：获取页面标题和 URL
    const pages = await this.browser.pages();
    if (pages.length === 0) return { empty: true };
    const page = pages[0];
    return {
      url: page.url(),
      title: await page.title(),
      // 实际实现应包含更全面的 DOM 感知
    };
  }

  private async reason(
    instruction: string,
    perception: Record<string, any>,
    history: PerceptionRecord[]
  ): Promise<{ summary: string; action: AgentAction; taskComplete: boolean; canRecover: boolean }> {
    // 使用 LLM 进行推理
    const response = await this.openai.chat.completions.create({
      model: "gpt-4o",
      messages: [
        { role: "system", content: "你是一个桌面 Agent。根据当前页面状态和用户指令，决定下一步行动。以JSON格式返回。" },
        { role: "user", content: JSON.stringify({ instruction, perception, history: history.slice(-3) }) }
      ],
      response_format: { type: "json_object" }
    });
    return JSON.parse(response.choices[0].message.content!);
  }

  private async execute(action: AgentAction, config: PRELoopConfig): Promise<ExecutionResult> {
    // 危险操作需要确认
    if (config.dangerousActionsRequireConfirmation && this.isDangerous(action)) {
      // 实际实现中这里应该请求用户确认
      console.warn(`[Agent] 危险操作待确认: ${JSON.stringify(action)}`);
    }

    // 执行行动...
    return { success: true };
  }

  private isDangerous(action: AgentAction): boolean {
    // 定义危险操作的标准
    return ["delete", "send_email", "submit_payment"].includes(action.type);
  }
}