**从手动配置到一键提取:ScrapeGraphAI 如何用大模型彻底改变网页数据采集**

**从手动配置到一键提取:ScrapeGraphAI 如何用大模型彻底改变网页数据采集**

从手动配置到一键提取:ScrapeGraphAI 如何用大模型彻底改变网页数据采集


为什么值得关注

在传统爬虫时代,采集网页数据是一项令人头疼的工作。你需要分析 HTML 结构、编写复杂的 CSS 选择器或 XPath 表达式、应对各种反爬机制、处理动态加载内容……每遇到一个网站,可能就要推翻重来一遍。更糟糕的是,很多网站的结构说变就变,昨天还能正常运行的脚本今天可能就彻底报废。

ScrapeGraphAI 的出现彻底颠覆了这一局面。这是一个专门为大语言模型设计的网页爬虫框架,它的核心思想非常简单:用 AI 来理解网页内容,用自然语言来描述你要提取的数据。你不需要再关心网站的 HTML 结构,只需要告诉它:“帮我把这个产品页面里的所有商品名称、价格和评分提取出来”——它就能自动完成。

这个项目在 GitHub 上已经获得了大量关注,它的核心优势包括:

零配置即可提取结构化数据。传统方法需要你手动分析网页结构,而 ScrapeGraphAI 可以根据你的自然语言描述自动推断要提取的内容格式,并返回 JSON、字典或你指定的数据结构。

多源支持,开箱即用。它同时支持 ScrapeGraph(通用网页)、SmartScraper(单页深度提取)、SearchGraph(搜索引擎结果采集)等多种模式,满足不同的采集需求。

灵活的大模型后端。你可以使用 OpenAI、Azure、Google Gemini 等商业大模型,也可以部署本地模型(如 Ollama 支持的模型),完全掌控成本和隐私。

智能处理复杂网页。对于需要滚动加载、点击加载更多、登录后才能查看内容等复杂场景,ScrapeGraphAI 都提供了相应的处理机制。


环境搭建

系统要求

在开始之前,确认你的开发环境满足以下要求:

  • Python 版本 3.8 或更高(推荐使用 Python 3.10 或 3.11,能获得最佳的依赖兼容性)
  • 稳定的网络连接(用于访问目标网站和调用大模型 API)
  • 足够的磁盘空间(用于安装依赖包和可能的本地模型缓存)

使用 pip 安装

ScrapeGraphAI 的安装非常简单,打开终端或命令行窗口,执行以下命令即可:

pip install scrapegraphai

如果你需要使用特定的额外功能,比如语音输出或更详细的日志,可以安装带有完整依赖的版本:

pip install scrapegraphai[full]

安装过程中,pip 会自动处理所有必要的依赖项,包括 requests、BeautifulSoup、langchain 等核心库。

安装可选的大模型依赖

ScrapeGraphAI 支持多种大模型后端,你可以根据自己的需求选择性安装:

# OpenAI GPT 系列(最常用)
pip install openai

# Azure OpenAI
pip install openai azure-identity

# Google Gemini
pip install google-genai

# 本地模型支持(配合 Ollama 使用)
pip install ollama

建议先从 OpenAI 入手体验完整功能,后续可以根据需要切换到其他后端或本地部署。

验证安装

安装完成后,在 Python 环境中验证一切正常:

import scrapegraphai

print("ScrapeGraphAI 版本:", scrapegraphai.__version__)

# 检查核心模块是否可用
from scrapegraphai.nodes import FetchNode, ParseNode, GenerateAnswerNode
print("核心模块加载成功!")

如果没有报错,说明安装完全正常,可以开始使用了。


核心功能详解

ScrapeGraphAI 提供了几个核心组件和预置的采集管道,理解它们是灵活运用的基础。

核心组件

Graph 架构

ScrapeGraphAI 的核心是“图”(Graph)的概念。一个采集任务被分解成多个节点(Node),每个节点负责特定的功能,数据在这些节点之间流动,最终产出结果。你可以把它想象成一条流水线:网页内容进入,经过各个处理环节,最终变成结构化的数据输出。

