1. 项目概述：为AI智能体赋予真实的网页访问能力

最近在折腾AI智能体（Agent）项目时，我遇到了一个非常普遍但棘手的问题：如何让我的Agent真正地“看到”并“理解”网页内容？很多教程会告诉你用简单的HTTP请求去抓取HTML，但现实是，现代网页充满了动态加载、复杂交互和反爬机制，一个简单的 requests.get() 往往只能抓回一堆骨架代码或者一个登录页面。这就像给你的Agent戴上了一副度数不对的眼镜，它看到的是一片模糊，根本无法进行有效的分析和决策。

这个项目的核心目标，就是解决这个痛点： 为AI Agent构建一套能够模拟真实用户、全面感知网页内容并结构化提取关键信息的“眼睛”和“手” 。我们不仅要能获取渲染后的完整页面截图（视觉感知），还要能抓取动态加载后的HTML源码（内容解析），更要能从中精准提取出结构化的JSON数据（信息消化）。这套组合拳打下来，你的Agent才能真正具备在复杂网络环境中自主导航、采集和分析信息的能力，无论是用于市场监控、数据聚合、竞品分析还是自动化流程，都将如虎添翼。

2. 核心需求与方案选型背后的考量

2.1 为什么简单的HTTP请求不够用？

在深入技术方案前，我们必须先理解为什么传统方法会失效。这决定了我们技术栈的选型。

1. JavaScript渲染问题 ：如今大量网站（尤其是单页应用SPA，如使用React、Vue.js构建的）依赖客户端JavaScript来渲染内容。一个初始的HTML请求返回的只是一个几乎为空的 <div id="root"></div> ，所有有意义的内容都需要浏览器执行JS后才能生成。简单的HTTP库对此无能为力。
2. 反爬虫机制 ：网站会通过检测请求头（如 User-Agent 是否像浏览器）、Cookie/Session管理、IP频率限制、甚至验证码来阻止自动化脚本。一个“不像人”的请求会被轻易拦截。
3. 交互式内容 ：有些信息需要点击“加载更多”、滚动页面、或与某些页面元素交互后才会出现。这超出了静态抓取的范畴。
4. 内容结构复杂性 ：即使拿到了完整的HTML，如何从中精准定位并提取出我们关心的数据（如商品价格、文章标题、评论列表）？正则表达式脆弱，需要更智能的解析方法。

2.2 技术方案的三层架构

基于上述挑战，一个健壮的方案需要分层解决：

• 浏览器自动化层（模拟真人） ：这是基础。我们需要一个能真正启动浏览器（如Chrome）、执行JavaScript、模拟用户操作（点击、滚动）的工具。 Puppeteer 和 Playwright 是当前的两大主流选择。它们都提供了高级API来控制无头（Headless）或有头浏览器。在这个项目中，我选择 Playwright ，因为它对多浏览器（Chromium, Firefox, WebKit）的原生支持更好，API设计更现代，并且在处理导航、等待元素等场景时更可靠。
• 网页内容捕获层（获取信息） ：在浏览器成功加载页面后，我们需要从中获取两种形式的信息：
  • 视觉信息（截图） ：将整个页面或特定元素保存为图片。这对于需要视觉确认、存档或基于图像分析的后续流程至关重要。Playwright提供了简单的 screenshot() 方法。
  • 文本/结构信息（抓取与提取） ：获取渲染后的完整HTML。然后，我们需要一个强大的解析器来遍历这个DOM树。 BeautifulSoup 是Python生态中的老牌冠军，语法友好。但 Parsel （Scrapy框架使用的选择器库）或 lxml 在纯解析速度和XPath支持上更胜一筹。对于复杂提取，我会结合使用。
