mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
feat(community): add bounded retries for Jina web fetch (#6143)
* feat(community): add bounded retries for Jina web fetch Closes #6116 Keep retries opt-in and share a total deadline across requests and backoff. Preserve cancellation and the existing Error contract. Include offline DNS and local-wheel fixture corrections needed to run the complete backend suite in the isolated verifier. Signed-off-by: 97three <2212371308@qq.com> * docs(community): scope Jina retry guidance to its provider Signed-off-by: 97three <2212371308@qq.com> * fix(community): jitter bounded Jina retry backoff * docs(community): apply provider-scoped Jina guidance Signed-off-by: 97three <2212371308@qq.com> --------- Signed-off-by: 97three <2212371308@qq.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
creed
GitHub
parent
d62d9e786c
commit
f4af75a7fd
@ -142,6 +142,8 @@ It is disabled by default; see the linked guide to enable it.
|
||||
|
||||
Jina, Browserless, and InfoQuest web fetches resolve relative links and image sources using the requested page URL (or a usable HTML base URL), so returned Markdown includes complete destinations. Link resolution preserves the surrounding HTML source, including malformed-page formatting.
|
||||
|
||||
Jina fetches support opt-in bounded retries via `max_retries` (default `0`) and `retry_budget_seconds` (default `30`) in the tool configuration. Retry waits are randomized within the time budget. Retries may increase upstream requests and cost; see [Jina fetch retries](backend/docs/CONFIGURATION.md#jina-fetch-retries).
|
||||
|
||||
Run `make doctor` at any time to verify your setup and get actionable fix hints.
|
||||
If you are opening a GitHub issue about a local setup or runtime problem, run
|
||||
`make support-bundle`. The command prints reporter next steps, writes a
|
||||
|
||||
@ -575,6 +575,41 @@ empty or omitted `include_domains`. These filters compose with `max_results`
|
||||
and the model's optional `time_range`. The model-visible arguments remain `query`
|
||||
and `time_range`; the filters do not apply to `web_fetch` or other search providers.
|
||||
|
||||
#### Jina fetch retries
|
||||
|
||||
Jina's `web_fetch` keeps one attempt by default. Configure retries on its existing
|
||||
`tools` entry; model-facing arguments remain unchanged:
|
||||
|
||||
```yaml
|
||||
tools:
|
||||
- name: web_fetch
|
||||
group: web
|
||||
use: deerflow.community.jina_ai.tools:web_fetch_tool
|
||||
timeout: 10
|
||||
max_retries: 2 # Additional attempts; default 0 (disabled)
|
||||
retry_budget_seconds: 30 # Total request + backoff budget; default 30
|
||||
```
|
||||
|
||||
`max_retries` must be a non-negative integer and `retry_budget_seconds` a finite,
|
||||
positive number (YAML numbers, not strings or booleans). Invalid values return an
|
||||
`Error:` without sending a request. The budget applies only when retries are enabled,
|
||||
starts before HTTP client creation, and covers all attempts and waits. Each HTTP
|
||||
request timeout is capped by the remaining budget; the outer deadline also bounds
|
||||
responses that keep delivering data. The existing `timeout` remains Jina's
|
||||
`X-Timeout` header and the per-request HTTP timeout limit.
|
||||
|
||||
Only HTTP 502/503/504 and HTTPX connection-establishment errors (`ConnectError`,
|
||||
`ConnectTimeout`) are retried. Authentication/client errors, 429, other statuses,
|
||||
empty successful responses, read/write timeouts and arbitrary exceptions are not
|
||||
retried. `Retry-After` is not interpreted. Backoff ceilings start at 0.5 seconds,
|
||||
double to 1 and 2 seconds, then stay at 4 seconds. Each asynchronous wait caps its
|
||||
ceiling by the remaining budget and independently samples a uniform factor from
|
||||
0.5 to 1.0, reducing synchronized retries without increasing the wait cap.
|
||||
Cancellation propagates during requests and waits. This stops local work; it
|
||||
cannot cancel work already started by Jina. Enabling retries can send up to `1 + max_retries` upstream requests
|
||||
and incur additional cost. Successful content and final `Error:` results retain
|
||||
the existing contract.
|
||||
|
||||
Serper `web_search` also accepts the optional model argument
|
||||
`time_range: "day" | "week" | "month" | "year"`. For example,
|
||||
`{"query": "Python releases", "time_range": "week"}` sends `tbs: "qdr:w"`
|
||||
|
||||
@ -0,0 +1,9 @@
|
||||
# Jina web fetch
|
||||
|
||||
Jina opt-in retries stay provider-local: `max_retries=0` preserves one attempt;
|
||||
`retry_budget_seconds` bounds enabled request sequences and asynchronous backoff
|
||||
with one asyncio deadline. Only 502/503/504 and connection-establishment failures
|
||||
retry; cancellation propagates. Backoff ceilings double from 0.5 to 4 seconds.
|
||||
Each wait caps the ceiling by the remaining budget, then multiplies it by a fresh
|
||||
uniform random factor from 0.5 to 1.0 to avoid synchronized retries.
|
||||
Offline coverage: `tests/test_jina_retries.py`.
|
||||
@ -1,5 +1,8 @@
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
import random
|
||||
|
||||
import httpx
|
||||
|
||||
@ -9,7 +12,8 @@ _api_key_warned = False
|
||||
|
||||
|
||||
class JinaClient:
|
||||
async def crawl(self, url: str, return_format: str = "html", timeout: int = 10, proxy: str | None = None, trust_env: bool = True) -> str:
|
||||
async def crawl(self, url: str, return_format: str = "html", timeout: int = 10, proxy: str | None = None, trust_env: bool = True, *, max_retries: int = 0, retry_budget_seconds: float = 30.0) -> str:
|
||||
"""Fetch with optional bounded retries; cancellation always propagates."""
|
||||
global _api_key_warned
|
||||
headers = {
|
||||
"Content-Type": "application/json",
|
||||
@ -23,24 +27,52 @@ class JinaClient:
|
||||
logger.warning("Jina API key is not set. Provide your own key to access a higher rate limit. See https://jina.ai/reader for more information.")
|
||||
data = {"url": url}
|
||||
try:
|
||||
client_kwargs: dict[str, object] = {"trust_env": trust_env}
|
||||
if proxy:
|
||||
client_kwargs["proxy"] = proxy
|
||||
async with httpx.AsyncClient(**client_kwargs) as client:
|
||||
response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=timeout)
|
||||
if isinstance(max_retries, bool) or not isinstance(max_retries, int) or max_retries < 0:
|
||||
raise ValueError("max_retries must be a non-negative integer")
|
||||
if isinstance(retry_budget_seconds, bool) or not isinstance(retry_budget_seconds, (int, float)) or not math.isfinite(retry_budget_seconds) or retry_budget_seconds <= 0:
|
||||
raise ValueError("retry_budget_seconds must be a finite positive number")
|
||||
|
||||
if response.status_code != 200:
|
||||
error_message = f"Jina API returned status {response.status_code}: {response.text}"
|
||||
logger.error(error_message)
|
||||
return f"Error: {error_message}"
|
||||
# HTTPX timeouts are per network phase, so use an outer deadline to
|
||||
# bound the complete request sequence (including waits and cleanup).
|
||||
deadline = asyncio.get_running_loop().time() + retry_budget_seconds if max_retries else None
|
||||
async with asyncio.timeout_at(deadline):
|
||||
client_kwargs: dict[str, object] = {"trust_env": trust_env}
|
||||
if proxy:
|
||||
client_kwargs["proxy"] = proxy
|
||||
async with httpx.AsyncClient(**client_kwargs) as client:
|
||||
delay = 0.5
|
||||
for attempt in range(max_retries + 1):
|
||||
remaining = deadline - asyncio.get_running_loop().time() if deadline is not None else None
|
||||
if remaining is not None and remaining <= 0:
|
||||
raise TimeoutError
|
||||
request_timeout = min(timeout, remaining) if remaining is not None else timeout
|
||||
try:
|
||||
response = await client.post("https://r.jina.ai/", headers=headers, json=data, timeout=request_timeout)
|
||||
except (httpx.ConnectError, httpx.ConnectTimeout):
|
||||
if attempt == max_retries:
|
||||
raise
|
||||
else:
|
||||
if response.status_code == 200:
|
||||
if response.text and response.text.strip():
|
||||
return response.text
|
||||
error_message = "Jina API returned empty response"
|
||||
logger.error(error_message)
|
||||
return f"Error: {error_message}"
|
||||
if response.status_code not in {502, 503, 504} or attempt == max_retries:
|
||||
error_message = f"Jina API returned status {response.status_code}: {response.text}"
|
||||
logger.error(error_message)
|
||||
return f"Error: {error_message}"
|
||||
|
||||
if not response.text or not response.text.strip():
|
||||
error_message = "Jina API returned empty response"
|
||||
logger.error(error_message)
|
||||
return f"Error: {error_message}"
|
||||
|
||||
return response.text
|
||||
# Only allowlisted failures reach the non-blocking wait.
|
||||
remaining = deadline - asyncio.get_running_loop().time()
|
||||
if remaining <= 0:
|
||||
raise TimeoutError
|
||||
await asyncio.sleep(min(delay, remaining) * random.uniform(0.5, 1.0))
|
||||
delay = min(delay * 2, 4.0)
|
||||
except Exception as e:
|
||||
error_message = f"Request to Jina API failed: {type(e).__name__}: {e}"
|
||||
if isinstance(e, TimeoutError) and max_retries and asyncio.get_running_loop().time() >= deadline:
|
||||
error_message = "Request to Jina API failed: retry time budget exhausted"
|
||||
else:
|
||||
error_message = f"Request to Jina API failed: {type(e).__name__}: {e}"
|
||||
logger.warning(error_message)
|
||||
return f"Error: {error_message}"
|
||||
|
||||
@ -56,12 +56,14 @@ async def web_fetch_tool(url: str) -> str:
|
||||
timeout = 10
|
||||
proxy = None
|
||||
trust_env = True
|
||||
retry_options = {}
|
||||
config = get_app_config().get_tool_config("web_fetch")
|
||||
if config is not None:
|
||||
timeout = _coerce_timeout(config.model_extra.get("timeout"), timeout)
|
||||
proxy = _coerce_proxy(config.model_extra.get("proxy"))
|
||||
trust_env = _coerce_bool(config.model_extra.get("trust_env"), trust_env)
|
||||
html_content = await jina_client.crawl(url, return_format="html", timeout=timeout, proxy=proxy, trust_env=trust_env)
|
||||
retry_options = {key: config.model_extra[key] for key in ("max_retries", "retry_budget_seconds") if key in config.model_extra}
|
||||
html_content = await jina_client.crawl(url, return_format="html", timeout=timeout, proxy=proxy, trust_env=trust_env, **retry_options)
|
||||
if isinstance(html_content, str) and html_content.startswith("Error:"):
|
||||
return html_content
|
||||
article = await asyncio.to_thread(readability_extractor.extract_article, html_content, url=url)
|
||||
|
||||
@ -19,6 +19,7 @@ EXPECTED_GUIDANCE_PATHS = {
|
||||
"backend/packages/harness/deerflow/agents/middlewares/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/agents/memory/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/agents/task_continuity/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/community/jina_ai/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/community/ragflow/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/community/tavily/AGENTS.md",
|
||||
"backend/packages/harness/deerflow/community/e2b_sandbox/AGENTS.md",
|
||||
|
||||
@ -9,6 +9,7 @@ from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import gc
|
||||
import ipaddress
|
||||
import sys
|
||||
from types import ModuleType, SimpleNamespace
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
@ -27,6 +28,17 @@ from deerflow.community.browser_automation.session import (
|
||||
SnapshotElement,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def public_dns():
|
||||
"""Keep mocked tool calls offline while exercising the real URL safety check."""
|
||||
with patch(
|
||||
"deerflow.community.browser_automation.tools._resolve_host_addresses",
|
||||
return_value=[ipaddress.ip_address("93.184.216.34")],
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
PlaywrightTimeoutError = type(
|
||||
"TimeoutError",
|
||||
(TimeoutError,),
|
||||
@ -70,6 +82,7 @@ class TestBrowserTools:
|
||||
manager.get_session.return_value = session
|
||||
return patch.object(tools, "get_browser_session_manager", return_value=manager), manager
|
||||
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_navigate_returns_snapshot(self):
|
||||
session = MagicMock()
|
||||
session.navigate = AsyncMock(return_value=_snapshot())
|
||||
@ -86,6 +99,7 @@ class TestBrowserTools:
|
||||
session.navigate.assert_awaited_once_with("https://example.com")
|
||||
manager.get_session.assert_called_once()
|
||||
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_navigate_emits_screenshot_artifact_and_browser_view(self, tmp_path):
|
||||
outputs = tmp_path / "outputs"
|
||||
outputs.mkdir()
|
||||
@ -113,6 +127,7 @@ class TestBrowserTools:
|
||||
assert meta["url"] == "https://example.com/"
|
||||
assert meta["title"] == "Example"
|
||||
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_navigate_screenshot_failure_does_not_break_action(self, tmp_path):
|
||||
outputs = tmp_path / "outputs"
|
||||
outputs.mkdir()
|
||||
|
||||
@ -9,6 +9,16 @@ import pytest
|
||||
from deerflow.community.crawl4ai.crawl4ai_client import Crawl4AiClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def public_dns():
|
||||
"""Keep mocked tool calls offline while exercising the real URL safety check."""
|
||||
with patch(
|
||||
"deerflow.community.url_safety.resolve_host_addresses",
|
||||
return_value=[ipaddress.ip_address("93.184.216.34")],
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
class AsyncMock(MagicMock):
|
||||
"""Mock that supports async call."""
|
||||
|
||||
@ -164,6 +174,7 @@ class TestCrawl4AiTools:
|
||||
"""Tests for the Crawl4AI tool functions."""
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_web_fetch_tool_success(self, mock_build):
|
||||
from deerflow.community.crawl4ai import tools
|
||||
|
||||
@ -178,6 +189,7 @@ class TestCrawl4AiTools:
|
||||
assert "Error:" not in result
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_web_fetch_tool_truncates_to_4096(self, mock_build):
|
||||
from deerflow.community.crawl4ai import tools
|
||||
|
||||
@ -191,6 +203,7 @@ class TestCrawl4AiTools:
|
||||
assert len(result) == 4096
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_web_fetch_tool_error_passthrough(self, mock_build):
|
||||
from deerflow.community.crawl4ai import tools
|
||||
|
||||
@ -202,8 +215,10 @@ class TestCrawl4AiTools:
|
||||
result = await tools.web_fetch_tool.ainvoke("https://example.com")
|
||||
|
||||
assert result.startswith("Error:")
|
||||
mock_client.fetch_markdown.assert_called_once()
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_web_fetch_tool_exception(self, mock_build):
|
||||
from deerflow.community.crawl4ai import tools
|
||||
|
||||
@ -215,6 +230,7 @@ class TestCrawl4AiTools:
|
||||
result = await tools.web_fetch_tool.ainvoke("https://example.com")
|
||||
|
||||
assert result.startswith("Error:")
|
||||
mock_client.fetch_markdown.assert_called_once()
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
async def test_web_fetch_tool_rejects_metadata_ip(self, mock_build):
|
||||
@ -255,6 +271,7 @@ class TestCrawl4AiTools:
|
||||
mock_client.fetch_markdown.assert_called_once()
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_web_fetch_tool_reads_config_once(self, mock_build):
|
||||
"""Config is read exactly once per invocation (no split read on hot-reload)."""
|
||||
from deerflow.community.crawl4ai import tools
|
||||
@ -269,6 +286,7 @@ class TestCrawl4AiTools:
|
||||
mock_cfg.assert_called_once_with("web_fetch")
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_web_fetch_tool_passes_configured_filter(self, mock_build):
|
||||
from deerflow.community.crawl4ai import tools
|
||||
|
||||
@ -283,6 +301,7 @@ class TestCrawl4AiTools:
|
||||
assert mock_client.fetch_markdown.call_args.kwargs.get("filter_mode") == "raw"
|
||||
|
||||
@patch("deerflow.community.crawl4ai.tools._build_client")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
async def test_web_fetch_tool_invalid_filter_falls_back_to_fit(self, mock_build):
|
||||
from deerflow.community.crawl4ai import tools
|
||||
|
||||
|
||||
@ -92,7 +92,7 @@ packages = ["demo_extension"]
|
||||
)
|
||||
|
||||
|
||||
def _write_host_project(root: Path) -> None:
|
||||
def _write_host_project(root: Path, *, no_index: bool = False) -> None:
|
||||
backend = root / "backend"
|
||||
backend.mkdir()
|
||||
(backend / "pyproject.toml").write_text(
|
||||
@ -111,6 +111,10 @@ default-groups = ["extensions"]
|
||||
""",
|
||||
encoding="utf-8",
|
||||
)
|
||||
if no_index:
|
||||
# Use uv's supported project setting; UV_NO_INDEX is not recognized.
|
||||
with (backend / "pyproject.toml").open("a", encoding="utf-8") as project:
|
||||
project.write("no-index = true\n")
|
||||
(root / "config.yaml").write_text("config_version: 1\n", encoding="utf-8")
|
||||
|
||||
|
||||
@ -820,10 +824,9 @@ def test_install_rejects_a_pypi_requirement_resolved_from_an_external_local_whee
|
||||
root = tmp_path / "deer-flow"
|
||||
wheels = tmp_path / "wheels"
|
||||
root.mkdir()
|
||||
_write_host_project(root)
|
||||
_write_host_project(root, no_index=True)
|
||||
_write_demo_wheel(wheels)
|
||||
monkeypatch.setenv("UV_FIND_LINKS", str(wheels))
|
||||
monkeypatch.setenv("UV_NO_INDEX", "1")
|
||||
pyproject_path = root / "backend" / "pyproject.toml"
|
||||
config_path = root / "config.yaml"
|
||||
before = (pyproject_path.read_bytes(), config_path.read_bytes())
|
||||
@ -843,12 +846,11 @@ def test_install_rejects_a_local_wheel_directory_ignored_by_the_docker_context(
|
||||
) -> None:
|
||||
root = tmp_path / "deer-flow"
|
||||
root.mkdir()
|
||||
_write_host_project(root)
|
||||
_write_host_project(root, no_index=True)
|
||||
wheels = root / "backend" / relative_wheels
|
||||
wheels.parent.mkdir(parents=True, exist_ok=True)
|
||||
_write_demo_wheel(wheels)
|
||||
monkeypatch.setenv("UV_FIND_LINKS", str(wheels))
|
||||
monkeypatch.setenv("UV_NO_INDEX", "1")
|
||||
pyproject_path = root / "backend" / "pyproject.toml"
|
||||
config_path = root / "config.yaml"
|
||||
before = (pyproject_path.read_bytes(), config_path.read_bytes())
|
||||
@ -866,13 +868,12 @@ def test_install_rejects_a_relative_find_links_wheelhouse_outside_the_build_cont
|
||||
) -> None:
|
||||
root = tmp_path / "deer-flow"
|
||||
root.mkdir()
|
||||
_write_host_project(root)
|
||||
_write_host_project(root, no_index=True)
|
||||
_write_demo_wheel(root / "backend" / "wheelhouse")
|
||||
# uv resolves a relative UV_FIND_LINKS against its working directory (the
|
||||
# backend project), so the lock records a relative registry that only
|
||||
# exists on this host.
|
||||
monkeypatch.setenv("UV_FIND_LINKS", "wheelhouse")
|
||||
monkeypatch.setenv("UV_NO_INDEX", "1")
|
||||
pyproject_path = root / "backend" / "pyproject.toml"
|
||||
config_path = root / "config.yaml"
|
||||
before = (pyproject_path.read_bytes(), config_path.read_bytes())
|
||||
|
||||
@ -4,6 +4,18 @@ import ipaddress
|
||||
import json
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def public_dns():
|
||||
"""Keep mocked tool calls offline while exercising the real URL safety check."""
|
||||
with patch(
|
||||
"deerflow.community.url_safety.resolve_host_addresses",
|
||||
return_value=[ipaddress.ip_address("93.184.216.34")],
|
||||
):
|
||||
yield
|
||||
|
||||
|
||||
class TestWebSearchTool:
|
||||
@patch.dict("os.environ", {}, clear=True)
|
||||
@ -69,6 +81,7 @@ class TestWebFetchTool:
|
||||
@patch.dict("os.environ", {}, clear=True)
|
||||
@patch("deerflow.community.fastcrw.tools.FirecrawlApp")
|
||||
@patch("deerflow.community.fastcrw.tools.get_app_config")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
def test_fetch_uses_web_fetch_config(self, mock_get_app_config, mock_fastcrw_cls):
|
||||
fetch_config = MagicMock()
|
||||
fetch_config.model_extra = {"api_key": "fastcrw-fetch-key", "base_url": "http://localhost:3000"}
|
||||
@ -100,6 +113,7 @@ class TestWebFetchTool:
|
||||
@patch.dict("os.environ", {}, clear=True)
|
||||
@patch("deerflow.community.fastcrw.tools.FirecrawlApp")
|
||||
@patch("deerflow.community.fastcrw.tools.get_app_config")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
def test_fetch_returns_error_when_no_content(self, mock_get_app_config, mock_fastcrw_cls):
|
||||
mock_get_app_config.return_value.get_tool_config.return_value = None
|
||||
|
||||
@ -115,6 +129,7 @@ class TestWebFetchTool:
|
||||
@patch.dict("os.environ", {}, clear=True)
|
||||
@patch("deerflow.community.fastcrw.tools.FirecrawlApp")
|
||||
@patch("deerflow.community.fastcrw.tools.get_app_config")
|
||||
@pytest.mark.usefixtures("public_dns")
|
||||
def test_fetch_returns_error_string_on_exception(self, mock_get_app_config, mock_fastcrw_cls):
|
||||
mock_get_app_config.return_value.get_tool_config.return_value = None
|
||||
mock_fastcrw_cls.return_value.scrape.side_effect = RuntimeError("scrape failed")
|
||||
|
||||
181
backend/tests/test_jina_retries.py
Normal file
181
backend/tests/test_jina_retries.py
Normal file
@ -0,0 +1,181 @@
|
||||
"""Offline coverage of Jina's opt-in retry policy."""
|
||||
|
||||
import asyncio
|
||||
import random
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import AsyncMock, Mock
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from deerflow.community.jina_ai.jina_client import JinaClient
|
||||
|
||||
pytestmark = pytest.mark.anyio
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def anyio_backend():
|
||||
return "asyncio"
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def requests(monkeypatch):
|
||||
post = AsyncMock()
|
||||
client = AsyncMock()
|
||||
client.__aenter__.return_value = client
|
||||
client.post = post
|
||||
monkeypatch.setattr(httpx, "AsyncClient", lambda **kwargs: client)
|
||||
monkeypatch.setenv("JINA_API_KEY", "test-key")
|
||||
return post
|
||||
|
||||
|
||||
async def test_default_single_attempt(requests, monkeypatch):
|
||||
jitter = Mock()
|
||||
monkeypatch.setattr(random, "uniform", jitter)
|
||||
requests.return_value = httpx.Response(503, text="unavailable")
|
||||
assert "503" in await JinaClient().crawl("https://example.com")
|
||||
assert requests.await_count == 1
|
||||
jitter.assert_not_called()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", [httpx.Response(502), httpx.Response(503), httpx.Response(504), httpx.ConnectError("offline"), httpx.ConnectTimeout("offline")])
|
||||
async def test_transient_recovers(requests, monkeypatch, failure):
|
||||
monkeypatch.setattr(asyncio, "sleep", AsyncMock())
|
||||
requests.side_effect = [failure, httpx.Response(200, text="success")]
|
||||
assert await JinaClient().crawl("https://example.com", max_retries=1) == "success"
|
||||
assert requests.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize("failure", [httpx.Response(code) for code in (400, 401, 403, 404, 429, 500)] + [httpx.Response(200, text=" "), httpx.ReadTimeout("slow"), httpx.WriteError("write"), RuntimeError("unexpected")])
|
||||
async def test_permanent_failure(requests, failure):
|
||||
requests.side_effect = [failure, httpx.Response(200, text="unexpected")]
|
||||
assert (await JinaClient().crawl("https://example.com", max_retries=2)).startswith("Error:")
|
||||
assert requests.await_count == 1
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("jitter_factor", "expected_waits"),
|
||||
[(0.5, [0.25, 0.5, 1, 2, 2, 2]), (0.75, [0.375, 0.75, 1.5, 3, 3, 3]), (1.0, [0.5, 1, 2, 4, 4, 4])],
|
||||
)
|
||||
async def test_exhaustion_and_capped_backoff(requests, monkeypatch, jitter_factor, expected_waits):
|
||||
sleep = AsyncMock()
|
||||
jitter = Mock(return_value=jitter_factor)
|
||||
monkeypatch.setattr(asyncio, "sleep", sleep)
|
||||
monkeypatch.setattr(random, "uniform", jitter)
|
||||
requests.return_value = httpx.Response(503, text="unavailable")
|
||||
assert "503" in await JinaClient().crawl("https://example.com", max_retries=6)
|
||||
assert requests.await_count == 7
|
||||
assert [call.args[0] for call in sleep.await_args_list] == expected_waits
|
||||
assert jitter.call_count == 6
|
||||
assert all(call.args == (0.5, 1.0) for call in jitter.call_args_list)
|
||||
|
||||
|
||||
async def test_each_retry_samples_new_jitter(requests, monkeypatch):
|
||||
sleep = AsyncMock()
|
||||
jitter = Mock(side_effect=[0.5, 1.0, 0.75])
|
||||
monkeypatch.setattr(asyncio, "sleep", sleep)
|
||||
monkeypatch.setattr(random, "uniform", jitter)
|
||||
requests.return_value = httpx.Response(503, text="unavailable")
|
||||
|
||||
assert "503" in await JinaClient().crawl("https://example.com", max_retries=3)
|
||||
|
||||
assert requests.await_count == 4
|
||||
assert [call.args[0] for call in sleep.await_args_list] == [0.25, 1, 1.5]
|
||||
assert jitter.call_count == 3
|
||||
|
||||
|
||||
async def test_backoff_jitter_is_capped_by_remaining_budget(requests, monkeypatch):
|
||||
sleep = AsyncMock()
|
||||
monkeypatch.setattr(asyncio, "sleep", sleep)
|
||||
monkeypatch.setattr(random, "uniform", lambda low, high: 0.5)
|
||||
requests.side_effect = [httpx.Response(503), httpx.Response(200, text="success")]
|
||||
|
||||
assert await JinaClient().crawl("https://example.com", max_retries=1, retry_budget_seconds=0.1) == "success"
|
||||
|
||||
assert requests.await_count == 2
|
||||
sleep.assert_awaited_once()
|
||||
assert 0 < sleep.await_args.args[0] <= 0.05
|
||||
|
||||
|
||||
@pytest.mark.parametrize("during_backoff", [False, True])
|
||||
async def test_shared_deadline(requests, monkeypatch, during_backoff):
|
||||
monkeypatch.setattr(random, "uniform", lambda low, high: 1.0)
|
||||
|
||||
async def post(*args, **kwargs):
|
||||
assert 0 < kwargs["timeout"] <= 0.05
|
||||
if not during_backoff:
|
||||
await asyncio.Event().wait()
|
||||
return httpx.Response(503)
|
||||
|
||||
requests.side_effect = post
|
||||
result = await asyncio.wait_for(JinaClient().crawl("https://example.com", max_retries=2, retry_budget_seconds=0.05), timeout=1)
|
||||
assert result.startswith("Error:")
|
||||
assert "budget" in result
|
||||
assert requests.await_count == 1
|
||||
|
||||
|
||||
async def test_budget_is_shared_across_attempts(requests, monkeypatch):
|
||||
timeouts = []
|
||||
real_sleep = asyncio.sleep
|
||||
|
||||
async def post(*args, **kwargs):
|
||||
timeouts.append(kwargs["timeout"])
|
||||
if len(timeouts) == 1:
|
||||
await real_sleep(0.02)
|
||||
return httpx.Response(503)
|
||||
await asyncio.Event().wait()
|
||||
|
||||
monkeypatch.setattr(asyncio, "sleep", AsyncMock())
|
||||
requests.side_effect = post
|
||||
result = await asyncio.wait_for(JinaClient().crawl("https://example.com", max_retries=2, retry_budget_seconds=0.1), 1)
|
||||
assert "budget" in result
|
||||
assert len(timeouts) == 2
|
||||
assert timeouts[1] < timeouts[0]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("during_backoff", [False, True])
|
||||
async def test_cancellation(requests, monkeypatch, during_backoff):
|
||||
entered = asyncio.Event()
|
||||
cancelled = asyncio.Event()
|
||||
|
||||
async def block(*args, **kwargs):
|
||||
entered.set()
|
||||
try:
|
||||
await asyncio.Event().wait()
|
||||
finally:
|
||||
cancelled.set()
|
||||
|
||||
if during_backoff:
|
||||
requests.return_value = httpx.Response(503)
|
||||
monkeypatch.setattr(asyncio, "sleep", block)
|
||||
else:
|
||||
requests.side_effect = block
|
||||
task = asyncio.create_task(JinaClient().crawl("https://example.com", max_retries=2))
|
||||
try:
|
||||
await asyncio.wait_for(entered.wait(), 1)
|
||||
task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await asyncio.wait_for(task, 1)
|
||||
assert cancelled.is_set()
|
||||
assert requests.await_count == 1
|
||||
finally:
|
||||
task.cancel()
|
||||
await asyncio.gather(task, return_exceptions=True)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("options", [{"max_retries": value} for value in (-1, True, 1.5, "2", None)] + [{"retry_budget_seconds": value} for value in (0, -1, True, float("nan"), float("inf"), "10", None)])
|
||||
async def test_invalid_settings_do_not_send(requests, options):
|
||||
assert (await JinaClient().crawl("https://example.com", **options)).startswith("Error:")
|
||||
requests.assert_not_awaited()
|
||||
|
||||
|
||||
async def test_tool_forwards_settings(monkeypatch):
|
||||
from deerflow.community.jina_ai import tools
|
||||
|
||||
crawl = AsyncMock(return_value="Error: test")
|
||||
config = SimpleNamespace(model_extra={"max_retries": 2, "retry_budget_seconds": 12.5})
|
||||
monkeypatch.setattr(tools, "get_app_config", lambda: SimpleNamespace(get_tool_config=lambda name: config))
|
||||
monkeypatch.setattr(JinaClient, "crawl", crawl)
|
||||
assert await tools.web_fetch_tool.ainvoke("https://example.com") == "Error: test"
|
||||
assert crawl.call_args.kwargs["max_retries"] == 2
|
||||
assert crawl.call_args.kwargs["retry_budget_seconds"] == 12.5
|
||||
@ -1164,6 +1164,9 @@ tools:
|
||||
group: web
|
||||
use: deerflow.community.jina_ai.tools:web_fetch_tool
|
||||
timeout: 10
|
||||
# Opt-in retries may increase upstream requests and cost.
|
||||
# max_retries: 0
|
||||
# retry_budget_seconds: 30
|
||||
# Optional proxy for restricted networks / Docker / WSL.
|
||||
# Use host.docker.internal instead of 127.0.0.1 when the proxy runs on the host.
|
||||
# proxy: $HTTPS_PROXY
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user