FireCrawl 网页抓取平台

随着企业竞相将空前的在线数据量转换为 LLM 友好格式，网页抓取和数据提取已成为必不可少的工具。Firecrawl 强大的网页抓取 API 通过企业级自动化和可扩展性功能简化了此过程。

本综合指南重点介绍 Firecrawl 最强大的功能 — /crawl 端点，该端点支持大规模自动网站抓取。

1、Firecrawl简介

网页抓取（Web Scraping）是指从单个网页（如维基百科文章或技术教程）中提取特定数据。它主要用于需要从具有已知 URL 的页面获取特定信息的情况。

另一方面，网页爬取（Web Crawling）涉及通过跟踪链接系统地浏览和发现网页。它专注于网站导航和 URL 发现。

例如，要构建一个回答有关 Stripe 文档问题的聊天机器人，你需要：

Web 爬取以发现并遍历 Stripe 文档站点中的所有页面
Web 抓取以从每个发现的页面中提取实际内容

Firecrawl 如何结合两者

Firecrawl 的 /crawl 端点结合了这两种功能：

URL 分析：通过站点地图或页面遍历识别链接
递归遍历：跟踪链接以发现子页面
内容抓取：从每个页面中提取干净的内容
结果编译：将所有内容转换为结构化数据

当你将 URL https://docs.stripe.com/api 传递给端点时，它会自动发现并抓取所有文档子页面。端点以你喜欢的格式返回内容 - 无论是 markdown、HTML、屏幕截图、链接还是元数据。

2、使用 Firecrawl API 进行网络爬取

Firecrawl 是一个以 REST API 形式公开的网络抓取引擎。你可以通过 cURL 从命令行使用它，也可以使用 Python、Node、Go 或 Rust 语言 SDK 之一使用它。在本教程中，我们将重点介绍其 Python SDK。

开始使用：

在 firecrawl.dev 注册并复制你的 API 密钥。

将密钥保存为环境变量：

export FIRECRAWL_API_KEY='fc-YOUR-KEY-HERE'

或者使用 dot-env 文件：

touch .env
echo "FIRECRAWL_API_KEY='fc-YOUR-KEY-HERE'" >> .env

然后使用 Python SDK：

from firecrawl import FirecrawlApp
from dotenv import load_dotenv

load_dotenv()
app = FirecrawlApp()

加载 API 密钥后， FirecrawlApp 类将使用它与 Firecrawl API 引擎建立连接。

首先，我们将抓取 https://books.toscrape.com/ 网站，该网站专为网页抓取实践而构建：

你无需使用 beautifulsoup4 或 lxml 等库编写数十行代码来解析 HTML 元素、处理分页和数据检索，Firecrawl 的 crawl_url 端点可让你在一行中完成此操作：

base_url = "https://books.toscrape.com/"
crawl_result = app.crawl_url(url=base_url)

结果是一个包含以下键的字典：

crawl_result.keys()

内容如下：

dict_keys(['success', 'status', 'completed', 'total', 'creditsUsed', 'expiresAt', 'data'])

首先，我们对抓取作业的状态感兴趣：

crawl_result['status']

'completed'

如果已完成，让我们看看如何抓取了多少页面：

crawl_result['total']

将近 1200 页（在我的机器上花了大约 70 秒；速度取决于你的连接速度）。让我们看看数据列表的其中一个元素：

sample_page = crawl_result['data'][10]
markdown_content = sample_page['markdown']

print(markdown_content[:500])

- [Home](../../../../index.html)
- [Books](../../books_1/index.html)
- Womens Fiction

# Womens Fiction
**17** results.
**Warning!** This is a demo website for web scraping purposes. Prices and ratings here were randomly assigned and have no real meaning.
01. [![I Had a Nice Time And Other Lies...: How to find love & sh*t like that](../../../../media/cache/5f/72/5f72c8a0d5a7292e2929a354ec8a022f.jpg)](../../../i-had-a-nice-time-and-other-lies-how-to-find-love-sht-like-that_814/index.html)

该页面对应于女性小说页面：

Firecrawl 还在元素的字典中包含页面元数据：

sample_page['metadata']

