1. 项目概述当AI学会“上网冲浪”最近在折腾一个挺有意思的东西我把它叫做“AI的浏览器”。简单来说就是让一个大型语言模型LLM能够像我们人类一样打开浏览器输入网址点击按钮填写表单从网页上抓取信息然后基于这些信息做出决策或回答问题。这听起来是不是有点像科幻电影里的场景其实这就是“Web Agent”网页智能体的核心概念。我这次聚焦的项目是围绕“Google搜索”这个场景来构建一个Web Agent。为什么是Google搜索因为它几乎是互联网信息获取的“第一入口”也是检验一个智能体能否在真实、开放、动态的网页环境中有效工作的绝佳试金石。想象一下你问AI“今年最值得关注的AI芯片有哪些”一个传统的、基于静态知识库的AI可能会给出一个过时的列表。但一个配备了Web Agent能力的AI可以实时打开Google搜索最新的行业新闻、技术博客和产品发布然后为你整理出一份新鲜、全面且附有来源的报告。这背后的价值对于信息时效性要求极高的领域如金融分析、舆情监控、学术研究、竞品追踪来说是颠覆性的。这个项目的目标就是打造一个能够自主、准确、高效地完成Google搜索任务并解析结果、提取关键信息的智能体。它不仅仅是调用一下搜索API那么简单而是要模拟人类用户的完整交互流程打开浏览器、定位搜索框、输入查询词、点击搜索按钮、等待结果加载、从结果页中识别和提取标题、链接、摘要甚至进行翻页和筛选。这整个过程涉及到浏览器自动化、网页元素定位、自然语言指令理解、动作规划与执行等一系列技术的深度融合。接下来我就把自己在构建这个“Google搜索智能体”过程中的核心思路、技术选型、实操细节以及踩过的那些坑毫无保留地分享出来。2. 核心架构与工具选型解析要构建一个能操作浏览器的AI我们得先给它准备好“手”和“眼睛”并设计好“大脑”的指挥逻辑。整个架构可以清晰地分为三层环境层、驱动层和智能体层。2.1 环境层浏览器的选择与控制浏览器是我们的智能体与网页世界交互的唯一窗口。选择哪个浏览器以及如何控制它是第一步。为什么选择Chrome/Chromium几乎没有任何悬念Chrome或其开源版本Chromium是自动化测试和爬虫领域的绝对主流。理由很充分广泛的工具支持Chrome DevTools Protocol (CDP) 提供了极其丰富的底层控制接口远超其他浏览器。无头模式Headless Mode这是关键。无头模式意味着浏览器在后台运行没有图形用户界面。这大大节省了系统资源CPU、内存特别适合在服务器上部署。从Chrome 59版本开始无头模式已经非常稳定。强大的开发者生态Selenium、Puppeteer、Playwright等主流自动化工具都对Chrome提供了最优先、最完整的支持。控制工具选型Puppeteer vs. Playwright这是两个最优秀的现代浏览器自动化库。我最终选择了Playwright原因如下特性PuppeteerPlaywright我们的选择理由浏览器支持主要为ChromeChrome, Firefox, Safari, EdgePlaywright的多浏览器支持意味着我们的智能体适应性更强。虽然我们主用Chrome但此架构为未来扩展留有余地。API设计优秀但相对较早更现代一致性更好Playwright的API在跨浏览器时保持高度一致且在一些复杂交互如iframe处理、文件上传上更优雅。自动等待基础支持智能等待Auto-waiting这是决定性因素之一。Playwright在执行点击、输入等操作前会自动等待元素变为可交互状态如可见、启用、稳定这极大地简化了代码避免了手动添加sleep或复杂等待条件让智能体的动作更稳健。录制与调试有DevTools集成强大的Codegen录制工具Playwright Test自带一个可视化录制工具可以边操作浏览器边生成代码对于快速构建和调试智能体的基础动作序列非常有帮助。提示无头模式虽然高效但在开发调试阶段建议先使用有头模式以便直观观察智能体的每一步操作确认元素定位是否准确动作执行是否符合预期。2.2 驱动层为AI提供感知与执行接口智能体AI大脑不能直接操作Playwright的API。我们需要一个中间层将网页的视觉/结构信息“翻译”给AI并将AI的决策“翻译”成具体的浏览器操作指令。这部分我设计了一个WebDriver类来封装。核心功能设计页面状态获取这是智能体的“眼睛”。不仅要获取页面的HTML源码page.content()更重要的是获取一个简化的、结构化的DOM表示。我会过滤掉脚本、样式等无关元素提取出带有关键属性如id,class,aria-label,placeholder的交互元素按钮、输入框、链接及其层级关系形成一个轻量化的“可操作元素列表”提供给AI。动作执行引擎这是智能体的“手”。根据AI的指令如CLICK [idsearch-button]TYPE [nameq] OpenAI调用对应的Playwright方法。这里的关键是错误处理与重试机制。网络延迟、元素加载稍慢都可能导致动作失败。我的驱动层会捕获这些异常进行有限次数的重试并将明确的错误信息如“元素未找到”、“元素不可点击”反馈给AI以便它调整策略。结果解析模块专门针对Google搜索结果页进行结构化信息提取。当AI判定当前页面是搜索结果页时会调用此模块。它会定位搜索结果区域通常是div#search下的div.g然后提取每个结果的标题文本和链接a标签显示URLcite标签内容摘要/片段可能的相关信息如发布时间、评分 将这些信息整理成JSON格式作为AI进行信息综合和回答用户问题的依据。2.3 智能体层LLM作为决策大脑这是整个系统的核心。我选用GPT-4 Turbo作为“大脑”因为它具有强大的上下文理解、推理和指令遵循能力。智能体的工作流程是一个经典的感知-思考-行动循环ReAct模式感知从WebDriver获取当前页面的简化DOM和屏幕截图可选用于复杂布局理解。思考将当前页面信息、用户原始问题、以及之前的操作历史组合成一个详细的提示词Prompt发送给LLM。Prompt会明确要求LLM分析当前页面状态决定下一步是“继续操作”还是“已回答”并严格以指定格式如JSON输出决策。行动解析LLM的返回结果。如果是操作指令则交给WebDriver执行如果LLM认为已获得足够信息并生成最终答案则循环结束。Prompt工程是关键中的关键。一个糟糕的Prompt会让AI胡言乱语或陷入死循环。我的Prompt结构大致如下你是一个专业的网页操作助手。你的目标是通过操作浏览器来回答用户的问题。 当前URL[当前页面URL] 当前页面可操作元素摘要[简化后的DOM列表包含元素描述和唯一选择器] 操作历史[之前的动作和结果] 用户问题[用户最初的问题] 请根据以上信息决定下一步行动。你只能选择以下两种行动之一 A. 执行一个操作。格式{action: click/type/goto/etc., selector: [元素选择器], value: 输入值仅type需要} B. 认为任务已完成给出最终答案。格式{action: answer, answer: 你的完整回答并引用信息来源} 请严格输出JSON不要有任何其他解释。这个Prompt明确了角色、上下文、可用动作和输出格式将LLM的“自由发挥”约束在一个可控的框架内极大地提高了任务的可靠性和成功率。3. 实操构建从零搭建Google搜索智能体理论讲完了我们动手把它搭起来。我会以Python环境为例展示核心步骤。3.1 环境准备与依赖安装首先确保你的Python版本在3.8以上。然后我们使用pip安装核心库。# 安装Playwright库 pip install playwright # 安装Playwright所需的浏览器驱动这步会下载Chromium、Firefox和WebKit playwright install chromium我强烈建议创建一个独立的虚拟环境如venv或conda来做这个项目以避免依赖冲突。3.2 实现WebDriver核心类我们来构建驱动层的骨架。这个类负责所有与浏览器的直接对话。import asyncio from playwright.async_api import async_playwright import json class GoogleSearchDriver: def __init__(self, headlessTrue): self.headless headless self.browser None self.page None self.context None async def start(self): 启动浏览器和页面 playwright await async_playwright().start() # 使用Chromium开启无头模式生产环境建议为True self.browser await playwright.chromium.launch(headlessself.headless, args[--no-sandbox]) # 创建一个新的上下文可以模拟特定设备或设置视口大小 self.context await self.browser.new_context(viewport{width: 1280, height: 800}) self.page await self.context.new_page() # 设置默认超时时间 self.page.set_default_timeout(30000) # 30秒 print(浏览器启动成功。) async def goto_google(self): 导航到Google首页 await self.page.goto(https://www.google.com) # 等待页面关键元素加载比如搜索框 await self.page.wait_for_selector(textarea[nameq], input[nameq], statevisible) print(已导航至Google首页。) async def get_page_state(self): 获取当前页面的结构化信息供AI分析 state { url: self.page.url, title: await self.page.title(), interactive_elements: [] } # 获取所有可交互元素输入框、按钮、链接 # 这里是一个简化示例实际中需要更精细的过滤和特征提取 all_inputs await self.page.query_selector_all(input, textarea, button, a[href], [rolebutton]) for i, element in enumerate(all_inputs[:20]): # 限制数量避免上下文过长 try: # 获取元素的一些特征属性 tag await element.evaluate(el el.tagName.toLowerCase()) elem_id await element.get_attribute(id) or name await element.get_attribute(name) or placeholder await element.get_attribute(placeholder) or aria_label await element.get_attribute(aria-label) or text await element.text_content() or # 生成一个相对稳定的选择器优先id其次name最后其他属性 if elem_id: selector f#{elem_id} elif name: selector f[name{name}] elif aria_label: selector f[aria-label{aria_label}] else: # 作为后备使用playwright的selector selector fnth{i} element_info { selector: selector, tag: tag, id: elem_id, name: name, placeholder: placeholder, aria_label: aria_label, text: text[:50] # 截断长文本 } state[interactive_elements].append(element_info) except Exception as e: # 如果元素已脱离DOM跳过 continue return state async def execute_action(self, action_dict): 执行AI发出的动作指令 action action_dict.get(action) selector action_dict.get(selector) value action_dict.get(value) if action goto: await self.page.goto(value) return f已导航至 {value} elif action type: # 先点击输入框确保焦点再清空并输入 await self.page.click(selector) await self.page.fill(selector, ) # 清空现有内容 await self.page.type(selector, value, delay50) # 模拟人工输入延迟50毫秒/字符 return f已在 {selector} 中输入文本{value} elif action click: await self.page.click(selector) return f已点击 {selector} elif action press: await self.page.press(selector, value) # 如按Enter键 return f已在 {selector} 上按下按键 {value} else: raise ValueError(f不支持的 action 类型{action}) async def extract_search_results(self): 专门解析Google搜索结果页 # 等待搜索结果容器加载 await self.page.wait_for_selector(div#search, statevisible, timeout10000) results [] # 定位每个搜索结果块Google的经典结构 result_elements await self.page.query_selector_all(div.g) # 注意此选择器可能随Google改版而变化 for elem in result_elements[:10]: # 取前10个结果 try: title_elem await elem.query_selector(h3) link_elem await elem.query_selector(a) snippet_elem await elem.query_selector(div.VwiC3b) # 摘要的class可能变化 title await title_elem.text_content() if title_elem else N/A url await link_elem.get_attribute(href) if link_elem else N/A snippet await snippet_elem.text_content() if snippet_elem else N/A if title ! N/A and url ! N/A: results.append({ title: title.strip(), url: url, snippet: snippet.strip() if snippet else }) except Exception as e: # 某个结果解析失败跳过 continue return results async def close(self): 关闭浏览器 if self.browser: await self.browser.close() print(浏览器已关闭。)3.3 构建智能体Agent主循环接下来我们创建智能体它将整合驱动层和LLM。import openai # 或其他LLM API客户端 import os class GoogleSearchAgent: def __init__(self, driver, llm_api_key): self.driver driver self.llm_client openai.OpenAI(api_keyllm_api_key) self.conversation_history [] # 记录交互历史 def _build_prompt(self, user_query, page_state): 构建发送给LLM的提示词 # 将交互元素列表格式化为易读的文本 elements_text for elem in page_state[interactive_elements]: desc f- 选择器: {elem[selector]} | 标签: {elem[tag]} if elem[text]: desc f | 文本: {elem[text]} if elem[placeholder]: desc f | 占位符: {elem[placeholder]} elements_text desc \n prompt f 你是一个网页操作智能体正在使用浏览器。你的终极目标是**回答用户的问题**。 当前任务通过操作Google搜索来回答用户的问题。 【当前页面状态】 - 页面标题{page_state[title]} - 页面URL{page_state[url]} - 当前页面上可交互的元素有 {elements_text} 【操作历史】 {self._format_history()} 【用户原始问题】 {user_query} 请分析当前页面和任务决定下一步操作。你**必须且只能**输出一个合法的JSON对象格式如下 1. 如果你需要操作浏览器来获取更多信息 {{action: 操作类型, selector: 元素选择器, value: 可选值}} 可用的操作类型: goto (跳转URL), type (输入文本), click (点击), press (按键)。 例如在Google首页输入搜索词并搜索 - {{action: type, selector: textarea[nameq], value: 人工智能最新进展}} 例如点击搜索按钮 - {{action: click, selector: input[valueGoogle 搜索]}} 例如在输入框按回车 - {{action: press, selector: textarea[nameq], value: Enter}} 2. 如果你认为已经获得足够信息可以回答用户问题 {{action: answer, answer: 你的回答内容请尽量详细并引用来源。}} 请直接输出JSON不要有任何其他文字。 return prompt def _format_history(self): 格式化操作历史 if not self.conversation_history: return 无 return \n.join([f- {h} for h in self.conversation_history[-5:]]) # 只保留最近5条 async def run(self, user_query): 执行主循环 print(f\n 开始处理查询: {user_query} ) # 第一步导航到Google await self.driver.goto_google() max_steps 10 # 防止陷入无限循环 for step in range(max_steps): print(f\n--- 第 {step1} 步 ---) # 1. 感知获取当前页面状态 page_state await self.driver.get_page_state() print(f当前URL: {page_state[url]}) # 2. 思考调用LLM做决策 prompt self._build_prompt(user_query, page_state) try: response self.llm_client.chat.completions.create( modelgpt-4-turbo-preview, # 或 gpt-3.5-turbo messages[{role: user, content: prompt}], temperature0.1, # 低温度保证输出稳定性 max_tokens500 ) llm_output response.choices[0].message.content.strip() print(fLLM原始输出:\n{llm_output}) except Exception as e: print(f调用LLM API失败: {e}) break # 3. 解析LLM输出 try: decision json.loads(llm_output) except json.JSONDecodeError: print(错误LLM未返回有效JSON。) # 可以尝试让LLM修正或直接退出 decision {action: answer, answer: 抱歉我在处理请求时遇到了内部错误。} self.conversation_history.append(f步骤{step1}: LLM返回了无效JSON。) break # 4. 行动执行或回答 if decision.get(action) answer: final_answer decision.get(answer, 未提供答案。) print(f\n 任务完成 \n最终答案: {final_answer}) self.conversation_history.append(f步骤{step1}: 任务完成给出答案。) return final_answer else: # 执行浏览器操作 try: result_msg await self.driver.execute_action(decision) print(f执行动作: {decision} - {result_msg}) self.conversation_history.append(f步骤{step1}: 执行 {decision} - {result_msg}) # 如果是搜索动作可以等待一下结果加载 if decision[action] type and q in decision.get(selector, ): await asyncio.sleep(1) # 等待输入完成 await self.driver.page.press(decision[selector], Enter) self.conversation_history.append(f步骤{step1}附加: 按下Enter键执行搜索。) await asyncio.sleep(2) # 等待搜索结果加载 except Exception as e: error_msg f执行动作失败: {e} print(error_msg) self.conversation_history.append(f步骤{step1}: 动作失败 - {error_msg}) # 将错误信息反馈给下一轮循环让LLM调整策略 # 这里简化处理直接跳出 break # 如果循环结束仍未给出答案 timeout_answer 经过多次尝试未能完成搜索并获取答案。可能是页面结构发生变化或查询过于复杂。 print(f\n 任务超时 \n{timeout_answer}) return timeout_answer3.4 运行你的第一个智能体最后我们写一个主函数把它们串起来。import asyncio from your_driver_module import GoogleSearchDriver # 假设上面的类保存在这个模块 from your_agent_module import GoogleSearchAgent # 假设上面的类保存在这个模块 async def main(): # 1. 初始化驱动开发阶段可设置headlessFalse以便观察 driver GoogleSearchDriver(headlessFalse) await driver.start() # 2. 初始化智能体需要你的OpenAI API Key # 重要请将API Key存储在环境变量中不要硬编码在代码里 api_key os.getenv(OPENAI_API_KEY) if not api_key: print(错误未设置 OPENAI_API_KEY 环境变量。) return agent GoogleSearchAgent(driver, api_key) # 3. 运行一个查询示例 user_question Playwright和Selenium在浏览器自动化上主要区别是什么 answer await agent.run(user_question) # 4. 清理 await driver.close() print(\n智能体运行结束。) if __name__ __main__: asyncio.run(main())运行这段代码你将看到一个浏览器窗口自动打开导航到Google在搜索框中输入你的问题按下回车然后智能体会尝试解析搜索结果页。最终LLM会基于抓取到的搜索结果摘要生成一个综合性的答案。整个过程完全自动化就像有一个隐形的助手在帮你搜索和整理信息。4. 避坑指南与性能优化实战在实际构建和运行过程中我遇到了不少挑战。下面这些经验希望能帮你少走弯路。4.1 元素定位动态网页的“猫鼠游戏”问题Google的页面结构并非一成不变A/B测试、地区差异、登录状态都会导致HTML元素的选择器发生变化。今天还能用的input[nameq]明天可能就变成了textarea[nameq]。解决方案使用多种属性组合不要依赖单一属性。构建选择器时优先使用id其次是name、aria-label、>async def robust_click(self, selectors): 尝试一组选择器直到成功点击一个 for selector in selectors: try: await self.page.click(selector, timeout5000) # 给每个选择器5秒超时 return True except Exception: continue raise Exception(f所有选择器均失败{selectors})定期更新与监控将关键页面的元素选择器作为配置项管理并建立简单的健康检查脚本定期运行测试用例一旦发现失败立即报警并更新选择器。4.2 处理验证码与反爬机制问题频繁的自动化请求可能会触发Google的验证码如reCAPTCHA。应对策略合规前提降低请求频率在动作之间添加随机延迟await asyncio.sleep(random.uniform(1, 3))模拟人类操作节奏。避免在极短时间内进行大量搜索。使用高质量代理IP如果需要在服务器端大规模运行考虑使用住宅代理IP池来分散请求源。设置合理的User-AgentPlaywright启动的浏览器有默认UA可以接受。无需特意伪装但也不要频繁更换。识别到验证码时优雅失败在get_page_state或执行动作后检查页面是否包含验证码的关键词如“captcha”、“我不是机器人”。如果检测到则中止任务并向上层返回明确错误“触发反爬机制”而不是继续硬闯。注意绝对不要尝试自动破解验证码这违反大多数网站的服务条款且从技术伦理上讲也不可取。我们的智能体应设计为在遇到不可逾越的障碍时礼貌地告知用户任务失败。4.3 控制LLM的成本与稳定性问题LLM API调用是按Token收费的且存在速率限制。冗长的页面描述和复杂的思考过程会迅速消耗预算。优化技巧压缩页面信息传给LLM的页面状态信息是成本大头。get_page_state函数中的all_inputs[:20]就是一种压缩。更激进的做法是只提取当前视口内的交互元素或者使用AI视觉模型如GPT-4V分析截图只提取关键文本区域。也可以训练一个轻量级模型来预测哪些元素是当前任务相关的。设置清晰的停止条件在Prompt中明确告诉LLM一旦从搜索结果中找到了足够的信息就立即给出答案不要无意义地翻页或点击不相关的链接。可以限制最大步数如我们代码中的max_steps10。使用更便宜的模型进行“粗筛”可以采用双模型策略。用一个快速、廉价的模型如GPT-3.5 Turbo来解析页面并决定下一步的简单操作如“点击第一个结果”只有当需要复杂阅读理解、信息综合时才调用GPT-4 Turbo。实现缓存对于相同的用户查询和页面状态LLM的决策很可能是相同的。可以建立一个简单的缓存系统如使用functools.lru_cache将(prompt_hash)映射到decision在一定时间内复用决策减少API调用。4.4 错误处理与鲁棒性提升一个健壮的智能体必须能处理各种意外。网络异常与超时Playwright的所有操作都应包裹在try-except中并设置合理的超时时间。对于goto、click等操作可以加入重试逻辑。async def goto_with_retry(self, url, retries3): for i in range(retries): try: await self.page.goto(url, timeout60000) return True except Exception as e: print(f导航失败第{i1}次重试。错误{e}) await asyncio.sleep(2 ** i) # 指数退避 return False页面加载状态判断不要只依赖goto完成。关键操作后使用page.wait_for_selector或page.wait_for_function等待特定元素出现确保页面已加载到可交互状态。LLM输出格式错误代码中虽然有了json.loads的异常捕获但还可以更进一步。如果解析失败可以将错误信息和原始Prompt再次发给LLM要求它修正输出格式。这通常比直接让任务失败更好。结果解析的容错性extract_search_results函数中的选择器div.g是脆弱的。应该增加更多后备选择器并定期检查其有效性。解析每个结果块时使用try-except包裹确保一个结果的解析失败不会影响其他结果。5. 进阶思路与应用场景拓展一个基础的Google搜索智能体已经成型但它的潜力远不止于此。以下是一些可以深入探索的方向5.1 从“搜索”到“任务执行”当前的智能体主要完成“搜索-获取信息”的任务。我们可以将其扩展为更通用的“网页任务执行器”。多步骤操作例如用户指令“在GitHub上找到Playwright的仓库并给我star数”。这需要智能体1) 搜索“Playwright GitHub”2) 点击第一个结果3) 在GitHub页面定位star数元素并提取。表单填写与提交训练智能体理解更复杂的表单结构完成如“在电商网站搜索‘无线鼠标’按价格排序将前三名加入购物车”这类任务。这需要更精细的页面状态表示和动作规划能力。5.2 融合多模态感知目前我们主要依赖HTML结构。加入视觉信息能极大提升智能体对复杂、非标准UI的理解能力。截图分析在get_page_state中除了HTML还可以对页面进行截图并使用多模态大模型如GPT-4V来描述截图内容特别是那些难以用HTML属性描述的元素如图标、验证码图片、复杂图表。坐标点击作为最后的手段当所有基于属性的选择器都失效时可以指示AI描述元素在屏幕上的大致位置如“右上角的红色按钮”然后通过图像识别或坐标估算来点击。但这通常不够精确和稳定。5.3 记忆与学习能力让智能体变得更“聪明”。会话记忆在conversation_history中不仅记录动作还可以记录成功提取的信息片段。这样在后续步骤中LLM可以引用之前获得的信息做出更连贯的决策。选择器学习库建立一个持久化存储记录针对特定网站如Google、GitHub、亚马逊的、经过验证的可靠选择器。智能体在新任务中优先使用库中的选择器如果失败再尝试通用方法成功后还可以更新这个库。5.4 部署与规模化考量要将这个Demo变成可用的服务还需要异步与并发利用asyncio处理多个并发的智能体任务提高资源利用率。容器化使用Docker将整个环境Python、Playwright、浏览器打包确保在不同服务器上运行一致。任务队列对于大量任务使用像Celery或RQ这样的任务队列进行管理实现异步执行和结果回调。监控与日志详细记录每个智能体运行的生命周期每一步的页面状态、LLM决策、执行结果这对于调试和优化至关重要。构建一个稳定可靠的Web Agent是一项充满挑战但也极具成就感的工作。它不仅仅是API的堆砌更是对不确定性环境的理解和适应。从最基础的Google搜索开始逐步增加其感知、决策和执行的能力你会发现一个能够真正理解并操作数字世界的AI助手正在你的手中一点点变为现实。这个过程里最大的收获或许不是最终的成品而是在解决一个个具体问题——比如那个总是定位不到的按钮或者那次意外的验证码——时对自动化边界和AI能力的更深层次思考。