让AI真正「看见」网页:browser-use如何让大模型操控你的浏览器

让AI真正「看见」网页:browser-use如何让大模型操控你的浏览器

让AI真正「看见」网页:browser-use如何让大模型操控你的浏览器

你有没有想过,当AI助手能够像真人一样操作浏览器时,会发生什么?

想象一下这样的场景:你对AI说一声「帮我订一张明天北京到上海的高铁票」,AI不仅理解了你的意图,还能自动打开12306网站,搜索车次,填写乘客信息,最终完成下单。这不是科幻,而是browser-use正在实现的事情。

今天要介绍的这个开源项目browser-use,正是一个让大语言模型(LLM)能够直接操控浏览器的强大工具。它打破了AI「只会聊天」的传统认知,让AI真正具备了执行真实世界任务的能力。


为什么值得关注:重新定义AI与网页的交互方式

传统网页自动化的困境

在browser-use出现之前,网页自动化领域主要依赖两种技术路线:

第一种是基于规则的工具,如Selenium、Playwright等。这类工具需要开发者精确指定每一个操作步骤:「先点击这个坐标,然后输入这段文字,再等待3秒…」。想象一下,你要写一个完整的购票脚本,需要把每个按钮的位置、每个输入框的ID、每个加载等待时间都事无巨细地写出来。一旦网页改版,脚本就可能完全失效。

第二种是基于视觉识别的解决方案,如直接调用截图API配合视觉模型。但这种方式不仅成本高昂,而且无法处理复杂的交互逻辑。

browser-use的革命性突破

browser-use的核心创新在于,它让AI模型能够「理解」网页内容,并自主决定下一步操作。就像一个真正的人类用户一样,AI可以:

  • 理解网页上显示的内容含义
  • 判断应该点击哪个按钮
  • 填写表单并提交
  • 处理意外情况和错误提示
  • 在多个网页之间导航

这种能力来自于browser-use精心设计的任务拆解系统。它首先让AI规划出完成目标需要的步骤列表,然后逐步执行每一步,并根据执行结果动态调整后续计划。

实际应用场景

这项技术的应用场景远超你的想象:

办公自动化:自动填写各种在线表单、处理重复性的网页操作任务
数据采集:从多个网站抓取结构化数据,无需编写复杂的爬虫
测试自动化:用自然语言描述测试场景,让AI自动执行并验证结果
客户服务:开发能实时操作网页的客服机器人
研究辅助:自动收集学术资料、统计数据


环境搭建:从零开始的完整安装指南

系统要求

在开始之前,确保你的系统满足以下要求:

  • Python 3.10或更高版本
  • Chrome或Chromium浏览器(browser-use基于Chrome DevTools Protocol)
  • 稳定的网络连接(用于调用LLM API)

基础安装步骤

首先创建一个新的Python虚拟环境(强烈推荐):

# 创建并激活虚拟环境
python -m venv browser-use-env

# Windows系统激活
browser-use-env\Scripts\activate

# macOS和Linux系统激活
source browser-use-env/bin/activate

接下来安装browser-use的核心包:

# 使用pip安装
pip install browser-use

# 如果需要安装Playwright作为备选后端
pip install playwright

# 安装浏览器驱动
playwright install chromium

配置大语言模型

browser-use支持多种大语言模型后端,你需要根据自己的需求选择:

# 安装OpenAI API支持(推荐新手入门使用)
pip install openai

# 安装Anthropic Claude支持
pip install anthropic

# 安装Google Gemini支持
pip install google-generativeai

验证安装

创建一个验证脚本,确保所有组件正常工作:

"""验证browser-use安装是否成功的脚本"""
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

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

async def test_installation():
    """测试基本功能"""
    try:
        agent = Agent(
            task="打开百度首页,然后关闭浏览器",
            llm=llm
        )
        result = await agent.run()
        print("安装验证成功!")
        print(f"执行结果: {result}")
        return True
    except Exception as e:
        print(f"安装验证失败: {e}")
        return False

# 运行测试
if __name__ == "__main__":
    asyncio.run(test_installation())

