Langgraph学习5：AI爬取工具crawl4ai

随着大语言模型(LLM)应用的发展，获取高质量的网络内容成为构建智能应用的重要环节。本文将介绍一个专为AI设计的爬虫工具——crawl4ai，并展示如何将其集成到Langgraph框架中，为AI助手提供更强大的信息获取能力。

1. crawl4ai简介

GitHub：https://github.com/unclecode/crawl4ai
文档地址：https://docs.crawl4ai.com/

是一个专为LLM应用设计的Python爬虫库，旨在简化网页爬取并提取有价值的信息。与传统爬虫工具相比，它具有以下特点：

  • LLM友好的输出格式：支持JSON、HTML和Markdown等多种输出格式
  • 异步架构：支持并行爬取多个URL，提高效率
  • 智能内容提取：可自动过滤网页中的广告、导航栏等无关内容
  • 自定义功能丰富：支持JavaScript执行、CSS选择器等高级功能

这些特性使crawl4ai成为构建智能应用时的理想选择，尤其适合需要从多个网页获取信息的场景。

2. 安装与环境准备

首先，需要安装crawl4ai及其依赖：

pip install crawl4ai

crawl4ai基于Playwright实现浏览器自动化，因此需要安装相关的浏览器驱动。安装完成后，我们可以开始使用该工具了。

3. 基础爬虫工具实现

让我们创建一个基础的爬虫工具函数，用于爬取指定URL的内容：

首先根据官方文档进行相关配置：

# 浏览器配置
browser_config = BrowserConfig(
    headless=True,  # 启用无头模式
    user_agent_mode="random", # 随机生成user_agent
    text_mode=True, # 只返回文本内容
)

# 爬虫配置
run_conf = CrawlerRunConfig(
    cache_mode=CacheMode.ENABLED, # 缓存模式
    stream=True,  # 是否启用流式模式
    excluded_tags=["form", "header", "footer", "nav"],
    exclude_external_links=True, # 是否排除外部链接
    exclude_social_media_links=True, # 是否排除社交媒体链接
    remove_forms=True, # 移除表单
    exclude_external_images=True, # 是否排除外部图片
)

这部分代码设置了两个重要的配置对象：

浏览器配置 (BrowserConfig):

  • headless=True: 无头模式，不显示浏览器界面
  • user_agent_mode=”random”: 随机生成User-Agent，降低被网站识别为爬虫的风险
  • text_mode=True: 只返回文本内容，忽略图片等多媒体

爬虫配置 (CrawlerRunConfig):

  • cache_mode=CacheMode.ENABLED: 启用缓存，避免重复爬取
  • stream=True: 启用流式模式，可以异步处理结果
  • excluded_tags=[“form”, “header”, “footer”, “nav”]: 排除不需要的HTML标签
    其他选项用于过滤无关内容，如外部链接、社交媒体链接、表单等

异步爬取与结果处理

async with AsyncWebCrawler(config=browser_config) as crawler:
    results = await crawler.arun_many(urls, config=run_conf)
    
    # 获取当前工作目录
    current_dir = os.getcwd()
    print('current_dir------>', current_dir)
    
    # 生成文件名 (使用时间戳确保唯一性)
    timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_file = os.path.join(current_dir, f"crawl_results_{timestamp}.md")
    
    search_results = ''
    # 创建或打开文件用于写入
    with open(output_file, 'w', encoding='utf-8') as f:
       async for res in results:
            if res.success:
                print(f"[OK] {res.url}, length: {len(res.markdown.raw_markdown)}")
                # 写入URL和内容到文件
                f.write(f"# URL: {res.url}\n\n")
                f.write(f"{res.markdown.raw_markdown}\n\n")
                f.write("---\n\n")  # 分隔符
                search_results += f"{res.markdown.raw_markdown}\n\n"
            else:
                print(f"[ERROR] {res.url} => {res.error_message}")
                # 写入错误信息到文件
                f.write(f"# ERROR URL: {res.url}\n")
                f.write(f"Error: {res.error_message}\n\n")
                f.write("---\n\n")
    
    print(f"所有结果已保存到文件: {output_file}")
    return search_results

