让 AI 亲自上网干活：browser-use 开源项目实战拆解与踩坑

[1] <input type="search" placeholder="搜索商品">
[2] <button>搜索</button>
[3] <a> 电饭煲 ¥299 </a>
[4] <button>下一页</button>

# 方式一：pip 直接装
pip install browser-use

# 方式二：要体验 2026 年最新的 Rust 内核 beta 版
pip install "browser-use[core]"

# 装 Playwright 的浏览器内核（首次需要）
playwright install chromium --with-deps

# .env
OPENAI_API_KEY=sk-你的key
# 国产模型也行，DeepSeek、通义、Kimi 都有对应接入方式

import asyncio
from dotenv import load_dotenv
from browser_use import Agent, Browser
from browser_use.llm import ChatOpenAI

load_dotenv()

async def main():
    browser = Browser(
        headless=False,  # 设成 True 就看不到浏览器窗口了
    )
    agent = Agent(
        task="去 GitHub 打开 browser-use 仓库页面，告诉我它现在有多少 Star、多少 Fork，最近一次提交是什么时候",
        llm=ChatOpenAI(model="gpt-4o"),
        browser=browser,
    )
    history = await agent.run()
    print("最终结果：", history.final_result())

asyncio.run(main())

import asyncio
from dotenv import load_dotenv
from pydantic import BaseModel
from browser_use import Agent, Browser
from browser_use.llm import ChatOpenAI

load_dotenv()

class JobInfo(BaseModel):
    """单条招聘岗位信息"""
    title: str          # 岗位名称
    company: str        # 公司名
    salary: str         # 薪资范围
    location: str       # 城市
    tags: list[str]     # 技术标签

class JobList(BaseModel):
    """岗位列表"""
    jobs: list[JobInfo]
    total_found: int

async def main():
    browser = Browser(headless=True)
    agent = Agent(
        task=(
            "去招聘网站搜索'大模型算法工程师'岗位，"
            "提取前两页的岗位信息，包括岗位名、公司、薪资、城市和技术标签。"
        ),
        llm=ChatOpenAI(model="gpt-4o"),
        browser=browser,
        output_schema=JobList,   # 关键：指定输出结构
    )
    history = await agent.run()
    result = JobList.model_validate_json(history.final_result())
    for job in result.jobs:
        print(f"{job.company} | {job.title} | {job.salary} | {job.location}")

asyncio.run(main())

import asyncio
import json
from datetime import datetime
from dotenv import load_dotenv
from pydantic import BaseModel
from browser_use import Agent, Browser, BrowserProfile
from browser_use.llm import ChatOpenAI

load_dotenv()

# ---------- 数据结构 ----------
class PriceRecord(BaseModel):
    platform: str      # 平台：京东/淘宝/拼多多
    product: str       # 商品名
    price: float       # 当前价
    shop_type: str     # 自营/旗舰店/第三方
    coupon: str        # 优惠信息，没有则写"无"

class PriceReport(BaseModel):
    records: list[PriceRecord]
    collected_at: str

# ---------- 昨天的价格基线（实际项目里从数据库读） ----------
BASELINE = {
    "品牌A空气炸锅 5L": 299.0,
    "竞品B空气炸锅 4.5L": 349.0,
    "竞品C可视化炸锅": 399.0,
}

async def collect_one(browser: Browser, platform: str, task: str) -> PriceReport:
    """单个平台的采集任务，独立 Agent 跑，互不干扰"""
    agent = Agent(
        task=task,
        llm=ChatOpenAI(model="gpt-4o"),
        browser=browser,
        output_schema=PriceReport,
        browser_profile=BrowserProfile(
            allowed_domains=[f"*{platform}.com"],  # 安全边界：只准去这个平台
        ),
    )
    history = await agent.run()
    return PriceReport.model_validate_json(history.final_result())

async def main():
    browser = Browser(headless=True)

    tasks = [
        ("jd",  "打开京东，搜索'空气炸锅'，找到'品牌A空气炸锅 5L'、'竞品B空气炸锅 4.5L'、'竞品C可视化炸锅'这三个商品，记录当前价格、店铺类型和优惠信息"),
        ("taobao", "打开淘宝，搜索上述三款空气炸锅，记录各店铺当前价格和是否有优惠券"),
        ("pinduoduo", "打开拼多多，搜索上述三款空气炸锅，记录百亿补贴价"),
    ]

    all_records = []
    for platform, desc in tasks:
        try:
            report = await collect_one(browser, platform, desc)
            all_records.extend(report.records)
        except Exception as e:
            # 单个平台失败不影响整体，记录后继续
            print(f"[WARN] {platform} 采集失败: {e}")

    await browser.close()

    # ---------- 对比基线，生成日报 ----------
    lines = ["# 竞品价格日报 " + datetime.now().strftime("%Y-%m-%d"), ""]
    for r in all_records:
        old = BASELINE.get(r.product)
        if old is None:
            continue
        diff = round(r.price - old, 2)
        if diff < 0:
            flag = f"🔴 降价 {abs(diff)} 元！"       # 降价标红
        elif diff > 0:
            flag = f"🟢 涨价 {diff} 元"
        else:
            flag = "➖ 持平"
        lines.append(f"- 【{r.platform}】{r.product}：¥{r.price}（{flag}）优惠券：{r.coupon}")

    daily = "\n".join(lines)
    with open(f"price_report_{datetime.now():%Y%m%d}.md", "w", encoding="utf-8") as f:
        f.write(daily)
    print(daily)
    # 生产环境里，这里接企业微信/钉钉 webhook 把日报推给老板

asyncio.run(main())

# 每天早上 7:00 执行价格监控
0 7 * * * cd /opt/price-monitor && /usr/bin/python3 monitor.py >> logs/run.log 2>&1