Crawl4AI爬虫获取页面内容

2025-11-29 10:39:22 分类：爬虫 热度：19 评论： 0

爬取网页内容 转为对LLM友好的MarkDown格式

一、Crawl4AI安装

基本安装

pip install crawl4ai

这将安装核心 Crawl4AI 库以及必要的依赖项。目前还没有包含任何高级功能（如 transformer 或 PyTorch）。

初始设置和诊断

运行 Setup 命令

安装后，调用：

crawl4ai-setup

它有什么作用？- 安装或更新所需的 Playwright 浏览器（Chromium、Firefox 等） - 执行作系统级别的检查（例如，Linux 上缺少库） - 确认您的环境已准备好进行爬网

诊断

或者，您可以运行诊断来确认一切正常：

crawl4ai-doctor

此命令尝试： - 检查 Python 版本兼容性 - 验证 Playwright 安装 - 检查环境变量或库冲突

二、使用方法

视频教程：【🚀 吐血爆肝整理最强大生成 AI 知识库工具！Crawl4AI 核心教程｜附核心代码】https://www.bilibili.***/video/BV1xTMGz1E3m?vd_source=7b1d5b62108146d6c57a160c3984dd92

github项目地址：https://github.***/unclecode/crawl4ai?tab=readme-ov-file

crawl4ai使用文档：https://docs.crawl4ai.***/

crawl4ai有三种解析方法：修剪内容过滤器（PruningContentFilter），BM25内容过滤器，LLM内容过滤器（LLMContentFilter）

由于LLM需要调用API有成本，且响应慢，这里优先介绍Pruning和BM25

经过测试BM25对中文效果不好，对英文内容搜索更好，所以推荐使用Pruning

1.Pruning（修剪）

修剪，自动分析内容重要度，只保留重要的内容

适合AI知识库构建

import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from crawl4ai.content_filter_strategy import PruningContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

"""
Pruning: 修剪，只保留重要的内容，自动分析重要度

在Crawl4AI中，Pruning是通过DefaultMarkdownGenerator类实现的。

options:
    ignore_links: 是否在最终markdown中移除所有超链接
    ignore_images: 是否在最终markdown中移除所有图片
"""

async def main():
    browser_config = BrowserConfig(
        headless=True, # 是否无头模式，True：不打卡浏览器
        viewport_width=1280, # 视口宽度
        viewport_height=720, # 视口高度
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # 用户代理
        text_mode=True, # 文本模式,禁用图片加载
    )  # 浏览器配置
    run_config = CrawlerRunConfig(
        cache_mode=CacheMode.DISABLED, # 禁用缓存模式，获取最新内容
        markdown_generator = DefaultMarkdownGenerator(
            content_filter=PruningContentFilter(
                # min_word_threshold = 10, # 丢掉少于10个单词的块，因为他们可能太短或者无用
                threshold = 0.76, # 丢掉重要度低于0.76的块,越高过滤越严格
                threshold_type = "fixed", # 重要度阈值类型，fixed：固定值，dynamic：相对值
                # threshold_type = "dynamic"
            ),
            options = {
                "ignore_links": True, # 是否在最终markdown中移除所有超链接
                "ignore_images": True, # 是否在最终markdown中移除所有图片
            }
        )
    )  # 爬虫运行配置

    async with AsyncWebCrawler(config=browser_config) as crawler:
        try:
            result = await crawler.arun(
                url="https://www.anthropic.***/news/agent-capabilities-api",
                config=run_config
            )
        except Exception as e:
            print(f"错误：{e}")
        with open(f"2.2.2result-{len(result.markdown.fit_markdown)}.md", "w", encoding="utf-8") as f:
            f.write(result.markdown.fit_markdown)
        print(f"内容长度：{len(result.markdown.fit_markdown)}")
        print(f"已保存在：{f.name}")

if __name__ == "__main__":
    asyncio.run(main())

2.BM25

向量检索，根据用户问题生成向量，然后根据向量检索文章相似内容，并返回文章相似内容片段

适合AI+联网搜索

经过测试BM25对中文效果不好，对英文内容搜索更好，所以推荐使用Pruning

import asyncio
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from crawl4ai.content_filter_strategy import BM25ContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

"""
bm25: 向量检索，根据内容生成向量，然后根据向量检索相似内容
"""

