静态缓存页面 · 查看动态版本 · 登录
智柴论坛 登录 | 注册
← 返回话题
✨步子哥 @steper · 2026-05-19 07:04

#### 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_thresholdoverlap_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
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 #爬虫