Computer Use 实战：让 AI Agent 操控浏览器与操作系统

用户指令 → [LLM 推理] → 调用 computer 工具(screenshot)
                                    ↓
                            截取当前屏幕 → Base64 图像
                                    ↓
                         [LLM 视觉推理] 分析屏幕内容
                                    ↓
                      生成操作指令(click/type/key/scroll)
                                    ↓
                        在操作系统层面执行操作 → 界面状态变化
                                    ↓
                         再次截图 → 回到视觉推理步骤
                                    ↓
                    判断任务是否完成？→ 是 → 返回结果
                                     → 否 → 继续循环

import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=4096,
    tools=[
        {
            "type": "computer_20250124",
            "name": "computer",
            "display_width_px": 1920,
            "display_height_px": 1080,
            "display_number": 1,
        },
        {
            "type": "text_editor_20250124",
            "name": "str_replace_editor",
        },
        {
            "type": "bash_20250124",
            "name": "bash",
        },
    ],
    messages=[
        {
            "role": "user",
            "content": "打开 Firefox 浏览器，访问 qubittool.com，截图给我看首页内容。",
        }
    ],
)

{
  "type": "tool_use",
  "name": "computer",
  "input": {
    "action": "mouse_move",
    "coordinate": [960, 540]
  }
}

# 推荐的屏幕分辨率配置
# Anthropic 建议使用 XGA（1024x768）以获得最佳性能
RECOMMENDED_RESOLUTION = (1024, 768)

# 如果实际屏幕分辨率更高，需要进行缩放
def scale_coordinates(x, y, actual_width, actual_height):
    scale_x = actual_width / RECOMMENDED_RESOLUTION[0]
    scale_y = actual_height / RECOMMENDED_RESOLUTION[1]
    return int(x * scale_x), int(y * scale_y)

import { chromium, Browser, Page } from 'playwright';
import Anthropic from '@anthropic-ai/sdk';

interface StepResult {
  success: boolean;
  screenshot?: string;
}

class HybridAgent {
  private browser: Browser;
  private page: Page;
  private client: Anthropic;

  async executeWorkflow(steps: WorkflowStep[]): Promise<void> {
    for (const step of steps) {
      if (step.type === 'deterministic') {
        // 确定性步骤：用 Playwright 精确执行
        await this.playwrightExecute(step);
      } else if (step.type === 'visual') {
        // 模糊步骤：用 Computer Use 视觉推理
        await this.computerUseExecute(step);
      }
    }
  }

  private async playwrightExecute(step: WorkflowStep): Promise<StepResult> {
    // 直接用选择器操作，速度快且确定
    await this.page.goto(step.url);
    await this.page.fill(step.selector, step.value);
    await this.page.click(step.submitSelector);
    return { success: true };
  }

  private async computerUseExecute(step: WorkflowStep): Promise<StepResult> {
    // 截图当前页面
    const screenshot = await this.page.screenshot({
      type: 'png',
      encoding: 'base64',
    });

    // 发送给 Claude 进行视觉推理
    const response = await this.client.messages.create({
      model: 'claude-sonnet-4-20250514',
      max_tokens: 1024,
      messages: [
        {
          role: 'user',
          content: [
            {
              type: 'image',
              source: {
                type: 'base64',
                media_type: 'image/png',
                data: screenshot,
              },
            },
            {
              type: 'text',
              text: step.instruction,
            },
          ],
        },
      ],
    });

    // 解析模型返回的坐标，用 Playwright 执行点击
    const action = this.parseAction(response);
    if (action.type === 'click') {
      await this.page.mouse.click(action.x, action.y);
    } else if (action.type === 'type') {
      await this.page.keyboard.type(action.text);
    }

    return { success: true };
  }
}

# Computer Use 专用沙箱
FROM ubuntu:22.04