• 结构化数据提取层（消化信息） ：这是将杂乱HTML转化为Agent可直接使用的“食物”的关键。我们不仅要提取文本，还要理解数据之间的关系，并输出规整的JSON。这里需要定义清晰的“提取模式”。对于简单固定的页面，可以直接用CSS选择器或XPath定位。对于多变或复杂的页面，可以考虑：
  • 微调LLM提取 ：将HTML片段和自然语言指令（如“提取所有产品的名称、价格和图片链接”）发送给像GPT-4、Claude 3这样的LLM，让其返回JSON。这非常灵活，但成本较高且有延迟。
  • 专用提取库 ：例如 extruct ，它可以提取嵌入在HTML中的结构化数据（如JSON-LD, Microdata, RDFa），很多电商和内容网站都使用这些语义标记，是高质量的数据源。
  • 自定义解析逻辑 ：对于核心稳定数据源，编写稳健的解析函数依然是最可靠、最快的方式。

2.3 工具链最终选型

经过综合评估，我的技术栈如下：

• 浏览器自动化 ：Playwright for Python。它异步特性好，能同时处理多个页面，且自动等待机制减少了编写大量 sleep 语句的需要。
• HTML解析 ：主要使用 lxml 结合 cssselect ，在需要更复杂链式选择时用 parsel 。BeautifulSoup作为备用，因其 find_all 语法对新手更直观。
• 结构化提取 ：采用混合策略。首先尝试用 extruct 提取结构化标记，如果没有或不全，则使用预先配置好的XPath/CSS选择器字典进行提取。对于极其复杂或无规则的情况，预留LLM调用的接口。
• 环境管理 ：使用 conda 或 venv 创建独立Python环境，用 requirements.txt 或 pyproject.toml 严格管理依赖版本，确保复现性。

注意 ：使用Playwright需要安装浏览器二进制文件。初次使用时会自动下载，但请确保网络环境通畅。在国内，可能需要配置镜像源或手动下载。

3. 环境搭建与核心工具详解

3.1 Playwright 安装与初始化

Playwright的安装不是简单的一个 pip install 就完事了。为了稳定和可控，我推荐以下步骤：

# 1. 创建并进入项目目录
mkdir ai-web-agent && cd ai-web-agent

# 2. 创建虚拟环境（以conda为例）
conda create -n web_agent python=3.10 -y
conda activate web_agent

# 3. 使用pip安装playwright
# 指定版本以避免未来API变更导致的问题
pip install playwright==1.40.0

# 4. 安装Playwright所需的浏览器内核
# 这一步会下载Chromium, Firefox和WebKit，体积较大，耐心等待
playwright install
# 如果只需要Chromium，可以运行 `playwright install chromium`

为什么选择Python 3.10？ 这是一个在稳定性和对新库支持上取得平衡的版本。大多数主流库都对其有良好支持，且不像3.11早期版本可能存在一些兼容性问题。

安装完成后，验证一下：

import playwright.sync_api as p

with p.sync_playwright() as playwright:
    browser = playwright.chromium.launch(headless=False) # 先有头模式看看效果
    page = browser.new_page()
    page.goto("https://example.com")
    print(page.title())
    browser.close()

如果能成功打印出“Example Domain”，说明环境基本OK。

3.2 解析库与辅助工具安装

接下来安装HTML解析和数据提取相关的库：

pip install lxml parsel beautifulsoup4 extruct
# 用于处理可能的异步LLM调用和JSON操作
pip install aiohttp httpx
# 用于更优雅地处理数据
pip install pydantic

pydantic 在这里扮演重要角色。我们将用它来定义期望提取的数据结构，它不仅能自动验证数据类型，还能为生成的JSON Schema提供清晰的定义，方便AI Agent理解。

3.3 配置浏览器上下文与常见参数

直接启动浏览器可能会被一些网站识别为自动化脚本。我们需要通过“上下文（Context）”来为浏览器实例添加伪装。

import playwright.sync_api as p