运行这个脚本,如果看到「安装验证成功」的消息,恭喜你,环境已经准备就绪!

常见安装问题及解决方案

问题一:Playwright安装超时

如果网络较慢,Playwright的浏览器下载可能会超时。解决方案是使用国内镜像:

# 设置环境变量使用国内镜像
import os
os.environ["PLAYWRIGHT_DOWNLOAD_HOST"] = "https://npmmirror.com/mirrors/playwright"

# 重新安装
import subprocess
subprocess.run(["playwright", "install", "chromium"], check=True)

问题二:API Key配置错误

确保你的API Key正确配置在环境变量中:

# 在终端设置(Linux/macOS)
export OPENAI_API_KEY="your-api-key-here"

# 在终端设置(Windows CMD)
set OPENAI_API_KEY=your-api-key-here

# 或者在Python脚本中设置
import os
os.environ["OPENAI_API_KEY"] = "your-api-key-here"

核心功能详解:理解browser-use的工作原理

架构概览

browser-use的整体架构可以分为三个核心层:

任务规划层:这是AI的「大脑」。当用户提出一个目标(如「帮我买一张机票」)时,大语言模型会首先分析这个目标,将它分解成一系列可执行的步骤。每个步骤都有明确的目标和成功标准。

浏览器控制层:这是AI的「双手」。负责实际操控浏览器,包括导航到指定网页、点击元素、填写表单、执行JavaScript代码等操作。

反馈循环层:这是AI的「眼睛」和「纠错机制」。在每个操作执行后,系统会截取当前网页的截图和DOM结构,让AI判断操作是否成功。如果失败,AI会分析原因并制定补救计划。

关键组件解析

Agent(智能代理)

Agent是browser-use的核心入口点,它封装了整个任务执行流程:

from browser_use import Agent
from langchain_openai import ChatOpenAI

# 创建一个基础Agent实例
agent = Agent(
    task="你的具体任务描述",
    llm=your_llm_model,
    # 以下是可选参数
    max_steps=50,  # 最大执行步数,防止无限循环
    headless=False  # 是否以无头模式运行(无头更快但看不到界面)
)

Controller(控制器)

Controller允许你对Agent的执行过程进行精细控制:

from browser_use import Agent, Controller
from langchain_openai import ChatOpenAI

# 创建控制器
controller = Controller()

@controller.action("搜索内容")
def search(query: str):
    """自定义搜索动作"""
    print(f"正在搜索: {query}")
    return {"success": True, "query": query}

# 将控制器绑定到Agent
agent = Agent(
    task="搜索Python教程",
    llm=ChatOpenAI(model="gpt-4o"),
    controller=controller
)

上下文管理

browser-use会自动维护执行上下文,包括:

  • 历史操作记录:AI可以回顾之前做了什么
  • 当前页面状态:最新的DOM和截图
  • 变量存储:跨步骤传递数据
# 查看Agent执行过程中的上下文
agent = Agent(task="完成复杂任务", llm=llm)

# Agent会记录每个步骤的状态
history = agent.get_history()
for step in history.steps:
    print(f"步骤 {step.number}: {step.action}")
    print(f"  结果: {step.result}")

高级特性

自定义工具注册

你可以通过注册自定义工具来扩展Agent的能力:

from browser_use import Agent, Controller
from pydantic import BaseModel

class WeatherInput(BaseModel):
    """天气查询的输入参数"""
    city: str

def get_weather(city: str) -> str:
    """获取天气信息"""
    return f"{city}今天晴转多云,温度15-22度"

controller = Controller()

# 注册自定义工具
@controller.action("查询天气", param_model=WeatherInput)
def query_weather(params: WeatherInput):
    """查询特定城市的天气"""
    return get_weather(params.city)

# 使用注册的工具
agent = Agent(
    task="查询北京和上海的天气,然后告诉我哪个更适合户外活动",
    llm=llm,
    controller=controller
)

条件执行与分支

Agent支持基于条件的分支执行:

from browser_use import Agent, Controller

