Browser Use 网页自动化 Agent 完全教程
从原理到实战，掌握 LLM 驱动的智能浏览器自动化技术
目录
- Browser Use 技术原理
- Browser Use 框架安装与配置
- Playwright/Puppeteer 集成
- 网页元素定位与交互
- 表单填写与数据提取
- 多页面导航与状态管理
- 反爬虫对抗与指纹伪装
- 与传统爬虫（Selenium）对比
- 企业级 Browser Use 应用场景
- 性能优化与成本控制
1. Browser Use 技术原理
1.1 什么是 Browser Use
Browser Use 是一种新兴的网页自动化范式，其核心思想是将大语言模型（LLM）作为"大脑"，浏览器自动化工具（如 Playwright）作为"手脚"，让 AI 自主理解网页内容并执行操作。与传统的基于 CSS 选择器或 XPath 的脚本自动化不同，Browser Use Agent 能够像人类一样"看懂"网页，并根据自然语言指令完成复杂的交互任务。
1.2 技术架构
Browser Use 的典型架构包含以下核心组件：
┌─────────────────────────────────────────────────────┐
│ 用户指令（自然语言） │
└─────────────────┬───────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────┐
│ LLM 推理引擎（决策层） │
│ ┌─────────┐ ┌──────────┐ ┌───────────────────┐ │
│ │任务规划器│ │动作生成器│ │上下文记忆管理器 │ │
│ └─────────┘ └──────────┘ └───────────────────┘ │
└─────────────────┬───────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────┐
│ 浏览器控制层（执行层） │
│ ┌──────────┐ ┌──────────┐ ┌──────────────────┐ │
│ │Playwright│ │DOM 解析器│ │截图/快照捕获器 │ │
│ └──────────┘ └──────────┘ └──────────────────┘ │
└─────────────────┬───────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────┐
│ 目标网页（Chrome/Firefox） │
└─────────────────────────────────────────────────────┘
工作流程：
- 感知阶段：Agent 通过浏览器获取当前页面的 DOM 快照或截图
- 理解阶段：LLM 分析页面内容，理解当前页面状态
- 决策阶段：LLM 根据任务目标决定下一步操作（点击、输入、滚动等）
- 执行阶段：浏览器自动化工具执行具体操作
- 循环：重复上述步骤直到任务完成或达到最大步数
1.3 与传统自动化的本质区别
| 特性 | 传统自动化（Selenium/Playwright） | Browser Use Agent |
|---|---|---|
| 元素定位 | 硬编码选择器（CSS/XPath） | AI 理解语义自动定位 |
| 容错能力 | 页面结构变化即失败 | 自适应页面变化 |
| 开发方式 | 编写确定性脚本 | 自然语言描述任务 |
| 维护成本 | 高（频繁更新选择器） | 低（LLM 自动适应） |
| 适用场景 | 固定流程的重复任务 | 动态、复杂的探索任务 |
| 执行速度 | 快（直接操作） | 较慢（需 LLM 推理） |
1.4 核心 Prompt 工程
Browser Use 的核心在于给 LLM 提供高质量的上下文。典型的系统提示词包含：
SYSTEM_PROMPT = """你是一个网页自动化助手。你将看到当前浏览器页面的内容。
你的任务是根据用户指令，决定下一步应该执行的操作。
可用操作：
- click(element_id): 点击页面元素
- type(element_id, text): 在输入框中输入文字
- scroll(direction): 滚动页面（up/down）
- navigate(url): 导航到指定URL
- wait(seconds): 等待指定秒数
- done(result): 任务完成，返回结果
当前页面元素：
{dom_snapshot}
用户任务：{task}
已执行步骤：{history}
请输出下一步操作（JSON格式）：
"""
2. Browser Use 框架安装与配置
2.1 环境准备
# 创建 Python 虚拟环境（推荐 Python 3.10+）
python -m venv browser-use-env
source browser-use-env/bin/activate # Linux/Mac
# browser-use-env\Scripts\activate # Windows
# 安装核心依赖
pip install browser-use playwright
# 安装浏览器驱动
playwright install chromium
# 或安装所有浏览器
# playwright install
2.2 配置 LLM 后端
Browser Use 支持多种 LLM 后端，以下是最常用的配置方式：
# config.py
import os
# OpenAI 配置
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "your-api-key-here")
# 也可以使用本地模型（如 Ollama）
OLLAMA_BASE_URL = "http://localhost:11434/v1"
OLLAMA_MODEL = "qwen2.5:14b"
# 或使用其他兼容 OpenAI API 的服务
CUSTOM_API_BASE = "https://your-api-endpoint.com/v1"
2.3 Browser Use 基础配置
# browser_config.py
from browser_use import Browser, BrowserConfig
# 基础配置
browser_config = BrowserConfig(
headless=False, # 是否无头模式（调试时建议 False）
disable_security=True, # 禁用同源策略（某些场景需要）
extra_chromium_args=[ # 额外的 Chromium 启动参数
"--disable-blink-features=AutomationControlled",
"--no-sandbox",
],
new_context_config={
"viewport": {"width": 1920, "height": 1080},
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
}
)
browser = Browser(config=browser_config)
2.4 快速上手示例
# quickstart.py
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI
async def main():
# 初始化 LLM
llm = ChatOpenAI(
model="gpt-4o",
temperature=0,
api_key="your-api-key"
)
# 创建 Agent
agent = Agent(
task="打开百度，搜索 'Browser Use 教程'，返回前3个搜索结果的标题和链接",
llm=llm,
)
# 执行任务
result = await agent.run()
print(f"任务结果: {result}")
if __name__ == "__main__":
asyncio.run(main())
3. Playwright/Puppeteer 集成
3.1 Playwright 深度集成
Playwright 是 Browser Use 最常用的底层引擎。以下是深度集成示例：
# playwright_integration.py
import asyncio
from playwright.async_api import async_playwright, Page, BrowserContext
class PlaywrightBrowser:
"""封装 Playwright 提供更精细的浏览器控制"""
def __init__(self, headless: bool = False):
self.headless = headless
self.playwright = None
self.browser = None
self.context = None
self.page = None
async def launch(self):
"""启动浏览器"""
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch(
headless=self.headless,
args=[
"--disable-blink-features=AutomationControlled",
"--disable-dev-shm-usage",
"--no-sandbox",
]
)
self.context = await self.browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent=(
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36"
),
locale="zh-CN",
timezone_id="Asia/Shanghai",
)
# 注入反检测脚本
await self.context.add_init_script("""
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3] });
Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh', 'en'] });
window.chrome = { runtime: {} };
""")
self.page = await self.context.new_page()
return self.page
async def get_dom_snapshot(self) -> str:
"""获取当前页面的 DOM 快照，用于 LLM 分析"""
snapshot = await self.page.evaluate("""
() => {
const elements = document.querySelectorAll(
'a, button, input, textarea, select, [role="button"], [onclick]'
);
return Array.from(elements).map((el, idx) => ({
id: idx,
tag: el.tagName.toLowerCase(),
text: el.innerText?.slice(0, 100) || '',
type: el.type || '',
name: el.name || '',
placeholder: el.placeholder || '',
href: el.href || '',
visible: el.offsetParent !== null,
})).filter(el => el.visible);
}
""")
# 格式化为 LLM 可读的文本
lines = ["页面可交互元素列表："]
for el in snapshot:
line = f"[{el['id']}] <{el['tag']}>"
if el['text']:
line += f" 文本: \"{el['text']}\""
if el['placeholder']:
line += f" 占位符: \"{el['placeholder']}\""
if el['href']:
line += f" 链接: {el['href']}"
lines.append(line)
return "\n".join(lines)
async def take_screenshot(self, path: str = "screenshot.png"):
"""截图用于视觉分析"""
await self.page.screenshot(path=path, full_page=False)
return path
async def execute_action(self, action: dict):
"""执行 LLM 决策的操作"""
action_type = action.get("type")
if action_type == "click":
element_id = action["element_id"]
selector = f"xpath=(//*[@tabindex or @role or @href or @onclick])[{element_id + 1}]"
await self.page.click(selector, timeout=5000)
elif action_type == "type":
element_id = action["element_id"]
text = action["text"]
selector = f"input, textarea"
elements = await self.page.query_selector_all(selector)
if element_id < len(elements):
await elements[element_id].fill(text)
elif action_type == "scroll":
direction = action.get("direction", "down")
delta = 500 if direction == "down" else -500
await self.page.mouse.wheel(0, delta)
elif action_type == "navigate":
await self.page.goto(action["url"], wait_until="domcontentloaded")
elif action_type == "wait":
await asyncio.sleep(action.get("seconds", 2))
async def close(self):
"""清理资源"""
if self.context:
await self.context.close()
if self.browser:
await self.browser.close()
if self.playwright:
await self.playwright.stop()
3.2 Puppeteer（Node.js）集成方案
对于 Node.js 生态，可以使用 Puppeteer 实现类似功能：
// puppeteer-browser.js
const puppeteer = require('puppeteer');
class PuppeteerBrowser {
constructor(options = {}) {
this.headless = options.headless ?? false;
this.browser = null;
this.page = null;
}
async launch() {
this.browser = await puppeteer.launch({
headless: this.headless,
args: [
'--disable-blink-features=AutomationControlled',
'--no-sandbox',
'--disable-dev-shm-usage',
],
});
this.page = await this.browser.newPage();
// 设置视口
await this.page.setViewport({ width: 1920, height: 1080 });
// 设置 User-Agent
await this.page.setUserAgent(
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
);
// 注入反检测
await this.page.evaluateOnNewDocument(() => {
Object.defineProperty(navigator, 'webdriver', { get: () => false });
window.chrome = { runtime: {} };
});
return this.page;
}
async getDomSnapshot() {
return await this.page.evaluate(() => {
const elements = document.querySelectorAll(
'a, button, input, textarea, select, [role="button"]'
);
return Array.from(elements)
.filter(el => el.offsetParent !== null)
.map((el, idx) => ({
id: idx,
tag: el.tagName.toLowerCase(),
text: (el.innerText || '').slice(0, 100),
type: el.type || '',
placeholder: el.placeholder || '',
}));
});
}
async close() {
if (this.browser) await this.browser.close();
}
}
module.exports = PuppeteerBrowser;
4. 网页元素定位与交互
4.1 多策略元素定位
在 Browser Use 中，元素定位不是靠硬编码选择器，而是通过 AI 理解语义来定位。以下是实现这一能力的核心策略：
# element_locator.py
from dataclasses import dataclass
from typing import Optional
from playwright.async_api import Page
@dataclass
class ElementInfo:
index: int
tag: str
text: str
role: str
attributes: dict
bbox: Optional[dict] = None
class SmartLocator:
"""智能元素定位器 - 结合多种策略"""
def __init__(self, page: Page):
self.page = page
async def build_element_map(self) -> list[ElementInfo]:
"""构建页面元素映射表，供 LLM 参考"""
return await self.page.evaluate("""
() => {
const interactable = document.querySelectorAll(`
a[href], button, input, textarea, select,
[role="button"], [role="link"], [role="tab"],
[role="menuitem"], [role="checkbox"], [role="radio"],
[onclick], [tabindex]:not([tabindex="-1"])
`);
return Array.from(interactable)
.filter(el => {
const rect = el.getBoundingClientRect();
return rect.width > 0 && rect.height > 0 &&
rect.top >= 0 && rect.top < window.innerHeight;
})
.map((el, idx) => {
const rect = el.getBoundingClientRect();
const computed = window.getComputedStyle(el);
return {
index: idx,
tag: el.tagName.toLowerCase(),
text: (el.innerText || '').trim().slice(0, 200),
role: el.getAttribute('role') || '',
attributes: {
id: el.id || '',
name: el.name || '',
type: el.type || '',
class: el.className?.toString().slice(0, 100) || '',
placeholder: el.placeholder || '',
href: el.href || '',
ariaLabel: el.getAttribute('aria-label') || '',
value: el.value?.slice(0, 100) || '',
},
bbox: {
x: Math.round(rect.x),
y: Math.round(rect.y),
width: Math.round(rect.width),
height: Math.round(rect.height),
}
};
});
}
""")
async def locate_by_semantic(self, description: str, element_map: list[dict]) -> Optional[int]:
"""根据语义描述定位元素（需配合 LLM 使用）"""
# 将元素映射格式化为 LLM 可理解的文本
prompt = f"""根据以下描述，找到最匹配的页面元素。
描述：{description}
可选元素：
"""
for el in element_map:
prompt += f"[{el['index']}] <{el['tag']}>"
if el['text']:
prompt += f" \"{el['text']}\""
if el['attributes'].get('ariaLabel'):
prompt += f" aria-label=\"{el['attributes']['ariaLabel']}\""
if el['attributes'].get('placeholder'):
prompt += f" placeholder=\"{el['attributes']['placeholder']}\""
prompt += "\n"
prompt += "\n请只返回最匹配的元素索引编号（纯数字）。"
# 此处调用 LLM 获取结果
# response = await llm.ainvoke(prompt)
# return int(response.content.strip())
return None # 示例占位
async def locate_by_text(self, text: str) -> Optional[int]:
"""通过文本内容定位元素"""
elements = await self.build_element_map()
for el in elements:
if text.lower() in el['text'].lower():
return el['index']
return None
async def locate_by_role(self, role: str, name: str = "") -> Optional[int]:
"""通过 ARIA 角色定位元素"""
selector = f'[role="{role}"]'
if name:
selector += f'[aria-label*="{name}"], [role="{role}"]'
elements = await self.page.query_selector_all(selector)
if elements:
return 0 # 返回第一个匹配
return None
4.2 交互操作封装
# interactions.py
import asyncio
from playwright.async_api import Page, TimeoutError as PlaywrightTimeout
class BrowserInteractions:
"""封装常用浏览器交互操作"""
def __init__(self, page: Page):
self.page = page
async def safe_click(self, selector: str, timeout: int = 5000) -> bool:
"""安全点击 - 等待元素可见后再点击"""
try:
await self.page.wait_for_selector(selector, state="visible", timeout=timeout)
await self.page.click(selector, timeout=timeout)
return True
except PlaywrightTimeout:
print(f"⚠️ 点击超时: {selector}")
return False
async def smart_type(self, selector: str, text: str, clear_first: bool = True):
"""智能输入 - 先清空再输入"""
await self.page.wait_for_selector(selector, state="visible")
element = await self.page.query_selector(selector)
if element:
if clear_first:
await element.click(click_count=3) # 全选
await self.page.keyboard.press("Backspace")
await element.type(text, delay=50) # 模拟真实打字速度
async def wait_and_click_text(self, text: str, timeout: int = 10000) -> bool:
"""等待包含指定文本的元素出现并点击"""
try:
await self.page.wait_for_function(
f"""() => {{
const elements = document.querySelectorAll('a, button, span, div');
return Array.from(elements).some(el =>
el.innerText.trim().includes('{text}') &&
el.offsetParent !== null
);
}}""",
timeout=timeout
)
# 找到并点击
await self.page.click(f"text={text}")
return True
except PlaywrightTimeout:
print(f"⚠️ 未找到包含文本 '{text}' 的元素")
return False
async def scroll_to_bottom(self, max_scrolls: int = 10):
"""滚动到页面底部"""
previous_height = 0
for i in range(max_scrolls):
current_height = await self.page.evaluate("document.body.scrollHeight")
if current_height == previous_height:
break
await self.page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
await asyncio.sleep(1)
previous_height = current_height
async def handle_popup(self, action: str = "accept"):
"""处理弹窗（alert/confirm/prompt）"""
self.page.on("dialog", lambda dialog: dialog.accept() if action == "accept" else dialog.dismiss())
5. 表单填写与数据提取
5.1 智能表单填写
# form_filler.py
import asyncio
from typing import Any
from playwright.async_api import Page
class SmartFormFiller:
"""智能表单填写器 - 配合 LLM 理解表单结构"""
def __init__(self, page: Page):
self.page = page
async def analyze_form(self) -> list[dict]:
"""分析页面表单结构"""
forms = await self.page.evaluate("""
() => {
const forms = document.querySelectorAll('form');
return Array.from(forms).map((form, formIdx) => ({
formIndex: formIdx,
action: form.action || '',
method: form.method || 'GET',
fields: Array.from(form.querySelectorAll(
'input, textarea, select'
)).map((field, fieldIdx) => ({
fieldIndex: fieldIdx,
tag: field.tagName.toLowerCase(),
type: field.type || 'text',
name: field.name || '',
id: field.id || '',
placeholder: field.placeholder || '',
label: field.labels?.[0]?.innerText?.trim() || '',
required: field.required,
value: field.value || '',
options: field.tagName === 'SELECT'
? Array.from(field.options).map(opt => ({
value: opt.value, text: opt.text
}))
: []
}))
}));
}
""")
return forms
async def fill_form_by_mapping(self, form_data: dict[str, Any]):
"""
根据映射关系填写表单
form_data 示例: {
"username": "张三",
"email": "zhangsan@example.com",
"phone": "13800138000",
"city": "北京"
}
"""
forms = await self.analyze_form()
for form in forms:
for field in form['fields']:
field_name = field['name'] or field['id'] or field['placeholder']
# 尝试匹配表单数据
matched_value = None
for key, value in form_data.items():
if key.lower() in field_name.lower() or \
key.lower() in field['label'].lower():
matched_value = value
break
if matched_value is None:
continue
# 构建选择器
if field['id']:
selector = f"#{field['id']}"
elif field['name']:
selector = f"[name='{field['name']}']"
else:
continue
# 根据字段类型填写
if field['tag'] == 'select':
await self.page.select_option(selector, label=str(matched_value))
elif field['type'] == 'checkbox':
if matched_value and not await self.page.is_checked(selector):
await self.page.check(selector)
elif not matched_value and await self.page.is_checked(selector):
await self.page.uncheck(selector)
elif field['type'] == 'radio':
await self.page.check(f"{selector}[value='{matched_value}']")
else:
await self.page.fill(selector, str(matched_value))
await asyncio.sleep(0.3) # 模拟人类操作间隔
async def extract_form_data(self) -> dict:
"""提取当前表单中已填写的数据"""
return await self.page.evaluate("""
() => {
const data = {};
const forms = document.querySelectorAll('form');
forms.forEach(form => {
const formData = new FormData(form);
for (const [key, value] of formData.entries()) {
data[key] = value;
}
});
return data;
}
""")
5.2 高级数据提取
# data_extractor.py
import json
from typing import Optional
from playwright.async_api import Page
class DataExtractor:
"""从网页中提取结构化数据"""
def __init__(self, page: Page):
self.page = page
async def extract_table(self, table_selector: str = "table") -> list[dict]:
"""提取表格数据为字典列表"""
return await self.page.evaluate(f"""
(selector) => {{
const table = document.querySelector(selector);
if (!table) return [];
const headers = Array.from(table.querySelectorAll('th'))
.map(th => th.innerText.trim());
const rows = Array.from(table.querySelectorAll('tbody tr'));
return rows.map(row => {{
const cells = Array.from(row.querySelectorAll('td'))
.map(td => td.innerText.trim());
const obj = {{}};
headers.forEach((h, i) => obj[h] = cells[i] || '');
return obj;
}});
}}
""", table_selector)
async def extract_list(self, list_selector: str) -> list[str]:
"""提取列表数据"""
return await self.page.evaluate("""
(selector) => {
return Array.from(document.querySelectorAll(selector))
.map(el => el.innerText.trim())
.filter(text => text.length > 0);
}
""", list_selector)
async def extract_structured_data(self, schema: dict) -> dict:
"""
根据自定义 schema 提取结构化数据
schema 示例: {
"title": "h1",
"price": ".price-value",
"description": ".product-desc",
"images": ["img.product-image", "src"]
}
"""
result = {}
for key, selector in schema.items():
if isinstance(selector, list):
# 提取属性值列表
result[key] = await self.page.evaluate("""
(sel) => Array.from(document.querySelectorAll(sel[0]))
.map(el => el.getAttribute(sel[1]) || '')
""", selector)
elif selector.startswith("meta["):
# 提取 meta 标签
result[key] = await self.page.evaluate(f"""
() => document.querySelector('{selector}')?.content || ''
""")
else:
result[key] = await self.page.evaluate(f"""
() => document.querySelector('{selector}')?.innerText?.trim() || ''
""")
return result
async def extract_json_ld(self) -> list[dict]:
"""提取页面中的 JSON-LD 结构化数据"""
return await self.page.evaluate("""
() => {
const scripts = document.querySelectorAll('script[type="application/ld+json"]');
return Array.from(scripts).map(s => {
try { return JSON.parse(s.textContent); }
catch { return null; }
}).filter(Boolean);
}
""")
async def extract_infinite_scroll_data(
self,
item_selector: str,
extract_fn: str,
max_items: int = 100,
scroll_pause: float = 2.0
) -> list:
"""处理无限滚动页面的数据提取"""
import asyncio
collected = []
seen_count = 0
while len(collected) < max_items:
# 提取当前可见数据
items = await self.page.evaluate(f"""
() => {{
const items = document.querySelectorAll('{item_selector}');
return Array.from(items).slice({seen_count}).map(el => {{
{extract_fn}
}});
}}
""")
if not items:
# 滚动加载更多
await self.page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
await asyncio.sleep(scroll_pause)
# 检查是否有新内容
new_count = await self.page.evaluate(f"""
() => document.querySelectorAll('{item_selector}').length
""")
if new_count <= seen_count:
break # 没有更多数据
continue
collected.extend(items)
seen_count += len(items)
return collected[:max_items]
6. 多页面导航与状态管理
6.1 会话状态管理
# state_manager.py
import json
import asyncio
from dataclasses import dataclass, field, asdict
from typing import Optional
from playwright.async_api import Page, BrowserContext
@dataclass
class NavigationState:
"""导航状态记录"""
current_url: str = ""
history: list[str] = field(default_factory=list)
cookies: dict = field(default_factory=dict)
local_storage: dict = field(default_factory=dict)
session_storage: dict = field(default_factory=dict)
page_count: int = 0
errors: list[str] = field(default_factory=list)
class MultiPageManager:
"""多页面管理器"""
def __init__(self, context: BrowserContext):
self.context = context
self.pages: dict[str, Page] = {}
self.state = NavigationState()
self._active_page: Optional[Page] = None
async def open_page(self, name: str, url: str) -> Page:
"""打开新页面并注册命名"""
page = await self.context.new_page()
await page.goto(url, wait_until="domcontentloaded")
self.pages[name] = page
self.state.page_count += 1
self.state.history.append(url)
self._active_page = page
return page
async def switch_page(self, name: str) -> Page:
"""切换到指定页面"""
if name not in self.pages:
raise ValueError(f"页面 '{name}' 不存在")
page = self.pages[name]
await page.bring_to_front()
self._active_page = page
return page
async def close_page(self, name: str):
"""关闭指定页面"""
if name in self.pages:
await self.pages[name].close()
del self.pages[name]
if self._active_page == self.pages.get(name):
self._active_page = None
async def save_state(self, path: str = "browser_state.json"):
"""保存当前浏览器状态"""
if self._active_page:
self.state.current_url = self._active_page.url
self.state.cookies = await self.context.cookies()
self.state.local_storage = await self._active_page.evaluate(
"() => ({...localStorage})"
)
self.state.session_storage = await self._active_page.evaluate(
"() => ({...sessionStorage})"
)
with open(path, "w", encoding="utf-8") as f:
json.dump(asdict(self.state), f, ensure_ascii=False, indent=2)
async def restore_state(self, path: str = "browser_state.json"):
"""从文件恢复浏览器状态"""
with open(path, "r", encoding="utf-8") as f:
saved = json.load(f)
# 恢复 cookies
if saved.get("cookies"):
await self.context.add_cookies(saved["cookies"])
# 恢复 localStorage
if saved.get("local_storage") and self._active_page:
for key, value in saved["local_storage"].items():
await self._active_page.evaluate(
f"(k, v) => localStorage.setItem(k, v)",
[key, value]
)
# 导航到之前的 URL
if saved.get("current_url") and self._active_page:
await self._active_page.goto(saved["current_url"])
async def parallel_navigate(self, tasks: list[dict]) -> dict[str, any]:
"""并行导航多个页面"""
async def navigate_one(name, url):
page = await self.open_page(name, url)
return name, await page.title()
results = await asyncio.gather(
*[navigate_one(t['name'], t['url']) for t in tasks],
return_exceptions=True
)
return {r[0]: r[1] for r in results if not isinstance(r, Exception)}
6.2 多 Tab 工作流
# tab_workflow.py
import asyncio
from playwright.async_api import BrowserContext
class TabWorkflow:
"""基于多 Tab 的复杂工作流编排"""
def __init__(self, context: BrowserContext):
self.context = context
self.results = {}
async def workflow_compare_prices(self, product_name: str, sites: list[dict]):
"""
价格对比工作流：同时打开多个电商网站搜索同一商品
sites: [{"name": "京东", "url": "https://jd.com", "search_selector": "#key"}, ...]
"""
async def search_site(site_info):
page = await self.context.new_page()
try:
await page.goto(site_info['url'], timeout=15000)
await page.fill(site_info['search_selector'], product_name)
await page.press(site_info['search_selector'], "Enter")
await page.wait_for_load_state("networkidle", timeout=10000)
# 提取价格（需根据具体网站调整选择器）
prices = await page.evaluate("""
(selector) => {
return Array.from(document.querySelectorAll(selector))
.slice(0, 5)
.map(el => el.innerText.trim());
}
""", site_info.get('price_selector', '.price'))
return site_info['name'], prices
except Exception as e:
return site_info['name'], f"错误: {str(e)}"
finally:
await page.close()
# 并行执行所有搜索
results = await asyncio.gather(
*[search_site(site) for site in sites]
)
return {name: prices for name, prices in results}
7. 反爬虫对抗与指纹伪装
7.1 浏览器指纹伪装
# anti_detection.py
import random
from playwright.async_api import BrowserContext
class StealthBrowser:
"""反检测浏览器 - 伪装真实用户行为"""
# 常见的 User-Agent 池
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.2 Safari/605.1.15",
]
# 常见屏幕分辨率
VIEWPORTS = [
{"width": 1920, "height": 1080},
{"width": 1366, "height": 768},
{"width": 1536, "height": 864},
{"width": 1440, "height": 900},
{"width": 2560, "height": 1440},
]
@classmethod
async def setup_stealth_context(cls, context: BrowserContext):
"""为浏览器上下文注入反检测脚本"""
# 隐藏 webdriver 特征
await context.add_init_script("""
// 1. 隐藏 webdriver 属性
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined,
});
// 2. 伪装 plugins
Object.defineProperty(navigator, 'plugins', {
get: () => {
const plugins = [
{ name: 'Chrome PDF Plugin', filename: 'internal-pdf-viewer' },
{ name: 'Chrome PDF Viewer', filename: 'mhjfbmdgcfjbbpaeojofohoefgiehjai' },
{ name: 'Native Client', filename: 'internal-nacl-plugin' },
];
plugins.length = 3;
return plugins;
},
});
// 3. 伪装 languages
Object.defineProperty(navigator, 'languages', {
get: () => ['zh-CN', 'zh', 'en-US', 'en'],
});
// 4. 伪装 chrome 对象
window.chrome = {
runtime: {},
loadTimes: function() {},
csi: function() {},
app: {},
};
// 5. 隐藏自动化特征
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) =>
parameters.name === 'notifications'
? Promise.resolve({ state: Notification.permission })
: originalQuery(parameters);
// 6. 伪装 WebGL 渲染器
const getParameter = WebGLRenderingContext.prototype.getParameter;
WebGLRenderingContext.prototype.getParameter = function(parameter) {
if (parameter === 37445) return 'Intel Inc.';
if (parameter === 37446) return 'Intel Iris OpenGL Engine';
return getParameter.call(this, parameter);
};
// 7. 修改 canvas 指纹
const toDataURL = HTMLCanvasElement.prototype.toDataURL;
HTMLCanvasElement.prototype.toDataURL = function(type) {
if (type === 'image/png') {
// 添加微小噪声改变 canvas 指纹
const ctx = this.getContext('2d');
if (ctx) {
const style = ctx.fillStyle;
ctx.fillStyle = 'rgba(0,0,0,0.01)';
ctx.fillRect(0, 0, 1, 1);
ctx.fillStyle = style;
}
}
return toDataURL.apply(this, arguments);
};
""")
@classmethod
async def human_like_delay(cls, min_ms: int = 100, max_ms: int = 500):
"""模拟人类操作延迟"""
import asyncio
delay = random.uniform(min_ms / 1000, max_ms / 1000)
await asyncio.sleep(delay)
@classmethod
async def human_like_mouse_move(cls, page, target_x: int, target_y: int):
"""模拟人类鼠标移动轨迹（贝塞尔曲线）"""
# 获取当前位置（或使用默认起点）
start_x = random.randint(100, 500)
start_y = random.randint(100, 500)
# 生成贝塞尔曲线控制点
cp1_x = start_x + (target_x - start_x) * random.uniform(0.2, 0.4)
cp1_y = start_y + (target_y - start_y) * random.uniform(0.1, 0.3)
cp2_x = start_x + (target_x - start_x) * random.uniform(0.6, 0.8)
cp2_y = start_y + (target_y - start_y) * random.uniform(0.7, 0.9)
steps = random.randint(10, 25)
for i in range(steps + 1):
t = i / steps
# 贝塞尔曲线公式
x = (1-t)**3 * start_x + 3*(1-t)**2*t * cp1_x + 3*(1-t)*t**2 * cp2_x + t**3 * target_x
y = (1-t)**3 * start_y + 3*(1-t)**2*t * cp1_y + 3*(1-t)*t**2 * cp2_y + t**3 * target_y
await page.mouse.move(x, y)
await asyncio.sleep(random.uniform(0.01, 0.03))
@classmethod
def get_random_config(cls) -> dict:
"""获取随机浏览器配置"""
return {
"user_agent": random.choice(cls.USER_AGENTS),
"viewport": random.choice(cls.VIEWPORTS),
"locale": random.choice(["zh-CN", "zh-TW", "en-US"]),
"timezone_id": random.choice(["Asia/Shanghai", "Asia/Tokyo", "America/New_York"]),
}
7.2 代理与 IP 轮换
# proxy_manager.py
import random
from typing import Optional
class ProxyManager:
"""代理 IP 管理器"""
def __init__(self, proxy_list: list[dict]):
"""
proxy_list 格式: [
{"server": "http://proxy1:8080", "username": "user", "password": "pass"},
{"server": "socks5://proxy2:1080"},
...
]
"""
self.proxies = proxy_list
self.current_index = 0
self.failed_proxies = set()
def get_next_proxy(self) -> Optional[dict]:
"""获取下一个可用代理"""
available = [p for i, p in enumerate(self.proxies) if i not in self.failed_proxies]
if not available:
# 重置失败列表
self.failed_proxies.clear()
available = self.proxies
proxy = random.choice(available)
return {
"server": proxy["server"],
"username": proxy.get("username"),
"password": proxy.get("password"),
}
def mark_failed(self, proxy_server: str):
"""标记代理为失败"""
for i, p in enumerate(self.proxies):
if p["server"] == proxy_server:
self.failed_proxies.add(i)
break
async def get_context_with_proxy(self, browser, config: dict = None):
"""使用随机代理创建浏览器上下文"""
proxy = self.get_next_proxy()
context_config = {
"proxy": proxy,
**(config or {})
}
return await browser.new_context(**context_config)
8. 与传统爬虫（Selenium）对比
8.1 全面对比分析
| 维度 | Selenium | Browser Use (Playwright) | 说明 |
|---|---|---|---|
| 架构 | WebDriver 协议 | CDP (Chrome DevTools Protocol) | CDP 更现代、更高效 |
| 速度 | 较慢 | 快 2-3 倍 | Playwright 原生异步 |
| 等待机制 | 显式/隐式等待 | 自动等待（Auto-waiting） | Playwright 自动等待元素就绪 |
| 浏览器支持 | Chrome/Firefox/Safari/Edge | Chrome/Firefox/Safari/WebKit | 覆盖面相当 |
| 语言支持 | Java/Python/C#/Ruby/JS | Python/JS/Java/C# | 覆盖面相当 |
| 移动端测试 | Appium 集成 | 原生移动端模拟 | Selenium 生态更成熟 |
| AI 集成 | 需自行实现 | 原生支持 LLM Agent | Browser Use 天然优势 |
| 社区生态 | 极其成熟（20年） | 快速增长中 | Selenium 更稳定 |
| 学习曲线 | 中等 | 低（自然语言驱动） | Browser Use 更友好 |
| 维护性 | 差（选择器易失效） | 好（AI 自适应） | Browser Use 显著优势 |
8.2 迁移指南：从 Selenium 到 Browser Use
# 迁移示例
# ===== Selenium 旧代码 =====
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
driver = webdriver.Chrome()
driver.get("https://example.com/login")
# 硬编码选择器
username_input = WebDriverWait(driver, 10).until(
EC.presence_of_element_located((By.CSS_SELECTOR, "#username"))
)
username_input.send_keys("my_user")
password_input = driver.find_element(By.CSS_SELECTOR, "#password")
password_input.send_keys("my_pass")
login_button = driver.find_element(By.XPATH, "//button[contains(text(), '登录')]")
login_button.click()
WebDriverWait(driver, 10).until(
EC.url_contains("/dashboard")
)
# 提取数据
items = driver.find_elements(By.CSS_SELECTOR, ".item-card")
for item in items:
title = item.find_element(By.CSS_SELECTOR, ".title").text
price = item.find_element(By.CSS_SELECTOR, ".price").text
print(f"{title}: {price}")
driver.quit()
# ===== Browser Use 新代码 =====
import asyncio
from browser_use import Agent
from langchain_openai import ChatOpenAI
async def main():
agent = Agent(
task="""
1. 打开 https://example.com/login
2. 在用户名输入框输入 'my_user'
3. 在密码输入框输入 'my_pass'
4. 点击登录按钮
5. 等待进入仪表盘页面
6. 提取所有商品卡片的标题和价格
7. 返回提取的数据
""",
llm=ChatOpenAI(model="gpt-4o", temperature=0),
)
result = await agent.run()
print(result)
asyncio.run(main())
8.3 混合策略：传统 + AI
在实际项目中，最佳实践往往是混合使用：
# hybrid_approach.py
import asyncio
from playwright.async_api import Page
class HybridAutomation:
"""混合策略：固定流程用传统方式，动态部分用 AI"""
def __init__(self, page: Page, llm):
self.page = page
self.llm = llm
async def login(self, username: str, password: str):
"""登录 - 固定流程，直接用选择器"""
await self.page.goto("https://example.com/login")
await self.page.fill("#username", username)
await self.page.fill("#password", password)
await self.page.click("button[type='submit']")
await self.page.wait_for_url("**/dashboard**")
async def search_and_extract(self, query: str):
"""搜索并提取 - 动态页面，用 AI 理解"""
# AI 分析页面结构
snapshot = await self._get_page_snapshot()
prompt = f"""当前页面快照：
{snapshot}
任务：搜索 "{query}" 并提取前5条结果
请返回操作步骤（JSON 格式）"""
actions = await self.llm.ainvoke(prompt)
# 执行 AI 返回的操作序列...
async def _get_page_snapshot(self) -> str:
"""获取页面快照"""
return await self.page.evaluate("""
() => document.body.innerText.slice(0, 3000)
""")
9. 企业级 Browser Use 应用场景
9.1 场景一：智能客服系统
# smart_customer_service.py
import asyncio
from dataclasses import dataclass
@dataclass
class CustomerQuery:
user_id: str
question: str
context: dict
class SmartCustomerService:
"""基于 Browser Use 的智能客服系统"""
def __init__(self, llm, knowledge_base_url: str):
self.llm = llm
self.kb_url = knowledge_base_url
async def handle_query(self, query: CustomerQuery) -> str:
"""处理客户查询"""
# 1. 在知识库中搜索
kb_results = await self._search_knowledge_base(query.question)
# 2. 在工单系统中查找历史
ticket_history = await self._search_tickets(query.user_id)
# 3. 在产品文档中查找
docs = await self._search_docs(query.question)
# 4. AI 综合分析并回答
prompt = f"""客户问题：{query.question}
知识库结果：{kb_results}
历史工单：{ticket_history}
产品文档：{docs}
请生成专业、友好的客服回复。"""
response = await self.llm.ainvoke(prompt)
return response.content
async def _search_knowledge_base(self, query: str) -> str:
"""通过 Browser Use 搜索知识库"""
from browser_use import Agent
agent = Agent(
task=f"在知识库中搜索 '{query}' 并返回相关结果",
llm=self.llm,
)
return await agent.run()
async def _search_tickets(self, user_id: str) -> str:
"""搜索工单系统"""
# 类似实现...
return ""
async def _search_docs(self, query: str) -> str:
"""搜索产品文档"""
# 类似实现...
return ""
9.2 场景二：竞品监控系统
# competitor_monitor.py
import asyncio
import json
from datetime import datetime
class CompetitorMonitor:
"""竞品价格和动态监控"""
def __init__(self, llm, competitors: list[dict]):
self.llm = llm
self.competitors = competitors # [{"name": "竞品A", "url": "...", "selectors": {...}}]
async def daily_check(self) -> dict:
"""每日竞品检查"""
results = {}
for competitor in self.competitors:
try:
data = await self._scrape_competitor(competitor)
results[competitor['name']] = {
"timestamp": datetime.now().isoformat(),
"data": data,
"status": "success"
}
except Exception as e:
results[competitor['name']] = {
"timestamp": datetime.now().isoformat(),
"error": str(e),
"status": "failed"
}
# AI 分析竞品变化
analysis = await self._analyze_changes(results)
results["analysis"] = analysis
return results
async def _scrape_competitor(self, competitor: dict) -> dict:
"""抓取单个竞品数据"""
from browser_use import Agent
agent = Agent(
task=f"""
访问 {competitor['url']}
提取以下信息：
1. 首页主要产品/服务及价格
2. 最新公告或新闻
3. 促销活动信息
4. 页面设计的主要变化
以 JSON 格式返回结果。
""",
llm=self.llm,
)
result = await agent.run()
return json.loads(result) if result else {}
async def _analyze_changes(self, results: dict) -> str:
"""AI 分析竞品变化"""
prompt = f"""以下是今日竞品监控数据：
{json.dumps(results, ensure_ascii=False, indent=2)}
请分析：
1. 各竞品的主要变化
2. 值得关注的市场趋势
3. 对我们的影响和建议"""
response = await self.llm.ainvoke(prompt)
return response.content
9.3 场景三：自动化测试 Agent
# test_agent.py
import asyncio
from typing import Optional
class AutomatedTestAgent:
"""基于 AI 的自动化测试 Agent"""
def __init__(self, llm, base_url: str):
self.llm = llm
self.base_url = base_url
self.test_results = []
async def run_test_suite(self, test_cases: list[dict]):
"""执行测试套件"""
for case in test_cases:
result = await self._execute_test_case(case)
self.test_results.append(result)
status = "✅ PASS" if result['passed'] else "❌ FAIL"
print(f"{status} [{case['id']}] {case['name']}")
# 生成测试报告
report = await self._generate_report()
return report
async def _execute_test_case(self, case: dict) -> dict:
"""执行单个测试用例"""
from browser_use import Agent
agent = Agent(
task=f"""
测试目标：{case['name']}
前置条件：{case.get('precondition', '无')}
测试步骤：
{chr(10).join(f'{i+1}. {step}' for i, step in enumerate(case['steps']))}
预期结果：{case['expected']}
请执行测试并判断是否通过。返回 JSON 格式：
{{"passed": true/false, "actual_result": "...", "screenshot": "..."}}
""",
llm=self.llm,
)
try:
result = await agent.run()
return {
"case_id": case['id'],
"passed": "PASS" in str(result) or "true" in str(result).lower(),
"details": result,
}
except Exception as e:
return {
"case_id": case['id'],
"passed": False,
"error": str(e),
}
async def _generate_report(self) -> str:
"""生成测试报告"""
total = len(self.test_results)
passed = sum(1 for r in self.test_results if r['passed'])
failed = total - passed
report = f"""
# 自动化测试报告
- 总计: {total}
- 通过: {passed} ✅
- 失败: {failed} ❌
- 通过率: {passed/total*100:.1f}%
## 详细结果
"""
for r in self.test_results:
status = "✅" if r['passed'] else "❌"
report += f"- {status} [{r['case_id']}] {r.get('details', r.get('error', ''))}\n"
return report
10. 性能优化与成本控制
10.1 性能优化策略
# performance.py
import asyncio
from functools import lru_cache
class PerformanceOptimizer:
"""Browser Use 性能优化器"""
@staticmethod
def optimize_dom_snapshot(full_html: str, max_tokens: int = 4000) -> str:
"""优化 DOM 快照，减少 LLM 输入 token"""
# 1. 移除 script/style 标签
import re
cleaned = re.sub(r'<(script|style)[^>]*>.*?</\1>', '', full_html, flags=re.DOTALL)
# 2. 移除注释
cleaned = re.sub(r'<!--.*?-->', '', cleaned, flags=re.DOTALL)
# 3. 移除不可见元素
cleaned = re.sub(r'<[^>]*style="[^"]*display\s*:\s*none[^"]*"[^>]*>.*?</[^>]+>',
'', cleaned, flags=re.DOTALL)
# 4. 截断到指定 token 数（粗略估算 1 token ≈ 4 字符）
max_chars = max_tokens * 4
if len(cleaned) > max_chars:
cleaned = cleaned[:max_chars] + "\n... (内容已截断)"
return cleaned
@staticmethod
async def batch_operations(page, operations: list[dict]):
"""批量执行操作，减少页面渲染等待"""
# 将多个 DOM 操作合并执行
script = "const results = [];\n"
for op in operations:
if op['type'] == 'click':
script += f"document.querySelector('{op['selector']}')?.click();\n"
elif op['type'] == 'fill':
script += f"""
const el_{op['selector'].replace('#','').replace('.','')} =
document.querySelector('{op['selector']}');
if (el_{op['selector'].replace('#','').replace('.','')}) {{
el_{op['selector'].replace('#','').replace('.','')}.value = '{op['value']}';
el_{op['selector'].replace('#','').replace('.','')}.dispatchEvent(
new Event('input', {{bubbles: true}})
);
}}
"""
script += "return results;"
return await page.evaluate(script)
@staticmethod
def estimate_cost(task_description: str, avg_steps: int = 10, model: str = "gpt-4o") -> dict:
"""估算任务成本"""
# 粗略估算每次交互的 token 消耗
tokens_per_step = {
"gpt-4o": {"input": 3000, "output": 500},
"gpt-4o-mini": {"input": 3000, "output": 500},
"claude-3-5-sonnet": {"input": 3000, "output": 500},
}
pricing = {
"gpt-4o": {"input": 2.50, "output": 10.00}, # per 1M tokens
"gpt-4o-mini": {"input": 0.15, "output": 0.60},
"claude-3-5-sonnet": {"input": 3.00, "output": 15.00},
}
tokens = tokens_per_step.get(model, tokens_per_step["gpt-4o"])
price = pricing.get(model, pricing["gpt-4o"])
total_input_tokens = tokens["input"] * avg_steps
total_output_tokens = tokens["output"] * avg_steps
cost = (total_input_tokens * price["input"] + total_output_tokens * price["output"]) / 1_000_000
return {
"model": model,
"estimated_steps": avg_steps,
"total_input_tokens": total_input_tokens,
"total_output_tokens": total_output_tokens,
"estimated_cost_usd": round(cost, 4),
"estimated_cost_cny": round(cost * 7.2, 4),
}
10.2 成本控制最佳实践
# cost_control.py
class CostController:
"""成本控制策略"""
def __init__(self, max_cost_per_task: float = 0.50, max_steps: int = 20):
self.max_cost_per_task = max_cost_per_task # 每任务最大成本（美元）
self.max_steps = max_steps
self.current_cost = 0.0
self.step_count = 0
def check_budget(self, step_cost: float) -> bool:
"""检查是否超出预算"""
self.current_cost += step_cost
self.step_count += 1
if self.current_cost >= self.max_cost_per_task:
print(f"⚠️ 已达到成本上限 ${self.max_cost_per_task}")
return False
if self.step_count >= self.max_steps:
print(f"⚠️ 已达到最大步数 {self.max_steps}")
return False
return True
@staticmethod
def get_cost_tips() -> list[str]:
return [
"1. 使用 gpt-4o-mini 替代 gpt-4o（成本降低 90%+）",
"2. 优化 DOM 快照，只传递关键元素",
"3. 缓存常见页面结构的 LLM 分析结果",
"4. 对简单任务使用规则引擎，仅复杂判断调用 LLM",
"5. 设置每任务和每日成本上限",
"6. 使用本地模型（如 Qwen2.5）处理非关键任务",
"7. 批量处理相似任务，减少重复推理",
"8. 监控并分析 token 使用情况，持续优化 prompt",
"9. 对已知结构的页面使用传统选择器，仅在不确定时调用 AI",
"10. 合理设置最大步数，避免 Agent 陷入循环",
]
10.3 本地模型替代方案
# local_model.py
from langchain_openai import ChatOpenAI
def get_cost_effective_llm(task_complexity: str = "medium"):
"""根据任务复杂度选择性价比最优的模型"""
if task_complexity == "simple":
# 简单任务：使用本地小模型
return ChatOpenAI(
model="qwen2.5:7b",
base_url="http://localhost:11434/v1",
api_key="ollama",
temperature=0,
)
elif task_complexity == "medium":
# 中等任务：使用便宜的云端模型
return ChatOpenAI(
model="gpt-4o-mini",
temperature=0,
)
else:
# 复杂任务：使用最强模型
return ChatOpenAI(
model="gpt-4o",
temperature=0,
)
# 路由策略示例
async def smart_route_task(task: str) -> str:
"""智能路由：根据任务复杂度选择模型"""
# 简单启发式判断
simple_keywords = ["点击", "输入", "打开", "关闭", "滚动"]
complex_keywords = ["分析", "对比", "总结", "判断", "决策"]
if any(kw in task for kw in complex_keywords):
complexity = "complex"
elif any(kw in task for kw in simple_keywords):
complexity = "simple"
else:
complexity = "medium"
llm = get_cost_effective_llm(complexity)
from browser_use import Agent
agent = Agent(task=task, llm=llm)
return await agent.run()
总结
Browser Use 代表了网页自动化的新范式——从"编写脚本"到"描述意图"。通过将 LLM 的理解能力与浏览器自动化工具的执行能力结合，开发者可以用更少的代码完成更复杂的任务，同时获得更好的容错性和可维护性。
关键要点回顾：
- 架构理解：Browser Use = LLM（决策） + Playwright（执行） + DOM 快照（感知）
- 工程实践：合理配置浏览器反检测、做好状态管理、优化 DOM 快照
- 成本控制：根据任务复杂度选择模型、优化 token 消耗、设置预算上限
- 混合策略：固定流程用传统选择器，动态部分用 AI，取两者之长
- 企业应用：智能客服、竞品监控、自动化测试等场景均有成熟方案
随着 LLM 能力的持续提升和成本的不断下降，Browser Use 将在更多场景中替代传统自动化方案，成为企业数字化转型的重要工具。
📅 最后更新：2026年5月
📝 本文内容基于公开技术文档与实践经验整理，仅供学习参考
