Crawl4AI：开源 LLM 友好型 Web 爬虫框架深度解析

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 + 语义过滤
                 ✅ 语义理解、智能提取

策略选择速查表

第七章：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
  Star & Fork 支持项目发展

• Discord: https://discord.gg/jP8KfhDhyN
  社区讨论、爬取技巧、AI 工作流分享

• 赞助: 支持持续开发和社区增长

  • 🌱 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 解析代码难以维护
• ❌ 处理动态内容痛苦不堪
• ❌ 难以扩展到大规模爬取

Crawl4AI：

• ✅ 简洁的两阶段配置，分离"全局设置"和"单次任务"
• ✅ 内置浏览器管理，自动处理生命周期
• ✅ 多种提取策略，从 CSS 选择器到 LLM 智能提取
• ✅ 生成干净的 Markdown，直接喂给 LLM
• ✅ 批量并发、自适应爬取、会话管理——开箱即用
• ✅ 完全开源，无强制 API 密钥，无付费墙

Crawl4AI 不只是一个工具，它是一种数据民主化的理念——让每个人都能自由地访问、提取和使用网络上的数据。

正如费曼所说："如果你不能向一个六岁的孩子解释清楚，那你就没有真正理解它。"我希望这篇解析能让你真正理解 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 #爬虫