# Computer Use Agents > 构建像人类一样与计算机交互的 AI 智能体 - 查看屏幕、移动光标、点击按钮和输入文本。涵盖 Anthropic 的 Computer Use、OpenAI 的 Operator/CUA 以及开源替代方案。 - Skill: `kscz0000/computer-use-agents` (Agent Skill) - Install (CLI): `npx skillmds@latest add kscz0000/computer-use-agents` - Raw SKILL.md: https://api.skillmd.com/api/skills/kscz0000/computer-use-agents/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - Author: kscz0000 (https://skillmd.com/u/kscz0000) - Updated: 2026-09-21 - Page: https://skillmd.com/skills/kscz0000/computer-use-agents --- # 计算机使用智能体 构建像人类一样与计算机交互的 AI 智能体 - 查看屏幕、移动光标、点击按钮和输入文本。涵盖 Anthropic 的 Computer Use、OpenAI 的 Operator/CUA 以及开源替代方案。重点关注沙箱化、安全性以及处理基于视觉控制的独特挑战。 ## 模式 ### 感知-推理-行动循环 计算机使用智能体的基础架构：观察屏幕，推理下一步行动，执行行动，重复。该循环通过迭代管道将视觉模型与行动执行集成。 关键组件： 1. 感知：截图捕获当前屏幕状态 2. 推理：视觉-语言模型分析并规划 3. 行动：执行鼠标/键盘操作 4. 反馈：观察结果，继续或纠正 关键洞察：视觉智能体在"思考"阶段（1-5秒）完全静止，形成可检测的暂停模式。 **何时使用**：从零构建任何计算机使用智能体，将视觉模型与桌面控制集成，理解智能体行为模式 from anthropic import Anthropic from PIL import Image import base64 import pyautogui import time class ComputerUseAgent: """ 感知-推理-行动循环实现。 基于 Anthropic Computer Use 模式。 """ def __init__(self, client: Anthropic, model: str = "claude-sonnet-4-20250514"): self.client = client self.model = model self.max_steps = 50 # 防止无限循环 self.action_delay = 0.5 # 操作间隔秒数 def capture_screenshot(self) -> str: """捕获屏幕并返回 base64 编码图像。""" screenshot = pyautogui.screenshot() # 调整大小以节省 token（1280x800 是较好的平衡） screenshot = screenshot.resize((1280, 800), Image.LANCZOS) import io buffer = io.BytesIO() screenshot.save(buffer, format="PNG") return base64.b64encode(buffer.getvalue()).decode() def execute_action(self, action: dict) -> dict: """在计算机上执行鼠标/键盘操作。""" action_type = action.get("type") if action_type == "click": x, y = action["x"], action["y"] button = action.get("button", "left") pyautogui.click(x, y, button=button) return {"success": True, "action": f"clicked at ({x}, {y})"} elif action_type == "type": text = action["text"] pyautogui.typewrite(text, interval=0.02) return {"success": True, "action": f"typed {len(text)} chars"} elif action_type == "key": key = action["key"] pyautogui.press(key) return {"success": True, "action": f"pressed {key}"} elif action_type == "scroll": direction = action.get("direction", "down") amount = action.get("amount", 3) scroll = -amount if direction == "down" else amount pyautogui.scroll(scroll) return {"success": True, "action": f"scrolled {direction}"} elif action_type == "move": x, y = action["x"], action["y"] pyautogui.moveTo(x, y) return {"success": True, "action": f"moved to ({x}, {y})"} else: return {"success": False, "error": f"Unknown action: {action_type}"} def run(self, task: str) -> dict: """ 运行感知-推理-行动循环直到任务完成。 循环流程： 1. 截图当前状态 2. 发送至视觉模型并附带任务上下文 3. 从响应中解析行动 4. 执行行动 5. 重复直到完成或达到最大步数 """ messages = [] step_count = 0 system_prompt = """你是一个计算机使用智能体。你可以看到屏幕 并控制鼠标/键盘。 可用操作（以 JSON 响应）： - {"type": "click", "x": 100, "y": 200, "button": "left"} - {"type": "type", "text": "hello world"} - {"type": "key", "key": "enter"} - {"type": "scroll", "direction": "down", "amount": 3} - {"type": "done", "result": "task completed successfully"} 始终仅以 JSON 操作对象响应。 坐标要精确 - 准确点击所需位置。 如果发现错误，尝试恢复。 """ while step_count < self.max_steps: step_count += 1 # 1. 感知：捕获当前屏幕 screenshot_b64 = self.capture_screenshot() # 2. 推理：发送至视觉模型 user_content = [ {"type": "text", "text": f"Task: {task}\n\nStep {step_count}. What action should I take?"}, {"type": "image", "source": { "type": "base64", "media_type": "image/png", "data": screenshot_b64 }} ] messages.append({"role": "user", "content": user_content}) response = self.client.messages.create( model=self.model, max_tokens=1024, system=system_prompt, messages=messages ) assistant_message = response.content[0].text messages.append({"role": "assistant", "content": assistant_message}) # 3. 从响应中解析行动 import json try: action = json.loads(assistant_message) except json.JSONDecodeError: # 尝试从响应中提取 JSON import re match = re.search(r'\{[^}]+\}', assistant_message) if match: action = json.loads(match.group()) else: continue # 检查是否完成 if action.get("type") == "done": return { "success": True, "result": action.get("result"), "steps": step_count } # 4. 行动：执行 result = self.execute_action(action) # 短暂延迟等待 UI 更新 time.sleep(self.action_delay) return { "success": False, "error": "Max steps reached", "steps": step_count } # 使用示例 agent = ComputerUseAgent(Anthropic()) result = agent.run("Open Chrome and search for 'weather today'") ### 反模式 - 不设步数限制运行（无限循环） - 操作之间无延迟（UI 来不及响应） - 全分辨率截图（token 爆炸） - 忽略操作失败（无恢复机制） ### 沙箱化环境模式 计算机使用智能体必须在隔离的沙箱环境中运行。 切勿让智能体直接访问你的主系统 - 安全风险太高。使用带有虚拟桌面的 Docker 容器。 关键隔离要求： 1. 网络：仅限必要的端点 2. 文件系统：只读或限定在临时目录 3. 凭证：无法访问主机凭证 4. 系统调用：过滤危险系统调用 5. 资源：限制 CPU、内存、时间 目标是"爆炸半径最小化" - 如果智能体出错，损害仅限于沙箱内。 **何时使用**：部署任何计算机使用智能体，安全测试智能体行为，运行不受信任的自动化任务 # 沙箱化计算机使用环境的 Dockerfile # 基于 Anthropic 参考实现模式 FROM ubuntu:22.04 # 安装桌面环境 RUN apt-get update && apt-get install -y \ xvfb \ x11vnc \ fluxbox \ xterm \ firefox \ python3 \ python3-pip \ supervisor # 安全：创建非 root 用户 RUN useradd -m -s /bin/bash agent && \ mkdir -p /home/agent/.vnc # 安装 Python 依赖 COPY requirements.txt /tmp/ RUN pip3 install -r /tmp/requirements.txt # 安全：移除能力 RUN apt-get install -y --no-install-recommends libcap2-bin && \ setcap -r /usr/bin/python3 || true # 复制智能体代码 COPY --chown=agent:agent . /app WORKDIR /app # 虚拟显示器 + VNC 的 Supervisor 配置 COPY supervisord.conf /etc/supervisor/conf.d/ # 仅暴露 VNC 端口（不直接暴露桌面） EXPOSE 5900 # 以非 root 用户运行 USER agent CMD ["/usr/bin/supervisord", "-c", "/etc/supervisor/conf.d/supervisord.conf"] --- # 带安全约束的 docker-compose.yml version: '3.8' services: computer-use-agent: build: . ports: - "5900:5900" # VNC 用于观察 - "8080:8080" # API 用于控制 # 安全约束 security_opt: - no-new-privileges:true - seccomp:seccomp-profile.json # 资源限制 deploy: resources: limits: cpus: '2' memory: 4G reservations: cpus: '0.5' memory: 1G # 网络隔离 networks: - agent-network # 无法访问主机文件系统 volumes: - agent-tmp:/tmp # 只读根文件系统 read_only: true tmpfs: - /run - /var/run # 环境变量 environment: - DISPLAY=:99 - NO_PROXY=localhost networks: agent-network: driver: bridge internal: true # 默认无互联网访问 volumes: agent-tmp: --- # 带额外运行时沙箱化的 Python 封装 import subprocess import os from dataclasses import dataclass from typing import Optional @dataclass class SandboxConfig: """智能体沙箱配置。""" network_allowed: list[str] = None # 允许的域名 max_runtime_seconds: int = 300 max_memory_mb: int = 2048 allow_downloads: bool = False allow_clipboard: bool = False class SandboxedAgent: """ 在 Docker 沙箱中运行计算机使用智能体。 """ def __init__(self, config: SandboxConfig): self.config = config self.container_id: Optional[str] = None def start(self): """启动沙箱化环境。""" # 构建网络规则 network_rules = "" if self.config.network_allowed: for domain in self.config.network_allowed: network_rules += f"--add-host={domain}:$(dig +short {domain}) " else: network_rules = "--network=none" cmd = f""" docker run -d \ --name computer-use-sandbox-$$ \ --security-opt no-new-privileges \ --cap-drop ALL \ --memory {self.config.max_memory_mb}m \ --cpus 2 \ --read-only \ --tmpfs /tmp \ {network_rules} \ computer-use-agent:latest """ result = subprocess.run(cmd, shell=True, capture_output=True) self.container_id = result.stdout.decode().strip() # 设置终止计时器 subprocess.Popen([ "sh", "-c", f"sleep {self.config.max_runtime_seconds} && docker kill {self.container_id}" ]) return self.container_id def execute_task(self, task: str) -> dict: """在沙箱中执行任务。""" if not self.container_id: self.start() # 通过 API 发送任务给智能体 import requests response = requests.post( f"http://localhost:8080/task", json={"task": task}, timeout=self.config.max_runtime_seconds ) return response.json() def stop(self): """停止并移除沙箱。""" if self.container_id: subprocess.run(f"docker rm -f {self.container_id}", shell=True) self.container_id = None ### 反模式 - 直接在主机系统上运行智能体 - 给沙箱完整的网络访问权限 - 在容器中以 root 运行 - 无资源限制（拒绝服务风险） - 持久化存储（数据可能在运行间泄露） ### Anthropic Computer Use 实现 使用 Claude 计算机使用能力的官方实现模式。 Claude 3.5 Sonnet 是首个提供计算机使用能力的前沿模型。 Claude Opus 4.5 现在是"世界上最好的计算机使用模型"。 关键能力： - screenshot：捕获当前屏幕状态 - mouse：点击、移动、拖拽操作 - keyboard：输入文本、按键 - bash：运行 shell 命令 - text_editor：查看和编辑文件 工具版本： - computer_20251124（Opus 4.5）：新增缩放操作用于详细检查 - computer_20250124（所有其他模型）：标准能力 关键限制："某些 UI 元素（如下拉菜单和滚动条）对 Claude 来说可能难以操作" - Anthropic 文档 **何时使用**：构建生产级计算机使用智能体，需要最高质量的视觉理解，完整的桌面控制（不仅是浏览器） from anthropic import Anthropic from anthropic.types.beta import ( BetaToolComputerUse20241022, BetaToolBash20241022, BetaToolTextEditor20241022, ) import subprocess import base64 from PIL import Image import io class AnthropicComputerUse: """ 官方 Anthropic Computer Use 实现。 需要： - 带虚拟显示器的 Docker 容器 - 用于查看智能体操作的 VNC - 正确的工具实现 """ def __init__(self): self.client = Anthropic() self.model = "claude-sonnet-4-20250514" # 最适合计算机使用 self.screen_size = (1280, 800) def get_tools(self) -> list: """定义计算机使用工具。""" return [ BetaToolComputerUse20241022( type="computer_20241022", name="computer", display_width_px=self.screen_size[0], display_height_px=self.screen_size[1], ), BetaToolBash20241022( type="bash_20241022", name="bash", ), BetaToolTextEditor20241022( type="text_editor_20241022", name="str_replace_editor", ), ] def execute_tool(self, name: str, input: dict) -> dict: """执行工具并返回结果。""" if name == "computer": return self._handle_computer_action(input) elif name == "bash": return self._handle_bash(input) elif name == "str_replace_editor": return self._handle_editor(input) else: return {"error": f"Unknown tool: {name}"} def _handle_computer_action(self, input: dict) -> dict: """处理计算机控制操作。""" action = input.get("action") if action == "screenshot": # 通过 xdotool/scrot 捕获 subprocess.run(["scrot", "/tmp/screenshot.png"]) with open("/tmp/screenshot.png", "rb") as f: img_data = f.read() # 调整大小以提高效率 img = Image.open(io.BytesIO(img_data)) img = img.resize(self.screen_size, Image.LANCZOS) buffer = io.BytesIO() img.save(buffer, format="PNG") return { "type": "image", "source": { "type": "base64", "media_type": "image/png", "data": base64.b64encode(buffer.getvalue()).decode() } } elif action == "mouse_move": x, y = input.get("coordinate", [0, 0]) subprocess.run(["xdotool", "mousemove", str(x), str(y)]) return {"success": True} elif action == "left_click": subprocess.run(["xdotool", "click", "1"]) return {"success": True} elif action == "right_click": subprocess.run(["xdotool", "click", "3"]) return {"success": True} elif action == "double_click": subprocess.run(["xdotool", "click", "--repeat", "2", "1"]) return {"success": True} elif action == "type": text = input.get("text", "") # 使用 xdotool type 并加延迟以提高可靠性 subprocess.run(["xdotool", "type", "--delay", "50", text]) return {"success": True} elif action == "key": key = input.get("key", "") # 映射常见键名 key_map = { "return": "Return", "enter": "Return", "tab": "Tab", "escape": "Escape", "backspace": "BackSpace", } xdotool_key = key_map.get(key.lower(), key) subprocess.run(["xdotool", "key", xdotool_key]) return {"success": True} elif action == "scroll": direction = input.get("direction", "down") amount = input.get("amount", 3) button = "5" if direction == "down" else "4" for _ in range(amount): subprocess.run(["xdotool", "click", button]) return {"success": True} return {"error": f"Unknown action: {action}"} def _handle_bash(self, input: dict) -> dict: """执行 bash 命令。""" command = input.get("command", "") # 安全：清理和限制命令 dangerous_patterns = ["rm -rf", "mkfs", "dd if=", "> /dev/"] for pattern in dangerous_patterns: if pattern in command: return {"error": "Dangerous command blocked"} try: result = subprocess.run( command, shell=True, capture_output=True, text=True, timeout=30 ) return { "stdout": result.stdout[:10000], # 限制输出 "stderr": result.stderr[:1000], "returncode": result.returncode } except subprocess.TimeoutExpired: return {"error": "Command timed out"} def _handle_editor(self, input: dict) -> dict: """处理文本编辑器操作。""" command = input.get("command") path = input.get("path") if command == "view": try: with open(path, "r") as f: content = f.read() return {"content": content[:50000]} # 限制大小 except Exception as e: return {"error": str(e)} elif command == "str_replace": old_str = input.get("old_str") new_str = input.get("new_str") try: with open(path, "r") as f: content = f.read() if old_str not in content: return {"error": "old_str not found in file"} content = content.replace(old_str, new_str, 1) with open(path, "w") as f: f.write(content) return {"success": True} except Exception as e: return {"error": str(e)} return {"error": f"Unknown editor command: {command}"} def run_task(self, task: str, max_steps: int = 50) -> dict: """以智能体循环运行计算机使用任务。""" messages = [{"role": "user", "content": task}] tools = self.get_tools() for step in range(max_steps): response = self.client.beta.messages.create( model=self.model, max_tokens=4096, tools=tools, messages=messages, betas=["computer-use-2024-10-22"] ) # 检查是否完成 if response.stop_reason == "end_turn": return { "success": True, "result": response.content[0].text if response.content else "", "steps": step + 1 } # 处理工具使用 if response.stop_reason == "tool_use": messages.append({"role": "assistant", "content": response.content}) tool_results = [] for block in response.content: if block.type == "tool_use": result = self.execute_tool(block.name, block.input) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": result }) messages.append({"role": "user", "content": tool_results}) return {"success": False, "error": "Max steps reached"} ### 反模式 - 不使用 betas=['computer-use-2024-10-22'] 标志 - 全分辨率截图（浪费资源） - bash 工具无命令清理 - 无限执行时间 ### 浏览器使用模式（基于 Playwright） 对于仅浏览器的自动化，使用结构化 DOM 访问比基于像素的计算机使用更高效。Playwright MCP 允许 LLM 使用无障碍快照而非截图来控制浏览器。 相比基于视觉的优势： - 更快：无需图像处理 - 更便宜：文本 token vs 图像 token - 更精确：直接元素定位 - 更可靠：无坐标漂移 何时使用视觉 vs 结构化： - 视觉：桌面应用、复杂 UI、视觉验证 - 结构化：Web 自动化、表单填写、数据提取 **何时使用**：仅浏览器自动化任务，表单填写和 Web 交互，速度和成本比视觉理解更重要时 from playwright.async_api import async_playwright from dataclasses import dataclass from typing import Optional import asyncio @dataclass class BrowserAction: """结构化浏览器操作。""" action: str # click, type, navigate, scroll, extract selector: Optional[str] = None text: Optional[str] = None url: Optional[str] = None class BrowserUseAgent: """ 使用 Playwright 和结构化命令的浏览器自动化。 对于 Web 任务比基于像素的方式更高效。 """ def __init__(self): self.browser = None self.page = None async def start(self, headless: bool = True): """启动浏览器会话。""" self.playwright = await async_playwright().start() self.browser = await self.playwright.chromium.launch(headless=headless) self.page = await self.browser.new_page() async def get_page_snapshot(self) -> dict: """ 获取页面的结构化快照供 LLM 使用。 使用无障碍树提高效率。 """ # 获取无障碍树 snapshot = await self.page.accessibility.snapshot() # 获取简化的 DOM 信息 elements = await self.page.evaluate('''() => { const interactable = []; const selector = 'a, button, input, select, textarea, [role="button"]'; document.querySelectorAll(selector).forEach((el, i) => { const rect = el.getBoundingClientRect(); if (rect.width > 0 && rect.height > 0) { interactable.push({ index: i, tag: el.tagName.toLowerCase(), text: el.textContent?.trim().slice(0, 100), type: el.type, placeholder: el.placeholder, name: el.name, id: el.id, class: el.className }); } }); return interactable; }''') return { "url": self.page.url, "title": await self.page.title(), "accessibility_tree": snapshot, "interactable_elements": elements[:50] # 限制以节省 token } async def execute_action(self, action: BrowserAction) -> dict: """执行结构化浏览器操作。""" try: if action.action == "navigate": await self.page.goto(action.url, wait_until="domcontentloaded") return {"success": True, "url": self.page.url} elif action.action == "click": await self.page.click(action.selector, timeout=5000) await self.page.wait_for_load_state("networkidle", timeout=5000) return {"success": True} elif action.action == "type": await self.page.fill(action.selector, action.text) return {"success": True} elif action.action == "scroll": direction = action.text or "down" distance = 500 if direction == "down" else -500 await self.page.evaluate(f"window.scrollBy(0, {distance})") return {"success": True} elif action.action == "extract": # 提取文本内容 if action.selector: text = await self.page.text_content(action.selector) else: text = await self.page.text_content("body") return {"success": True, "text": text[:5000]} elif action.action == "screenshot": # 需要时回退到视觉方式 screenshot = await self.page.screenshot(type="png") import base64 return { "success": True, "image": base64.b64encode(screenshot).decode() } except Exception as e: return {"success": False, "error": str(e)} return {"success": False, "error": f"Unknown action: {action.action}"} async def run_with_llm(self, task: str, llm_client, max_steps: int = 20): """ 使用 LLM 决策运行浏览器任务。 使用结构化 DOM 而非截图。 """ system_prompt = """你是一个浏览器自动化智能体。你接收 包含可交互元素的页面快照并决定操作。 以 JSON 操作响应： - {"action": "navigate", "url": "https://..."} - {"action": "click", "selector": "button.submit"} - {"action": "type", "selector": "input[name='email']", "text": "..."} - {"action": "scroll", "text": "down"} - {"action": "extract", "selector": ".results"} - {"action": "done", "result": "task completed"} 根据提供的元素信息使用 CSS 选择器。 选择器优先级：id > name > class > 文本内容。 """ messages = [] for step in range(max_steps): # 获取当前页面状态 snapshot = await self.get_page_snapshot() user_message = f"""Task: {task} Current page: URL: {snapshot['url']} Title: {snapshot['title']} Interactable elements: {snapshot['interactable_elements']} What action should I take?""" messages.append({"role": "user", "content": user_message}) # 获取 LLM 决策 response = llm_client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, system=system_prompt, messages=messages ) assistant_text = response.content[0].text messages.append({"role": "assistant", "content": assistant_text}) # 解析并执行 import json action_dict = json.loads(assistant_text) if action_dict.get("action") == "done": return {"success": True, "result": action_dict.get("result")} action = BrowserAction(**action_dict) result = await self.execute_action(action) if not result.get("success"): messages.append({ "role": "user", "content": f"Action failed: {result.get('error')}" }) await asyncio.sleep(0.5) # 速率限制 return {"success": False, "error": "Max steps reached"} async def close(self): """清理浏览器。""" if self.browser: await self.browser.close() if hasattr(self, 'playwright'): await self.playwright.stop() # 使用示例 async def main(): agent = BrowserUseAgent() await agent.start(headless=False) from anthropic import Anthropic result = await agent.run_with_llm( "Go to weather.com and find the weather for New York", Anthropic() ) print(result) await agent.close() asyncio.run(main()) ### 反模式 - DOM 访问可用时仍使用截图 - 不等待页面加载 - 使用会失效的硬编码选择器 - 对过期元素无错误恢复 ### 用户确认模式 对于敏感操作，智能体应暂停并请求人工确认。 "ChatGPT 智能体在执行购买等敏感步骤前也会暂停并请求确认。" 敏感级别： 1. 低：导航、读取（自动批准） 2. 中：表单填写、点击（记录日志，可能确认） 3. 高：购买、认证、文件操作（始终确认） 4. 关键：凭证输入、金融交易（确认 + 审查） **何时使用**：具有现实后果的操作，金融交易，认证流程，文件修改 from enum import Enum from dataclasses import dataclass from typing import Callable, Optional import asyncio class ActionSeverity(Enum): LOW = "low" # 自动批准 MEDIUM = "medium" # 记录日志，可选确认 HIGH = "high" # 始终确认 CRITICAL = "critical" # 确认 + 审查详情 @dataclass class SensitiveAction: """可能需要用户确认的操作。""" action_type: str description: str severity: ActionSeverity details: dict class ConfirmationGate: """ 通过用户确认来门控敏感操作。 """ # 操作类型 -> 敏感级别映射 ACTION_SEVERITY = { # 低 - 自动批准 "navigate": ActionSeverity.LOW, "scroll": ActionSeverity.LOW, "read": ActionSeverity.LOW, "screenshot": ActionSeverity.LOW, # 中 - 记录日志并可能确认 "click": ActionSeverity.MEDIUM, "type": ActionSeverity.MEDIUM, "search": ActionSeverity.MEDIUM, # 高 - 始终确认 "download": ActionSeverity.HIGH, "submit_form": ActionSeverity.HIGH, "login": ActionSeverity.HIGH, "file_write": ActionSeverity.HIGH, # 关键 - 确认并完整审查 "purchase": ActionSeverity.CRITICAL, "enter_password": ActionSeverity.CRITICAL, "enter_credit_card": ActionSeverity.CRITICAL, "send_money": ActionSeverity.CRITICAL, "delete": ActionSeverity.CRITICAL, } def __init__( self, confirm_callback: Callable[[SensitiveAction], bool] = None, auto_confirm_low: bool = True, auto_confirm_medium: bool = False ): self.confirm_callback = confirm_callback or self._default_confirm self.auto_confirm_low = auto_confirm_low self.auto_confirm_medium = auto_confirm_medium self.action_log = [] def _default_confirm(self, action: SensitiveAction) -> bool: """通过 CLI 提示进行默认确认。""" print(f"\n{'='*60}") print(f"ACTION CONFIRMATION REQUIRED") print(f"{'='*60}") print(f"Type: {action.action_type}") print(f"Severity: {action.severity.value.upper()}") print(f"Description: {action.description}") print(f"Details: {action.details}") print(f"{'='*60}") while True: response = input("Allow this action? [y/n]: ").lower().strip() if response in ['y', 'yes']: return True elif response in ['n', 'no']: return False def classify_action(self, action_type: str, context: dict) -> ActionSeverity: """分类操作敏感级别，考虑上下文。""" base_severity = self.ACTION_SEVERITY.get(action_type, ActionSeverity.MEDIUM) # 根据上下文升级 if context.get("involves_credentials"): return ActionSeverity.CRITICAL if context.get("involves_money"): return ActionSeverity.CRITICAL if context.get("irreversible"): return max(base_severity, ActionSeverity.HIGH, key=lambda x: x.value) return base_severity def check_action( self, action_type: str, description: str, details: dict = None ) -> tuple[bool, str]: """ 检查操作是否应继续。 返回 (是否批准, 原因)。 """ details = details or {} severity = self.classify_action(action_type, details) action = SensitiveAction( action_type=action_type, description=description, severity=severity, details=details ) # 记录所有操作 self.action_log.append({ "action": action, "timestamp": __import__('datetime').datetime.now().isoformat() }) # 自动批准低敏感级别 if severity == ActionSeverity.LOW and self.auto_confirm_low: return True, "auto-approved (low severity)" # 可能自动批准中敏感级别 if severity == ActionSeverity.MEDIUM and self.auto_confirm_medium: return True, "auto-approved (medium severity)" # 请求确认 approved = self.confirm_callback(action) if approved: return True, "user approved" else: return False, "user rejected" class ConfirmedComputerUseAgent: """ 带确认门控的计算机使用智能体。 """ def __init__(self, base_agent, confirmation_gate: ConfirmationGate): self.agent = base_agent self.gate = confirmation_gate def execute_action(self, action: dict) -> dict: """带确认检查的操作执行。""" action_type = action.get("type", "unknown") # 构建描述 if action_type == "click": desc = f"Click at ({action.get('x')}, {action.get('y')})" elif action_type == "type": text = action.get('text', '') # 如果看起来像密码则遮蔽 if self._looks_sensitive(text): desc = f"Type sensitive text ({len(text)} chars)" else: desc = f"Type: {text[:50]}..." else: desc = f"Execute: {action_type}" # 用于敏感级别分类的上下文 context = { "involves_credentials": self._looks_sensitive(action.get("text", "")), "involves_money": self._mentions_money(action), } # 通过门控检查 approved, reason = self.gate.check_action( action_type, desc, context ) if not approved: return { "success": False, "error": f"Action blocked: {reason}", "action": action_type } # 批准后执行 return self.agent.execute_action(action) def _looks_sensitive(self, text: str) -> bool: """检查文本是否看起来像敏感数据。""" if not text: return False # 常见模式 patterns = [ r'\b\d{16}\b', # 信用卡 r'\b\d{3,4}\b.*\b\d{3,4}\b', # 类似 CVV r'password', r'secret', r'api.?key', r'token' ] import re return any(re.search(p, text.lower()) for p in patterns) def _mentions_money(self, action: dict) -> bool: """检查操作是否涉及金钱。""" text = str(action) money_patterns = [ r'\$\d+', r'pay', r'purchase', r'buy', r'checkout', r'credit', r'debit', r'invoice', r'payment' ] import re return any(re.search(p, text.lower()) for p in money_patterns) # 使用示例 gate = ConfirmationGate( auto_confirm_low=True, auto_confirm_medium=False # 确认点击、输入等操作 ) agent = ConfirmedComputerUseAgent(base_agent, gate) result = agent.execute_action({"type": "click", "x": 500, "y": 300}) ### 反模式 - 自动批准所有操作 - 不记录被拒绝的操作 - 在确认中显示完整密码 - 确认无超时（永远挂起） ### 操作日志模式 所有计算机使用智能体的操作都应记录，用于： 1. 调试失败的自动化 2. 安全审计 3. 可复现性 4. 合规要求 日志格式应捕获： - 时间戳 - 操作类型和参数 - 操作前/后截图 - 成功/失败状态 - 模型推理（如可用） **何时使用**：生产级计算机使用部署，调试自动化失败，安全敏感环境 from dataclasses import dataclass, field from datetime import datetime from typing import Optional, Any import json import os @dataclass class ActionLogEntry: """单条操作日志条目。""" timestamp: datetime action_type: str parameters: dict success: bool error: Optional[str] = None screenshot_before: Optional[str] = None # 截图路径 screenshot_after: Optional[str] = None model_reasoning: Optional[str] = None duration_ms: Optional[int] = None def to_dict(self) -> dict: return { "timestamp": self.timestamp.isoformat(), "action_type": self.action_type, "parameters": self._sanitize_params(self.parameters), "success": self.success, "error": self.error, "screenshot_before": self.screenshot_before, "screenshot_after": self.screenshot_after, "model_reasoning": self.model_reasoning, "duration_ms": self.duration_ms } def _sanitize_params(self, params: dict) -> dict: """从参数中移除敏感数据。""" sanitized = {} sensitive_keys = ['password', 'secret', 'token', 'key', 'credit_card'] for k, v in params.items(): if any(s in k.lower() for s in sensitive_keys): sanitized[k] = "[REDACTED]" elif isinstance(v, str) and len(v) > 100: sanitized[k] = v[:100] + "...[truncated]" else: sanitized[k] = v return sanitized @dataclass class TaskSession: """完整的任务执行会话。""" session_id: str task: str start_time: datetime end_time: Optional[datetime] = None actions: list[ActionLogEntry] = field(default_factory=list) success: bool = False final_result: Optional[str] = None class ActionLogger: """ 计算机使用智能体的全面操作日志记录。 """ def __init__(self, log_dir: str = "./agent_logs"): self.log_dir = log_dir self.screenshot_dir = os.path.join(log_dir, "screenshots") os.makedirs(self.screenshot_dir, exist_ok=True) self.current_session: Optional[TaskSession] = None def start_session(self, task: str) -> str: """启动新的任务会话。""" import uuid session_id = str(uuid.uuid4())[:8] self.current_session = TaskSession( session_id=session_id, task=task, start_time=datetime.now() ) return session_id def log_action( self, action_type: str, parameters: dict, success: bool, error: Optional[str] = None, screenshot_before: bytes = None, screenshot_after: bytes = None, model_reasoning: str = None, duration_ms: int = None ): """记录单条操作。""" if not self.current_session: raise RuntimeError("No active session") # 保存截图（如提供） screenshot_paths = {} timestamp_str = datetime.now().strftime("%Y%m%d_%H%M%S_%f") if screenshot_before: path = os.path.join( self.screenshot_dir, f"{self.current_session.session_id}_{timestamp_str}_before.png" ) with open(path, "wb") as f: f.write(screenshot_before) screenshot_paths["before"] = path if screenshot_after: path = os.path.join( self.screenshot_dir, f"{self.current_session.session_id}_{timestamp_str}_after.png" ) with open(path, "wb") as f: f.write(screenshot_after) screenshot_paths["after"] = path # 创建日志条目 entry = ActionLogEntry( timestamp=datetime.now(), action_type=action_type, parameters=parameters, success=success, error=error, screenshot_before=screenshot_paths.get("before"), screenshot_after=screenshot_paths.get("after"), model_reasoning=model_reasoning, duration_ms=duration_ms ) self.current_session.actions.append(entry) # 同时追加到运行中的日志文件 self._append_to_log(entry) def _append_to_log(self, entry: ActionLogEntry): """追加条目到 JSONL 日志文件。""" log_file = os.path.join( self.log_dir, f"session_{self.current_session.session_id}.jsonl" ) with open(log_file, "a") as f: f.write(json.dumps(entry.to_dict()) + "\n") def end_session(self, success: bool, result: str = None): """结束当前会话。""" if not self.current_session: return self.current_session.end_time = datetime.now() self.current_session.success = success self.current_session.final_result = result # 写入会话摘要 summary_file = os.path.join( self.log_dir, f"session_{self.current_session.session_id}_summary.json" ) summary = { "session_id": self.current_session.session_id, "task": self.current_session.task, "start_time": self.current_session.start_time.isoformat(), "end_time": self.current_session.end_time.isoformat(), "duration_seconds": ( self.current_session.end_time - self.current_session.start_time ).total_seconds(), "total_actions": len(self.current_session.actions), "successful_actions": sum( 1 for a in self.current_session.actions if a.success ), "failed_actions": sum( 1 for a in self.current_session.actions if not a.success ), "success": success, "final_result": result } with open(summary_file, "w") as f: json.dump(summary, f, indent=2) self.current_session = None def get_session_replay(self, session_id: str) -> list[dict]: """获取会话的所有操作用于回放/调试。""" log_file = os.path.join(self.log_dir, f"session_{session_id}.jsonl") actions = [] with open(log_file, "r") as f: for line in f: actions.append(json.loads(line)) return actions # 与智能体集成 class LoggedComputerUseAgent: """带全面日志记录的计算机使用智能体。""" def __init__(self, base_agent, logger: ActionLogger): self.agent = base_agent self.logger = logger def run_task(self, task: str) -> dict: """带完整日志记录运行任务。""" session_id = self.logger.start_session(task) try: result = self._run_with_logging(task) self.logger.end_session( success=result.get("success", False), result=result.get("result") ) return result except Exception as e: self.logger.end_session(success=False, result=str(e)) raise def _run_with_logging(self, task: str) -> dict: """带操作日志记录的内部运行。""" # 这将封装基础智能体的 run 方法 # 并记录每个操作 pass ### 反模式 - 日志中未清理敏感数据 - 无限期存储截图（存储成本） - 不轮转日志文件 - 同步日志记录（阻塞智能体） ## 关键陷阱 ### Web 内容可以劫持你的智能体 严重程度：关键 场景：计算机使用智能体浏览网页 症状： 智能体突然执行意外操作。点击恶意链接。在钓鱼网站上输入凭证。下载不该下载的文件。忽略你的指令而遵循嵌入的命令。 为什么会出问题： "虽然所有处理不受信任内容的智能体都面临提示注入风险，但浏览器使用以两种方式放大了这一风险。首先，攻击面非常广泛：每个网页、嵌入的文档、广告和动态加载的脚本都代表潜在的恶意指令载体。其次，浏览器智能体可以执行许多不同的操作——导航到 URL、填写表单、点击按钮、下载文件——攻击者可以利用这些操作。" 实际攻击已经发生： - "Microsoft Copilot 智能体被包含恶意指令的电子邮件劫持，攻击者借此提取了整个 CRM 数据库。" - "Google 的 Workspace 服务被操纵——日历邀请和电子邮件中隐藏的提示欺骗了 Gemini 智能体删除事件和暴露敏感消息。" 即使 1% 的攻击成功率在大规模下也意味着重大风险。 推荐修复方案： ## 纵深防御 - 没有单一解决方案有效 1. 沙箱化（最有效）： ```python # 带严格隔离的 Docker docker run \ --security-opt no-new-privileges \ --cap-drop ALL \ --network none \ # 无互联网！ --read-only \ computer-use-agent ``` 2. 基于分类器的检测： ```python def scan_for_injection(content: str) -> bool: """检测提示注入尝试。""" patterns = [ r"ignore.*instructions", r"disregard.*previous", r"new.*instructions", r"you are now", r"act as if", r"pretend to be", ] return any(re.search(p, content.lower()) for p in patterns) # 处理前检查页面内容 page_text = await page.text_content("body") if scan_for_injection(page_text): return {"error": "Potential injection detected"} ``` 3. 敏感操作的用户确认： ```python SENSITIVE_ACTIONS = {"download", "submit", "login", "purchase"} if action_type in SENSITIVE_ACTIONS: if not await get_user_confirmation(action): return {"error": "User rejected action"} ``` 4. 限定范围凭证： - 永远不要给智能体访问所有凭证的权限 - 使用临时的、有限制的令牌 - 任务完成后撤销 ### 视觉智能体点击精确中心 严重程度：中 场景：智能体点击 UI 元素 症状： 智能体的点击可被检测为非人类。网站可能阻止智能体或弹出 CAPTCHA。反机器人系统标记交互。 为什么会出问题： "当视觉模型识别一个按钮时，它计算中心点。点击坐标落在数学上精确的位置——通常是精确的元素中心或网格对齐的像素值。人类不会点击中心；他们的点击分布遵循目标周围的高斯模式。" 截图循环也产生可检测的模式： "可预测的暂停。视觉智能体在'思考'阶段完全静止。模式看起来像：操作 → 完全静止（1-5秒）→ 操作 → 完全静止 → 操作。" 复杂的反机器人 …(truncated)