3个最流行的开源大模型网络爬虫框架

在传统网络爬虫中，主要的挑战一直是手动操作的工作量。使用像 Beautiful Soup（BS4）和 Selenium 这样的工具时，我们需要为每个新网站编写解析代码，需要适配和适应不同的 HTML 结构。这种不断的修改既耗时又容易出错。然而，当出现了大模型之后就没那么复杂了。随着具备视觉功能的大型语言模型（LLM）的出现，我们现在可以创建几乎通用的网络爬虫代理，大大简化和自动化了这一过程。

在这篇博客中，我们的主要关注点是探讨三种强大的工具：ScrapeGraph、FireCrawl 和 AgentQL。这些创新的库在革命性地改变网络爬虫领域方面发挥了关键作用，提供了先进的功能，使我们能够创建高效且多功能的爬虫代理。通过深入的讨论和实际示例，我们将详细探讨这些工具如何简化网络爬虫过程，并实现构建由 LLM 模型驱动的爬虫代理的目标。

ScrapeGraph

ScrapeGraphAI 是一个开源框架，它利用大型语言模型（LLM）和直接图逻辑的力量来简化网络爬虫过程。使用 ScrapeGraphAI，为网站、文档和 XML 文件创建爬虫管道变得轻而易举。你只需指定要提取的信息，其余的工作由库来处理。其直观的界面和先进的功能使其成为开发人员寻求高效、精确的网络爬虫解决方案时的首选。

import os
from dotenv import load_dotenv
from scrapegraphai.graphs import SmartScraperGraph
from scrapegraphai.utils import prettify_exec_info
import json

# 从.env加载环境变量
load_dotenv()

# 从环境变量加载openai key
openai_key = os.getenv("OPENAI_APIKEY")

# 配置SmartScraperGraph
graph_config = {
   "llm": {
      "api_key": openai_key,
      "model": "gpt-3.5-turbo",
   },
}

# 创建SmartScraperGraph并运行
smart_scraper_graph = SmartScraperGraph(
   prompt="列举所有的产品和他们的价格",
   # 接收一个html网页页面
   source="https://s.taobao.com/search?page=1&q=iphone",
   config=graph_config
)

# 执行爬虫并保存结果
result = smart_scraper_graph.run()
with open("results.json", 'w', encoding='utf-8') as f:
      json.dump(result, f, indent=4)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

我们导入了必要的模块和库，如 os、dotenv、SmartScraperGraph 和 json。首先，我们从 .env 文件中加载环境变量，这是安全存储像 API 密钥等敏感信息的常见做法。graph_config 字典包含 SmartScraperGraph 所需的配置设置。在本例中，它包含 OpenAI API 密钥并指定使用的 GPT 模型（gpt-3.5-turbo）。接下来，我们创建 SmartScraperGraph 类的实例，向其提供提示（查询）、源（要爬取的网页 URL）和配置设置。在 smart_scraper_graph 实例上调用 run() 方法来执行爬虫并从网页中提取数据。提取的数据存储在 result 变量中。最后，使用 json.dump() 方法将提取的数据保存到名为 “results.json” 的 JSON 文件中，供进一步处理或分析。

