🎯 AI浏览器自动化革命：browser-use/WebUI如何让任何网站都能被GPT-4控制

python --version

git clone https://github.com/browser-use/web-ui.git

cd web-ui

.\setup.bat

chmod +x setup.sh
./setup.sh

python -m venv venv

venv\Scripts\activate

source venv/bin/activate

pip install -r requirements.txt

playwright install chromium

# OpenAI配置
OPENAI_API_KEY=sk-your-openai-api-key

# 或者使用Anthropic Claude
ANTHROPIC_API_KEY=sk-ant-your-claude-api-key

# 如果使用本地模型（如Ollama）
OLLAMA_BASE_URL=http://localhost:11434

python scripts/run.Windows.bat

cd browser-ui
streamlit run app.py --server.port 8501

打开Google，搜索'Python入门教程'，找出搜索结果中播放量最高的视频。

import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

# 初始化语言模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 创建Agent实例
agent = Agent(
    task="打开Google，搜索'Python入门教程'，找出搜索结果中播放量最高的视频。",
    llm=llm,
)

# 执行任务
async def main():
    result = await agent.run()
    print(f"任务完成！最终结果：{result}")

    # 查看执行轨迹
    for step in agent.history:
        print(f"步骤 {step.number}: {step.action}")
        print(f"  观察: {step.observation}")

# 运行异步任务
asyncio.run(main())

import json
import csv
from datetime import datetime

class DataExporter:
    """数据导出器，支持多种格式"""

    def __init__(self):
        self.movies = []
        self.timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

    def add_movie(self, movie_info):
        """
        添加电影信息
        movie_info: dict，包含name, director, actors, rating, summary
        """
        self.movies.append(movie_info)
        print(f"已添加：{movie_info['name']}")

    def export_to_json(self):
        """导出为JSON格式"""
        filename = f"movies_{self.timestamp}.json"
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.movies, f, ensure_ascii=False, indent=2)
        print(f"已导出JSON文件: {filename}")
        return filename

    def export_to_csv(self):
        """导出为CSV格式"""
        filename = f"movies_{self.timestamp}.csv"
        with open(filename, 'w', newline='', encoding='utf-8-sig') as f:
            writer = csv.DictWriter(f, fieldnames=[
                'name', 'director', 'actors', 'rating', 'summary'
            ])
            writer.writeheader()
            writer.writerows(self.movies)
        print(f"已导出CSV文件: {filename}")
        return filename

# 使用示例
exporter = DataExporter()

import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

# 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 数据导出器实例
exporter = DataExporter()

# 定义任务
task = """
请完成以下数据采集任务：
1. 打开豆瓣电影Top250页面 (https://movie.douban.com/top250)
2. 采集当前页面的25部电影信息：
   - 电影名称
   - 导演
   - 主演演员
   - 评分
   - 一句话简介
3. 点击下一页，重复步骤2
4. 直到采集完100部电影
5. 最后输出所有采集到的数据
"""

# 创建Agent
agent = Agent(
    task=task,
    llm=llm,
)

# 执行并处理结果
async def main():
    result = await agent.run()

    # 解析结果并导出
    # 假设result包含采集到的数据
    if result and 'movies' in result:
        for movie in result['movies']:
            exporter.add_movie(movie)

        exporter.export_to_json()
        exporter.export_to_csv()

    return result

asyncio.run(main())

python douban_scraper.py

import asyncio
from browser_use import Agent, ActionResult
from langchain_openai import ChatOpenAI
from pydantic import BaseModel

# 定义表单数据结构
class SurveyForm(BaseModel):
    """调查问卷表单"""
    name: str = ""
    email: str = ""
    age_group: str = ""
    occupation: str = ""
    comments: str = ""

# 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 准备表单数据
form_data = SurveyForm(
    name="张三",
    email="zhangsan@example.com",
    age_group="25-35岁",
    occupation="软件工程师",
    comments="希望AI技术越来越成熟！"
)

# 构建详细的任务描述
task = f"""
请帮我填写一份在线调查问卷。

表单数据如下：
- 姓名：{form_data.name}
- 邮箱：{form_data.email}
- 年龄段：{form_data.age_group}
- 职业：{form_data.occupation}
- 留言：{form_data.comments}

请按以下步骤操作：
1. 打开表单页面（URL: https://www.wjx.cn/vm/h8XKz8P.aspx）
2. 找到并填写姓名输入框
3. 找到并填写邮箱输入框
4. 找到年龄段下拉菜单，选择对应选项
5. 找到职业下拉菜单，选择对应选项
6. 如果有留言框，填写留言内容
7. 检查所有填写内容是否正确
8. 找到并点击提交按钮
9. 等待提交结果，确认是否成功提交
"""