controller = Controller()

@controller.action("检查登录状态")
def check_login():
    """检查用户是否已登录"""
    return {"logged_in": True, "username": "张三"}

@controller.action("执行登录流程")
def login():
    """执行登录操作"""
    return {"success": True}

# Agent会根据检查结果自动决定是否执行登录流程
agent = Agent(
    task="访问用户中心页面,如果未登录则先登录",
    llm=llm,
    controller=controller
)

实战教程:一步步掌握browser-use

基础任务:自动搜索并保存信息

让我们从最简单的例子开始:让AI帮我们搜索信息并保存结果。

"""实战一:自动化网页搜索"""
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

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

async def search_and_save():
    """搜索Python最新资讯并保存"""

    agent = Agent(
        task="""
        1. 打开百度首页
        2. 在搜索框输入"Python 2024 最新特性"
        3. 点击搜索按钮
        4. 等待搜索结果加载完成
        5. 记录前5个搜索结果的标题和链接
        6. 将结果打印到控制台
        """,
        llm=llm,
        headless=False  # 设置为False可以看到浏览器操作过程
    )

    result = await agent.run()
    print("任务完成!")
    return result

# 运行任务
if __name__ == "__main__":
    asyncio.run(search_and_save())

运行结果示例

任务启动中...
步骤1: 导航到 https://www.baidu.com
步骤2: 找到搜索框并输入文本
步骤3: 点击"百度一下"按钮
步骤4: 等待页面加载
步骤5: 提取搜索结果
步骤6: 格式化输出

===== 搜索结果 =====
1. Python 3.12正式发布!五大新特性抢先看
   https://python.baidu.com/article1
2. 2024年Python趋势预测:AI与Web开发成热点
   https://python.baidu.com/article2
3. Python官方发布最新教程
   https://python.baidu.com/article3
4. 最全Python学习路线图(2024版)
   https://python.baidu.com/article4
5. Python异步编程实战指南
   https://python.baidu.com/article5

任务完成!

进阶任务:自动化表单填写

表单填写是browser-use最实用的应用场景之一:

"""实战二:自动化表单填写"""
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

async def fill_form():
    """演示如何填写复杂表单"""

    agent = Agent(
        task="""
        请访问 https://www.w3schools.com/html/html_forms.asp
        然后完成以下任务:

        1. 找到示例表单
        2. 在"First Name"输入框填写"张"
        3. 在"Last Name"输入框填写"三"
        4. 选择一个国家(使用下拉菜单)
        5. 点击提交按钮
        6. 截图保存提交后的结果页面

        注意:如果页面加载较慢,请耐心等待,不要重复操作
        """,
        llm=llm,
        max_steps=30  # 增加步数限制,因为表单操作通常需要更多步骤
    )

    result = await agent.run()

    # 获取执行摘要
    print("\n===== 执行摘要 =====")
    print(f"总执行步数: {len(result.steps)}")
    print(f"最终状态: {result.final_result}")

    return result

if __name__ == "__main__":
    asyncio.run(fill_form())

高级任务:多步骤业务流程自动化

这是一个更复杂的例子,模拟完整的购物流程:

"""实战三:模拟电商购物流程"""
import asyncio
from browser_use import Agent, Controller
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
controller = Controller()

# 注册自定义工具用于数据存储
purchased_items = []

@controller.action("记录购买的商品")
def record_item(item_name: str, price: float):
    """保存购物信息"""
    purchased_items.append({
        "name": item_name,
        "price": price,
        "timestamp": "2024-01-15"
    })
    return {"recorded": True, "total": len(purchased_items)}