FetchNode – 获取节点

负责从目标 URL 抓取原始 HTML 内容。它可以处理普通的静态页面,也可以配合 Selenium 或 Playwright 处理需要 JavaScript 渲染的动态页面。

ParseNode – 解析节点

负责将 HTML 转换为更易于处理的格式,同时可能进行初步的文本清洗和结构化。ParseNode 的输出通常是一个包含页面文本内容的文档对象。

GenerateAnswerNode – 生成节点

这是整个流程的“大脑”。它接收用户的自然语言指令和解析后的页面内容,调用大模型来理解和提取用户需要的信息。

预置管道

SmartScraperGraph

最常用的采集管道,适合单页深度提取。你只需要提供一个 URL 和一个自然语言描述,就能获得结构化的提取结果。

输入:URL + "提取页面上所有文章的标题、作者和发布日期"
输出:结构化的 JSON 数据

SearchGraph

专为搜索引擎结果设计。你提供一个搜索关键词,它会自动抓取搜索结果页,提取标题、链接、摘要等信息。

输入:搜索关键词 + "提取前 10 条搜索结果的标题和链接"
输出:搜索结果列表

ScriptCreatorGraph

这是一个很有趣的功能——它可以根据你的采集需求,自动生成一个可重复使用的 Python 脚本。这样你就不必每次都调用 AI,而是可以运行生成的脚本来完成相同的任务,效率更高。


实战教程

第一个示例:提取电商产品信息

让我们从最常见的场景开始:提取电商网站的产品信息。假设我们要从某个产品页面提取商品名称、价格、评分和评论数量。

步骤一:准备配置

首先,你需要设置大模型的 API 密钥。如果你使用 OpenAI:

import os

os.environ["OPENAI_API_KEY"] = "你的-API-密钥"

步骤二:编写采集代码

from scrapegraphai.graphs import SmartScraperGraph

# 定义采集配置
graph_config = {
    "llm": {
        "model": "gpt-4o-mini",  # 可以选择 gpt-4o、gpt-4-turbo 等
        "api_key": os.environ["OPENAI_API_KEY"],
        "temperature": 0.1,  # 温度参数,越低越确定性输出
    },
    "verbose": True,  # 显示详细的执行过程
    "headless": True,  # 是否使用无头模式
}

# 创建采集器实例
smart_scraper = SmartScraperGraph(
    prompt="提取这个页面的商品名称、价格、评分(满分5分)、评论数量和主要规格参数",
    source="https://example.com/products/some-product",
    config=graph_config
)

# 执行采集
result = smart_scraper.run()

# 查看结果
print("采集结果:")
print(result)

运行后,你会得到一个结构化的字典,类似这样:

{
    "商品名称": "iPhone 15 Pro Max 256GB 钛金属色",
    "价格": "¥9999",
    "评分": 4.8,
    "评论数量": 12580,
    "主要规格": {
        "存储容量": "256GB",
        "颜色": "钛金属色",
        "芯片": "A17 Pro",
        "屏幕尺寸": "6.7英寸"
    }
}

整个过程就这么简单!你不需要写任何 CSS 选择器,不需要分析页面结构,只需要用自然语言描述你要什么。

示例二:批量采集搜索结果

当你需要收集某个主题的多个相关信息时,SearchGraph 就派上用场了。比如你要采集“机器学习教程”相关的 YouTube 视频信息:

from scrapegraphai.graphs import SearchGraph

search_config = {
    "llm": {
        "model": "gpt-4o-mini",
        "api_key": os.environ["OPENAI_API_KEY"],
    },
    "max_results": 10,  # 最多采集 10 条结果
}

search_scraper = SearchGraph(
    prompt="""提取以下信息:
    - 视频标题
    - 视频链接
    - 频道名称
    - 播放量
    - 上传日期""",
    config=search_config
)

result = search_scraper.run("Python 机器学习教程")

