Crawl4AI Web Crawler

基于 Crawl4AI 的网页爬取与结构化数据提取,支持 Markdown 转换。

已扫描
适合谁
需要批量获取网页内容的研究人员、从事内容聚合与知识库构建的团队
不适合谁
无服务器部署能力的初学者、对网络请求或数据合规性不敏感的用户
国内可用性
需网络配置。可能需要网络配置或第三方服务可访问。
安装难度
新手友好(★☆☆)。基于终端操作、依赖、API Key 和本地环境要求的初步判断。

安装与下载

openclaw skills install @openlark/crawl4ai-web-crawler

Skill 说明

命令、参数、文件名以原文为准

Crawl4AI 网页爬虫

[Crawl4AI](https://github.com/unclecode/crawl4ai) 是一个开源且对大语言模型友好的网页爬虫工具,可在 GitHub 上获取。它能将网页内容转换为干净的 Markdown 或结构化 JSON 格式,非常适合用于 RAG(检索增强生成)、AI 代理和数据流水线。

有关详细的 API 参数说明,请参阅 [references/api-reference.md](references/api-reference.md)。

触发关键词

“抓取”、“爬取”、“提取网页”、“将网页转为 Markdown”、“结构化提取”等。

安装

pip install -U crawl4ai
crawl4ai-setup          # 自动安装 Playwright 浏览器
crawl4ai-doctor         # 验证安装是否成功

如果浏览器安装失败,可手动运行:

python -m playwright install --with-deps chromium

核心架构

三个核心类:

类名用途
AsyncWebCrawler主要的异步爬虫类,负责管理浏览器生命周期
BrowserConfig浏览器配置(无头模式、用户代理、代理、视口等)
CrawlerRunConfig每次爬取的配置(缓存模式、提取策略、JavaScript 执行、截图等)

基本用法

最简爬取

import asyncio
from crawl4ai import AsyncWebCrawler

async def main():
    async with AsyncWebCrawler() as crawler:
        result = await crawler.arun(url="https://example.com")
        print(result.markdown)  # 适合 LLM 使用的 Markdown 内容

asyncio.run(main())

带配置的爬取

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode

browser_cfg = BrowserConfig(headless=True, verbose=True)
run_cfg = CrawlerRunConfig(
    cache_mode=CacheMode.BYPASS,     # BYPASS=不使用缓存,ENABLED=启用,WRITE_ONLY,READ_ONLY
    css_selector="main.article",     # 仅提取指定区域的内容
    word_count_threshold=10,         # 过滤掉字数过少的文本块
    screenshot=True,                 # 截取屏幕截图
)

async with AsyncWebCrawler(config=browser_cfg) as crawler:
    result = await crawler.arun(url="https://example.com", config=run_cfg)
    print(result.markdown)
    if result.screenshot:
        print(f"截图大小:{len(result.screenshot)} 字节(Base64 编码)")

命令行工具

# 基础爬取
crwl https://example.com -o markdown

# 深度爬取(广度优先,最多 10 页)
crwl https://docs.crawl4ai.com --deep-crawl bfs --max-pages 10

# 使用 LLM 提取内容
crwl https://example.com/products -q "提取所有商品价格"

Markdown 生成

使用内容过滤器

默认生成原始 Markdown。可通过 DefaultMarkdownGenerator 配合内容过滤器获得更清晰的结果:

from crawl4ai.content_filter_strategy import PruningContentFilter, BM25ContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

# 方法一:PruningContentFilter —— 基于密度的清理
md_gen = DefaultMarkdownGenerator(
    content_filter=PruningContentFilter(
        threshold=0.48,           # 0-1;值越低,清理越多
        threshold_type="fixed",   # "fixed" 或 "dynamic"
        min_word_threshold=0
    )
)

# 方法二:BM25ContentFilter —— 基于查询相关性的过滤
md_gen = DefaultMarkdownGenerator(
    content_filter=BM25ContentFilter(
        user_query="机器学习",  # 关注的关键字
        bm25_threshold=1.0
    )
)

run_cfg = CrawlerRunConfig(markdown_generator=md_gen)

async with AsyncWebCrawler() as crawler:
    result = await crawler.arun(url="...", config=run_cfg)
    print(len(result.markdown.raw_markdown))   # 原始 Markdown
    print(len(result.markdown.fit_markdown))   # 过滤后 Markdown

结构化数据提取

CSS/XPath 提取(无需 LLM,快速且免费)

from crawl4ai import JsonCssExtractionStrategy
import json

schema = {
    "name": "文章列表",
    "baseSelector": "article.post",     # 重复元素的容器选择器
    "fields": [
        {"name": "标题", "selector": "h2", "type": "text"},
        {"name": "链接", "selector": "a", "type": "attribute", "attribute": "href"},
        {"name": "图片", "selector": "img", "type": "attribute", "attribute": "src"},
    ]
}

run_cfg = CrawlerRunConfig(
    extraction_strategy=JsonCssExtractionStrategy(schema)
)

async with AsyncWebCrawler() as crawler:
    result = await crawler.arun(url="https://example.com/blog", config=run_cfg)
    data = json.loads(result.extracted_content)
    print(data)  # [{"标题": "...", "链接": "...", "图片": "..."}, ...]

自动生成 Schema(一次性 LLM 花费,之后可免费复用):

from crawl4ai import LLMConfig

schema = JsonCssExtractionStrategy.generate_schema(
    html="<div class='product'>...",
    llm_config=LLMConfig(provider="openai/gpt-4o", api_token="your-key")
    # 或使用本地模型:LLMConfig(provider="ollama/llama3.3", api_token=None)
)

LLM 提取(适用于非结构化内容)

from pydantic import BaseModel, Field
from crawl4ai import LLMExtractionStrategy, LLMConfig

class 产品(BaseModel):
    名称: str = Field(..., description="商品名称")
    价格: str = Field(..., description="价格字符串")
    描述: str = Field(..., description="简短描述")

llm_strategy = LLMExtractionStrategy(
    llm_config=LLMConfig(
        provider="openai/gpt-4o-mini",     # 支持 ollama/llama3、anthropic/claude-3 等
        api_token="your-api-key"
    ),
    schema=产品.model_json_schema(),
    extraction_type="schema",              # "schema" 或 "block"
    instruction="提取所有包含名称、价格和描述的商品对象。",
    chunk_token_threshold=1000,            # 超过此 token 数量时自动分块
    overlap_rate=0.1,                      # 块间重叠率 10%
    apply_chunking=True,
    input_format="markdown",               # "markdown" | "html" | "fit_markdown"
    extra_args={"temperature": 0.0, "max_tokens": 800}
)

run_cfg = CrawlerRunConfig(extraction_strategy=llm_strategy)

async with AsyncWebCrawler() as crawler:
    result = await crawler.arun(url="https://example.com/products", config=run_cfg)
    data = json.loads(result.extracted_content)
    llm_strategy.show_usage()  # 输出 token 使用统计

提取策略选择指南

场景推荐策略
重复列表(商品、文章、搜索结果)JsonCssExtractionStrategy
非结构化文本需 AI 理解LLMExtractionStrategy
高频爬取同一站点先用 LLM 生成 Schema,再通过 CSS 提取

动态页面处理

run_cfg = CrawlerRunConfig(
    js_code=[                          # 在页面上执行 JavaScript
        "window.scrollTo(0, document.body.scrollHeight)",
        "await new Promise(r => setTimeout(r, 2000))",
    ],
    wait_for="css:.content-loaded",     # 等待特定元素出现
    delay_before_return_html=2.0,       # 返回前额外等待时间(秒)
)

批量爬取

urls = ["https://example.com/page1", "https://example.com/page2", ...]

async with AsyncWebCrawler() as crawler:
    results = await crawler.arun_many(urls=urls, config=run_cfg)
    for result in results:
        if result.success:
            print(result.markdown[:200])

arun_many() 会自动处理速率限制、内存监控和并发控制。

浏览器管理

browser_cfg = BrowserConfig(
    browser_type="chromium",       # "chromium" | "firefox" | "webkit"
    headless=True,
    viewport_width=1920,
    viewport_height=1080,
    user_agent="Mozilla/5.0 ...",
    proxy="http://user:pass@proxy:8080",
    use_managed_browser=True,      # 使用已有浏览器实例
    user_data_dir="/path/to/profile",  # 保留登录状态的持久化配置目录
)

深度爬取(站点级爬取)

from crawl4ai import DeepCrawlStrategy, BFSDeepCrawlStrategy

deep_crawl = BFSDeepCrawlStrategy(
    max_depth=3,                    # 最大深度
    max_pages=50,                   # 最多爬取页面数
    include_paths=["/docs/*"],      # 仅爬取指定路径
    exclude_paths=["/blog/*"],      # 排除指定路径
)

run_cfg = CrawlerRunConfig(deep_crawl_strategy=deep_crawl)

async with AsyncWebCrawler() as crawler:
    results = await crawler.arun(url="https://example.com", config=run_cfg)
    for r in results:
        print(f"{r.url} → {len(r.markdown)} 字符")

Docker 部署

docker pull unclecode/crawl4ai:latest
docker run -d -p 11235:11235 --name crawl4ai --shm-size=1g unclecode/crawl4ai:latest

# 控制台:http://localhost:11235/dashboard
# 体验界面:http://localhost:11235/playground

Python 客户端调用:

import requests

resp = requests.post("http://localhost:11235/crawl",
    json={"urls": ["https://example.com"], "priority": 10})

task_id = resp.json()["task_id"]
result = requests.get(f"http://localhost:11235/task/{task_id}")
print(result.json())

CrawlResult 关键字段

result.url              # 最终 URL(经过重定向后)
result.html             # 原始 HTML
result.cleaned_html     # 清理后的 HTML
result.markdown         # Markdown 格式输出(包含 raw_markdown 和 fit_markdown)
result.extracted_content # 提取策略返回的 JSON 字符串
result.screenshot       # Base64 编码的截图
result.media            # 图片/视频信息
result.links            # 内部与外部链接信息
result.success          # 爬取是否成功
result.error_message    # 错误信息

常见问题

Playwright 浏览器未安装:

python -m playwright install --with-deps chromium

缓存导致返回旧数据:

设置 cache_mode=CacheMode.BYPASS 可跳过缓存。

动态内容未加载:

使用 wait_for="css:selector" 等待目标元素出现,或通过 js_code 执行滚动操作。

内存不足(批量爬取时):

降低并发数;arun_many() 会自动监控内存并调整。

反爬虫 / 检测机制:

BrowserConfig 中启用 use_managed_browser=True,或配置代理。

O
@openlark

已收录 26 个 Skill

相关推荐