爬虫在deepseek加持下所向无敌

一、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的代码结构清晰，模块化设计便于维护和扩展。以下是对其主要功能和代码实现的解析：

1. 异步爬虫（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())