详解AI采集框架Crawl4AI，打造智能网络爬虫

大家好，Crawl4AI作为开源Python库，专门用来简化网页爬取和数据提取的工作。它不仅功能强大、灵活，而且全异步的设计让处理速度更快，稳定性更好。无论是构建AI项目还是提升语言模型的性能，Crawl4AI都能帮您简化工作流程。它可以直接在Python项目中使用，或者将其集成到REST API中，实现快速、稳定的数据爬取和处理。这样，无论是数据的实时获取还是后续的分析处理，都能更加得心应手。

1.快速使用

以下是个简单的例子，展示了Crawl4AI强大的异步能力：

import asyncio
from crawl4ai import AsyncWebCrawlerasyncdef main():# 初始化异步网页爬虫asyncwith AsyncWebCrawler(verbose=True) as crawler:# 爬取指定的 URLresult = await crawler.arun(url="https://www.nbcnews.com/business")# 以 Markdown 格式显示提取的内容print(result.markdown)# 执行异步主函数
if __name__ == "__main__":asyncio.run(main())

解释：

导入库：从crawl4ai库中导入AsyncWebCrawler和asyncio模块。
创建异步上下文：使用异步上下文管理器实例化AsyncWebCrawler。
运行爬虫：使用arun() 法异步爬取指定的 URL 并提取有意义的内容。
打印结果：输出提取的内容，格式化为 Markdown。
执行异步函数：使用asyncio.run()执行异步的main函数。

2.特性亮点

Crawl4AI具备以下核心特性，让网页爬取和数据提取工作更加高效：

开源免费：无额外费用，开源可信赖。
快速性能：速度超越许多付费工具。
多样输出：支持JSON、清洁HTML、Markdown格式。
多URL并发：一次性处理多个网页，提升效率。
媒体提取：全面抓取图片、音频、视频等。
链接全收集：不遗漏任何内外链接。
元数据抽取：深入提取网页信息。
自定义操作：自定义请求头、认证，修改页面后再爬取。
用户代理模拟：模拟不同设备访问。
页面截图：快速获取网页视觉快照。
JavaScript支持：执行JS获取动态内容。
数据结构化：精确提取结构化数据。
智能提取技术：使用余弦聚类和LLM技术。
CSS选择器：精准定位数据。
指令优化：通过指令提升提取效果。
代理配置：增强访问权限和隐私保护。
会话管理：轻松处理多页爬取。
异步架构：提升性能和可扩展性。

3.安装指南

Crawl4AI提供了多种安装方式，以适应不同的使用场景。以下是几种常用的安装方法：

3.1 基本安装（推荐）

对于大多数网页爬取和数据抓取任务，可以直接使用pip进行安装：

pip install crawl4ai

这样，默认安装的是Crawl4AI的异步版本，使用Playwright进行网页爬取。

如果安装时遇到Playwright相关错误，可以通过以下命令手动安装Playwright：

playwright install

或者，安装特定版本的Chromium：

python -m playwright install chromium

3.2 同步版本安装

如果需要使用Selenium的同步版本，可以使用以下命令：

pip install crawl4ai[sync]

3.3 开发者安装

对于想要参与项目开发，修改源代码的贡献者，可以通过以下步骤进行安装：

git clone https://github.com/unclecode/crawl4ai.git
cd crawl4ai
pip install -e .

4.高级应用

要想充分发挥Crawl4AI的能力，可以看看这些高级功能和应用案例：

4.1 执行JavaScript和使用CSS选择器

可以利用Crawl4AI执行自定义JavaScript代码，以及通过CSS选择器精准定位页面元素，从而提升爬取任务的效率和精确度。这让你能够更灵活地处理复杂的网页数据抓取需求。

import asyncio
from crawl4ai import AsyncWebCrawlerasyncdef main():asyncwith AsyncWebCrawler(verbose=True) as crawler:js_code = ["const loadMoreButton = Array.from(document.querySelectorAll('button')).find(button => button.textContent.includes('Load More')); loadMoreButton && loadMoreButton.click();"]result = await crawler.arun(url="https://www.nbcnews.com/business",js_code=js_code,css_selector="article.tease-card",bypass_cache=True)print(result.extracted_content)if __name__ == "__main__":asyncio.run(main())

4.2 使用代理

通过将爬取任务路由到代理，增强隐私和访问权限。

import asyncio
from crawl4ai import AsyncWebCrawlerasync def main():async with AsyncWebCrawler(verbose=True, proxy="http://127.0.0.1:7890") as crawler:result = await crawler.arun(url="https://www.nbcnews.com/business",bypass_cache=True)print(result.markdown)if __name__ == "__main__":asyncio.run(main())

4.3 不使用 LLM 提取结构化数据