{
    'url': 'https://books.toscrape.com/catalogue/category/books/womens-fiction_9/index.html',
    'title': 'Womens Fiction | Books to Scrape - Sandbox',
    'robots': 'NOARCHIVE,NOCACHE', 
    'created': '24th Jun 2016 09:29',
    'language': 'en-us',
    'viewport': 'width=device-width',
    'sourceURL': 'https://books.toscrape.com/catalogue/category/books/womens-fiction_9/index.html',
    'statusCode': 200,
    'description': '',
    'ogLocaleAlternate': []
}

我们没有提到的一件事是 Firecrawl 如何处理分页。如果你滚动到 Books-to-Scrape 的底部，将看到在其上有一个“下一步”按钮。

在转到 books.toscrape.com/category 等子页面之前，Firecrawl 首先从主页抓取所有子页面。之后，如果子页面包含已抓取页面的链接，则会忽略这些链接。

3、高级 Web 抓取配置和最佳实践

Firecrawl 提供多种类型的参数来配置端点如何抓取网站。我们将在此处概述它们及其用例。

3.1 抓取选项

在实际项目中，你将最频繁地调整此参数。它允许你控制如何保存网页的内容。 Firecrawl 允许以下格式：

Markdown — 默认格式
HTML
原始 HTML（整个网页的简单复制/粘贴）
链接
屏幕截图

以下是抓取 Stripe API 的四种格式的示例请求：

# Crawl the first 5 pages of the stripe API documentation
stripe_crawl_result = app.crawl_url(
    url="https://docs.stripe.com/api",
    params={
        "limit": 5,  # Only scrape the first 5 pages including the base-url
        "scrapeOptions": {
            "formats": ["markdown", "html", "links", "screenshot"]
        }
    }
)

当你指定多种格式时，每个网页的数据包含每种格式内容的单独键：

stripe_crawl_result['data'][0].keys()

dict_keys(['html', 'links', 'markdown', 'metadata', 'screenshot'])

screenshot 键的值是指向存储在 Firecrawl 服务器上的 PNG 文件的临时链接，将在 24 小时内过期。以下是 Stripe 的 API 文档主页：

from IPython.display import Image

Image(stripe_crawl_result['data'][0]['screenshot'])

请注意，指定更多格式来转换页面内容会显著减慢该过程。

另一个耗时的操作可能是抓取整个页面内容，而不仅仅是你想要的元素。对于这种情况，Firecrawl 允许你使用 onlyMainContent、 includeTags 和 excludeTags 参数来控制抓取网页的哪些元素。

启用 onlyMainContent 参数（默认情况下禁用）可排除导航、页眉和页脚：

stripe_crawl_result = app.crawl_url(
    url="https://docs.stripe.com/api",
    params={
        "limit": 5,
        "scrapeOptions": {
            "formats": ["markdown", "html"], 
            "onlyMainContent": True,
        },
    },
)

includeTags 和 excludeTags 接受白名单/黑名单 HTML 标签、类和 ID 列表：

# Crawl the first 5 pages of the stripe API documentation
stripe_crawl_result = app.crawl_url(
    url="https://docs.stripe.com/api",
    params={
        "limit": 5,
        "scrapeOptions": {
            "formats": ["markdown", "html"],
            "includeTags": ["code", "#page-header"],
            "excludeTags": ["h1", "h2", ".main-content"],
        },
    },
)

抓取大型网站可能需要很长时间，如果合适，这些小调整可能会对运行时间产生很大影响。

3.2 URL 控制

除了抓取配置外，还有四个选项可以指定在抓取过程中要包含或排除的 URL 模式：

includePaths - 定位特定部分
excludePaths - 避免不需要的内容
allowBackwardLinks - 处理交叉引用
allowExternalLinks - 管理外部内容

以下是使用这些参数的示例请求：

# Example of URL control parameters
url_control_result = app.crawl_url(
    url="https://docs.stripe.com/",
    params={
        # Only crawl pages under the /payments path
        "includePaths": ["/payments/*"],
        # Skip the terminal and financial-connections sections
        "excludePaths": ["/terminal/*", "/financial-connections/*"],
        # Allow crawling links that point to already visited pages
        "allowBackwardLinks": False,
        # Don't follow links to external domains
        "allowExternalLinks": False,
        "scrapeOptions": {
            "formats": ["html"]
        }
    }
)