# 安装虚拟桌面环境
RUN apt-get update && apt-get install -y \
    xvfb \
    x11vnc \
    fluxbox \
    firefox-esr \
    xdotool \
    && rm -rf /var/lib/apt/lists/*

# 创建受限用户
RUN useradd -m -s /bin/bash agent && \
    echo "agent ALL=(ALL) NOPASSWD: /usr/bin/apt-get" >> /etc/sudoers

# 限制网络访问（仅允许特定域名）
COPY iptables-rules.sh /etc/init.d/

# 启动虚拟桌面
ENV DISPLAY=:99
CMD ["bash", "-c", "Xvfb :99 -screen 0 1024x768x24 & fluxbox & x11vnc -display :99 -forever -nopw & python3 /app/agent.py"]

class SafetyGuard:
    # 禁止访问的 URL 模式
    BLOCKED_URLS = [
        r".*bank.*\.com",
        r".*paypal\.com",
        r".*\.gov",
        r"chrome://settings/passwords",
    ]

    # 危险操作拦截
    BLOCKED_ACTIONS = [
        "rm -rf",
        "format",
        "DROP TABLE",
        "DELETE FROM",
        "sudo rm",
    ]

    # 操作频率限制
    MAX_ACTIONS_PER_MINUTE = 30
    MAX_TOTAL_ACTIONS = 200

    def validate_action(self, action: dict) -> bool:
        # 检查是否触发了安全规则
        if action.get("action") == "type":
            text = action.get("text", "")
            for pattern in self.BLOCKED_ACTIONS:
                if pattern.lower() in text.lower():
                    return False

        # 检查操作频率
        if self.action_count_last_minute > self.MAX_ACTIONS_PER_MINUTE:
            return False

        return True

    def require_human_confirmation(self, action: dict) -> bool:
        # 不可逆操作需要人工确认
        irreversible_patterns = ["submit", "confirm", "delete", "send", "pay"]
        if action.get("action") == "click":
            # 如果即将点击的按钮文字包含危险关键词，要求确认
            return any(p in str(action).lower() for p in irreversible_patterns)
        return False

async def visual_e2e_test(page_url: str, test_scenario: str):
    """
    用自然语言描述测试意图，让 Computer Use 执行并验证
    """
    system_prompt = """你是一个 QA 测试工程师。根据测试场景描述执行操作，
    并在每一步验证界面状态是否符合预期。如果发现异常，立即报告。
    不要点击任何与测试无关的元素。"""

    messages = [
        {
            "role": "user",
            "content": f"请在 {page_url} 上执行以下测试场景：\n{test_scenario}"
        }
    ]

    # 执行截图-操作循环，直到测试完成或失败
    max_steps = 20
    for step in range(max_steps):
        response = await client.messages.create(
            model="claude-sonnet-4-20250514",
            max_tokens=2048,
            system=system_prompt,
            tools=[computer_tool],
            messages=messages,
        )

        # 处理工具调用结果
        tool_result = await execute_tool_action(response)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_result})

        # 检查模型是否报告测试完成
        if is_task_complete(response):
            return parse_test_result(response)

    return {"status": "timeout", "steps": max_steps}

async def legacy_data_entry(records: list[dict]):
    """批量向遗留 ERP 系统录入数据"""
    for i, record in enumerate(records):
        instruction = f"""
        在当前的 ERP 录入界面上：
        1. 在"客户名称"字段输入：{record['name']}
        2. 在"联系电话"字段输入：{record['phone']}
        3. 在"订单金额"字段输入：{record['amount']}
        4. 点击"保存"按钮
        5. 等待出现"保存成功"提示后，点击"新增"按钮准备下一条
        如果出现错误提示，请截图并报告具体错误信息。
        """
        result = await computer_use_execute(instruction)

        if not result.success:
            log.error(f"第 {i+1} 条记录录入失败: {result.error}")
            # 失败时暂停，等待人工介入
            await notify_human(record, result.screenshot)
            break

SYSTEM_PROMPT = """你是一个计算机操作 Agent。请遵循以下规则：
1. 每次操作前先截图确认当前界面状态
2. 点击前确认目标元素可见且可交互
3. 输入文字前先点击对应的输入框确保获得焦点
4. 如果遇到意外弹窗，优先关闭弹窗再继续任务
5. 如果连续 3 次操作未产生预期效果，停止并报告问题
6. 不要访问与任务无关的网站
7. 不要在任何地方输入密码或敏感信息"""

class CheckpointManager:
    def __init__(self):
        self.checkpoints = []

    async def save_checkpoint(self, page, step_name: str):
        screenshot = await page.screenshot()
        url = page.url
        self.checkpoints.append({
            "step": step_name,
            "url": url,
            "screenshot": screenshot,
            "timestamp": time.time(),
        })

    async def rollback_to(self, page, step_name: str):
        checkpoint = next(
            c for c in self.checkpoints if c["step"] == step_name
        )
        await page.goto(checkpoint["url"])