1. 项目概述:当Web自动化遇上大语言模型
最近在搞自动化测试和RPA(机器人流程自动化)的朋友,估计都遇到过同一个头疼的问题:现代Web应用越来越“花里胡哨”了。动态加载、状态切换、元素属性随机生成……传统的基于XPath或CSS选择器的脚本,动不动就“找不到元素”而崩溃,维护成本高得吓人。我自己就曾为了一个购物车按钮的># 示例:获取页面关键信息 async def get_page_context(page): # 1. 页面标题和URL title = await page.title() url = page.url # 2. 获取所有可交互元素的简化表示 elements = [] all_buttons = await page.locator('button, a, input, [role="button"]').all() for btn in all_buttons: if await btn.is_visible(): # 只关心可见元素 text = await btn.text_content() or await btn.get_attribute('aria-label') or '' # 获取一个相对稳定的选择器(优先使用Playwright的定位器逻辑) # 这里简化处理,实际可使用 `btn._impl_obj._description` 或自定义算法生成唯一ID selector = await generate_stable_selector(btn) elements.append({ 'type': await btn.evaluate('el => el.tagName.toLowerCase()'), 'text': text[:50], # 截断长文本 'selector': selector }) # 3. 获取页面主要文本内容(段落、标题) main_content = await page.locator('body').inner_text() main_content = ' '.join(main_content.split()[:200]) # 摘要 return { 'title': title, 'url': url, 'elements': elements, 'main_text': main_content }
生成稳定元素标识:这是连接LLM决策和Playwright执行的桥梁。不要依赖LLM生成XPath。可以在提取元素时,利用Playwright内置的智能定位器,或者计算元素的文本、角色、名称属性的组合哈希,生成一个唯一ID。LLM在决策时,只需引用这个ID或与之关联的简短描述。
实操心得:
page.get_by_role()和page.get_by_text()是Playwright提供的非常强大的、面向可访问性的定位方式,它们比脆弱的CSS选择器稳定得多。在构建上下文时,可以优先尝试用这些方法为元素生成描述。例如,一个按钮可能同时有<button>提交</button>和get_by_role("button", name="提交")两种表示,后者对LLM更友好,对代码也更具可读性。
3.3 动作执行与循环控制
有了LLM的决策和稳定的元素标识,执行就相对直接了。核心是构建一个鲁棒的循环。
核心循环逻辑:
async def ai_automation_loop(page, initial_instruction): context = await get_page_context(page) task_stack = decompose_task(initial_instruction) # 任务分解 while task_stack: current_step = task_stack[0] # 构建包含当前步骤和页面上下文的Prompt prompt = build_prompt(current_step, context) # 调用LLM API (如OpenAI, Claude, 或本地模型) llm_response = await call_llm_api(prompt) action = parse_llm_response(llm_response) # 解析为JSON if action['name'] == 'done': task_stack.pop(0) print("当前步骤完成!") continue elif action['name'] == 'click': # 根据LLM返回的元素描述或ID,找到对应的Playwright定位器 locator = find_locator_by_description(page, action['element_description']) if locator: await locator.click() else: # 找不到元素,可能是页面变了,重新获取上下文 print(f"未找到元素: {action['element_description']}") elif action['name'] == 'fill': # ... 类似处理填充动作 # ... 处理其他动作 # 执行后,等待页面稳定并获取新上下文 await page.wait_for_load_state('networkidle') context = await get_page_context(page) # 可选:验证步骤是否成功,失败则重试或加入异常处理循环控制中的关键点:
- 超时与重试:每个动作执行后都应设置合理的超时。失败后不应立即放弃,可以重试几次,或者让LLM基于新上下文重新决策。
- 状态验证:如何判断一个步骤(如“登录成功”)完成了?可以定义一些验证规则,例如检查URL是否变化、特定元素(如用户头像)是否出现。这个验证逻辑也可以部分交给LLM:“请判断当前页面是否已成功登录。”
- 错误处理与降级:当LLM连续做出错误决策时,系统应能进入安全状态(如暂停),并可能回退到传统的、预定义的选择器流程,作为降级方案。
4. 实战:构建一个智能登录与搜索Agent
理论说了这么多,我们动手实现一个具体的例子:一个能登录Github并搜索指定仓库的智能Agent。这里我们使用OpenAI的GPT-4 API作为“大脑”,Playwright作为“手脚”。
4.1 环境搭建与依赖安装
首先,准备好Python环境(建议3.8+),然后安装核心库:
pip install playwright openai # 安装Playwright的浏览器 playwright install chromium你需要一个OpenAI的API密钥,并将其设置为环境变量OPENAI_API_KEY。
4.2 核心代码实现
我们创建一个名为github_ai_agent.py的文件。
import asyncio import json from openai import AsyncOpenAI from playwright.async_api import async_playwright # 初始化OpenAI客户端 client = AsyncOpenAI(api_key=os.environ.get("OPENAI_API_KEY")) class GithubAIAgent: def __init__(self): self.context = {} self.task_steps = [] async def get_page_context(self, page): """获取结构化页面上下文""" # 等待页面基本稳定 await page.wait_for_load_state("domcontentloaded") elements = [] # 获取所有可能的交互元素 locators = page.locator("button, a, input, textarea, [role='button'], [role='link']") count = await locators.count() for i in range(min(count, 50)): # 限制数量,避免过长 locator = locators.nth(i) if await locator.is_visible(): tag = await locator.evaluate("el => el.tagName.toLowerCase()") text = (await locator.text_content() or "").strip() placeholder = await locator.get_attribute("placeholder") or "" name = await locator.get_attribute("name") or "" type_attr = await locator.get_attribute("type") or "" # 生成一个描述性字符串和唯一标识 desc_parts = [] if text: desc_parts.append(f"文本:'{text[:30]}'") if placeholder: desc_parts.append(f"提示:'{placeholder[:20]}'") if name: desc_parts.append(f"名称:'{name}'") if type_attr: desc_parts.append(f"类型:{type_attr}") if not desc_parts: desc_parts.append(f"标签:<{tag}>") description = ",".join(desc_parts) # 使用Playwright的selector作为ID(在实际中可能需要更稳定的算法) selector = await locator.evaluate("el => { if(el.id) return '#'+el.id; // 这里可以实现更复杂的稳定选择器生成算法 return ''; }") elements.append({ "description": description, "selector": selector if selector else f"nth={i}", "tag": tag }) # 获取页面主要文本 main_text = await page.locator("body").inner_text() main_text = " ".join(main_text.split()[:100]) # 摘要 return { "url": page.url, "title": await page.title(), "elements": elements, "main_text": main_text } async def ask_llm(self, prompt): """调用LLM获取决策""" try: response = await client.chat.completions.create( model="gpt-4-turbo-preview", # 可根据需要调整模型 messages=[{"role": "user", "content": prompt}], temperature=0.1, # 低温度,保证输出稳定 max_tokens=500 ) return response.choices[0].message.content except Exception as e: print(f"调用LLM失败: {e}") return None def parse_llm_action(self, response_text): """解析LLM的回复,期望是JSON格式""" try: # 尝试从回复中提取JSON部分 lines = response_text.strip().split('\n') json_str = None for line in lines: if line.startswith('{') and line.endswith('}'): json_str = line break if not json_str and '```json' in response_text: # 处理代码块格式 start = response_text.find('```json') + 7 end = response_text.find('```', start) json_str = response_text[start:end].strip() action = json.loads(json_str) if json_str else json.loads(response_text) # 验证必要字段 if "action" not in action: raise ValueError("响应中缺少 'action' 字段") return action except json.JSONDecodeError as e: print(f"解析LLM响应为JSON失败: {e}") print(f"原始响应: {response_text}") return {"action": "retry", "reason": "parse_error"} async def execute_action(self, page, action): """执行LLM决策的动作""" action_type = action.get("action") if action_type == "navigate": url = action.get("url") if url: print(f"导航至: {url}") await page.goto(url, wait_until="networkidle") return True elif action_type == "click": element_desc = action.get("element_description", "") print(f"尝试点击: {element_desc}") # 这里简化处理:在实际应用中,需要将element_description映射回具体的定位器 # 我们可以根据描述中的关键词,在上下文元素列表中寻找最匹配的 # 本例中,我们简单使用Playwright的get_by_text进行模拟 if element_desc: # 提取描述中的文本部分(这是一个非常简单的启发式方法) import re match = re.search(r"文本:'([^']+)'", element_desc) if match: text_to_click = match.group(1) try: await page.get_by_text(text_to_click, exact=True).first.click(timeout=5000) print("点击成功") return True except Exception as e: print(f"点击失败: {e}") return False elif action_type == "fill": element_desc = action.get("element_description", "") text = action.get("text", "") print(f"尝试在 {element_desc} 中填写: {text}") # 类似click,需要映射到具体输入框 # 简化:寻找输入框并填充 try: # 假设第一个可见的输入框是目标 await page.locator("input:visible").first.fill(text) print("填充成功") return True except Exception as e: print(f"填充失败: {e}") return False elif action_type == "done": print("任务步骤完成!") return True elif action_type == "retry" or action_type == "wait": print("等待或重试...") await page.wait_for_timeout(2000) return True else: print(f"未知动作类型: {action_type}") return False async def run(self, task_description): """主运行循环""" print(f"开始任务: {task_description}") async with async_playwright() as p: browser = await p.chromium.launch(headless=False) # 设为True可无头运行 page = await browser.new_page() # 初始导航到Github await page.goto("https://github.com") # 分解任务(这里简化,实际可用LLM进行任务规划) self.task_steps = [ "导航到Github登录页面或找到登录入口", "在登录页输入用户名", "在登录页输入密码", "点击登录按钮", "登录后,找到搜索框", "在搜索框中输入‘playwright’并搜索" ] for step in self.task_steps: print(f"\n=== 当前步骤: {step} ===") max_retries = 3 for retry in range(max_retries): # 1. 获取当前页面上下文 self.context = await self.get_page_context(page) # 2. 构建Prompt prompt = f""" 你是一个网页自动化助手。请根据当前页面状态和任务步骤,决定下一步做什么。 当前页面信息: - 标题:{self.context['title']} - URL:{self.context['url']} - 主要文本摘要:{self.context['main_text'][:200]}... - 可交互元素(最多20个): {json.dumps(self.context['elements'][:20], indent=2, ensure_ascii=False)} 当前需要完成的任务步骤是:{step} 请从以下动作中选择一个,并严格按照JSON格式输出: - `navigate`: 导航到新URL。需要提供 "url"。 - `click`: 点击元素。需要提供 "element_description"(参考上方元素描述)。 - `fill`: 填写输入框。需要提供 "element_description" 和 "text"。 - `wait`: 等待2秒。 - `done`: 此步骤已完成。 请只输出JSON,不要有其他任何解释。 示例:{{"action": "click", "element_description": "文本:'Sign in',标签:<button>"}} 你的决策: """ # 3. 获取LLM决策 print("正在咨询LLM...") llm_response = await self.ask_llm(prompt) if not llm_response: print("LLM无响应,重试...") continue print(f"LLM回复: {llm_response[:200]}...") # 4. 解析并执行动作 action = self.parse_llm_action(llm_response) if action.get("action") == "done": print(f"步骤 '{step}' 标记为完成。") break success = await self.execute_action(page, action) if success: # 动作执行成功,短暂等待后进入下一步 await page.wait_for_timeout(1000) break else: print(f"动作执行失败,重试 ({retry+1}/{max_retries})...") await page.wait_for_timeout(1000) else: print(f"步骤 '{step}' 重试多次后仍失败,任务终止。") break print("\n=== 任务执行结束 ===") await page.wait_for_timeout(3000) await browser.close() # 运行主程序 async def main(): agent = GithubAIAgent() await agent.run("登录Github并搜索playwright仓库") if __name__ == "__main__": asyncio.run(main())4.3 代码解读与运行说明
这个示例虽然简化,但涵盖了核心流程:
- 初始化:启动浏览器,打开Github。
- 任务分解:我们预先定义了一个简单的步骤列表。在实际更复杂的场景中,这个“任务分解”步骤本身也可以由LLM完成。
- 循环执行:对每个步骤,获取页面上下文,构建Prompt,调用LLM,解析并执行动作。
- 动作映射:
execute_action函数是薄弱环节。示例中用了简单的文本匹配来点击元素,这在实际中是不可靠的。生产环境需要更鲁棒的映射机制,比如在get_page_context中为每个元素生成一个唯一ID,LLM决策时返回这个ID,执行时直接用ID找到对应的Playwright定位器。 - 错误处理:包含了重试机制,当动作执行失败时会重试最多3次。
运行它:将你的OpenAI API密钥设置为环境变量后,直接运行python github_ai_agent.py。你会看到浏览器启动,并尝试自动完成登录和搜索流程。由于Github的页面结构可能变化,且示例中的元素映射逻辑很初级,很可能无法一次成功,但这正是我们需要优化和调试的地方。
5. 成本、优化与常见问题排查
5.1 成本考量与优化策略
使用商用LLM API(如GPT-4)最大的顾虑就是成本。一次简单的任务可能需要多次调用,每次调用都消耗token。
优化策略:
- 上下文压缩:这是最有效的省钱方法。不要传送整个DOM。
- 过滤:只传送可见的、可交互的元素。
- 摘要:对长文本进行摘要。甚至可以先让一个“廉价”的模型(如GPT-3.5)对页面进行总结,再把总结送给“昂贵”的决策模型(如GPT-4)。
- 分块:对于超长页面,可以分区域(如“头部”、“主内容区”、“侧边栏”)获取上下文,一次只处理一个区域。
- 模型选型:对于决策逻辑,不一定需要最顶级的模型。
gpt-4-turbo比gpt-4便宜且快。对于简单的页面,gpt-3.5-turbo可能就足够了。可以设计一个评估机制,先用小模型,如果置信度低再fallback到大模型。 - 缓存:对于常见的页面和操作(如各种网站的登录页面),可以将LLM的决策结果(Prompt + 页面特征哈希 -> Action)缓存起来。下次遇到相似页面,直接使用缓存结果,无需再次调用API。
- 本地模型:如果对延迟和成本极度敏感,可以考虑使用开源的、可在本地部署的小型LLM(如Llama 3.1 8B, Qwen2.5 7B)。虽然能力可能稍弱,但经过特定任务的微调后,对于模式固定的自动化任务,表现可能出乎意料的好。
5.2 常见问题与调试技巧
在实际开发中,你会遇到各种各样的问题。下面是一些典型问题及其排查思路:
| 问题现象 | 可能原因 | 排查与解决思路 |
|---|---|---|
| LLM输出格式错误 | Prompt没有严格限制输出格式;模型温度参数过高。 | 1. 在Prompt中使用更明确的指令,如“必须输出JSON,且只包含如下字段...”。 2. 在代码中增加更健壮的JSON解析,尝试从回复中提取JSON块。 3. 将 temperature参数设为0或接近0的值。 |
| LLM决策循环(原地踏步) | 页面状态未发生预期变化,但LLM每次看到的上下文一样,于是做出相同决策。 | 1. 在上下文中加入“历史动作”记录,告诉LLM刚才做了什么,避免重复。 2. 增加超时和最大重试次数限制,超过后触发异常处理或请求人工干预。 3. 让LLM在决策时能选择“等待页面变化”或“滚动”等动作来获取新信息。 |
| 找不到元素/点击无效 | LLM的元素描述无法准确映射到Playwright定位器;页面尚未加载完成。 | 1.强化元素标识映射:这是核心。在提供上下文时,为每个元素计算一个唯一、稳定的签名(如基于文本、角色、邻近属性的哈希)。 2.增加等待:在执行动作前,使用 page.wait_for_selector或page.wait_for_function等待目标元素处于可交互状态。3.使用更鲁棒的定位器:优先使用 page.get_by_role()和page.get_by_text(),它们比CSS选择器更稳定。 |
| 任务分解不合理 | 初始的步骤分解过于笼统或错误,导致LLM无法执行。 | 1. 将任务分解也交给LLM去做,使用思维链(Chain-of-Thought)Prompting:“为了完成目标X,请列出具体的原子步骤”。 2. 建立常见任务的模板库,如“登录模板”、“搜索模板”、“表单填写模板”。 |
| API调用缓慢或失败 | 网络问题或API服务限流。 | 1. 实现指数退避重试机制。 2. 考虑使用异步调用并设置合理的超时。 3. 对于关键业务,需要有降级方案(如切换到规则引擎或停止任务)。 |
调试心得:初期,不要追求全自动。一定要加入详细的日志,打印出每一步的页面URL、提供给LLM的上下文摘要、LLM的完整回复以及执行的动作。这能帮你快速定位问题出在“看”(上下文)、“想”(LLM决策)还是“做”(Playwright执行)的环节。可以先将headless设为False,亲眼观察自动化的执行过程,这对理解问题有巨大帮助。
6. 进阶方向与应用场景拓展
这个“LLM + Playwright”的范式,其潜力远不止于自动登录和搜索。一旦跑通基础流程,你可以将它应用到更多令人兴奋的场景中:
- 智能端到端测试:不再是断言某个元素是否存在,而是让AI Agent模拟真实用户,执行完整的用户旅程(如注册-浏览-下单-支付),并基于页面反馈(如错误提示、成功消息)自主判断测试用例是否通过。它能发现那些脚本未覆盖但用户可能遇到的逻辑问题。
- 复杂业务流程自动化:处理需要跨多个系统、决策树复杂的流程。例如,“从邮箱读取客户询盘,登录CRM系统创建客户记录,根据产品类型在内部Wiki搜索解决方案模板,起草回复邮件”。LLM可以理解邮件内容,做出分支判断。
- 无障碍测试与监控:让Agent模拟屏幕阅读器用户或键盘导航用户,自动检测网站的可访问性问题,并生成报告。
- 反爬虫对抗的逆向工程:对于一些带有复杂反爬机制(如动态令牌、鼠标轨迹验证)的网站,可以尝试让LLM观察正常人类操作的模式,并模仿该模式进行操作。当然,这需要严格遵守法律法规和网站的使用条款。
- 结合视觉模型:纯DOM文本有时会丢失关键视觉信息(如图片验证码、图形位置)。可以结合多模态模型(如GPT-4V),对页面截图进行分析,实现“看到什么点什么”的真正视觉驱动自动化。
我个人的体会是,这项技术目前正处于“可用”到“好用”的过渡期。直接用它替代所有传统自动化是不现实的,成本和稳定性都是挑战。但它是一个强大的“增强”工具。最适合的场景是那些变化频繁、逻辑复杂、传统脚本维护成本极高的流程。你可以从一个小而具体的痛点开始,比如“自动处理每天收到的特定格式的工单”,让它与你的现有系统协同工作,逐步积累经验和优化策略。最大的收获不是完全解放双手,而是找到了一种让机器更“懂”人意图的交互方式,这本身就是一个充满可能性的方向。