def create_stealthy_browser_context(playwright_instance, headless=True):
    """
    创建一个模拟真实用户的浏览器上下文。
    """
    # 启动浏览器，可以添加一些启动参数
    browser = playwright_instance.chromium.launch(
        headless=headless,
        args=[
            '--disable-blink-features=AutomationControlled', # 禁用自动化控制特征
            '--no-sandbox',
            '--disable-dev-shm-usage'
        ]
    )
    
    # 创建上下文，设置视口大小、User-Agent、语言等
    context = 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='en-US',
        timezone_id='America/New_York',
        # 可以额外加载存储的cookies文件，模拟已登录状态
        # storage_state='path/to/cookies.json'
    )
    
    # 至关重要：在Page加载任何页面之前，执行一个脚本以覆盖navigator.webdriver属性
    context.add_init_script("""
        Object.defineProperty(navigator, 'webdriver', {
            get: () => undefined
        });
        // 覆盖其他可能暴露的自动化属性
        window.chrome = { runtime: {} };
    """)
    
    page = context.new_page()
    return browser, context, page

关键点解释 ：

• --disable-blink-features=AutomationControlled ：禁用Chrome的一些自动化检测特征。
• add_init_script ：这是反检测的核心。网站JavaScript可以通过检查 navigator.webdriver 属性来判断是否被自动化工具控制。我们将其覆盖为 undefined 。
• viewport , user_agent ：模拟一个普通桌面用户的配置。

4. 核心功能实现：截图、抓取与提取

4.1 实现可靠的全页与元素截图

截图功能看似简单，但要想截得完整、清晰，有几个坑需要避开。

def capture_screenshot(page, url, output_path='screenshot.png', full_page=True, selector=None, delay=2000):
    """
    捕获网页截图。
    
    Args:
        page: Playwright page对象。
        url: 要访问的URL。
        output_path: 截图保存路径。
        full_page: 是否截取整个可滚动页面。
        selector: 可选，仅截取该CSS选择器匹配的元素。
        delay: 页面加载后额外的等待时间（毫秒），用于等待动态内容。
    """
    try:
        # 导航到页面，使用`wait_until`确保页面加载状态
        page.goto(url, wait_until='networkidle') # 'networkidle' 表示网络空闲，适合SPA
        # 等待额外时间，确保所有动态内容（如懒加载图片）就位
        page.wait_for_timeout(delay)
        
        screenshot_options = {'path': output_path, 'type': 'png'}
        
        if selector:
            # 等待特定元素出现
            element = page.wait_for_selector(selector, state='visible', timeout=10000)
            screenshot_options['clip'] = element.bounding_box()
            full_page = False # 元素截图时忽略full_page
        
        if full_page:
            screenshot_options['full_page'] = True
        
        page.screenshot(**screenshot_options)
        print(f"Screenshot saved to {output_path}")
        return output_path
        
    except p.TimeoutError:
        print(f"Timeout while loading or waiting for element on {url}")
        # 即使超时，也可以尝试截取当前状态
        page.screenshot(path=f'timeout_{output_path}')
        return None
    except Exception as e:
        print(f"Error capturing screenshot: {e}")
        return None

实操心得 ：

1. wait_until='networkidle' 比默认的 'load' 更适合现代网页，但它也可能因某些长期活跃的连接而一直等待。有时结合 page.wait_for_selector 等待一个关键元素出现会更可靠。
2. delay 参数非常实用。对于依赖setTimeout或异步接口渲染的内容，硬编码等待几秒往往是必要的。更好的做法是使用 page.wait_for_function 检查某个JS变量或DOM状态，但 delay 在快速原型阶段够用。
3. 截取元素时，一定要用 wait_for_selector 确保元素可见，否则 bounding_box() 可能返回 None 。

4.2 抓取渲染后的HTML与动态内容处理

获取“所见即所得”的HTML是后续所有处理的基础。