# 创建Agent
agent = Agent(
    task=task,
    llm=llm,
    max_steps=20,  # 增加步数，因为表单填写需要多个步骤
)

# 执行任务
async def main():
    print("开始填写表单...")
    result = await agent.run()

    if result:
        print(f"表单填写完成！")
        print(f"最终结果：{result}")
    else:
        print("表单填写过程中遇到问题")

asyncio.run(main())

import asyncio
import os
from browser_use import Agent
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

# 加载环境变量
load_dotenv()

# 从环境变量获取凭证（不要硬编码！）
USERNAME = os.getenv("EXAMPLE_SITE_USERNAME")
PASSWORD = os.getenv("EXAMPLE_SITE_PASSWORD")

# 检查凭证是否配置
if not USERNAME or not PASSWORD:
    raise ValueError("请在.env文件中配置 EXAMPLE_SITE_USERNAME 和 EXAMPLE_SITE_PASSWORD")

# 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 登录任务
task = f"""
请帮我完成以下网站的登录操作：
URL: https://example.com/login

步骤：
1. 打开登录页面
2. 在用户名输入框中输入：{USERNAME}
3. 在密码输入框中输入：{PASSWORD}
4. 如果有"记住我"选项，勾选它
5. 点击登录按钮
6. 等待登录成功，检查是否跳转到首页
7. 确认登录状态，告诉我是否成功登录
"""

# 创建Agent
agent = Agent(
    task=task,
    llm=llm,
)

# 执行登录
async def main():
    print(f"正在使用用户名 {USERNAME} 登录...")
    result = await agent.run()

    if result and "success" in result.lower():
        print("登录成功！")
    else:
        print(f"登录结果：{result}")

asyncio.run(main())

import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI
import json

# 初始化模型
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 准备批量数据
users = [
    {"username": "user001", "email": "user001@example.com", "phone": "13800001001"},
    {"username": "user002", "email": "user002@example.com", "phone": "13800001002"},
    {"username": "user003", "email": "user003@example.com", "phone": "13800001003"},
]

# 注册任务模板
REGISTER_TASK_TEMPLATE = """
请帮我注册一个新账号：

用户信息：
- 用户名：{username}
- 邮箱：{email}
- 手机号：{phone}

步骤：
1. 打开注册页面
2. 填写用户名
3. 填写邮箱
4. 填写手机号
5. 设置密码（使用用户名+123作为密码）
6. 确认密码
7. 勾选用户协议
8. 点击注册按钮
9. 如果需要邮箱验证，跳过验证步骤
10. 记录注册结果

最后告诉我注册是否成功。
"""

# 批量执行函数
async def register_batch():
    results = []

    for i, user in enumerate(users):
        print(f"\n{'='*50}")
        print(f"开始注册第 {i+1}/{len(users)} 个账号: {user['username']}")
        print('='*50)

        # 构建当前用户的任务
        task = REGISTER_TASK_TEMPLATE.format(**user)

        # 创建Agent
        agent = Agent(task=task, llm=llm)

        try:
            # 执行注册
            result = await agent.run()
            results.append({
                "user": user,
                "status": "success",
                "result": result
            })
            print(f"✓ {user['username']} 注册成功")

        except Exception as e:
            results.append({
                "user": user,
                "status": "failed",
                "error": str(e)
            })
            print(f"✗ {user['username']} 注册失败: {e}")

        # 每个任务之间稍作休息，避免被网站限流
        if i < len(users) - 1:
            await asyncio.sleep(5)

    # 汇总结果
    success_count = sum(1 for r in results if r["status"] == "success")
    print(f"\n{'='*50}")
    print(f"批量注册完成！成功: {success_count}/{len(users)}")
    print('='*50)

    # 保存结果
    with open("registration_results.json", "w", encoding="utf-8") as f:
        json.dump(results, f, ensure_ascii=False, indent=2)

    return results

# 运行批量任务
asyncio.run(register_batch())