这部分代码包含以下核心功能：
异步爬取多个URL：使用arun_many方法并行爬取所有URL

结果保存：

  • 使用当前时间戳创建唯一的Markdown文件名
  • 将每个URL的爬取结果写入文件，包括URL、内容或错误信息
  • 使用分隔符区分不同URL的内容

结果收集与返回：

  • 将成功爬取的内容累加到search_results变量
  • 最终返回所有爬取内容的字符串

4. 高级功能与注意事项

crawl4ai还提供了许多高级功能，如：

  • 结构化数据提取：使用JsonCssExtractionStrategy可以精确提取网页中的结构化数据
  • JavaScript执行：可以在爬取前执行自定义JavaScript，适用于需要与动态网页交互的场景
  • CSS选择器：支持使用CSS选择器精确定位网页元素

使用crawl4ai时，需要注意以下几点：

  • 遵守网站规则：始终遵守robots.txt规则和网站服务条款
  • 适当的爬取速率：避免短时间内发送过多请求，可能被网站封禁
  • 处理异常：网络不稳定或网站内容变化可能导致爬取失败，需要妥善处理异常

5. 总结

crawl4ai是一个功能强大且易于使用的爬虫工具，尤其适合AI应用场景。通过本文介绍的代码示例，我们可以看到它如何高效地：

  • 配置浏览器和爬虫行为
  • 异步并行爬取多个网页
  • 处理和保存爬取结果
  • 这些功能使得crawl4ai成为构建需要网络信息的AI应用的理想选择，无论是简单的信息检索还是复杂的数据分析都能轻松应对。

完整代码：

from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode, BrowserConfig
import os
import asyncio
from datetime import datetime

urls = [
    "https://www.huangli.com/huangli/2025/04_12.html",
    "https://www.zhihu.com/question/609483833/answer/3420895685"
]

# 爬虫工具
async def quick_crawl_tool(urls: list[str]):
    print('urls------>',urls)
    

    # 浏览器配置
    browser_config = BrowserConfig(
        headless=True,  # 启用无头模式
        user_agent_mode="random", # 随机生成user_agent
        text_mode=True, # 只返回文本内容
    )

    # 爬虫配置
    run_conf = CrawlerRunConfig(
        cache_mode=CacheMode.ENABLED, # 缓存模式
        stream=True,  # 是否启用流式模式
        excluded_tags=["form", "header", "footer", "nav"],
        exclude_external_links=True, # 是否排除外部链接
        exclude_social_media_links=True, # 是否排除社交媒体链接
        remove_forms=True, # 移除表单
        exclude_external_images=True, # 是否排除外部图片
    )

    async with AsyncWebCrawler(config=browser_config) as crawler:
        # 或者一次性获取所有结果(默认行为)
        results = await crawler.arun_many(urls, config=run_conf)
        
        # 获取当前工作目录
        current_dir = os.getcwd()
        print('current_dir------>',current_dir)
        
        # 生成文件名 (使用时间戳确保唯一性)
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_file = os.path.join(current_dir, f"crawl_results_{timestamp}.md")
        
        search_results = ''
        # 创建或打开文件用于写入
        with open(output_file, 'w', encoding='utf-8') as f:
           async for res in results:
                if res.success:
                    print(f"[OK] {res.url}, length: {len(res.markdown.raw_markdown)}")
                    # 写入URL和内容到文件
                    f.write(f"# URL: {res.url}\n\n")
                    f.write(f"{res.markdown.raw_markdown}\n\n")
                    f.write("---\n\n")  # 分隔符
                    search_results += f"{res.markdown.raw_markdown}\n\n"
                else:
                    print(f"[ERROR] {res.url} => {res.error_message}")
                    # 写入错误信息到文件
                    f.write(f"# ERROR URL: {res.url}\n")
                    f.write(f"Error: {res.error_message}\n\n")
                    f.write("---\n\n")
        
        print(f"所有结果已保存到文件: {output_file}")
        return search_results


if __name__ == "__main__":
   asyncio.run(quick_crawl_tool(urls))