自适应爬虫

这AdaptiveCrawler该类实现了智能网页爬取，可以自动判断何时已收集到足够的信息来回答查询。它使用三层评分系统来评估覆盖率、一致性和饱和度。

构造函数

AdaptiveCrawler(
    crawler: AsyncWebCrawler,
    config: Optional[AdaptiveConfig] = None
)

参数

  • 爬虫（AsyncWebCrawler ): 用于抓取页面的底层网络爬虫实例
  • 配置（Optional[AdaptiveConfig] ): 自适应抓取行为的配置设置。如果未提供，则使用默认设置。

主要方法

消化（）

从具有特定查询的 URL 开始执行自适应爬取的主要方法。

async def digest(
    start_url: str,
    query: str,
    resume_from: Optional[Union[str, Path]] = None
) -> CrawlState

参数

  • 起始网址（str ): 抓取的起始URL
  • 询问 （str ): 引导抓取过程的搜索查询
  • 恢复_从（Optional[Union[str, Path]] ): 保存状态文件的路径，用于恢复

返回

  • CrawlState：包含所有已抓取的 URL、知识库和指标的最终抓取状态

例子

async with AsyncWebCrawler() as crawler:
    adaptive = AdaptiveCrawler(crawler)
    state = await adaptive.digest(
        start_url="https://docs.python.org",
        query="async context managers"
    )

特性

信心

当前置信度分数（0-1）表示信息充分性。

@property
def confidence(self) -> float

覆盖率统计

包含详细覆盖率统计数据的字典。

@property  
def coverage_stats(self) -> Dict[str, float]

返回： - 覆盖率：查询术语覆盖率得分 - 一致性：信息一致性得分 - 饱和度：内容饱和度得分 - 置信度：总体置信度得分

足够

布尔值，指示是否已收集到足够的信息。

@property
def is_sufficient(self) -> bool

状态

访问当前爬网状态。

@property
def state(self) -> CrawlState

方法

获取相关内容（）

从知识库中检索最相关的内容。

def get_relevant_content(
    self,
    top_k: int = 5
) -> List[Dict[str, Any]]

参数

  • 顶部_k（int ): 返回的最相关文档数量（默认值：5）

返回

包含以下内容的词典列表： - url：页面的 URL - content：页面内容 - score：相关性分数 - metadata：附加页面元数据

以格式化的输出显示爬网统计信息。

def print_stats(
    self,
    detailed: bool = False
) -> None

参数

  • 详细的 （bool ): 如果为 True，则显示带有颜色的详细指标。如果为 False，则显示摘要表。

导出知识库（）

将收集的知识库导出到 JSONL 文件。

def export_knowledge_base(
    self,
    path: Union[str, Path]
) -> None

参数

  • 小路 （Union[str, Path] ): JSONL 导出的输出文件路径

例子

adaptive.export_knowledge_base("my_knowledge.jsonl")

导入知识库（）

导入先前导出的知识库。

def import_knowledge_base(
    self,
    path: Union[str, Path]
) -> None

参数

  • 小路 （Union[str, Path] ): 要导入的 JSONL 文件的路径

配置

这AdaptiveConfig类控制自适应爬取的行为：

@dataclass
class AdaptiveConfig:
    confidence_threshold: float = 0.8      # Stop when confidence reaches this
    max_pages: int = 50                    # Maximum pages to crawl
    top_k_links: int = 5                   # Links to follow per page
    min_gain_threshold: float = 0.1        # Minimum expected gain to continue
    save_state: bool = False               # Auto-save crawl state
    state_path: Optional[str] = None       # Path for state persistence

自定义配置示例

config = AdaptiveConfig(
    confidence_threshold=0.7,
    max_pages=20,
    top_k_links=3
)

adaptive = AdaptiveCrawler(crawler, config=config)

完整示例

import asyncio
from crawl4ai import AsyncWebCrawler, AdaptiveCrawler, AdaptiveConfig

async def main():
    # Configure adaptive crawling
    config = AdaptiveConfig(
        confidence_threshold=0.75,
        max_pages=15,
        save_state=True,
        state_path="my_crawl.json"
    )

    async with AsyncWebCrawler() as crawler:
        adaptive = AdaptiveCrawler(crawler, config)

        # Start crawling
        state = await adaptive.digest(
            start_url="https://example.com/docs",
            query="authentication oauth2 jwt"
        )

        # Check results
        print(f"Confidence achieved: {adaptive.confidence:.0%}")
        adaptive.print_stats()

        # Get most relevant pages
        for page in adaptive.get_relevant_content(top_k=3):
            print(f"- {page['url']} (score: {page['score']:.2f})")

        # Export for later use
        adaptive.export_knowledge_base("auth_knowledge.jsonl")

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

参见

> Feedback