print("搜索结果:")
for item in result:
    print(f"标题:{item['视频标题']}")
    print(f"链接:{item['视频链接']}")
    print(f"频道:{item['频道名称']}")
    print(f"播放量:{item['播放量']}")
    print("---")

示例三:处理需要交互的页面

有些页面的内容不是一次性加载完成的,需要滚动、点击或登录才能看到。ScrapeGraphAI 支持与 Playwright 集成来处理这类场景:

from scrapegraphai.graphs import SmartScraperGraph

# 配置中启用 Playwright
graph_config = {
    "llm": {
        "model": "gpt-4o-mini",
        "api_key": os.environ["OPENAI_API_KEY"],
    },
    "headless": True,
    "playwright": {
        "headless": True,
        "dynamic_wait": 5000,  # 等待动态内容加载的时间(毫秒)
    },
}

smart_scraper = SmartScraperGraph(
    prompt="提取页面加载完成后显示的所有评论内容和评论者用户名",
    source="https://example.com/article/123/comments",
    config=graph_config
)

result = smart_scraper.run()

对于需要登录才能访问的页面,你也可以在配置中提供 cookies 或使用代理登录后的会话:

graph_config = {
    "llm": {
        "model": "gpt-4o-mini",
        "api_key": os.environ["OPENAI_API_KEY"],
    },
    "headless": True,
    "cookies": [
        {"name": "session_id", "value": "你的-session-id", "domain": ".example.com"},
        # 可以添加更多 cookies
    ],
}

示例四:使用本地模型降低成本

如果你对数据隐私有要求,或者想要节省 API 调用费用,可以使用 Ollama 部署本地模型:

from scrapegraphai.graphs import SmartScraperGraph

# 配置使用 Ollama 本地模型
graph_config = {
    "llm": {
        "provider": "ollama",
        "model": "llama3.2",  # 或其他你已下载的模型
        "base_url": "http://localhost:11434",  # Ollama 默认地址
        "temperature": 0.1,
    },
    "headless": True,
}

smart_scraper = SmartScraperGraph(
    prompt="用中文总结这篇新闻的主要内容",
    source="https://news.example.com/tech/ai-breakthrough",
    config=graph_config
)

result = smart_scraper.run()
print("摘要结果:", result)

使用本地模型的优势是完全没有 API 费用,而且数据不会离开你的机器。但需要注意,本地模型的响应速度和效果可能不如云端的大模型,需要根据实际情况权衡。

示例五:生成可复用的脚本

对于需要定期执行的采集任务,每次都调用 AI 可能会显得过于奢侈。ScriptCreatorGraph 可以帮你生成一个标准的 Python 脚本:

from scrapegraphai.graphs import ScriptCreatorGraph

script_config = {
    "llm": {
        "model": "gpt-4o-mini",
        "api_key": os.environ["OPENAI_API_KEY"],
    },
}

script_creator = ScriptCreatorGraph(
    prompt="创建一个脚本来提取新闻网站的文章标题、发布日期和正文内容",
    config=script_config
)

script_code = script_creator.run()

# 将生成的脚本保存到文件
with open("news_scraper.py", "w", encoding="utf-8") as f:
    f.write(script_code)

print("脚本已保存到 news_scraper.py")
print("你可以在不调用 AI 的情况下重复运行这个脚本")

生成的脚本会包含标准的 requests 和 BeautifulSoup 操作,完全独立运行,不需要再消耗 AI token。


常见使用场景

理解了基本用法之后,让我们看看 ScrapeGraphAI 在实际工作中能帮你解决哪些问题。

场景一:竞品价格监控

电商运营人员经常需要监控竞品的价格变动。使用传统方法,你可能需要为每个电商平台编写专门的爬虫脚本,维护成本很高。ScrapeGraphAI 可以统一处理:

from scrapegraphai.graphs import SmartScraperGraph

urls = [
    "https://www.jd.com/product/123456.html",
    "https://www.tmall.com/product/789012.html",
    "https://www.suning.com/product/345678.html",
]