# 自动化测试示例
test_task = """
测试任务：验证用户注册流程

测试步骤：
1. 打开网站注册页面
2. 填写测试用户信息：
   - 用户名：test_user_001
   - 邮箱：test@example.com
   - 密码：Test123456
3. 点击注册按钮
4. 验证是否出现"注册成功"提示
5. 验证是否自动跳转到登录页或首页
6. 使用注册的账号尝试登录
7. 验证登录是否成功

预期结果：
- 注册过程无错误
- 注册后可以成功登录

请详细报告每个步骤的执行结果。
"""

import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI
from datetime import datetime

# 监控任务
monitor_task = """
监控任务：检查某产品的价格和库存

1. 打开电商网站，搜索"iPhone 15 Pro"
2. 找到最便宜的商品链接
3. 进入商品详情页
4. 记录当前价格
5. 检查库存状态（是否有货/缺货/预售）
6. 检查配送信息

请以结构化方式返回监控结果：
{
    "product_name": "...",
    "price": "...",
    "stock_status": "...",
    "delivery_info": "...",
    "checked_at": "..."
}
"""

# 监控执行函数
async def run_monitoring():
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    agent = Agent(task=monitor_task, llm=llm)

    result = await agent.run()
    result["checked_at"] = datetime.now().isoformat()

    # 发送通知（可以根据result判断是否需要通知）
    if "缺货" in result.get("stock_status", ""):
        print("⚠️ 产品缺货，发送通知...")
        # 集成微信、邮件或钉钉通知

    return result

# 设置定时执行（每小时检查一次）
async def scheduled_monitor():
    while True:
        try:
            result = await run_monitoring()
            print(f"监控结果: {result}")
        except Exception as e:
            print(f"监控出错: {e}")

        # 等待1小时
        await asyncio.sleep(3600)

asyncio.run(scheduled_monitor())

content_task = """
内容聚合任务：从科技新闻网站收集今日热点

请完成以下步骤：
1. 依次访问以下网站：
   - 36氪 (https://36kr.com/)
   - 虎嗅 (https://www.huxiu.com/)
   - 少数派 (https://sspai.com/)
2. 在每个网站中：
   - 找出今日点击量最高的前3篇文章
   - 记录文章标题、链接和摘要
3. 将所有结果整理成一个统一的列表
4. 按点击量排序
5. 输出一份简报，包含：
   - 今日Top10科技新闻
   - 每个来源的特点简评
"""

# 执行内容聚合
async def aggregate_content():
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    agent = Agent(task=content_task, llm=llm)

    result = await agent.run()

    # 保存或发送聚合结果
    with open(f"tech_news_{datetime.now().strftime('%Y%m%d')}.md", "w", encoding="utf-8") as f:
        f.write(result)

    return result

asyncio.run(aggregate_content())

update_task = """
更新任务：更新维基百科词条中的某段内容

1. 打开维基百科Python词条页面
2. 找到"历史"或"发展历程"部分
3. 在适当位置添加以下新内容：
   "2024年，Python在TIOBE编程社区指数中继续位居榜首，成为最受欢迎的教学语言。"
4. 点击编辑保存按钮
5. 在编辑摘要中填写："添加2024年Python发展趋势信息"

注意：
- 保持原有格式和风格
- 不要破坏页面的其他内容
- 添加的内容要客观准确
"""

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# 使用Azure OpenAI
llm = ChatOpenAI(
    model="gpt-4o",
    temperature=0,
    api_key=os.getenv("AZURE_OPENAI_API_KEY"),
    azure_endpoint=os.getenv("AZURE_OPENAI_ENDPOINT"),
    api_version="2024-02-01"
)

# 使用Claude 3.5 Sonnet
claude = ChatAnthropic(
    model="claude-3-5-sonnet-20240620",
    anthropic_api_key=os.getenv("ANTHROPIC_API_KEY")
)

# 使用本地Ollama模型
ollama_llm = ChatOpenAI(
    model="llama3",
    temperature=0,
    base_url="http://localhost:11434/v1",
    api_key="ollama"  # Ollama不需要真正的API key
)

# 快速模式配置
agent = Agent(
    task=task,
    llm=llm,
    # 减少思考时间
    max_tokens=500,  # 限制响应长度
    temperature=0,   # 确定性输出

    # 减少等待时间
    page_load_timeout=10,  # 页面加载超时（秒）
    wait_for_timeout=2,    # 等待元素超时（秒）
)

import asyncio
from browser_use import Agent
from browser_use.exceptions import AgentError