def get_rendered_html(page, url, wait_for_selector=None, scroll_to_bottom=False):
    """
    获取JavaScript完全执行后的页面HTML。
    
    Args:
        page: Playwright page对象。
        url: 要访问的URL。
        wait_for_selector: 可选，等待该选择器出现后再获取HTML。
        scroll_to_bottom: 是否模拟滚动到底部以触发无限滚动加载。
    """
    page.goto(url, wait_until='domcontentloaded') # 先等待DOM结构加载
    # 也可以使用 'networkidle'，但视情况而定
    
    if wait_for_selector:
        page.wait_for_selector(wait_for_selector, state='visible', timeout=15000)
    
    # 处理滚动加载
    if scroll_to_bottom:
        last_height = page.evaluate('document.body.scrollHeight')
        while True:
            page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
            page.wait_for_timeout(2000) # 等待新内容加载
            new_height = page.evaluate('document.body.scrollHeight')
            if new_height == last_height:
                break
            last_height = new_height
        # 滚动回顶部，方便后续可能基于视口的操作
        page.evaluate('window.scrollTo(0, 0)')
        page.wait_for_timeout(500)
    
    # 获取最终HTML
    html_content = page.content()
    return html_content

关键技巧 ：

• scroll_to_bottom 循环是处理“加载更多”式无限滚动页面的经典模式。但需要设置超时或最大滚动次数限制，防止死循环。
• 获取HTML后，建议立即保存到本地文件，便于调试和避免重复请求： with open('page_dump.html', 'w', encoding='utf-8') as f: f.write(html_content)

4.3 从HTML到结构化JSON：多层提取策略

这是最核心的部分。我们将实现一个分层的提取器，优先使用高质量的结构化数据，再回退到基于规则的解析。

from lxml import html
import extruct
import json
from pydantic import BaseModel, ValidationError
from typing import List, Optional
import re

# 首先，用Pydantic定义我们希望提取的数据模型
class Product(BaseModel):
    name: str
    price: Optional[str] # 价格可能是字符串如"$19.99"
    currency: Optional[str] = "USD"
    image_url: Optional[str]
    description: Optional[str]
    url: Optional[str]

class Article(BaseModel):
    headline: str
    author: Optional[str]
    publish_date: Optional[str]
    body_text: str
    keywords: List[str] = []

class WebExtractor:
    def __init__(self, html_string, source_url=None):
        self.html = html_string
        self.source_url = source_url
        self.tree = html.fromstring(html_string)
        # 预先提取所有结构化数据
        self.structured_data = extruct.extract(html_string, 
                                               syntaxes=['json-ld', 'microdata', 'opengraph'], 
                                               uniform=True)
    
    def try_extract_structured(self, data_type='product'):
        """
        尝试从嵌入的结构化数据中提取信息。
        返回匹配的字典列表，或空列表。
        """
        results = []
        # 处理json-ld，它通常是列表形式
        for item in self.structured_data.get('json-ld', []):
            if data_type == 'product' and item.get('@type') in ['Product', 'IndividualProduct']:
                product_data = {
                    'name': item.get('name'),
                    'price': item.get('offers', {}).get('price') if item.get('offers') else item.get('price'),
                    'currency': item.get('offers', {}).get('priceCurrency') if item.get('offers') else 'USD',
                    'image_url': item.get('image'),
                    'description': item.get('description'),
                    'url': self.source_url
                }
                # 清理空值
                product_data = {k: v for k, v in product_data.items() if v is not None}
                if product_data.get('name'): # 至少要有名字
                    results.append(product_data)
            # 可以添加对Article, Event等其他类型的支持
        return results
    
    def extract_by_rules(self, rule_set):
        """
        基于预定义的CSS选择器/XPath规则进行提取。
        rule_set 是一个字典，例如：
        {
            'title': 'h1.product-title::text',
            'price': '.price::attr(data-amount)',
            'images': '.product-gallery img::attr(src)'
        }
        返回一个字典。
        """
        extracted = {}
        for key, selector in rule_set.items():
            try:
                if '::attr(' in selector:
                    # 处理属性提取，如 `div::attr(data-id)`
                    sel, attr = selector.split('::attr(')
                    attr = attr.rstrip(')')
                    elements = self.tree.cssselect(sel)
                    if elements:
                        extracted[key] = [elem.get(attr) for elem in elements if elem.get(attr)]
                    else:
                        extracted[key] = None
                elif '::text' in selector:
                    # 处理文本提取
                    sel = selector.replace('::text', '')
                    elements = self.tree.cssselect(sel)
                    if elements:
                        texts = [elem.text_content().strip() for elem in elements]
                        extracted[key] = texts if len(texts) > 1 else (texts[0] if texts else None)
                    else:
                        extracted[key] = None
                else:
                    # 默认返回元素本身（outer HTML或进一步处理）
                    elements = self.tree.cssselect(selector)
                    extracted[key] = elements if elements else None
            except Exception as e:
                print(f"Error extracting with selector '{selector}': {e}")
                extracted[key] = None
        return extracted
    
    def extract_articles_generic(self):
        """
        一个通用的文章内容提取示例，基于常见的HTML语义标签。
        这是一个启发式方法，并不完美。
        """
        # 尝试找<article>标签
        article_elements = self.tree.cssselect('article')
        if not article_elements:
            # 回退到包含大量文本的main content区域
            article_elements = self.tree.cssselect('main, [role="main"], .post-content, .article-body')
        
        articles = []
        for elem in article_elements[:5]: # 限制前几个
            headline = elem.cssselect('h1, h2, [itemprop="headline"]')
            headline = headline[0].text_content().strip() if headline else "No Headline"
            
            # 获取所有段落文本
            paragraphs = elem.cssselect('p')
            body = ' '.join([p.text_content().strip() for p in paragraphs if p.text_content().strip()])
            
            # 简单提取作者和日期（正则示例）
            author = None
            date = None
            # 可以在这里添加更复杂的正则匹配逻辑
            
            articles.append({
                'headline': headline,
                'body_text': body[:2000], # 限制长度
                'author': author,
                'publish_date': date
            })
        return articles
    
    def to_json(self, data, indent=2):
        """将提取的数据转换为格式化的JSON字符串。"""
        # 如果数据是Pydantic模型列表，先转换为字典
        if data and isinstance(data[0], BaseModel):
            data = [item.dict() for item in data]
        return json.dumps(data, ensure_ascii=False, indent=indent)

