一、AI时代的爬虫人需要改变了
AI大模型时代下的爬虫人也需要紧跟智能的潮流，抓住模型发展的契机，使用AI创建新的爬虫定义新的爬虫范式！数据的解析、整理、格式化可以让大模型来提高处理的效率！
介绍一个开源llm爬虫框架：Crawl4AI是一个功能全面、性能优越的网络爬虫工具，特别适合需要处理大量网页数据并进行智能分析的场景。
1.1 还在编写xpath、css选择器定位数据吗
爬虫人花费了大量的时间在元素的定位和数据的解析获取上，我们为此招募了许多的xpath、css规则编写人，就为了适应上百、上千的web页面的数据处理。
首先我们需要明确的一个点就是，爬虫的数据处理与源码的获取不是一个概念，AI并不能帮助我们获取到所有的网站的源代码！为什么不能获取呢？由于现在的数据安全意识的增强，许多的站点都有反爬虫以及风控措施、模型不能直接与这些防护做对抗！
那么AI可以帮助做些什么呢？AI可以使用他的推理能力和智能体的能力，帮助用户使用自动化的工具打开一些简单的站点。可以帮助我们在源代码里面提取一些表格、列表等结构化的数据并处理后输出！
1.2 开源模型Crawl4AI
Crawl4AI是一个开源的网络爬虫和数据提取工具，专为大型语言模型（LLM）设计，旨在简化网页数据的抓取和提取过程。它通过异步操作、高效的数据处理和智能提取策略，为开发者提供了一个强大且灵活的工具，能够应对现代网页的复杂性和动态性。Crawl4AI不仅支持传统的爬虫功能，还融入了AI技术，使其在处理大规模数据和动态内容时表现出色。
Crawl4AI的核心目标是提供一个高效、灵活且易于集成的网络爬虫工具，特别适合与大型语言模型和AI应用配合使用。以下是Crawl4AI的主要特点：
异步操作：采用异步架构，能够同时处理多个URL，提高爬取效率，特别适合大规模数据抓取。 LLM友好：输出的数据格式（如JSON、Markdown）经过优化，易于LLM处理，方便集成到AI应用中。 动态内容处理：能够执行JavaScript，处理动态加载的内容，确保抓取的数据完整性。 智能数据提取：利用AI技术自动分析网页结构，生成数据提取规则，减少手动配置的工作量。 多浏览器支持：支持Chromium、Firefox和WebKit等多种浏览器，适应不同网站的兼容性需求。 灵活的配置：提供丰富的配置选项，如钩子函数、缓存管理、代理设置等，满足各种定制化需求。
1.3 功能代码解析
Crawl4AI的代码结构清晰，模块化设计便于维护和扩展。以下是对其主要功能和代码实现的解析：
异步爬虫（AsyncWebCrawler）
import asyncio
from crawl4ai import AsyncWebCrawler
async def main():
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(url="http://zhaomeng.net")
print(result.markdown)
asyncio.run(main())
代码解析：
AsyncWebCrawler类负责管理爬虫的生命周期，包括浏览器的启动和关闭。 arun方法执行爬取任务，返回一个CrawlResult对象，包含网页内容和其他元数据。 asyncio.run(main())用于运行异步主函数。
1.4 数据提取策略
Crawl4AI提供了多种数据提取策略，包括基于CSS/XPath的传统方法和基于LLM的智能提取。以下是使用LLM提取策略的示例：
from crawl4ai.extraction_strategy import LLMExtractionStrategy
INSTRUCTION_TO_LLM = "Extract all rows from the main table as objects with 'CASNo','purity','MF','MW','SMILES','size', 'price' ,'stock' from the content."
class Product(BaseModel):
CASNo:str
size: str
price: str
stock:str
purity:str
MF:str
MW:str
SMILES:str
llm_strategy = LLMExtractionStrategy(
provider="deepseek/deepseek-chat",
api_token=apikey,
schema=Product.model_json_schema(),
extraction_type="schema",
instruction=INSTRUCTION_TO_LLM,
chunk_token_threshold=1000,
overlap_rate=0.0,
apply_chunking=True,
input_format="markdown",
extra_args={"temperature": 0.0, "max_tokens": 800},
)
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(
url="https://www.chemshuttle.com/building-blocks/amino-acids/fmoc-r-3-amino-4-4-nitrophenyl-butyric-acid.html",
extraction_strategy=extraction_strategy
)
print(result.extracted_content)
解析：
LLMExtractionStrategy类利用LLM的语义理解能力，自动提取网页中的关键信息。 provider和api_token指定LLM服务提供商和API密钥。 schema定义需要提取的字段，instruction提供提取指导。
二、动态内容处理
Crawl4AI能够处理通过JavaScript动态加载的内容。以下是配置爬虫执行JavaScript的示例：
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(
url="https://example.com",
js_code="window.scrollTo(0, document.body.scrollHeight);",
wait_for="document.querySelector('.content-loaded')"
)
print(result.markdown)
解析：
add_hook方法注册钩子函数，可在爬取前、后等阶段执行自定义逻辑。 这种机制便于扩展功能，如日志记录或异常处理。
2.1 错误处理和健壮性
Crawl4AI实现了全面的错误处理机制，确保在网络不稳定或网页结构变化时稳定运行。以下是错误处理的示例：
try:
result = await crawler.arun(url="https://example.com")
except Exception as e:
print(f"An error occurred: {e}")
解析：
内部捕获并处理网络错误、超时等异常，确保爬虫稳定性。 用户可通过try-except块处理爬取过程中的异常。
2.2 案例实战
背景导入：
获取化学生物医药行业的站点的产品信息以及产品的价格、规格、纯度等信息
2.3.1 deepseek部署
ollama安装：https://ollama.com/
deepseek-r1本地部署
ollama run deepseek-r1:14b
官网注册获取API
官网：https://platform.deepseek.com/usage
注册api_key
安装Crawl4AI
pip install crawl4ai
playwright install
2.3.2 AI爬虫开发
上述的相关分析，和正常做一些爬虫业务需求是一样的，不会因为需要对接就有什么特别不太一样的，所以按正常的需求分析。
配置数据对象
class Product(BaseModel):
CASNo:str
size: str
price: str
stock:str
purity:str
MF:str
MW:str
SMILES:str
配置AI采集
llm_strategy = LLMExtractionStrategy(
provider="deepseek/deepseek-chat",
api_token="sk-1561f1bf223f41df908dc96cd3e5b403",
schema=Product.model_json_schema(),
extraction_type="schema",
instruction=INSTRUCTION_TO_LLM,
chunk_token_threshold=1000,
overlap_rate=0.0,
apply_chunking=True,
input_format="markdown",
extra_args={"temperature": 0.0, "max_tokens": 800},
)
crawl_config = CrawlerRunConfig(
extraction_strategy=llm_strategy,
cache_mode=CacheMode.BYPASS,
process_iframes=False,
remove_overlay_elements=True,
exclude_external_links=True,
)
无头浏览器配置及采集
browser_cfg = BrowserConfig(headless=True, verbose=True)
async with AsyncWebCrawler(config=browser_cfg) as crawler:
try:
result = await crawler.arun(url=URL_TO_SCRAPE, config=crawl_config)
if result.success:
data = json.loads(result.extracted_content)
print("Extracted items:", data)
llm_strategy.show_usage()
else:
print("Error:", result.error_message)
except Exception as e:
print(traceback.print_exc())
结果展示
数据展示：
Extracted items: [{'CASNo': '269398-78-9', 'size': '1g', 'price': '$150.00', 'stock': 'Typically in stock', 'purity': '95%', 'MF': 'C25H22N2O6', 'MW': '446.459', 'SMILES': 'OC(=O)C[C@@H](CC1=CC=C(C=C1)[N+]([O-])=O)NC(=O)OCC1C2=CC=CC=C2C2=C1C=CC=C2', 'error': False}, {'CASNo': '269398-78-9', 'size': '5g', 'price': '$450.00', 'stock': 'Typically in stock', 'purity': '95%', 'MF': 'C25H22N2O6', 'MW': '446.459', 'SMILES': 'OC(=O)C[C@@H](CC1=CC=C(C=C1)[N+]([O-])=O)NC(=O)OCC1C2=CC=CC=C2C2=C1C=CC=C2', 'error': False}, {'CASNo': '269398-78-9', 'size': '10g', 'price': 'Inquire', 'stock': 'Inquire', 'purity': '95%', 'MF': 'C25H22N2O6', 'MW': '446.459', 'SMILES': 'OC(=O)C[C@@H](CC1=CC=C(C=C1)[N+]([O-])=O)NC(=O)OCC1C2=CC=CC=C2C2=C1C=CC=C2', 'error': False}, {'CASNo': '269398-78-9', 'size': '100g', 'price': '$6980.00', 'stock': 'Inquire', 'purity': '95%', 'MF': 'C25H22N2O6', 'MW': '446.459', 'SMILES': 'OC(=O)C[C@@H](CC1=CC=C(C=C1)[N+]([O-])=O)NC(=O)OCC1C2=CC=CC=C2C2=C1C=CC=C2', 'error': False}]
完整代码如下
import asyncio
import json
import os
import traceback
from typing import List
from crawl4ai import AsyncWebCrawler, BrowserConfig, CacheMode, CrawlerRunConfig
from crawl4ai.extraction_strategy import LLMExtractionStrategy
from pydantic import BaseModel, Field
# URL_TO_SCRAPE = "https://nstchemicals.com/product/s-pro-xylane-cas-868156-46-1/"
# INSTRUCTION_TO_LLM = "Extract all rows from the main table as objects with 'specs', 'price' from the content."
URL_TO_SCRAPE = "https://www.chemshuttle.com/building-blocks/amino-acids/fmoc-r-3-amino-4-4-nitrophenyl-butyric-acid.html"
INSTRUCTION_TO_LLM = "Extract all rows from the main table as objects with 'CASNo','purity','MF','MW','SMILES','size', 'price' ,'stock' from the content."
class Product(BaseModel):
CASNo:str
size: str
price: str
stock:str
purity:str
MF:str
MW:str
SMILES:str
async def main():
llm_strategy = LLMExtractionStrategy(
provider="deepseek/deepseek-chat",
api_token="api-key",
schema=Product.model_json_schema(),
extraction_type="schema",
instruction=INSTRUCTION_TO_LLM,
chunk_token_threshold=1000,
overlap_rate=0.0,
apply_chunking=True,
input_format="markdown",
extra_args={"temperature": 0.0, "max_tokens": 800},
)
crawl_config = CrawlerRunConfig(
extraction_strategy=llm_strategy,
cache_mode=CacheMode.BYPASS,
process_iframes=False,
remove_overlay_elements=True,
exclude_external_links=True,
)
browser_cfg = BrowserConfig(headless=True, verbose=True)
async with AsyncWebCrawler(config=browser_cfg) as crawler:
try:
result = await crawler.arun(url=URL_TO_SCRAPE, config=crawl_config)
if result.success:
data = json.loads(result.extracted_content)
print("Extracted items:", data)
llm_strategy.show_usage()
else:
print("Error:", result.error_message)
except Exception as e:
print(traceback.print_exc())
if __name__ == "__main__":
asyncio.run(main())