async def main():
    browser_config = BrowserConfig(
        headless=True, # 是否无头模式，True：不打卡浏览器
        viewport_width=1280, # 视口宽度
        viewport_height=720, # 视口高度
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # 用户代理
        text_mode=True, # 文本模式,禁用图片加载
    )  # 浏览器配置
    run_config = CrawlerRunConfig(
        cache_mode=CacheMode.DISABLED, # 禁用缓存模式，获取最新内容
        markdown_generator = DefaultMarkdownGenerator(
            content_filter=BM25ContentFilter(
                user_query="Anthropic API",
                bm25_threshold=1.2, #数值越高块越少
            ),
            options = {
                "ignore_links": True, # 是否在最终markdown中移除所有超链接
                "ignore_images": True, # 是否在最终markdown中移除所有图片
            }
        )
    )  # 爬虫运行配置

    async with AsyncWebCrawler(config=browser_config) as crawler:
        try:
            result = await crawler.arun(
                url="https://www.anthropic.***/news/agent-capabilities-api",
                config=run_config
            )
        except Exception as e:
            print(f"错误：{e}")
        with open(f"2.2.3result-{len(result.markdown.fit_markdown)}.md", "w", encoding="utf-8") as f:
            f.write(result.markdown.fit_markdown)
        print(f"内容长度：{len(result.markdown.fit_markdown)}")
        print(f"已保存在：{f.name}")

if __name__ == "__main__":
    asyncio.run(main())

3.并发url请求爬取（arun_many）

import asyncio
import os
import re
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from crawl4ai.content_filter_strategy import BM25ContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

"""
bm25: 向量检索，根据内容生成向量，然后根据向量检索相似内容
"""

async def main():
    browser_config = BrowserConfig(
        headless=True, # 是否无头模式，True：不打卡浏览器
        viewport_width=1280, # 视口宽度
        viewport_height=720, # 视口高度
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # 用户代理
        text_mode=True, # 文本模式,禁用图片加载
    )  # 浏览器配置
    run_config = CrawlerRunConfig(
        cache_mode=CacheMode.DISABLED, # 禁用缓存模式，获取最新内容
        markdown_generator = DefaultMarkdownGenerator(
            content_filter=BM25ContentFilter(
                user_query="Anthropic API",
                bm25_threshold=1.2,
            ),
            options = {
                "ignore_links": True, # 是否在最终markdown中移除所有超链接
                "ignore_images": True, # 是否在最终markdown中移除所有图片
            }
        )
    )  # 爬虫运行配置

    async with AsyncWebCrawler(config=browser_config) as crawler:
        try:
            results = await crawler.arun_many(
                urls=[
                    "https://www.anthropic.***/news/agent-capabilities-api",
                    "https://www.anthropic.***/news/claude-4",
                    "https://www.anthropic.***/news/the-anthropic-economic-index"
                ],
                config=run_config
            )
        except Exception as e:
            print(f"错误：{e}")
            return
            
        # 创建输出目录（如果不存在）
        output_dir = "anthropic_articles"
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
            
        for result in results:
            # 从URL生成文件名
            filename = re.sub(r'https?://(?:www\.)?anthropic\.***/news/', '', result.url)
            filename = re.sub(r'[^\w\-]', '_', filename)  # 替换非字母数字字符为下划线
            filepath = os.path.join(output_dir, f"{filename}-{len(result.markdown.fit_markdown)}.md")
            
            # 保存到markdown文件
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(result.markdown.fit_markdown)
                
            print(f"已保存: {filepath}")


if __name__ == "__main__":
    asyncio.run(main())

三、配合serp api 搜索进行爬虫

1.定义search.py方法

import requests
import json

api_key = "xxxxx申请的key"

def serp_search(query, num_results=10, country="us", language="en", exclude_images=False, exclude_videos=False):
    """
    使用SERP API执行搜索，并返回结果URL列表
    
    参数:
        query (str): 搜索查询词
        num_results (int): 希望返回的结果数量
        country (str): 搜索结果的国家代码，默认为'us'
        language (str): 搜索结果的语言代码，默认为'en'
        exclude_images (bool): 是否排除图片结果，默认为False
        exclude_videos (bool): 是否排除视频结果，默认为False
        
    返回:
        list: 搜索结果URL列表
    """
    url = "https://serpapi.***/search"
    params = {
        "api_key": api_key,
        "q": query,
        "num": num_results,
        "gl": country,
        "hl": language,
        "engine": "google"  # 使用Google搜索引擎
    }
    
    # 如果需要排除图片和视频，添加过滤器
    if exclude_images or exclude_videos:
        filter_terms = []
        if exclude_images:
            filter_terms.append("-filetype:jpg -filetype:png -filetype:gif -filetype:webp -filetype:jpeg")
        if exclude_videos:
            filter_terms.append("-filetype:mp4 -filetype:avi -filetype:mov -filetype:wmv -filetype:flv -site:youtube.*** -site:youtu.be -site:vimeo.*** -site:bilibili.*** -site:tiktok.***")
        
        # 将过滤器添加到查询中
        if filter_terms:
            params["q"] = f"{query} {' '.join(filter_terms)}"
    
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        search_results = response.json()
        
        # 视频网站域名列表
        video_domains = [
            'youtube.***', 'youtu.be', 'vimeo.***', 'bilibili.***', 'b23.tv',
            'tiktok.***', 'douyin.***', 'facebook.***/watch', 'facebook.***/videos',
            'instagram.***/tv', 'instagram.***/reel', 'twitch.tv', 'huya.***',
            'douyu.***', 'iqiyi.***', 'youku.***', 'v.qq.***', 'tudou.***','zhihu.***'
        ]
        
        urls = []
        if "organic_results" in search_results:
            for result in search_results["organic_results"]:
                # 跳过图片和视频结果
                result_type = result.get("type", "").lower()
                if (exclude_images and "image" in result_type) or (exclude_videos and "video" in result_type):
                    continue
                
                if "link" in result:
                    link = result["link"]
                    
                    # 如果需要排除视频，检查URL是否包含视频网站域名
                    if exclude_videos:
                        is_video_site = any(domain in link.lower() for domain in video_domains)
                        if is_video_site:
                            continue
                    
                    urls.append(link)
                    if len(urls) >= num_results:
                        break
        print(urls)
        return urls
    except requests.exceptions.RequestException as e:
        print(f"SERP API请求出错: {e}")
        return []
    except (KeyError, json.JSONDecodeError) as e:
        print(f"解析SERP API结果出错: {e}")
        return []