这个 WebExtractor 类的精妙之处在于其分层策略 ：

1. 第一层（最优） ： try_extract_structured 。直接抓取网站开发者主动提供的结构化数据（JSON-LD等）。这些数据质量最高、最规范。很多电商网站（亚马逊、BestBuy）、新闻网站（CNN）和内容平台都大量使用。
2. 第二层（稳健） ： extract_by_rules 。针对特定网站或特定类型的页面，预先配置好选择器规则。这需要一些前期分析工作，但一旦配置好，非常稳定和快速。适合核心数据源。
3. 第三层（通用/后备） ： extract_articles_generic 。这是一个启发式方法，当没有结构化数据也没有预定义规则时，尝试根据常见的HTML语义标签（ <article> , <main> , <p> ）来提取内容。它可能不精确，但能提供一个基线结果。
4. 第四层（终极武器） ：可以集成LLM。当以上所有方法都失效或结果不理想时，将关键的HTML片段和自然语言指令发送给LLM API。由于成本和延迟，这应作为最后手段。

5. 整合与实战：构建一个完整的AI Agent网页访问模块

现在，我们将上述所有功能整合到一个可重用的模块中，并模拟一个AI Agent的调用场景。

# web_agent_toolkit.py
import asyncio
from typing import Dict, Any, List
import json
from pathlib import Path

