Browser Agent：Computer Use 工程实践

# Anthropic Claude 4.x 系列
import anthropic

client = anthropic.Anthropic()

response = client.messages.create(
    model="claude-sonnet-4-5",
    tools=[
        {
            "type": "computer_20251124",  # 最新 computer use 工具
            "name": "computer",
            "display_width_px": 1280,
            "display_height_px": 720,
        }
    ],
    messages=[{
        "role": "user",
        "content": [
            {"type": "image", "source": {"type": "base64", "data": screenshot_b64}},
            {"type": "text", "text": "在搜索框输入 'iPhone 15' 然后点击搜索按钮"}
        ]
    }]
)

# 模型返回 action
action = response.content[0].input
# {"action": "left_click", "coordinate": [423, 156]}

# browser-use 是开源的 browser agent 框架
from browser_use import Agent

agent = Agent(
    task="在 GitHub 上给 tanteng/blog 仓库点 star",
    llm=ChatOpenAI(model="gpt-4o"),
    browser=Browser(
        viewport={"width": 1280, "height": 720},
        # 关键：用独立 profile 隔离
        user_data_dir="/tmp/agent-profile-001",
    ),
)

result = await agent.run()
# Agent 自动：打开 GitHub → 登录 → 搜索 → 点 star

# 用 Playwright 启动隔离浏览器
from playwright.async_api import async_playwright

async def get_isolated_browser(task_id: str):
    playwright = await async_playwright().start()
    
    # 每个任务独立 user data dir
    context = await playwright.chromium.launch_persistent_context(
        user_data_dir=f"/tmp/agent-profile-{task_id}",
        headless=False,  # 有些场景需要 headless=False
        viewport={"width": 1280, "height": 720},
        # 关键：禁用 web security，避开 CORS
        args=["--disable-web-security"],
    )
    
    return context

# ✅ 用 Browserless 等托管服务
browser = await playwright.chromium.connect(
    "wss://production.browserless.io?token=YOUR_API_KEY"
)

# 关键配置
browser_args = {
    " stealth: True,  # 绕过 bot 检测
    "proxy": {
        "server": "residential-proxy.example.com:8000",
        "username": "...",
        "password": "..."
    },
    "region": "us-east"  # 离 agent 服务近
}

# 训练数据集中在 1024-1440 宽度
# 1280×720 是黄金选择
VIEWPORT = {"width": 1280, "height": 720}

# 模型训练时见过的尺寸占比 ~80%
# 太宽（如 1920×1080）会让按钮太小
# 太窄（如 800×600）会让模型没见过

# Browserless Session API
session = await browserless.create_session({
    "session_id": "user-12345",  # 复用之前的登录态
    "cookies": user_cookies,  # 注入登录 cookie
})

# 同一用户的多次任务共享登录态
# 不需要每次重新登录

# 住宅代理（residential proxy）才能绕过 bot 检测
proxy_config = {
    "type": "residential",
    "country": "US",  # 按目标站点选择
    "rotate": "per_request"  # 每次请求换 IP
}

SECURITY_CHECKS = [
    # 1. 最小权限：每个任务独立 profile
    "separate_browser_profile_per_task",
    
    # 2. 域白名单：只允许访问可信站点
    "domain_allowlist",
    
    # 3. 人在回路：不可逆操作必须确认
    "human_in_the_loop_for_irreversible_actions",
    
    # 4. 沙箱环境：Docker + 虚拟显示
    "docker_sandbox_with_virtual_display",
    
    # 5. 短期凭证：避免 admin 权限
    "short_lived_scoped_tokens",
    
    # 6. 计划后执行：先展示动作再执行
    "plan_then_execute",
    
    # 7. 可中止：用户随时能停
    "user_can_abort_anytime",
    
    # 8. 完整审计：所有截图和动作
    "audit_log_all_screenshots_and_actions",
]