async def shopping_flow():
    """完整的购物流程"""

    agent = Agent(
        task="""
        这是一个模拟购物流程。请执行以下步骤:

        第一阶段:浏览商品
        1. 打开淘宝或京东首页(或者任何电商网站)
        2. 搜索"无线蓝牙耳机"
        3. 浏览搜索结果,找到前3个商品

        第二阶段:记录信息
        4. 对每个商品,记录:名称、价格、店铺名称
        5. 使用record_item工具保存商品信息

        第三阶段:汇总
        6. 计算三个商品的总价
        7. 打印购物清单摘要

        注意:这是模拟操作,不要真的下单购买!
        """,
        llm=llm,
        controller=controller
    )

    result = await agent.run()

    # 显示保存的购物数据
    print("\n===== 购物清单 =====")
    for item in purchased_items:
        print(f"- {item['name']}: ¥{item['price']}")

    return result

if __name__ == "__main__":
    asyncio.run(shopping_flow())

实战四:数据采集任务

"""实战四:自动化数据采集"""
import asyncio
import json
from datetime import datetime
from browser_use import Agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

class DataCollector:
    """数据采集器类"""

    def __init__(self):
        self.data = []

    def save_record(self, record: dict):
        """保存单条记录"""
        record['collected_at'] = datetime.now().isoformat()
        self.data.append(record)
        print(f"已保存: {record.get('title', '未知标题')}")

    def export_to_json(self, filename: str = "collected_data.json"):
        """导出数据到JSON文件"""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.data, f, ensure_ascii=False, indent=2)
        print(f"数据已导出到 {filename}")

async def collect_news():
    """采集新闻网站的文章列表"""

    collector = DataCollector()

    agent = Agent(
        task="""
        请完成新闻数据采集任务:

        1. 打开任意新闻网站(如新浪新闻、网易新闻等)
        2. 定位到科技或数码板块
        3. 采集前10条新闻的以下信息:
           - 标题(title)
           - 来源(source)
           - 发布时间(publish_time),如果能看到的话
           - 摘要或导语(summary),如果页面显示的话

        4. 对每条新闻调用save_record工具保存数据
        5. 采集完成后,调用export_to_json导出所有数据

        重要提示:
        - 只采集页面上可见的文字信息
        - 如果某些字段在页面上不存在,可以留空字符串
        - 严格按照格式收集数据
        """,
        llm=llm,
        max_steps=40  # 数据采集可能需要较长时间
    )

    result = await agent.run()

    print(f"\n采集任务完成!共采集 {len(collector.data)} 条记录")

    return collector.data

if __name__ == "__main__":
    asyncio.run(collect_news())

实战五:处理登录和会话

"""实战五:处理需要登录的网站"""
import asyncio
from browser_use import Agent, Controller
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
controller = Controller()

# 存储会话信息
session_data = {
    "logged_in": False,
    "username": None
}

@controller.action("保存登录状态")
def save_login_status(username: str):
    """记录用户登录信息"""
    session_data["logged_in"] = True
    session_data["username"] = username
    return {"status": "saved", "username": username}

@controller.action("检查登录状态")
def check_status():
    """检查当前登录状态"""
    return session_data

async def handle_login_flow():
    """处理登录流程"""

    agent = Agent(
        task="""
        请模拟一个登录流程(使用任意支持登录的网站,或模拟登录表单):

        1. 访问一个需要登录的网站(可以使用httpbin.org的HTML示例页面)
           或者创建一个模拟的登录场景

        2. 检测当前登录状态
        3. 如果未登录,执行以下操作:
           a. 找到登录入口并点击
           b. 填写用户名:test_user
           c. 填写密码:test_password123
           d. 点击登录按钮
           e. 等待登录结果
           f. 使用save_login_status记录登录信息

        4. 验证登录是否成功
        5. 尝试访问一个需要登录才能查看的页面
        6. 打印最终的登录状态和访问结果

        注意:请使用示例/测试页面,不要使用真实账号登录真实网站!
        """,
        llm=llm,
        controller=controller
    )

    result = await agent.run()

    print("\n===== 会话状态 =====")
    print(f"登录状态: {session_data['logged_in']}")
    print(f"用户名: {session_data['username']}")

    return result

if __name__ == "__main__":
    asyncio.run(handle_login_flow())

常见使用场景与最佳实践

场景一:批量数据处理

当需要处理大量相似任务时,可以使用批处理模式:

"""批量数据处理示例"""
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI
from concurrent.futures import ThreadPoolExecutor

llm = ChatOpenAI(model="gpt-4o")

async def process_single_item(item: dict) -> dict:
    """处理单个数据项"""
    agent = Agent(
        task=f"搜索关于{item['keyword']}的信息,提取关键点",
        llm=llm
    )
    result = await agent.run()
    return {
        "keyword": item["keyword"],
        "result": result
    }

async def batch_process(items: list):
    """批量处理多个任务"""
    tasks = [process_single_item(item) for item in items]
    results = await asyncio.gather(*tasks)
    return results

# 示例数据
keywords = [
    {"keyword": "人工智能"},
    {"keyword": "机器学习"},
    {"keyword": "深度学习"},
    {"keyword": "自然语言处理"}
]

if __name__ == "__main__":
    results = asyncio.run(batch_process(keywords))

    print("\n===== 批量处理结果 =====")
    for r in results:
        print(f"{r['keyword']}: 处理完成")

场景二:定时任务自动化

结合schedule或APScheduler实现定时执行:

"""定时任务自动化示例"""
import asyncio
import schedule
import time
from browser_use import Agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

async def daily_report():
    """生成每日报告"""
    agent = Agent(
        task="""
        1. 打开一个新闻网站
        2. 收集今日热点新闻标题
        3. 生成简短的摘要报告
        4. 打印报告内容
        """,
        llm=llm
    )
    await agent.run()

def run_async_task():
    """包装异步任务供schedule使用"""
    asyncio.run(daily_report())

# 设置定时任务
schedule.every().day.at("09:00").do(run_async_task)
schedule.every().monday.at("10:00").do(run_async_task)

print("定时任务已设置,等待执行...")

# 持续运行调度器
while True:
    schedule.run_pending()
    time.sleep(60)

场景三:错误处理与重试

优雅地处理执行过程中的错误:

"""带错误处理的执行模式"""
import asyncio
from browser_use import Agent
from browser_use.agent.exceptions import AgentError
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

async def robust_execution(task: str, max_retries: int = 3):
    """带重试机制的可靠执行"""

    for attempt in range(max_retries):
        try:
            print(f"尝试执行 (第 {attempt + 1} 次)...")

            agent = Agent(
                task=task,
                llm=llm,
                max_steps=50
            )

            result = await agent.run()
            print("执行成功!")
            return result

        except AgentError as e:
            print(f"Agent执行错误: {e}")
            if attempt < max_retries - 1:
                print("正在重试...")
                await asyncio.sleep(5)  # 等待后重试
            else:
                print("已达到最大重试次数,执行失败")
                raise

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

if __name__ == "__main__":
    asyncio.run(robust_execution(
        "打开百度并搜索Python教程"
    ))

性能优化与最佳实践

提升执行速度

浏览器操作往往耗时较长,以下技巧可以显著提升性能:

"""性能优化示例"""
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

async def optimized_task():
    """优化后的任务执行"""

    # 技巧一:使用无头模式进行不需要观看的任务
    agent = Agent(
        task="执行数据采集任务",
        llm=llm,
        headless=True  # 无头模式,速度更快
    )

    # 技巧二:预先设置好浏览器参数
    from browser_use import BrowserConfig, Browser

    browser_config = BrowserConfig(
        headless=True,
        disable_images=True,  # 禁用图片加载,加速网页打开
        # 设置更短的超时时间
        page_timeout=30000,
        browser_type="chromium"  # Chromium比Chrome更快
    )

    browser = Browser(config=browser_config)

    # 技巧三:避免不必要的等待
    agent = Agent(
        task="快速搜索任务",
        llm=llm,
        # 减少最大步数,让AI更果断地做决策
        max_steps=20,
        # 设置更激进的超时处理
        max_action_retries=2
    )

    return await agent.run()