class AIWebAgent:
    def __init__(self, headless=True, browser_type='chromium'):
        self.playwright = None
        self.browser = None
        self.context = None
        self.headless = headless
        self.browser_type = browser_type
        self.extraction_rules = self._load_rules() # 从文件加载规则
        
    def _load_rules(self):
        """从JSON文件加载针对特定域名或URL模式的提取规则。"""
        rules_file = Path('extraction_rules.json')
        if rules_file.exists():
            with open(rules_file, 'r') as f:
                return json.load(f)
        return {} # 返回空字典，例如：{'example.com/product': {'title': 'h1', 'price': '.price'}}
    
    async def __aenter__(self):
        """异步上下文管理器入口，用于初始化。"""
        from playwright.async_api import async_playwright
        self.playwright = await async_playwright().start()
        browser_launcher = getattr(self.playwright, self.browser_type)
        self.browser = await browser_launcher.launch(headless=self.headless)
        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...',
        )
        await self.context.add_init_script("""
            Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
        """)
        return self
    
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        """异步上下文管理器出口，用于清理。"""
        await self.browser.close()
        await self.playwright.stop()
    
    async def fetch_and_extract(self, url: str, 
                                 actions: List[str] = ['screenshot', 'html', 'extract'],
                                 output_dir: str = './output') -> Dict[str, Any]:
        """
        AI Agent调用的主要方法。
        
        Args:
            url: 目标网页URL。
            actions: 要执行的操作列表，可选 'screenshot', 'html', 'extract'。
            output_dir: 输出文件目录。
        
        Returns:
            包含所有结果（截图路径、HTML、提取的JSON等）的字典。
        """
        Path(output_dir).mkdir(parents=True, exist_ok=True)
        results = {'url': url, 'success': False}
        
        page = await self.context.new_page()
        try:
            # 1. 导航并等待
            await page.goto(url, wait_until='networkidle', timeout=30000)
            domain = url.split('/')[2]
            
            # 2. 执行请求的操作
            if 'screenshot' in actions:
                screenshot_path = Path(output_dir) / f"screenshot_{domain}_{int(time.time())}.png"
                await page.screenshot(path=str(screenshot_path), full_page=True)
                results['screenshot'] = str(screenshot_path)
            
            if 'html' in actions or 'extract' in actions:
                html_content = await page.content()
                results['html_saved'] = False
                if 'html' in actions:
                    html_path = Path(output_dir) / f"page_{domain}_{int(time.time())}.html"
                    html_path.write_text(html_content, encoding='utf-8')
                    results['html_path'] = str(html_path)
                    results['html_saved'] = True
                
                if 'extract' in actions:
                    extractor = WebExtractor(html_content, source_url=url)
                    
                    # 策略1: 尝试结构化数据
                    structured_products = extractor.try_extract_structured('product')
                    if structured_products:
                        results['extraction_method'] = 'structured_data'
                        results['data'] = structured_products
                    else:
                        # 策略2: 查找匹配的预定义规则
                        matched_rule = None
                        for pattern, rule in self.extraction_rules.items():
                            if pattern in url: # 简单匹配，可用正则增强
                                matched_rule = rule
                                break
                        
                        if matched_rule:
                            extracted = extractor.extract_by_rules(matched_rule)
                            results['extraction_method'] = 'predefined_rules'
                            results['data'] = extracted
                        else:
                            # 策略3: 通用提取（如文章）
                            articles = extractor.extract_articles_generic()
                            if articles:
                                results['extraction_method'] = 'generic_heuristic'
                                results['data'] = articles
                            else:
                                results['extraction_method'] = 'none'
                                results['data'] = None
                    
                    # 将提取的数据也保存为JSON文件
                    if results.get('data'):
                        json_path = Path(output_dir) / f"data_{domain}_{int(time.time())}.json"
                        json_path.write_text(extractor.to_json(results['data']), encoding='utf-8')
                        results['json_path'] = str(json_path)
            
            results['success'] = True
            results['title'] = await page.title()
            
        except Exception as e:
            results['error'] = str(e)
            print(f"Error processing {url}: {e}")
        finally:
            await page.close()
        
        return results

# 使用示例 (异步)
async def main():
    async with AIWebAgent(headless=True) as agent:
        # 模拟AI Agent决策后调用
        target_url = "https://news.ycombinator.com"
        result = await agent.fetch_and_extract(
            target_url,
            actions=['screenshot', 'extract'],
            output_dir='./hn_output'
        )
        
        if result['success']:
            print(f"Title: {result['title']}")
            if result.get('data'):
                print(f"Extracted data sample: {json.dumps(result['data'][:1], indent=2)}") # 打印第一条
            if result.get('screenshot'):
                print(f"Screenshot saved at: {result['screenshot']}")
        else:
            print(f"Failed: {result.get('error')}")

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