class SafeBrowserAgent:
    ALLOWED_DOMAINS = {"github.com", "stackoverflow.com", "docs.example.com"}
    
    IRREVERSIBLE_ACTIONS = {
        "submit_form_payment",
        "delete_account",
        "send_email_to_external",
        "submit_github_issue",
        "merge_pull_request"
    }
    
    async def execute_action(self, action: dict):
        # 1. 域白名单检查
        current_url = self.page.url
        domain = urlparse(current_url).netloc
        
        if not any(allowed in domain for allowed in self.ALLOWED_DOMAINS):
            raise SecurityError(f"Domain {domain} not in allowlist")
        
        # 2. 不可逆操作拦截
        if action.get("type") in self.IRREVERSIBLE_ACTIONS:
            await self.request_human_approval(action)
        
        # 3. 执行
        await self.page.dispatch_action(action)
    
    async def request_human_approval(self, action):
        # 推到审批队列，等人类确认
        approval_id = await self.approval_queue.enqueue(action)
        print(f"⚠️ 等待人工审批: {approval_id}")
        # ...

# 先让 agent 输出计划，人类批准后再执行
async def plan_then_execute(task: str):
    # Step 1: 让 agent 列出计划（不实际操作）
    plan = await agent.plan(task)
    
    print("计划：")
    for i, step in enumerate(plan.steps):
        print(f"{i+1}. {step.description}")
    
    # Step 2: 人类审批
    if not await get_user_approval(plan):
        return
    
    # Step 3: 执行
    result = await agent.execute(plan)
    return result

# 启用 zoom 模式（Claude 4.6+）
computer_tool = {
    "type": "computer_20251124",
    "display_width_px": 1280,
    "display_height_px": 720,
    "enable_zoom": True,        # 关键
    "adaptive_thinking": True,  # 让模型自纠
}

# 提示小目标用键盘而非鼠标
# 模型默认会用 pixel click，但对小元素（系统托盘、小勾选框）容易失败
# 显式提示：能 tab + enter 解决的用键盘

SYSTEM_PROMPT = """你是一个浏览器操作助手。

## 工作流程
1. 先看截图，描述你看到了什么
2. 决定下一步动作
3. 执行

## 注意事项
- 小目标优先用键盘（Tab + Enter）而不是鼠标点击
- 表单提交前再次确认所有字段
- 遇到验证码立即停止并报告
- 不要主动访问登录后才能看到的页面（除非已登录）

## 错误恢复
- 点击没反应 → 等待 1 秒再点
- 页面未加载 → 截图确认后等 2 秒
- 操作失败 → 撤销前一步
"""

VALIDATION_ASSERTIONS = {
    # 每个网络请求必须命中允许的 origin
    "every_request_to_allowed_origin",
    
    # 运行在 N 步 / N 秒内终止
    "terminates_within_steps_and_time",
    
    # 单次 token 消耗 < B
    "tokens_per_run_below_budget",
    
    # 每个不可逆操作有确认 hook
    "every_irreversible_action_has_approval_hook",
    
    # Action log 重放产出相同状态
    "action_log_replay_produces_identical_state",
    
    # 单任务成本上限
    "cost_ceiling_enforced_per_task"
}

async def validate_agent(agent_task):
    results = []
    
    # 1. 域白名单
    requests = await agent_task.get_all_requests()
    for req in requests:
        if not is_allowed_domain(req.url):
            results.append({"assertion": "allowed_origin", "pass": False, "url": req.url})
    
    # 2. 步骤限制
    if agent_task.steps > MAX_STEPS:
        results.append({"assertion": "step_limit", "pass": False})
    
    # 3. 成本限制
    if agent_task.cost > MAX_COST:
        results.append({"assertion": "cost_limit", "pass": False})
    
    # 4. 不可逆操作必须有 approval
    for action in agent_task.actions:
        if action.type in IRREVERSIBLE_ACTIONS:
            if not action.had_approval:
                results.append({"assertion": "approval_required", "pass": False})
    
    return all(r["pass"] for r in results)