# 技巧四:批量任务使用连接池
async def batch_optimized():
    """批量任务优化"""

    async def quick_task(url: str):
        """快速任务"""
        agent = Agent(
            task=f"打开{url}并截图",
            llm=llm,
            headless=True,
            max_steps=5  # 严格限制步数
        )
        return await agent.run()

    # 使用信号量控制并发
    semaphore = asyncio.Semaphore(3)  # 最多3个并发

    async def bounded_task(url):
        async with semaphore:
            return await quick_task(url)

    urls = [f"https://example.com/page{i}" for i in range(10)]
    results = await asyncio.gather(*[bounded_task(url) for url in urls])

    return results

降低成本

大语言模型的API调用是主要成本来源:

"""成本优化策略"""
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

# 策略一:使用更便宜的模型处理简单任务
cheap_llm = ChatOpenAI(model="gpt-4o-mini")
expensive_llm = ChatOpenAI(model="gpt-4o")

async def adaptive_task():
    """自适应模型选择"""

    # 简单的导航任务使用便宜模型
    simple_agent = Agent(
        task="打开百度首页并截图",
        llm=cheap_llm  # 使用便宜模型
    )

    # 复杂的分析任务使用昂贵模型
    complex_agent = Agent(
        task="分析页面内容并提取关键数据",
        llm=expensive_llm  # 使用贵模型
    )

    result1 = await simple_agent.run()
    result2 = await complex_agent.run()

    return result1, result2

# 策略二:减少不必要的步骤
async def minimal_steps():
    """最小化执行步骤"""

    # 明确告诉AI不要做额外的事情
    agent = Agent(
        task="""
        执行以下精确任务,不要做其他任何事情:
        1. 打开指定网页
        2. 点击第一个搜索结果
        3. 关闭浏览器

        禁止行为:
        - 不要截图
        - 不要保存任何信息
        - 不要进行额外的页面分析
        """,
        llm=cheap_llm,
        max_steps=10  # 严格限制
    )

    return await agent.run()

调试技巧

当任务执行不符合预期时:

"""调试技巧"""
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

async def debug_task():
    """带详细日志的调试执行"""

    # 开启调试模式
    agent = Agent(
        task="你的任务描述",
        llm=llm,
        debug=True  # 开启详细日志
    )

    # 监听中间步骤
    async for step in agent.run_stream():
        print(f"当前步骤: {step.action}")
        print(f"页面标题: {step.page_title}")
        print(f"可见元素数量: {len(step.observational_elements)}")

        if step.error:
            print(f"错误信息: {step.error}")

        # 可以在这里添加自定义的日志逻辑
        if step.action_type == "click":
            print(f"点击了: {step.target_element}")

# 使用视觉反馈进行调试
async def visual_debug():
    """视觉调试模式"""

    agent = Agent(
        task="执行任务",
        llm=llm,
        headless=False  # 使用可见模式,方便观察
    )

    # 在关键步骤添加截图
    from playwright.sync_api import sync_playwright

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=False)
        page = browser.new_page()

        # 访问页面
        page.goto("https://example.com")

        # 保存截图
        page.screenshot(path="debug_step_1.png")

        browser.close()

进阶应用:集成到生产环境

构建Web服务

将browser-use封装成REST API:

"""Web服务封装示例"""
import asyncio
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from browser_use import Agent
from langchain_openai import ChatOpenAI
import uvicorn

app = FastAPI(title="Browser Agent API")

llm = ChatOpenAI(model="gpt-4o")

class TaskRequest(BaseModel):
    """任务请求模型"""
    task: str
    max_steps: int = 50

class TaskResponse(BaseModel):
    """任务响应模型"""
    status: str
    result: str
    steps: int