这个 AIWebAgent 类封装了完整流程 ：

1. 异步设计 ：使用 async/await ，允许Agent同时处理多个网页任务，提高效率。
2. 上下文管理器 ：通过 __aenter__ 和 __aexit__ 确保浏览器资源的正确初始化和清理，避免资源泄漏。
3. 可配置操作 ：Agent可以通过 actions 参数指定需要执行哪些操作（截图、保存HTML、提取数据），按需所取。
4. 分层提取自动化 ：在 fetch_and_extract 方法中，自动按照“结构化数据 -> 预定义规则 -> 通用启发式”的优先级进行数据提取，并将结果统一返回。
5. 结果持久化 ：所有输出（截图、HTML、JSON）都自动保存到指定目录，并附带时间戳，方便追溯和管理。

6. 高级技巧、性能优化与错误处理

6.1 处理登录、Cookie与会话保持

很多有价值的信息在登录墙后面。我们需要让Agent能够管理会话。

async def login_and_save_context(agent, login_url, login_actions, context_save_path='./auth_context.json'):
    """
    执行登录操作，并保存认证后的浏览器上下文状态。
    login_actions: 一个异步函数，接收page参数，并在其上执行登录步骤（如填充表单、点击）。
    """
    page = await agent.context.new_page()
    try:
        await page.goto(login_url)
        # 执行自定义登录逻辑
        await login_actions(page)
        # 等待登录成功，例如导航到特定页面或出现特定元素
        await page.wait_for_selector('.user-avatar', timeout=10000) # 示例
        # 将登录状态（cookies, local storage）保存到文件
        await agent.context.storage_state(path=context_save_path)
        print(f"Login successful, context saved to {context_save_path}")
    except Exception as e:
        print(f"Login failed: {e}")
    finally:
        await page.close()

# 后续使用保存的上下文创建Agent
async def create_agent_with_saved_context(headless=True, context_path='./auth_context.json'):
    from playwright.async_api import async_playwright
    playwright = await async_playwright().start()
    browser = await playwright.chromium.launch(headless=headless)
    # 加载之前保存的上下文状态
    context = await browser.new_context(storage_state=context_path)
    # ... 添加其他初始化脚本 ...
    return browser, context

6.2 性能优化：并发控制与资源管理

同时打开太多页面会耗尽内存。我们需要一个池化机制。

import asyncio
from asyncio import Semaphore

class BoundedBrowserPool:
    def __init__(self, max_concurrent_pages=5):
        self.semaphore = Semaphore(max_concurrent_pages)
        self.browser = None
        self.context = None
    
    async def process_url(self, url):
        """在信号量控制下处理单个URL。"""
        async with self.semaphore:
            # 每个任务使用独立的Page，共享Browser Context
            page = await self.context.new_page()
            try:
                # ... 使用page执行任务 ...
                result = await some_fetch_logic(page, url)
                return result
            finally:
                await page.close()