{
    "products": [
        {
            "Name": "Apple/苹果 iPhone 13 Pro Max苹果13promax 苹果13 pro 手机",
            "Price": "¥3238.00"
        },
        {
            "Name": "新款Apple/苹果 iPhone 15 Pro Max 苹果5G手机15ProMax 国行正品",
            "Price": "¥7428.00"
        },
        {
            "Name": "Apple/苹果 iPhone 15 支持移动联通电信5G 双卡双待手机",
            "Price": "¥6999.00"
        },
        {
            "Name": "Apple/苹果 iPhone14ProMax双卡原装正品苹果14Promax全网通全新",
            "Price": "¥5999.00"
        },
        {
            "Name": "Apple/苹果 iPhone 15 Pro Max",
            "Price": "¥9999.00"
        }
    ]
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

如果你想从不同的来源爬取数据，只需在代码中更改 URL。SmartScraperGraph 的灵活性允许你在不显著修改代码的情况下，针对各种网站或网页。这意味着你可以根据具体需求调整爬虫过程，轻松从各种来源收集数据。虽然 SmartScraperGraph 在处理某些网站的弹出窗口或拦截器时可能遇到限制，但需要注意的是，SmartScraperGraph 是一个开源库，这意味着你可以根据具体要求对其进行定制。

FireCrawl

Firecrawl 作为一个强大的解决方案，配备了一系列功能，旨在克服网络爬虫工作中的固有挑战。它高效地管理代理、缓存、速率限制等复杂性，确保数据检索过程的顺畅。Firecrawl 的爬取能力扩展到网站的所有可访问子页面，无论是否存在站点地图，保证全面的数据提取。即使面对通过 JavaScript 动态渲染的内容，Firecrawl 也能非常高效的地捕获每一条有价值的信息。其输出经过 Markdown 格式化，简化了与大型语言模型（LLM）和其他应用程序的集成。

你可以注册 Firecrawl 的免费套餐，获得基本的爬虫功能。通过在这注册，你可以爬取最多 500 个页面，限制为每分钟 5 次爬取以及 1 个并发爬取任务。

下面是一个使用FireCrawl爬取

from firecrawl import FirecrawlApp
from openai import OpenAI
from dotenv import load_dotenv
import os
import json
import pandas as pd
from datetime import datetime

def scrape_data(url):
    load_dotenv()
    # 初始化FirecrawlApp实例
    app = FirecrawlApp(api_key=os.getenv('FIRECRAWL_API_KEY'))

    # 爬取单个URL
    scraped_data = app.scrape_url(url, {'pageOptions': {'onlyMainContent': True}})

    # 检查是否爬取到了markdown数据
    if 'markdown' in scraped_data:
        return scraped_data['markdown']
    else:
        raise KeyError("The key 'markdown' does not exist in the scraped data.")

def save_raw_data(raw_data, timestamp, output_folder='output'):
    # 确保输出文件夹存在
    os.makedirs(output_folder, exist_ok=True)

    # 保存原始数据到Markdown文件
    raw_output_path = os.path.join(output_folder, f'rawData_{timestamp}.md')
    with open(raw_output_path, 'w', encoding='utf-8') as f:
        f.write(raw_data)
    print(f"Raw data saved to {raw_output_path}")

def format_data(data, fields=None):
    load_dotenv()
    # 初始化OpenAI实例
    client = OpenAI(api_key=os.getenv('OPENAI_APIKEY'))

    # 如果未提供字段列表，则使用默认字段
    if fields is None:
        fields = ["名称","价格","地址", "链接"]

    # 定义系统消息内容
    system_message = f"""你是一个智能文本提取和转换助手。你的任务是从给定的文本中提取结构化信息，并将其转换为纯JSON格式。JSON
    应仅包含从文本中提取的结构化数据，不包含任何额外的评论、解释或无关的信息。你可能会遇到无法找到所需字段数据的情况，或者数据会以外语形式出现。请处理以下文本，并以纯JSON格式提供输出，JSON前后不应有任何文字。:"""

    # 定义用户消息内容
    user_message = f"请提供要处理的文本和需要提取的信息字段。:\nPage content:\n\n{data}\n\nInformation to extract: {fields}"

    response = client.chat.completions.create(
        model="gpt3.5",
        response_format={"type": "json_object"},
        messages=[
            {
                "role": "system",
                "content": system_message
            },
            {
                "role": "user",
                "content": user_message
            }
        ]
    )

    # 检查API响应是否包含选择数据
    if response and response.choices:
        formatted_data = response.choices[0].message.content.strip()
        print(f"Formatted data received from API: {formatted_data}")

        try:
            parsed_json = json.loads(formatted_data)
        except json.JSONDecodeError as e:
            print(f"JSON decoding error: {e}")
            print(f"Formatted data that caused the error: {formatted_data}")
            raise ValueError("The formatted data could not be decoded into JSON.")

        return parsed_json
    else:
        raise ValueError("The OpenAI API response did not contain the expected choices data.")

def save_formatted_data(formatted_data, timestamp, output_folder='output'):
    # Ensure the output folder exists
    os.makedirs(output_folder, exist_ok=True)

    # 保存格式化数据到JSON文件
    output_path = os.path.join(output_folder, f'sorted_data_{timestamp}.json')

    with open(output_path, 'w', encoding='utf-8') as f:
        json.dump(formatted_data, f, indent=4)
    print(f"Formatted data saved to {output_path}")

    # 检查格式化数据是否为字典且只包含一个键
    if isinstance(formatted_data, dict) and len(formatted_data) == 1:
        key = next(iter(formatted_data))  # Get the single key
        formatted_data = formatted_data[key]

    # 转换格式化数据为pandas DataFrame
    df = pd.DataFrame(formatted_data)

    # 准换格式化数据为pandas DataFrame
    if isinstance(formatted_data, dict):
        formatted_data = [formatted_data]

    df = pd.DataFrame(formatted_data)

    # 保存格式化数据到CSV文件
    # excel_output_path = os.path.join(output_folder, f'sorted_data_{timestamp}.xlsx')
    df.to_csv(f"{timestamp}.csv", index=False)

if __name__ == "__main__":
    # 爬取的URL
    url = 'https://bj.ke.com/ershoufang/rs/'

    try:
        # 生成时间戳
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')

        # 爬取数据
        raw_data = scrape_data(url)

        # 保存原始数据
        save_raw_data(raw_data, timestamp)

        # 格式化数据
        formatted_data = format_data(raw_data)

        #   保存格式化数据
        save_formatted_data(formatted_data, timestamp)
    except Exception as e:
        print(f"An error occurred: {e}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

• 导入库 导入必要的库，例如 firecrawl、OpenAI、dotenv、os、json、pandas 和 datetime，以便实现爬取和数据处理所需的各种功能。

import firecrawl
import OpenAI
import dotenv
import os
import json
import pandas as pd
from datetime import datetime

1
2
3
4
5
6
7
8

• 爬取数据 scrape_data() 函数利用你的 API 密钥初始化一个 FirecrawlApp，并使用 Firecrawl 爬取一个 URL。它检索网页的主要内容并以 Markdown 格式返回。

def scrape_data(api_key, url):
    app = firecrawl.FirecrawlApp(api_key)
    content = app.scrape(url)
    return content

1
2
3
4
5

• 保存原始数据 save_raw_data() 函数将原始的 Markdown 数据保存到文件中，为便于识别，文件名附加时间戳。

def save_raw_data(data):
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    filename = f"raw_data_{timestamp}.md"
    with open(filename, 'w') as file:
        file.write(data)

1
2
3
4
5
6

• 格式化数据 format_data() 函数利用 OpenAI 的 GPT 模型从原始数据中提取结构化信息。它构建系统和用户消息，提示模型从文本中提取指定字段。提取的数据以 JSON 格式返回。

def format_data(api_key, raw_data):
    openai.api_key = api_key
    response = openai.Completion.create(
        model="text-davinci-003",
        prompt=f"从文本中提取结构化数据:\n\n{raw_data}",
        max_tokens=1000
    )
    return json.loads(response.choices[0].text)

1
2
3
4
5
6
7
8
9

• 保存格式化数据 save_formatted_data() 函数将格式化的数据以 JSON 格式保存到文件中，文件名附加时间戳。此外，它将 JSON 数据转换为 pandas DataFrame 并保存为 CSV 文件，以便进一步分析。

def save_formatted_data(data):
    timestamp = datetime.now().strftime("%Y%m%d%H%M%S")
    json_filename = f"formatted_data_{timestamp}.json"
    csv_filename = f"formatted_data_{timestamp}.csv"

    with open(json_filename, 'w') as json_file:
        json.dump(data, json_file)

    df = pd.DataFrame(data)
    df.to_csv(csv_filename, index=False)

1
2
3
4
5
6
7
8
9
10
11

• 主函数 在主函数中，指定要爬取的 URL（url）。脚本然后尝试爬取数据、保存原始数据、格式化数据并保存格式化数据。如果在这些步骤中的任何一步发生错误，它将被捕获并显示。

if __name__ == "__main__":
    api_key = os.getenv("FIRECRAWL_API_KEY")
    url = "https://example.com"  # 替换为要爬取的实际 URL

    try:
        raw_data = scrape_data(api_key, url)
        save_raw_data(raw_data)

        formatted_data = format_data(api_key, raw_data)
        save_formatted_data(formatted_data)

    except Exception as e:
        print(f"An error occurred: {e}")
1
2
3
4
5
6
7
8
9
10
11
12
13

JSON 格式的输出

[
    {
        "名称": "南向一居室户型好，中间楼层，老少居住皆宜。",
        "地址": "二里庄北里",
        "价格": "345万",
        "链接": "https://bj.ke.com/ershoufang/101125066609.html"
    },
    {
        "名称": "三里河 正规2 居室 随时可看 随时签约",
        "地址": "三里河三区",
        "价格": "798万",
        "链接": "https://bj.ke.com/ershoufang/101125059884.html"
    },
    {
        "名称": "骊龙园 3室2厅 南 北",
        "地址": "骊龙园",
        "价格": "559万",
        "链接": "https://bj.ke.com/ershoufang/101125057747.html"
    },
    {
        "名称": "南三环草桥马家堡南向一居室商品房社区",
        "地址": "玺萌鹏苑",
        "价格": "270万",
        "链接": "https://bj.ke.com/ershoufang/101125055143.html"
    },
    {
        "名称": "2018年次新小区 19号线10号线机场线草桥站",
        "地址": "今日草桥",
        "价格": "710万",
        "链接": "https://bj.ke.com/ershoufang/101125049365.html"
    },
    {
        "名称": "裕华园一区 1室1厅 南 北",
        "地址": "裕华园一区",
        "价格": "71万",
        "链接": "https://bj.ke.com/ershoufang/101125059731.html"
    },
    {
        "名称": "化工大院 南北通透两居室 不临街 平改坡 停车方便",
        "地址": "化工大院",
        "价格": "620万",
        "链接": "https://bj.ke.com/ershoufang/101125053322.html"
    },
    {
        "名称": "北辰福第二号院 高楼层采光视野好 满五唯一 无抵押",
        "地址": "北辰福第二号院",
        "价格": "409万",
        "链接": "https://bj.ke.com/ershoufang/101125048788.html"
    },
    {
        "名称": "海特花园西区三居室，户型方正，南北通透",
        "地址": "海特花园西区",
        "价格": "445万",
        "链接": "https://bj.ke.com/ershoufang/101125045278.html"
    },
    {
        "名称": "中国铁建国际城 3室1厅 南 北",
        "地址": "中国铁建国际城",
        "价格": "680万",
        "链接": "https://bj.ke.com/ershoufang/101125045266.html"
    }
]

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63

使用 Firecrawl，我们可以毫不费力地爬取整个网页，而无需担心复杂的细节。只需更改 URL，我们就可以轻松地将爬取工作适应不同的来源。尽管 Firecrawl 在处理动态内容和速率限制等爬取挑战方面表现出色，但仍然需要爬取网站上的所有页面，并导航到后续页面直到到达最后一个页面。这时，AgentQL 提供了一个解决方案。

AgentQL

AgentQL for Web 通过使用自然语言查询提供了一种革命性的方法来与网页元素交互。借助 AgentQL，用户可以轻松定位和交互网页元素，而无需复杂的代码或特定选择器。这种直观的界面简化了网页自动化的过程，使用户能够轻松高效地执行任务。无论是点击按钮、填写表单还是浏览页面，AgentQL for Web 都简化了交互过程，使网页自动化对各类用户都变得易于访问。

from dotenv import load_dotenv
import agentql
#from agentql.sync_api import ScrollDirection
import csv

load_dotenv()

PRODUCTS = """
{
    results{
        products[]{
            product_name
            product_price
            num_reviews
            rating
        }
    }
}
"""
NEXT_PAGE_BTN ="""
{
    next_page_button_enabled
    next_page_button_disabled
}
"""
session = agentql.start_session("https://s.taobao.com/search?page=1&q=iphone")

session.driver.scroll_to_bottom()

pagination = session.query(NEXT_PAGE_BTN)
print("get pagination")
print(pagination.next_page_button_enabled)

with open("products.csv", "a",newline="") as file:
    fieldnames = ["product_name","product_price","num_reviews","rating"]
    writer = csv.DictWriter(file,fieldnames=fieldnames)
    writer.writeheader()
    
    print(f"enabled button : {pagination.to_data()['next_page_button_enabled']}")
    print(f"disabled button : {pagination.to_data()['next_page_button_disabled']}")
    
    while(
        pagination.to_data()['next_page_button_enabled'] and
        pagination.to_data()['next_page_button_disabled'] is None
    ):
        products = session.query(PRODUCTS)
        print("scraped product data")
        print(products.to_data())
        
        for product in products.to_data()['results']['products']:
            print(f"product: {product}")
            writer.writerow(product)
        print("data written to csv")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

• 查询定义：定义了两个 GraphQL 查询：PRODUCTS 用于检索产品详细信息，而 NEXT_PAGE_BTN 用于检查分页的下一页按钮是否可用。
• 会话初始化：使用 agentql.start_session() 与目标 URL（"https://s.taobao.com/search?page=1&q=iphone"）建立会话。
• 滚动到底部：使用 session.driver.scroll_to_bottom() 将浏览器窗口滚动到页面底部。
• 分页检查：使用 NEXT_PAGE_BTN 查询确定下一页按钮的状态（启用或禁用），结果存储在 pagination 变量中。
• CSV 文件初始化：创建一个名为“products.csv”的 CSV 文件，并定义产品信息的字段名称。
• 数据提取循环：在下一页按钮启用且未找到禁用按钮的情况下：使用 PRODUCTS 查询提取产品数据。每个产品的详细信息写入 CSV 文件。该过程持续进行，直到爬取完所有页面。

结论

ScrapeGraph、Firecrawl 和 AgentQL 代表了最新一代的网络爬虫框架。每个框架都有自己的优势，可以满足不同的爬取需求。 ScrapeGraph 利用 LLM 和直接图形逻辑实现多功能爬取管道，Firecrawl 擅长高效处理复杂的 Web 场景，AgentQL 引入自然语言交互以实现无缝 Web 元素操作。这些框架共同简化了从网络中提取有价值数据的过程，为开发人员提供了强大的工具来轻松处理爬取任务。