From 372ba797ab18ed6773b0eb8f2bf590714b61d702 Mon Sep 17 00:00:00 2001 From: LeonSGP43 Date: Wed, 17 Jun 2026 11:58:29 +0800 Subject: [PATCH] fix(web): parallelize firecrawl extract fanout --- plugins/web/firecrawl/provider.py | 132 ++++++++---------- .../web/test_web_search_provider_plugins.py | 41 ++++++ 2 files changed, 101 insertions(+), 72 deletions(-) diff --git a/plugins/web/firecrawl/provider.py b/plugins/web/firecrawl/provider.py index 0fa99bf58f6cc..bbf1445423764 100644 --- a/plugins/web/firecrawl/provider.py +++ b/plugins/web/firecrawl/provider.py @@ -420,9 +420,9 @@ def search(self, query: str, limit: int = 5) -> Dict[str, Any]: async def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]: """Extract content from one or more URLs via Firecrawl. - Async; each URL is scraped in a background thread with a 60s - timeout. After scraping, the final URL (post-redirect) is - re-checked against website-access policy. + Async; URLs are scraped concurrently in background threads with a + 60s per-URL timeout. After scraping, the final URL (post-redirect) + is re-checked against website-access policy. Accepted kwargs (others ignored for forward compat): - ``format``: ``"markdown"`` or ``"html"``; default is both @@ -450,12 +450,12 @@ async def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]: # module level (lazy-friendly because the website_policy import is # cheap) so monkeypatching it in tests works as expected. - results: List[Dict[str, Any]] = [] + concurrency_limit = min(max(len(urls), 1), 4) + scrape_slots = asyncio.Semaphore(concurrency_limit) - for url in urls: + async def extract_one(url: str) -> Dict[str, Any]: if _is_interrupted(): - results.append({"url": url, "error": "Interrupted", "title": ""}) - continue + return {"url": url, "error": "Interrupted", "title": ""} # Pre-scrape website policy gate blocked = check_website_access(url) @@ -465,36 +465,33 @@ async def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]: blocked["host"], blocked["rule"], ) - results.append( - { - "url": url, - "title": "", - "content": "", - "error": blocked["message"], - "blocked_by_policy": { - "host": blocked["host"], - "rule": blocked["rule"], - "source": blocked["source"], - }, - } - ) - continue + return { + "url": url, + "title": "", + "content": "", + "error": blocked["message"], + "blocked_by_policy": { + "host": blocked["host"], + "rule": blocked["rule"], + "source": blocked["source"], + }, + } try: - logger.info("Firecrawl scraping: %s", url) - try: - scrape_result = await asyncio.wait_for( - asyncio.to_thread( - _get_firecrawl_client().scrape, - url=url, - formats=formats, - ), - timeout=60, - ) - except asyncio.TimeoutError: - logger.warning("Firecrawl scrape timed out for %s", url) - results.append( - { + async with scrape_slots: + logger.info("Firecrawl scraping: %s", url) + try: + scrape_result = await asyncio.wait_for( + asyncio.to_thread( + _get_firecrawl_client().scrape, + url=url, + formats=formats, + ), + timeout=60, + ) + except asyncio.TimeoutError: + logger.warning("Firecrawl scrape timed out for %s", url) + return { "url": url, "title": "", "content": "", @@ -503,8 +500,6 @@ async def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]: "or unresponsive. Try browser_navigate instead." ), } - ) - continue scrape_payload = _extract_scrape_payload(scrape_result) metadata = scrape_payload.get("metadata", {}) @@ -531,21 +526,18 @@ async def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]: final_blocked["host"], final_blocked["rule"], ) - results.append( - { - "url": final_url, - "title": title, - "content": "", - "raw_content": "", - "error": final_blocked["message"], - "blocked_by_policy": { - "host": final_blocked["host"], - "rule": final_blocked["rule"], - "source": final_blocked["source"], - }, - } - ) - continue + return { + "url": final_url, + "title": title, + "content": "", + "raw_content": "", + "error": final_blocked["message"], + "blocked_by_policy": { + "host": final_blocked["host"], + "rule": final_blocked["rule"], + "source": final_blocked["source"], + }, + } # Choose markdown vs html according to the requested format if format == "markdown" or (format is None and content_markdown): @@ -553,28 +545,24 @@ async def extract(self, urls: List[str], **kwargs: Any) -> List[Dict[str, Any]]: else: chosen_content = content_html or content_markdown or "" - results.append( - { - "url": final_url, - "title": title, - "content": chosen_content, - "raw_content": chosen_content, - "metadata": metadata, - } - ) + return { + "url": final_url, + "title": title, + "content": chosen_content, + "raw_content": chosen_content, + "metadata": metadata, + } except Exception as scrape_err: # noqa: BLE001 logger.debug("Firecrawl scrape failed for %s: %s", url, scrape_err) - results.append( - { - "url": url, - "title": "", - "content": "", - "raw_content": "", - "error": str(scrape_err), - } - ) - - return results + return { + "url": url, + "title": "", + "content": "", + "raw_content": "", + "error": str(scrape_err), + } + + return await asyncio.gather(*(extract_one(url) for url in urls)) def get_setup_schema(self) -> Dict[str, Any]: return { diff --git a/tests/plugins/web/test_web_search_provider_plugins.py b/tests/plugins/web/test_web_search_provider_plugins.py index 2177d875c4b1e..ebdd696fd2cbf 100644 --- a/tests/plugins/web/test_web_search_provider_plugins.py +++ b/tests/plugins/web/test_web_search_provider_plugins.py @@ -20,6 +20,7 @@ import asyncio import inspect +import time import pytest @@ -353,6 +354,46 @@ def test_firecrawl_extract_is_async(self) -> None: assert p is not None assert inspect.iscoroutinefunction(p.extract) is True + def test_firecrawl_extract_scrapes_multiple_urls_concurrently( + self, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + from plugins.web.firecrawl.provider import FirecrawlWebSearchProvider + + class DummyClient: + def scrape(self, *, url, formats): + time.sleep(0.20) + return { + "metadata": { + "title": f"title:{url}", + "sourceURL": url, + }, + "markdown": f"markdown:{url}", + } + + monkeypatch.setattr( + "plugins.web.firecrawl.provider._get_firecrawl_client", + lambda: DummyClient(), + ) + monkeypatch.setattr( + "plugins.web.firecrawl.provider.check_website_access", + lambda url: None, + ) + + provider = FirecrawlWebSearchProvider() + urls = ["https://example.com/one", "https://example.com/two"] + + started = time.monotonic() + results = asyncio.run(provider.extract(urls, format="markdown")) + elapsed = time.monotonic() - started + + assert [result["url"] for result in results] == urls + assert [result["content"] for result in results] == [ + "markdown:https://example.com/one", + "markdown:https://example.com/two", + ] + assert elapsed < 0.32 + def test_exa_extract_is_sync(self) -> None: _ensure_plugins_loaded() from agent.web_search_registry import get_provider