#### 3.1.3 Schema 自动生成:AI 辅助的一次性工作
手动写 schema 可能比较繁琐。Crawl4AI 提供了一个巧妙的功能:用 LLM 一次性生成 schema,后续爬取就不需要再调用 LLM 了。
from crawl4ai import LLMConfig
# 让 LLM 分析页面结构,自动生成 schema
schema = JsonCssExtractionStrategy.generate_schema(
html=page_html,
llm_config=LLMConfig(
provider="openai/gpt-4o",
api_token="your-api-key"
)
)
# 保存 schema,后续重复使用
import json
with open("schema.json", "w") as f:
json.dump(schema, f)
# 后续爬取:零 LLM 调用
strategy = JsonCssExtractionStrategy(schema)
#### 3.1.4 RegexExtractionStrategy:正则提取
对于简单的模式匹配(邮箱、电话、URL 等),正则是最快的选择。
from crawl4ai import RegexExtractionStrategy
# 使用内置模式
strategy = RegexExtractionStrategy(
pattern = RegexExtractionStrategy.Email |
RegexExtractionStrategy.PhoneUS |
RegexExtractionStrategy.Url
)
# 内置模式包括:
# - Email, PhoneUS, PhoneIntl, Url, IPv4, DateIso, Currency...
3.2 LLM 策略:复杂内容的"智能大脑"
当页面结构不规则、语义复杂,或者你需要理解内容含义时,就需要动用 LLM 了。
from pydantic import BaseModel
from crawl4ai import LLMExtractionStrategy, LLMConfig
# 定义数据结构
class Product(BaseModel):
name: str
price: str
description: str = "" # 可选字段
# 配置 LLM 提取策略
llm_strategy = LLMExtractionStrategy(
llm_config=LLMConfig(
provider="openai/gpt-4o-mini",
api_token="your-api-key"
),
schema=Product.model_json_schema(),
extraction_type="schema",
instruction="Extract all products with name, price, and description",
# 分块策略(处理长页面)
chunk_token_threshold=1000, # 每块最大 token 数
overlap_rate=0.1, # 块间重叠率(保持上下文连贯)
apply_chunking=True, # 启用自动分块
input_format="markdown" # 输入格式:markdown | fit_markdown | html
)
#### 3.2.1 分块策略:处理超长页面
想象你在读一本厚厚的书,但你的助手一次只能记住 10 页。你会怎么做?
1. 把书分成多个 10 页的段落 2. 让助手一段一段读 3. 在段落之间保留一些重叠(比如每段重复前一段的最后 1 页),确保不会遗漏跨段落的上下文
这就是 Crawl4AI 的 chunk_token_threshold 和 overlap_rate 在做的事情。
# 示例:处理一个超长的文档页面
llm_strategy = LLMExtractionStrategy(
llm_config=LLMConfig(provider="openai/gpt-4o"),
schema=Article.schema_json(),
extraction_type="schema",
instruction="Extract key findings from this research paper",
chunk_token_threshold=2000, # GPT-4 能处理 8k,我们留安全余量
overlap_rate=0.15, # 15% 重叠,确保上下文不丢失
apply_chunking=True
)
# 提取后会自动合并各块的结果
#### 3.2.2 多提供商支持
Crawl4AI 使用 LiteLLM 作为后端,支持几乎所有主流 LLM 提供商:
# OpenAI
LLMConfig(provider="openai/gpt-4o", api_token="sk-...")
# Anthropic
LLMConfig(provider="anthropic/claude-3-opus-20240229", api_token="sk-ant-...")
# 本地模型(Ollama)
LLMConfig(provider="ollama/llama2")
# Google
LLMConfig(provider="gemini/gemini-pro", api_token="...")
# Azure
LLMConfig(provider="azure/gpt-4", api_token="...", base_url="...")
3.3 Markdown 生成策略:RAG 的黄金原料
如果你在做 RAG(Retrieval-Augmented Generation),你需要的是干净、结构化的文本,而不是一团 HTML。
from crawl4ai.content_filter_strategy import PruningContentFilter, BM25ContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator
# 策略 1:剪枝过滤器(基于启发式规则)
pruning_filter = PruningContentFilter(
threshold=0.48, # 内容质量阈值
threshold_type="fixed", # 固定阈值(相对阈值用 "dynamic")
min_word_threshold=50 # 最小词数
)
# 策略 2:BM25 过滤器(基于查询相关性)
bm25_filter = BM25ContentFilter(
user_query="machine learning deployment",
bm25_threshold=1.0
)
# 创建 Markdown 生成器
md_generator = DefaultMarkdownGenerator(content_filter=pruning_filter)
config = CrawlerRunConfig(markdown_generator=md_generator)
三种 Markdown 输出:
result = await crawler.arun(url="...", config=config)
# 原始 Markdown(完整内容)
raw = result.markdown.raw_markdown
# 过滤后的 Markdown(去除噪音)
fit = result.markdown.fit_markdown
# 带引用的 Markdown(链接转为编号引用)
cited = result.markdown.markdown_with_citations
---
第四章:实战教程——从入门到精通
4.1 入门示例:爬取博客文章
import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
async def scrape_blog():
browser_config = BrowserConfig(headless=True)
run_config = CrawlerRunConfig(
cache_mode=CacheMode.ENABLED
)
async with AsyncWebCrawler(config=browser_config) as crawler:
result = await crawler.arun(
url="https://docs.crawl4ai.com/",
config=run_config
)
if result.success:
print(f"✅ 成功爬取: {result.url}")
print(f"📄 标题: {result.metadata.get('title', 'N/A')}")
print(f"📝 Markdown 长度: {len(result.markdown.raw_markdown)} 字符")
# 保存到文件
with open("output.md", "w", encoding="utf-8") as f:
f.write(result.markdown.fit_markdown)
else:
print(f"❌ 失败: {result.error_message}")
asyncio.run(scrape_blog())
4.2 进阶示例:结构化数据提取
import asyncio
import json
from crawl4ai import AsyncWebCrawler, JsonCssExtractionStrategy
async def extract_products():
schema = {
"name": "Product List",
"baseSelector": "div.product-card",
"fields": [
{"name": "name", "selector": "h3.title", "type": "text"},
{"name": "price", "selector": "span.price", "type": "text"},
{"name": "rating", "selector": "div.rating", "type": "text"},
{"name": "link", "selector": "a", "type": "attribute", "attribute": "href"}
]
}
strategy = JsonCssExtractionStrategy(schema)
browser_config = BrowserConfig(headless=True)
run_config = CrawlerRunConfig(
extraction_strategy=strategy
)
async with AsyncWebCrawler(config=browser_config) as crawler:
result = await crawler.arun(
url="https://example-shop.com/products",
config=run_config
)
if result.success:
# 提取的内容是 JSON 字符串
products = json.loads(result.extracted_content)
print(f"提取到 {len(products)} 个产品")
print(json.dumps(products[:3], indent=2, ensure_ascii=False))
asyncio.run(extract_products())
4.3 高级示例:动态内容处理
import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
async def scrape_dynamic_content():
# JavaScript 代码:点击所有标签页
js_code = """
(async () => {
const tabs = document.querySelectorAll(".tab-button");
for (let tab of tabs) {
tab.click();
await new Promise(r => setTimeout(r, 500));
}
// 滚动到页面底部加载更多内容
window.scrollTo(0, document.body.scrollHeight);
await new Promise(r => setTimeout(r, 1000));
})();
"""
run_config = CrawlerRunConfig(
js_code=[js_code], # 执行自定义 JS
wait_for="css:.loaded", # 等待加载完成标识
screenshot=True, # 截图留档
process_iframes=True # 处理 iframe
)
async with AsyncWebCrawler() as crawler:
result = await crawler.arun(
url="https://example.com/dynamic-page",
config=run_config
)
if result.screenshot:
# 保存截图
import base64
with open("screenshot.png", "wb") as f:
f.write(base64.b64decode(result.screenshot))
asyncio.run(scrape_dynamic_content())
---
第五章:高级特性——工业级爬虫的武器库
5.1 批量并发爬取:MemoryAdaptiveDispatcher
当你需要爬取数千个页面时,简单的 for 循环会慢得让人抓狂。Crawl4AI 提供了智能调度器:
import asyncio
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
async def batch_crawl():
urls = [
"https://example.com/page1",
"https://example.com/page2",
# ... 更多 URL
]
# 方式 1:流式处理(内存友好)
run_config = CrawlerRunConfig(stream=True)
async with AsyncWebCrawler() as crawler:
async for result in await crawler.arun_many(urls, config=run_config):
if result.success:
print(f"✅ {result.url}: {len(result.markdown.raw_markdown)} chars")
else:
print(f"❌ {result.url}: {result.error_message}")
# 方式 2:批量等待(简单直接)
run_config = CrawlerRunConfig(stream=False)
async with AsyncWebCrawler() as crawler:
results = await crawler.arun_many(urls, config=run_config)
successful = [r for r in results if r.success]
print(f"成功: {len(successful)}/{len(urls)}")
MemoryAdaptiveDispatcher 的智能之处:
- 监控系统内存使用
- 自适应调整并发度
- 遇到 429/503 自动退避重试
- 实时进度监控
5.2 自适应爬取(Adaptive Crawling)
这是 Crawl4AI 的杀手级功能之一。想象你在一个巨大的图书馆找资料,你不知道哪本书有你需要的答案。传统做法是"全量爬取"——把所有书都搬回家慢慢看。自适应爬取则像一位聪明的图书管理员:
1. 你先告诉她你要找什么(query)
2. 她根据书名和目录判断哪些书可能相关
3. 她先拿几本最可能相关的给你看
4. 如果发现信息还不够,她再去找更多
5. 当收集到足够信息时,她主动告诉你:"应该够了"
from crawl4ai import AdaptiveCrawler
async def adaptive_search():
async with AsyncWebCrawler() as crawler:
adaptive = AdaptiveCrawler(crawler)
result = await adaptive.digest(
start_url="https://docs.python.org/3/",
query="async context managers" # 你想了解的内容
)
# 查看统计信息
adaptive.print_stats()
print(f"置信度: {adaptive.confidence:.0%}")
# 获取最相关的内容
relevant_docs = adaptive.get_relevant_content(top_k=10)
for doc in relevant_docs:
print(f"- {doc['url']}: {doc['score']}")
asyncio.run(adaptive_search())
自适应爬取的特点:
- 自动停止:当收集到足够信息时自动停止,不浪费资源
- 智能链接选择:只跟踪与查询相关的链接
- 置信度评分:了解信息完整度
5.3 会话管理与身份验证
爬取需要登录的页面是爬虫的噩梦之一。Crawl4AI 让这件事变得简单:
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig
async def authenticated_crawl():
# 方法 1:使用 cookies
browser_config = BrowserConfig()
run_config = CrawlerRunConfig(
cookies={
"session_id": "abc123",
"auth_token": "xyz789"
}
)
# 方法 2:使用 headers(Bearer Token)
run_config = CrawlerRunConfig(
headers={
"Authorization": "Bearer your-jwt-token",
"X-Custom-Header": "custom-value"
}
)
# 方法 3:会话复用(登录一次,多次使用)
async with AsyncWebCrawler() as crawler:
# 第一步:登录
login_result = await crawler.arun(
url="https://example.com/login",
config=CrawlerRunConfig(
js_code=["document.querySelector('#user').value='admin';"]
)
)
# 第二步:复用同一会话访问受保护页面
protected_result = await crawler.arun(
url="https://example.com/protected"
# 自动携带登录后的 cookies
)
5.4 反爬虫绕过(Stealth Mode)
现代网站有各种反爬虫机制:检测自动化浏览器、分析行为模式等。Crawl4AI 提供了"隐身模式":
browser_config = BrowserConfig(
browser_type="undetected", # 使用 undetected Chrome
headless=True,
extra_args=[
"--disable-blink-features=AutomationControlled",
"--disable-web-security",
"--disable-features=IsolateOrigins,site-per-process"
]
)
# 或者手动设置防检测
run_config = CrawlerRunConfig(
# 模拟真实用户行为
wait_for="css:body", # 等待页面完全加载
delay_before_return_html=2.0, # 额外等待 2 秒
)
---
第六章:策略选择指南——决策树
面对不同的爬取任务,你应该选择哪种策略?以下是决策指南:
开始
│
├─ 页面结构是否清晰、有规律?
│ ├─ 是 → 使用 JsonCssExtractionStrategy 或 JsonXPathExtractionStrategy
│ │ ✅ 最快、零成本、零幻觉
│ │
│ └─ 否 → 数据类型是什么?
│ ├─ 简单模式(邮箱、电话、URL)
│ │ └─ 使用 RegexExtractionStrategy
│ │ ✅ 极速、无 API 调用
│ │
│ ├─ 复杂/非结构化内容
│ │ └─ 使用 LLMExtractionStrategy
│ │ ✅ AI 理解语义、处理不规则数据
│ │
│ └─ 需要理解内容语义(如提取"关于AI安全的内容")
└─ 使用 LLMExtractionStrategy + 语义过滤
✅ 语义理解、智能提取
策略选择速查表
| 场景 | 推荐策略 | 理由 |
|---|---|---|
| 电商产品列表 | JsonCssExtractionStrategy | 结构化、可并行、零成本 |
| 提取所有邮箱/电话 | RegexExtractionStrategy | 极速、无 API 调用 |
| 新闻文章正文 | Markdown + PruningContentFilter | 干净文本、适合 RAG |
| 不规则表格数据 | LLMExtractionStrategy | 理解表格语义 |
| 从文档中提取特定主题 | LLM + BM25ContentFilter | 语义相关性过滤 |
| 需要理解上下文关系 | LLMExtractionStrategy + Pydantic Schema | 关系推理 |
第七章:CrawlResult 输出结构全解析
result.url # 最终 URL(含重定向后的地址)
result.html # 原始 HTML
result.cleaned_html # 清理后的 HTML(移除了脚本、样式等)
result.markdown # Markdown 对象(包含多种格式)
result.success # 是否成功(布尔值)
result.status_code # HTTP 状态码
result.error_message # 错误信息(如果失败)
# 媒体资源
result.media = {
"images": [...], # 图片 URL 列表
"videos": [...], # 视频 URL 列表
"audios": [...] # 音频 URL 列表
}
# 链接分析
result.links = {
"internal": [...], # 站内链接
"external": [...] # 站外链接
}
# 提取的内容(如果你配置了提取策略)
result.extracted_content # JSON 字符串
# 元数据
result.metadata = {
"title": "...",
"description": "...",
"keywords": "...",
"author": "..."
}
# 截图(如果配置了 screenshot=True)
result.screenshot # base64 编码的图片
# PDF(如果配置了 pdf=True)
result.pdf # PDF 二进制内容
---
第八章:未来展望与社区生态
8.1 Crawl4AI Cloud API
Crawl4AI 即将推出 Cloud API(Closed Beta),目标是:
- 大规模网页提取
- 比现有方案更具成本效益
- 无需管理基础设施
8.2 社区与支持
- GitHub: https://github.com/unclecode/crawl4ai
- Discord: https://discord.gg/jP8KfhDhyN
- 赞助: 支持持续开发和社区增长
- 🌱 Believer ($5/mo)
- 🚀 Builder ($50/mo)
- 💼 Growing Team ($500/mo)
- 🏢 Data Infrastructure Partner ($2000/mo)
8.3 数据民主化的愿景
Crawl4AI 的使命不仅仅是做一个好用的爬虫工具。创始人 Unclecode 的愿景是:
> "解锁个人和企业数据的价值,将数字足迹转化为结构化、可交易的数据资产。"
这个愿景包含三个层次:
1. 开源工具:社区驱动的透明数据提取平台(现在已经实现) 2. 数字资产结构化:组织和评估数字知识的工具 3. 道德数据市场:一个安全、公平的数据交换平台
Crawl4AI 相信,AI 的未来应该由真实的人类知识驱动,数据创造者应该直接从他们的贡献中受益。
---
结语:为什么选择 Crawl4AI?
让我们回到开头的那个周五下午。
传统方案:
- ❌ 需要手动处理浏览器驱动
- ❌ 写大量等待和重试逻辑
- ❌ HTML 解析代码难以维护
- ❌ 处理动态内容痛苦不堪
- ❌ 难以扩展到大规模爬取
- ✅ 简洁的两阶段配置,分离"全局设置"和"单次任务"
- ✅ 内置浏览器管理,自动处理生命周期
- ✅ 多种提取策略,从 CSS 选择器到 LLM 智能提取
- ✅ 生成干净的 Markdown,直接喂给 LLM
- ✅ 批量并发、自适应爬取、会话管理——开箱即用
- ✅ 完全开源,无强制 API 密钥,无付费墙
正如费曼所说:"如果你不能向一个六岁的孩子解释清楚,那你就没有真正理解它。"我希望这篇解析能让你真正理解 Crawl4AI,并在你的下一个数据项目中大放异彩。
Happy Crawling! 🕸️🚀
---
参考资源
- 官方文档: https://docs.crawl4ai.com/
- GitHub: https://github.com/unclecode/crawl4ai
- Discord 社区: https://discord.gg/jP8KfhDhyN
- 官方示例: https://github.com/unclecode/crawl4ai/tree/main/docs/examples
#Crawl4AI #WebScraping #LLM #数据提取 #开源 #费曼风格 #Python #爬虫