async def robust_execute(task, max_retries=3):
    """带重试机制的执行函数"""

    for attempt in range(max_retries):
        try:
            llm = ChatOpenAI(model="gpt-4o", temperature=0)
            agent = Agent(task=task, llm=llm)

            result = await agent.run()

            # 验证结果是否合理
            if result and is_valid_result(result):
                return result

            print(f"第 {attempt + 1} 次尝试结果无效，尝试重试...")

        except AgentError as e:
            print(f"Agent执行错误: {e}")
            if attempt < max_retries - 1:
                await asyncio.sleep(5)  # 等待后重试
            else:
                raise

        except Exception as e:
            print(f"未知错误: {e}")
            raise

    return None

def is_valid_result(result):
    """验证结果是否有效"""
    # 根据具体任务定义验证逻辑
    if not result:
        return False

    # 检查结果是否包含预期内容
    expected_keywords = ["success", "完成", "成功"]
    return any(kw in result.lower() for kw in expected_keywords)

# 启用详细日志
import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("browser_use")

# 捕获每一步的详细信息
async def debug_execute(task):
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    agent = Agent(task=task, llm=llm)

    # 保存中间状态
    screenshots = []
    page_states = []

    # 设置回调收集信息
    original_step = agent._execute_step

    async def debug_step(step):
        result = await original_step(step)

        # 收集调试信息
        screenshots.append(agent.browser.get_screenshot())
        page_states.append(agent.browser.get_page_state())

        # 打印详细信息
        print(f"\n{'='*60}")
        print(f"步骤 {len(screenshots)}:")
        print(f"动作: {step.action}")
        print(f"观察: {step.observation}")
        print(f"下一步: {step.next_action}")
        print('='*60)

        return result

    agent._execute_step = debug_step

    result = await agent.run()

    # 保存调试信息
    save_debug_info(screenshots, page_states)

    return result

# ✅ 推荐：从环境变量读取
password = os.getenv("SITE_PASSWORD")

# ❌ 避免：硬编码密码
password = "MySecretPassword123"

async def rate_limited_request():
    """带速率限制的请求"""
    async with semaphore:  # 控制并发数
        await asyncio.sleep(2)  # 每次请求间隔2秒
        # 执行请求

SYSTEM_PROMPT = """
你是一个专业的网页自动化助手。请严格遵循以下规则：

【操作规范】
1. 每次操作前先观察页面状态，确保目标元素存在
2. 优先使用语义化的元素定位（如"点击提交按钮"）而非坐标
3. 输入操作前先清空原有内容
4. 操作完成后等待页面响应再进行下一步

【容错处理】
1. 如果找不到目标元素，尝试滚动页面或等待加载
2. 如果页面出现弹窗，先处理弹窗
3. 遇到错误时记录错误信息并尝试替代方案

【效率优化】
1. 避免不必要的页面刷新
2. 合理利用页面的搜索功能定位元素
3. 批量相似操作可以一次性描述

【输出格式】
完成每个任务后，请输出结构化的结果摘要。
"""

agent = Agent(
    task=task,
    llm=llm,
    system_prompt=SYSTEM_PROMPT
)

# 多标签页操作示例
multi_tab_task = """
任务：比较两个电商平台的同一商品价格

1. 在第一个标签页打开京东，搜索"小米手机14"
2. 记录第一款商品的价格
3. 打开新标签页，淘宝搜索同样的商品
4. 记录价格
5. 关闭淘宝标签页
6. 在京东标签页继续浏览，查看用户评价
7. 输出两个平台的价格比较结果
"""

agent = Agent(task=multi_tab_task, llm=llm)
await agent.run()

# 结合Selenium使用（处理特殊场景）
from selenium import webdriver

async def hybrid_approach():
    """
    混合使用方案：
    - browser-use处理需要AI决策的任务
    - Selenium处理标准化、重复性的任务
    """

    # 使用Selenium进行基础设置
    driver = webdriver.Chrome()
    driver.get("https://example.com")

    # 登录
    driver.find_element("name", "username").send_keys("user")
    driver.find_element("name", "password").send_keys("pass")
    driver.find_element("css", "button[type='submit']").click()

    # 切换到browser-use处理复杂决策
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    agent = Agent(
        task="完成表单填写，需要理解页面内容",
        llm=llm,
        browser=driver  # 复用Selenium的driver
    )

    result = await agent.run()

    return result