async def batch_fetch(urls, max_concurrent=5):
    """批量获取多个URL。"""
    pool = BoundedBrowserPool(max_concurrent_pages=max_concurrent)
    async with async_playwright() as p:
        pool.browser = await p.chromium.launch(headless=True)
        pool.context = await pool.browser.new_context()
        
        tasks = [pool.process_url(url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        await pool.browser.close()
    
    # 处理结果和异常
    processed_results = []
    for r in results:
        if isinstance(r, Exception):
            processed_results.append({'error': str(r)})
        else:
            processed_results.append(r)
    return processed_results

6.3 常见反爬策略与应对方案

1. Cloudflare等5秒盾 ：检测到异常流量会弹出验证码或等待页面。
   • 应对 ：最有效的方法是使用更难以检测的浏览器指纹，或者使用专业的反反爬服务（如 puppeteer-extra-plugin-stealth 的Playwright版本）。简单情况下，可以尝试增加初始导航的延迟，或使用 wait_for_selector 等待页面真正加载的内容，而不是网络空闲。
2. 请求频率限制 ：过快请求会导致IP被暂时封禁。
   • 应对 ：在请求间添加随机延迟（ await asyncio.sleep(random.uniform(1, 5)) ），使用代理IP池轮换。 playwright 本身支持通过 context 设置代理。
3. Canvas指纹识别 ：通过Canvas API生成图像来识别浏览器。
   • 应对 ： add_init_script 中可以注入代码来覆盖Canvas API，使其返回一个一致但随机的指纹，但这属于较深的水下攻防。
4. WebDriver检测 ：我们已经通过 add_init_script 覆盖了 navigator.webdriver 。

基本原则 ：对于重要的商业目标，尊重网站的 robots.txt ，并考虑使用官方API。自动化访问应遵循道德和法律界限，避免对目标网站造成过大负担。

6.4 错误处理与重试机制

网络请求天生不稳定，必须要有健壮的错误处理。

import asyncio
import random
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

class RobustWebAgent(AIWebAgent):
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type((TimeoutError, asyncio.TimeoutError))
    )
    async def robust_fetch(self, url, timeout=30000):
        """带有重试机制的获取方法。"""
        page = await self.context.new_page()
        try:
            # 设置页面超时
            page.set_default_timeout(timeout)
            response = await page.goto(url, wait_until='domcontentloaded')
            if response and response.status >= 400:
                raise Exception(f"HTTP {response.status} for {url}")
            # ... 后续操作 ...
            return await page.content()
        except Exception as e:
            print(f"Attempt failed for {url}: {e}")
            raise # 让tenacity捕获并决定是否重试
        finally:
            await page.close()

使用 tenacity 库可以优雅地实现指数退避重试，对于处理暂时的网络波动或服务器过载非常有效。

7. 将能力封装给AI Agent：定义清晰的工具与提示词

最后，我们需要让上层的AI Agent（比如使用LangChain、AutoGPT或ChatGPT Function Calling）能够方便地调用这些能力。关键在于提供清晰的工具描述（Schema）。

例如，为基于OpenAI Assistant或类似框架的Agent定义工具：

{
  "type": "function",
  "function": {
    "name": "fetch_and_analyze_webpage",
    "description": "访问一个网页，捕获其视觉截图，并提取其中的结构化信息（如产品列表、文章内容、价格等）。适用于信息搜集、竞品分析、内容监控等场景。",
    "parameters": {
      "type": "object",
      "properties": {
        "url": {
          "type": "string",
          "description": "需要访问和分析的网页完整URL。"
        },
        "extraction_focus": {
          "type": "string",
          "enum": ["auto", "products", "articles", "contact_info", "prices"],
          "description": "指定提取信息的重点。'auto'将自动探测。"
        },
        "save_output": {
          "type": "boolean",
          "description": "是否将截图和提取的数据保存到本地文件。"
        }
      },
      "required": ["url"]
    }
  }
}

然后，在Agent的系统中提示词里加入这样的说明：

“你拥有浏览网页的能力。当你需要获取最新网页内容、查看页面视觉效果或从特定网页提取结构化数据时，可以使用 fetch_and_analyze_webpage 工具。只需提供URL和你关心的信息类型即可。”

当Agent决定调用此工具时，后端就执行我们构建的 AIWebAgent.fetch_and_extract 方法，并将结果（如图片路径、提取的JSON）返回给Agent，Agent再基于这些真实、结构化的信息进行后续的推理、总结或决策。

至此，你的AI Agent就真正拥有了穿透现代网络复杂性的“眼睛”和“手”，能够可靠地获取并理解互联网上的动态信息，为其自主任务执行打下了坚实的基础。这套方案不是银弹，需要根据具体目标网站进行微调和规则维护，但它提供了一个强大、可扩展的起点，足以应对大多数常见的网页访问与信息提取挑战。