AI浏览器自动化与Web Agent完全教程
本教程全面讲解AI浏览器自动化与Web Agent的核心技术，通过丰富的代码示例和实战案例，帮助开发者掌握AI驱动的浏览器自动化技术。
目录
- 概述与背景
- 核心技术栈
- Playwright自动化基础
- Browser Use框架详解
- Computer Use多模态操作
- 网页元素识别与交互
- 表单自动填写
- 数据采集与爬取
- 视觉定位与点击
- 多步骤任务规划
- 错误恢复机制
- 实战：自动化数据采集Agent
- 最佳实践与注意事项
- 总结
概述与背景
传统的浏览器自动化依赖于Selenium、Puppeteer等工具，通过CSS选择器、XPath等方式定位网页元素。这种方式虽然有效，但存在明显的局限性：当网页结构发生变化时，脚本往往会失效；面对复杂的动态页面，编写和维护选择器的成本很高。
AI浏览器自动化的出现改变了这一局面。通过将大语言模型（LLM）与浏览器控制相结合，我们可以用自然语言描述任务，让AI Agent自主理解页面内容、规划操作步骤、执行交互动作。这种范式转变使得自动化脚本更加鲁棒、灵活，能够适应网页的变化。
AI浏览器自动化的核心优势：
- 自然语言驱动：用自然语言描述任务，无需编写复杂的选择器
- 视觉理解能力：AI可以"看懂"页面内容，理解语义关系
- 自适应能力：页面结构变化时，AI能自动调整操作策略
- 复杂推理能力：处理多步骤、有依赖关系的复杂任务
目前主流的AI浏览器自动化技术包括：
- Browser Use：基于Playwright的AI Agent框架
- Computer Use：Anthropic的多模态计算机操作能力
- WebVoyager：多模态Web导航Agent
- SeeAct：基于视觉语言模型的网页操作
核心技术栈
在深入学习之前，我们先了解一下AI浏览器自动化涉及的核心技术栈：
┌─────────────────────────────────────────────┐
│ 用户自然语言指令 │
├─────────────────────────────────────────────┤
│ AI Agent (LLM核心) │
│ ┌──────────┬──────────┬──────────────┐ │
│ │ 任务规划 │ 元素理解 │ 动作决策 │ │
│ └──────────┴──────────┴──────────────┘ │
├─────────────────────────────────────────────┤
│ 浏览器控制层 │
│ ┌──────────┬──────────┬──────────────┐ │
│ │ Playwright│ Selenium │ Chrome DevTools│ │
│ └──────────┴──────────┴──────────────┘ │
├─────────────────────────────────────────────┤
│ 网页渲染层 │
│ ┌──────────┬──────────┬──────────────┐ │
│ │ Chrome │ Firefox │ WebKit │ │
│ └──────────┴──────────┴──────────────┘ │
└─────────────────────────────────────────────┘
关键依赖库：
# 安装核心依赖
pip install playwright browser-use langchain-openai langchain-anthropic
# 安装Playwright浏览器
playwright install chromium
# 可选：安装其他依赖
pip install beautifulsoup4 lxml pillow
Playwright自动化基础
Playwright是微软开发的现代浏览器自动化库，支持Chromium、Firefox和WebKit三大浏览器引擎。它是Browser Use等AI Agent框架的底层基础。
基础操作
import asyncio
from playwright.async_api import async_playwright
async def basic_demo():
"""Playwright基础操作演示"""
async with async_playwright() as p:
# 启动浏览器
browser = await p.chromium.launch(headless=False)
context = await browser.new_context(
viewport={"width": 1280, "height": 720},
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
)
page = await context.new_page()
# 导航到目标页面
await page.goto("https://example.com", wait_until="networkidle")
# 等待元素出现
await page.wait_for_selector("h1", timeout=10000)
# 获取页面标题
title = await page.title()
print(f"页面标题: {title}")
# 获取元素文本
heading = await page.inner_text("h1")
print(f"标题文本: {heading}")
# 截图
await page.screenshot(path="screenshot.png", full_page=True)
# 关闭浏览器
await browser.close()
asyncio.run(basic_demo())
元素定位与交互
async def element_interaction():
"""元素定位与交互演示"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False)
page = await browser.new_page()
await page.goto("https://example.com/form")
# 多种定位方式
# 1. CSS选择器
button = page.locator("button.submit-btn")
# 2. 文本内容定位
link = page.get_by_text("了解更多")
# 3. 角色定位（推荐的语义化定位方式）
login_button = page.get_by_role("button", name="登录")
# 4. 标签定位
email_input = page.get_by_label("邮箱地址")
# 5. 占位符定位
search_input = page.get_by_placeholder("搜索...")
# 6. 测试ID定位
submit = page.get_by_test_id("submit-button")
# 执行交互
await email_input.fill("user@example.com")
await search_input.fill("Playwright自动化")
await login_button.click()
# 等待导航完成
await page.wait_for_load_state("networkidle")
await browser.close()
高级功能：网络拦截与请求监听
async def network_interception():
"""网络请求拦截与监听"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False)
page = await browser.new_page()
# 监听所有网络请求
requests_log = []
def on_request(request):
requests_log.append({
"url": request.url,
"method": request.method,
"resource_type": request.resource_type
})
page.on("request", on_request)
# 拦截特定请求
async def handle_route(route):
if route.request.resource_type == "image":
await route.abort() # 阻止图片加载，加速爬取
else:
await route.continue_()
await page.route("**/*", handle_route)
await page.goto("https://news.ycombinator.com")
await page.wait_for_load_state("networkidle")
print(f"捕获到 {len(requests_log)} 个请求")
for req in requests_log[:5]:
print(f" {req['method']} {req['url'][:80]}")
await browser.close()
Browser Use框架详解
Browser Use是一个将AI Agent与浏览器自动化相结合的开源框架。它封装了Playwright的复杂性，让开发者可以用自然语言驱动浏览器操作。
架构设计
Browser Use的核心架构包含以下组件：
- Agent：核心调度器，负责任务理解、步骤规划和执行控制
- Browser：浏览器实例管理，封装Playwright操作
- Controller：动作注册与执行，定义Agent可执行的操作
- DOM Service：网页内容提取与元素映射
- Message Manager：管理与LLM的对话上下文
基础使用
import asyncio
from langchain_openai import ChatOpenAI
from browser_use import Agent, Browser, BrowserConfig
async def basic_browser_use():
"""Browser Use基础用法"""
# 配置浏览器
browser_config = BrowserConfig(
headless=False,
disable_security=True, # 禁用同源策略（仅用于开发）
extra_chromium_args=[
"--window-size=1280,720",
"--disable-blink-features=AutomationControlled"
]
)
# 初始化LLM
llm = ChatOpenAI(
model="gpt-4o",
temperature=0
)
# 创建Agent
agent = Agent(
task="打开Google，搜索 'Python browser automation'，获取前3个搜索结果的标题和链接",
llm=llm,
browser=Browser(config=browser_config),
max_actions_per_step=5 # 每步最多执行5个动作
)
# 执行任务
result = await agent.run(max_steps=20)
print("任务执行结果:")
print(result)
asyncio.run(basic_browser_use())
自定义工具与动作
Browser Use允许开发者注册自定义工具，扩展Agent的能力：
from browser_use import Agent, Controller
from langchain_openai import ChatOpenAI
from pydantic import BaseModel
# 定义输出格式
class ProductInfo(BaseModel):
name: str
price: str
rating: str
url: str
class ProductList(BaseModel):
products: list[ProductInfo]
# 创建控制器并注册自定义动作
controller = Controller()
@controller.action("Save product data to file", param_model=ProductList)
async def save_products(data: ProductList):
"""将产品数据保存到文件"""
with open("products.json", "w", encoding="utf-8") as f:
import json
json.dump(
[p.model_dump() for p in data.products],
f,
ensure_ascii=False,
indent=2
)
return f"已保存 {len(data.products)} 个产品信息"
@controller.action("Take a screenshot with custom name")
async def take_screenshot(browser, name: str):
"""自定义截图功能"""
page = await browser.get_current_page()
await page.screenshot(path=f"screenshots/{name}.png")
return f"截图已保存: screenshots/{name}.png"
@controller.action("Extract all links from current page")
async def extract_links(browser):
"""提取当前页面所有链接"""
page = await browser.get_current_page()
links = await page.evaluate("""
() => Array.from(document.querySelectorAll('a[href]')).map(a => ({
text: a.innerText.trim(),
href: a.href
})).filter(l => l.text && l.href.startsWith('http'))
""")
return {"links": links, "count": len(links)}
async def custom_tools_demo():
"""自定义工具演示"""
llm = ChatOpenAI(model="gpt-4o", temperature=0)
agent = Agent(
task="访问电商网站，搜索'机械键盘'，提取前5个产品的名称、价格、评分和链接，然后保存到文件",
llm=llm,
controller=controller,
max_actions_per_step=3
)
result = await agent.run(max_steps=30)
print(result)
多标签页管理
async def multi_tab_demo():
"""多标签页操作演示"""
from browser_use import Browser
browser = Browser()
async with browser:
# 创建多个标签页
page1 = await browser.new_page()
await page1.goto("https://example.com")
page2 = await browser.new_page()
await page2.goto("https://example.org")
# 获取所有标签页
pages = await browser.get_pages()
print(f"当前有 {len(pages)} 个标签页")
# 切换标签页
await browser.switch_to_tab(0) # 切换到第一个标签页
# 在不同标签页中执行操作
title1 = await page1.title()
title2 = await page2.title()
print(f"标签页1: {title1}")
print(f"标签页2: {title2}")
# 关闭指定标签页
await browser.close_tab(1)
Computer Use多模态操作
Anthropic的Computer Use能力让Claude能够通过截图理解计算机屏幕，并执行鼠标点击、键盘输入等操作。这是一种完全基于视觉的交互方式。
工作原理
Computer Use的工作流程：
- 截取屏幕截图发送给Claude
- Claude分析截图，确定需要执行的操作
- 返回具体的操作指令（点击坐标、按键等）
- 执行操作，再次截图，循环直到任务完成
实现Computer Use Agent
import anthropic
import pyautogui
import base64
import time
from PIL import Image
class ComputerUseAgent:
"""基于Anthropic Computer Use的Agent"""
def __init__(self, api_key: str):
self.client = anthropic.Anthropic(api_key=api_key)
self.screen_width, self.screen_height = pyautogui.size()
def take_screenshot(self) -> str:
"""截取屏幕并返回base64编码"""
screenshot = pyautogui.screenshot()
# 缩放到合理大小
screenshot = screenshot.resize((1280, 800))
import io
buffer = io.BytesIO()
screenshot.save(buffer, format="PNG")
return base64.standard_b64encode(buffer.getvalue()).decode("utf-8")
def execute_action(self, action: dict):
"""执行Claude返回的操作指令"""
action_type = action.get("type")
if action_type == "mouse_move":
x, y = action["coordinate"]
pyautogui.moveTo(x, y, duration=0.3)
elif action_type == "left_click":
x, y = action["coordinate"]
pyautogui.click(x, y)
elif action_type == "double_click":
x, y = action["coordinate"]
pyautogui.doubleClick(x, y)
elif action_type == "right_click":
x, y = action["coordinate"]
pyautogui.rightClick(x, y)
elif action_type == "type":
text = action["text"]
pyautogui.typewrite(text, interval=0.05)
elif action_type == "key":
key = action["key"]
pyautogui.press(key)
elif action_type == "scroll":
x, y = action["coordinate"]
direction = action["direction"]
amount = action.get("amount", 3)
pyautogui.moveTo(x, y)
if direction == "down":
pyautogui.scroll(-amount)
else:
pyautogui.scroll(amount)
elif action_type == "screenshot":
pass # 截图操作在主循环中处理
async def run_task(self, task: str, max_turns: int = 20):
"""执行自然语言任务"""
messages = []
for turn in range(max_turns):
# 截取当前屏幕
screenshot_b64 = self.take_screenshot()
# 构建消息
if turn == 0:
messages.append({
"role": "user",
"content": [
{
"type": "text",
"text": f"请完成以下任务: {task}"
},
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": screenshot_b64
}
}
]
})
else:
messages.append({
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": "image/png",
"data": screenshot_b64
}
},
{
"type": "text",
"text": "这是当前屏幕截图，请继续执行任务。"
}
]
})
# 调用Claude
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=1024,
tools=[{
"type": "computer_20250124",
"name": "computer",
"display_width_px": 1280,
"display_height_px": 800,
"display_number": 1
}],
messages=messages
)
# 处理响应
assistant_content = response.content
messages.append({"role": "assistant", "content": assistant_content})
# 检查是否完成
if response.stop_reason == "end_turn":
text_blocks = [b for b in assistant_content if b.type == "text"]
if text_blocks:
print(f"任务完成: {text_blocks[0].text}")
return
# 执行工具调用
tool_results = []
for block in assistant_content:
if block.type == "tool_use":
print(f"执行操作: {block.name} - {block.input}")
self.execute_action(block.input)
time.sleep(0.5) # 等待操作生效
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": [{"type": "text", "text": "操作已执行"}]
})
if tool_results:
messages.append({"role": "user", "content": tool_results})
print("达到最大轮次限制")
混合方案：Playwright + Computer Use
import asyncio
from playwright.async_api import async_playwright
class HybridBrowserAgent:
"""结合Playwright精确控制和Computer Use视觉理解的混合Agent"""
def __init__(self, llm):
self.llm = llm
async def analyze_page_visually(self, page):
"""通过截图让AI分析页面"""
screenshot = await page.screenshot(type="png")
import base64
screenshot_b64 = base64.b64encode(screenshot).decode()
# 使用多模态LLM分析截图
response = await self.llm.ainvoke([
{
"role": "user",
"content": [
{"type": "text", "text": "分析这个网页截图，识别主要的交互元素和它们的大致位置"},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{screenshot_b64}"}}
]
}
])
return response.content
async def precise_interact(self, page, selector, action, value=None):
"""使用Playwright进行精确的元素交互"""
element = page.locator(selector)
if action == "click":
await element.click()
elif action == "fill":
await element.fill(value)
elif action == "select":
await element.select_option(value)
elif action == "hover":
await element.hover()
# 操作后等待页面稳定
await page.wait_for_load_state("networkidle")
async def smart_interact(self, page, description):
"""当无法通过选择器定位时，使用视觉定位"""
# 获取页面可交互元素的坐标
elements = await page.evaluate("""
() => {
const interactive = document.querySelectorAll(
'a, button, input, select, textarea, [role="button"], [onclick]'
);
return Array.from(interactive).map(el => {
const rect = el.getBoundingClientRect();
return {
tag: el.tagName,
text: el.innerText?.substring(0, 50),
type: el.type || '',
x: rect.x + rect.width / 2,
y: rect.y + rect.height / 2,
width: rect.width,
height: rect.height
};
}).filter(el => el.width > 0 && el.height > 0);
}
""")
# 让AI选择最匹配的元素
response = await self.llm.ainvoke([
{"role": "system", "content": "你是网页元素识别专家。根据用户描述和元素列表，返回最匹配元素的索引。只返回数字。"},
{"role": "user", "content": f"描述: {description}\n元素列表: {elements}"}
])
try:
index = int(response.content.strip())
target = elements[index]
await page.mouse.click(target["x"], target["y"])
return True
except (ValueError, IndexError):
return False
网页元素识别与交互
在AI浏览器自动化中，准确识别和理解网页元素是关键。现代方案结合了DOM解析、视觉理解和语义分析。
DOM内容提取
async def extract_page_content(page) -> dict:
"""提取页面的结构化内容"""
# 提取文本内容
text_content = await page.evaluate("""
() => {
// 获取所有可见文本
const walker = document.createTreeWalker(
document.body,
NodeFilter.SHOW_TEXT,
{
acceptNode: (node) => {
const parent = node.parentElement;
if (!parent) return NodeFilter.FILTER_REJECT;
const style = window.getComputedStyle(parent);
if (style.display === 'none' || style.visibility === 'hidden')
return NodeFilter.FILTER_REJECT;
return NodeFilter.FILTER_ACCEPT;
}
}
);
const texts = [];
while (walker.nextNode()) {
const text = walker.currentNode.textContent.trim();
if (text) texts.push(text);
}
return texts;
}
""")
# 提取交互元素
interactive_elements = await page.evaluate("""
() => {
const selectors = [
'a[href]', 'button', 'input', 'select', 'textarea',
'[role="button"]', '[role="link"]', '[role="tab"]',
'[onclick]', '[tabindex]'
];
const elements = [];
const seen = new Set();
for (const sel of selectors) {
document.querySelectorAll(sel).forEach(el => {
if (seen.has(el)) return;
seen.add(el);
const rect = el.getBoundingClientRect();
if (rect.width === 0 || rect.height === 0) return;
elements.push({
tag: el.tagName.toLowerCase(),
text: el.innerText?.substring(0, 100)?.trim() || '',
href: el.href || '',
type: el.type || '',
name: el.name || '',
id: el.id || '',
placeholder: el.placeholder || '',
ariaLabel: el.getAttribute('aria-label') || '',
role: el.getAttribute('role') || '',
rect: {
x: Math.round(rect.x),
y: Math.round(rect.y),
width: Math.round(rect.width),
height: Math.round(rect.height)
}
});
});
}
return elements;
}
""")
# 提取页面元数据
metadata = await page.evaluate("""
() => ({
title: document.title,
url: window.location.href,
forms: Array.from(document.forms).map(f => ({
action: f.action,
method: f.method,
fields: Array.from(f.elements).map(e => ({
name: e.name,
type: e.type,
value: e.value
}))
}))
})
""")
return {
"text": text_content,
"elements": interactive_elements,
"metadata": metadata
}
基于AI的元素理解
async def ai_element_understanding(llm, page_content: dict, user_intent: str) -> dict:
"""使用AI理解页面元素与用户意图的关系"""
prompt = f"""你是一个网页分析专家。根据以下页面信息和用户意图，找出最相关的交互元素。
用户意图: {user_intent}
页面交互元素:
{json.dumps(page_content['elements'], ensure_ascii=False, indent=2)}
请返回JSON格式，包含:
- target_element: 目标元素的详细信息
- action: 建议执行的操作 (click/fill/select/hover)
- value: 如果需要填入的值
- confidence: 置信度 (0-1)
- reasoning: 选择理由
只返回JSON，不要其他内容。"""
response = await llm.ainvoke([{"role": "user", "content": prompt}])
return json.loads(response.content)
表单自动填写
表单自动填写是AI浏览器自动化的重要应用场景。AI可以理解表单结构，自动匹配字段并填入正确的内容。
智能表单填写器
import asyncio
from dataclasses import dataclass
from typing import Any
@dataclass
class FormData:
"""表单数据模型"""
fields: dict[str, Any]
metadata: dict[str, Any] = None
class SmartFormFiller:
"""AI驱动的智能表单填写器"""
def __init__(self, llm, browser):
self.llm = llm
self.browser = browser
async def analyze_form(self, page) -> list[dict]:
"""分析页面中的表单结构"""
form_info = await page.evaluate("""
() => {
const forms = [];
document.querySelectorAll('form').forEach(form => {
const fields = [];
form.querySelectorAll('input, select, textarea').forEach(el => {
const rect = el.getBoundingClientRect();
const label = el.labels?.[0]?.innerText?.trim() ||
el.getAttribute('aria-label') ||
el.getAttribute('placeholder') ||
el.name || el.id;
fields.push({
name: el.name,
id: el.id,
type: el.type || el.tagName.toLowerCase(),
label: label,
required: el.required,
value: el.value,
options: el.tagName === 'SELECT'
? Array.from(el.options).map(o => ({
value: o.value,
text: o.text
}))
: [],
rect: {
x: rect.x + rect.width / 2,
y: rect.y + rect.height / 2
}
});
});
forms.append({
action: form.action,
method: form.method,
fields: fields
});
});
return forms;
}
""")
return form_info
async def match_fields(self, form_fields: list, user_data: dict) -> dict:
"""使用AI匹配表单字段与用户数据"""
prompt = f"""将用户数据匹配到表单字段。
用户数据:
{json.dumps(user_data, ensure_ascii=False, indent=2)}
表单字段:
{json.dumps(form_fields, ensure_ascii=False, indent=2)}
返回JSON格式的匹配结果，key为字段name或id，value为要填入的值。
对于select字段，返回最匹配的option value。
对于无法匹配的字段，value设为null。
只返回JSON。"""
response = await self.llm.ainvoke([{"role": "user", "content": prompt}])
return json.loads(response.content)
async def fill_form(self, page, form_index: int, field_values: dict):
"""填写表单"""
forms = await self.analyze_form(page)
if form_index >= len(forms):
raise ValueError(f"表单索引 {form_index} 超出范围")
form = forms[form_index]
for field in form["fields"]:
field_name = field["name"] or field["id"]
value = field_values.get(field_name)
if value is None:
continue
# 构建选择器
if field["id"]:
selector = f"#{field['id']}"
elif field["name"]:
selector = f"[name='{field['name']}']"
else:
# 使用坐标点击
await page.mouse.click(field["rect"]["x"], field["rect"]["y"])
continue
element = page.locator(selector)
field_type = field["type"]
if field_type in ("text", "email", "tel", "url", "password", "number"):
await element.fill(str(value))
elif field_type == "select":
await element.select_option(value=str(value))
elif field_type == "checkbox":
if value and not await element.is_checked():
await element.check()
elif not value and await element.is_checked():
await element.uncheck()
elif field_type == "radio":
await element.check()
elif field_type == "textarea":
await element.fill(str(value))
await asyncio.sleep(0.3) # 模拟人工输入间隔
async def fill_form_natural(self, page, description: str, user_data: dict):
"""自然语言驱动的表单填写"""
# 分析表单
forms = await self.analyze_form(page)
if not forms:
print("未找到表单")
return
# AI匹配字段
matched_fields = await self.match_fields(forms[0]["fields"], user_data)
# 过滤null值
valid_fields = {k: v for k, v in matched_fields.items() if v is not None}
print(f"匹配到 {len(valid_fields)} 个字段")
# 填写表单
await self.fill_form(page, 0, valid_fields)
print("表单填写完成")
数据采集与爬取
AI驱动的数据采集相比传统爬虫更加智能和灵活，能够处理复杂的页面结构和反爬机制。
智能数据采集Agent
import asyncio
import json
from datetime import datetime
class AIDataCollector:
"""AI驱动的智能数据采集器"""
def __init__(self, llm, browser):
self.llm = llm
self.browser = browser
self.collected_data = []
async def collect_from_page(self, page, schema: dict) -> list[dict]:
"""从当前页面按schema提取数据"""
# 获取页面HTML结构
html = await page.content()
# 使用AI提取数据
prompt = f"""从以下HTML中提取数据，按照给定的schema格式化。
Schema: {json.dumps(schema, ensure_ascii=False)}
HTML片段（前5000字符）:
{html[:5000]}
返回JSON数组，每个元素对应一条数据记录。
只返回JSON，不要其他内容。"""
response = await self.llm.ainvoke([{"role": "user", "content": prompt}])
try:
data = json.loads(response.content)
return data if isinstance(data, list) else [data]
except json.JSONDecodeError:
print(f"AI返回的JSON解析失败: {response.content[:200]}")
return []
async def auto_paginate(self, page, next_button_selector: str = None) -> bool:
"""自动翻页"""
if next_button_selector:
next_btn = page.locator(next_button_selector)
if await next_btn.count() > 0 and await next_btn.is_enabled():
await next_btn.click()
await page.wait_for_load_state("networkidle")
return True
# AI自动寻找翻页按钮
page_info = await page.evaluate("""
() => {
const candidates = document.querySelectorAll(
'a.next, button.next, [aria-label="Next"], [aria-label="下一页"], a[rel="next"]'
);
return Array.from(candidates).map(el => ({
tag: el.tagName,
text: el.innerText,
href: el.href,
disabled: el.disabled
}));
}
""")
if page_info:
next_link = page.locator("a.next, a[rel='next'], [aria-label='Next'], [aria-label='下一页']").first
if await next_link.count() > 0:
await next_link.click()
await page.wait_for_load_state("networkidle")
return True
return False
async def collect_with_pagination(self, page, schema: dict,
max_pages: int = 10,
next_selector: str = None) -> list[dict]:
"""带翻页的数据采集"""
all_data = []
for page_num in range(max_pages):
print(f"正在采集第 {page_num + 1} 页...")
# 采集当前页数据
page_data = await self.collect_from_page(page, schema)
all_data.extend(page_data)
print(f" 采集到 {len(page_data)} 条数据")
# 尝试翻页
has_next = await self.auto_paginate(page, next_selector)
if not has_next:
print("没有更多页面")
break
await asyncio.sleep(1) # 礼貌性延迟
self.collected_data.extend(all_data)
return all_data
async def save_data(self, filename: str, format: str = "json"):
"""保存采集到的数据"""
if format == "json":
with open(filename, "w", encoding="utf-8") as f:
json.dump(self.collected_data, f, ensure_ascii=False, indent=2)
elif format == "csv":
import csv
if self.collected_data:
keys = self.collected_data[0].keys()
with open(filename, "w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=keys)
writer.writeheader()
writer.writerows(self.collected_data)
print(f"数据已保存到 {filename}，共 {len(self.collected_data)} 条")
反检测策略
async def anti_detection_setup(browser_context):
"""设置反检测策略"""
# 注入反检测脚本
await browser_context.add_init_script("""
// 隐藏webdriver标识
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});
// 模拟真实浏览器的plugins
Object.defineProperty(navigator, 'plugins', {
get: () => [
{ name: 'Chrome PDF Plugin' },
{ name: 'Chrome PDF Viewer' },
{ name: 'Native Client' }
]
});
// 修改navigator.languages
Object.defineProperty(navigator, 'languages', {
get: () => ['zh-CN', 'zh', 'en-US', 'en']
});
// 隐藏自动化特征
window.chrome = {
runtime: {},
loadTimes: function() {},
csi: function() {},
app: {}
};
// 修改权限查询
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
parameters.name === 'notifications'
? Promise.resolve({ state: Notification.permission })
: originalQuery(parameters)
);
""")
# 设置真实的请求头
await browser_context.set_extra_http_headers({
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Accept-Encoding": "gzip, deflate, br",
"Connection": "keep-alive",
"Upgrade-Insecure-Requests": "1"
})
# 随机化视口大小
import random
width = random.randint(1200, 1400)
height = random.randint(700, 900)
await browser_context.set_viewport_size({"width": width, "height": height})
return browser_context
视觉定位与点击
当DOM结构不可靠或元素无法通过选择器定位时，视觉定位成为关键能力。
截图标注与坐标定位
from PIL import Image, ImageDraw, ImageFont
import base64
class VisualLocator:
"""视觉定位器"""
def __init__(self, llm):
self.llm = llm
async def locate_element_by_screenshot(self, page, description: str) -> tuple[int, int]:
"""通过截图和AI定位元素坐标"""
# 截取页面
screenshot = await page.screenshot(type="png")
img = Image.open(io.BytesIO(screenshot))
width, height = img.size
# 转为base64
screenshot_b64 = base64.b64encode(screenshot).decode()
# 使用多模态AI定位
response = await self.llm.ainvoke([
{
"role": "user",
"content": [
{
"type": "text",
"text": f"""请在这个截图中找到"{description}"的位置。
返回JSON格式: {{"x": 像素坐标x, "y": 像素坐标y, "confidence": 置信度0-1}}
图片尺寸: {width}x{height}
只返回JSON。"""
},
{
"type": "image_url",
"image_url": {"url": f"data:image/png;base64,{screenshot_b64}"}
}
]
}
])
result = json.loads(response.content)
return (result["x"], result["y"]), result.get("confidence", 0.5)
def annotate_screenshot(self, image: Image.Image,
points: list[dict]) -> Image.Image:
"""在截图上标注定位点"""
draw = ImageDraw.Draw(image)
for point in points:
x, y = point["x"], point["y"]
color = point.get("color", "red")
label = point.get("label", "")
# 画十字标记
size = 15
draw.line([(x - size, y), (x + size, y)], fill=color, width=2)
draw.line([(x, y - size), (x, y + size)], fill=color, width=2)
# 画圆圈
draw.ellipse(
[(x - size, y - size), (x + size, y + size)],
outline=color, width=2
)
# 添加标签
if label:
draw.text((x + size + 5, y - 10), label, fill=color)
return image
async def click_by_visual(self, page, description: str, llm):
"""通过视觉定位并点击"""
coords, confidence = await self.locate_element_by_screenshot(page, description)
if confidence < 0.3:
print(f"警告: 视觉定位置信度较低 ({confidence})")
return False
x, y = coords
print(f"点击坐标: ({x}, {y}), 置信度: {confidence}")
# 移动鼠标到目标位置
await page.mouse.move(x, y)
await asyncio.sleep(0.2)
# 点击
await page.mouse.click(x, y)
return True
多步骤任务规划
复杂的自动化任务需要合理的规划和分解。AI Agent需要理解任务的依赖关系，制定执行计划。
任务规划器
from dataclasses import dataclass, field
from enum import Enum
class TaskStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class TaskStep:
"""任务步骤"""
id: int
description: str
action: str
params: dict
status: TaskStatus = TaskStatus.PENDING
result: Any = None
error: str = None
dependencies: list[int] = field(default_factory=list)
retry_count: int = 0
max_retries: int = 3
class TaskPlanner:
"""AI任务规划器"""
def __init__(self, llm):
self.llm = llm
self.steps: list[TaskStep] = []
self.context: dict = {}
async def create_plan(self, task_description: str, page_info: dict = None) -> list[TaskStep]:
"""根据任务描述创建执行计划"""
context_str = ""
if page_info:
context_str = f"\n当前页面信息:\n{json.dumps(page_info, ensure_ascii=False, indent=2)}"
prompt = f"""请将以下任务分解为具体的执行步骤。
任务描述: {task_description}
{context_str}
返回JSON数组，每个步骤包含:
- description: 步骤描述
- action: 动作类型 (navigate/click/fill/extract/scroll/wait/conditional)
- params: 动作参数
- dependencies: 依赖的步骤索引数组
步骤应该具体、可执行、有明确的输入输出。"""
response = await self.llm.ainvoke([{"role": "user", "content": prompt}])
plan_data = json.loads(response.content)
self.steps = []
for i, step in enumerate(plan_data):
self.steps.append(TaskStep(
id=i,
description=step["description"],
action=step["action"],
params=step["params"],
dependencies=step.get("dependencies", [])
))
return self.steps
def get_executable_steps(self) -> list[TaskStep]:
"""获取当前可执行的步骤（所有依赖已完成）"""
executable = []
for step in self.steps:
if step.status != TaskStatus.PENDING:
continue
deps_met = all(
self.steps[dep].status == TaskStatus.COMPLETED
for dep in step.dependencies
)
if deps_met:
executable.append(step)
return executable
async def execute_plan(self, browser_agent) -> dict:
"""执行完整计划"""
max_iterations = len(self.steps) * 2 # 防止无限循环
iteration = 0
while iteration < max_iterations:
executable = self.get_executable_steps()
if not executable:
break
for step in executable:
step.status = TaskStatus.RUNNING
print(f"执行步骤 {step.id}: {step.description}")
try:
result = await self.execute_step(step, browser_agent)
step.result = result
step.status = TaskStatus.COMPLETED
print(f" ✓ 完成")
except Exception as e:
step.error = str(e)
step.retry_count += 1
if step.retry_count < step.max_retries:
step.status = TaskStatus.PENDING
print(f" ⟳ 重试 ({step.retry_count}/{step.max_retries}): {e}")
else:
step.status = TaskStatus.FAILED
print(f" ✗ 失败: {e}")
iteration += 1
# 汇总结果
return {
"completed": [s for s in self.steps if s.status == TaskStatus.COMPLETED],
"failed": [s for s in self.steps if s.status == TaskStatus.FAILED],
"results": [s.result for s in self.steps if s.result]
}
async def execute_step(self, step: TaskStep, browser_agent) -> Any:
"""执行单个步骤"""
page = browser_agent.page
action = step.action
params = step.params
if action == "navigate":
await page.goto(params["url"])
await page.wait_for_load_state("networkidle")
elif action == "click":
if "selector" in params:
await page.click(params["selector"])
elif "text" in params:
await page.get_by_text(params["text"]).click()
await page.wait_for_load_state("networkidle")
elif action == "fill":
await page.fill(params["selector"], params["value"])
elif action == "extract":
data = await page.evaluate(params["js_code"])
return data
elif action == "scroll":
await page.evaluate(f"window.scrollBy(0, {params.get('amount', 500)})")
elif action == "wait":
if "selector" in params:
await page.wait_for_selector(params["selector"], timeout=params.get("timeout", 10000))
else:
await asyncio.sleep(params.get("seconds", 2))
elif action == "conditional":
# 条件判断，让AI决定下一步
condition_result = await page.evaluate(params["condition_js"])
if condition_result:
return params.get("true_result")
else:
return params.get("false_result")
return None
错误恢复机制
健壮的错误恢复机制是生产级自动化系统的关键。
重试与恢复策略
import asyncio
import functools
import logging
from typing import Callable
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def with_retry(max_retries: int = 3,
delay: float = 1.0,
backoff: float = 2.0,
exceptions: tuple = (Exception,)):
"""重试装饰器，支持指数退避"""
def decorator(func: Callable):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
last_exception = None
current_delay = delay
for attempt in range(max_retries + 1):
try:
return await func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt < max_retries:
logger.warning(
f"{func.__name__} 第{attempt + 1}次尝试失败: {e}, "
f"{current_delay:.1f}秒后重试..."
)
await asyncio.sleep(current_delay)
current_delay *= backoff
else:
logger.error(f"{func.__name__} 所有重试均失败: {e}")
raise last_exception
return wrapper
return decorator
class ErrorRecoveryManager:
"""错误恢复管理器"""
def __init__(self, llm):
self.llm = llm
self.error_history = []
self.recovery_strategies = {
"element_not_found": self._recover_element_not_found,
"page_load_timeout": self._recover_page_load_timeout,
"navigation_error": self._recover_navigation_error,
"captcha_detected": self._recover_captcha,
"rate_limited": self._recover_rate_limit,
}
async def handle_error(self, error: Exception, page, context: dict) -> bool:
"""处理错误并尝试恢复"""
error_type = self._classify_error(error)
self.error_history.append({
"type": error_type,
"message": str(error),
"timestamp": datetime.now().isoformat(),
"context": context
})
logger.error(f"遇到错误 [{error_type}]: {error}")
# 尝试预定义的恢复策略
if error_type in self.recovery_strategies:
recovery_fn = self.recovery_strategies[error_type]
try:
recovered = await recovery_fn(page, error, context)
if recovered:
logger.info(f"错误恢复成功: {error_type}")
return True
except Exception as recovery_error:
logger.error(f"恢复策略执行失败: {recovery_error}")
# 使用AI分析错误并尝试恢复
return await self._ai_recovery(page, error, context)
def _classify_error(self, error: Exception) -> str:
"""错误分类"""
error_str = str(error).lower()
error_type = type(error).__name__
if "timeout" in error_str or "TimeoutError" in error_type:
return "page_load_timeout"
if "element" in error_str and ("not found" in error_str or "not visible" in error_str):
return "element_not_found"
if "navigation" in error_str:
return "navigation_error"
if "captcha" in error_str or "recaptcha" in error_str:
return "captcha_detected"
if "429" in error_str or "rate" in error_str:
return "rate_limited"
return "unknown"
async def _recover_element_not_found(self, page, error, context):
"""元素未找到的恢复策略"""
selector = context.get("selector", "")
description = context.get("description", "")
# 策略1: 等待更长时间
try:
await page.wait_for_selector(selector, timeout=15000)
return True
except:
pass
# 策略2: 滚动页面后重试
await page.evaluate("window.scrollTo(0, document.body.scrollHeight / 2)")
await asyncio.sleep(1)
try:
await page.wait_for_selector(selector, timeout=5000)
return True
except:
pass
# 策略3: 使用AI寻找替代元素
if description:
page_content = await extract_page_content(page)
match = await ai_element_understanding(
self.llm, page_content, f"找到 {description}"
)
if match.get("confidence", 0) > 0.5:
element = match["target_element"]
await page.click(
element["rect"]["x"] + element["rect"]["width"] / 2,
element["rect"]["y"] + element["rect"]["height"] / 2
)
return True
return False
async def _recover_page_load_timeout(self, page, error, context):
"""页面加载超时的恢复策略"""
# 尝试停止加载并继续
try:
await page.evaluate("window.stop()")
await asyncio.sleep(2)
# 检查页面是否可用
title = await page.title()
if title:
return True
except:
pass
# 重新导航
url = context.get("url")
if url:
try:
await page.goto(url, timeout=30000, wait_until="domcontentloaded")
return True
except:
pass
return False
async def _recover_navigation_error(self, page, error, context):
"""导航错误的恢复策略"""
url = context.get("url")
if url:
# 尝试不同的等待策略
try:
await page.goto(url, wait_until="commit", timeout=30000)
await asyncio.sleep(3)
return True
except:
pass
return False
async def _recover_captcha(self, page, error, context):
"""验证码恢复策略"""
logger.warning("检测到验证码，暂停30秒后重试...")
await asyncio.sleep(30)
# 可以在这里集成验证码识别服务
return False
async def _recover_rate_limit(self, page, error, context):
"""频率限制恢复策略"""
logger.warning("触发频率限制，等待60秒...")
await asyncio.sleep(60)
return True
async def _ai_recovery(self, page, error, context) -> bool:
"""使用AI进行智能错误恢复"""
try:
screenshot = await page.screenshot(type="png")
screenshot_b64 = base64.b64encode(screenshot).decode()
page_url = page.url
page_title = await page.title()
prompt = f"""浏览器自动化遇到错误，请分析并提供恢复建议。
错误信息: {error}
错误类型: {type(error).__name__}
当前URL: {page_url}
页面标题: {page_title}
任务上下文: {json.dumps(context, ensure_ascii=False)}
请分析截图和错误信息，返回JSON格式的恢复方案:
{{
"action": "恢复动作 (retry/navigate/scroll/click/wait/abort)",
"params": {{}},
"reasoning": "分析原因"
}}"""
response = await self.llm.ainvoke([
{
"role": "user",
"content": [
{"type": "text", "text": prompt},
{"type": "image_url", "image_url": {"url": f"data:image/png;base64,{screenshot_b64}"}}
]
}
])
recovery = json.loads(response.content)
action = recovery["action"]
params = recovery.get("params", {})
logger.info(f"AI恢复方案: {action} - {recovery.get('reasoning', '')}")
if action == "retry":
return True
elif action == "navigate" and "url" in params:
await page.goto(params["url"])
return True
elif action == "scroll":
amount = params.get("amount", 500)
await page.evaluate(f"window.scrollBy(0, {amount})")
return True
elif action == "wait":
await asyncio.sleep(params.get("seconds", 5))
return True
except Exception as ai_error:
logger.error(f"AI恢复分析失败: {ai_error}")
return False
实战：自动化数据采集Agent
综合以上技术，我们构建一个完整的自动化数据采集Agent。
import asyncio
import json
import logging
from datetime import datetime
from pathlib import Path
from langchain_openai import ChatOpenAI
from playwright.async_api import async_playwright
from browser_use import Agent, Browser, BrowserConfig
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class DataCollectionAgent:
"""完整的自动化数据采集Agent"""
def __init__(self, config: dict):
self.config = config
self.llm = ChatOpenAI(
model=config.get("model", "gpt-4o"),
temperature=0
)
self.error_manager = ErrorRecoveryManager(self.llm)
self.collector = AIDataCollector(self.llm, None) # browser稍后设置
self.results = []
async def initialize(self):
"""初始化浏览器和Agent"""
self.playwright = await async_playwright().start()
self.browser = await self.playwright.chromium.launch(
headless=self.config.get("headless", True),
args=[
"--disable-blink-features=AutomationControlled",
"--no-sandbox"
]
)
self.context = await self.browser.new_context(
viewport={"width": 1280, "height": 720},
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
)
# 应用反检测
await anti_detection_setup(self.context)
self.page = await self.context.new_page()
self.collector.browser = type('Browser', (), {'get_current_page': lambda s: self.page})()
async def run_task(self, task: str, data_schema: dict = None,
max_pages: int = 5, output_file: str = None):
"""执行数据采集任务"""
logger.info(f"开始执行任务: {task}")
# 创建执行计划
planner = TaskPlanner(self.llm)
# 先导航到目标页面以获取页面信息
target_url = self.config.get("target_url")
if target_url:
await self.page.goto(target_url, wait_until="domcontentloaded")
await asyncio.sleep(2)
page_info = await extract_page_content(self.page)
plan = await planner.create_plan(task, {
"url": self.page.url,
"title": await self.page.title(),
"elements_count": len(page_info.get("elements", []))
})
logger.info(f"执行计划包含 {len(plan)} 个步骤")
for step in plan:
logger.info(f" Step {step.id}: {step.description}")
# 执行计划
execution_result = await planner.execute_plan(self)
# 如果有数据schema，执行数据采集
if data_schema:
logger.info("开始数据采集...")
data = await self.collector.collect_with_pagination(
self.page, data_schema, max_pages=max_pages
)
self.results.extend(data)
# 保存结果
if output_file:
await self.save_results(output_file)
return {
"plan_result": execution_result,
"collected_data": self.results,
"total_items": len(self.results)
}
async def save_results(self, output_file: str):
"""保存采集结果"""
output_path = Path(output_file)
output_path.parent.mkdir(parents=True, exist_ok=True)
if output_file.endswith(".json"):
with open(output_file, "w", encoding="utf-8") as f:
json.dump({
"timestamp": datetime.now().isoformat(),
"config": self.config,
"total_items": len(self.results),
"data": self.results
}, f, ensure_ascii=False, indent=2)
elif output_file.endswith(".csv"):
await self.collector.save_data(output_file, format="csv")
logger.info(f"结果已保存到 {output_file}")
async def cleanup(self):
"""清理资源"""
await self.page.close()
await self.context.close()
await self.browser.close()
await self.playwright.stop()
# 主函数
async def main():
"""主入口"""
config = {
"model": "gpt-4o",
"headless": True,
"target_url": "https://news.ycombinator.com"
}
agent = DataCollectionAgent(config)
await agent.initialize()
try:
result = await agent.run_task(
task="采集Hacker News首页的所有新闻标题、链接和分数",
data_schema={
"title": "新闻标题",
"url": "新闻链接",
"score": "分数",
"author": "作者"
},
max_pages=3,
output_file="output/hn_data.json"
)
print(f"\n采集完成！共采集 {result['total_items']} 条数据")
for item in result["collected_data"][:5]:
print(f" - {item.get('title', 'N/A')}")
finally:
await agent.cleanup()
if __name__ == "__main__":
asyncio.run(main())
最佳实践与注意事项
1. 性能优化
# 使用headless模式提高速度
browser = await p.chromium.launch(headless=True)
# 禁用不必要的资源加载
async def block_resources(route):
blocked_types = {"image", "stylesheet", "font", "media"}
if route.request.resource_type in blocked_types:
await route.abort()
else:
await route.continue_()
await page.route("**/*", block_resources)
# 设置合理的超时时间
page.set_default_timeout(15000)
page.set_default_navigation_timeout(30000)
2. 成本控制
- 选择合适的模型：简单任务使用小模型，复杂推理使用大模型
- 减少截图频率：能用DOM交互的场景不必截图
- 缓存AI决策：相同页面结构的决策可以缓存复用
- 限制LLM调用次数：设置
max_actions_per_step和max_steps
3. 遵守规范
# 遵守robots.txt
import urllib.robotparser
def can_crawl(url: str, user_agent: str = "*") -> bool:
"""检查是否允许爬取"""
from urllib.parse import urlparse
parsed = urlparse(url)
robots_url = f"{parsed.scheme}://{parsed.netloc}/robots.txt"
rp = urllib.robotparser.RobotFileParser()
rp.set_url(robots_url)
try:
rp.read()
return rp.can_fetch(user_agent, url)
except:
return True # 无法读取时默认允许
# 礼貌性延迟
import random
async def polite_delay(min_seconds=1, max_seconds=3):
"""随机延迟，避免对服务器造成压力"""
delay = random.uniform(min_seconds, max_seconds)
await asyncio.sleep(delay)
4. 日志与监控
import logging
from datetime import datetime
class AgentLogger:
"""Agent操作日志记录器"""
def __init__(self, log_file: str = "agent.log"):
self.logger = logging.getLogger("AgentLogger")
self.logger.setLevel(logging.INFO)
handler = logging.FileHandler(log_file, encoding="utf-8")
formatter = logging.Formatter(
'%(asctime)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
self.logger.addHandler(handler)
self.action_count = 0
self.start_time = datetime.now()
def log_action(self, action: str, result: str, details: dict = None):
self.action_count += 1
self.logger.info(
f"[Action #{self.action_count}] {action} -> {result} | "
f"Details: {json.dumps(details or {}, ensure_ascii=False)}"
)
def log_error(self, error: str, context: dict = None):
self.logger.error(
f"[Error] {error} | Context: {json.dumps(context or {}, ensure_ascii=False)}"
)
def get_summary(self) -> dict:
elapsed = (datetime.now() - self.start_time).total_seconds()
return {
"total_actions": self.action_count,
"elapsed_seconds": elapsed,
"actions_per_minute": self.action_count / (elapsed / 60) if elapsed > 0 else 0
}
总结
AI浏览器自动化与Web Agent代表了自动化技术的重要发展方向。通过将大语言模型的推理能力与浏览器控制相结合，我们能够构建更加智能、灵活、鲁棒的自动化系统。
本教程涵盖的核心要点：
- Playwright基础：现代浏览器自动化的基石，提供强大的页面控制能力
- Browser Use框架：将AI Agent与浏览器自动化的优雅结合
- Computer Use：纯视觉驱动的计算机操作方案
- 元素识别与交互：DOM解析与AI视觉理解的结合
- 表单自动填写：智能表单识别与数据填充
- 数据采集：AI驱动的智能数据采集与翻页策略
- 视觉定位：基于截图的元素定位与交互
- 任务规划：复杂任务的分解与依赖管理
- 错误恢复：多层次的错误处理与智能恢复
随着多模态AI模型的持续进步，浏览器自动化将变得更加智能和通用。建议开发者从简单的任务开始，逐步构建更复杂的自动化工作流，在实践中掌握AI浏览器自动化的核心技术。
本教程持续更新中，欢迎反馈和建议。
