1. 從“手動點點點”到“智能自動化”的跨越如果你還在用Selenium或者Puppeteer寫一長串find_element和click來模擬瀏覽器操作每次頁面結構一變就得重新調試腳本那感覺就像在用算盤處理大數據。我最初接觸瀏覽器自動化也是從這個階段過來的直到后來項目里需要處理大量動態渲染、反爬策略復雜的頁面以及實現一些需要“觀察-決策-執行”的智能流程時傳統的腳本模式就徹底不夠用了。這時候一個更底層的協議——Chrome DevTools Protocol也就是CDP進入了我的視野。CDP不是一個新的工具而是Chrome瀏覽器內置的調試協議。它像是一把能直接與瀏覽器內核對話的“手術刀”讓你能控制標簽頁、攔截網絡請求、執行JavaScript、獲取DOM快照甚至監聽內存變化?；贑DP構建自動化意味著你跳過了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版)開始。原因如下多語言支持團隊可以根據技術棧選擇核心邏輯可以共享?？煽啃栽O計內置的自動等待、智能定位器如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系統其價值不僅在于自動化了多少流程更在于它能否像一名可靠的員工一樣在變化的環境中持續、穩定地完成工作。