最近在尝试爬取一些电商平台数据时发现传统的Web逆向如JS逆向、反调试绕过越来越难搞。复杂的混淆、动态加密、环境检测让逆向分析耗时耗力甚至可能刚分析完对方就更新了防护。有没有一种更“智能”、更通用的方法答案是肯定的那就是Agent智能体。本文要探讨的不是传统意义上用于自动化测试的Selenium/Puppeteer而是结合了大型语言模型LLM推理能力的AI Agent。它能够理解网页结构、模拟人类操作逻辑、处理动态内容甚至应对一些简单的反爬策略实现“通杀”级别的数据采集。下面我将从概念到实战手把手带你搭建一个面向电商平台的AI Agent并附上核心源码和完整的搭建流程。无论你是数据采集开发者还是对AI应用感兴趣的同学都能从中获得一套可直接复用的解决方案。1. 背景与核心概念为什么说Agent是Web逆向的新思路在深入之前我们首先要厘清几个关键概念并理解传统Web逆向的痛点与Agent方案的突破点。1.1 传统Web逆向的困境传统的Web数据采集尤其是针对反爬严密的网站通常遵循“分析-破解-模拟”的路径静态分析查看网页源码、Network请求寻找数据接口。JS逆向如果接口参数被JavaScript动态加密如sign、token则需要使用开发者工具调试定位加密函数并用Python如PyExecJS、Node.js子进程还原算法。环境模拟应对WebDriver检测、window.navigator属性检测、Canvas指纹等。痛点在于高成本每个网站、甚至同一网站的不同接口其加密逻辑都可能不同需要重复投入大量分析时间。高维护成本网站前端稍作更新加密逻辑或环境检测就可能变化导致爬虫失效需要重新分析。技术门槛高需要深厚的JavaScript、浏览器原理、密码学知识。脆弱性对抗性升级道高一尺魔高一丈。1.2 什么是AI AgentAI Agent智能体在此语境下指的是一个能够感知环境、自主决策、执行动作以实现目标的软件实体。结合LLM如GPT-4、Claude、国产大模型它具备了强大的自然语言理解、逻辑推理和代码生成能力。一个用于Web操作的AI Agent通常包含以下核心组件规划器Planner将用户的高层目标如“获取商品X的价格和评论”分解为一系列可执行的步骤如“打开网站-搜索商品-进入详情页-提取价格和评论”。记忆Memory存储历史操作、网页状态、提取到的数据用于上下文理解和避免重复操作。工具ToolsAgent可以调用的外部能力如打开浏览器、点击元素、输入文本、滚动页面、提取文本、执行JavaScript等。执行器Actuator实际驱动工具执行的模块如通过Playwright或Selenium控制浏览器。1.3 Agent如何“通杀”Web逆向Agent的思路是**“降维打击”**。它不直接与复杂的加密逻辑硬碰硬而是模拟真人操作通过浏览器自动化工具以真实用户的身份访问网站加载完整的JavaScript环境让网站自己完成所有加密和渲染。这直接绕过了静态分析和JS还原的步骤。智能解析与决策LLM可以理解网页的语义结构。你无需编写复杂的XPath或CSS Selector来定位元素只需用自然语言告诉Agent“找到那个显示价格的红色数字”或“点击‘下一页’按钮”。Agent能理解指令并找到对应元素。处理动态与异常遇到验证码、弹窗、页面布局变化你可以为Agent设计应对策略例如“如果看到滑块验证调用打码平台API”或者“如果找不到商品列表尝试滚动页面”。LLM可以根据当前页面状态做出判断。简而言之Agent将爬虫开发从“密码学工程”部分转变为了“流程设计与自然语言指令工程”大大降低了技术壁垒并提升了方案的通用性和鲁棒性。2. 环境准备与版本说明我们将构建一个基于Python、Playwright浏览器自动化和OpenAI APILLM能力的电商数据采集Agent。你也可以替换为其他LLM如通过Ollama部署的本地模型。核心环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文示例在 macOS/Linux 下完成。Python: 3.8 或更高版本。推荐使用 3.10。包管理pip或conda。主要依赖库及版本建议使用虚拟环境# 创建并激活虚拟环境 (可选) python -m venv web_agent_env source web_agent_env/bin/activate # Linux/macOS # web_agent_env\Scripts\activate # Windows # 安装核心依赖 pip install playwright1.40.0 # 浏览器自动化 pip install openai1.12.0 # OpenAI API 客户端 pip install python-dotenv1.0.0 # 环境变量管理 pip install beautifulsoup44.12.3 # 备用HTML解析 pip install lxml5.2.1 # 解析器 # 安装Playwright浏览器内核 playwright install chromium重要配置OpenAI API Key你需要从 OpenAI平台 获取一个有效的API Key。我们将它存储在环境变量中以保证安全。项目结构建议如下组织你的代码。web_scraping_agent/ ├── .env # 存储API KEY等敏感信息 ├── config.py # 配置文件 ├── agent_core.py # Agent核心逻辑 ├── tools.py # 自定义工具集 ├── planner.py # 规划器 ├── memory.py # 记忆模块 ├── main.py # 主程序入口 ├── requirements.txt # 依赖列表 └── logs/ # 日志目录3. 核心组件拆解构建Agent的四大支柱我们的Agent架构将围绕规划、工具、记忆、执行来构建。下面逐一拆解其原理和实现。3.1 工具ToolsAgent的手和脚工具是Agent与浏览器环境交互的接口。我们基于Playwright封装一系列原子操作。tools.py核心代码import asyncio from playwright.async_api import async_playwright, Page, BrowserContext from typing import Any, Dict, Optional import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class BrowserTools: 浏览器操作工具集 def __init__(self): self.playwright None self.browser None self.context: Optional[BrowserContext] None self.page: Optional[Page] None async def start_browser(self, headless: bool False): 启动浏览器实例。headlessFalse便于调试。 self.playwright await async_playwright().start() # 使用Chromium更稳定且指纹相对统一 self.browser await self.playwright.chromium.launch( headlessheadless, args[--disable-blink-featuresAutomationControlled] # 禁用自动化控制特征 ) # 创建上下文可以统一设置User-Agent、视口等 self.context await self.browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 ) self.page await self.context.new_page() logger.info(浏览器启动成功) async def goto(self, url: str): 导航到指定URL if not self.page: raise RuntimeError(浏览器页面未初始化请先调用start_browser) await self.page.goto(url, wait_untilnetworkidle) # 等待网络空闲 logger.info(f已导航至: {url}) return await self.get_page_summary() async def click(self, selector: str, description: str ): 点击元素 if not self.page: raise RuntimeError(浏览器页面未初始化) try: await self.page.wait_for_selector(selector, statevisible, timeout10000) await self.page.click(selector) logger.info(f点击元素: {description or selector}) await asyncio.sleep(1) # 等待页面反应 return await self.get_page_summary() except Exception as e: logger.error(f点击元素失败 {selector}: {e}) return f操作失败: 无法点击元素 {selector}错误: {e} async def type_text(self, selector: str, text: str): 在输入框输入文本 if not self.page: raise RuntimeError(浏览器页面未初始化) try: await self.page.wait_for_selector(selector, statevisible, timeout10000) await self.page.fill(selector, text) logger.info(f在 {selector} 中输入文本: {text}) await asyncio.sleep(0.5) return await self.get_page_summary() except Exception as e: logger.error(f输入文本失败 {selector}: {e}) return f操作失败: 无法在 {selector} 中输入文本错误: {e} async def scroll(self, direction: str down, pixels: int 500): 滚动页面 if not self.page: raise RuntimeError(浏览器页面未初始化) try: if direction down: await self.page.mouse.wheel(0, pixels) elif direction up: await self.page.mouse.wheel(0, -pixels) logger.info(f向{direction}滚动{pixels}像素) await asyncio.sleep(1) # 等待内容加载 return await self.get_page_summary() except Exception as e: logger.error(f滚动失败: {e}) return f滚动操作失败: {e} async def extract_content(self, selector: str body): 提取指定选择器的文本内容 if not self.page: raise RuntimeError(浏览器页面未初始化) try: element await self.page.wait_for_selector(selector, timeout10000) text await element.inner_text() logger.info(f从 {selector} 提取内容长度: {len(text)}) # 返回精简后的内容避免上下文过长 return text[:2000] (... if len(text) 2000 else ) except Exception as e: logger.error(f提取内容失败 {selector}: {e}) return f提取失败: 无法从 {selector} 获取内容错误: {e} async def get_page_summary(self) - str: 获取当前页面的关键信息摘要用于反馈给LLM if not self.page: return 页面未就绪 try: # 获取页面标题和URL title await self.page.title() url self.page.url # 获取可见区域的主要文本简化版 main_content await self.page.locator(body).inner_text() summary f页面标题: {title}\n当前URL: {url}\n页面内容预览:\n{main_content[:500]}... return summary except Exception as e: return f获取页面摘要失败: {e} async def close(self): 关闭浏览器资源 if self.browser: await self.browser.close() if self.playwright: await self.playwright.stop() logger.info(浏览器资源已释放)关键点解析反检测args[--disable-blink-featuresAutomationControlled]是Playwright提供的禁用自动化控制特征的参数能有效降低被简单脚本检测的风险。等待策略wait_untilnetworkidle和wait_for_selector确保了页面或元素加载完成后再操作提高了稳定性。错误处理每个工具函数都包含try-except并将错误信息以自然语言形式返回便于LLM理解并调整策略。页面摘要get_page_summary函数至关重要。它将当前页面的状态标题、URL、核心内容总结成一段文本作为LLM决策的“观察”。3.2 规划器Planner与LLM集成Agent的大脑规划器的核心是调用LLM根据用户目标和当前页面状态决定下一步该执行哪个工具函数。我们使用OpenAI的Chat Completion API。planner.py核心代码import openai import os from dotenv import load_dotenv import json import logging from typing import List, Dict, Any load_dotenv() # 加载.env文件中的环境变量 openai.api_key os.getenv(OPENAI_API_KEY) logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class AgentPlanner: 基于LLM的规划器决定下一步行动 def __init__(self, model: str gpt-3.5-turbo): self.model model self.tools_description self._get_tools_description() def _get_tools_description(self) - str: 定义可供Agent使用的工具列表及其描述 tools [ { name: goto, description: 导航到一个新的URL。输入应为完整的网址。, parameters: {url: string} }, { name: click, description: 点击页面上的一个元素。你需要提供CSS选择器或对元素的描述。, parameters: {selector: string, description: string (optional)} }, { name: type_text, description: 在输入框内输入文本。你需要提供输入框的选择器和要输入的文本。, parameters: {selector: string, text: string} }, { name: scroll, description: 向上或向下滚动页面。用于加载更多内容。, parameters: {direction: string (down/up), pixels: integer} }, { name: extract_content, description: 从页面中提取特定元素的文本内容。默认提取整个body。, parameters: {selector: string (optional)} } ] # 将工具描述格式化为LLM容易理解的文本 desc_lines [你可以使用以下工具] for tool in tools: desc_lines.append(f- {tool[name]}: {tool[description]} 参数: {tool[parameters]}) return \n.join(desc_lines) def plan_next_action(self, goal: str, current_page_summary: str, action_history: List[str]) - Dict[str, Any]: 根据目标、当前页面状态和历史规划下一个动作。 返回一个包含工具名和参数的字典。 # 构建给LLM的提示词Prompt system_prompt f你是一个专业的网页操作智能体Web Agent。你的任务是理解用户目标并通过操作浏览器来达成目标。 {self.tools_description} 行动准则 1. 仔细分析当前页面摘要理解你在网站中的位置。 2. 回顾历史操作避免重复无效动作。 3. 一次只执行一个最直接、最必要的动作。 4. 如果页面摘要显示目标已达成如已找到所需信息则返回工具名为 extract_content 来提取最终数据。 5. 如果卡住或找不到元素尝试 scroll 或重新评估。 6. 你的输出必须是严格的JSON格式{{tool: 工具名, parameters: {{参数1: 值1, ...}}}} user_prompt f 用户目标{goal} 当前页面状态 {current_page_summary} 已执行的操作历史最近5条 {chr(10).join(action_history[-5:]) if action_history else 无} 请根据以上信息决定下一步应该执行哪个工具操作并给出具体的参数。 只输出JSON不要有其他任何解释。 messages [ {role: system, content: system_prompt}, {role: user, content: user_prompt} ] try: response openai.chat.completions.create( modelself.model, messagesmessages, temperature0.1, # 低随机性保证决策稳定 max_tokens500 ) llm_output response.choices[0].message.content.strip() logger.info(fLLM原始输出: {llm_output}) # 解析LLM返回的JSON action json.loads(llm_output) # 基本验证 if tool not in action or parameters not in action: raise ValueError(LLM返回的JSON格式不正确缺少tool或parameters字段) return action except json.JSONDecodeError as e: logger.error(f解析LLM输出的JSON失败: {e}, 输出内容: {llm_output}) # 降级策略返回一个安全的默认动作比如重新获取页面状态 return {tool: extract_content, parameters: {selector: body}} except Exception as e: logger.error(f调用LLM规划时发生错误: {e}) raise关键点解析提示词工程Prompt Engineering这是Agent智能度的核心。system_prompt定义了Agent的角色、可用工具和行动准则。user_prompt提供了当前任务的具体上下文目标、页面状态、历史。清晰的提示词能极大提升LLM决策的准确性。输出约束强制要求LLM输出严格的JSON格式便于程序化解析。这是实现LLM与代码协作的关键。错误处理与降级当LLM输出不符合JSON格式时我们有降级策略例如默认执行extract_content避免整个流程崩溃。历史记录将最近的操作历史提供给LLM有助于它避免循环操作比如反复点击同一个无效按钮。3.3 记忆Memory与执行循环连接大脑与手脚记忆模块负责存储操作历史和中间数据。执行循环则是驱动整个Agent运行的主逻辑。memory.py核心代码from typing import List, Dict, Any class SimpleMemory: 简单的记忆模块存储会话历史和数据 def __init__(self, max_history_len: int 50): self.action_history: List[str] [] # 记录每一步操作描述 self.extracted_data: List[Dict[str, Any]] [] # 记录提取到的数据 self.max_history_len max_history_len def add_action(self, action: str, parameters: Dict, result: str): 记录一次操作及其结果 entry f动作: {action}({parameters}) - 结果摘要: {result[:100]}... self.action_history.append(entry) # 限制历史长度防止上下文过长 if len(self.action_history) self.max_history_len: self.action_history.pop(0) def add_data(self, data: Dict[str, Any]): 记录提取到的结构化数据 self.extracted_data.append(data) def get_recent_history(self, n: int 5) - List[str]: 获取最近N条操作历史 return self.action_history[-n:] if self.action_history else [] def get_all_data(self) - List[Dict[str, Any]]: 获取所有已提取的数据 return self.extracted_dataagent_core.py- 执行循环核心import asyncio from typing import Dict, Any, Optional from tools import BrowserTools from planner import AgentPlanner from memory import SimpleMemory import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class WebScrapingAgent: Web采集Agent的核心执行器 def __init__(self, planner: AgentPlanner, headless: bool False): self.planner planner self.browser_tools BrowserTools() self.memory SimpleMemory() self.headless headless self.is_running False async def run(self, goal: str, start_url: Optional[str] None): 运行Agent尝试达成目标 self.is_running True try: # 1. 启动浏览器 await self.browser_tools.start_browser(headlessself.headless) logger.info(f开始执行目标: {goal}) # 2. 如果有起始URL先导航过去 current_state 浏览器已启动等待指令。 if start_url: current_state await self.browser_tools.goto(start_url) self.memory.add_action(goto, {url: start_url}, current_state) max_steps 20 # 防止无限循环 step 0 # 3. 主循环规划 - 执行 - 观察 - 记录 while self.is_running and step max_steps: step 1 logger.info(f--- 第 {step} 步 ---) # a. 规划下一步 recent_history self.memory.get_recent_history() next_action self.planner.plan_next_action(goal, current_state, recent_history) tool_name next_action[tool] params next_action[parameters] logger.info(f规划动作: {tool_name} with {params}) # b. 执行动作 tool_func getattr(self.browser_tools, tool_name, None) if not tool_func: logger.error(f未知工具: {tool_name}) current_state f错误: 未知工具 {tool_name} self.memory.add_action(tool_name, params, current_state) break try: # 调用工具函数 result await tool_func(**params) current_state result # 更新当前状态为执行结果 logger.info(f动作执行结果: {result[:200]}...) # c. 记录到记忆 self.memory.add_action(tool_name, params, result) # d. 简单目标检查示例如果结果中包含“价格”等关键词可能意味着成功 # 这里可以设计更复杂的目标达成判定逻辑 if 价格 in result or 商品详情 in result: logger.info(可能已找到目标信息尝试提取最终数据。) # 触发一次最终数据提取 final_data await self.browser_tools.extract_content(body) self.memory.add_data({final_extraction: final_data}) # 可以根据需要决定是否继续或停止 # break except Exception as e: error_msg f执行工具 {tool_name} 时出错: {e} logger.error(error_msg) current_state error_msg self.memory.add_action(tool_name, params, current_state) # 出错后可以尝试让LLM规划恢复策略这里简单跳过或停止 # 例如可以等待一下再继续 await asyncio.sleep(2) # 短暂暂停模拟人类操作间隔也避免请求过快 await asyncio.sleep(1) logger.info(fAgent执行结束共执行 {step} 步。) # 返回最终收集到的所有数据 return self.memory.get_all_data() except Exception as e: logger.error(fAgent运行过程中发生严重错误: {e}) raise finally: # 4. 清理资源 await self.browser_tools.close() self.is_running False关键点解析状态驱动循环Agent的核心是一个循环。每次迭代它都基于当前页面状态current_state和历史recent_history通过规划器planner决定下一个动作然后调用工具tool执行并将结果更新为新的状态如此往复。可配置的停止条件我们设置了最大步数max_steps来防止无限循环。在实际应用中你还可以设计更智能的停止条件例如当LLM规划出stop动作时、当成功提取到目标数据时、或当长时间无进展时。异常处理与鲁棒性在执行工具时进行异常捕获并将错误信息记录到记忆和状态中。这样在下一轮规划时LLM就能知道上一步出错了并可能尝试不同的策略。资源管理在finally块中确保浏览器被正确关闭避免资源泄漏。4. 完整实战案例抓取电商平台商品信息现在我们将上述组件组装起来完成一个具体任务在某个电商平台以示例网站为例上搜索“Python编程书籍”并获取第一页的商品列表信息名称、价格。注意出于法律和道德考虑本示例将使用一个公开的、允许爬取的测试网站或模拟环境如https://books.toscrape.com/进行演示。在实际应用中请务必遵守目标网站的robots.txt协议和服务条款。4.1 项目主程序与配置.env文件OPENAI_API_KEY你的OpenAI_API_Key_放在这里main.py主程序import asyncio import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from agent_core import WebScrapingAgent from planner import AgentPlanner import json async def main(): 主函数定义任务并运行Agent # 1. 初始化规划器使用GPT-3.5-turbo成本较低 planner AgentPlanner(modelgpt-3.5-turbo) # 2. 初始化AgentheadlessFalse便于观察调试 agent WebScrapingAgent(plannerplanner, headlessFalse) # 3. 定义目标任务 goal 请访问网站 https://books.toscrape.com/。 这是一个在线书店。 你的目标是找到所有关于“Python”主题的书籍并列出它们的书名和价格。 如果第一页没有请尝试搜索或翻页。 start_url https://books.toscrape.com/ print(f开始执行目标\n{goal}) print(*50) try: # 4. 运行Agent collected_data await agent.run(goalgoal, start_urlstart_url) # 5. 输出结果 print(\n *50) print(任务执行完成收集到的数据如下) print(*50) for i, data_item in enumerate(collected_data): print(f\n--- 数据块 {i1} ---) # 美化打印JSON print(json.dumps(data_item, indent2, ensure_asciiFalse)) except Exception as e: print(f运行过程中出错: {e}) finally: print(\n程序结束。) if __name__ __main__: asyncio.run(main())4.2 运行与结果分析运行程序在终端中执行python main.py。观察过程由于设置了headlessFalse你会看到一个浏览器窗口自动打开导航到书籍网站。然后你会看到Agent在“思考”调用LLM并自动执行点击、滚动等操作。可能的执行序列步骤1LLM根据目标“找到Python书籍”看到首页是分类可能规划动作{tool: click, parameters: {selector: a[href*python], description: 点击Python分类链接}}。步骤2如果直接点击成功页面跳转到Python书籍列表。LLM看到列表后可能规划动作{tool: extract_content, parameters: {selector: .product_pod}}来提取商品信息。步骤3提取的内容会存入记忆。LLM可能会判断是否已获取足够信息或者是否需要翻页scroll或点击next按钮。最终输出程序会将memory中收集到的所有extracted_data打印出来。数据是原始文本你可以看到包含了书名和价格信息的HTML片段或纯文本。4.3 数据后处理Agent提取的往往是原始文本我们需要一个后处理步骤来清洗和结构化数据。可以在Agent循环结束后或者设计一个专门的post_process工具。示例后处理函数可添加到tools.py或新建utils.pyimport re from bs4 import BeautifulSoup def parse_book_list(html_content: str): 从书籍列表页的HTML中解析结构化数据 soup BeautifulSoup(html_content, html.parser) books [] # 根据目标网站结构定位商品块 product_pods soup.select(.product_pod) for pod in product_pods: title_elem pod.select_one(h3 a) price_elem pod.select_one(.price_color) book {} if title_elem: book[title] title_elem.get(title, ).strip() if price_elem: # 去除货币符号并转换为浮点数 price_text price_elem.get_text().strip() book[price] float(re.sub(r[^\d.], , price_text)) if book: # 只添加有数据的项 books.append(book) return books # 在主程序中调用 # final_html collected_data[-1][final_extraction] # 假设最后一条是提取的HTML # structured_books parse_book_list(final_html) # print(json.dumps(structured_books, indent2, ensure_asciiFalse))5. 常见问题与排查思路在搭建和运行Agent过程中你可能会遇到以下问题问题现象常见原因解决思路浏览器无法启动或闪退1. Playwright浏览器未安装。2. 系统缺少依赖库Linux常见。3. 端口冲突或已有浏览器实例。1. 运行playwright install chromium。2. 根据Playwright官方文档安装系统依赖。3. 检查并关闭冲突的浏览器进程。LLM不返回JSON或返回格式错误1. Prompt中JSON格式要求不明确。2. LLM的temperature参数过高导致输出随机。3. 上下文过长或混乱。1. 强化System Prompt中的输出格式指令使用“必须”、“严格”等词。2. 将temperature调低如0.1。3. 精简action_history和page_summary的长度只保留关键信息。Agent陷入循环或重复无效操作1. 页面状态识别错误LLM无法感知操作已生效或失败。2. 目标定义不清晰。3. 缺乏有效的停止条件。1. 优化get_page_summary函数使其能更准确地反映页面关键变化如URL变化、特定元素出现。2. 将大目标拆解为更原子化的子目标。3. 在规划器中加入更明确的成功/失败判断逻辑并设置max_steps。被目标网站检测并屏蔽1. 浏览器指纹被识别为自动化工具。2. 操作频率过高像机器人。1. 使用更真实的user_agent并考虑使用playwright的stealth模式或第三方反检测插件需谨慎评估。2. 在操作间增加随机延迟 (await asyncio.sleep(random.uniform(1, 3)))。3. 考虑使用代理IP池。务必遵守网站规则。OpenAI API调用超时或报错1. API Key无效或余额不足。2. 网络连接问题。3. 请求速率超限。1. 检查.env文件中的OPENAI_API_KEY并在OpenAI后台检查余额和用量。2. 检查网络或设置API请求的timeout参数。3. 降低请求频率或升级API套餐。元素定位失败Click/Type不生效1. 选择器Selector不正确或不稳定。2. 元素尚未加载完成或位于iframe内。3. 页面是SPA单页应用URL未变但内容已变。1. 让LLM尝试更通用的选择器如通过文本内容text或使用Playwright的locatorAPI结合多种策略。2. 增加wait_for_selector的超时时间或检查是否存在iframe。3. 在get_page_summary中更多依赖页面文本内容而非URL来判断状态。6. 最佳实践与工程建议要将这个Demo级别的Agent用于更严肃的项目需要考虑以下工程化改进6.1 提升Agent的智能与可靠性更精细的工具设计增加wait_for_element、screenshot用于视觉理解、solve_captcha集成打码服务等工具。分层规划Hierarchical Planning不要指望LLM一次规划所有步骤。可以设计一个顶层规划器分解高级目标再由底层规划器执行具体步骤。丰富页面状态描述get_page_summary可以不仅包含文本还可以提取关键元素的属性如按钮状态、输入框值、URL哈希等为LLM提供更全面的“观察”。集成视觉模型对于极度依赖页面布局的网站可以考虑使用多模态模型如GPT-4V来分析屏幕截图实现更接近人类的视觉定位。6.2 性能与成本优化本地模型替代对于简单任务或对成本敏感的场景可以使用本地部署的轻量级LLM如通过Ollama运行的Llama 3、Qwen等并通过函数调用Function Calling格式与之交互。缓存与记忆优化缓存LLM对相似页面状态的规划结果避免重复调用。使用向量数据库存储历史页面摘要实现更高效的上下文检索。并发与异步如果需要采集大量页面可以设计一个主控Agent负责规划多个Worker Agent并行执行浏览器操作注意管理浏览器实例资源。6.3 工程化与可维护性配置化管理将目标网站的特定规则如登录流程、商品列表选择器、翻页逻辑抽象为配置文件或知识库让Agent动态加载而不是硬编码在Prompt中。日志与监控记录详细的运行日志包括每一步的规划决策、执行结果、页面截图。这便于调试和优化Agent策略。模块化设计将规划器、工具集、记忆模块设计为可插拔的接口方便替换不同的LLM提供商、浏览器驱动或算法。伦理与合规严格遵守robots.txt在访问任何网站前先检查其robots.txt文件尊重Disallow规则。控制请求速率在工具函数中内置延迟避免对目标服务器造成压力。明确数据用途仅采集公开数据用于合法合规的分析、研究或个人学习不得用于商业侵权或恶意竞争。用户代理标识考虑在User-Agent中友好地标识你的Bot并提供一个联系邮箱以示尊重。6.4 应对复杂场景登录与会话保持实现login工具处理表单填写、Cookie/Session保存。Playwright的context.storage_state()可以方便地保存和恢复登录状态。处理验证码集成第三方验证码识别服务如2Captcha、DeathByCaptcha当检测到验证码时自动调用服务并填写。处理无限滚动设计智能的滚动策略结合scroll工具和页面内容变化检测判断何时停止滚动。Web逆向的战场正在从“逆向工程”转向“智能交互”。AI Agent通过模拟人类浏览行为和理解能力为动态网页的数据采集提供了一条新的、更具通用性的路径。它并不能解决所有问题例如对于数据量极大、对实时性要求极高的场景直接调用API仍是首选但它极大地降低了许多中等复杂度网站的采集门槛和维护成本。本文提供的代码是一个完整的起点你可以在此基础上根据具体业务需求扩展工具集、优化提示词、集成更强大的模型。技术的本质是解决问题Agent为我们打开了一扇新的大门。 基于AI Agent的Web数据采集：绕过JS逆向与反爬的智能解决方案
最近在尝试爬取一些电商平台数据时发现传统的Web逆向如JS逆向、反调试绕过越来越难搞。复杂的混淆、动态加密、环境检测让逆向分析耗时耗力甚至可能刚分析完对方就更新了防护。有没有一种更“智能”、更通用的方法答案是肯定的那就是Agent智能体。本文要探讨的不是传统意义上用于自动化测试的Selenium/Puppeteer而是结合了大型语言模型LLM推理能力的AI Agent。它能够理解网页结构、模拟人类操作逻辑、处理动态内容甚至应对一些简单的反爬策略实现“通杀”级别的数据采集。下面我将从概念到实战手把手带你搭建一个面向电商平台的AI Agent并附上核心源码和完整的搭建流程。无论你是数据采集开发者还是对AI应用感兴趣的同学都能从中获得一套可直接复用的解决方案。1. 背景与核心概念为什么说Agent是Web逆向的新思路在深入之前我们首先要厘清几个关键概念并理解传统Web逆向的痛点与Agent方案的突破点。1.1 传统Web逆向的困境传统的Web数据采集尤其是针对反爬严密的网站通常遵循“分析-破解-模拟”的路径静态分析查看网页源码、Network请求寻找数据接口。JS逆向如果接口参数被JavaScript动态加密如sign、token则需要使用开发者工具调试定位加密函数并用Python如PyExecJS、Node.js子进程还原算法。环境模拟应对WebDriver检测、window.navigator属性检测、Canvas指纹等。痛点在于高成本每个网站、甚至同一网站的不同接口其加密逻辑都可能不同需要重复投入大量分析时间。高维护成本网站前端稍作更新加密逻辑或环境检测就可能变化导致爬虫失效需要重新分析。技术门槛高需要深厚的JavaScript、浏览器原理、密码学知识。脆弱性对抗性升级道高一尺魔高一丈。1.2 什么是AI AgentAI Agent智能体在此语境下指的是一个能够感知环境、自主决策、执行动作以实现目标的软件实体。结合LLM如GPT-4、Claude、国产大模型它具备了强大的自然语言理解、逻辑推理和代码生成能力。一个用于Web操作的AI Agent通常包含以下核心组件规划器Planner将用户的高层目标如“获取商品X的价格和评论”分解为一系列可执行的步骤如“打开网站-搜索商品-进入详情页-提取价格和评论”。记忆Memory存储历史操作、网页状态、提取到的数据用于上下文理解和避免重复操作。工具ToolsAgent可以调用的外部能力如打开浏览器、点击元素、输入文本、滚动页面、提取文本、执行JavaScript等。执行器Actuator实际驱动工具执行的模块如通过Playwright或Selenium控制浏览器。1.3 Agent如何“通杀”Web逆向Agent的思路是**“降维打击”**。它不直接与复杂的加密逻辑硬碰硬而是模拟真人操作通过浏览器自动化工具以真实用户的身份访问网站加载完整的JavaScript环境让网站自己完成所有加密和渲染。这直接绕过了静态分析和JS还原的步骤。智能解析与决策LLM可以理解网页的语义结构。你无需编写复杂的XPath或CSS Selector来定位元素只需用自然语言告诉Agent“找到那个显示价格的红色数字”或“点击‘下一页’按钮”。Agent能理解指令并找到对应元素。处理动态与异常遇到验证码、弹窗、页面布局变化你可以为Agent设计应对策略例如“如果看到滑块验证调用打码平台API”或者“如果找不到商品列表尝试滚动页面”。LLM可以根据当前页面状态做出判断。简而言之Agent将爬虫开发从“密码学工程”部分转变为了“流程设计与自然语言指令工程”大大降低了技术壁垒并提升了方案的通用性和鲁棒性。2. 环境准备与版本说明我们将构建一个基于Python、Playwright浏览器自动化和OpenAI APILLM能力的电商数据采集Agent。你也可以替换为其他LLM如通过Ollama部署的本地模型。核心环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文示例在 macOS/Linux 下完成。Python: 3.8 或更高版本。推荐使用 3.10。包管理pip或conda。主要依赖库及版本建议使用虚拟环境# 创建并激活虚拟环境 (可选) python -m venv web_agent_env source web_agent_env/bin/activate # Linux/macOS # web_agent_env\Scripts\activate # Windows # 安装核心依赖 pip install playwright1.40.0 # 浏览器自动化 pip install openai1.12.0 # OpenAI API 客户端 pip install python-dotenv1.0.0 # 环境变量管理 pip install beautifulsoup44.12.3 # 备用HTML解析 pip install lxml5.2.1 # 解析器 # 安装Playwright浏览器内核 playwright install chromium重要配置OpenAI API Key你需要从 OpenAI平台 获取一个有效的API Key。我们将它存储在环境变量中以保证安全。项目结构建议如下组织你的代码。web_scraping_agent/ ├── .env # 存储API KEY等敏感信息 ├── config.py # 配置文件 ├── agent_core.py # Agent核心逻辑 ├── tools.py # 自定义工具集 ├── planner.py # 规划器 ├── memory.py # 记忆模块 ├── main.py # 主程序入口 ├── requirements.txt # 依赖列表 └── logs/ # 日志目录3. 核心组件拆解构建Agent的四大支柱我们的Agent架构将围绕规划、工具、记忆、执行来构建。下面逐一拆解其原理和实现。3.1 工具ToolsAgent的手和脚工具是Agent与浏览器环境交互的接口。我们基于Playwright封装一系列原子操作。tools.py核心代码import asyncio from playwright.async_api import async_playwright, Page, BrowserContext from typing import Any, Dict, Optional import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class BrowserTools: 浏览器操作工具集 def __init__(self): self.playwright None self.browser None self.context: Optional[BrowserContext] None self.page: Optional[Page] None async def start_browser(self, headless: bool False): 启动浏览器实例。headlessFalse便于调试。 self.playwright await async_playwright().start() # 使用Chromium更稳定且指纹相对统一 self.browser await self.playwright.chromium.launch( headlessheadless, args[--disable-blink-featuresAutomationControlled] # 禁用自动化控制特征 ) # 创建上下文可以统一设置User-Agent、视口等 self.context await self.browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 ) self.page await self.context.new_page() logger.info(浏览器启动成功) async def goto(self, url: str): 导航到指定URL if not self.page: raise RuntimeError(浏览器页面未初始化请先调用start_browser) await self.page.goto(url, wait_untilnetworkidle) # 等待网络空闲 logger.info(f已导航至: {url}) return await self.get_page_summary() async def click(self, selector: str, description: str ): 点击元素 if not self.page: raise RuntimeError(浏览器页面未初始化) try: await self.page.wait_for_selector(selector, statevisible, timeout10000) await self.page.click(selector) logger.info(f点击元素: {description or selector}) await asyncio.sleep(1) # 等待页面反应 return await self.get_page_summary() except Exception as e: logger.error(f点击元素失败 {selector}: {e}) return f操作失败: 无法点击元素 {selector}错误: {e} async def type_text(self, selector: str, text: str): 在输入框输入文本 if not self.page: raise RuntimeError(浏览器页面未初始化) try: await self.page.wait_for_selector(selector, statevisible, timeout10000) await self.page.fill(selector, text) logger.info(f在 {selector} 中输入文本: {text}) await asyncio.sleep(0.5) return await self.get_page_summary() except Exception as e: logger.error(f输入文本失败 {selector}: {e}) return f操作失败: 无法在 {selector} 中输入文本错误: {e} async def scroll(self, direction: str down, pixels: int 500): 滚动页面 if not self.page: raise RuntimeError(浏览器页面未初始化) try: if direction down: await self.page.mouse.wheel(0, pixels) elif direction up: await self.page.mouse.wheel(0, -pixels) logger.info(f向{direction}滚动{pixels}像素) await asyncio.sleep(1) # 等待内容加载 return await self.get_page_summary() except Exception as e: logger.error(f滚动失败: {e}) return f滚动操作失败: {e} async def extract_content(self, selector: str body): 提取指定选择器的文本内容 if not self.page: raise RuntimeError(浏览器页面未初始化) try: element await self.page.wait_for_selector(selector, timeout10000) text await element.inner_text() logger.info(f从 {selector} 提取内容长度: {len(text)}) # 返回精简后的内容避免上下文过长 return text[:2000] (... if len(text) 2000 else ) except Exception as e: logger.error(f提取内容失败 {selector}: {e}) return f提取失败: 无法从 {selector} 获取内容错误: {e} async def get_page_summary(self) - str: 获取当前页面的关键信息摘要用于反馈给LLM if not self.page: return 页面未就绪 try: # 获取页面标题和URL title await self.page.title() url self.page.url # 获取可见区域的主要文本简化版 main_content await self.page.locator(body).inner_text() summary f页面标题: {title}\n当前URL: {url}\n页面内容预览:\n{main_content[:500]}... return summary except Exception as e: return f获取页面摘要失败: {e} async def close(self): 关闭浏览器资源 if self.browser: await self.browser.close() if self.playwright: await self.playwright.stop() logger.info(浏览器资源已释放)关键点解析反检测args[--disable-blink-featuresAutomationControlled]是Playwright提供的禁用自动化控制特征的参数能有效降低被简单脚本检测的风险。等待策略wait_untilnetworkidle和wait_for_selector确保了页面或元素加载完成后再操作提高了稳定性。错误处理每个工具函数都包含try-except并将错误信息以自然语言形式返回便于LLM理解并调整策略。页面摘要get_page_summary函数至关重要。它将当前页面的状态标题、URL、核心内容总结成一段文本作为LLM决策的“观察”。3.2 规划器Planner与LLM集成Agent的大脑规划器的核心是调用LLM根据用户目标和当前页面状态决定下一步该执行哪个工具函数。我们使用OpenAI的Chat Completion API。planner.py核心代码import openai import os from dotenv import load_dotenv import json import logging from typing import List, Dict, Any load_dotenv() # 加载.env文件中的环境变量 openai.api_key os.getenv(OPENAI_API_KEY) logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class AgentPlanner: 基于LLM的规划器决定下一步行动 def __init__(self, model: str gpt-3.5-turbo): self.model model self.tools_description self._get_tools_description() def _get_tools_description(self) - str: 定义可供Agent使用的工具列表及其描述 tools [ { name: goto, description: 导航到一个新的URL。输入应为完整的网址。, parameters: {url: string} }, { name: click, description: 点击页面上的一个元素。你需要提供CSS选择器或对元素的描述。, parameters: {selector: string, description: string (optional)} }, { name: type_text, description: 在输入框内输入文本。你需要提供输入框的选择器和要输入的文本。, parameters: {selector: string, text: string} }, { name: scroll, description: 向上或向下滚动页面。用于加载更多内容。, parameters: {direction: string (down/up), pixels: integer} }, { name: extract_content, description: 从页面中提取特定元素的文本内容。默认提取整个body。, parameters: {selector: string (optional)} } ] # 将工具描述格式化为LLM容易理解的文本 desc_lines [你可以使用以下工具] for tool in tools: desc_lines.append(f- {tool[name]}: {tool[description]} 参数: {tool[parameters]}) return \n.join(desc_lines) def plan_next_action(self, goal: str, current_page_summary: str, action_history: List[str]) - Dict[str, Any]: 根据目标、当前页面状态和历史规划下一个动作。 返回一个包含工具名和参数的字典。 # 构建给LLM的提示词Prompt system_prompt f你是一个专业的网页操作智能体Web Agent。你的任务是理解用户目标并通过操作浏览器来达成目标。 {self.tools_description} 行动准则 1. 仔细分析当前页面摘要理解你在网站中的位置。 2. 回顾历史操作避免重复无效动作。 3. 一次只执行一个最直接、最必要的动作。 4. 如果页面摘要显示目标已达成如已找到所需信息则返回工具名为 extract_content 来提取最终数据。 5. 如果卡住或找不到元素尝试 scroll 或重新评估。 6. 你的输出必须是严格的JSON格式{{tool: 工具名, parameters: {{参数1: 值1, ...}}}} user_prompt f 用户目标{goal} 当前页面状态 {current_page_summary} 已执行的操作历史最近5条 {chr(10).join(action_history[-5:]) if action_history else 无} 请根据以上信息决定下一步应该执行哪个工具操作并给出具体的参数。 只输出JSON不要有其他任何解释。 messages [ {role: system, content: system_prompt}, {role: user, content: user_prompt} ] try: response openai.chat.completions.create( modelself.model, messagesmessages, temperature0.1, # 低随机性保证决策稳定 max_tokens500 ) llm_output response.choices[0].message.content.strip() logger.info(fLLM原始输出: {llm_output}) # 解析LLM返回的JSON action json.loads(llm_output) # 基本验证 if tool not in action or parameters not in action: raise ValueError(LLM返回的JSON格式不正确缺少tool或parameters字段) return action except json.JSONDecodeError as e: logger.error(f解析LLM输出的JSON失败: {e}, 输出内容: {llm_output}) # 降级策略返回一个安全的默认动作比如重新获取页面状态 return {tool: extract_content, parameters: {selector: body}} except Exception as e: logger.error(f调用LLM规划时发生错误: {e}) raise关键点解析提示词工程Prompt Engineering这是Agent智能度的核心。system_prompt定义了Agent的角色、可用工具和行动准则。user_prompt提供了当前任务的具体上下文目标、页面状态、历史。清晰的提示词能极大提升LLM决策的准确性。输出约束强制要求LLM输出严格的JSON格式便于程序化解析。这是实现LLM与代码协作的关键。错误处理与降级当LLM输出不符合JSON格式时我们有降级策略例如默认执行extract_content避免整个流程崩溃。历史记录将最近的操作历史提供给LLM有助于它避免循环操作比如反复点击同一个无效按钮。3.3 记忆Memory与执行循环连接大脑与手脚记忆模块负责存储操作历史和中间数据。执行循环则是驱动整个Agent运行的主逻辑。memory.py核心代码from typing import List, Dict, Any class SimpleMemory: 简单的记忆模块存储会话历史和数据 def __init__(self, max_history_len: int 50): self.action_history: List[str] [] # 记录每一步操作描述 self.extracted_data: List[Dict[str, Any]] [] # 记录提取到的数据 self.max_history_len max_history_len def add_action(self, action: str, parameters: Dict, result: str): 记录一次操作及其结果 entry f动作: {action}({parameters}) - 结果摘要: {result[:100]}... self.action_history.append(entry) # 限制历史长度防止上下文过长 if len(self.action_history) self.max_history_len: self.action_history.pop(0) def add_data(self, data: Dict[str, Any]): 记录提取到的结构化数据 self.extracted_data.append(data) def get_recent_history(self, n: int 5) - List[str]: 获取最近N条操作历史 return self.action_history[-n:] if self.action_history else [] def get_all_data(self) - List[Dict[str, Any]]: 获取所有已提取的数据 return self.extracted_dataagent_core.py- 执行循环核心import asyncio from typing import Dict, Any, Optional from tools import BrowserTools from planner import AgentPlanner from memory import SimpleMemory import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class WebScrapingAgent: Web采集Agent的核心执行器 def __init__(self, planner: AgentPlanner, headless: bool False): self.planner planner self.browser_tools BrowserTools() self.memory SimpleMemory() self.headless headless self.is_running False async def run(self, goal: str, start_url: Optional[str] None): 运行Agent尝试达成目标 self.is_running True try: # 1. 启动浏览器 await self.browser_tools.start_browser(headlessself.headless) logger.info(f开始执行目标: {goal}) # 2. 如果有起始URL先导航过去 current_state 浏览器已启动等待指令。 if start_url: current_state await self.browser_tools.goto(start_url) self.memory.add_action(goto, {url: start_url}, current_state) max_steps 20 # 防止无限循环 step 0 # 3. 主循环规划 - 执行 - 观察 - 记录 while self.is_running and step max_steps: step 1 logger.info(f--- 第 {step} 步 ---) # a. 规划下一步 recent_history self.memory.get_recent_history() next_action self.planner.plan_next_action(goal, current_state, recent_history) tool_name next_action[tool] params next_action[parameters] logger.info(f规划动作: {tool_name} with {params}) # b. 执行动作 tool_func getattr(self.browser_tools, tool_name, None) if not tool_func: logger.error(f未知工具: {tool_name}) current_state f错误: 未知工具 {tool_name} self.memory.add_action(tool_name, params, current_state) break try: # 调用工具函数 result await tool_func(**params) current_state result # 更新当前状态为执行结果 logger.info(f动作执行结果: {result[:200]}...) # c. 记录到记忆 self.memory.add_action(tool_name, params, result) # d. 简单目标检查示例如果结果中包含“价格”等关键词可能意味着成功 # 这里可以设计更复杂的目标达成判定逻辑 if 价格 in result or 商品详情 in result: logger.info(可能已找到目标信息尝试提取最终数据。) # 触发一次最终数据提取 final_data await self.browser_tools.extract_content(body) self.memory.add_data({final_extraction: final_data}) # 可以根据需要决定是否继续或停止 # break except Exception as e: error_msg f执行工具 {tool_name} 时出错: {e} logger.error(error_msg) current_state error_msg self.memory.add_action(tool_name, params, current_state) # 出错后可以尝试让LLM规划恢复策略这里简单跳过或停止 # 例如可以等待一下再继续 await asyncio.sleep(2) # 短暂暂停模拟人类操作间隔也避免请求过快 await asyncio.sleep(1) logger.info(fAgent执行结束共执行 {step} 步。) # 返回最终收集到的所有数据 return self.memory.get_all_data() except Exception as e: logger.error(fAgent运行过程中发生严重错误: {e}) raise finally: # 4. 清理资源 await self.browser_tools.close() self.is_running False关键点解析状态驱动循环Agent的核心是一个循环。每次迭代它都基于当前页面状态current_state和历史recent_history通过规划器planner决定下一个动作然后调用工具tool执行并将结果更新为新的状态如此往复。可配置的停止条件我们设置了最大步数max_steps来防止无限循环。在实际应用中你还可以设计更智能的停止条件例如当LLM规划出stop动作时、当成功提取到目标数据时、或当长时间无进展时。异常处理与鲁棒性在执行工具时进行异常捕获并将错误信息记录到记忆和状态中。这样在下一轮规划时LLM就能知道上一步出错了并可能尝试不同的策略。资源管理在finally块中确保浏览器被正确关闭避免资源泄漏。4. 完整实战案例抓取电商平台商品信息现在我们将上述组件组装起来完成一个具体任务在某个电商平台以示例网站为例上搜索“Python编程书籍”并获取第一页的商品列表信息名称、价格。注意出于法律和道德考虑本示例将使用一个公开的、允许爬取的测试网站或模拟环境如https://books.toscrape.com/进行演示。在实际应用中请务必遵守目标网站的robots.txt协议和服务条款。4.1 项目主程序与配置.env文件OPENAI_API_KEY你的OpenAI_API_Key_放在这里main.py主程序import asyncio import sys import os sys.path.append(os.path.dirname(os.path.abspath(__file__))) from agent_core import WebScrapingAgent from planner import AgentPlanner import json async def main(): 主函数定义任务并运行Agent # 1. 初始化规划器使用GPT-3.5-turbo成本较低 planner AgentPlanner(modelgpt-3.5-turbo) # 2. 初始化AgentheadlessFalse便于观察调试 agent WebScrapingAgent(plannerplanner, headlessFalse) # 3. 定义目标任务 goal 请访问网站 https://books.toscrape.com/。 这是一个在线书店。 你的目标是找到所有关于“Python”主题的书籍并列出它们的书名和价格。 如果第一页没有请尝试搜索或翻页。 start_url https://books.toscrape.com/ print(f开始执行目标\n{goal}) print(*50) try: # 4. 运行Agent collected_data await agent.run(goalgoal, start_urlstart_url) # 5. 输出结果 print(\n *50) print(任务执行完成收集到的数据如下) print(*50) for i, data_item in enumerate(collected_data): print(f\n--- 数据块 {i1} ---) # 美化打印JSON print(json.dumps(data_item, indent2, ensure_asciiFalse)) except Exception as e: print(f运行过程中出错: {e}) finally: print(\n程序结束。) if __name__ __main__: asyncio.run(main())4.2 运行与结果分析运行程序在终端中执行python main.py。观察过程由于设置了headlessFalse你会看到一个浏览器窗口自动打开导航到书籍网站。然后你会看到Agent在“思考”调用LLM并自动执行点击、滚动等操作。可能的执行序列步骤1LLM根据目标“找到Python书籍”看到首页是分类可能规划动作{tool: click, parameters: {selector: a[href*python], description: 点击Python分类链接}}。步骤2如果直接点击成功页面跳转到Python书籍列表。LLM看到列表后可能规划动作{tool: extract_content, parameters: {selector: .product_pod}}来提取商品信息。步骤3提取的内容会存入记忆。LLM可能会判断是否已获取足够信息或者是否需要翻页scroll或点击next按钮。最终输出程序会将memory中收集到的所有extracted_data打印出来。数据是原始文本你可以看到包含了书名和价格信息的HTML片段或纯文本。4.3 数据后处理Agent提取的往往是原始文本我们需要一个后处理步骤来清洗和结构化数据。可以在Agent循环结束后或者设计一个专门的post_process工具。示例后处理函数可添加到tools.py或新建utils.pyimport re from bs4 import BeautifulSoup def parse_book_list(html_content: str): 从书籍列表页的HTML中解析结构化数据 soup BeautifulSoup(html_content, html.parser) books [] # 根据目标网站结构定位商品块 product_pods soup.select(.product_pod) for pod in product_pods: title_elem pod.select_one(h3 a) price_elem pod.select_one(.price_color) book {} if title_elem: book[title] title_elem.get(title, ).strip() if price_elem: # 去除货币符号并转换为浮点数 price_text price_elem.get_text().strip() book[price] float(re.sub(r[^\d.], , price_text)) if book: # 只添加有数据的项 books.append(book) return books # 在主程序中调用 # final_html collected_data[-1][final_extraction] # 假设最后一条是提取的HTML # structured_books parse_book_list(final_html) # print(json.dumps(structured_books, indent2, ensure_asciiFalse))5. 常见问题与排查思路在搭建和运行Agent过程中你可能会遇到以下问题问题现象常见原因解决思路浏览器无法启动或闪退1. Playwright浏览器未安装。2. 系统缺少依赖库Linux常见。3. 端口冲突或已有浏览器实例。1. 运行playwright install chromium。2. 根据Playwright官方文档安装系统依赖。3. 检查并关闭冲突的浏览器进程。LLM不返回JSON或返回格式错误1. Prompt中JSON格式要求不明确。2. LLM的temperature参数过高导致输出随机。3. 上下文过长或混乱。1. 强化System Prompt中的输出格式指令使用“必须”、“严格”等词。2. 将temperature调低如0.1。3. 精简action_history和page_summary的长度只保留关键信息。Agent陷入循环或重复无效操作1. 页面状态识别错误LLM无法感知操作已生效或失败。2. 目标定义不清晰。3. 缺乏有效的停止条件。1. 优化get_page_summary函数使其能更准确地反映页面关键变化如URL变化、特定元素出现。2. 将大目标拆解为更原子化的子目标。3. 在规划器中加入更明确的成功/失败判断逻辑并设置max_steps。被目标网站检测并屏蔽1. 浏览器指纹被识别为自动化工具。2. 操作频率过高像机器人。1. 使用更真实的user_agent并考虑使用playwright的stealth模式或第三方反检测插件需谨慎评估。2. 在操作间增加随机延迟 (await asyncio.sleep(random.uniform(1, 3)))。3. 考虑使用代理IP池。务必遵守网站规则。OpenAI API调用超时或报错1. API Key无效或余额不足。2. 网络连接问题。3. 请求速率超限。1. 检查.env文件中的OPENAI_API_KEY并在OpenAI后台检查余额和用量。2. 检查网络或设置API请求的timeout参数。3. 降低请求频率或升级API套餐。元素定位失败Click/Type不生效1. 选择器Selector不正确或不稳定。2. 元素尚未加载完成或位于iframe内。3. 页面是SPA单页应用URL未变但内容已变。1. 让LLM尝试更通用的选择器如通过文本内容text或使用Playwright的locatorAPI结合多种策略。2. 增加wait_for_selector的超时时间或检查是否存在iframe。3. 在get_page_summary中更多依赖页面文本内容而非URL来判断状态。6. 最佳实践与工程建议要将这个Demo级别的Agent用于更严肃的项目需要考虑以下工程化改进6.1 提升Agent的智能与可靠性更精细的工具设计增加wait_for_element、screenshot用于视觉理解、solve_captcha集成打码服务等工具。分层规划Hierarchical Planning不要指望LLM一次规划所有步骤。可以设计一个顶层规划器分解高级目标再由底层规划器执行具体步骤。丰富页面状态描述get_page_summary可以不仅包含文本还可以提取关键元素的属性如按钮状态、输入框值、URL哈希等为LLM提供更全面的“观察”。集成视觉模型对于极度依赖页面布局的网站可以考虑使用多模态模型如GPT-4V来分析屏幕截图实现更接近人类的视觉定位。6.2 性能与成本优化本地模型替代对于简单任务或对成本敏感的场景可以使用本地部署的轻量级LLM如通过Ollama运行的Llama 3、Qwen等并通过函数调用Function Calling格式与之交互。缓存与记忆优化缓存LLM对相似页面状态的规划结果避免重复调用。使用向量数据库存储历史页面摘要实现更高效的上下文检索。并发与异步如果需要采集大量页面可以设计一个主控Agent负责规划多个Worker Agent并行执行浏览器操作注意管理浏览器实例资源。6.3 工程化与可维护性配置化管理将目标网站的特定规则如登录流程、商品列表选择器、翻页逻辑抽象为配置文件或知识库让Agent动态加载而不是硬编码在Prompt中。日志与监控记录详细的运行日志包括每一步的规划决策、执行结果、页面截图。这便于调试和优化Agent策略。模块化设计将规划器、工具集、记忆模块设计为可插拔的接口方便替换不同的LLM提供商、浏览器驱动或算法。伦理与合规严格遵守robots.txt在访问任何网站前先检查其robots.txt文件尊重Disallow规则。控制请求速率在工具函数中内置延迟避免对目标服务器造成压力。明确数据用途仅采集公开数据用于合法合规的分析、研究或个人学习不得用于商业侵权或恶意竞争。用户代理标识考虑在User-Agent中友好地标识你的Bot并提供一个联系邮箱以示尊重。6.4 应对复杂场景登录与会话保持实现login工具处理表单填写、Cookie/Session保存。Playwright的context.storage_state()可以方便地保存和恢复登录状态。处理验证码集成第三方验证码识别服务如2Captcha、DeathByCaptcha当检测到验证码时自动调用服务并填写。处理无限滚动设计智能的滚动策略结合scroll工具和页面内容变化检测判断何时停止滚动。Web逆向的战场正在从“逆向工程”转向“智能交互”。AI Agent通过模拟人类浏览行为和理解能力为动态网页的数据采集提供了一条新的、更具通用性的路径。它并不能解决所有问题例如对于数据量极大、对实时性要求极高的场景直接调用API仍是首选但它极大地降低了许多中等复杂度网站的采集门槛和维护成本。本文提供的代码是一个完整的起点你可以在此基础上根据具体业务需求扩展工具集、优化提示词、集成更强大的模型。技术的本质是解决问题Agent为我们打开了一扇新的大门。