Firecrawl 网页数据采集与 AI 爬虫完全教程
适用版本： Firecrawl v2.x（自托管 + 云服务）
最后更新： 2025 年 5 月
阅读时间： 约 25 分钟
目录
- Firecrawl 是什么
- 架构与核心原理
- 安装与部署
- 网页抓取 API
- 结构化数据提取
- LLM 驱动内容解析
- 批量采集工作流
- 与传统爬虫方案对比
- 反爬虫处理策略
- 数据清洗与格式化
- 与 RAG 系统集成
- 企业级部署方案
- 最佳实践与注意事项
- 总结
1. Firecrawl 是什么
Firecrawl 是一个专为 AI 应用设计的网页数据采集引擎。它不是传统的爬虫框架，而是一个将任意网页转化为 LLM 可直接消费的干净数据的服务层。
1.1 核心价值
传统爬虫的工作流程通常是：下载 HTML → 编写选择器 → 解析 DOM → 清洗数据 → 存储。每换一个网站就要重写解析逻辑。Firecrawl 彻底改变了这个范式：
你只需要告诉它一个 URL，它返回干净的 Markdown 或结构化数据。
# 传统爬虫：几十行代码解析一个网站
# Firecrawl：两行代码搞定
from firecrawl import FirecrawlApp
app = FirecrawlApp(api_key="your-key")
result = app.scrape_url("https://example.com/article")
print(result["markdown"]) # 干净的 Markdown 内容
1.2 核心能力一览
| 能力 | 说明 |
|---|---|
| Scrape（抓取） | 单页面抓取，返回 Markdown / HTML / 结构化数据 |
| Crawl（爬取） | 多页面递归爬取，自动发现子链接 |
| Map（映射） | 快速获取网站的 URL 结构图 |
| Extract（提取） | 基于 Schema 的结构化数据提取 |
| Batch（批量） | 异步批量抓取，支持数千 URL 并发 |
| LLM 解析 | 自动使用 LLM 理解页面语义 |
1.3 适用场景
- RAG 知识库构建：将网站内容转化为向量数据库的语料
- 竞品监控：定期抓取竞品页面，提取价格、功能变化
- 内容聚合：从多个来源采集内容，构建信息平台
- 研究数据采集：学术论文、新闻、公开数据的批量采集
- SEO 分析：爬取目标站点，分析页面结构与关键词
2. 架构与核心原理
2.1 系统架构
Firecrawl 的设计哲学是"浏览器即服务"。它内部管理了一套浏览器集群来渲染 JavaScript 密集型页面。
┌──────────────────────────────────────────────────────────┐
│ Firecrawl 服务层 │
├─────────────┬──────────────┬──────────────┬──────────────┤
│ Scrape API │ Crawl API │ Map API │ Extract API │
├─────────────┴──────────────┴──────────────┴──────────────┤
│ 核心引擎层 │
│ ┌───────────┐ ┌───────────┐ ┌────────────────────┐ │
│ │ 浏览器集群 │ │ HTML 解析器│ │ LLM 解析管线 │ │
│ │ (Playwright)│ │ (Turndown) │ │ (Markdown + 结构化) │ │
│ └───────────┘ └───────────┘ └────────────────────┘ │
├──────────────────────────────────────────────────────────┤
│ ┌───────────┐ ┌───────────┐ ┌────────────────────┐ │
│ │ 代理管理 │ │ 任务队列 │ │ 缓存与去重 │ │
│ │ (Proxy) │ │ (Redis) │ │ (Cache Layer) │ │
│ └───────────┘ └───────────┘ └────────────────────┘ │
└──────────────────────────────────────────────────────────┘
2.2 工作流程
- 请求接收：客户端发送 URL 和配置参数
- 浏览器渲染：使用 Playwright 无头浏览器加载页面，执行 JavaScript
- 内容提取：从渲染后的 DOM 中提取正文内容
- Markdown 转换：使用 Turndown 等库将 HTML 转换为干净的 Markdown
- LLM 增强（可选）：使用 LLM 进一步清洗、提取结构化信息
- 返回结果：以 JSON 格式返回给客户端
2.3 为什么选择浏览器渲染
现代网站大量使用 JavaScript 渲染（React、Vue、Next.js、SPA 应用等）。传统的 requests + BeautifulSoup 方案只能获取初始 HTML，无法获取动态加载的内容。Firecrawl 使用真实浏览器渲染，可以：
- 处理 SPA（单页应用）的动态内容
- 执行页面交互（滚动、点击加载更多）
- 获取渲染后的最终 DOM
- 处理需要登录才能访问的页面
3. 安装与部署
3.1 云服务（快速开始）
最简单的方式是使用 Firecrawl 的云服务，无需自行部署：
pip install firecrawl
from firecrawl import FirecrawlApp
app = FirecrawlApp(api_key="fc-your-api-key")
前往 firecrawl.dev 注册获取 API Key。免费层级包含每月 500 次抓取配额。
3.2 自托管部署（Docker）
对于企业级场景或数据安全要求高的场景，可以自托管部署：
# 克隆仓库
git clone https://github.com/mendableai/firecrawl
cd firecrawl
# 使用 Docker Compose 启动
docker compose up -d
3.3 自托管配置
# docker-compose.yml 中的关键环境变量
services:
firecrawl:
environment:
# Redis 连接
REDIS_URL: redis://redis:6379
# 浏览器配置
NUM_WORKERS: 4 # 并发浏览器实例数
MAX_BROWSERS: 10 # 最大浏览器数
# 代理配置（可选）
PROXY_SERVER: "" # 代理服务器地址
# API 配置
PORT: 3002
# LLM 配置（用于 Extract 功能）
OPENAI_API_KEY: "sk-..." # 或使用其他 LLM 提供商
# 认证
API_KEY: "your-secret-key"
3.4 验证安装
from firecrawl import FirecrawlApp
# 指定自托管地址
app = FirecrawlApp(
api_key="your-secret-key",
api_url="http://localhost:3002"
)
# 测试抓取
result = app.scrape_url("https://example.com")
print(result["metadata"]["title"])
print(f"内容长度: {len(result['markdown'])} 字符")
4. 网页抓取 API
scrape_url 是最基础的操作，用于抓取单个页面。
4.1 基本抓取
from firecrawl import FirecrawlApp
app = FirecrawlApp(api_key="your-key")
# 最简用法
result = app.scrape_url("https://news.ycombinator.com")
# 返回结构
# {
# "markdown": "# Hacker News\n\n...", # Markdown 格式正文
# "html": "<html>...</html>", # 原始 HTML
# "metadata": {
# "title": "Hacker News",
# "description": "...",
# "language": "en",
# "sourceURL": "https://...",
# "statusCode": 200
# }
# }
4.2 高级参数
result = app.scrape_url(
"https://example.com/article",
params={
# 内容格式
"formats": ["markdown", "html"],
# 只提取主内容（去除导航、页脚等）
"onlyMainContent": True,
# 等待特定元素加载（针对 SPA）
"waitFor": 2000, # 等待 2 秒
# 包含特定标签的内容
"includeTags": ["article", "main", "section"],
# 排除特定标签
"excludeTags": ["nav", "footer", "aside", ".ad"],
# 超时设置
"timeout": 30000, # 30 秒
}
)
4.3 处理动态页面
对于需要滚动加载的页面（如社交媒体 feed）：
result = app.scrape_url(
"https://example.com/infinite-scroll",
params={
"formats": ["markdown"],
"waitFor": 5000, # 给更多加载时间
"actions": [
{"type": "scroll", "direction": "down"},
{"type": "wait", "milliseconds": 2000},
{"type": "scroll", "direction": "down"},
]
}
)
4.4 实战案例：新闻聚合采集器
from firecrawl import FirecrawlApp
from datetime import datetime
import json
class NewsCollector:
def __init__(self, api_key: str):
self.app = FirecrawlApp(api_key=api_key)
def collect_article(self, url: str) -> dict:
"""采集单篇新闻"""
result = self.app.scrape_url(url, params={
"formats": ["markdown"],
"onlyMainContent": True,
"excludeTags": ["nav", "footer", ".comments", ".sidebar"],
})
return {
"url": url,
"title": result["metadata"].get("title", ""),
"content": result["markdown"],
"collected_at": datetime.utcnow().isoformat(),
"word_count": len(result["markdown"].split()),
}
def collect_batch(self, urls: list[str]) -> list[dict]:
"""批量采集新闻"""
articles = []
for url in urls:
try:
article = self.collect_article(url)
articles.append(article)
print(f"✅ {article['title'][:50]}...")
except Exception as e:
print(f"❌ {url}: {e}")
return articles
# 使用
collector = NewsCollector("fc-your-key")
articles = collector.collect_batch([
"https://techcrunch.com/2025/05/29/...",
"https://arstechnica.com/ai/2025/05/...",
])
# 保存
with open("articles.json", "w") as f:
json.dump(articles, f, ensure_ascii=False, indent=2)
5. 结构化数据提取
Firecrawl 的 extract 功能可以根据你提供的 Schema，从网页中自动提取结构化数据。
5.1 基于 Schema 提取
from firecrawl import FirecrawlApp
from pydantic import BaseModel
from typing import Optional
app = FirecrawlApp(api_key="your-key")
# 定义提取 Schema
class Product(BaseModel):
name: str
price: float
currency: str
description: str
rating: Optional[float]
review_count: Optional[int]
in_stock: bool
# 从产品页面提取
result = app.scrape_url(
"https://shop.example.com/product/123",
params={
"formats": ["extract"],
"extract": {
"schema": Product.model_json_schema(),
}
}
)
product_data = result["extract"]
print(f"商品: {product_data['name']}, 价格: {product_data['price']}")
5.2 多字段复杂提取
class JobPosting(BaseModel):
title: str
company: str
location: str
salary_range: Optional[str]
job_type: str # full-time, part-time, contract
requirements: list[str]
benefits: list[str]
posted_date: Optional[str]
result = app.scrape_url(
"https://jobs.example.com/posting/456",
params={
"formats": ["extract"],
"extract": {
"schema": JobPosting.model_json_schema(),
}
}
)
5.3 无 Schema 的自由提取
当页面结构不确定时，可以让 LLM 自行判断提取什么：
result = app.scrape_url(
"https://example.com/page",
params={
"formats": ["extract"],
"extract": {
"prompt": "提取页面中所有的价格信息、产品名称和促销活动，以 JSON 数组格式返回"
}
}
)
5.4 实战案例：竞品价格监控
import schedule
import time
from datetime import datetime
class PriceMonitor:
def __init__(self, api_key: str):
self.app = FirecrawlApp(api_key=api_key)
self.price_history = []
def check_price(self, url: str, product_name: str):
"""检查单个产品的当前价格"""
result = self.app.scrape_url(url, params={
"formats": ["extract"],
"extract": {
"schema": {
"type": "object",
"properties": {
"product_name": {"type": "string"},
"current_price": {"type": "number"},
"original_price": {"type": "number"},
"discount_percent": {"type": "number"},
"in_stock": {"type": "boolean"}
},
"required": ["product_name", "current_price"]
}
}
})
price_data = result["extract"]
record = {
**price_data,
"url": url,
"checked_at": datetime.utcnow().isoformat()
}
self.price_history.append(record)
# 检查价格变化
if len(self.price_history) > 1:
prev = self.price_history[-2]
if record["current_price"] != prev["current_price"]:
change = record["current_price"] - prev["current_price"]
print(f"⚠️ 价格变化: {product_name} {change:+.2f}")
return record
def monitor_loop(self, products: list[dict], interval_minutes: int = 60):
"""持续监控循环"""
def check_all():
for product in products:
self.check_price(product["url"], product["name"])
time.sleep(2) # 避免过快请求
check_all()
schedule.every(interval_minutes).minutes.do(check_all)
while True:
schedule.run_pending()
time.sleep(60)
6. LLM 驱动内容解析
Firecrawl 集成了 LLM 能力，可以在抓取的同时进行智能内容解析。
6.1 智能内容清洗
对于内容杂乱的页面，LLM 可以自动清洗并组织内容：
result = app.scrape_url(
"https://messy-forum.example.com/thread/789",
params={
"formats": ["markdown"],
"onlyMainContent": True,
# LLM 自动去除广告、无关内容
"extractorOptions": {
"mode": "llm-extraction"
}
}
)
6.2 页面摘要生成
result = app.scrape_url(
"https://long-article.example.com/post",
params={
"formats": ["markdown", "extract"],
"extract": {
"prompt": "请生成这篇文章的中文摘要，不超过200字，包含核心论点和结论"
}
}
)
print(result["extract"]) # LLM 生成的摘要
6.3 多语言内容处理
# 抓取外文页面并翻译
result = app.scrape_url(
"https://jp.example.com/article",
params={
"formats": ["extract"],
"extract": {
"prompt": "将这篇文章翻译为中文，保留原文的结构和格式"
}
}
)
6.4 实战案例：智能知识库构建器
from firecrawl import FirecrawlApp
import json
class KnowledgeBaseBuilder:
def __init__(self, api_key: str):
self.app = FirecrawlApp(api_key=api_key)
self.entries = []
def ingest_url(self, url: str, topic: str):
"""抓取并智能解析页面，构建知识条目"""
result = self.app.scrape_url(url, params={
"formats": ["markdown", "extract"],
"onlyMainContent": True,
"extract": {
"schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"summary": {"type": "string", "description": "200字以内的中文摘要"},
"key_points": {
"type": "array",
"items": {"type": "string"},
"description": "3-5个核心知识点"
},
"tags": {
"type": "array",
"items": {"type": "string"},
"description": "相关标签"
},
"difficulty": {
"type": "string",
"enum": ["beginner", "intermediate", "advanced"]
}
},
"required": ["title", "summary", "key_points"]
}
}
})
entry = {
"source_url": url,
"topic": topic,
"markdown_content": result["markdown"],
"metadata": result.get("extract", {}),
"indexed_at": datetime.utcnow().isoformat()
}
self.entries.append(entry)
return entry
def build_from_sitemap(self, base_url: str, topic: str):
"""从站点地图批量构建知识库"""
# 使用 Map API 获取站点所有 URL
site_map = app.map_url(base_url, params={
"limit": 100
})
for url in site_map.get("urls", []):
try:
self.ingest_url(url, topic)
print(f"✅ 已索引: {url}")
except Exception as e:
print(f"⚠️ 跳过: {url} ({e})")
def export_for_rag(self, output_path: str):
"""导出为 RAG 系统可用的格式"""
chunks = []
for entry in self.entries:
# 按段落切分长文本
content = entry["markdown_content"]
paragraphs = content.split("\n\n")
for i, para in enumerate(paragraphs):
if len(para.strip()) > 50: # 过滤过短的段落
chunks.append({
"text": para.strip(),
"source": entry["source_url"],
"title": entry["metadata"].get("title", ""),
"topic": entry["topic"],
"chunk_index": i
})
with open(output_path, "w") as f:
json.dump(chunks, f, ensure_ascii=False, indent=2)
print(f"导出 {len(chunks)} 个文本块到 {output_path}")
7. 批量采集工作流
7.1 Crawl API：递归爬取
Crawl 可以自动发现并爬取网站内的所有子页面：
from firecrawl import FirecrawlApp
app = FirecrawlApp(api_key="your-key")
# 启动异步爬取任务
crawl_result = app.crawl_url(
"https://docs.example.com",
params={
"limit": 50, # 最多爬取 50 页
"scrapeOptions": {
"formats": ["markdown"],
"onlyMainContent": True,
},
# URL 过滤
"includePaths": ["/docs/.*"], # 只爬取 /docs/ 路径
"excludePaths": ["/changelog"], # 排除 changelog
},
poll_interval=5 # 每 5 秒检查进度
)
# 获取结果
for page in crawl_result["data"]:
print(f"📄 {page['metadata']['title']}")
print(f" URL: {page['metadata']['sourceURL']}")
print(f" 内容: {len(page['markdown'])} 字符\n")
7.2 Map API：快速站点映射
Map 是最快的获取网站所有 URL 的方式（不抓取内容）：
# 快速获取站点 URL 列表
site_map = app.map_url(
"https://blog.example.com",
params={
"limit": 200, # 最多返回 200 个 URL
"includeSubdomains": False,
"search": "tutorial", # 只返回包含 "tutorial" 的 URL
}
)
print(f"发现 {len(site_map['urls'])} 个 URL")
for url in site_map["urls"]:
print(f" - {url}")
7.3 异步批量抓取
对于大规模采集，使用异步 API 更高效：
import asyncio
from firecrawl import AsyncFirecrawlApp
async def batch_scrape(urls: list[str], concurrency: int = 5):
"""异步批量抓取"""
app = AsyncFirecrawlApp(api_key="your-key")
# 使用批量 API
batch_result = await app.async_batch_scrape_urls(
urls,
params={
"formats": ["markdown"],
"onlyMainContent": True,
}
)
job_id = batch_result["id"]
print(f"批量任务已启动: {job_id}")
# 轮询等待完成
while True:
status = await app.check_batch_scrape_status(job_id)
if status["status"] == "completed":
break
print(f"进度: {status['completed']}/{status['total']}")
await asyncio.sleep(10)
return status["data"]
# 使用
urls = [
"https://example.com/page1",
"https://example.com/page2",
# ... 数百个 URL
]
results = asyncio.run(batch_scrape(urls))
7.4 实战案例：全站文档采集管道
from firecrawl import FirecrawlApp
import json
import os
class DocumentationPipeline:
"""完整的文档采集处理管道"""
def __init__(self, api_key: str, output_dir: str = "./docs_output"):
self.app = FirecrawlApp(api_key=api_key)
self.output_dir = output_dir
os.makedirs(output_dir, exist_ok=True)
def discover_urls(self, base_url: str) -> list[str]:
"""第一步：发现所有文档页面"""
site_map = self.app.map_url(base_url, params={"limit": 500})
urls = site_map.get("urls", [])
print(f"🔍 发现 {len(urls)} 个页面")
return urls
def filter_urls(self, urls: list[str], patterns: list[str]) -> list[str]:
"""第二步：过滤不需要的 URL"""
import re
filtered = []
skip_patterns = ["/tag/", "/author/", "/page/", "#", "?page="]
for url in urls:
# 跳过不需要的页面
if any(skip in url for skip in skip_patterns):
continue
# 只保留匹配模式的 URL
if patterns and not any(re.search(p, url) for p in patterns):
continue
filtered.append(url)
print(f"📋 过滤后剩余 {len(filtered)} 个页面")
return filtered
def scrape_all(self, urls: list[str]) -> list[dict]:
"""第三步：批量抓取"""
results = []
total = len(urls)
for i, url in enumerate(urls, 1):
try:
result = self.app.scrape_url(url, params={
"formats": ["markdown"],
"onlyMainContent": True,
"excludeTags": ["nav", "footer", ".sidebar", ".breadcrumb"],
})
results.append({
"url": url,
"title": result["metadata"].get("title", ""),
"content": result["markdown"],
"word_count": len(result["markdown"].split()),
})
if i % 10 == 0:
print(f"⏳ 进度: {i}/{total}")
except Exception as e:
print(f"❌ 失败: {url} - {e}")
return results
def save_results(self, results: list[dict]):
"""第四步：保存结果"""
# 保存为 JSON
json_path = os.path.join(self.output_dir, "all_docs.json")
with open(json_path, "w") as f:
json.dump(results, f, ensure_ascii=False, indent=2)
# 保存为独立 Markdown 文件
md_dir = os.path.join(self.output_dir, "markdown")
os.makedirs(md_dir, exist_ok=True)
for i, doc in enumerate(results):
filename = doc["title"][:50].replace("/", "-").replace(" ", "_") or f"doc_{i}"
md_path = os.path.join(md_dir, f"{filename}.md")
with open(md_path, "w") as f:
f.write(f"# {doc['title']}\n\n")
f.write(f"> 来源: {doc['url']}\n\n")
f.write(doc["content"])
print(f"💾 已保存 {len(results)} 个文档到 {self.output_dir}")
def run(self, base_url: str, url_patterns: list[str] = None):
"""执行完整管道"""
print(f"🚀 开始采集: {base_url}")
urls = self.discover_urls(base_url)
filtered = self.filter_urls(urls, url_patterns or [])
results = self.scrape_all(filtered)
self.save_results(results)
total_words = sum(r["word_count"] for r in results)
print(f"\n✅ 采集完成!")
print(f" 页面数: {len(results)}")
print(f" 总字数: {total_words:,}")
# 使用
pipeline = DocumentationPipeline("fc-your-key")
pipeline.run(
"https://docs.example.com",
url_patterns=["/docs/guides/.*", "/docs/api/.*"]
)
8. 与传统爬虫方案对比
8.1 全面对比
| 对比维度 | Firecrawl | Scrapy | Playwright + BS4 | Selenium + BS4 |
|---|---|---|---|---|
| 学习曲线 | 极低 | 中等 | 中等 | 低 |
| JS 渲染 | ✅ 原生支持 | ❌ 需额外中间件 | ✅ 原生支持 | ✅ 原生支持 |
| HTML → Markdown | ✅ 自动转换 | ❌ 手动实现 | ❌ 手动实现 | ❌ 手动实现 |
| 结构化提取 | ✅ LLM 驱动 | ❌ 手写选择器 | ❌ 手写选择器 | ❌ 手写选择器 |
| 反爬处理 | ✅ 内置代理 | ❌ 需自行实现 | ❌ 需自行实现 | ❌ 需自行实现 |
| 并发能力 | 高（云端） | 极高 | 中等 | 低 |
| 维护成本 | 低 | 高（选择器常变） | 高 | 高 |
| 数据质量 | 高（LLM 清洗） | 取决于编写者 | 取决于编写者 | 取决于编写者 |
| 适用场景 | AI/RAG 数据源 | 大规模通用爬虫 | 需精确控制的场景 | 简单自动化 |
8.2 代码对比
传统 Scrapy 方式（以新闻网站为例）：
# Scrapy: 需要为每个网站写选择器
class NewsSpider(scrapy.Spider):
name = "news"
def parse(self, response):
# 每个网站的选择器都不同，维护成本高
for article in response.css("article.post"):
yield {
"title": article.css("h2.title::text").get(),
"content": "\n".join(article.css("div.body p::text").getall()),
"date": article.css("time::attr(datetime)").get(),
"author": article.css("span.author::text").get(),
}
# 翻页逻辑
next_page = response.css("a.next::attr(href)").get()
if next_page:
yield response.follow(next_page, self.parse)
Firecrawl 方式：
# Firecrawl: 无需了解页面结构
result = app.scrape_url(
"https://news.example.com/article/123",
params={
"formats": ["markdown", "extract"],
"onlyMainContent": True,
"extract": {
"schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"content": {"type": "string"},
"date": {"type": "string"},
"author": {"type": "string"},
}
}
}
}
)
# 完成！无需维护选择器
8.3 选择建议
- 选 Firecrawl：AI 应用数据源、RAG 知识库、快速原型、不想维护爬虫代码
- 选 Scrapy：需要百万级采集量、非 JS 页面为主、需要极致性能控制
- 选 Playwright：需要模拟复杂用户交互、需要精确控制浏览器行为
9. 反爬虫处理策略
9.1 Firecrawl 内置的反爬能力
Firecrawl 在底层已经处理了大量反爬措施：
- 自动代理轮换：云服务内置代理池
- 浏览器指纹管理：每次使用不同的 User-Agent 和浏览器配置
- 自动重试：遇到临时封锁自动重试
- 速率控制：内置请求间隔控制
9.2 自托管的反爬配置
# 配置代理
app = FirecrawlApp(
api_key="your-key",
api_url="http://localhost:3002"
)
# 抓取时指定代理
result = app.scrape_url(
"https://protected-site.com/page",
params={
"formats": ["markdown"],
"proxy": {
"server": "http://proxy-server:8080",
"username": "proxy-user",
"password": "proxy-pass"
},
# 模拟真实用户行为
"waitFor": 3000,
"headers": {
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
"Referer": "https://www.google.com/"
}
}
)
9.3 速率限制与礼貌爬取
import time
from dataclasses import dataclass
@dataclass
class PoliteCrawler:
"""礼貌爬虫：控制速率，避免给目标网站造成压力"""
api_key: str
requests_per_minute: int = 20
min_delay: float = 3.0
def __post_init__(self):
self.app = FirecrawlApp(api_key=self.api_key)
self.last_request_time = 0
def _wait(self):
"""确保请求间隔"""
elapsed = time.time() - self.last_request_time
if elapsed < self.min_delay:
time.sleep(self.min_delay - elapsed)
self.last_request_time = time.time()
def scrape(self, url: str, **kwargs) -> dict:
"""带速率控制的抓取"""
self._wait()
for attempt in range(3):
try:
result = self.app.scrape_url(url, params=kwargs)
return result
except Exception as e:
if "429" in str(e) or "rate" in str(e).lower():
wait = 60 * (attempt + 1)
print(f"⏳ 触发限流，等待 {wait}s...")
time.sleep(wait)
else:
raise
raise Exception(f"重试耗尽: {url}")
9.4 登录页面抓取
对于需要认证的页面：
# 使用 cookies 抓取登录后的页面
result = app.scrape_url(
"https://dashboard.example.com/data",
params={
"formats": ["markdown"],
"cookies": [
{"name": "session_id", "value": "abc123", "domain": ".example.com"},
{"name": "auth_token", "value": "xyz789", "domain": ".example.com"},
],
"headers": {
"Authorization": "Bearer your-token-here"
}
}
)
10. 数据清洗与格式化
10.1 Markdown 清洗
Firecrawl 返回的 Markdown 通常已经很干净，但有时还需要进一步处理：
import re
def clean_markdown(text: str) -> str:
"""进一步清洗 Markdown 内容"""
# 移除多余的空行
text = re.sub(r'\n{3,}', '\n\n', text)
# 移除图片链接（如不需要）
text = re.sub(r'!\[.*?\]\(.*?\)', '', text)
# 移除 HTML 标签残留
text = re.sub(r'<[^>]+>', '', text)
# 移除广告相关文本
ad_patterns = [
r'(?i)(advertisement|sponsored|promoted)',
r'(?i)(subscribe to|sign up for|newsletter)',
r'(?i)(cookie policy|privacy policy|terms of service)',
]
for pattern in ad_patterns:
text = re.sub(pattern, '', text)
# 标准化空白
text = re.sub(r'[ \t]+', ' ', text)
text = text.strip()
return text
10.2 文本分块策略
将长文档切分为适合 RAG 系统使用的块：
def chunk_text(
text: str,
max_chunk_size: int = 1000,
overlap: int = 100,
split_on: str = "\n\n"
) -> list[dict]:
"""将长文本分块，保留元数据"""
paragraphs = text.split(split_on)
chunks = []
current_chunk = []
current_size = 0
for para in paragraphs:
para_size = len(para)
if current_size + para_size > max_chunk_size and current_chunk:
# 保存当前块
chunk_text = split_on.join(current_chunk)
chunks.append({
"text": chunk_text,
"char_count": len(chunk_text),
"paragraph_count": len(current_chunk)
})
# 保留重叠部分
if overlap > 0:
overlap_text = chunk_text[-overlap:]
current_chunk = [overlap_text]
current_size = len(overlap_text)
else:
current_chunk = []
current_size = 0
current_chunk.append(para)
current_size += para_size
# 处理最后一个块
if current_chunk:
chunk_text = split_on.join(current_chunk)
chunks.append({
"text": chunk_text,
"char_count": len(chunk_text),
"paragraph_count": len(current_chunk)
})
return chunks
10.3 实战案例：数据清洗管道
from dataclasses import dataclass
from typing import Optional
@dataclass
class CleanedDocument:
url: str
title: str
content: str
summary: Optional[str]
word_count: int
language: str
chunks: list[dict]
def process_document(raw_result: dict, chunk_size: int = 800) -> CleanedDocument:
"""完整的数据清洗管道"""
# 1. 提取原始内容
markdown = raw_result.get("markdown", "")
metadata = raw_result.get("metadata", {})
# 2. 清洗
cleaned = clean_markdown(markdown)
# 3. 检测语言
# 简单检测：中文字符占比
chinese_chars = len(re.findall(r'[\u4e00-\u9fff]', cleaned))
language = "zh" if chinese_chars / max(len(cleaned), 1) > 0.1 else "en"
# 4. 分块
chunks = chunk_text(cleaned, max_chunk_size=chunk_size)
# 5. 为每个块添加上下文
title = metadata.get("title", "未知标题")
for i, chunk in enumerate(chunks):
chunk["metadata"] = {
"source_url": raw_result.get("url", metadata.get("sourceURL", "")),
"title": title,
"chunk_index": i,
"total_chunks": len(chunks),
"language": language,
}
return CleanedDocument(
url=raw_result.get("url", metadata.get("sourceURL", "")),
title=title,
content=cleaned,
summary=None, # 可后续由 LLM 生成
word_count=len(cleaned.split()),
language=language,
chunks=chunks
)
11. 与 RAG 系统集成
Firecrawl 是构建 RAG（检索增强生成）知识库的理想数据源。
11.1 完整 RAG 数据管线
from firecrawl import FirecrawlApp
from typing import Optional
import hashlib
import json
class RAGDataPipeline:
"""从网页到向量数据库的完整管线"""
def __init__(self, firecrawl_key: str):
self.app = FirecrawlApp(api_key=firecrawl_key)
def ingest_url(self, url: str) -> list[dict]:
"""抓取单个 URL 并生成 RAG 块"""
result = self.app.scrape_url(url, params={
"formats": ["markdown"],
"onlyMainContent": True,
})
cleaned = clean_markdown(result["markdown"])
chunks = chunk_text(cleaned, max_chunk_size=800, overlap=100)
# 为每个块生成唯一 ID
for chunk in chunks:
content_hash = hashlib.md5(chunk["text"].encode()).hexdigest()[:12]
chunk["id"] = f"{url}_{content_hash}"
chunk["metadata"]["source_url"] = url
chunk["metadata"]["title"] = result["metadata"].get("title", "")
return chunks
def ingest_site(self, base_url: str, max_pages: int = 100) -> list[dict]:
"""爬取整个站点并生成 RAG 块"""
all_chunks = []
# 1. 获取站点地图
site_map = self.app.map_url(base_url, params={"limit": max_pages})
urls = site_map.get("urls", [])
# 2. 逐页抓取和处理
for url in urls:
try:
chunks = self.ingest_url(url)
all_chunks.extend(chunks)
print(f"✅ {url} → {len(chunks)} 个块")
except Exception as e:
print(f"⚠️ 跳过 {url}: {e}")
print(f"\n📊 总计: {len(all_chunks)} 个文本块来自 {len(urls)} 个页面")
return all_chunks
def export_for_vector_db(
self,
chunks: list[dict],
format: str = "jsonl",
output_path: str = "rag_data.jsonl"
):
"""导出为向量数据库可导入的格式"""
if format == "jsonl":
with open(output_path, "w") as f:
for chunk in chunks:
f.write(json.dumps(chunk, ensure_ascii=False) + "\n")
elif format == "langchain":
# LangChain Document 格式
from langchain.schema import Document
documents = [
Document(
page_content=chunk["text"],
metadata=chunk.get("metadata", {})
)
for chunk in chunks
]
return documents
print(f"💾 已导出到 {output_path}")
11.2 与主流向量数据库集成
# 示例：将数据导入 Pinecone
import pinecone
def upload_to_pinecone(chunks: list[dict], index_name: str):
"""将 Firecrawl 采集的文本块上传到 Pinecone"""
from sentence_transformers import SentenceTransformer
# 初始化嵌入模型
model = SentenceTransformer("all-MiniLM-L6-v2")
# 初始化 Pinecone
pinecone.init(api_key="your-pinecone-key", environment="us-west1-gcp")
index = pinecone.Index(index_name)
# 批量上传
batch_size = 100
for i in range(0, len(chunks), batch_size):
batch = chunks[i:i + batch_size]
vectors = []
for chunk in batch:
embedding = model.encode(chunk["text"]).tolist()
vectors.append({
"id": chunk["id"],
"values": embedding,
"metadata": {
"text": chunk["text"][:1000], # 限制元数据大小
**chunk.get("metadata", {})
}
})
index.upsert(vectors=vectors)
print(f"⬆️ 已上传 {min(i + batch_size, len(chunks))}/{len(chunks)}")
11.3 增量更新策略
import hashlib
import json
import os
class IncrementalIndexer:
"""增量索引：只处理新增或变更的页面"""
def __init__(self, state_file: str = "crawl_state.json"):
self.state_file = state_file
self.state = self._load_state()
def _load_state(self) -> dict:
if os.path.exists(self.state_file):
with open(self.state_file) as f:
return json.load(f)
return {"indexed_urls": {}}
def _save_state(self):
with open(self.state_file, "w") as f:
json.dump(self.state, f, indent=2)
def get_content_hash(self, content: str) -> str:
return hashlib.md5(content.encode()).hexdigest()
def should_update(self, url: str, content: str) -> bool:
"""检查 URL 的内容是否已变更"""
new_hash = self.get_content_hash(content)
old_hash = self.state["indexed_urls"].get(url, {}).get("hash")
if old_hash != new_hash:
self.state["indexed_urls"][url] = {
"hash": new_hash,
"updated_at": datetime.utcnow().isoformat()
}
return True
return False
def process_updates(self, urls: list[str], app: FirecrawlApp) -> list[dict]:
"""只处理有变化的页面"""
updated_chunks = []
for url in urls:
result = app.scrape_url(url, params={"formats": ["markdown"]})
content = result["markdown"]
if self.should_update(url, content):
chunks = chunk_text(clean_markdown(content))
for chunk in chunks:
chunk["metadata"] = {"source_url": url}
updated_chunks.extend(chunks)
print(f"🔄 已更新: {url}")
else:
print(f"⏭️ 未变化: {url}")
self._save_state()
return updated_chunks
12. 企业级部署方案
12.1 自托管架构
┌──────────────────────────────────────────────────────────────┐
│ Kubernetes 集群 │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Firecrawl API │ │ Firecrawl API │ │ Firecrawl API │ │
│ │ Pod 1 │ │ Pod 2 │ │ Pod 3 │ │
│ └──────┬───────┘ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │ │
│ ┌──────┴─────────────────┴─────────────────┴──────┐ │
│ │ Redis 集群（任务队列 + 缓存） │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ ┌─────────────────────────────────────────────────┐ │
│ │ Playwright 浏览器池 │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │ │
│ │ │Browser │ │Browser │ │Browser │ │Browser │ │ │
│ │ │ 1 │ │ 2 │ │ 3 │ │ 4 │ │ │
│ │ └────────┘ └────────┘ └────────┘ └────────┘ │ │
│ └─────────────────────────────────────────────────┘ │
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ Prometheus │ │ Grafana │ │ 代理池 │ │
│ │ 监控 │ │ 仪表板 │ │ (SquidProxy) │ │
│ └──────────────┘ └──────────────┘ └──────────────┘ │
└──────────────────────────────────────────────────────────────┘
12.2 Kubernetes 部署配置
# firecrawl-deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: firecrawl
spec:
replicas: 3
selector:
matchLabels:
app: firecrawl
template:
metadata:
labels:
app: firecrawl
spec:
containers:
- name: firecrawl
image: ghcr.io/mendableai/firecrawl:latest
ports:
- containerPort: 3002
env:
- name: REDIS_URL
value: "redis://redis-cluster:6379"
- name: NUM_WORKERS
value: "4"
- name: API_KEY
valueFrom:
secretKeyRef:
name: firecrawl-secrets
key: api-key
resources:
requests:
memory: "2Gi"
cpu: "1000m"
limits:
memory: "4Gi"
cpu: "2000m"
readinessProbe:
httpGet:
path: /health
port: 3002
initialDelaySeconds: 30
periodSeconds: 10
---
apiVersion: v1
kind: Service
metadata:
name: firecrawl
spec:
selector:
app: firecrawl
ports:
- port: 3002
targetPort: 3002
type: ClusterIP
12.3 监控与告警
# 监控指标采集示例
from prometheus_client import Counter, Histogram, start_http_server
# 定义指标
scrape_requests = Counter(
'firecrawl_scrape_total',
'Total scrape requests',
['status', 'domain']
)
scrape_duration = Histogram(
'firecrawl_scrape_duration_seconds',
'Scrape request duration',
['domain']
)
content_size = Histogram(
'firecrawl_content_size_bytes',
'Scraped content size',
['domain']
)
def monitored_scrape(url: str) -> dict:
"""带监控的抓取"""
from urllib.parse import urlparse
domain = urlparse(url).netloc
with scrape_duration.labels(domain=domain).time():
try:
result = app.scrape_url(url)
scrape_requests.labels(status="success", domain=domain).inc()
content_size.labels(domain=domain).observe(
len(result.get("markdown", "").encode())
)
return result
except Exception as e:
scrape_requests.labels(status="error", domain=domain).inc()
raise
# 启动 Prometheus 指标服务
start_http_server(8000)
12.4 成本估算
| 部署方式 | 月成本估算 | 适用规模 | 优势 |
|---|---|---|---|
| Firecrawl Cloud Free | $0 | 500 次/月 | 零运维 |
| Cloud Starter | $85 | 5,000 次/月 | 简单易用 |
| Cloud Growth | $340 | 50,000 次/月 | 适合中小团队 |
| 自托管（小规模） | ~$100（服务器） | 10,000+ 次/月 | 数据可控 |
| 自托管（大规模） | ~$500（K8s 集群） | 100,000+ 次/月 | 完全自主 |
13. 最佳实践与注意事项
✅ 最佳实践
- 先 Map 后 Scrape：先用 Map API 获取 URL 列表，筛选后再抓取，避免浪费配额
- 合理设置
onlyMainContent：绝大多数场景应设为True，去除导航栏、侧边栏等噪音 - 使用 excludeTags：排除广告、评论区、页脚等非核心内容
- 控制并发：即使是云服务，也不要瞬间发送大量请求
- 缓存结果：对同一 URL 的抓取结果做本地缓存
- 增量更新：实现内容哈希比对，只处理变化的页面
- 监控配额：定期检查 API 使用量，避免意外超出限制
❌ 常见陷阱
| 陷阱 | 说明 | 解决方案 |
|---|---|---|
| 未设置 onlyMainContent | 返回大量导航、页脚等噪音 | 始终设置为 True |
| 忽略 robots.txt | 违反网站爬取规则 | 先检查 robots.txt 合规性 |
| 不处理异常 | 单页失败导致整个任务中断 | 实现 try-except 和重试逻辑 |
| Token 爆炸 | 抓取大量页面后一次性发给 LLM | 流式处理，分批调用 LLM |
| 未去重 | 同一内容从多个 URL 采集 | 使用内容哈希去重 |
| 忽略速率限制 | 过快请求导致 IP 被封 | 实现礼貌爬取，控制请求间隔 |
| 硬编码 API Key | 安全风险 | 使用环境变量或 Secret Manager |
法律与道德注意
- 遵守 robots.txt：在爬取前检查目标网站的 robots.txt
- 尊重版权：抓取的内容可能有版权，注意合理使用
- 控制频率：不要对目标网站造成过大压力
- 隐私保护：不要采集和存储个人隐私信息
- 遵守 ToS：遵守目标网站的服务条款
14. 总结
核心要点
- Firecrawl 的核心价值是将任意网页转化为 LLM 可消费的干净数据
- Scrape 适合单页、Crawl 适合递归爬取、Map 适合快速发现 URL
- LLM 驱动的结构化提取是 Firecrawl 区别于传统爬虫的核心能力
- 自托管部署适合数据安全要求高的企业场景
- 与 RAG 系统集成是最主流的使用场景
典型工作流
发现 URL (Map) → 过滤 → 批量抓取 (Scrape) → 清洗 → 分块 → 嵌入 → 向量数据库
适用场景选择
| 需求 | 推荐方案 |
|---|---|
| 快速原型 / 低量级 | Firecrawl Cloud Free |
| RAG 知识库（中小规模） | Firecrawl Cloud + LangChain |
| 企业级大规模采集 | 自托管 Firecrawl + Kubernetes |
| 需要极致控制的特殊场景 | Playwright + 自定义逻辑 |
学习资源
本教程基于 Firecrawl v2.x 版本编写。API 细节可能随版本更新变化，请以官方文档为准。
