Browser Use 实战教程第2课：高级技巧与企业级应用

多标签页管理

from browser_use import Agent

task = """
1. 打开 tab 1: jd.com 搜索 'MacBook Pro' 记录价格
2. 打开 tab 2: taobao.com 搜索 'MacBook Pro' 记录价格  
3. 打开 tab 3: pdd.com 搜索 'MacBook Pro' 记录价格
4. 对比三个平台的价格，返回最低价的商品链接
"""

agent = Agent(
    task=task,
    llm=ChatOpenAI(model="gpt-4o"),
    max_actions_per_step=10,  # 允许多步操作
)

登录态保持

from browser_use import Agent, BrowserConfig
import json

# 方式1：Cookie注入
cookies = [
    {"name": "session_id", "value": "abc123", "domain": ".taobao.com"},
    {"name": "token", "value": "xyz789", "domain": ".jd.com"},
]

config = BrowserConfig(
    cookies=cookies,  # 注入Cookie
)

# 方式2：复用浏览器Profile
config = BrowserConfig(
    user_data_dir="/tmp/browser_profile",  # 保存登录态
    headless=False,  # 首次需要手动登录
)

# 首次登录后，后续运行自动复用登录态
agent = Agent(
    task="访问我的淘宝订单页面，提取最近10个订单信息",
    llm=llm,
    browser_config=config,
)

文件下载

task = """
1. 打开 arxiv.org/abs/2401.12345
2. 点击 'Download PDF' 
3. 等待下载完成
4. 返回下载文件的路径
"""

# Browser Use自动处理下载
config = BrowserConfig(
    downloads_path="/root/downloads/",  # 下载目录
)

错误处理与重试

from browser_use import Agent
import asyncio

async def robust_run(task: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            agent = Agent(
                task=task,
                llm=ChatOpenAI(model="gpt-4o"),
                max_actions_per_step=5,
            )
            result = await agent.run()
            return result
        except Exception as e:
            print(f"Attempt {attempt+1} failed: {e}")
            if attempt == max_retries - 1:
                raise
            await asyncio.sleep(5)  # 等待后重试

# 使用
result = asyncio.run(robust_run("提取京东商品数据"))

批量任务管道

import asyncio
from browser_use import Agent
from dataclasses import dataclass

@dataclass
class Task:
    name: str
    prompt: str
    priority: int

tasks = [
    Task("竞品监控", "监控竞品价格...", 1),
    Task("数据采集", "采集供应商信息...", 2),
    Task("评价分析", "分析商品评价...", 3),
]

async def run_pipeline(tasks: list[Task]):
    results = {}
    
    # 按优先级排序
    tasks.sort(key=lambda t: t.priority)
    
    for task in tasks:
        print(f"执行: {task.name}")
        agent = Agent(task=task.prompt, llm=ChatOpenAI(model="gpt-4o"))
        try:
            result = await agent.run()
            results[task.name] = {"status": "success", "data": result}
        except Exception as e:
            results[task.name] = {"status": "error", "error": str(e)}
    
    return results

results = asyncio.run(run_pipeline(tasks))

与Hermes Agent集成

# 把Browser Use做成Hermes MCP工具
# /root/tools/browser-use-free/mcp_server.py

from hermes_mcp import HermesMCP

mcp = HermesMCP("browser-use", "AI浏览器自动化")

@mcp.tool()
async def browse_and_extract(url: str, instruction: str) -> str:
    """用AI浏览网页并提取信息"""
    from browser_use import Agent
    agent = Agent(
        task=f"打开 {url}，{instruction}",
        llm=get_llm(),
    )
    return await agent.run()

@mcp.tool()
async def auto_fill_form(url: str, form_data: dict) -> str:
    """自动填写网页表单"""
    fields = ", ".join([f"'{k}'填'{v}'" for k, v in form_data.items()])
    task = f"打开 {url}，填写表单：{fields}，然后提交"
    agent = Agent(task=task, llm=get_llm())
    return await agent.run()

@mcp.tool()
async def price_monitor(urls: list[str], product: str) -> str:
    """监控多个网站的商品价格"""
    results = []
    for url in urls:
        agent = Agent(
            task=f"打开 {url}，搜索 '{product}'，提取前5个商品的价格",
            llm=get_llm(),
        )
        result = await agent.run()
        results.append({"url": url, "data": result})
    return json.dumps(results, ensure_ascii=False)

mcp.run()

性能优化

1. 用本地模型降低成本

from langchain_ollama import ChatOllama

# 用本地Qwen做简单任务
llm_local = ChatOllama(model="qwen2.5:14b")

# 复杂任务才用GPT-4o
llm_cloud = ChatOpenAI(model="gpt-4o")

# 根据任务复杂度选择模型
def get_llm(task_complexity: str):
    if task_complexity == "simple":
        return llm_local  # 免费
    return llm_cloud  # 收费但更准

2. 缓存页面分析结果

import hashlib
import json

cache = {}

async def cached_browse(url: str, instruction: str):
    key = hashlib.md5(f"{url}:{instruction}".encode()).hexdigest()
    
    if key in cache:
        return cache[key]
    
    agent = Agent(task=f"打开 {url}，{instruction}", llm=llm)
    result = await agent.run()
    
    cache[key] = result
    return result

3. 并行执行独立任务

import asyncio

async def parallel_tasks():
    tasks = [
        Agent(task="监控京东价格", llm=llm).run(),
        Agent(task="监控淘宝价格", llm=llm).run(),
        Agent(task="监控拼多多价格", llm=llm).run(),
    ]
    results = await asyncio.gather(*tasks)
    return results

企业级架构

┌─────────────────────────────────────────────┐
│              Browser Use 集群                │
├─────────────────────────────────────────────┤
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ Worker 1 │  │ Worker 2 │  │ Worker 3 │  │
│  │ (Chrome) │  │ (Chrome) │  │ (Chrome) │  │
│  └──────────┘  └──────────┘  └──────────┘  │
├─────────────────────────────────────────────┤
│  任务队列 (Redis/RabbitMQ)                   │
├─────────────────────────────────────────────┤
│  调度器 (定时/事件触发)                       │
├─────────────────────────────────────────────┤
│  数据管道 (清洗→存储→报告)                    │
└─────────────────────────────────────────────┘

下节预告

第3课：Browser Use商业化——如何把AI浏览器自动化包装成产品卖钱。