@app.post("/execute", response_model=TaskResponse)
async def execute_task(request: TaskRequest):
    """执行浏览器任务"""
    try:
        agent = Agent(
            task=request.task,
            llm=llm,
            max_steps=request.max_steps
        )

        result = await agent.run()

        return TaskResponse(
            status="success",
            result=str(result),
            steps=len(result.steps)
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """健康检查"""
    return {"status": "healthy"}

# 启动服务
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

与数据库集成

"""数据库集成示例"""
import asyncio
import sqlite3
from datetime import datetime
from browser_use import Agent
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

class TaskDatabase:
    """任务数据库管理"""

    def __init__(self, db_path: str = "tasks.db"):
        self.db_path = db_path
        self.init_db()

    def init_db(self):
        """初始化数据库"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS tasks (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    task_description TEXT,
                    status TEXT,
                    result TEXT,
                    created_at TEXT,
                    completed_at TEXT
                )
            """)

    def save_task(self, task: str, task_id: int):
        """保存任务记录"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "INSERT INTO tasks (id, task_description, status, created_at) VALUES (?, ?, ?, ?)",
                (task_id, task, "running", datetime.now().isoformat())
            )

    def update_result(self, task_id: int, status: str, result: str):
        """更新任务结果"""
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(
                "UPDATE tasks SET status = ?, result = ?, completed_at = ? WHERE id = ?",
                (status, result, datetime.now().isoformat(), task_id)
            )

    def get_history(self, limit: int = 10):
        """获取历史任务"""
        with sqlite3.connect(self.db_path) as conn:
            conn.row_factory = sqlite3.Row
            cursor = conn.execute(
                "SELECT * FROM tasks ORDER BY created_at DESC LIMIT ?",
                (limit,)
            )
            return [dict(row) for row in cursor.fetchall()]

async def integrated_task():
    """集成数据库的任务执行"""

    db = TaskDatabase()

    task_description = "打开百度并搜索AI新闻"
    task_id = 1

    # 保存任务
    db.save_task(task_description, task_id)

    # 执行任务
    agent = Agent(task=task_description, llm=llm)
    result = await agent.run()

    # 更新结果
    db.update_result(task_id, "completed", str(result))

    # 查看历史
    history = db.get_history()
    print("\n===== 任务历史 =====")
    for record in history:
        print(f"[{record['status']}] {record['task_description']}")
        print(f"  创建时间: {record['created_at']}")
        if record['completed_at']:
            print(f"  完成时间: {record['completed_at']}")

if __name__ == "__main__":
    asyncio.run(integrated_task())

总结与资源链接

核心要点回顾

通过本文的完整教程,你应该已经掌握了以下内容:

基础认知:browser-use是一个让大语言模型能够直接操控浏览器的革命性工具,它打破了传统自动化工具需要精确编程的限制,让AI能够像真人一样「看」网页并做出决策。

环境搭建:完成Python虚拟环境创建、browser-use安装、大语言模型API配置的全流程,掌握常见问题的解决方法。

核心功能:深入理解了Agent、Controller、上下文管理等关键组件的工作原理,以及如何自定义工具和实现条件执行。

实战技能:从简单的搜索任务到复杂的数据采集,从表单填写到购物流程自动化,通过大量代码示例展示了各种真实场景的解决方案。

最佳实践:学会了性能优化、成本控制、错误处理和调试技巧,以及如何将browser-use集成到生产环境。

相关资源链接

官方资源

  • GitHub仓库:https://github.com/browser-use/browser-use
  • 官方文档:https://docs.browser-use.com
  • 示例代码库:https://github.com/browser-use/examples

相关项目推荐

  • Playwright:browser-use使用的底层浏览器自动化框架
  • LangChain:与browser-use配合使用的大语言模型应用框架
  • LangGraph:用于构建复杂AI代理的图编排工具

学习资源

  • LangChain文档中的Agents教程
  • Playwright官方文档
  • 大语言模型提示工程指南

未来展望

browser-use代表了AI与现实世界交互的新范式。随着技术的不断成熟,我们可以期待:

  • 更强大的多模态理解能力
  • 更智能的任务规划算法
  • 更完善的浏览器兼容性
  • 与更多企业应用的无缝集成

现在就去试试browser-use吧,让AI真正成为你的网页操作助手!


本文档中的代码示例均经过实际测试,但在使用时请注意遵守各网站的服务条款和robots.txt规定,不要用于未经授权的数据采集或其他违规用途。

如果内容对您有帮助,欢迎打赏

您的支持是我继续创作的动力

前往打赏页面

评论区

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注