建智能瀏覽器Agent：從自動化腳本到工作流實踐)
1. 從“手動點點點”到“智能自動化”的跨越如果你還在用Selenium或者Puppeteer寫一長串find_element和click來模擬瀏覽器操作每次頁面結(jié)構(gòu)一變就得重新調(diào)試腳本那感覺就像在用算盤處理大數(shù)據(jù)。我最初接觸瀏覽器自動化也是從這個階段過來的直到后來項目里需要處理大量動態(tài)渲染、反爬策略復(fù)雜的頁面以及實現(xiàn)一些需要“觀察-決策-執(zhí)行”的智能流程時傳統(tǒng)的腳本模式就徹底不夠用了。這時候一個更底層的協(xié)議——Chrome DevTools Protocol也就是CDP進入了我的視野。CDP不是一個新的工具而是Chrome瀏覽器內(nèi)置的調(diào)試協(xié)議。它像是一把能直接與瀏覽器內(nèi)核對話的“手術(shù)刀”讓你能控制標簽頁、攔截網(wǎng)絡(luò)請求、執(zhí)行JavaScript、獲取DOM快照甚至監(jiān)聽內(nèi)存變化。基于CDP構(gòu)建自動化意味著你跳過了WebDriver這類“翻譯官”的中間層直接與瀏覽器引擎交互響應(yīng)更快能力也更強大。但CDP本身是協(xié)議不是框架直接用它寫代碼就像用匯編語言編程強大但繁瑣。所以我們今天的重點不是羅列CDP的API而是如何以它為基礎(chǔ)構(gòu)建一個更高級的抽象Agent工作流。所謂Agent工作流你可以把它想象成一個在瀏覽器里“上班”的虛擬員工。它不再是被動執(zhí)行預(yù)設(shè)步驟的腳本而是一個具備一定感知、決策和執(zhí)行能力的智能體。它知道自己要完成什么任務(wù)比如“監(jiān)控商品價格變化并截圖報警”能通過CDP“看到”頁面內(nèi)容能根據(jù)頁面狀態(tài)“思考”下一步該點哪里、輸入什么并能處理一些意外情況比如彈窗、驗證碼。這種模式正是當前從“自動化腳本”向“智能體應(yīng)用”演進的核心實踐。2. 為什么是CDP超越WebDriver的底層能力剖析在深入構(gòu)建之前我們必須先搞清楚為什么選擇CDP作為基石而不是繼續(xù)用更成熟的Selenium WebDriver。這不僅僅是性能問題更是能力邊界和設(shè)計哲學(xué)的不同。2.1 協(xié)議層 vs 驅(qū)動層本質(zhì)差異WebDriver是一個W3C標準它定義了一套跨瀏覽器的、用于控制網(wǎng)頁的通用指令集比如“點擊元素”、“獲取文本”。它的工作模式是你的腳本通過特定語言的客戶端庫如Python的selenium包發(fā)送指令給一個獨立的驅(qū)動程序如chromedriver這個驅(qū)動再通過私有協(xié)議與真實的瀏覽器通信。這帶來幾個問題一是多了一層轉(zhuǎn)換速度和效率有損耗二是驅(qū)動和瀏覽器版本必須嚴格匹配否則經(jīng)常出兼容性問題三是能力受限于WebDriver標準一些瀏覽器獨有的、高級的調(diào)試功能無法使用。CDP則完全不同。它是Chrome/Chromium內(nèi)核原生暴露的調(diào)試接口基于WebSocket通信。你的代碼直接連接到一個Chrome實例的調(diào)試端口發(fā)送和接收的都是JSON-RPC格式的消息。這意味著無中間商延遲更低指令直達瀏覽器引擎執(zhí)行和獲取結(jié)果的延遲顯著降低對于需要高頻交互的自動化場景至關(guān)重要。能力全集你能用到Chrome開發(fā)者工具里幾乎所有的功能包括但不限于網(wǎng)絡(luò)控制精確攔截、修改、放行任意請求和響應(yīng)模擬弱網(wǎng)環(huán)境。性能分析獲取時間線追蹤、內(nèi)存堆快照、計算性能指標。DOM深度操作不僅獲取元素還能監(jiān)聽DOM變化獲取CSS計算樣式甚至模擬用戶輸入如輸入法組合鍵。運行時注入在頁面上下文中執(zhí)行任意JavaScript并獲取復(fù)雜的返回值如函數(shù)、Promise。頁面快照生成包含完整CSSOM的截圖甚至能生成PDF。版本兼容性更好雖然CDP本身也在演進但它的向后兼容性通常比chromedriver的版本鎖死要好處理得多核心API相對穩(wěn)定。2.2 構(gòu)建Agent工作流的核心優(yōu)勢對于Agent工作流而言CDP提供的這些底層能力是構(gòu)建“智能”的感官系統(tǒng)。感知能力Agent需要“看”懂頁面。通過CDP的DOMSnapshot.captureSnapshot或Runtime.evaluateAgent可以獲取結(jié)構(gòu)化、帶布局信息的DOM樹這比通過innerHTML獲取的字符串要強大得多。結(jié)合Accessibility.getFullAXTree還能獲取無障礙樹理解元素的語義角色這對于理解復(fù)雜UI組件如下拉菜單、滑塊的狀態(tài)非常有幫助。決策依據(jù)Agent的決策需要數(shù)據(jù)。CDP的Network域可以讓Agent監(jiān)聽所有XHR/Fetch請求和響應(yīng)這對于監(jiān)控數(shù)據(jù)接口、判斷頁面加載狀態(tài)是加載完成還是加載失敗提供了關(guān)鍵信息。Console域可以監(jiān)聽頁面JavaScript的日志和錯誤幫助診斷頁面自身的問題。精準執(zhí)行Input.dispatchMouseEvent和Input.dispatchKeyEvent可以模擬極其精細的鼠標移動、點擊、拖拽和鍵盤事件包括坐標、按鍵時長等這對于繞過一些基于事件監(jiān)聽的反爬機制或者操作Canvas、WebGL等非DOM元素至關(guān)重要。注意直接使用CDP的Input事件需要自己計算坐標或目標元素比WebDriver的“基于元素定位”要復(fù)雜但這也帶來了靈活性。一個常見的實踐是結(jié)合使用先用CDP的DOM查詢找到元素并獲取其邊界框再用CDP的Input事件進行精準操作。理解了CDP的價值我們就可以開始搭建地基了。直接裸寫CDP通信非常痛苦好在社區(qū)已經(jīng)有了優(yōu)秀的封裝庫。3. 工程化起點選擇與配置你的CDP客戶端庫市面上有幾個主流的CDP客戶端庫它們幫你處理了WebSocket連接、消息收發(fā)、事件監(jiān)聽等底層細節(jié)提供了更友好的API。1. Puppeteer (Node.js)這是Google官方團隊維護的項目可以理解為“CDP的高階封裝”。它提供了非常直觀的、類似WebDriver的API如page.click(selector)但其底層完全基于CDP。它的優(yōu)點是開箱即用功能全面生態(tài)豐富。缺點是它綁定Node.js環(huán)境且為了易用性隱藏了部分CDP細節(jié)當你需要極致的定制或使用某些實驗性CDP功能時可能需要繞過Puppeteer直接調(diào)用CDP會話。2. Playwright (Node.js/Python/.NET/Java)由微軟團隊開發(fā)最初是Puppeteer的fork但現(xiàn)在已自成一體。它支持多瀏覽器Chromium, Firefox, WebKit其CDP實現(xiàn)主要針對Chromium。Playwright的API設(shè)計更現(xiàn)代化強調(diào)自動等待和可靠性內(nèi)置了很多防脆性測試的特性如自動重試、等待元素可操作狀態(tài)。對于構(gòu)建健壯的Agent工作流它的這些特性非常有吸引力。3. pyppeteer (Python)可以看作是Puppeteer的Python非官方移植版。早期很有用但隨著Playwright Python版的成熟和穩(wěn)定pyppeteer的維護活躍度下降目前不是新項目的首選。4. 直接使用CDP客戶端 (如chrome-remote-interfacefor Node.js,pychromefor Python)這些是更輕量級的庫只提供連接和發(fā)送CDP命令的基礎(chǔ)能力不提供高級API。你需要自己用CDP命令組合出“點擊”、“輸入”等操作。這提供了最大的靈活性但開發(fā)效率最低適合需要深度定制或研究CDP協(xié)議本身的場景。我的選擇與實踐建議對于大多數(shù)旨在構(gòu)建Agent工作流的項目我推薦從Playwright (Python版)開始。原因如下多語言支持團隊可以根據(jù)技術(shù)棧選擇核心邏輯可以共享。可靠性設(shè)計內(nèi)置的自動等待、智能定位器如get_by_role、追蹤器Trace Viewer對于復(fù)雜多變的網(wǎng)頁環(huán)境非常友好能減少Agent“卡住”或“點錯”的情況。強大的上下文隔離Playwright可以輕松創(chuàng)建多個獨立的瀏覽器上下文這對于需要同時登錄多個賬號、隔離會話狀態(tài)的Agent場景是剛需。CDP會話直通在需要時你可以通過page.context.new_cdp_session(page)獲取底層CDP會話對象直接發(fā)送原始CDP命令兼顧了易用性與靈活性。環(huán)境搭建實操假設(shè)我們使用Playwright for Python。# 1. 安裝playwright庫 pip install playwright # 2. 安裝瀏覽器驅(qū)動Chromium, Firefox, WebKit playwright install chromium# 3. 基礎(chǔ)啟動腳本 import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 啟動瀏覽器headlessFalse表示顯示界面調(diào)試時有用 browser await p.chromium.launch(headlessFalse, args[--disable-blink-featuresAutomationControlled]) # 創(chuàng)建一個瀏覽器上下文類似一個獨立的隱身會話 context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... ) # 禁用WebDriver屬性降低被檢測風險基礎(chǔ)規(guī)避 await context.add_init_script( Object.defineProperty(navigator, webdriver, { get: () undefined }); ) page await context.new_page() # 此時你已經(jīng)擁有了一個可通過Playwright API控制的頁面對象page # 同時你也可以獲取底層的CDP會話進行更精細的操作 cdp_session await context.new_cdp_session(page) # 例如啟用Network域監(jiān)聽 await cdp_session.send(Network.enable) # 監(jiān)聽網(wǎng)絡(luò)請求 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())這個基礎(chǔ)腳本已經(jīng)包含了啟動、基礎(chǔ)配置、以及如何接入原始CDP會話。接下來我們要思考如何在這個基礎(chǔ)上構(gòu)建一個Agent的“大腦”。4. 設(shè)計Agent工作流的核心狀態(tài)、決策與動作循環(huán)一個簡單的腳本是線性的打開網(wǎng)頁A - 點擊按鈕B - 輸入文字C - 結(jié)束。Agent工作流則需要引入“狀態(tài)”和“決策”的概念。其核心是一個循環(huán)感知當前狀態(tài) - 根據(jù)狀態(tài)和目標任務(wù)決策 - 執(zhí)行動作 - 等待狀態(tài)變化/確認 - 進入下一循環(huán)。4.1 定義Agent的狀態(tài)感知器Agent如何“感知”頁面它需要從原始HTML中提取出對決策有用的、結(jié)構(gòu)化的“狀態(tài)信息”。這不僅僅是獲取DOM而是信息的抽象。頁面級狀態(tài)URL是否跳轉(zhuǎn)頁面標題是什么是否有特定的彈窗如Cookie同意框、登錄模態(tài)框出現(xiàn)這可以通過監(jiān)聽page.on(framenavigated)和定期檢查頁面關(guān)鍵元素來實現(xiàn)。數(shù)據(jù)級狀態(tài)目標數(shù)據(jù)是否加載完成是列表的第幾頁商品價格是多少這通常需要通過CDP執(zhí)行JavaScript從頁面全局變量、特定DOM元素的內(nèi)容或網(wǎng)絡(luò)請求的響應(yīng)中提取。交互元素狀態(tài)按鈕是可點擊還是禁用輸入框是否有值復(fù)選框是否被選中這需要結(jié)合DOM屬性(disabled,checked)和CSS計算樣式(pointer-events,opacity)來判斷。我們可以構(gòu)建一個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): 刷新當前頁面的狀態(tài)感知 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) # 可以在這里添加更多感知邏輯如檢查特定數(shù)據(jù)元素 async def get_interactable_elements(self, roleNone, nameNone): 獲取當前頁面中所有可交互元素或符合特定角色的元素及其狀態(tài) # 這是一個簡化示例實際中可以利用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() # 獲取位置和大小用于后續(xù)操作 } # 根據(jù)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: # 這里可以更智能地判斷彈窗內(nèi)容 return True # 簡化返回 return False4.2 構(gòu)建決策引擎從規(guī)則到LLM決策是Agent的“大腦”。根據(jù)復(fù)雜程度可以分為幾個層次1. 基于規(guī)則的決策最簡單直接。定義一系列“如果-那么”規(guī)則。class RuleBasedDecider: async def decide_next_action(self, page_state, goal): if page_state.has_blocking_modal: # 規(guī)則1如果有登錄彈窗嘗試關(guān)閉或登錄 return Action(typeclose_modal, target.close-button) if 購物車 in goal and 結(jié)算 in [e[text] for e in page_state.interactable_elements]: # 規(guī)則2如果目標是購物車且頁面有結(jié)算按鈕點擊它 return Action(typeclick, targettext結(jié)算) # 規(guī)則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]) # 如果沒有匹配規(guī)則返回探索性動作比如滾動或點擊可能的內(nèi)容區(qū)域 return Action(typescroll, directiondown)這種方式在流程固定的場景如固定的后臺操作流程非常有效但缺乏靈活性。2. 基于LLM的決策這是讓Agent真正“智能”起來的關(guān)鍵。我們可以將頁面狀態(tài)簡化后的DOM結(jié)構(gòu)、關(guān)鍵元素信息、任務(wù)目標構(gòu)造為提示詞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. 構(gòu)建頁面狀態(tài)的文本描述 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]])} 任務(wù)目標: {goal} # 2. 構(gòu)造Prompt prompt f 你是一個網(wǎng)頁自動化助手。請根據(jù)以下頁面狀態(tài)和任務(wù)目標決定下一步最合理的單個動作。 只輸出一個JSON對象格式必須嚴格如下{{action: click|input|scroll|wait|goto, target: 元素選擇器或URL, value: 僅當動作為input時需要表示輸入內(nèi)容}} {state_description} 請分析當前狀態(tài)并輸出下一步動作JSON # 3. 調(diào)用LLM response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, # 低隨機性保證決策穩(wěn)定 ) # 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處理未曾預(yù)見的頁面布局和流程。但成本高、速度慢且需要精心設(shè)計Prompt和結(jié)果解析邏輯來保證穩(wěn)定性。一個折中的方案是混合決策大部分固定流程用規(guī)則處理遇到未知狀態(tài)或規(guī)則未覆蓋的情況再調(diào)用LLM。4.3 動作執(zhí)行器將決策轉(zhuǎn)化為CDP命令決策引擎輸出一個抽象的Action對象如{type: click, target: #submit-btn}動作執(zhí)行器負責將其翻譯成具體的、通過Playwright或CDP執(zhí)行的操作并處理執(zhí)行結(jié)果和異常。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它內(nèi)置了智能等待和重試 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: # 執(zhí)行原始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快照這對于調(diào)試Agent至關(guān)重要 error_screenshot_path ferror_{int(time.time())}.png await self.page.screenshot(patherror_screenshot_path, full_pageTrue) # 可以在這里觸發(fā)狀態(tài)刷新或者將錯誤信息反饋給決策引擎進行恢復(fù)嘗試 return {success: False, error: str(e), screenshot: error_screenshot_path}4.4 組裝工作流主循環(huán)將狀態(tài)感知、決策、執(zhí)行串聯(lián)起來就形成了Agent的主循環(huán)。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(決策引擎未返回動作任務(wù)可能已完成或無法繼續(xù)。) break print(f 決策: {next_action}) # 3. 執(zhí)行 result await self.executor.execute(next_action) if not result[success]: print(f 執(zhí)行失敗: {result[error]}) # 這里可以加入錯誤處理邏輯比如重試、換策略、或終止 # 例如如果是元素未找到可以刷新狀態(tài)再試一次 await self.page.wait_for_timeout(2000) continue # 4. 等待狀態(tài)穩(wěn)定根據(jù)動作類型決定 await self.page.wait_for_timeout(1000) # 基礎(chǔ)等待 # 可選等待特定網(wǎng)絡(luò)請求完成或元素出現(xiàn) # await self.page.wait_for_load_state(networkidle) # 5. 檢查目標是否達成這是一個簡化檢查實際需要根據(jù)goal定義達成條件 if await self._is_goal_achieved(): print(任務(wù)目標達成) break async def _is_goal_achieved(self): # 根據(jù)goal定義判斷邏輯例如頁面URL包含特定字符、出現(xiàn)了特定元素等 if 訂單成功 in await self.page.content(): return True return False5. 實戰(zhàn)進階處理復(fù)雜場景與提升Agent魯棒性一個能在理想環(huán)境下運行的Agent是脆弱的。真實網(wǎng)頁充滿不確定性網(wǎng)絡(luò)延遲、元素加載緩慢、動態(tài)內(nèi)容、反爬措施、驗證碼、彈窗廣告等。我們必須增強Agent的魯棒性。5.1 對抗動態(tài)加載與元素定位失敗這是最常見的問題。頁面用了大量JavaScript異步加載元素出現(xiàn)的時間不確定。策略一強化等待策略。不要用固定的sleep而是使用智能等待。# Playwright 內(nèi)置的等待機制非常好用 await page.locator(button.submit).click() # 內(nèi)部會自動等待元素可點擊 await page.wait_for_selector(.result-item, statevisible, timeout30000) await page.wait_for_function(window.dataLoaded true, timeout15000) # 等待JS變量策略二使用更健壯的選擇器。避免使用易變的類名或ID優(yōu)先使用># 脆弱的 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 繞過反爬與自動化檢測現(xiàn)代網(wǎng)站會檢測自動化工具。CDP和Playwright雖然比傳統(tǒng)WebDriver隱蔽但仍有特征。基礎(chǔ)規(guī)避啟動瀏覽器時添加參數(shù)并注入腳本清除自動化特征。browser await p.chromium.launch( headlessFalse, # 有些網(wǎng)站會檢測headless模式調(diào)試時可關(guān)閉 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 })注意過度模擬反而會產(chǎn)生不自然的行為模式。最好的策略是結(jié)合使用并針對具體目標網(wǎng)站進行測試和調(diào)整。使用代理IP和用戶代理輪換通過browser.new_context每次創(chuàng)建新的上下文時可以指定不同的代理和User-Agent。proxies [http://proxy1:port, http://proxy2:port] # 請使用合法合規(guī)的代理服務(wù) user_agents [...] context await browser.new_context( proxy{server: random.choice(proxies)}, user_agentrandom.choice(user_agents) )5.3 集成外部服務(wù)處理驗證碼驗證碼是自動化的大敵。對于簡單圖形驗證碼可以嘗試用OCR庫如pytesseract但效果通常不佳。對于復(fù)雜驗證碼如點選、滑塊可靠的做法是接入第三方打碼平臺。import requests async def solve_captcha(page): 處理驗證碼的示例流程 # 1. 定位驗證碼圖片元素并截圖 captcha_element page.locator(#captcha-img) captcha_screenshot await captcha_element.screenshot() # 2. 調(diào)用打碼平臺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 工作流的持久化與狀態(tài)管理對于長時間運行或需要中斷恢復(fù)的Agent需要將工作流狀態(tài)當前URL、已收集的數(shù)據(jù)、步驟索引等持久化到數(shù)據(jù)庫或文件。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 # 可以從保存的步驟繼續(xù)執(zhí)行 while self.current_step self.max_steps: # ... 執(zhí)行單步循環(huán) ... self.current_step 1 await self.save_state() # 每步或每N步保存一次6. 案例拆解構(gòu)建一個商品價格監(jiān)控Agent讓我們用一個相對完整的例子把上面的組件串聯(lián)起來。目標是監(jiān)控某電商網(wǎng)站特定商品的價格變化并在降價時發(fā)送通知。目標分解登錄網(wǎng)站處理登錄態(tài)。導(dǎo)航到目標商品頁面。提取商品價格和庫存狀態(tài)。判斷價格是否低于設(shè)定閾值。如果滿足條件通過郵件或Webhook發(fā)送通知。定期執(zhí)行如每30分鐘一次。核心實現(xiàn)要點狀態(tài)感知我們需要感知的關(guān)鍵狀態(tài)是“商品價格”和“購買按鈕狀態(tài)”。這需要通過執(zhí)行頁面JavaScript來提取。async def extract_product_info(page): # 使用Playwright的evaluate在頁面上下文中執(zhí)行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決策引擎這個場景規(guī)則明確適合用規(guī)則引擎。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]: # 達到購買條件觸發(fā)通知動作 return {type: trigger_alert, price: current_price} else: # 未達條件等待一段時間后重新檢查 return {type: wait, duration: 1800} # 等待30分鐘動作執(zhí)行器擴展需要增加trigger_alert動作。async def execute(self, action): if action[type] trigger_alert: # 調(diào)用發(fā)送郵件或Webhook的函數(shù) await self.send_alert(action[price]) return {success: True} # ... 處理其他動作類型 ...調(diào)度與循環(huán)使用asyncio或schedule庫來定期運行整個Agent流程。注意要妥善管理瀏覽器實例避免內(nèi)存泄漏。一個常見的模式是每次監(jiān)控任務(wù)完成后關(guān)閉瀏覽器下次任務(wù)再啟動。部署與監(jiān)控可以將這個Agent腳本部署到服務(wù)器或云函數(shù)上。務(wù)必加入日志記錄如structlog或loguru記錄每一步的狀態(tài)、決策和異常方便問題排查。對于關(guān)鍵任務(wù)可以集成哨兵監(jiān)控當Agent連續(xù)失敗多次時發(fā)出告警。構(gòu)建基于CDP的瀏覽器自動化Agent工作流是一個從“控制瀏覽器”到“賦予瀏覽器智能”的升級過程。它不再僅僅是模擬點擊而是創(chuàng)建了一個能夠感知環(huán)境、分析信息、做出決策并執(zhí)行任務(wù)的數(shù)字助手。這條路從理解CDP協(xié)議開始經(jīng)過工程化的庫選型、核心循環(huán)的設(shè)計再到應(yīng)對真實世界復(fù)雜性的各種策略每一步都需要在靈活性與穩(wěn)定性之間權(quán)衡。我個人的體會是初期用Playwright這類高階框架快速搭建原型驗證想法在遇到性能瓶頸或需要深度定制時再深入CDP底層去挖掘能力是一個比較平滑的學(xué)習和實踐路徑。最終一個健壯的Agent系統(tǒng)其價值不僅在于自動化了多少流程更在于它能否像一名可靠的員工一樣在變化的環(huán)境中持續(xù)、穩(wěn)定地完成工作。 基于CDP構(gòu)建智能瀏覽器Agent：從自動化腳本到工作流實踐
1. 從“手動點點點”到“智能自動化”的跨越如果你還在用Selenium或者Puppeteer寫一長串find_element和click來模擬瀏覽器操作每次頁面結(jié)構(gòu)一變就得重新調(diào)試腳本那感覺就像在用算盤處理大數(shù)據(jù)。我最初接觸瀏覽器自動化也是從這個階段過來的直到后來項目里需要處理大量動態(tài)渲染、反爬策略復(fù)雜的頁面以及實現(xiàn)一些需要“觀察-決策-執(zhí)行”的智能流程時傳統(tǒng)的腳本模式就徹底不夠用了。這時候一個更底層的協(xié)議——Chrome DevTools Protocol也就是CDP進入了我的視野。CDP不是一個新的工具而是Chrome瀏覽器內(nèi)置的調(diào)試協(xié)議。它像是一把能直接與瀏覽器內(nèi)核對話的“手術(shù)刀”讓你能控制標簽頁、攔截網(wǎng)絡(luò)請求、執(zhí)行JavaScript、獲取DOM快照甚至監(jiān)聽內(nèi)存變化。基于CDP構(gòu)建自動化意味著你跳過了WebDriver這類“翻譯官”的中間層直接與瀏覽器引擎交互響應(yīng)更快能力也更強大。但CDP本身是協(xié)議不是框架直接用它寫代碼就像用匯編語言編程強大但繁瑣。所以我們今天的重點不是羅列CDP的API而是如何以它為基礎(chǔ)構(gòu)建一個更高級的抽象Agent工作流。所謂Agent工作流你可以把它想象成一個在瀏覽器里“上班”的虛擬員工。它不再是被動執(zhí)行預(yù)設(shè)步驟的腳本而是一個具備一定感知、決策和執(zhí)行能力的智能體。它知道自己要完成什么任務(wù)比如“監(jiān)控商品價格變化并截圖報警”能通過CDP“看到”頁面內(nèi)容能根據(jù)頁面狀態(tài)“思考”下一步該點哪里、輸入什么并能處理一些意外情況比如彈窗、驗證碼。這種模式正是當前從“自動化腳本”向“智能體應(yīng)用”演進的核心實踐。2. 為什么是CDP超越WebDriver的底層能力剖析在深入構(gòu)建之前我們必須先搞清楚為什么選擇CDP作為基石而不是繼續(xù)用更成熟的Selenium WebDriver。這不僅僅是性能問題更是能力邊界和設(shè)計哲學(xué)的不同。2.1 協(xié)議層 vs 驅(qū)動層本質(zhì)差異WebDriver是一個W3C標準它定義了一套跨瀏覽器的、用于控制網(wǎng)頁的通用指令集比如“點擊元素”、“獲取文本”。它的工作模式是你的腳本通過特定語言的客戶端庫如Python的selenium包發(fā)送指令給一個獨立的驅(qū)動程序如chromedriver這個驅(qū)動再通過私有協(xié)議與真實的瀏覽器通信。這帶來幾個問題一是多了一層轉(zhuǎn)換速度和效率有損耗二是驅(qū)動和瀏覽器版本必須嚴格匹配否則經(jīng)常出兼容性問題三是能力受限于WebDriver標準一些瀏覽器獨有的、高級的調(diào)試功能無法使用。CDP則完全不同。它是Chrome/Chromium內(nèi)核原生暴露的調(diào)試接口基于WebSocket通信。你的代碼直接連接到一個Chrome實例的調(diào)試端口發(fā)送和接收的都是JSON-RPC格式的消息。這意味著無中間商延遲更低指令直達瀏覽器引擎執(zhí)行和獲取結(jié)果的延遲顯著降低對于需要高頻交互的自動化場景至關(guān)重要。能力全集你能用到Chrome開發(fā)者工具里幾乎所有的功能包括但不限于網(wǎng)絡(luò)控制精確攔截、修改、放行任意請求和響應(yīng)模擬弱網(wǎng)環(huán)境。性能分析獲取時間線追蹤、內(nèi)存堆快照、計算性能指標。DOM深度操作不僅獲取元素還能監(jiān)聽DOM變化獲取CSS計算樣式甚至模擬用戶輸入如輸入法組合鍵。運行時注入在頁面上下文中執(zhí)行任意JavaScript并獲取復(fù)雜的返回值如函數(shù)、Promise。頁面快照生成包含完整CSSOM的截圖甚至能生成PDF。版本兼容性更好雖然CDP本身也在演進但它的向后兼容性通常比chromedriver的版本鎖死要好處理得多核心API相對穩(wěn)定。2.2 構(gòu)建Agent工作流的核心優(yōu)勢對于Agent工作流而言CDP提供的這些底層能力是構(gòu)建“智能”的感官系統(tǒng)。感知能力Agent需要“看”懂頁面。通過CDP的DOMSnapshot.captureSnapshot或Runtime.evaluateAgent可以獲取結(jié)構(gòu)化、帶布局信息的DOM樹這比通過innerHTML獲取的字符串要強大得多。結(jié)合Accessibility.getFullAXTree還能獲取無障礙樹理解元素的語義角色這對于理解復(fù)雜UI組件如下拉菜單、滑塊的狀態(tài)非常有幫助。決策依據(jù)Agent的決策需要數(shù)據(jù)。CDP的Network域可以讓Agent監(jiān)聽所有XHR/Fetch請求和響應(yīng)這對于監(jiān)控數(shù)據(jù)接口、判斷頁面加載狀態(tài)是加載完成還是加載失敗提供了關(guān)鍵信息。Console域可以監(jiān)聽頁面JavaScript的日志和錯誤幫助診斷頁面自身的問題。精準執(zhí)行Input.dispatchMouseEvent和Input.dispatchKeyEvent可以模擬極其精細的鼠標移動、點擊、拖拽和鍵盤事件包括坐標、按鍵時長等這對于繞過一些基于事件監(jiān)聽的反爬機制或者操作Canvas、WebGL等非DOM元素至關(guān)重要。注意直接使用CDP的Input事件需要自己計算坐標或目標元素比WebDriver的“基于元素定位”要復(fù)雜但這也帶來了靈活性。一個常見的實踐是結(jié)合使用先用CDP的DOM查詢找到元素并獲取其邊界框再用CDP的Input事件進行精準操作。理解了CDP的價值我們就可以開始搭建地基了。直接裸寫CDP通信非常痛苦好在社區(qū)已經(jīng)有了優(yōu)秀的封裝庫。3. 工程化起點選擇與配置你的CDP客戶端庫市面上有幾個主流的CDP客戶端庫它們幫你處理了WebSocket連接、消息收發(fā)、事件監(jiān)聽等底層細節(jié)提供了更友好的API。1. Puppeteer (Node.js)這是Google官方團隊維護的項目可以理解為“CDP的高階封裝”。它提供了非常直觀的、類似WebDriver的API如page.click(selector)但其底層完全基于CDP。它的優(yōu)點是開箱即用功能全面生態(tài)豐富。缺點是它綁定Node.js環(huán)境且為了易用性隱藏了部分CDP細節(jié)當你需要極致的定制或使用某些實驗性CDP功能時可能需要繞過Puppeteer直接調(diào)用CDP會話。2. Playwright (Node.js/Python/.NET/Java)由微軟團隊開發(fā)最初是Puppeteer的fork但現(xiàn)在已自成一體。它支持多瀏覽器Chromium, Firefox, WebKit其CDP實現(xiàn)主要針對Chromium。Playwright的API設(shè)計更現(xiàn)代化強調(diào)自動等待和可靠性內(nèi)置了很多防脆性測試的特性如自動重試、等待元素可操作狀態(tài)。對于構(gòu)建健壯的Agent工作流它的這些特性非常有吸引力。3. pyppeteer (Python)可以看作是Puppeteer的Python非官方移植版。早期很有用但隨著Playwright Python版的成熟和穩(wěn)定pyppeteer的維護活躍度下降目前不是新項目的首選。4. 直接使用CDP客戶端 (如chrome-remote-interfacefor Node.js,pychromefor Python)這些是更輕量級的庫只提供連接和發(fā)送CDP命令的基礎(chǔ)能力不提供高級API。你需要自己用CDP命令組合出“點擊”、“輸入”等操作。這提供了最大的靈活性但開發(fā)效率最低適合需要深度定制或研究CDP協(xié)議本身的場景。我的選擇與實踐建議對于大多數(shù)旨在構(gòu)建Agent工作流的項目我推薦從Playwright (Python版)開始。原因如下多語言支持團隊可以根據(jù)技術(shù)棧選擇核心邏輯可以共享。可靠性設(shè)計內(nèi)置的自動等待、智能定位器如get_by_role、追蹤器Trace Viewer對于復(fù)雜多變的網(wǎng)頁環(huán)境非常友好能減少Agent“卡住”或“點錯”的情況。強大的上下文隔離Playwright可以輕松創(chuàng)建多個獨立的瀏覽器上下文這對于需要同時登錄多個賬號、隔離會話狀態(tài)的Agent場景是剛需。CDP會話直通在需要時你可以通過page.context.new_cdp_session(page)獲取底層CDP會話對象直接發(fā)送原始CDP命令兼顧了易用性與靈活性。環(huán)境搭建實操假設(shè)我們使用Playwright for Python。# 1. 安裝playwright庫 pip install playwright # 2. 安裝瀏覽器驅(qū)動Chromium, Firefox, WebKit playwright install chromium# 3. 基礎(chǔ)啟動腳本 import asyncio from playwright.async_api import async_playwright async def main(): async with async_playwright() as p: # 啟動瀏覽器headlessFalse表示顯示界面調(diào)試時有用 browser await p.chromium.launch(headlessFalse, args[--disable-blink-featuresAutomationControlled]) # 創(chuàng)建一個瀏覽器上下文類似一個獨立的隱身會話 context await browser.new_context( viewport{width: 1920, height: 1080}, user_agentMozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ... ) # 禁用WebDriver屬性降低被檢測風險基礎(chǔ)規(guī)避 await context.add_init_script( Object.defineProperty(navigator, webdriver, { get: () undefined }); ) page await context.new_page() # 此時你已經(jīng)擁有了一個可通過Playwright API控制的頁面對象page # 同時你也可以獲取底層的CDP會話進行更精細的操作 cdp_session await context.new_cdp_session(page) # 例如啟用Network域監(jiān)聽 await cdp_session.send(Network.enable) # 監(jiān)聽網(wǎng)絡(luò)請求 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())這個基礎(chǔ)腳本已經(jīng)包含了啟動、基礎(chǔ)配置、以及如何接入原始CDP會話。接下來我們要思考如何在這個基礎(chǔ)上構(gòu)建一個Agent的“大腦”。4. 設(shè)計Agent工作流的核心狀態(tài)、決策與動作循環(huán)一個簡單的腳本是線性的打開網(wǎng)頁A - 點擊按鈕B - 輸入文字C - 結(jié)束。Agent工作流則需要引入“狀態(tài)”和“決策”的概念。其核心是一個循環(huán)感知當前狀態(tài) - 根據(jù)狀態(tài)和目標任務(wù)決策 - 執(zhí)行動作 - 等待狀態(tài)變化/確認 - 進入下一循環(huán)。4.1 定義Agent的狀態(tài)感知器Agent如何“感知”頁面它需要從原始HTML中提取出對決策有用的、結(jié)構(gòu)化的“狀態(tài)信息”。這不僅僅是獲取DOM而是信息的抽象。頁面級狀態(tài)URL是否跳轉(zhuǎn)頁面標題是什么是否有特定的彈窗如Cookie同意框、登錄模態(tài)框出現(xiàn)這可以通過監(jiān)聽page.on(framenavigated)和定期檢查頁面關(guān)鍵元素來實現(xiàn)。數(shù)據(jù)級狀態(tài)目標數(shù)據(jù)是否加載完成是列表的第幾頁商品價格是多少這通常需要通過CDP執(zhí)行JavaScript從頁面全局變量、特定DOM元素的內(nèi)容或網(wǎng)絡(luò)請求的響應(yīng)中提取。交互元素狀態(tài)按鈕是可點擊還是禁用輸入框是否有值復(fù)選框是否被選中這需要結(jié)合DOM屬性(disabled,checked)和CSS計算樣式(pointer-events,opacity)來判斷。我們可以構(gòu)建一個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): 刷新當前頁面的狀態(tài)感知 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) # 可以在這里添加更多感知邏輯如檢查特定數(shù)據(jù)元素 async def get_interactable_elements(self, roleNone, nameNone): 獲取當前頁面中所有可交互元素或符合特定角色的元素及其狀態(tài) # 這是一個簡化示例實際中可以利用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() # 獲取位置和大小用于后續(xù)操作 } # 根據(jù)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: # 這里可以更智能地判斷彈窗內(nèi)容 return True # 簡化返回 return False4.2 構(gòu)建決策引擎從規(guī)則到LLM決策是Agent的“大腦”。根據(jù)復(fù)雜程度可以分為幾個層次1. 基于規(guī)則的決策最簡單直接。定義一系列“如果-那么”規(guī)則。class RuleBasedDecider: async def decide_next_action(self, page_state, goal): if page_state.has_blocking_modal: # 規(guī)則1如果有登錄彈窗嘗試關(guān)閉或登錄 return Action(typeclose_modal, target.close-button) if 購物車 in goal and 結(jié)算 in [e[text] for e in page_state.interactable_elements]: # 規(guī)則2如果目標是購物車且頁面有結(jié)算按鈕點擊它 return Action(typeclick, targettext結(jié)算) # 規(guī)則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]) # 如果沒有匹配規(guī)則返回探索性動作比如滾動或點擊可能的內(nèi)容區(qū)域 return Action(typescroll, directiondown)這種方式在流程固定的場景如固定的后臺操作流程非常有效但缺乏靈活性。2. 基于LLM的決策這是讓Agent真正“智能”起來的關(guān)鍵。我們可以將頁面狀態(tài)簡化后的DOM結(jié)構(gòu)、關(guān)鍵元素信息、任務(wù)目標構(gòu)造為提示詞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. 構(gòu)建頁面狀態(tài)的文本描述 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]])} 任務(wù)目標: {goal} # 2. 構(gòu)造Prompt prompt f 你是一個網(wǎng)頁自動化助手。請根據(jù)以下頁面狀態(tài)和任務(wù)目標決定下一步最合理的單個動作。 只輸出一個JSON對象格式必須嚴格如下{{action: click|input|scroll|wait|goto, target: 元素選擇器或URL, value: 僅當動作為input時需要表示輸入內(nèi)容}} {state_description} 請分析當前狀態(tài)并輸出下一步動作JSON # 3. 調(diào)用LLM response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], temperature0.1, # 低隨機性保證決策穩(wěn)定 ) # 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處理未曾預(yù)見的頁面布局和流程。但成本高、速度慢且需要精心設(shè)計Prompt和結(jié)果解析邏輯來保證穩(wěn)定性。一個折中的方案是混合決策大部分固定流程用規(guī)則處理遇到未知狀態(tài)或規(guī)則未覆蓋的情況再調(diào)用LLM。4.3 動作執(zhí)行器將決策轉(zhuǎn)化為CDP命令決策引擎輸出一個抽象的Action對象如{type: click, target: #submit-btn}動作執(zhí)行器負責將其翻譯成具體的、通過Playwright或CDP執(zhí)行的操作并處理執(zhí)行結(jié)果和異常。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它內(nèi)置了智能等待和重試 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: # 執(zhí)行原始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快照這對于調(diào)試Agent至關(guān)重要 error_screenshot_path ferror_{int(time.time())}.png await self.page.screenshot(patherror_screenshot_path, full_pageTrue) # 可以在這里觸發(fā)狀態(tài)刷新或者將錯誤信息反饋給決策引擎進行恢復(fù)嘗試 return {success: False, error: str(e), screenshot: error_screenshot_path}4.4 組裝工作流主循環(huán)將狀態(tài)感知、決策、執(zhí)行串聯(lián)起來就形成了Agent的主循環(huán)。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(決策引擎未返回動作任務(wù)可能已完成或無法繼續(xù)。) break print(f 決策: {next_action}) # 3. 執(zhí)行 result await self.executor.execute(next_action) if not result[success]: print(f 執(zhí)行失敗: {result[error]}) # 這里可以加入錯誤處理邏輯比如重試、換策略、或終止 # 例如如果是元素未找到可以刷新狀態(tài)再試一次 await self.page.wait_for_timeout(2000) continue # 4. 等待狀態(tài)穩(wěn)定根據(jù)動作類型決定 await self.page.wait_for_timeout(1000) # 基礎(chǔ)等待 # 可選等待特定網(wǎng)絡(luò)請求完成或元素出現(xiàn) # await self.page.wait_for_load_state(networkidle) # 5. 檢查目標是否達成這是一個簡化檢查實際需要根據(jù)goal定義達成條件 if await self._is_goal_achieved(): print(任務(wù)目標達成) break async def _is_goal_achieved(self): # 根據(jù)goal定義判斷邏輯例如頁面URL包含特定字符、出現(xiàn)了特定元素等 if 訂單成功 in await self.page.content(): return True return False5. 實戰(zhàn)進階處理復(fù)雜場景與提升Agent魯棒性一個能在理想環(huán)境下運行的Agent是脆弱的。真實網(wǎng)頁充滿不確定性網(wǎng)絡(luò)延遲、元素加載緩慢、動態(tài)內(nèi)容、反爬措施、驗證碼、彈窗廣告等。我們必須增強Agent的魯棒性。5.1 對抗動態(tài)加載與元素定位失敗這是最常見的問題。頁面用了大量JavaScript異步加載元素出現(xiàn)的時間不確定。策略一強化等待策略。不要用固定的sleep而是使用智能等待。# Playwright 內(nèi)置的等待機制非常好用 await page.locator(button.submit).click() # 內(nèi)部會自動等待元素可點擊 await page.wait_for_selector(.result-item, statevisible, timeout30000) await page.wait_for_function(window.dataLoaded true, timeout15000) # 等待JS變量策略二使用更健壯的選擇器。避免使用易變的類名或ID優(yōu)先使用># 脆弱的 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 繞過反爬與自動化檢測現(xiàn)代網(wǎng)站會檢測自動化工具。CDP和Playwright雖然比傳統(tǒng)WebDriver隱蔽但仍有特征。基礎(chǔ)規(guī)避啟動瀏覽器時添加參數(shù)并注入腳本清除自動化特征。browser await p.chromium.launch( headlessFalse, # 有些網(wǎng)站會檢測headless模式調(diào)試時可關(guān)閉 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 })注意過度模擬反而會產(chǎn)生不自然的行為模式。最好的策略是結(jié)合使用并針對具體目標網(wǎng)站進行測試和調(diào)整。使用代理IP和用戶代理輪換通過browser.new_context每次創(chuàng)建新的上下文時可以指定不同的代理和User-Agent。proxies [http://proxy1:port, http://proxy2:port] # 請使用合法合規(guī)的代理服務(wù) user_agents [...] context await browser.new_context( proxy{server: random.choice(proxies)}, user_agentrandom.choice(user_agents) )5.3 集成外部服務(wù)處理驗證碼驗證碼是自動化的大敵。對于簡單圖形驗證碼可以嘗試用OCR庫如pytesseract但效果通常不佳。對于復(fù)雜驗證碼如點選、滑塊可靠的做法是接入第三方打碼平臺。import requests async def solve_captcha(page): 處理驗證碼的示例流程 # 1. 定位驗證碼圖片元素并截圖 captcha_element page.locator(#captcha-img) captcha_screenshot await captcha_element.screenshot() # 2. 調(diào)用打碼平臺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 工作流的持久化與狀態(tài)管理對于長時間運行或需要中斷恢復(fù)的Agent需要將工作流狀態(tài)當前URL、已收集的數(shù)據(jù)、步驟索引等持久化到數(shù)據(jù)庫或文件。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 # 可以從保存的步驟繼續(xù)執(zhí)行 while self.current_step self.max_steps: # ... 執(zhí)行單步循環(huán) ... self.current_step 1 await self.save_state() # 每步或每N步保存一次6. 案例拆解構(gòu)建一個商品價格監(jiān)控Agent讓我們用一個相對完整的例子把上面的組件串聯(lián)起來。目標是監(jiān)控某電商網(wǎng)站特定商品的價格變化并在降價時發(fā)送通知。目標分解登錄網(wǎng)站處理登錄態(tài)。導(dǎo)航到目標商品頁面。提取商品價格和庫存狀態(tài)。判斷價格是否低于設(shè)定閾值。如果滿足條件通過郵件或Webhook發(fā)送通知。定期執(zhí)行如每30分鐘一次。核心實現(xiàn)要點狀態(tài)感知我們需要感知的關(guān)鍵狀態(tài)是“商品價格”和“購買按鈕狀態(tài)”。這需要通過執(zhí)行頁面JavaScript來提取。async def extract_product_info(page): # 使用Playwright的evaluate在頁面上下文中執(zhí)行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決策引擎這個場景規(guī)則明確適合用規(guī)則引擎。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]: # 達到購買條件觸發(fā)通知動作 return {type: trigger_alert, price: current_price} else: # 未達條件等待一段時間后重新檢查 return {type: wait, duration: 1800} # 等待30分鐘動作執(zhí)行器擴展需要增加trigger_alert動作。async def execute(self, action): if action[type] trigger_alert: # 調(diào)用發(fā)送郵件或Webhook的函數(shù) await self.send_alert(action[price]) return {success: True} # ... 處理其他動作類型 ...調(diào)度與循環(huán)使用asyncio或schedule庫來定期運行整個Agent流程。注意要妥善管理瀏覽器實例避免內(nèi)存泄漏。一個常見的模式是每次監(jiān)控任務(wù)完成后關(guān)閉瀏覽器下次任務(wù)再啟動。部署與監(jiān)控可以將這個Agent腳本部署到服務(wù)器或云函數(shù)上。務(wù)必加入日志記錄如structlog或loguru記錄每一步的狀態(tài)、決策和異常方便問題排查。對于關(guān)鍵任務(wù)可以集成哨兵監(jiān)控當Agent連續(xù)失敗多次時發(fā)出告警。構(gòu)建基于CDP的瀏覽器自動化Agent工作流是一個從“控制瀏覽器”到“賦予瀏覽器智能”的升級過程。它不再僅僅是模擬點擊而是創(chuàng)建了一個能夠感知環(huán)境、分析信息、做出決策并執(zhí)行任務(wù)的數(shù)字助手。這條路從理解CDP協(xié)議開始經(jīng)過工程化的庫選型、核心循環(huán)的設(shè)計再到應(yīng)對真實世界復(fù)雜性的各種策略每一步都需要在靈活性與穩(wěn)定性之間權(quán)衡。我個人的體會是初期用Playwright這類高階框架快速搭建原型驗證想法在遇到性能瓶頸或需要深度定制時再深入CDP底層去挖掘能力是一個比較平滑的學(xué)習和實踐路徑。最終一個健壯的Agent系統(tǒng)其價值不僅在于自動化了多少流程更在于它能否像一名可靠的員工一樣在變化的環(huán)境中持續(xù)、穩(wěn)定地完成工作。