Video Editing Tool Free
提供从赛道选择到交付的完整变现执行框架,助力个人创作者快速启动。
基于 Crawl4AI 的网页爬取与结构化数据提取,支持 Markdown 转换。
openclaw skills install @openlark/crawl4ai-web-crawler命令、参数、文件名以原文为准
[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。可通过 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)) # 过滤后 Markdownfrom 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)
)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 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/playgroundPython 客户端调用:
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())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,或配置代理。
已收录 26 个 Skill