尧图网页设计 
1. 项目概述为开放网络数据采集构建“安全护栏”在AI驱动的数据采集领域我们常常面临一个核心矛盾一方面我们希望智能体Agent能像人类一样在开放、动态、充满不确定性的互联网上自由探索高效收集信息另一方面我们又必须对这种“自由”施加严格的约束确保每一次操作都符合预期、可验证、且失败时不会引发灾难性后果。这就像训练一只嗅觉灵敏的猎犬在复杂的城市环境中执行搜索任务既要让它能自主追踪线索又必须给它戴上可靠的牵引绳和嘴套防止它跑丢、误伤他人或吞下有害物品。“Making Failure Safe: A Constrained, Verifiable Agent Framework for Open-Web Data Collection”这个项目正是为了解决这一矛盾而生。它不是一个简单的网络爬虫而是一个为大型语言模型LLM驱动的智能体设计的“安全操作框架”。其核心目标是让失败变得安全、可预测、可管理。在开放网络Open-Web这个“战场”上链接可能失效、网站结构可能突变、反爬策略层出不穷、甚至目标信息本身就可能不存在。传统的脚本要么一错全停要么在静默中采集到一堆垃圾数据。而这个框架通过引入“约束Constrained”和“可验证Verifiable”两大支柱为智能体的每一次行动划定了安全边界并提供了实时验伤的能力。它非常适合以下几类人希望将LLM能力应用于自动化、规模化数据采集的开发者需要从大量异构网站中提取结构化数据但对数据质量和过程可靠性有严苛要求的数据工程师以及任何厌倦了编写和维护脆弱、难以调试的定制爬虫脚本的技术团队。这个框架将采集逻辑由LLM理解与安全策略由框架强制执行解耦使得构建一个既强大又稳健的采集智能体变得像组装乐高积木一样清晰可控。2. 框架核心设计哲学约束、验证与韧性这个框架的设计不是从工具选型开始的而是从一系列核心原则出发的。理解这些原则比直接看代码更重要。2.1 为什么“约束”比“能力”更优先在AI智能体设计中一个常见的误区是过度追求其“自主能力”希望它能处理所有边界情况。但在开放网络场景下边界情况几乎是无限的。因此本框架采用了截然不同的思路首先定义智能体“绝对不能做什么”和“必须在什么范围内做”其次才是它“能做什么”。行动空间约束智能体不被允许执行任意的浏览器操作或网络请求。框架会提供一个有限的、经过审核的操作集Action Set例如navigate_to(url),extract_text(selector),click_button(selector),fill_form(selector, data)等。智能体只能从这些预设动作中选择和组合。这从根本上杜绝了它尝试访问恶意链接、执行危险JavaScript代码或进行未授权的POST请求。导航范围约束通过域名白名单、URL模式正则匹配等方式将智能体的活动范围严格限制在目标网站或可信域内。防止采集过程中“迷路”到无关或风险站点。资源消耗约束对单次任务的执行时间、网络请求次数、内存/CPU使用率设置硬性上限。一旦触及任务被优雅地终止并标记为“资源超限”而不是耗尽服务器资源。这种约束设计实质上是为LLM的“创造力”套上了一个安全的沙箱Sandbox。LLM负责在沙箱内制定最优策略而沙箱的墙壁保证了外部环境的安全。2.2 “可验证”如何贯穿采集生命周期验证不是事后的数据清洗而是贯穿采集行动前、中、后的连续过程。行动前验证计划验证智能体LLM根据任务目标如“获取某产品价格和评论”会生成一个初步的行动计划通常表示为一系列步骤。框架会首先对这个计划进行静态验证。例如检查计划中的URL是否在许可范围内计划使用的操作是否在允许的动作集中步骤间是否存在明显的逻辑循环或矛盾。这可以提前拦截掉大量不靠谱的“幻想”计划。行动中验证状态验证在每一步操作执行后框架会立即检查结果状态。这不仅仅是看HTTP状态码是否为200。它包括内容验证使用XPath、CSS选择器或正则表达式快速检查关键元素是否出现在响应中。如果寻找的“价格”标签消失了这一步就算失败。结构验证检查返回的HTML或JSON结构是否符合预期的大致模样防止因为页面完全跳转如跳到登录页或404页而导致后续步骤在错误的上下文中执行。业务规则验证集成简单的规则引擎例如“提取的价格数值必须大于0”“发布日期不能晚于今天”。这些规则可以即时过滤掉明显的脏数据。行动后验证输出验证这是最关键的一环即对智能体最终产出的结构化数据通常是JSON进行模式Schema验证。框架强制要求每个采集任务都必须对应一个严格的JSON Schema。在智能体宣称完成任务、输出JSON后框架会用这个Schema去校验数据的完整性、类型正确性和值域有效性。只有通过校验的数据才会被最终接纳。注意这里的验证逻辑不应再交给LLM去做“请你判断一下这个数据对不对”而必须使用确定性的、程序化的规则如JSON Schema验证库、正则表达式。LLM用于理解和生成框架用于校验和保障这是职责分离的关键。2.3 韧性设计让失败成为信息源传统系统的失败往往是二元的成功/失败且失败意味着流程终止。本框架将失败视为一种常态和有用的信息。分级失败处理定义不同级别的失败。例如轻微偏离页面布局微调导致某个选择器失效但备用选择器生效。任务继续记录警告。可恢复错误遇到验证码或网络临时错误。触发重试机制带指数退避或切换到备用数据源。预期内失败目标信息确实不存在如“缺货”。任务“成功”结束但输出中包含“NOT_FOUND”等特殊标记。不可恢复错误域名无法解析、网站结构彻底重构。任务终止并标记为特定错误类型触发人工审核或规则更新流程。状态快照与回放智能体的每一步操作、页面状态、验证结果都被详细记录。当任务失败时开发者可以完整回放Replay整个会话精确看到智能体在哪一步、基于什么页面状态、做出了什么决策、然后遇到了什么验证失败。这极大地简化了调试过程。失败驱动的自适应框架可以配置为当某种类型的失败在一定时间内频繁出现时自动将相关任务标记为“疑似规则过期”并通知维护人员或触发一个自动化的规则学习流程。3. 架构拆解从LLM指令到可信JSON的流水线理解了设计哲学我们来看具体的实现架构。整个框架可以看作一条高度自动化的流水线我将结合当前流行的技术栈来阐述一个可落地的实现方案。3.1 核心组件交互图概念层整个系统围绕“任务执行引擎”展开其与LLM、验证器、约束器的交互流程如下任务解析器接收一个高层级自然语言指令如“监测电商网站X上品牌Y所有手机的价格波动”或一个结构化任务描述。将其与对应的约束配置文件定义URL范围、允许动作、资源限制和输出JSON Schema绑定形成一个可执行的“任务包”。规划器LLM驱动任务包首先被送到规划器。规划器通常是一个提示词Prompt精心设计的LLM调用。Prompt中会包含任务目标、当前上下文如之前采集的数据、允许的动作列表、以及几个规划示例。LLM的输出必须是一个JSON格式的行动计划例如{steps: [{action: navigate_to, args: {url: ...}}, ...]}。这个计划会立即经过计划验证器的静态检查。执行器与状态管理器验证通过的计划交给执行器。执行器是一个封装了浏览器自动化如Playwright或HTTP客户端能力的模块。它严格按计划执行动作并在每一步后更新“世界状态”当前URL、页面HTML快照、提取的中间数据等。状态管理器负责维护这个状态。验证器集群这是框架的“免疫系统”。在每一步执行后运行时验证器被触发检查上一步的结果是否符合预期内容/结构验证。所有步骤执行完毕后输出验证器基于JSON Schema对最终产出的数据进行终极校验。裁决器根据验证结果和约束检查裁决器决定下一步动作继续执行下一步、重试当前步、切换到备用计划、还是宣告失败并分类。它将决策反馈给执行器或规划器进行重规划。3.2 关键技术栈选型与理由LLM层核心模型优先选择在推理、规划和指令跟随方面表现优秀的模型如GPT-4、Claude 3系列或开源的DeepSeek-V2、Qwen2.5系列。对于成本敏感的场景可以使用较小的模型如Qwen2.5-7B进行动作规划而用大模型进行复杂的页面内容理解。关键技巧必须使用结构化输出JSON Mode功能。在调用LLM API时强制指定返回格式为JSON并提供一个与计划Schema匹配的说明。这能极大提高输出解析的稳定性和可靠性。执行与自动化层首选Playwright相较于SeleniumPlaywright对现代Web技术支持更好自动等待机制更智能且能更可靠地捕获动态内容。它支持多浏览器并能生成用于调试的追踪记录Trace这与框架的“状态回放”理念完美契合。轻量级备选对于纯API接口或服务端渲染SSR页面可使用httpx或aiohttp等异步HTTP客户端搭配parselScrapy的选择器库进行解析速度更快。验证与约束层JSON Schema验证jsonschema库是Python生态的标准功能强大。定义Schema时要尽可能严格利用required,type,pattern,enum,minimum/maximum等关键字。运行时内容验证parsel库的XPath和CSS选择器速度快、表达能力强。可以预先为目标页面的关键元素定义一组“验证选择器”执行后立刻检查是否存在。编排与调度层Airflow DAG为什么是Airflow因为数据采集任务本质上是工作流有依赖先登录再采集、需要调度每日执行、要处理重试和失败告警。Airflow的DAG有向无环图能直观地表达这些任务流。DAG设计模式每个采集站点或品类可以是一个DAG。DAG中的每个Task不是直接执行采集代码而是向框架的任务队列提交一个“任务包”。框架自身的执行引擎作为常驻服务从队列中消费任务并执行。这样实现了调度与执行解耦执行器可以水平扩展。3.3 一个简单的约束与Schema定义示例# task_config.yaml - 约束配置 task_id: monitor_phone_price constraints: allowed_domains: - trusted-store.com - api.trusted-store.com allowed_actions: - navigate - extract_text - extract_attribute - screenshot # 用于失败调试 resource_limits: max_execution_time_seconds: 120 max_network_requests: 20 navigation_policy: stay_within_domain # 禁止跨域 validation: runtime_selectors: price_element: span[data-testidproduct-price] product_title_element: h1.product-title output_schema_ref: ./schemas/phone_price_schema.json// phone_price_schema.json - 输出数据模式 { $schema: http://json-schema.org/draft-07/schema#, type: object, required: [product_id, product_name, current_price, currency, timestamp, availability], properties: { product_id: { type: string, pattern: ^[A-Z0-9-]$ }, product_name: { type: string, minLength: 1, maxLength: 200 }, current_price: { type: number, minimum: 0 }, currency: { type: string, enum: [USD, EUR, CNY] }, timestamp: { type: string, format: date-time }, availability: { type: string, enum: [IN_STOCK, OUT_OF_STOCK, PRE_ORDER] }, discount: { type: [number, null], minimum: 0, maximum: 100 } }, additionalProperties: false // 严禁输出任何未定义的字段 }这个Schema定义得非常严格字段必填、类型明确、值域受限、甚至拒绝额外字段。这确保了输出数据质量的一致性方便下游系统直接消费。4. 实操构建从零搭建一个安全的价格监测智能体现在我们动手搭建一个具体的实例一个监测某电商网站手机价格的安全智能体。我们将使用Python、Playwright和OpenAI API或兼容的开源模型来实现核心逻辑。4.1 环境准备与基础模块搭建首先创建项目结构并安装依赖。# 项目目录结构 safe-web-agent/ ├── config/ │ ├── tasks/ # 存放各任务的约束配置YAML │ └── schemas/ # 存放各任务的输出JSON Schema ├── core/ │ ├── agent.py # 智能体核心规划、执行、学习循环 │ ├── constraints.py # 约束检查器 │ ├── verifier.py # 验证器运行时输出 │ ├── executor.py # 执行器Playwright封装 │ └── state.py # 状态管理器 ├── llm/ │ └── client.py # LLM客户端封装 ├── tasks/ │ └── phone_price_monitor.py # 具体任务定义 ├── utils/ ├── requirements.txt └── main.py # 服务入口或脚本入口requirements.txt关键依赖openai1.0.0 # 或 litellm, openai兼容库 playwright1.40.0 jsonschema4.20.0 pydantic2.0.0 # 用于数据模型验证作为jsonschema的补充 pyyaml6.0 asyncio安装Playwright浏览器pip install -r requirements.txt playwright install chromium4.2 核心模块代码实现要点1. 约束检查器 (core/constraints.py)这个模块负责加载YAML配置并在任务执行前、中、后施加硬性限制。import yaml from urllib.parse import urlparse import time class ConstraintChecker: def __init__(self, config_path): with open(config_path, r) as f: self.config yaml.safe_load(f) self.request_count 0 self.start_time None def check_navigation(self, url): 检查目标URL是否在允许的域名内 parsed urlparse(url) allowed self.config[constraints][allowed_domains] if parsed.netloc not in allowed: raise SecurityViolationError(fNavigation to disallowed domain: {parsed.netloc}) def check_action(self, action_name): 检查动作是否在允许列表中 allowed self.config[constraints][allowed_actions] if action_name not in allowed: raise SecurityViolationError(fAction not allowed: {action_name}) def start_execution(self): 开始执行记录资源消耗起点 self.start_time time.time() self.request_count 0 def record_request(self): 记录一次网络请求 self.request_count 1 max_req self.config[constraints][resource_limits][max_network_requests] if self.request_count max_req: raise ResourceLimitError(fExceeded max network requests: {max_req}) def check_time(self): 检查是否超时 if self.start_time: elapsed time.time() - self.start_time max_time self.config[constraints][resource_limits][max_execution_time_seconds] if elapsed max_time: raise ResourceLimitError(fExceeded max execution time: {max_time}s)2. LLM客户端与规划器 (llm/client.py和core/agent.py部分)规划器的核心是构造一个能引导LLM输出结构化计划的Prompt。# llm/client.py from openai import OpenAI import json class LLMClient: def __init__(self, modelgpt-4-turbo, api_keyNone, base_urlNone): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model async def create_plan(self, task_description, current_state, allowed_actions): prompt f 你是一个网络数据采集智能体。你的目标{task_description} 当前状态 - 当前URL: {current_state.get(url, N/A)} - 已提取数据: {json.dumps(current_state.get(extracted, {}), indent2)} 你可以执行以下动作 {json.dumps(allowed_actions, indent2)} 请生成一个JSON格式的行动计划来完成任务。计划应是一个步骤列表每个步骤包含“action”和“args”。 只使用上面允许的动作。如果当前页面已有需要的数据动作可以是“extract_data”。 输出必须是有效的JSON格式如下 {{ steps: [ {{action: action_name, args: {{arg1: value1}}}}, ... ] }} response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], response_format{type: json_object}, # 关键强制JSON输出 temperature0.1, # 低随机性保证稳定性 ) plan_json json.loads(response.choices[0].message.content) return plan_json3. 执行器与验证器的集成 (core/executor.py和core/verifier.py)执行器在每一步动作后都会调用验证器进行状态检查。# core/verifier.py import jsonschema from parsel import Selector class RuntimeVerifier: def __init__(self, runtime_selectors_config): self.selectors runtime_selectors_config def verify(self, html_content, step_name): 根据步骤名检查页面是否包含关键元素 selector self.selectors.get(step_name) if not selector: return True, No selector defined for this step sel Selector(texthtml_content) if sel.css(selector).get(): return True, Key element found else: return False, fKey element not found with selector: {selector} class OutputVerifier: def __init__(self, schema_path): with open(schema_path, r) as f: self.schema json.load(f) def verify(self, data): try: jsonschema.validate(instancedata, schemaself.schema) return True, None except jsonschema.ValidationError as e: return False, fSchema validation failed: {e.message}# core/executor.py (片段) async def execute_step(self, step, state, constraint_checker, runtime_verifier): action step[action] args step.get(args, {}) # 1. 检查动作约束 constraint_checker.check_action(action) # 2. 执行动作 if action navigate_to: url args[url] constraint_checker.check_navigation(url) # 导航前检查URL await self.page.goto(url, wait_untilnetworkidle) state[url] url constraint_checker.record_request() # 记录请求 elif action extract_text: selector args[selector] elements await self.page.locator(selector).all() values [await el.text_content() for el in elements] state[extracted][args[key]] values # ... 其他动作处理 # 3. 获取执行后状态HTML快照 html await self.page.content() state[html_snapshot] html # 4. 运行时验证 is_valid, msg runtime_verifier.verify(html, step.get(name, action)) if not is_valid: raise RuntimeVerificationError(fStep {action} failed runtime verification: {msg}) # 5. 检查资源约束 constraint_checker.check_time() return state4.3 组装智能体与任务循环 (core/agent.py)这是粘合所有组件的“大脑”。class ConstrainedVerifiableAgent: def __init__(self, task_config_path, llm_client): self.constraints ConstraintChecker(task_config_path) self.llm llm_client self.runtime_verifier RuntimeVerifier(self._load_runtime_selectors(task_config_path)) self.output_verifier OutputVerifier(self._load_schema_path(task_config_path)) self.executor PlaywrightExecutor() self.state {url: None, extracted: {}, history: []} async def run(self, task_description): self.constraints.start_execution() await self.executor.start() try: while not self._is_task_complete(): # 1. 规划 allowed_acts self.constraints.config[constraints][allowed_actions] plan await self.llm.create_plan(task_description, self.state, allowed_acts) # 静态计划验证简化示例检查步骤数 if len(plan.get(steps, [])) 10: raise PlanningError(Plan too long, potential loop.) # 2. 执行与验证循环 for step in plan[steps]: self.state await self.executor.execute_step( step, self.state, self.constraints, self.runtime_verifier ) self.state[history].append(step) # 3. 判断是否完成LLM或规则 if self._has_target_data(self.state): break # 否则基于新状态进行下一轮规划... # 4. 最终输出验证 final_data self._format_output(self.state) is_valid, error self.output_verifier.verify(final_data) if not is_valid: raise OutputValidationError(error) return {status: success, data: final_data} except (SecurityViolationError, ResourceLimitError, RuntimeVerificationError) as e: return {status: failed, error_type: type(e).__name__, message: str(e), state_snapshot: self.state} finally: await self.executor.close()4.4 集成到Airflow DAG最后我们将这个智能体包装成一个可以在Airflow中调用的算子。# tasks/phone_price_monitor.py from airflow.decorators import dag, task from datetime import datetime from core.agent import ConstrainedVerifiableAgent from llm.client import LLMClient default_args { owner: data_team, retries: 2, retry_delay: timedelta(minutes5), } dag( dag_idsafe_phone_price_monitor, default_argsdefault_args, schedule_interval0 */6 * * *, # 每6小时执行一次 start_datedatetime(2024, 1, 1), catchupFalse, ) def create_dag(): task def monitor_product(product_url): 单个产品监测任务 llm_client LLMClient(modelgpt-4-turbo) agent ConstrainedVerifiableAgent( task_config_pathconfig/tasks/phone_monitor.yaml, llm_clientllm_client ) task_desc fNavigate to {product_url} and extract the current price, product name, and availability status. result asyncio.run(agent.run(task_desc)) if result[status] success: # 将成功数据推送到数据仓库或消息队列 send_to_data_warehouse(result[data]) return result[data] else: # 将失败详情发送到告警系统如Slack、PagerDuty send_alert(fTask failed: {result[error_type]} - {result[message]}) # Airflow会将此标记为失败任务并根据DAG设置重试 raise ValueError(fAgent execution failed: {result}) # 假设我们有一个产品URL列表 product_urls get_product_urls_from_catalog() # 使用Airflow的动态任务映射为每个URL创建一个并行任务 monitor_product.expand(product_urlproduct_urls) # 创建DAG实例 phone_price_dag create_dag()这个DAG会并行启动多个智能体实例每个实例独立、安全地监测一个产品页面。任何一个实例的失败如触犯约束、验证失败都不会影响其他实例并且其完整的错误信息和状态快照会被记录下来用于排查。5. 避坑指南与实战心得在实际部署和运行此类框架时我踩过不少坑也积累了一些让系统更稳健的经验。5.1 LLM提示工程中的稳定性陷阱问题LLM有时会“幻想”出配置中不存在的动作或者输出格式不符合JSON要求。对策结构化输出是生命线务必使用LLM API提供的response_format{type: json_object}参数。对于不支持此功能的模型在Prompt的开头和结尾用json ...明确标记并在后处理中使用严格的json.loads()并捕获异常。少样本示例Few-Shot至关重要在Prompt中提供2-3个清晰、正确的规划示例。示例应涵盖正常流程和边界情况如“未找到元素”。温度Temperature设置规划阶段务必使用低温度如0.1-0.3以减少随机性。只有在需要创意性解决问题时如选择器失效后的备选方案生成才适当调高。设置重规划Re-plan机制当LLM输出的计划无法通过静态验证或连续几步执行失败时应将当前状态和错误信息反馈给LLM要求它重新规划。这通常比简单的重试更有效。5.2 验证逻辑的设计平衡问题验证规则太松垃圾数据会漏过太紧则误杀率高任务频繁失败。对策分层验证采用“宽松运行时验证 严格输出验证”策略。运行时验证只检查页面是否发生根本性错误如跳转到登录页、关键容器元素消失。而数据是否完整、格式是否正确交给最终的JSON Schema校验。这样避免了因页面微小改动导致任务中途失败。定义“软失败”状态不是所有验证失败都应导致任务终止。例如如果“折扣价”元素没找到但“原价”找到了可以将其记录为null并打上discount_missing的标记让任务继续并成功结束。下游系统可以处理这种部分数据。定期更新验证规则将运行时验证的选择器作为配置管理。可以建立一个简单的版本控制或A/B测试机制当某个选择器失败率持续升高时自动尝试切换到备选选择器。5.3 性能、成本与扩展性问题LLM API调用成本高Playwright执行速度慢大规模采集时资源消耗大。对策缓存LLM响应对于相同的任务描述和页面状态其最优计划很可能是相同的。可以使用Redis或内存缓存对LLM的规划请求进行缓存缓存键由任务描述页面关键特征哈希构成。分离“探索”与“采集”对于定期执行的重复任务如每日价格监测第一次运行时用LLM进行“探索”生成一个稳定的行动计划序列如导航路径、选择器。之后的任务可以直接复用这个“剧本”无需再次调用LLM除非验证失败触发重规划。这能节省90%以上的LLM成本。无头浏览器池使用playwright-pool或自定义连接池管理多个浏览器实例避免为每个任务启动/关闭浏览器的开销。结合异步IO可以显著提高吞吐量。降级策略准备一个基于规则Rule-based的备用提取器。当LLM智能体连续失败或遇到已知的、结构简单的页面时可以降级到使用预定义规则和XPath进行采集更快速、更稳定。5.4 调试与监控问题智能体在黑盒中失败难以定位是LLM规划问题、页面变化问题还是验证规则问题。对策全链路追踪为每个任务生成唯一的trace_id并将每一步的输入状态/计划、执行动作、页面快照HTML或截图、验证结果都记录到可查询的存储中如Elasticsearch。Playwright的trace.playwright.dev功能可以完美集成。构建调试回放界面开发一个简单的Web界面输入trace_id就能像播放视频一样逐步回放智能体的整个执行过程查看每一步的页面截图和LLM的“思考过程”规划输出。这是排查问题最强大的工具。定义关键指标监控任务成功率、平均执行时间、LLM调用次数、各类验证失败率运行时/输出、约束触发次数。设置告警当成功率下降或LLM成本异常升高时及时通知。构建一个“失败安全”的智能体框架初期投入在设计和基础设施上的精力会比较多但一旦建成它带来的长期收益是巨大的采集流程的可靠性从“碰运气”变成了“可度量、可管理”。你不再需要每天忙于处理各种奇怪的爬虫崩溃而是可以专注于定义更复杂的采集目标和优化数据质量规则。这套框架的核心思想——通过约束限制行动范围通过验证确保每一步的正确性——不仅适用于网络采集对于任何需要LLM与不确定环境交互的自动化任务如软件操作、机器人流程自动化RPA都具有重要的借鉴意义。 构建安全可靠的LLM智能体框架：约束、验证与韧性设计实践
1. 项目概述为开放网络数据采集构建“安全护栏”在AI驱动的数据采集领域我们常常面临一个核心矛盾一方面我们希望智能体Agent能像人类一样在开放、动态、充满不确定性的互联网上自由探索高效收集信息另一方面我们又必须对这种“自由”施加严格的约束确保每一次操作都符合预期、可验证、且失败时不会引发灾难性后果。这就像训练一只嗅觉灵敏的猎犬在复杂的城市环境中执行搜索任务既要让它能自主追踪线索又必须给它戴上可靠的牵引绳和嘴套防止它跑丢、误伤他人或吞下有害物品。“Making Failure Safe: A Constrained, Verifiable Agent Framework for Open-Web Data Collection”这个项目正是为了解决这一矛盾而生。它不是一个简单的网络爬虫而是一个为大型语言模型LLM驱动的智能体设计的“安全操作框架”。其核心目标是让失败变得安全、可预测、可管理。在开放网络Open-Web这个“战场”上链接可能失效、网站结构可能突变、反爬策略层出不穷、甚至目标信息本身就可能不存在。传统的脚本要么一错全停要么在静默中采集到一堆垃圾数据。而这个框架通过引入“约束Constrained”和“可验证Verifiable”两大支柱为智能体的每一次行动划定了安全边界并提供了实时验伤的能力。它非常适合以下几类人希望将LLM能力应用于自动化、规模化数据采集的开发者需要从大量异构网站中提取结构化数据但对数据质量和过程可靠性有严苛要求的数据工程师以及任何厌倦了编写和维护脆弱、难以调试的定制爬虫脚本的技术团队。这个框架将采集逻辑由LLM理解与安全策略由框架强制执行解耦使得构建一个既强大又稳健的采集智能体变得像组装乐高积木一样清晰可控。2. 框架核心设计哲学约束、验证与韧性这个框架的设计不是从工具选型开始的而是从一系列核心原则出发的。理解这些原则比直接看代码更重要。2.1 为什么“约束”比“能力”更优先在AI智能体设计中一个常见的误区是过度追求其“自主能力”希望它能处理所有边界情况。但在开放网络场景下边界情况几乎是无限的。因此本框架采用了截然不同的思路首先定义智能体“绝对不能做什么”和“必须在什么范围内做”其次才是它“能做什么”。行动空间约束智能体不被允许执行任意的浏览器操作或网络请求。框架会提供一个有限的、经过审核的操作集Action Set例如navigate_to(url),extract_text(selector),click_button(selector),fill_form(selector, data)等。智能体只能从这些预设动作中选择和组合。这从根本上杜绝了它尝试访问恶意链接、执行危险JavaScript代码或进行未授权的POST请求。导航范围约束通过域名白名单、URL模式正则匹配等方式将智能体的活动范围严格限制在目标网站或可信域内。防止采集过程中“迷路”到无关或风险站点。资源消耗约束对单次任务的执行时间、网络请求次数、内存/CPU使用率设置硬性上限。一旦触及任务被优雅地终止并标记为“资源超限”而不是耗尽服务器资源。这种约束设计实质上是为LLM的“创造力”套上了一个安全的沙箱Sandbox。LLM负责在沙箱内制定最优策略而沙箱的墙壁保证了外部环境的安全。2.2 “可验证”如何贯穿采集生命周期验证不是事后的数据清洗而是贯穿采集行动前、中、后的连续过程。行动前验证计划验证智能体LLM根据任务目标如“获取某产品价格和评论”会生成一个初步的行动计划通常表示为一系列步骤。框架会首先对这个计划进行静态验证。例如检查计划中的URL是否在许可范围内计划使用的操作是否在允许的动作集中步骤间是否存在明显的逻辑循环或矛盾。这可以提前拦截掉大量不靠谱的“幻想”计划。行动中验证状态验证在每一步操作执行后框架会立即检查结果状态。这不仅仅是看HTTP状态码是否为200。它包括内容验证使用XPath、CSS选择器或正则表达式快速检查关键元素是否出现在响应中。如果寻找的“价格”标签消失了这一步就算失败。结构验证检查返回的HTML或JSON结构是否符合预期的大致模样防止因为页面完全跳转如跳到登录页或404页而导致后续步骤在错误的上下文中执行。业务规则验证集成简单的规则引擎例如“提取的价格数值必须大于0”“发布日期不能晚于今天”。这些规则可以即时过滤掉明显的脏数据。行动后验证输出验证这是最关键的一环即对智能体最终产出的结构化数据通常是JSON进行模式Schema验证。框架强制要求每个采集任务都必须对应一个严格的JSON Schema。在智能体宣称完成任务、输出JSON后框架会用这个Schema去校验数据的完整性、类型正确性和值域有效性。只有通过校验的数据才会被最终接纳。注意这里的验证逻辑不应再交给LLM去做“请你判断一下这个数据对不对”而必须使用确定性的、程序化的规则如JSON Schema验证库、正则表达式。LLM用于理解和生成框架用于校验和保障这是职责分离的关键。2.3 韧性设计让失败成为信息源传统系统的失败往往是二元的成功/失败且失败意味着流程终止。本框架将失败视为一种常态和有用的信息。分级失败处理定义不同级别的失败。例如轻微偏离页面布局微调导致某个选择器失效但备用选择器生效。任务继续记录警告。可恢复错误遇到验证码或网络临时错误。触发重试机制带指数退避或切换到备用数据源。预期内失败目标信息确实不存在如“缺货”。任务“成功”结束但输出中包含“NOT_FOUND”等特殊标记。不可恢复错误域名无法解析、网站结构彻底重构。任务终止并标记为特定错误类型触发人工审核或规则更新流程。状态快照与回放智能体的每一步操作、页面状态、验证结果都被详细记录。当任务失败时开发者可以完整回放Replay整个会话精确看到智能体在哪一步、基于什么页面状态、做出了什么决策、然后遇到了什么验证失败。这极大地简化了调试过程。失败驱动的自适应框架可以配置为当某种类型的失败在一定时间内频繁出现时自动将相关任务标记为“疑似规则过期”并通知维护人员或触发一个自动化的规则学习流程。3. 架构拆解从LLM指令到可信JSON的流水线理解了设计哲学我们来看具体的实现架构。整个框架可以看作一条高度自动化的流水线我将结合当前流行的技术栈来阐述一个可落地的实现方案。3.1 核心组件交互图概念层整个系统围绕“任务执行引擎”展开其与LLM、验证器、约束器的交互流程如下任务解析器接收一个高层级自然语言指令如“监测电商网站X上品牌Y所有手机的价格波动”或一个结构化任务描述。将其与对应的约束配置文件定义URL范围、允许动作、资源限制和输出JSON Schema绑定形成一个可执行的“任务包”。规划器LLM驱动任务包首先被送到规划器。规划器通常是一个提示词Prompt精心设计的LLM调用。Prompt中会包含任务目标、当前上下文如之前采集的数据、允许的动作列表、以及几个规划示例。LLM的输出必须是一个JSON格式的行动计划例如{steps: [{action: navigate_to, args: {url: ...}}, ...]}。这个计划会立即经过计划验证器的静态检查。执行器与状态管理器验证通过的计划交给执行器。执行器是一个封装了浏览器自动化如Playwright或HTTP客户端能力的模块。它严格按计划执行动作并在每一步后更新“世界状态”当前URL、页面HTML快照、提取的中间数据等。状态管理器负责维护这个状态。验证器集群这是框架的“免疫系统”。在每一步执行后运行时验证器被触发检查上一步的结果是否符合预期内容/结构验证。所有步骤执行完毕后输出验证器基于JSON Schema对最终产出的数据进行终极校验。裁决器根据验证结果和约束检查裁决器决定下一步动作继续执行下一步、重试当前步、切换到备用计划、还是宣告失败并分类。它将决策反馈给执行器或规划器进行重规划。3.2 关键技术栈选型与理由LLM层核心模型优先选择在推理、规划和指令跟随方面表现优秀的模型如GPT-4、Claude 3系列或开源的DeepSeek-V2、Qwen2.5系列。对于成本敏感的场景可以使用较小的模型如Qwen2.5-7B进行动作规划而用大模型进行复杂的页面内容理解。关键技巧必须使用结构化输出JSON Mode功能。在调用LLM API时强制指定返回格式为JSON并提供一个与计划Schema匹配的说明。这能极大提高输出解析的稳定性和可靠性。执行与自动化层首选Playwright相较于SeleniumPlaywright对现代Web技术支持更好自动等待机制更智能且能更可靠地捕获动态内容。它支持多浏览器并能生成用于调试的追踪记录Trace这与框架的“状态回放”理念完美契合。轻量级备选对于纯API接口或服务端渲染SSR页面可使用httpx或aiohttp等异步HTTP客户端搭配parselScrapy的选择器库进行解析速度更快。验证与约束层JSON Schema验证jsonschema库是Python生态的标准功能强大。定义Schema时要尽可能严格利用required,type,pattern,enum,minimum/maximum等关键字。运行时内容验证parsel库的XPath和CSS选择器速度快、表达能力强。可以预先为目标页面的关键元素定义一组“验证选择器”执行后立刻检查是否存在。编排与调度层Airflow DAG为什么是Airflow因为数据采集任务本质上是工作流有依赖先登录再采集、需要调度每日执行、要处理重试和失败告警。Airflow的DAG有向无环图能直观地表达这些任务流。DAG设计模式每个采集站点或品类可以是一个DAG。DAG中的每个Task不是直接执行采集代码而是向框架的任务队列提交一个“任务包”。框架自身的执行引擎作为常驻服务从队列中消费任务并执行。这样实现了调度与执行解耦执行器可以水平扩展。3.3 一个简单的约束与Schema定义示例# task_config.yaml - 约束配置 task_id: monitor_phone_price constraints: allowed_domains: - trusted-store.com - api.trusted-store.com allowed_actions: - navigate - extract_text - extract_attribute - screenshot # 用于失败调试 resource_limits: max_execution_time_seconds: 120 max_network_requests: 20 navigation_policy: stay_within_domain # 禁止跨域 validation: runtime_selectors: price_element: span[data-testidproduct-price] product_title_element: h1.product-title output_schema_ref: ./schemas/phone_price_schema.json// phone_price_schema.json - 输出数据模式 { $schema: http://json-schema.org/draft-07/schema#, type: object, required: [product_id, product_name, current_price, currency, timestamp, availability], properties: { product_id: { type: string, pattern: ^[A-Z0-9-]$ }, product_name: { type: string, minLength: 1, maxLength: 200 }, current_price: { type: number, minimum: 0 }, currency: { type: string, enum: [USD, EUR, CNY] }, timestamp: { type: string, format: date-time }, availability: { type: string, enum: [IN_STOCK, OUT_OF_STOCK, PRE_ORDER] }, discount: { type: [number, null], minimum: 0, maximum: 100 } }, additionalProperties: false // 严禁输出任何未定义的字段 }这个Schema定义得非常严格字段必填、类型明确、值域受限、甚至拒绝额外字段。这确保了输出数据质量的一致性方便下游系统直接消费。4. 实操构建从零搭建一个安全的价格监测智能体现在我们动手搭建一个具体的实例一个监测某电商网站手机价格的安全智能体。我们将使用Python、Playwright和OpenAI API或兼容的开源模型来实现核心逻辑。4.1 环境准备与基础模块搭建首先创建项目结构并安装依赖。# 项目目录结构 safe-web-agent/ ├── config/ │ ├── tasks/ # 存放各任务的约束配置YAML │ └── schemas/ # 存放各任务的输出JSON Schema ├── core/ │ ├── agent.py # 智能体核心规划、执行、学习循环 │ ├── constraints.py # 约束检查器 │ ├── verifier.py # 验证器运行时输出 │ ├── executor.py # 执行器Playwright封装 │ └── state.py # 状态管理器 ├── llm/ │ └── client.py # LLM客户端封装 ├── tasks/ │ └── phone_price_monitor.py # 具体任务定义 ├── utils/ ├── requirements.txt └── main.py # 服务入口或脚本入口requirements.txt关键依赖openai1.0.0 # 或 litellm, openai兼容库 playwright1.40.0 jsonschema4.20.0 pydantic2.0.0 # 用于数据模型验证作为jsonschema的补充 pyyaml6.0 asyncio安装Playwright浏览器pip install -r requirements.txt playwright install chromium4.2 核心模块代码实现要点1. 约束检查器 (core/constraints.py)这个模块负责加载YAML配置并在任务执行前、中、后施加硬性限制。import yaml from urllib.parse import urlparse import time class ConstraintChecker: def __init__(self, config_path): with open(config_path, r) as f: self.config yaml.safe_load(f) self.request_count 0 self.start_time None def check_navigation(self, url): 检查目标URL是否在允许的域名内 parsed urlparse(url) allowed self.config[constraints][allowed_domains] if parsed.netloc not in allowed: raise SecurityViolationError(fNavigation to disallowed domain: {parsed.netloc}) def check_action(self, action_name): 检查动作是否在允许列表中 allowed self.config[constraints][allowed_actions] if action_name not in allowed: raise SecurityViolationError(fAction not allowed: {action_name}) def start_execution(self): 开始执行记录资源消耗起点 self.start_time time.time() self.request_count 0 def record_request(self): 记录一次网络请求 self.request_count 1 max_req self.config[constraints][resource_limits][max_network_requests] if self.request_count max_req: raise ResourceLimitError(fExceeded max network requests: {max_req}) def check_time(self): 检查是否超时 if self.start_time: elapsed time.time() - self.start_time max_time self.config[constraints][resource_limits][max_execution_time_seconds] if elapsed max_time: raise ResourceLimitError(fExceeded max execution time: {max_time}s)2. LLM客户端与规划器 (llm/client.py和core/agent.py部分)规划器的核心是构造一个能引导LLM输出结构化计划的Prompt。# llm/client.py from openai import OpenAI import json class LLMClient: def __init__(self, modelgpt-4-turbo, api_keyNone, base_urlNone): self.client OpenAI(api_keyapi_key, base_urlbase_url) self.model model async def create_plan(self, task_description, current_state, allowed_actions): prompt f 你是一个网络数据采集智能体。你的目标{task_description} 当前状态 - 当前URL: {current_state.get(url, N/A)} - 已提取数据: {json.dumps(current_state.get(extracted, {}), indent2)} 你可以执行以下动作 {json.dumps(allowed_actions, indent2)} 请生成一个JSON格式的行动计划来完成任务。计划应是一个步骤列表每个步骤包含“action”和“args”。 只使用上面允许的动作。如果当前页面已有需要的数据动作可以是“extract_data”。 输出必须是有效的JSON格式如下 {{ steps: [ {{action: action_name, args: {{arg1: value1}}}}, ... ] }} response self.client.chat.completions.create( modelself.model, messages[{role: user, content: prompt}], response_format{type: json_object}, # 关键强制JSON输出 temperature0.1, # 低随机性保证稳定性 ) plan_json json.loads(response.choices[0].message.content) return plan_json3. 执行器与验证器的集成 (core/executor.py和core/verifier.py)执行器在每一步动作后都会调用验证器进行状态检查。# core/verifier.py import jsonschema from parsel import Selector class RuntimeVerifier: def __init__(self, runtime_selectors_config): self.selectors runtime_selectors_config def verify(self, html_content, step_name): 根据步骤名检查页面是否包含关键元素 selector self.selectors.get(step_name) if not selector: return True, No selector defined for this step sel Selector(texthtml_content) if sel.css(selector).get(): return True, Key element found else: return False, fKey element not found with selector: {selector} class OutputVerifier: def __init__(self, schema_path): with open(schema_path, r) as f: self.schema json.load(f) def verify(self, data): try: jsonschema.validate(instancedata, schemaself.schema) return True, None except jsonschema.ValidationError as e: return False, fSchema validation failed: {e.message}# core/executor.py (片段) async def execute_step(self, step, state, constraint_checker, runtime_verifier): action step[action] args step.get(args, {}) # 1. 检查动作约束 constraint_checker.check_action(action) # 2. 执行动作 if action navigate_to: url args[url] constraint_checker.check_navigation(url) # 导航前检查URL await self.page.goto(url, wait_untilnetworkidle) state[url] url constraint_checker.record_request() # 记录请求 elif action extract_text: selector args[selector] elements await self.page.locator(selector).all() values [await el.text_content() for el in elements] state[extracted][args[key]] values # ... 其他动作处理 # 3. 获取执行后状态HTML快照 html await self.page.content() state[html_snapshot] html # 4. 运行时验证 is_valid, msg runtime_verifier.verify(html, step.get(name, action)) if not is_valid: raise RuntimeVerificationError(fStep {action} failed runtime verification: {msg}) # 5. 检查资源约束 constraint_checker.check_time() return state4.3 组装智能体与任务循环 (core/agent.py)这是粘合所有组件的“大脑”。class ConstrainedVerifiableAgent: def __init__(self, task_config_path, llm_client): self.constraints ConstraintChecker(task_config_path) self.llm llm_client self.runtime_verifier RuntimeVerifier(self._load_runtime_selectors(task_config_path)) self.output_verifier OutputVerifier(self._load_schema_path(task_config_path)) self.executor PlaywrightExecutor() self.state {url: None, extracted: {}, history: []} async def run(self, task_description): self.constraints.start_execution() await self.executor.start() try: while not self._is_task_complete(): # 1. 规划 allowed_acts self.constraints.config[constraints][allowed_actions] plan await self.llm.create_plan(task_description, self.state, allowed_acts) # 静态计划验证简化示例检查步骤数 if len(plan.get(steps, [])) 10: raise PlanningError(Plan too long, potential loop.) # 2. 执行与验证循环 for step in plan[steps]: self.state await self.executor.execute_step( step, self.state, self.constraints, self.runtime_verifier ) self.state[history].append(step) # 3. 判断是否完成LLM或规则 if self._has_target_data(self.state): break # 否则基于新状态进行下一轮规划... # 4. 最终输出验证 final_data self._format_output(self.state) is_valid, error self.output_verifier.verify(final_data) if not is_valid: raise OutputValidationError(error) return {status: success, data: final_data} except (SecurityViolationError, ResourceLimitError, RuntimeVerificationError) as e: return {status: failed, error_type: type(e).__name__, message: str(e), state_snapshot: self.state} finally: await self.executor.close()4.4 集成到Airflow DAG最后我们将这个智能体包装成一个可以在Airflow中调用的算子。# tasks/phone_price_monitor.py from airflow.decorators import dag, task from datetime import datetime from core.agent import ConstrainedVerifiableAgent from llm.client import LLMClient default_args { owner: data_team, retries: 2, retry_delay: timedelta(minutes5), } dag( dag_idsafe_phone_price_monitor, default_argsdefault_args, schedule_interval0 */6 * * *, # 每6小时执行一次 start_datedatetime(2024, 1, 1), catchupFalse, ) def create_dag(): task def monitor_product(product_url): 单个产品监测任务 llm_client LLMClient(modelgpt-4-turbo) agent ConstrainedVerifiableAgent( task_config_pathconfig/tasks/phone_monitor.yaml, llm_clientllm_client ) task_desc fNavigate to {product_url} and extract the current price, product name, and availability status. result asyncio.run(agent.run(task_desc)) if result[status] success: # 将成功数据推送到数据仓库或消息队列 send_to_data_warehouse(result[data]) return result[data] else: # 将失败详情发送到告警系统如Slack、PagerDuty send_alert(fTask failed: {result[error_type]} - {result[message]}) # Airflow会将此标记为失败任务并根据DAG设置重试 raise ValueError(fAgent execution failed: {result}) # 假设我们有一个产品URL列表 product_urls get_product_urls_from_catalog() # 使用Airflow的动态任务映射为每个URL创建一个并行任务 monitor_product.expand(product_urlproduct_urls) # 创建DAG实例 phone_price_dag create_dag()这个DAG会并行启动多个智能体实例每个实例独立、安全地监测一个产品页面。任何一个实例的失败如触犯约束、验证失败都不会影响其他实例并且其完整的错误信息和状态快照会被记录下来用于排查。5. 避坑指南与实战心得在实际部署和运行此类框架时我踩过不少坑也积累了一些让系统更稳健的经验。5.1 LLM提示工程中的稳定性陷阱问题LLM有时会“幻想”出配置中不存在的动作或者输出格式不符合JSON要求。对策结构化输出是生命线务必使用LLM API提供的response_format{type: json_object}参数。对于不支持此功能的模型在Prompt的开头和结尾用json ...明确标记并在后处理中使用严格的json.loads()并捕获异常。少样本示例Few-Shot至关重要在Prompt中提供2-3个清晰、正确的规划示例。示例应涵盖正常流程和边界情况如“未找到元素”。温度Temperature设置规划阶段务必使用低温度如0.1-0.3以减少随机性。只有在需要创意性解决问题时如选择器失效后的备选方案生成才适当调高。设置重规划Re-plan机制当LLM输出的计划无法通过静态验证或连续几步执行失败时应将当前状态和错误信息反馈给LLM要求它重新规划。这通常比简单的重试更有效。5.2 验证逻辑的设计平衡问题验证规则太松垃圾数据会漏过太紧则误杀率高任务频繁失败。对策分层验证采用“宽松运行时验证 严格输出验证”策略。运行时验证只检查页面是否发生根本性错误如跳转到登录页、关键容器元素消失。而数据是否完整、格式是否正确交给最终的JSON Schema校验。这样避免了因页面微小改动导致任务中途失败。定义“软失败”状态不是所有验证失败都应导致任务终止。例如如果“折扣价”元素没找到但“原价”找到了可以将其记录为null并打上discount_missing的标记让任务继续并成功结束。下游系统可以处理这种部分数据。定期更新验证规则将运行时验证的选择器作为配置管理。可以建立一个简单的版本控制或A/B测试机制当某个选择器失败率持续升高时自动尝试切换到备选选择器。5.3 性能、成本与扩展性问题LLM API调用成本高Playwright执行速度慢大规模采集时资源消耗大。对策缓存LLM响应对于相同的任务描述和页面状态其最优计划很可能是相同的。可以使用Redis或内存缓存对LLM的规划请求进行缓存缓存键由任务描述页面关键特征哈希构成。分离“探索”与“采集”对于定期执行的重复任务如每日价格监测第一次运行时用LLM进行“探索”生成一个稳定的行动计划序列如导航路径、选择器。之后的任务可以直接复用这个“剧本”无需再次调用LLM除非验证失败触发重规划。这能节省90%以上的LLM成本。无头浏览器池使用playwright-pool或自定义连接池管理多个浏览器实例避免为每个任务启动/关闭浏览器的开销。结合异步IO可以显著提高吞吐量。降级策略准备一个基于规则Rule-based的备用提取器。当LLM智能体连续失败或遇到已知的、结构简单的页面时可以降级到使用预定义规则和XPath进行采集更快速、更稳定。5.4 调试与监控问题智能体在黑盒中失败难以定位是LLM规划问题、页面变化问题还是验证规则问题。对策全链路追踪为每个任务生成唯一的trace_id并将每一步的输入状态/计划、执行动作、页面快照HTML或截图、验证结果都记录到可查询的存储中如Elasticsearch。Playwright的trace.playwright.dev功能可以完美集成。构建调试回放界面开发一个简单的Web界面输入trace_id就能像播放视频一样逐步回放智能体的整个执行过程查看每一步的页面截图和LLM的“思考过程”规划输出。这是排查问题最强大的工具。定义关键指标监控任务成功率、平均执行时间、LLM调用次数、各类验证失败率运行时/输出、约束触发次数。设置告警当成功率下降或LLM成本异常升高时及时通知。构建一个“失败安全”的智能体框架初期投入在设计和基础设施上的精力会比较多但一旦建成它带来的长期收益是巨大的采集流程的可靠性从“碰运气”变成了“可度量、可管理”。你不再需要每天忙于处理各种奇怪的爬虫崩溃而是可以专注于定义更复杂的采集目标和优化数据质量规则。这套框架的核心思想——通过约束限制行动范围通过验证确保每一步的正确性——不仅适用于网络采集对于任何需要LLM与不确定环境交互的自动化任务如软件操作、机器人流程自动化RPA都具有重要的借鉴意义。