1. 从“手动点点点”到“智能自动化”的跨越如果你还在用Selenium或者Puppeteer写一长串find_element和click来模拟浏览器操作每次页面结构一变就得重新调试脚本那感觉就像在用算盘处理大数据。我最初接触浏览器自动化也是从这个阶段过来的直到后来项目里需要处理大量动态渲染、反爬策略复杂的页面以及实现一些需要“观察-决策-执行”的智能流程时传统的脚本模式就彻底不够用了。这时候一个更底层的协议——Chrome DevTools Protocol也就是CDP进入了我的视野。CDP不是一个新的工具而是Chrome浏览器内置的调试协议。它像是一把能直接与浏览器内核对话的“手术刀”让你能控制标签页、拦截网络请求、执行JavaScript、获取DOM快照甚至监听内存变化。基于CDP构建自动化意味着你跳过了WebDriver这类“翻译官”的中间层直接与浏览器引擎交互响应更快能力也更强大。但CDP本身是协议不是框架直接用它写代码就像用汇编语言编程强大但繁琐。所以我们今天的重点不是罗列CDP的API而是如何以它为基础构建一个更高级的抽象Agent工作流。所谓Agent工作流你可以把它想象成一个在浏览器里“上班”的虚拟员工。它不再是被动执行预设步骤的脚本而是一个具备一定感知、决策和执行能力的智能体。它知道自己要完成什么任务比如“监控商品价格变化并截图报警”能通过CDP“看到”页面内容能根据页面状态“思考”下一步该点哪里、输入什么并能处理一些意外情况比如弹窗、验证码。这种模式正是当前从“自动化脚本”向“智能体应用”演进的核心实践。2. 为什么是CDP超越WebDriver的底层能力剖析在深入构建之前我们必须先搞清楚为什么选择CDP作为基石而不是继续用更成熟的Selenium WebDriver。这不仅仅是性能问题更是能力边界和设计哲学的不同。2.1 协议层 vs 驱动层本质差异WebDriver是一个W3C标准它定义了一套跨浏览器的、用于控制网页的通用指令集比如“点击元素”、“获取文本”。它的工作模式是你的脚本通过特定语言的客户端库如Python的selenium包发送指令给一个独立的驱动程序如chromedriver这个驱动再通过私有协议与真实的浏览器通信。这带来几个问题一是多了一层转换速度和效率有损耗二是驱动和浏览器版本必须严格匹配否则经常出兼容性问题三是能力受限于WebDriver标准一些浏览器独有的、高级的调试功能无法使用。CDP则完全不同。它是Chrome/Chromium内核原生暴露的调试接口基于WebSocket通信。你的代码直接连接到一个Chrome实例的调试端口发送和接收的都是JSON-RPC格式的消息。这意味着无中间商延迟更低指令直达浏览器引擎执行和获取结果的延迟显著降低对于需要高频交互的自动化场景至关重要。能力全集你能用到Chrome开发者工具里几乎所有的功能包括但不限于网络控制精确拦截、修改、放行任意请求和响应模拟弱网环境。性能分析获取时间线追踪、内存堆快照、计算性能指标。DOM深度操作不仅获取元素还能监听DOM变化获取CSS计算样式甚至模拟用户输入如输入法组合键。运行时注入在页面上下文中执行任意JavaScript并获取复杂的返回值如函数、Promise。页面快照生成包含完整CSSOM的截图甚至能生成PDF。版本兼容性更好虽然CDP本身也在演进但它的向后兼容性通常比chromedriver的版本锁死要好处理得多核心API相对稳定。2.2 构建Agent工作流的核心优势对于Agent工作流而言CDP提供的这些底层能力是构建“智能”的感官系统。感知能力Agent需要“看”懂页面。通过CDP的DOMSnapshot.captureSnapshot或Runtime.evaluateAgent可以获取结构化、带布局信息的DOM树这比通过innerHTML获取的字符串要强大得多。结合Accessibility.getFullAXTree还能获取无障碍树理解元素的语义角色这对于理解复杂UI组件如下拉菜单、滑块的状态非常有帮助。决策依据Agent的决策需要数据。CDP的Network域可以让Agent监听所有XHR/Fetch请求和响应这对于监控数据接口、判断页面加载状态是加载完成还是加载失败提供了关键信息。Console域可以监听页面JavaScript的日志和错误帮助诊断页面自身的问题。精准执行Input.dispatchMouseEvent和Input.dispatchKeyEvent可以模拟极其精细的鼠标移动、点击、拖拽和键盘事件包括坐标、按键时长等这对于绕过一些基于事件监听的反爬机制或者操作Canvas、WebGL等非DOM元素至关重要。注意直接使用CDP的Input事件需要自己计算坐标或目标元素比WebDriver的“基于元素定位”要复杂但这也带来了灵活性。一个常见的实践是结合使用先用CDP的DOM查询找到元素并获取其边界框再用CDP的Input事件进行精准操作。理解了CDP的价值我们就可以开始搭建地基了。直接裸写CDP通信非常痛苦好在社区已经有了优秀的封装库。3. 工程化起点选择与配置你的CDP客户端库市面上有几个主流的CDP客户端库它们帮你处理了WebSocket连接、消息收发、事件监听等底层细节提供了更友好的API。1. Puppeteer (Node.js)这是Google官方团队维护的项目可以理解为“CDP的高阶封装”。它提供了非常直观的、类似WebDriver的API如page.click(selector)但其底层完全基于CDP。它的优点是开箱即用功能全面生态丰富。缺点是它绑定Node.js环境且为了易用性隐藏了部分CDP细节当你需要极致的定制或使用某些实验性CDP功能时可能需要绕过Puppeteer直接调用CDP会话。2. Playwright (Node.js/Python/.NET/Java)由微软团队开发最初是Puppeteer的fork但现在已自成一体。它支持多浏览器Chromium, Firefox, WebKit其CDP实现主要针对Chromium。Playwright的API设计更现代化强调自动等待和可靠性内置了很多防脆性测试的特性如自动重试、等待元素可操作状态。对于构建健壮的Agent工作流它的这些特性非常有吸引力。3. pyppeteer (Python)可以看作是Puppeteer的Python非官方移植版。早期很有用但随着Playwright Python版的成熟和稳定pyppeteer的维护活跃度下降目前不是新项目的首选。4. 直接使用CDP客户端 (如chrome-remote-interfacefor Node.js,pychromefor Python)这些是更轻量级的库只提供连接和发送CDP命令的基础能力不提供高级API。你需要自己用CDP命令组合出“点击”、“输入”等操作。这提供了最大的灵活性但开发效率最低适合需要深度定制或研究CDP协议本身的场景。我的选择与实践建议对于大多数旨在构建Agent工作流的项目我推荐从Playwright (Python版)开始。原因如下多语言支持团队可以根据技术栈选择核心逻辑可以共享。可靠性设计内置的自动等待、智能定位器如get_by_role、追踪器Trace Viewer对于复杂多变的网页环境非常友好能减少Agent“卡住”或“点错”的情况。强大的上下文隔离Playwright可以轻松创建多个独立的浏览器上下文这对于需要同时登录多个账号、隔离会话状态的Agent场景是刚需。CDP会话直通在需要时你可以通过page.context.new_cdp_session(page)获取底层CDP会话对象直接发送原始CDP命令兼顾了易用性与灵活性。环境搭建实操假设我们使用Playwright for Python。# 1. 安装playwright库 pip install playwright # 2. 安装浏览器驱动Chromium, Firefox, WebKit playwright install chromium# 3. 基础启动脚本 import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 启动浏览器headlessFalse表示显示界面调试时有用 browser await p.chromium.launch(headlessFalse, args[--disable-blink-featuresAutomationControlled]) # 创建一个浏览器上下文类似一个独立的隐身会话 context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... ) # 禁用WebDriver属性降低被检测风险基础规避 await context.add_init_script( Object.defineProperty(navigator, webdriver, { get: () undefined }); ) page await context.new_page() # 此时你已经拥有了一个可通过Playwright API控制的页面对象page # 同时你也可以获取底层的CDP会话进行更精细的操作 cdp_session await context.new_cdp_session(page) # 例如启用Network域监听 await cdp_session.send(Network.enable) # 监听网络请求 cdp_session.on(Network.requestWillBeSent, lambda params: print(fRequest: {params[request][url]})) await page.goto(https://example.com) await page.screenshot(pathexample.png) await browser.close() asyncio.run(main())这个基础脚本已经包含了启动、基础配置、以及如何接入原始CDP会话。接下来我们要思考如何在这个基础上构建一个Agent的“大脑”。4. 设计Agent工作流的核心状态、决策与动作循环一个简单的脚本是线性的打开网页A - 点击按钮B - 输入文字C - 结束。Agent工作流则需要引入“状态”和“决策”的概念。其核心是一个循环感知当前状态 - 根据状态和目标任务决策 - 执行动作 - 等待状态变化/确认 - 进入下一循环。4.1 定义Agent的状态感知器Agent如何“感知”页面它需要从原始HTML中提取出对决策有用的、结构化的“状态信息”。这不仅仅是获取DOM而是信息的抽象。页面级状态URL是否跳转页面标题是什么是否有特定的弹窗如Cookie同意框、登录模态框出现这可以通过监听page.on(framenavigated)和定期检查页面关键元素来实现。数据级状态目标数据是否加载完成是列表的第几页商品价格是多少这通常需要通过CDP执行JavaScript从页面全局变量、特定DOM元素的内容或网络请求的响应中提取。交互元素状态按钮是可点击还是禁用输入框是否有值复选框是否被选中这需要结合DOM属性(disabled,checked)和CSS计算样式(pointer-events,opacity)来判断。我们可以构建一个PageState类来封装这些感知逻辑class PageState: def __init__(self, page, cdp_session): self.page page self.cdp_session cdp_session self._current_url None self._detected_modals [] async def refresh(self): 刷新当前页面的状态感知 self._current_url self.page.url # 感知常见弹窗 self._detected_modals [] modal_selectors [.modal, .dialog, [roledialog], .popup] for selector in modal_selectors: if await self.page.locator(selector).count() 0: self._detected_modals.append(selector) # 可以在这里添加更多感知逻辑如检查特定数据元素 async def get_interactable_elements(self, roleNone, nameNone): 获取当前页面中所有可交互元素或符合特定角色的元素及其状态 # 这是一个简化示例实际中可以利用CDP的Accessibility域或Playwright的get_by_role elements [] # 使用Playwright的locator API找到所有按钮、链接、输入框等 all_buttons self.page.locator(button, a, input, [rolebutton], [tabindex]) count await all_buttons.count() for i in range(count): elem all_buttons.nth(i) is_visible await elem.is_visible() is_enabled not await elem.get_attribute(disabled) # 更精确的判断可以计算样式 elem_info { selector: await elem.evaluate(el el.tagName (el.id ? #${el.id} : (el.className ? .${el.className.split( )[0]} : ))), text: (await elem.text_content() or ).strip()[:50], visible: is_visible, enabled: is_enabled, bounds: await elem.bounding_box() # 获取位置和大小用于后续操作 } # 根据role和name进行过滤这里简化处理 elements.append(elem_info) return elements property def has_blocking_modal(self): 判断是否存在阻塞性弹窗如登录框 blocking_modal_indicators [登录, Sign in, auth, modal-overlay] for modal in self._detected_modals: # 这里可以更智能地判断弹窗内容 return True # 简化返回 return False4.2 构建决策引擎从规则到LLM决策是Agent的“大脑”。根据复杂程度可以分为几个层次1. 基于规则的决策最简单直接。定义一系列“如果-那么”规则。class RuleBasedDecider: async def decide_next_action(self, page_state, goal): if page_state.has_blocking_modal: # 规则1如果有登录弹窗尝试关闭或登录 return Action(typeclose_modal, target.close-button) if 购物车 in goal and 结算 in [e[text] for e in page_state.interactable_elements]: # 规则2如果目标是购物车且页面有结算按钮点击它 return Action(typeclick, targettext结算) # 规则3默认行为尝试点击第一个可用的“下一步”或类似按钮 next_buttons [e for e in page_state.interactable_elements if 下一步 in e[text] or Next in e[text]] if next_buttons: return Action(typeclick, selectornext_buttons[0][selector]) # 如果没有匹配规则返回探索性动作比如滚动或点击可能的内容区域 return Action(typescroll, directiondown)这种方式在流程固定的场景如固定的后台操作流程非常有效但缺乏灵活性。2. 基于LLM的决策这是让Agent真正“智能”起来的关键。我们可以将页面状态简化后的DOM结构、关键元素信息、任务目标构造为提示词Prompt交给大语言模型如GPT-4, Claude, 或本地部署的Llama来生成下一步动作。import openai # 或使用其他LLM SDK class LLMBasedDecider: def __init__(self, api_key, modelgpt-4): self.client openai.OpenAI(api_keyapi_key) self.model model async def decide_next_action(self, page_state, goal): # 1. 构建页面状态的文本描述 state_description f 当前页面URL: {page_state.url} 页面标题: {await page_state.page.title()} 可见的主要交互元素: {chr(10).join([f- [{elem[text]}] (选择器: {elem[selector]}, 是否可用: {elem[enabled]}) for elem in page_state.interactable_elements[:10]])} 任务目标: {goal} # 2. 构造Prompt prompt f 你是一个网页自动化助手。请根据以下页面状态和任务目标决定下一步最合理的单个动作。 只输出一个JSON对象格式必须严格如下{{action: click|input|scroll|wait|goto, target: 元素选择器或URL, value: 仅当动作为input时需要表示输入内容}} {state_description} 请分析当前状态并输出下一步动作JSON # 3. 调用LLM response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, # 低随机性保证决策稳定 ) # 4. 解析LLM返回的JSON import json try: action_dict json.loads(response.choices[0].message.content.strip()) return Action(**action_dict) except json.JSONDecodeError: # 如果LLM输出不符合格式降级到安全动作如等待或滚动 return Action(typewait, duration2)这种方式非常强大能让Agent处理未曾预见的页面布局和流程。但成本高、速度慢且需要精心设计Prompt和结果解析逻辑来保证稳定性。一个折中的方案是混合决策大部分固定流程用规则处理遇到未知状态或规则未覆盖的情况再调用LLM。4.3 动作执行器将决策转化为CDP命令决策引擎输出一个抽象的Action对象如{type: click, target: #submit-btn}动作执行器负责将其翻译成具体的、通过Playwright或CDP执行的操作并处理执行结果和异常。class ActionExecutor: def __init__(self, page, cdp_session): self.page page self.cdp_session cdp_session async def execute(self, action): action_type action.get(type) target action.get(target, ) value action.get(value, ) try: if action_type click: # 使用Playwright的Locator API它内置了智能等待和重试 await self.page.locator(target).click(timeout10000) # 10秒超时 elif action_type input: await self.page.locator(target).fill(value) elif action_type scroll: if target down: await self.page.evaluate(window.scrollBy(0, window.innerHeight * 0.8)) elif target up: await self.page.evaluate(window.scrollBy(0, -window.innerHeight * 0.8)) elif action_type wait: await self.page.wait_for_timeout(int(value) * 1000 if value else 2000) elif action_type goto: await self.page.goto(target) elif action_type cdp_command: # 执行原始CDP命令用于高级或定制化操作 result await self.cdp_session.send(action[command], action.get(params, {})) return result else: raise ValueError(f未知的动作类型: {action_type}) return {success: True} except Exception as e: # 记录详细的错误信息包括截图和DOM快照这对于调试Agent至关重要 error_screenshot_path ferror_{int(time.time())}.png await self.page.screenshot(patherror_screenshot_path, full_pageTrue) # 可以在这里触发状态刷新或者将错误信息反馈给决策引擎进行恢复尝试 return {success: False, error: str(e), screenshot: error_screenshot_path}4.4 组装工作流主循环将状态感知、决策、执行串联起来就形成了Agent的主循环。class BrowserAgent: def __init__(self, page, cdp_session, decider, executor): self.page_state PageState(page, cdp_session) self.decider decider self.executor executor self.goal self.max_steps 100 async def run(self, goal): self.goal goal step 0 while step self.max_steps: step 1 print(f[Step {step}]) # 1. 感知 await self.page_state.refresh() # 2. 决策 next_action await self.decider.decide_next_action(self.page_state, self.goal) if not next_action: print(决策引擎未返回动作任务可能已完成或无法继续。) break print(f 决策: {next_action}) # 3. 执行 result await self.executor.execute(next_action) if not result[success]: print(f 执行失败: {result[error]}) # 这里可以加入错误处理逻辑比如重试、换策略、或终止 # 例如如果是元素未找到可以刷新状态再试一次 await self.page.wait_for_timeout(2000) continue # 4. 等待状态稳定根据动作类型决定 await self.page.wait_for_timeout(1000) # 基础等待 # 可选等待特定网络请求完成或元素出现 # await self.page.wait_for_load_state(networkidle) # 5. 检查目标是否达成这是一个简化检查实际需要根据goal定义达成条件 if await self._is_goal_achieved(): print(任务目标达成) break async def _is_goal_achieved(self): # 根据goal定义判断逻辑例如页面URL包含特定字符、出现了特定元素等 if 订单成功 in await self.page.content(): return True return False5. 实战进阶处理复杂场景与提升Agent鲁棒性一个能在理想环境下运行的Agent是脆弱的。真实网页充满不确定性网络延迟、元素加载缓慢、动态内容、反爬措施、验证码、弹窗广告等。我们必须增强Agent的鲁棒性。5.1 对抗动态加载与元素定位失败这是最常见的问题。页面用了大量JavaScript异步加载元素出现的时间不确定。策略一强化等待策略。不要用固定的sleep而是使用智能等待。# Playwright 内置的等待机制非常好用 await page.locator(button.submit).click() # 内部会自动等待元素可点击 await page.wait_for_selector(.result-item, statevisible, timeout30000) await page.wait_for_function(window.dataLoaded true, timeout15000) # 等待JS变量策略二使用更健壮的选择器。避免使用易变的类名或ID优先使用># 脆弱的 await page.click(.j-confirm-btn) # 更健壮的 await page.get_by_role(button, name确认).click() await page.locator(button).filter(has_text提交订单).click()策略三重试与降级定位。如果首选定位器失败尝试备用方案。async def robust_click(page, selectors): 尝试多个选择器进行点击 for selector in selectors: try: await page.locator(selector).click(timeout5000) return True except Exception as e: print(f选择器 {selector} 点击失败: {e}) continue return False await robust_click(page, [[data-testidcheckout], button:has-text(去支付), .checkout-btn])5.2 绕过反爬与自动化检测现代网站会检测自动化工具。CDP和Playwright虽然比传统WebDriver隐蔽但仍有特征。基础规避启动浏览器时添加参数并注入脚本清除自动化特征。browser await p.chromium.launch( headlessFalse, # 有些网站会检测headless模式调试时可关闭 args[ --disable-blink-featuresAutomationControlled, --disable-dev-shm-usage, --no-sandbox, --disable-web-security, # 谨慎使用可能带来安全问题 --disable-featuresIsolateOrigins,site-per-process, # 有时用于绕过同源策略检测 ] ) await context.add_init_script( // 覆盖 navigator.webdriver, plugins, languages 等属性 Object.defineProperty(navigator, webdriver, { get: () undefined }); Object.defineProperty(navigator, plugins, { get: () [1, 2, 3, 4, 5] }); Object.defineProperty(navigator, languages, { get: () [zh-CN, zh, en] }); // 覆盖 chrome 运行时属性 window.chrome { runtime: {} }; // 修改屏幕分辨率等硬件指纹需谨慎可能不自然 const originalDescriptor Object.getOwnPropertyDescriptor(HTMLCanvasElement.prototype, toDataURL); Object.defineProperty(HTMLCanvasElement.prototype, toDataURL, { apply: function() { const result originalDescriptor.apply.call(this, arguments); // 这里可以加入微小的随机噪声来干扰指纹识别 return result; } }); )模拟真人行为这是更高级的对抗。通过CDP的Input域模拟非匀速的鼠标移动、随机停留、不精确点击。async def human_like_click(page, selector): element page.locator(selector) box await element.bounding_box() if not box: raise Exception(元素未找到或不可见) # 计算点击中心点 x box[x] box[width] / 2 y box[y] box[height] / 2 # 模拟鼠标移动轨迹带随机偏移和延迟 cdp_session await page.context.new_cdp_session(page) await cdp_session.send(Input.dispatchMouseEvent, { type: mouseMoved, x: x random.uniform(-5, 5), y: y random.uniform(-5, 5), button: none, }) await page.wait_for_timeout(random.randint(50, 200)) # 按下 await cdp_session.send(Input.dispatchMouseEvent, { type: mousePressed, x: x, y: y, button: left, clickCount: 1 }) await page.wait_for_timeout(random.randint(30, 150)) # 释放 await cdp_session.send(Input.dispatchMouseEvent, { type: mouseReleased, x: x, y: y, button: left, clickCount: 1 })注意过度模拟反而会产生不自然的行为模式。最好的策略是结合使用并针对具体目标网站进行测试和调整。使用代理IP和用户代理轮换通过browser.new_context每次创建新的上下文时可以指定不同的代理和User-Agent。proxies [http://proxy1:port, http://proxy2:port] # 请使用合法合规的代理服务 user_agents [...] context await browser.new_context( proxy{server: random.choice(proxies)}, user_agentrandom.choice(user_agents) )5.3 集成外部服务处理验证码验证码是自动化的大敌。对于简单图形验证码可以尝试用OCR库如pytesseract但效果通常不佳。对于复杂验证码如点选、滑块可靠的做法是接入第三方打码平台。import requests async def solve_captcha(page): 处理验证码的示例流程 # 1. 定位验证码图片元素并截图 captcha_element page.locator(#captcha-img) captcha_screenshot await captcha_element.screenshot() # 2. 调用打码平台API api_url https://api.dama2.com/xxx # 示例需替换为真实平台 with open(captcha_screenshot, rb) as f: files {image: f} response requests.post(api_url, filesfiles, data{apikey: your_key}) if response.json().get(success): code response.json().get(code) # 3. 输入验证码 await page.locator(#captcha-input).fill(code) await page.locator(#submit-captcha).click() # 4. 检查是否正确 await page.wait_for_timeout(1000) if await page.locator(.captcha-error).is_visible(): # 识别错误可能需要重试或更换策略 return False return True return False5.4 工作流的持久化与状态管理对于长时间运行或需要中断恢复的Agent需要将工作流状态当前URL、已收集的数据、步骤索引等持久化到数据库或文件。import json import pickle class StatefulAgent(BrowserAgent): def __init__(self, state_fileagent_state.json, **kwargs): super().__init__(**kwargs) self.state_file state_file self.load_state() def load_state(self): try: with open(self.state_file, r) as f: state json.load(f) self.current_step state.get(current_step, 0) self.collected_data state.get(collected_data, []) except FileNotFoundError: self.current_step 0 self.collected_data [] async def save_state(self): state { current_url: self.page.url, current_step: self.current_step, collected_data: self.collected_data, goal: self.goal } with open(self.state_file, w) as f: json.dump(state, f) async def run_with_persistence(self, goal): self.goal goal # 可以从保存的步骤继续执行 while self.current_step self.max_steps: # ... 执行单步循环 ... self.current_step 1 await self.save_state() # 每步或每N步保存一次6. 案例拆解构建一个商品价格监控Agent让我们用一个相对完整的例子把上面的组件串联起来。目标是监控某电商网站特定商品的价格变化并在降价时发送通知。目标分解登录网站处理登录态。导航到目标商品页面。提取商品价格和库存状态。判断价格是否低于设定阈值。如果满足条件通过邮件或Webhook发送通知。定期执行如每30分钟一次。核心实现要点状态感知我们需要感知的关键状态是“商品价格”和“购买按钮状态”。这需要通过执行页面JavaScript来提取。async def extract_product_info(page): # 使用Playwright的evaluate在页面上下文中执行JS info await page.evaluate(() { const priceElem document.querySelector(.product-price); const stockElem document.querySelector(.stock-status); const buyButton document.querySelector(#buy-now-btn); return { price: priceElem ? priceElem.innerText.replace(/[^0-9.]/g, ) : null, inStock: stockElem ? !stockElem.innerText.includes(缺货) : false, canBuy: buyButton ? !buyButton.disabled : false }; }) return info决策引擎这个场景规则明确适合用规则引擎。class PriceMonitorDecider: def __init__(self, threshold_price): self.threshold threshold_price async def decide_next_action(self, page_state, goal, product_info): # goal 可能是 monitor_price if not product_info[price]: return {type: refresh} # 价格未加载刷新页面 current_price float(product_info[price]) if current_price self.threshold and product_info[canBuy]: # 达到购买条件触发通知动作 return {type: trigger_alert, price: current_price} else: # 未达条件等待一段时间后重新检查 return {type: wait, duration: 1800} # 等待30分钟动作执行器扩展需要增加trigger_alert动作。async def execute(self, action): if action[type] trigger_alert: # 调用发送邮件或Webhook的函数 await self.send_alert(action[price]) return {success: True} # ... 处理其他动作类型 ...调度与循环使用asyncio或schedule库来定期运行整个Agent流程。注意要妥善管理浏览器实例避免内存泄漏。一个常见的模式是每次监控任务完成后关闭浏览器下次任务再启动。部署与监控可以将这个Agent脚本部署到服务器或云函数上。务必加入日志记录如structlog或loguru记录每一步的状态、决策和异常方便问题排查。对于关键任务可以集成哨兵监控当Agent连续失败多次时发出告警。构建基于CDP的浏览器自动化Agent工作流是一个从“控制浏览器”到“赋予浏览器智能”的升级过程。它不再仅仅是模拟点击而是创建了一个能够感知环境、分析信息、做出决策并执行任务的数字助手。这条路从理解CDP协议开始经过工程化的库选型、核心循环的设计再到应对真实世界复杂性的各种策略每一步都需要在灵活性与稳定性之间权衡。我个人的体会是初期用Playwright这类高阶框架快速搭建原型验证想法在遇到性能瓶颈或需要深度定制时再深入CDP底层去挖掘能力是一个比较平滑的学习和实践路径。最终一个健壮的Agent系统其价值不仅在于自动化了多少流程更在于它能否像一名可靠的员工一样在变化的环境中持续、稳定地完成工作。 基于CDP构建智能浏览器Agent：从自动化脚本到工作流实践
1. 从“手动点点点”到“智能自动化”的跨越如果你还在用Selenium或者Puppeteer写一长串find_element和click来模拟浏览器操作每次页面结构一变就得重新调试脚本那感觉就像在用算盘处理大数据。我最初接触浏览器自动化也是从这个阶段过来的直到后来项目里需要处理大量动态渲染、反爬策略复杂的页面以及实现一些需要“观察-决策-执行”的智能流程时传统的脚本模式就彻底不够用了。这时候一个更底层的协议——Chrome DevTools Protocol也就是CDP进入了我的视野。CDP不是一个新的工具而是Chrome浏览器内置的调试协议。它像是一把能直接与浏览器内核对话的“手术刀”让你能控制标签页、拦截网络请求、执行JavaScript、获取DOM快照甚至监听内存变化。基于CDP构建自动化意味着你跳过了WebDriver这类“翻译官”的中间层直接与浏览器引擎交互响应更快能力也更强大。但CDP本身是协议不是框架直接用它写代码就像用汇编语言编程强大但繁琐。所以我们今天的重点不是罗列CDP的API而是如何以它为基础构建一个更高级的抽象Agent工作流。所谓Agent工作流你可以把它想象成一个在浏览器里“上班”的虚拟员工。它不再是被动执行预设步骤的脚本而是一个具备一定感知、决策和执行能力的智能体。它知道自己要完成什么任务比如“监控商品价格变化并截图报警”能通过CDP“看到”页面内容能根据页面状态“思考”下一步该点哪里、输入什么并能处理一些意外情况比如弹窗、验证码。这种模式正是当前从“自动化脚本”向“智能体应用”演进的核心实践。2. 为什么是CDP超越WebDriver的底层能力剖析在深入构建之前我们必须先搞清楚为什么选择CDP作为基石而不是继续用更成熟的Selenium WebDriver。这不仅仅是性能问题更是能力边界和设计哲学的不同。2.1 协议层 vs 驱动层本质差异WebDriver是一个W3C标准它定义了一套跨浏览器的、用于控制网页的通用指令集比如“点击元素”、“获取文本”。它的工作模式是你的脚本通过特定语言的客户端库如Python的selenium包发送指令给一个独立的驱动程序如chromedriver这个驱动再通过私有协议与真实的浏览器通信。这带来几个问题一是多了一层转换速度和效率有损耗二是驱动和浏览器版本必须严格匹配否则经常出兼容性问题三是能力受限于WebDriver标准一些浏览器独有的、高级的调试功能无法使用。CDP则完全不同。它是Chrome/Chromium内核原生暴露的调试接口基于WebSocket通信。你的代码直接连接到一个Chrome实例的调试端口发送和接收的都是JSON-RPC格式的消息。这意味着无中间商延迟更低指令直达浏览器引擎执行和获取结果的延迟显著降低对于需要高频交互的自动化场景至关重要。能力全集你能用到Chrome开发者工具里几乎所有的功能包括但不限于网络控制精确拦截、修改、放行任意请求和响应模拟弱网环境。性能分析获取时间线追踪、内存堆快照、计算性能指标。DOM深度操作不仅获取元素还能监听DOM变化获取CSS计算样式甚至模拟用户输入如输入法组合键。运行时注入在页面上下文中执行任意JavaScript并获取复杂的返回值如函数、Promise。页面快照生成包含完整CSSOM的截图甚至能生成PDF。版本兼容性更好虽然CDP本身也在演进但它的向后兼容性通常比chromedriver的版本锁死要好处理得多核心API相对稳定。2.2 构建Agent工作流的核心优势对于Agent工作流而言CDP提供的这些底层能力是构建“智能”的感官系统。感知能力Agent需要“看”懂页面。通过CDP的DOMSnapshot.captureSnapshot或Runtime.evaluateAgent可以获取结构化、带布局信息的DOM树这比通过innerHTML获取的字符串要强大得多。结合Accessibility.getFullAXTree还能获取无障碍树理解元素的语义角色这对于理解复杂UI组件如下拉菜单、滑块的状态非常有帮助。决策依据Agent的决策需要数据。CDP的Network域可以让Agent监听所有XHR/Fetch请求和响应这对于监控数据接口、判断页面加载状态是加载完成还是加载失败提供了关键信息。Console域可以监听页面JavaScript的日志和错误帮助诊断页面自身的问题。精准执行Input.dispatchMouseEvent和Input.dispatchKeyEvent可以模拟极其精细的鼠标移动、点击、拖拽和键盘事件包括坐标、按键时长等这对于绕过一些基于事件监听的反爬机制或者操作Canvas、WebGL等非DOM元素至关重要。注意直接使用CDP的Input事件需要自己计算坐标或目标元素比WebDriver的“基于元素定位”要复杂但这也带来了灵活性。一个常见的实践是结合使用先用CDP的DOM查询找到元素并获取其边界框再用CDP的Input事件进行精准操作。理解了CDP的价值我们就可以开始搭建地基了。直接裸写CDP通信非常痛苦好在社区已经有了优秀的封装库。3. 工程化起点选择与配置你的CDP客户端库市面上有几个主流的CDP客户端库它们帮你处理了WebSocket连接、消息收发、事件监听等底层细节提供了更友好的API。1. Puppeteer (Node.js)这是Google官方团队维护的项目可以理解为“CDP的高阶封装”。它提供了非常直观的、类似WebDriver的API如page.click(selector)但其底层完全基于CDP。它的优点是开箱即用功能全面生态丰富。缺点是它绑定Node.js环境且为了易用性隐藏了部分CDP细节当你需要极致的定制或使用某些实验性CDP功能时可能需要绕过Puppeteer直接调用CDP会话。2. Playwright (Node.js/Python/.NET/Java)由微软团队开发最初是Puppeteer的fork但现在已自成一体。它支持多浏览器Chromium, Firefox, WebKit其CDP实现主要针对Chromium。Playwright的API设计更现代化强调自动等待和可靠性内置了很多防脆性测试的特性如自动重试、等待元素可操作状态。对于构建健壮的Agent工作流它的这些特性非常有吸引力。3. pyppeteer (Python)可以看作是Puppeteer的Python非官方移植版。早期很有用但随着Playwright Python版的成熟和稳定pyppeteer的维护活跃度下降目前不是新项目的首选。4. 直接使用CDP客户端 (如chrome-remote-interfacefor Node.js,pychromefor Python)这些是更轻量级的库只提供连接和发送CDP命令的基础能力不提供高级API。你需要自己用CDP命令组合出“点击”、“输入”等操作。这提供了最大的灵活性但开发效率最低适合需要深度定制或研究CDP协议本身的场景。我的选择与实践建议对于大多数旨在构建Agent工作流的项目我推荐从Playwright (Python版)开始。原因如下多语言支持团队可以根据技术栈选择核心逻辑可以共享。可靠性设计内置的自动等待、智能定位器如get_by_role、追踪器Trace Viewer对于复杂多变的网页环境非常友好能减少Agent“卡住”或“点错”的情况。强大的上下文隔离Playwright可以轻松创建多个独立的浏览器上下文这对于需要同时登录多个账号、隔离会话状态的Agent场景是刚需。CDP会话直通在需要时你可以通过page.context.new_cdp_session(page)获取底层CDP会话对象直接发送原始CDP命令兼顾了易用性与灵活性。环境搭建实操假设我们使用Playwright for Python。# 1. 安装playwright库 pip install playwright # 2. 安装浏览器驱动Chromium, Firefox, WebKit playwright install chromium# 3. 基础启动脚本 import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 启动浏览器headlessFalse表示显示界面调试时有用 browser await p.chromium.launch(headlessFalse, args[--disable-blink-featuresAutomationControlled]) # 创建一个浏览器上下文类似一个独立的隐身会话 context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... ) # 禁用WebDriver属性降低被检测风险基础规避 await context.add_init_script( Object.defineProperty(navigator, webdriver, { get: () undefined }); ) page await context.new_page() # 此时你已经拥有了一个可通过Playwright API控制的页面对象page # 同时你也可以获取底层的CDP会话进行更精细的操作 cdp_session await context.new_cdp_session(page) # 例如启用Network域监听 await cdp_session.send(Network.enable) # 监听网络请求 cdp_session.on(Network.requestWillBeSent, lambda params: print(fRequest: {params[request][url]})) await page.goto(https://example.com) await page.screenshot(pathexample.png) await browser.close() asyncio.run(main())这个基础脚本已经包含了启动、基础配置、以及如何接入原始CDP会话。接下来我们要思考如何在这个基础上构建一个Agent的“大脑”。4. 设计Agent工作流的核心状态、决策与动作循环一个简单的脚本是线性的打开网页A - 点击按钮B - 输入文字C - 结束。Agent工作流则需要引入“状态”和“决策”的概念。其核心是一个循环感知当前状态 - 根据状态和目标任务决策 - 执行动作 - 等待状态变化/确认 - 进入下一循环。4.1 定义Agent的状态感知器Agent如何“感知”页面它需要从原始HTML中提取出对决策有用的、结构化的“状态信息”。这不仅仅是获取DOM而是信息的抽象。页面级状态URL是否跳转页面标题是什么是否有特定的弹窗如Cookie同意框、登录模态框出现这可以通过监听page.on(framenavigated)和定期检查页面关键元素来实现。数据级状态目标数据是否加载完成是列表的第几页商品价格是多少这通常需要通过CDP执行JavaScript从页面全局变量、特定DOM元素的内容或网络请求的响应中提取。交互元素状态按钮是可点击还是禁用输入框是否有值复选框是否被选中这需要结合DOM属性(disabled,checked)和CSS计算样式(pointer-events,opacity)来判断。我们可以构建一个PageState类来封装这些感知逻辑class PageState: def __init__(self, page, cdp_session): self.page page self.cdp_session cdp_session self._current_url None self._detected_modals [] async def refresh(self): 刷新当前页面的状态感知 self._current_url self.page.url # 感知常见弹窗 self._detected_modals [] modal_selectors [.modal, .dialog, [roledialog], .popup] for selector in modal_selectors: if await self.page.locator(selector).count() 0: self._detected_modals.append(selector) # 可以在这里添加更多感知逻辑如检查特定数据元素 async def get_interactable_elements(self, roleNone, nameNone): 获取当前页面中所有可交互元素或符合特定角色的元素及其状态 # 这是一个简化示例实际中可以利用CDP的Accessibility域或Playwright的get_by_role elements [] # 使用Playwright的locator API找到所有按钮、链接、输入框等 all_buttons self.page.locator(button, a, input, [rolebutton], [tabindex]) count await all_buttons.count() for i in range(count): elem all_buttons.nth(i) is_visible await elem.is_visible() is_enabled not await elem.get_attribute(disabled) # 更精确的判断可以计算样式 elem_info { selector: await elem.evaluate(el el.tagName (el.id ? #${el.id} : (el.className ? .${el.className.split( )[0]} : ))), text: (await elem.text_content() or ).strip()[:50], visible: is_visible, enabled: is_enabled, bounds: await elem.bounding_box() # 获取位置和大小用于后续操作 } # 根据role和name进行过滤这里简化处理 elements.append(elem_info) return elements property def has_blocking_modal(self): 判断是否存在阻塞性弹窗如登录框 blocking_modal_indicators [登录, Sign in, auth, modal-overlay] for modal in self._detected_modals: # 这里可以更智能地判断弹窗内容 return True # 简化返回 return False4.2 构建决策引擎从规则到LLM决策是Agent的“大脑”。根据复杂程度可以分为几个层次1. 基于规则的决策最简单直接。定义一系列“如果-那么”规则。class RuleBasedDecider: async def decide_next_action(self, page_state, goal): if page_state.has_blocking_modal: # 规则1如果有登录弹窗尝试关闭或登录 return Action(typeclose_modal, target.close-button) if 购物车 in goal and 结算 in [e[text] for e in page_state.interactable_elements]: # 规则2如果目标是购物车且页面有结算按钮点击它 return Action(typeclick, targettext结算) # 规则3默认行为尝试点击第一个可用的“下一步”或类似按钮 next_buttons [e for e in page_state.interactable_elements if 下一步 in e[text] or Next in e[text]] if next_buttons: return Action(typeclick, selectornext_buttons[0][selector]) # 如果没有匹配规则返回探索性动作比如滚动或点击可能的内容区域 return Action(typescroll, directiondown)这种方式在流程固定的场景如固定的后台操作流程非常有效但缺乏灵活性。2. 基于LLM的决策这是让Agent真正“智能”起来的关键。我们可以将页面状态简化后的DOM结构、关键元素信息、任务目标构造为提示词Prompt交给大语言模型如GPT-4, Claude, 或本地部署的Llama来生成下一步动作。import openai # 或使用其他LLM SDK class LLMBasedDecider: def __init__(self, api_key, modelgpt-4): self.client openai.OpenAI(api_keyapi_key) self.model model async def decide_next_action(self, page_state, goal): # 1. 构建页面状态的文本描述 state_description f 当前页面URL: {page_state.url} 页面标题: {await page_state.page.title()} 可见的主要交互元素: {chr(10).join([f- [{elem[text]}] (选择器: {elem[selector]}, 是否可用: {elem[enabled]}) for elem in page_state.interactable_elements[:10]])} 任务目标: {goal} # 2. 构造Prompt prompt f 你是一个网页自动化助手。请根据以下页面状态和任务目标决定下一步最合理的单个动作。 只输出一个JSON对象格式必须严格如下{{action: click|input|scroll|wait|goto, target: 元素选择器或URL, value: 仅当动作为input时需要表示输入内容}} {state_description} 请分析当前状态并输出下一步动作JSON # 3. 调用LLM response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, # 低随机性保证决策稳定 ) # 4. 解析LLM返回的JSON import json try: action_dict json.loads(response.choices[0].message.content.strip()) return Action(**action_dict) except json.JSONDecodeError: # 如果LLM输出不符合格式降级到安全动作如等待或滚动 return Action(typewait, duration2)这种方式非常强大能让Agent处理未曾预见的页面布局和流程。但成本高、速度慢且需要精心设计Prompt和结果解析逻辑来保证稳定性。一个折中的方案是混合决策大部分固定流程用规则处理遇到未知状态或规则未覆盖的情况再调用LLM。4.3 动作执行器将决策转化为CDP命令决策引擎输出一个抽象的Action对象如{type: click, target: #submit-btn}动作执行器负责将其翻译成具体的、通过Playwright或CDP执行的操作并处理执行结果和异常。class ActionExecutor: def __init__(self, page, cdp_session): self.page page self.cdp_session cdp_session async def execute(self, action): action_type action.get(type) target action.get(target, ) value action.get(value, ) try: if action_type click: # 使用Playwright的Locator API它内置了智能等待和重试 await self.page.locator(target).click(timeout10000) # 10秒超时 elif action_type input: await self.page.locator(target).fill(value) elif action_type scroll: if target down: await self.page.evaluate(window.scrollBy(0, window.innerHeight * 0.8)) elif target up: await self.page.evaluate(window.scrollBy(0, -window.innerHeight * 0.8)) elif action_type wait: await self.page.wait_for_timeout(int(value) * 1000 if value else 2000) elif action_type goto: await self.page.goto(target) elif action_type cdp_command: # 执行原始CDP命令用于高级或定制化操作 result await self.cdp_session.send(action[command], action.get(params, {})) return result else: raise ValueError(f未知的动作类型: {action_type}) return {success: True} except Exception as e: # 记录详细的错误信息包括截图和DOM快照这对于调试Agent至关重要 error_screenshot_path ferror_{int(time.time())}.png await self.page.screenshot(patherror_screenshot_path, full_pageTrue) # 可以在这里触发状态刷新或者将错误信息反馈给决策引擎进行恢复尝试 return {success: False, error: str(e), screenshot: error_screenshot_path}4.4 组装工作流主循环将状态感知、决策、执行串联起来就形成了Agent的主循环。class BrowserAgent: def __init__(self, page, cdp_session, decider, executor): self.page_state PageState(page, cdp_session) self.decider decider self.executor executor self.goal self.max_steps 100 async def run(self, goal): self.goal goal step 0 while step self.max_steps: step 1 print(f[Step {step}]) # 1. 感知 await self.page_state.refresh() # 2. 决策 next_action await self.decider.decide_next_action(self.page_state, self.goal) if not next_action: print(决策引擎未返回动作任务可能已完成或无法继续。) break print(f 决策: {next_action}) # 3. 执行 result await self.executor.execute(next_action) if not result[success]: print(f 执行失败: {result[error]}) # 这里可以加入错误处理逻辑比如重试、换策略、或终止 # 例如如果是元素未找到可以刷新状态再试一次 await self.page.wait_for_timeout(2000) continue # 4. 等待状态稳定根据动作类型决定 await self.page.wait_for_timeout(1000) # 基础等待 # 可选等待特定网络请求完成或元素出现 # await self.page.wait_for_load_state(networkidle) # 5. 检查目标是否达成这是一个简化检查实际需要根据goal定义达成条件 if await self._is_goal_achieved(): print(任务目标达成) break async def _is_goal_achieved(self): # 根据goal定义判断逻辑例如页面URL包含特定字符、出现了特定元素等 if 订单成功 in await self.page.content(): return True return False5. 实战进阶处理复杂场景与提升Agent鲁棒性一个能在理想环境下运行的Agent是脆弱的。真实网页充满不确定性网络延迟、元素加载缓慢、动态内容、反爬措施、验证码、弹窗广告等。我们必须增强Agent的鲁棒性。5.1 对抗动态加载与元素定位失败这是最常见的问题。页面用了大量JavaScript异步加载元素出现的时间不确定。策略一强化等待策略。不要用固定的sleep而是使用智能等待。# Playwright 内置的等待机制非常好用 await page.locator(button.submit).click() # 内部会自动等待元素可点击 await page.wait_for_selector(.result-item, statevisible, timeout30000) await page.wait_for_function(window.dataLoaded true, timeout15000) # 等待JS变量策略二使用更健壮的选择器。避免使用易变的类名或ID优先使用># 脆弱的 await page.click(.j-confirm-btn) # 更健壮的 await page.get_by_role(button, name确认).click() await page.locator(button).filter(has_text提交订单).click()策略三重试与降级定位。如果首选定位器失败尝试备用方案。async def robust_click(page, selectors): 尝试多个选择器进行点击 for selector in selectors: try: await page.locator(selector).click(timeout5000) return True except Exception as e: print(f选择器 {selector} 点击失败: {e}) continue return False await robust_click(page, [[data-testidcheckout], button:has-text(去支付), .checkout-btn])5.2 绕过反爬与自动化检测现代网站会检测自动化工具。CDP和Playwright虽然比传统WebDriver隐蔽但仍有特征。基础规避启动浏览器时添加参数并注入脚本清除自动化特征。browser await p.chromium.launch( headlessFalse, # 有些网站会检测headless模式调试时可关闭 args[ --disable-blink-featuresAutomationControlled, --disable-dev-shm-usage, --no-sandbox, --disable-web-security, # 谨慎使用可能带来安全问题 --disable-featuresIsolateOrigins,site-per-process, # 有时用于绕过同源策略检测 ] ) await context.add_init_script( // 覆盖 navigator.webdriver, plugins, languages 等属性 Object.defineProperty(navigator, webdriver, { get: () undefined }); Object.defineProperty(navigator, plugins, { get: () [1, 2, 3, 4, 5] }); Object.defineProperty(navigator, languages, { get: () [zh-CN, zh, en] }); // 覆盖 chrome 运行时属性 window.chrome { runtime: {} }; // 修改屏幕分辨率等硬件指纹需谨慎可能不自然 const originalDescriptor Object.getOwnPropertyDescriptor(HTMLCanvasElement.prototype, toDataURL); Object.defineProperty(HTMLCanvasElement.prototype, toDataURL, { apply: function() { const result originalDescriptor.apply.call(this, arguments); // 这里可以加入微小的随机噪声来干扰指纹识别 return result; } }); )模拟真人行为这是更高级的对抗。通过CDP的Input域模拟非匀速的鼠标移动、随机停留、不精确点击。async def human_like_click(page, selector): element page.locator(selector) box await element.bounding_box() if not box: raise Exception(元素未找到或不可见) # 计算点击中心点 x box[x] box[width] / 2 y box[y] box[height] / 2 # 模拟鼠标移动轨迹带随机偏移和延迟 cdp_session await page.context.new_cdp_session(page) await cdp_session.send(Input.dispatchMouseEvent, { type: mouseMoved, x: x random.uniform(-5, 5), y: y random.uniform(-5, 5), button: none, }) await page.wait_for_timeout(random.randint(50, 200)) # 按下 await cdp_session.send(Input.dispatchMouseEvent, { type: mousePressed, x: x, y: y, button: left, clickCount: 1 }) await page.wait_for_timeout(random.randint(30, 150)) # 释放 await cdp_session.send(Input.dispatchMouseEvent, { type: mouseReleased, x: x, y: y, button: left, clickCount: 1 })注意过度模拟反而会产生不自然的行为模式。最好的策略是结合使用并针对具体目标网站进行测试和调整。使用代理IP和用户代理轮换通过browser.new_context每次创建新的上下文时可以指定不同的代理和User-Agent。proxies [http://proxy1:port, http://proxy2:port] # 请使用合法合规的代理服务 user_agents [...] context await browser.new_context( proxy{server: random.choice(proxies)}, user_agentrandom.choice(user_agents) )5.3 集成外部服务处理验证码验证码是自动化的大敌。对于简单图形验证码可以尝试用OCR库如pytesseract但效果通常不佳。对于复杂验证码如点选、滑块可靠的做法是接入第三方打码平台。import requests async def solve_captcha(page): 处理验证码的示例流程 # 1. 定位验证码图片元素并截图 captcha_element page.locator(#captcha-img) captcha_screenshot await captcha_element.screenshot() # 2. 调用打码平台API api_url https://api.dama2.com/xxx # 示例需替换为真实平台 with open(captcha_screenshot, rb) as f: files {image: f} response requests.post(api_url, filesfiles, data{apikey: your_key}) if response.json().get(success): code response.json().get(code) # 3. 输入验证码 await page.locator(#captcha-input).fill(code) await page.locator(#submit-captcha).click() # 4. 检查是否正确 await page.wait_for_timeout(1000) if await page.locator(.captcha-error).is_visible(): # 识别错误可能需要重试或更换策略 return False return True return False5.4 工作流的持久化与状态管理对于长时间运行或需要中断恢复的Agent需要将工作流状态当前URL、已收集的数据、步骤索引等持久化到数据库或文件。import json import pickle class StatefulAgent(BrowserAgent): def __init__(self, state_fileagent_state.json, **kwargs): super().__init__(**kwargs) self.state_file state_file self.load_state() def load_state(self): try: with open(self.state_file, r) as f: state json.load(f) self.current_step state.get(current_step, 0) self.collected_data state.get(collected_data, []) except FileNotFoundError: self.current_step 0 self.collected_data [] async def save_state(self): state { current_url: self.page.url, current_step: self.current_step, collected_data: self.collected_data, goal: self.goal } with open(self.state_file, w) as f: json.dump(state, f) async def run_with_persistence(self, goal): self.goal goal # 可以从保存的步骤继续执行 while self.current_step self.max_steps: # ... 执行单步循环 ... self.current_step 1 await self.save_state() # 每步或每N步保存一次6. 案例拆解构建一个商品价格监控Agent让我们用一个相对完整的例子把上面的组件串联起来。目标是监控某电商网站特定商品的价格变化并在降价时发送通知。目标分解登录网站处理登录态。导航到目标商品页面。提取商品价格和库存状态。判断价格是否低于设定阈值。如果满足条件通过邮件或Webhook发送通知。定期执行如每30分钟一次。核心实现要点状态感知我们需要感知的关键状态是“商品价格”和“购买按钮状态”。这需要通过执行页面JavaScript来提取。async def extract_product_info(page): # 使用Playwright的evaluate在页面上下文中执行JS info await page.evaluate(() { const priceElem document.querySelector(.product-price); const stockElem document.querySelector(.stock-status); const buyButton document.querySelector(#buy-now-btn); return { price: priceElem ? priceElem.innerText.replace(/[^0-9.]/g, ) : null, inStock: stockElem ? !stockElem.innerText.includes(缺货) : false, canBuy: buyButton ? !buyButton.disabled : false }; }) return info决策引擎这个场景规则明确适合用规则引擎。class PriceMonitorDecider: def __init__(self, threshold_price): self.threshold threshold_price async def decide_next_action(self, page_state, goal, product_info): # goal 可能是 monitor_price if not product_info[price]: return {type: refresh} # 价格未加载刷新页面 current_price float(product_info[price]) if current_price self.threshold and product_info[canBuy]: # 达到购买条件触发通知动作 return {type: trigger_alert, price: current_price} else: # 未达条件等待一段时间后重新检查 return {type: wait, duration: 1800} # 等待30分钟动作执行器扩展需要增加trigger_alert动作。async def execute(self, action): if action[type] trigger_alert: # 调用发送邮件或Webhook的函数 await self.send_alert(action[price]) return {success: True} # ... 处理其他动作类型 ...调度与循环使用asyncio或schedule库来定期运行整个Agent流程。注意要妥善管理浏览器实例避免内存泄漏。一个常见的模式是每次监控任务完成后关闭浏览器下次任务再启动。部署与监控可以将这个Agent脚本部署到服务器或云函数上。务必加入日志记录如structlog或loguru记录每一步的状态、决策和异常方便问题排查。对于关键任务可以集成哨兵监控当Agent连续失败多次时发出告警。构建基于CDP的浏览器自动化Agent工作流是一个从“控制浏览器”到“赋予浏览器智能”的升级过程。它不再仅仅是模拟点击而是创建了一个能够感知环境、分析信息、做出决策并执行任务的数字助手。这条路从理解CDP协议开始经过工程化的库选型、核心循环的设计再到应对真实世界复杂性的各种策略每一步都需要在灵活性与稳定性之间权衡。我个人的体会是初期用Playwright这类高阶框架快速搭建原型验证想法在遇到性能瓶颈或需要深度定制时再深入CDP底层去挖掘能力是一个比较平滑的学习和实践路径。最终一个健壮的Agent系统其价值不仅在于自动化了多少流程更在于它能否像一名可靠的员工一样在变化的环境中持续、稳定地完成工作。