# Print the total number of pages crawled
print(f"Total pages crawled: {url_control_result['total']}")

Total pages crawled: 134

在此示例中，我们使用特定的 URL 控制参数抓取 Stripe 文档网站：

抓取程序从 https://docs.stripe.com/ 开始，仅抓取 /payments/*路径下的页面
它明确排除了 /terminal/和 /financial-connections/*部分
通过将 allowBackwardLinks 设置为 false，它不会重新访问已抓取的页面
外部链接被忽略（ allowExternalLinks： false）
抓取配置为仅捕获 HTML 内容

这种有针对性的方法有助于将抓取重点放在相关内容上，同时避免不必要的页面，从而使抓取更高效，并专注于我们需要的特定文档部分。

另一个关键参数是 maxDepth，它可让你控制爬虫程序从起始 URL 遍历的深度。例如， maxDepth 为 2 表示它将爬取初始页面及其链接的页面，但不会进一步爬取。

这是 Strip API文档上的另一个示例请求：

# Example of URL control parameters
url_control_result = app.crawl_url(
    url="https://docs.stripe.com/",
    params={
        "limit": 100,
        "maxDepth": 2,
        "allowBackwardLinks": False,
        "allowExternalLinks": False,
        "scrapeOptions": {"formats": ["html"]},
    },
)

# Print the total number of pages crawled
print(f"Total pages crawled: {url_control_result['total']}")

Total pages crawled: 99

注意：当页面有分页（例如第 2、3、4 页）时，使用 maxDepth 时这些分页页面不计入额外深度级别。

3.3 性能和限制

我们在前面的示例中使用的限制参数对于控制网络抓取的范围至关重要。它设置了将被抓取的最大页面数量，这在抓取大型网站或启用外部链接时尤为重要。如果没有此限制，抓取工具可能会遍历无休止的连接页面链，从而浪费不必要的资源和时间。

虽然限制参数有助于控制抓取的广度，但你可能还需要确保抓取的每个页面的质量和完整性。为了确保抓取所有所需内容，你可以启用等待期以让页面完全加载。例如，一些网站使用 JavaScript 来处理动态内容，并使用 iFrames 来嵌入内容或视频或 GIF 等重度媒体元素：

stripe_crawl_result = app.crawl_url(
    url="https://docs.stripe.com/api",
    params={
        "limit": 5,
        "scrapeOptions": {
            "formats": ["markdown", "html"],
            "waitFor": 1000,  # wait for a second for pages to load
            "timeout": 10000,  # timeout after 10 seconds
        },
    },
)

上述代码还将超时参数（timeout）设置为 10000 毫秒（10 秒），这确保如果页面加载时间过长，爬虫将继续前进而不是卡住。

注意： waitFor 时长适用于爬虫遇到的所有页面。

始终牢记计划的限制非常重要：

4、使用 Firecrawl 进行异步 Web 爬取

即使遵循了上一节中的提示和最佳实践，对于包含数千个页面的大型网站，爬取过程仍会非常漫长。为了高效处理此问题，Firecrawl 提供了异步爬取功能，可让你启动爬取并监控其进度，而不会阻止应用程序。这在构建需要在爬取过程中保持响应的 Web 应用程序或服务时特别有用。

4.1 异步编程简介

首先，让我们通过现实世界的类比来理解异步编程：

异步编程就像餐厅服务员一次接受多个订单。他们不必在一张桌子上等待顾客吃完饭后再移到下一张桌子，而是可以从多张桌子上接受订单，将订单提交给厨房，并在准备食物时处理其他任务。

从编程角度来说，这意味着你的代码可以发起多个操作（如 Web 请求或数据库查询），并在等待响应时继续执行其他任务，而不是按顺序处理所有内容。

这种方法在 Web 爬取中特别有价值，因为大多数时间都花在等待网络响应上 — 异步编程允许你同时处理多个页面，而不是在等待每个页面加载时冻结整个应用程序，从而大大提高效率。

4.2 使用 async_crawl_url 方法

Firecrawl 通过 async_crawl_url 提供了一种直观的异步抓取方法：

app = FirecrawlApp()

crawl_status = app.async_crawl_url("https://docs.stripe.com")
print(crawl_status)

{'success': True, 'id': 'c4a6a749-3445-454e-bf5a-f3e1e6befad7', 'url': 'https://api.firecrawl.dev/v1/crawl/c4a6a749-3445-454e-bf5a-f3e1e6befad7'}

它接受与 crawl_url 相同的参数和抓取选项，但返回抓取状态字典。

我们最感兴趣的是抓取作业 ID，可以使用 check_crawl_status 检查进程的状态：

checkpoint = app.check_crawl_status(crawl_status['id'])

print(len(checkpoint['data']))

check_crawl_status 返回的输出与 crawl_url 相同，但仅包含迄今为止抓取的页面。你可以多次运行它，并看到抓取的页面数量在增加。

如果你想取消作业，可以使用 cancel_crawl 传递作业 ID：

final_result = app.cancel_crawl(crawl_status['id'])

print(final_result)

{'status': 'cancelled'}

4.3 异步抓取的好处

与 crawl_url 相比，使用 async_crawl_url 有很多优势：

你可以创建多个抓取作业，而无需等待每个作业完成。
你可以更有效地监控进度并管理资源。
非常适合批处理或并行抓取任务。
在后台进行抓取时，应用程序可以保持响应
用户监控进度，而不是等待完成
允许实现进度条或状态更新
更容易与消息队列或作业调度程序集成
可以成为更大的自动化工作流程的一部分
更适合微服务架构

实际上，你几乎总是对大型网站使用异步抓取。

5、如何保存Web 抓取结果

抓取大型网站时，持久保存结果很重要。Firecrawl 以结构化格式提供抓取的数据，可以轻松保存到各种存储系统中。让我们探索一些常见的方法。

5.1 本地文件存储

最简单的方法是保存到本地文件。以下是如何以不同格式保存抓取的内容：

import json
from pathlib import Path

def save_crawl_results(crawl_result, output_dir="firecrawl_output"):
    # Create output directory if it doesn't exist
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    # Save full results as JSON
    with open(f"{output_dir}/full_results.json", "w") as f:
        json.dump(crawl_result, f, indent=2)
    # Save just the markdown content in separate files
    for idx, page in enumerate(crawl_result["data"]):
        # Create safe filename from URL
        filename = (
            page["metadata"]["url"].split("/")[-1].replace(".html", "") or f"page_{idx}"
        )
        # Save markdown content
        if "markdown" in page:
            with open(f"{output_dir}/{filename}.md", "w") as f:
                f.write(page["markdown"])

上述函数的作用如下：

如果不存在，则创建一个输出目录
将完整的抓取结果保存为具有适当缩进的 JSON 文件
对于每个抓取的页面：根据页面 URL 生成文件名，将 markdown 内容保存到单独的 .md 文件中

app = FirecrawlApp()

crawl_result = app.crawl_url(url="https://docs.stripe.com/api", params={"limit": 10})
save_crawl_results(crawl_result)

这是一项基本函数，需要针对其他抓取格式进行修改。

5.2 数据库存储

对于更复杂的应用程序，你可能希望将结果存储在数据库中。以下是使用 SQLite 的示例：

import sqlite3

def save_to_database(crawl_result, db_path="crawl_results.db"):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    # Create table if it doesn't exist
    cursor.execute(
        """
        CREATE TABLE IF NOT EXISTS pages (
            url TEXT PRIMARY KEY,
            title TEXT,
            content TEXT,
            metadata TEXT,
            crawl_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
    """
    )
    # Insert pages
    for page in crawl_result["data"]:
        cursor.execute(
            "INSERT OR REPLACE INTO pages (url, title, content, metadata) VALUES (?, ?, ?, ?)",
            (
                page["metadata"]["url"],
                page["metadata"]["title"],
                page.get("markdown", ""),
                json.dumps(page["metadata"]),
            ),
        )
    conn.commit()
    print(f"Saved {len(crawl_result['data'])} pages to {db_path}")
    conn.close()

该函数创建一个 SQLite 数据库，其中包含一个用于存储抓取数据的 pages 表。对于每个页面，它会保存 URL（作为主键）、标题、内容（markdown 格式）和元数据（JSON）。抓取日期会自动添加为时间戳。如果已存在具有相同 URL 的页面，则会将其替换为新数据。这提供了一种持久存储解决方案，以后可以轻松查询。

save_to_database(crawl_result)

Saved 9 pages to crawl_results.db

让我们查询数据库以进行复查：

# Query the database
conn = sqlite3.connect("crawl_results.db")
cursor = conn.cursor()
cursor.execute("SELECT url, title, metadata FROM pages")
print(cursor.fetchone())
conn.close()

5.3 云存储

对于生产应用程序，你可能希望将结果存储在云存储中。以下是使用 AWS S3 的示例：

import boto3
from datetime import datetime

def save_to_s3(crawl_result, bucket_name, prefix="crawls"):
    s3 = boto3.client("s3")
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    # Save full results
    full_results_key = f"{prefix}/{timestamp}/full_results.json"
    s3.put_object(
        Bucket=bucket_name,
        Key=full_results_key,
        Body=json.dumps(crawl_result, indent=2),
    )
    # Save individual pages
    for idx, page in enumerate(crawl_result["data"]):
        if "markdown" in page:
            page_key = f"{prefix}/{timestamp}/pages/{idx}.md"
            s3.put_object(Bucket=bucket_name, Key=page_key, Body=page["markdown"])
    print(f"Successfully saved {len(crawl_result['data'])} pages to {bucket_name}/{full_results_key}")

以下是该函数的作用：

以抓取结果字典、S3 bucket 名称和可选前缀作为输入
在 S3 中创建带时间戳的文件夹结构以组织数据
将完整抓取结果保存为单个 JSON 文件
对于每个包含 markdown 内容的抓取页面，将其保存为单独的 .md 文件
使用 boto3 处理 AWS S3 交互
保留层次结构抓取数据

要使此功能正常工作，你必须安装 boto3 并将 AWS 凭证保存在 ~/.aws/credentials 文件中，格式如下：

[default]
aws_access_key_id = your_access_key
aws_secret_access_key = your_secret_key
region = your_region

然后，只要你已经有一个 S3 bucket 来存储数据，就可以执行该功能：

save_to_s3(crawl_result, "sample-bucket-1801", "stripe-api-docs")

Successfully saved 9 pages to sample-bucket-1801/stripe-api-docs/20241118_142945/full_results.json

5.4 使用异步抓取进行增量保存

使用异步抓取时，你可能希望在结果进入时以增量方式保存它们：

import time

def save_incremental_results(app, crawl_id, output_dir="firecrawl_output"):
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    processed_urls = set()
    while True:
        # Check current status
        status = app.check_crawl_status(crawl_id)
        # Save new pages
        for page in status["data"]:
            url = page["metadata"]["url"]
            if url not in processed_urls:
                filename = f"{output_dir}/{len(processed_urls)}.md"
                with open(filename, "w") as f:
                    f.write(page.get("markdown", ""))
                processed_urls.add(url)
        # Break if crawl is complete
        if status["status"] == "completed":
            print(f"Saved {len(processed_urls)} pages.")
            break
        time.sleep(5)  # Wait before checking again

以下是该函数的作用：

如果不存在，则创建一个输出目录
维护一组已处理的 URL 以避免重复
持续检查抓取状态直至完成
对于找到的每个新页面，将其 markdown 内容保存到编号文件中
在状态检查之间休眠 5 秒以避免过多的 API 调用

让我们在应用程序抓取 Books-to-Scrape 网站时使用它：

# Start the crawl
crawl_status = app.async_crawl_url(url="https://books.toscrape.com/")

# Save results incrementally
save_incremental_results(app, crawl_status["id"])

Saved 705 pages.

6、使用 Firecrawl 构建 AI 驱动的 Web 爬虫

Firecrawl 与 LangChain 等流行开源库和其他平台集成。

在本节中，我们将了解如何使用 LangChain 集成在 LangChain 社区集成网站上构建一个基本的 QA 聊天机器人。

首先安装 LangChain 及其相关库：

pip install langchain langchain_community langchain_anthropic langchain_openai

然后，将你的 ANTHROPIC_API_KEY 和 OPENAI_API_KEY 作为变量添加到你的 .env 文件中。

接下来，从文档加载器模块导入 FireCrawlLoader 类并初始化它：

from dotenv import load_dotenv
from langchain_community.document_loaders.firecrawl import FireCrawlLoader

load_dotenv()

loader = FireCrawlLoader(
    url="https://python.langchain.com/docs/integrations/providers/",
    mode="crawl",
    params={"limit": 5, "scrapeOptions": {"onlyMainContent": True}},
)

由于我们使用 load_dotenv() 加载变量，因此该类可以自动读取你的 Firecrawl API 密钥。

要开始抓取，可以调用 loader 对象的 load() 方法，抓取的内容将转换为与 LangChain 兼容的文档：

# Start the crawl
docs = loader.load()

上面，我们使用 RecursiveCharacterTextSplitter 将文档拆分成更小的块。这有助于使文本更易于处理，并确保在创建嵌入和执行检索时获得更好的结果。1,000 个字符的块大小和 100 个字符的重叠在上下文保留和粒度之间提供了良好的平衡。

from langchain_chroma import Chroma
from langchain.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores.utils import filter_complex_metadata

# Create embeddings for the documents
embeddings = OpenAIEmbeddings()

# Create a vector store from the loaded documents
docs = filter_complex_metadata(docs)
vector_store = Chroma.from_documents(docs, embeddings)

接下来，我们使用 Chroma 和 OpenAI 嵌入创建一个向量存储。向量存储支持对我们的文档进行语义搜索和检索。我们还过滤掉可能导致存储问题的复杂元数据。

最后一步是使用 Claude 3.5 Sonnet 作为语言模型构建 QA 链：

from langchain.chains import RetrievalQA
from langchain_anthropic import ChatAnthropic

# Initialize the language model
llm = ChatAnthropic(model="claude-3-5-sonnet-20240620", streaming=True)
# Create a QA chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vector_store.as_retriever(),
)

现在，我们可以询问有关文档的问题：

# Example question
query = "What is the main topic of the website?"
answer = qa_chain.invoke(query)

print(answer)

{
    'query': 'What is the main topic of the website?',
    'result': """The main topic of the website is LangChain's integrations with Hugging Face. 
    The page provides an overview of various LangChain components that can be used with 
    Hugging Face models and services, including:
    1. Chat models
    2. LLMs (Language Models) 
    3. Embedding models
    4. Document loaders
    5. Tools
    The page focuses on showing how to use different Hugging Face functionalities within 
    the LangChain framework, such as embedding models, language models, datasets, and 
    other tools."""
}

本节演示了为使用 Firecrawl 抓取的内容构建基本 RAG 管道的过程。对于此版本，我们仅使用了 LangChain 文档中的 10 页。随着信息量的增加，管道将需要进一步细化。为了有效地扩展此管道，我们需要考虑几个因素，包括：

分块策略优化
嵌入模型选择
向量存储性能调整
针对较大文档集合的快速工程

7、结束语

在本指南中，我们探索了 Firecrawl 的 /crawl 端点及其大规模网页抓取功能。从基本用法到高级配置，我们涵盖了 URL 控制、性能优化和异步操作。我们还研究了实际实现，包括数据存储解决方案和与 LangChain 等框架的集成。

端点能够处理 JavaScript 内容、分页和各种输出格式，使其成为满足现代网页抓取需求的多功能工具。无论你是构建文档聊天机器人还是收集训练数据，Firecrawl 都能提供强大的基础。通过利用讨论的配置选项和最佳实践，你可以构建高效且可扩展的网页抓取解决方案，以满足你的特定需求。

原文链接：Firecrawl: How to Scrape Entire Websites With a Single Command in Python

汇智网翻译整理，转载请标明出处