# 使用示例
if __name__ == "__main__":
    search_query = "王者荣耀最好玩的组合"
    
    # SERP API搜索示例
    results = serp_search(search_query, 10, country="***", language="zh-***", exclude_images=True, exclude_videos=True)
    print(f"SERP API搜索 '{search_query}' 的结果:")
    for i, url in enumerate(results, 1):
        print(f"{i}. {url}")

2.使用crawl4ai

from search import serp_search

query = "王者荣耀最好玩的双人组合"

urls = serp_search(query, num_results=10, country="***", language="zh-***", exclude_images=True, exclude_videos=True)

import asyncio
import os
import re
from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, CacheMode
from crawl4ai.content_filter_strategy import BM25ContentFilter, PruningContentFilter
from crawl4ai.markdown_generation_strategy import DefaultMarkdownGenerator

"""
这里使用PruningContentFilter，根据内容生成重要度，然后根据重要度过滤掉不重要的内容

经过测试BM25对中文效果不好，所以使用PruningContentFilter
"""

async def main(query,urls):
    browser_config = BrowserConfig(
        headless=True, # 是否无头模式，True：不打卡浏览器
        viewport_width=1280, # 视口宽度
        viewport_height=720, # 视口高度
        user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36", # 用户代理
        text_mode=True, # 文本模式,禁用图片加载
    )  # 浏览器配置
    run_config = CrawlerRunConfig(
        cache_mode=CacheMode.DISABLED, # 禁用缓存模式，获取最新内容
        markdown_generator = DefaultMarkdownGenerator(
            content_filter=PruningContentFilter(
                # min_word_threshold = 10, # 丢掉少于10个单词的块，因为他们可能太短或者无用
                threshold = 0.7, # 丢掉重要度低于0.76的块,越高过滤越严格
                threshold_type = "dynamic", # 重要度阈值类型，fixed：固定值，dynamic：相对值
                # threshold_type = "fixed"
            ),
            options = {
                "ignore_links": True, # 是否在最终markdown中移除所有超链接
                "ignore_images": True, # 是否在最终markdown中移除所有图片
            }
        )
    )  # 爬虫运行配置

    async with AsyncWebCrawler(config=browser_config) as crawler:
        try:
            results = await crawler.arun_many(
                urls=urls,
                config=run_config
            )
        except Exception as e:
            print(f"错误：{e}")
            return
            
        # 创建输出目录（如果不存在）
        output_dir = "results"
        if not os.path.exists(output_dir):
            os.makedirs(output_dir)
            
        for result in results:
            # 检查爬取是否成功
            if result.markdown is None or result.markdown.fit_markdown is None:
                print(f"跳过失败的URL: {result.url}")
                continue
                
            # 从URL生成文件名
            filename = re.sub(r'https?://(?:www\.)?anthropic\.***/news/', '', result.url)
            filename = re.sub(r'[^\w\-]', '_', filename)  # 替换非字母数字字符为下划线
            filepath = os.path.join(output_dir, f"{filename}-{len(result.markdown.fit_markdown)}.md")
            
            # 保存到markdown文件
            with open(filepath, 'w', encoding='utf-8') as f:
                f.write(result.markdown.fit_markdown)
                
            print(f"已保存: {filepath}")

if __name__ == "__main__":
    asyncio.run(main(query,urls))

3.效果展示

转载请说明出处内容投诉
CSS教程网 » Crawl4AI爬虫获取页面内容
图小库

图小库

分享到：

发表评论

欢迎 访客 发表评论

一个令你着迷的主题！

查看演示 官网购买