BFSDeepCrawlStrategy 使用教程（crawl4ai）

适用版本：crawl4ai==0.7.8。

BFSDeepCrawlStrategy 是 crawl4ai 的“深度爬取策略”（Deep Crawl Strategy）之一：它会从一个 start_url 出发，按 BFS（广度优先） 一层层发现链接并抓取页面，直到达到 max_depth / max_pages 等限制。

---

1. 核心概念

1.1 深度（max_depth）

在 BFSDeepCrawlStrategy 中：

• depth=0：起始 URL（start_url）
• depth=1：从 start_url 页面里发现的链接
• depth=2：从 depth=1 页面里发现的链接
• ……

当 next_depth > max_depth 时，不再继续发现下一层链接（即停止“扩边”）。

1.2 页面数上限（max_pages）

max_pages 限制“成功抓取的页面数”（策略内部只对 result.success==True 的结果计数）。

策略会根据“剩余容量”在 link discovery 阶段裁剪下一层的 URL 数量：

• 若 valid_links > remaining_capacity：
  • 传了 url_scorer：按 score 倒序取前 remaining_capacity
  • 没传 url_scorer：按发现顺序取前 remaining_capacity

1.3 内链 / 外链（include_external）

策略的链接来源来自每个页面的 result.links：

• 默认只用 result.links["internal"]
• 当 include_external=True 时，会把 result.links["external"] 也加入候选

是否最终被抓取，还会经过 filter_chain（见下文）。

1.4 过滤链（filter_chain）

深爬的每个候选 URL（除了 depth=0 的起始 URL）都会走 filter_chain.apply(url)。

注意：BFSDeepCrawlStrategy.can_process_url() 对 depth=0 会绕过过滤，所以起始 URL 一定会抓一次。

---

2. 最小可运行示例（流式输出）

当 CrawlerRunConfig(stream=True) 时，深爬会返回一个 AsyncGenerator，你可以边抓边处理结果：

import asyncio

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig
from crawl4ai import FilterChain, URLPatternFilter
from crawl4ai.deep_crawling import BFSDeepCrawlStrategy


async def main():
    # 1) 过滤：只允许 example.com 域内 URL（示例）
    filter_chain = FilterChain([
        URLPatternFilter(["https://example.com/**"], use_glob=True),
    ])

    # 2) BFS 深爬策略
    deep = BFSDeepCrawlStrategy(
        max_depth=2,
        max_pages=50,
        include_external=False,
        filter_chain=filter_chain,
    )

    # 3) 把深爬策略挂到 run config 上
    run_cfg = CrawlerRunConfig(
        deep_crawl_strategy=deep,
        stream=True,        # 关键：流式返回
        verbose=False,
    )

    # 4) 启动 crawler 并开始深爬
    browser_cfg = BrowserConfig(headless=True)
    async with AsyncWebCrawler(browser_config=browser_cfg) as crawler:
        async for result in await crawler.arun("https://example.com", config=run_cfg):
            depth = (result.metadata or {}).get("depth")
            parent = (result.metadata or {}).get("parent_url")
            if result.success:
                print("OK ", depth, result.url, " <- ", parent)
            else:
                print("ERR", depth, result.url, getattr(result, "error_message", ""))


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

你会发现每个 result.metadata 里通常带有：

• depth：该页面在 BFS 中的层级
• parent_url：是从哪个页面发现出来的

---

3. 批量模式（一次性返回 List）

当 CrawlerRunConfig(stream=False)（默认）时，深爬会返回 List[CrawlResult]：

import asyncio

from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig
from crawl4ai.deep_crawling import BFSDeepCrawlStrategy


async def main():
    deep = BFSDeepCrawlStrategy(max_depth=1, max_pages=20)
    run_cfg = CrawlerRunConfig(deep_crawl_strategy=deep, stream=False, verbose=False)

    async with AsyncWebCrawler(browser_config=BrowserConfig(headless=True)) as crawler:
        results = await crawler.arun("https://example.com", config=run_cfg)
        for r in results:
            print(r.success, (r.metadata or {}).get("depth"), r.url)


asyncio.run(main())

---

4. 过滤：URLPatternFilter（最常用）

URLPatternFilter(patterns, use_glob=True, reverse=False) 支持多种模式：

• glob：**、*、{a,b}（内部会转成 regex）
• regex：以 ^ 开头 / 以 $ 结尾 / 包含 \\d 等会被视为正则
• reverse=True：把“命中”当作拒绝（用于 exclude）

示例：只允许 /docs/ 下的 URL，并排除 /docs/api/：

from crawl4ai import FilterChain, URLPatternFilter

filter_chain = FilterChain([
    URLPatternFilter(["https://example.com/docs/**"]),
    URLPatternFilter(["https://example.com/docs/api/**"], reverse=True),
])

---

5. 过滤：自定义 URLFilter（需要更强逻辑时）

当你要做“同域 + 路径前缀 + 子域”等复杂限制时，可以继承 URLFilter：

from urllib.parse import urlparse
from crawl4ai import URLFilter

class SameHostOnly(URLFilter):
    def __init__(self, seed_url: str):
        super().__init__(name="SameHostOnly")
        self.seed_host = (urlparse(seed_url).hostname or "").lower()

    def apply(self, url: str) -> bool:
        host = (urlparse(url).hostname or "").lower()
        ok = bool(host) and host == self.seed_host
        self._update_stats(ok)
        return ok

然后挂到 FilterChain([...]) 即可。

提示：FilterChain.apply() 会把 异步 filter 并发执行（如果 apply() 返回 awaitable），同步 filter 则是顺序快速判定；建议尽量写成同步快速逻辑。

---

6. URL 评分（url_scorer + score_threshold）

当你只想在大量链接中优先抓“更相关”的 URL，可以提供 url_scorer：

from crawl4ai.deep_crawling import BFSDeepCrawlStrategy
from crawl4ai.deep_crawling.scorers import KeywordRelevanceScorer, PathDepthScorer, CompositeScorer

scorer = CompositeScorer([
    KeywordRelevanceScorer(["pricing", "docs", "api"], weight=1.0),
    PathDepthScorer(optimal_depth=3, weight=0.3),
])

deep = BFSDeepCrawlStrategy(
    max_depth=3,
    max_pages=200,
    url_scorer=scorer,
    score_threshold=0.1,   # 分数低于阈值的 URL 会被跳过
)

评分的主要作用点：

• score < score_threshold：直接跳过
• 当 valid_links > remaining_capacity：按 score 排序截断，优先保留高分链接

---

7. 常见坑与排查

1. “只抓到很少页面”

• 检查 filter_chain 是否把大部分链接过滤掉了（特别是路径前缀限制）。
• max_depth=0 时只会抓起始 URL。
• include_external=False 会完全不扩展外链（即使页面上外链很多）。

2. “我设置了 deep_crawl_strategy，但只返回了一个页面”

• 确认你是通过 crawler.arun(start_url, config=run_cfg) 触发的（AsyncWebCrawler 会在初始化时用 DeepCrawlDecorator 包装 arun，当 run_cfg.deep_crawl_strategy 存在时才会走深爬逻辑）。

3. “流式 / 非流式写法不对”

• stream=True：async for result in await crawler.arun(...)
• stream=False：results = await crawler.arun(...); for r in results: ...