使用JsonCssExtractionStrategy精确提取使用 CSS 选择器的结构化数据。

import asyncio
import json
from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import JsonCssExtractionStrategyasyncdef extract_news_teasers():schema = {"name": "News Teaser Extractor","baseSelector": ".wide-tease-item__wrapper","fields": [{"name": "category", "selector": ".unibrow span[data-testid='unibrow-text']", "type": "text"},{"name": "headline", "selector": ".wide-tease-item__headline", "type": "text"},{"name": "summary", "selector": ".wide-tease-item__description", "type": "text"},{"name": "time", "selector": "[data-testid='wide-tease-date']", "type": "text"},{"name": "image","type": "nested","selector": "picture.teasePicture img","fields": [{"name": "src", "type": "attribute", "attribute": "src"},{"name": "alt", "type": "attribute", "attribute": "alt"},],},{"name": "link", "selector": "a[href]", "type": "attribute", "attribute": "href"},],}extraction_strategy = JsonCssExtractionStrategy(schema, verbose=True)asyncwith AsyncWebCrawler(verbose=True) as crawler:result = await crawler.arun(url="https://www.nbcnews.com/business",extraction_strategy=extraction_strategy,bypass_cache=True,)assert result.success, "Failed to crawl the page"news_teasers = json.loads(result.extracted_content)print(f"Successfully extracted {len(news_teasers)} news teasers")print(json.dumps(news_teasers[0], indent=2))if __name__ == "__main__":asyncio.run(extract_news_teasers())

4.4 使用 OpenAI 提取结构化数据

利用 OpenAI 的能力动态提取和结构化数据：

import os
import asyncio
from crawl4ai import AsyncWebCrawler
from crawl4ai.extraction_strategy import LLMExtractionStrategy
from pydantic import BaseModel, Fieldclass OpenAIModelFee(BaseModel):model_name: str = Field(..., description="Name of the OpenAI model.")input_fee: str = Field(..., description="Fee for input token for the OpenAI model.")output_fee: str = Field(..., description="Fee for output token for the OpenAI model.")asyncdef main():asyncwith AsyncWebCrawler(verbose=True) as crawler:result = await crawler.arun(url='https://openai.com/api/pricing/',word_count_threshold=1,extraction_strategy=LLMExtractionStrategy(provider="openai/gpt-4
o",api_token=os.getenv('OPENAI_API_KEY'), schema=OpenAIModelFee.schema(),extraction_type="schema",instruction="""From the crawled content, extract all mentioned model names along with their fees for input and output tokens. Do not miss any models in the entire content. One extracted model JSON format should look like this: {"model_name": "GPT-4", "input_fee": "US$10.00 / 1M tokens", "output_fee": "US$30.00 / 1M tokens"}."""),            bypass_cache=True,)print(result.extracted_content)if __name__ == "__main__":asyncio.run(main())

4.5 会话管理 & 动态内容爬取

处理复杂的场景，如爬取通过 JavaScript 加载动态内容的多个页面：

import asyncio
import re
from bs4 import BeautifulSoup
from crawl4ai import AsyncWebCrawlerasyncdef crawl_typescript_commits():first_commit = ""asyncdef on_execution_started(page):nonlocal first_commit try:whileTrue:await page.wait_for_selector('li.Box-sc-g0xbh4-0 h4')commit = await page.query_selector('li.Box-sc-g0xbh4-0 h4')commit = await commit.evaluate('(element) => element.textContent')commit = re.sub(r'\s+', '', commit)if commit and commit != first_commit:first_commit = commitbreakawait asyncio.sleep(0.5)except Exception as e:print(f"Warning: New content didn't appear after JavaScript execution: {e}")asyncwith AsyncWebCrawler(verbose=True) as crawler:crawler.crawler_strategy.set_hook('on_execution_started', on_execution_started)url = "https://github.com/microsoft/TypeScript/commits/main"session_id = "typescript_commits_session"all_commits = []js_next_page = """const button = document.querySelector('a[data-testid="pagination-next-button"]');if (button) button.click();"""for page in range(3):  # Crawl 3 pagesresult = await crawler.arun(url=url,session_id=session_id,css_selector="li.Box-sc-g0xbh4-0",js=js_next_page if page > 0elseNone,bypass_cache=True,js_only=page > 0)assert result.success, f"Failed to crawl page {page + 1}"soup = BeautifulSoup(result.cleaned_html, 'html.parser')commits = soup.select("li.Box-sc-g0xbh4-0")all_commits.extend(commits)print(f"Page {page + 1}: Found {len(commits)} commits")await crawler.crawler_strategy.kill_session(session_id)print(f"Successfully crawled {len(all_commits)} commits across 3 pages")if __name__ == "__main__":asyncio.run(crawl_typescript_commits())