graph_config = {
    "llm": {
        "model": "gpt-4o-mini",
        "api_key": os.environ["OPENAI_API_KEY"],
    },
}

for url in urls:
    scraper = SmartScraperGraph(
        prompt="提取商品名称、价格、原价、折扣信息",
        source=url,
        config=graph_config
    )
    result = scraper.run()
    print(f"URL: {url}")
    print(f"价格: {result}")
    print("---")

场景二:学术文献资料收集

研究人员在进行文献综述时,往往需要从多个数据库和网站收集论文信息:

search_keywords = [
    "transformer attention mechanism 2024",
    "large language model fine-tuning",
    "multimodal AI research"
]

all_results = []

for keyword in search_keywords:
    search_scraper = SearchGraph(
        prompt="""提取论文标题、作者、发表年份、期刊/会议名称、摘要""",
        config={"llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}, "max_results": 20}
    )
    results = search_scraper.run(keyword)
    all_results.extend(results)

# 将结果导出为 CSV
import csv

with open("literature_review.csv", "w", newline="", encoding="utf-8") as f:
    writer = csv.DictWriter(f, fieldnames=["标题", "作者", "年份", "来源", "摘要"])
    writer.writeheader()
    for item in all_results:
        writer.writerow(item)

print(f"共采集到 {len(all_results)} 条文献信息")

场景三:招聘信息和职位分析

求职者和 HR 都可以利用这个工具来收集和分析职位信息:

def extract_job_listing(url, keywords):
    """提取职位信息"""
    scraper = SmartScraperGraph(
        prompt=f"""从职位页面提取:
        - 职位名称
        - 公司名称
        - 薪资范围
        - 工作地点
        - 职位描述(关键职责)
        - 岗位要求(关键技能)
        - 是否有{keywords}相关要求""",
        source=url,
        config={"llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}}
    )
    return scraper.run()

# 批量采集多个招聘平台的职位
job_urls = [
    "https://www.zhipin.com/job_abc123.html",
    "https://www.lagou.com/jobs/xyz456.html",
    # 更多 URL...
]

jobs_data = [extract_job_listing(url, "Python") for url in job_urls]

# 分析哪些技能最受欢迎
skills_count = {}
for job in jobs_data:
    requirements = job.get("岗位要求", "")
    # 简单统计技能词频
    common_skills = ["Python", "SQL", "机器学习", "深度学习", "TensorFlow", "PyTorch"]
    for skill in common_skills:
        if skill in requirements:
            skills_count[skill] = skills_count.get(skill, 0) + 1

print("技能需求统计:")
for skill, count in sorted(skills_count.items(), key=lambda x: x[1], reverse=True):
    print(f"{skill}: {count} 个职位")

场景四:新闻舆情监控

企业公关或市场人员可能需要持续跟踪特定主题的新闻报道:

from datetime import datetime, timedelta

def collect_news(topic, days=7):
    """收集近期的新闻报道"""
    search_scraper = SearchGraph(
        prompt="""提取新闻标题、发布媒体、发布日期、新闻摘要""",
        config={"llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}, "max_results": 50}
    )
    results = search_scraper.run(f"{topic} 最新新闻")

    # 过滤时间范围
    recent_news = []
    cutoff_date = datetime.now() - timedelta(days=days)

    for news in results:
        # 假设日期格式为 "2024-01-15" 或类似格式
        pub_date_str = news.get("发布日期", "")
        try:
            pub_date = datetime.strptime(pub_date_str, "%Y-%m-%d")
            if pub_date >= cutoff_date:
                recent_news.append(news)
        except:
            recent_news.append(news)  # 无法解析日期时保留

    return recent_news

# 监控竞品相关报道
competitor_news = collect_news("某竞争对手公司", days=30)

print(f"近期共有 {len(competitor_news)} 条相关报道:")
for idx, news in enumerate(competitor_news, 1):
    print(f"{idx}. {news['标题']} - {news['发布媒体']} ({news['发布日期']})")

场景五:网站结构分析和内容审计

网站管理员和 SEO 工作人员可以用它来审计网站内容:

def audit_page(url):
    """审计单个页面的 SEO 相关信息"""
    scraper = SmartScraperGraph(
        prompt="""详细分析这个页面的内容质量:
        - 主标题(H1)和副标题结构
        - 图片 alt 属性情况
        - 内容长度和段落结构
        - 关键词密度(指出主要关键词)
        - 内外部链接情况
        - 内容可读性评分""",
        source=url,
        config={"llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}}
    )
    return scraper.run()

# 审计网站的所有主要页面
site_pages = [
    "https://example.com/",
    "https://example.com/about/",
    "https://example.com/products/",
    # 更多页面...
]

audit_results = {}
for page_url in site_pages:
    try:
        result = audit_page(page_url)
        audit_results[page_url] = result
    except Exception as e:
        audit_results[page_url] = {"error": str(e)}

# 生成审计报告
with open("seo_audit_report.md", "w", encoding="utf-8") as f:
    f.write("# 网站 SEO 审计报告\n\n")
    for url, result in audit_results.items():
        f.write(f"## {url}\n\n")
        f.write(f"```\n{result}\n```\n\n")

print("审计报告已生成:seo_audit_report.md")

技巧与最佳实践

提高采集稳定性

合理设置重试机制

网络请求偶尔会失败,添加重试逻辑可以显著提高采集成功率:

import time
from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
def robust_scrape(url, prompt, config):
    """带有重试机制的采集函数"""
    scraper = SmartScraperGraph(prompt=prompt, source=url, config=config)
    return scraper.run()

# 使用示例
result = robust_scrape(
    "https://example.com/page",
    "提取关键信息",
    {"llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}}
)

控制采集速度

避免过于频繁的请求,可以使用时间间隔来控制速率:

import time

urls_to_scrape = ["https://example.com/page1", "https://example.com/page2", ...]

for url in urls_to_scrape:
    result = scrape(url)
    print(f"已完成: {url}")
    time.sleep(2)  # 每次采集后等待 2 秒

优化成本控制

批量处理减少 API 调用

与其对每个页面单独调用 AI,不如批量处理:

def batch_scrape(urls, prompt, config):
    """批量采集多个页面"""
    results = []
    for url in urls:
        scraper = SmartScraperGraph(prompt=prompt, source=url, config=config)
        results.append({
            "url": url,
            "data": scraper.run()
        })
    return results

# 一次提取多个页面的共同字段
batch_results = batch_scrape(
    urls=["https://site.com/products/1", "https://site.com/products/2"],
    prompt="所有产品页都提取:产品名称、价格、库存状态",
    config={"llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}}
)

使用更高效的模型

对于简单的提取任务,使用更便宜更快的模型就足够了:

# 简单任务用 mini 模型
simple_config = {"llm": {"model": "gpt-4o-mini", ...}}

# 复杂任务用完整模型
complex_config = {"llm": {"model": "gpt-4o", ...}}

# 判断任务复杂度
def auto_select_config(task_type, url, prompt):
    if task_type == "simple":
        return simple_config
    else:
        return complex_config

处理复杂页面

处理登录和验证码

对于需要登录的页面,可以先手动获取 cookies 后使用:

from selenium import webdriver

# 使用 Selenium 获取登录后的 cookies
driver = webdriver.Chrome()
driver.get("https://example.com/login")
# 手动完成登录...

cookies = driver.get_cookies()
driver.quit()

# 将 cookies 用于 ScrapeGraphAI
graph_config = {
    "llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]},
    "cookies": cookies,
}

scraper = SmartScraperGraph(prompt="提取用户信息", source="https://example.com/profile", config=graph_config)

处理分页内容

对于分页显示的内容,可以先提取分页规则,然后循环采集:

def extract_pagination_info(base_url):
    """提取分页信息"""
    scraper = SmartScraperGraph(
        prompt="分析页面分页结构,找出分页 URL 模式(如 ?page=1, /page/2 等)",
        source=base_url,
        config={"llm": {"model": "gpt-4o-mini", "api_key": os.environ["OPENAI_API_KEY"]}}
    )
    return scraper.run()

# 获取分页模式后采集所有页面
pagination_info = extract_pagination_info("https://example.com/articles")
page_urls = [f"https://example.com/articles?page={i}" for i in range(1, 11)]

all_articles = []
for url in page_urls:
    result = scrape_single_page(url)
    all_articles.extend(result)

错误处理与日志

完善的错误处理

import logging
from datetime import datetime

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler(f"scrape_log_{datetime.now().strftime('%Y%m%d')}.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger(__name__)

def safe_scrape(url, prompt, config):
    """安全的采集函数,带有完整的错误处理"""
    try:
        logger.info(f"开始采集: {url}")
        scraper = SmartScraperGraph(prompt=prompt, source=url, config=config)
        result = scraper.run()
        logger.info(f"成功采集: {url}")
        return {"success": True, "data": result, "url": url}
    except Exception as e:
        logger.error(f"采集失败 {url}: {str(e)}")
        return {"success": False, "error": str(e), "url": url}

# 批量采集并记录结果
results = []
failed_urls = []

for url in urls:
    result = safe_scrape(url, prompt, config)
    results.append(result)
    if not result["success"]:
        failed_urls.append(url)

# 输出失败重试建议
if failed_urls:
    logger.warning(f"共有 {len(failed_urls)} 个页面采集失败,建议稍后重试")
    print("失败 URL 列表:", failed_urls)

总结与资源链接

核心要点回顾

通过这篇文章,你应该已经掌握了 ScrapeGraphAI 的基本使用方法:

它是什么:一个基于大语言模型的智能网页采集框架,用自然语言替代了传统的 CSS 选择器和 XPath 表达式。

为什么用它:大大降低了网页数据采集的门槛,一次编写可以适应多种网站结构,采集结果的格式可以自定义。

怎么用:SmartScraperGraph 用于单页提取,SearchGraph 用于搜索结果采集,配置好大模型后只需要提供 URL 和自然语言指令。

需要注意什么:控制采集频率以免对目标网站造成压力,合理选择模型以平衡成本和效果,添加错误处理以提高稳定性。

相关资源链接

  • GitHub 仓库:https://github.com/ScrapeGraphAI/Scrapegraph-ai
  • 官方文档:https://scrapegraph-doc.readthedocs.io/
  • 示例代码库:https://github.com/ScrapeGraphAI/scrapegraph-ai/tree/main/examples
  • 官方 Discord 社区:加入讨论获取最新更新和使用技巧

延伸学习方向

如果你对网页采集和大模型应用感兴趣,以下方向值得进一步探索:

深入理解 LangChain 集成。ScrapeGraphAI 底层使用了 LangChain 的 Chain 概念,学习 LangChain 可以帮助你更灵活地构建复杂的采集流程。

探索本地部署大模型。使用 Ollama 或 text-generation-webui 在本地运行模型,可以完全掌控数据隐私,同时探索 llama3、mistral 等开源模型的能力边界。

学习 Playwright 和 Selenium。虽然 ScrapeGraphAI 可以自动处理部分动态内容,但对于更复杂的交互场景,掌握浏览器自动化工具会让你如虎添翼。

数据处理和分析。采集到数据后,学习 pandas、matplotlib 等工具可以帮助你从数据中提取洞察。

ScrapeGraphAI 代表了网页采集领域的一个重要趋势:用 AI 来理解和适应网页结构,而不是让人去适应网页结构。随着大模型能力的不断提升,这类工具的功能和易用性都会越来越好。现在正是学习和入门的最佳时机,动手试试吧!

如果内容对您有帮助,欢迎打赏

您的支持是我继续创作的动力

前往打赏页面

评论区

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注