WEBDIENSTE/openswarm
1
0
Fork 0
mirror of https://github.com/openswarm-ai/openswarm.git synced 2026-09-11 12:17:45 +02:00
Code Issues Packages Projects Releases Wiki Activity

[eric] websearch: fast-first bounded search/fetch cascade for human-speed

Browse Source
This commit is contained in:
ciregenz
2026-06-03 18:23:08 -07:00
parent 928df4f324
commit cf8e1f25f7
2 changed files with 311 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/apps/web/web.py
+123 -79
View File
@@ -9,6 +9,7 @@ extraction in the MCP process.
Mounted at `/api/web`.
"""
import asyncio
from contextlib import asynccontextmanager
from typing import Any
@@ -76,6 +77,21 @@ GEMINI_GROUNDING_MODEL = "gemini-2.5-flash" # cheapest + fastest for grounded c
OPENAI_API_BASE = "https://api.openai.com/v1"
OPENAI_SEARCH_MODEL = "gpt-5-mini" # cheapest model that supports web_search_preview
# Per-attempt timeouts for the search/fetch cascade. The fast-first ORDERING is
# what fixes the ~75s stall (DDG answers in ~1s so the slow grounded backends are
# rarely reached); these bounds are hang safety-nets, set just ABOVE each path's
# own httpx timeout so a normally-slow call still completes and only a truly hung
# provider (no response at all) gets cut. Grounded native search legitimately
# takes 32-42s (httpx ceiling 45s), so its leash sits at 48s, NOT below 45, or
# we'd clip the slow tail of a valid paid call.
_DDG_ATTEMPT_TIMEOUT = 6.0 # DDG answers <1s; >6s is a network hang, fall through
_GROUNDED_ATTEMPT_TIMEOUT = 48.0 # just above the providers' own 45s httpx timeout
# Local httpx + trafilatura fetch of a real page; the fast path for /fetch
# (normal pages return in <2s). Set just above WebFetchTool's own 30s httpx
# ceiling so a valid-but-slow page still completes locally instead of being
# clipped down to a grounded summary; only a truly hung server gets cut.
_LOCAL_FETCH_TIMEOUT = 32.0
async def _gemini_grounded_call(api_key: str, prompt: str, *, use_url_context: bool) -> dict:
"""Call Gemini with googleSearch (+ optionally urlContext) grounding.
@@ -430,72 +446,70 @@ async def search(body: SearchBody) -> dict:
"backend": "openai_subscription",
}
# Ordered cascade: primary's native API key first (most direct), then
# the user's connected subscriptions (free via OAuth), then the
# opposite-provider native key, then DuckDuckGo last as a guaranteed
# fallback (which is rate-limit-prone but free).
if primary == "openai":
cascade = [
("openai_native", try_openai),
("openai_subscription", try_openai_subscription),
("gemini_native", try_gemini),
("gemini_subscription", try_gemini_subscription),
]
elif primary in ("gemini", "google"):
cascade = [
("gemini_native", try_gemini),
("gemini_subscription", try_gemini_subscription),
("openai_native", try_openai),
("openai_subscription", try_openai_subscription),
]
else:
cascade = [
("gemini_native", try_gemini),
("gemini_subscription", try_gemini_subscription),
("openai_native", try_openai),
("openai_subscription", try_openai_subscription),
]
for name, fn in cascade:
async def try_ddg():
# Fast path: direct HTML search, sub-second when DDG isn't throttling us.
# Returns None on a real no-hits OR a 202 throttle so the chain falls
# through to the slower-but-grounded backends.
from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited
try:
res = await fn()
text = await WebSearchTool._search_ddg(body.query, body.num_results)
except DDGRateLimited:
# Surface the throttle as a recorded error (not a silent None) so the
# caller can see WHY we fell through to a slower backend.
raise RuntimeError("DuckDuckGo rate-limited (HTTP 202)") from None
if not text:
return None
return {"query": body.query, "results": text, "backend": "ddg"}
# Fast-first cascade: DDG leads (~1s = human speed); the 30-42s LLM-grounded
# backends are the reliable fallback when DDG is throttled or empty. The
# primary hint only reorders the grounded tier (native key before the
# same-provider subscription). Every attempt is wait_for-bounded so a slow
# or hung provider fails over fast instead of stalling the whole request.
grounded = [
("gemini_native", try_gemini),
("gemini_subscription", try_gemini_subscription),
("openai_native", try_openai),
("openai_subscription", try_openai_subscription),
]
if primary == "openai":
grounded = grounded[2:] + grounded[:2]
cascade = [("ddg", try_ddg, _DDG_ATTEMPT_TIMEOUT)] + [
(name, fn, _GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded
]
for name, fn, timeout in cascade:
try:
res = await asyncio.wait_for(fn(), timeout=timeout)
if res is not None:
if errors:
res["cascade_errors"] = errors
return res
except asyncio.TimeoutError:
errors.append(f"{name}: timed out after {timeout:.0f}s")
except Exception as e:
errors.append(f"{name}: {str(e)[:150]}")
# DDG fallback.
from backend.apps.agents.tools.web import WebSearchTool
try:
tool = WebSearchTool()
parts = await tool.execute(
{"query": body.query, "num_results": body.num_results},
None,
# Everything failed. Be honest about why instead of an empty "no results".
connected = await _refresh_9r_connected()
has_subscription = bool(connected & {"codex", "antigravity", "gemini-cli"})
if not (gemini_key or openai_key or has_subscription):
tail = (
"No search backend is configured and DuckDuckGo is rate-limiting this "
"network. Connect Codex / Antigravity / Gemini CLI in Settings, or add "
"an OpenAI / Gemini API key, for reliable search."
)
else:
tail = (
"DuckDuckGo is rate-limiting this network and every configured provider "
"errored (see details below). Wait a moment and retry."
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Search failed: {e}")
text = _join_text(parts)
hint = ""
if text.startswith("No search results found"):
connected = await _refresh_9r_connected()
has_subscription = bool(connected & {"codex", "antigravity", "gemini-cli"})
if not (gemini_key or openai_key or has_subscription):
hint = (
"\n\n(DuckDuckGo returned no results; likely rate-limiting this IP. "
"Connect Codex / Antigravity / Gemini CLI in Settings, or add an "
"OpenAI / Gemini API key, for reliable native search.)"
)
else:
hint = (
"\n\n(DuckDuckGo returned no results and the connected providers "
"didn't return useful results either; try rephrasing the query.)"
)
return {
"query": body.query,
"results": text + hint,
"backend": "ddg",
**({"cascade_errors": errors} if errors else {}),
"results": f"No results for: {body.query}\n\n{tail}",
"backend": "none",
"cascade_errors": errors,
}
@@ -571,28 +585,58 @@ async def fetch(body: FetchBody) -> dict:
"backend": "openai_subscription",
}
if primary == "openai":
cascade = [try_openai, try_openai_subscription, try_gemini, try_gemini_subscription]
else:
cascade = [try_gemini, try_gemini_subscription, try_openai, try_openai_subscription]
# Remembered so a thin/errored local read is still returned as the last
# resort if every grounded fetcher also fails (never worse than before).
local_text: str | None = None
for fn in cascade:
try:
res = await fn()
if res is not None:
return res
except Exception:
continue
# Local httpx + trafilatura fallback.
from backend.apps.agents.tools.web import WebFetchTool
try:
tool = WebFetchTool()
parts = await tool.execute(
{"url": body.url, "prompt": body.prompt or ""},
None,
async def try_local():
# Fast path: direct httpx + trafilatura, sub-second to a few seconds and
# returns the page's ACTUAL text (the grounded fetchers summarize, which
# is slower and loses detail). Thin/errored reads (JS walls, paywalls,
# HTTP errors) fall through to the grounded fetchers that can render them.
nonlocal local_text
from backend.apps.agents.tools.web import WebFetchTool
parts = await WebFetchTool().execute(
{"url": body.url, "prompt": body.prompt or ""}, None,
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Fetch failed: {e}")
text = _join_text(parts)
local_text = text
if text.startswith(("HTTP error", "Error fetching", "Refused to fetch")):
return None
body_text = text.split("\n\n", 1)[-1] if "\n\n" in text else text
if len(body_text.strip()) < 200:
return None
return {"url": body.url, "content": text, "backend": "local"}
return {"url": body.url, "content": _join_text(parts), "backend": "local"}
grounded = [
("gemini_native", try_gemini),
("gemini_subscription", try_gemini_subscription),
("openai_native", try_openai),
("openai_subscription", try_openai_subscription),
]
if primary == "openai":
grounded = grounded[2:] + grounded[:2]
cascade = [("local", try_local, _LOCAL_FETCH_TIMEOUT)] + [
(name, fn, _GROUNDED_ATTEMPT_TIMEOUT) for name, fn in grounded
]
errors: list[str] = []
for name, fn, timeout in cascade:
try:
res = await asyncio.wait_for(fn(), timeout=timeout)
if res is not None:
if errors:
res["cascade_errors"] = errors
return res
except asyncio.TimeoutError:
errors.append(f"{name}: timed out after {timeout:.0f}s")
except Exception as e:
errors.append(f"{name}: {str(e)[:150]}")
# Grounded all failed; hand back whatever the local read got (even an error
# string is useful signal) rather than nothing.
if local_text is not None:
return {"url": body.url, "content": local_text, "backend": "local",
**({"cascade_errors": errors} if errors else {})}
raise HTTPException(status_code=502, detail=f"Fetch failed for {body.url}: " + "; ".join(errors)[:400])
backend/tests/test_web_search_cascade.py
+188
View File
@@ -0,0 +1,188 @@
"""Fast-first, bounded web-search cascade (/api/web/search).
Pins the behaviour that fixes the ~75s stall and the human-speed goal:
- DuckDuckGo is tried FIRST and short-circuits the chain when it has results.
- When DDG is throttled, the chain falls over to the grounded backends.
- Every attempt is wait_for-bounded, so a hung provider can't stall the request.
- The `primary` hint reorders only the grounded tier.
- When everything fails we return an honest message, not a bogus empty result.
All providers are mocked, so the test is deterministic and offline.
"""
import asyncio
import time
import pytest
import backend.apps.web.web as W
from backend.apps.web.web import search, SearchBody
from backend.apps.agents.tools.web import WebSearchTool, DDGRateLimited
@pytest.fixture(autouse=True)
def _no_network(monkeypatch):
# Default everything to "unavailable / no network"; each test opts paths in.
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: None)
monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: None)
async def _no_subs():
return set()
monkeypatch.setattr(W, "_refresh_9r_connected", _no_subs)
async def _empty(*a, **k):
return {}
# subscription helpers hit localhost:20128 otherwise
monkeypatch.setattr(W, "_gemini_grounded_via_9router", _empty)
monkeypatch.setattr(W, "_openai_websearch_via_9router", _empty)
def _ddg_returns(monkeypatch, text):
async def _f(query, num):
return text
monkeypatch.setattr(WebSearchTool, "_search_ddg", staticmethod(_f))
def _ddg_throttled(monkeypatch):
async def _f(query, num):
raise DDGRateLimited(query)
monkeypatch.setattr(WebSearchTool, "_search_ddg", staticmethod(_f))
@pytest.mark.asyncio
async def test_ddg_is_tried_first_and_wins(monkeypatch):
_ddg_returns(monkeypatch, "[1] Foo\n https://foo.example")
# grounded would raise if reached; prove it isn't
async def _boom(*a, **k):
raise AssertionError("grounded should not be called when DDG has results")
monkeypatch.setattr(W, "_gemini_grounded_call", _boom)
t = time.monotonic()
res = await search(SearchBody(query="foo"))
assert res["backend"] == "ddg"
assert "foo.example" in res["results"]
assert "cascade_errors" not in res
assert time.monotonic() - t < 1.0
@pytest.mark.asyncio
async def test_ddg_throttled_falls_over_to_openai(monkeypatch):
_ddg_throttled(monkeypatch)
monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey")
async def _openai(api_key, query):
return {"text": "grounded answer", "chunks": [("Title", "https://u.example")]}
monkeypatch.setattr(W, "_openai_websearch", _openai)
res = await search(SearchBody(query="x"))
assert res["backend"] == "openai_native"
assert "u.example" in res["results"]
# DDG's throttle is recorded so the caller knows why we fell through
assert any("ddg" in e for e in res.get("cascade_errors", []))
@pytest.mark.asyncio
async def test_a_hung_grounded_attempt_is_bounded(monkeypatch):
_ddg_throttled(monkeypatch)
monkeypatch.setattr(W, "_GROUNDED_ATTEMPT_TIMEOUT", 0.3)
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey")
async def _hangs(*a, **k):
await asyncio.sleep(30)
monkeypatch.setattr(W, "_gemini_grounded_call", _hangs)
t = time.monotonic()
res = await search(SearchBody(query="x"))
elapsed = time.monotonic() - t
assert elapsed < 2.0, f"hung provider should be bounded, took {elapsed:.2f}s"
assert res["backend"] == "none"
assert any("timed out" in e for e in res["cascade_errors"])
@pytest.mark.asyncio
async def test_primary_openai_reorders_grounded_tier(monkeypatch):
_ddg_throttled(monkeypatch)
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey")
monkeypatch.setattr(W, "_resolve_openai_api_key", lambda: "okey")
async def _gem(*a, **k):
return {"text": "GEM", "chunks": [("g", "https://gem.example")]}
async def _oai(api_key, query):
return {"text": "OAI", "chunks": [("o", "https://oai.example")]}
monkeypatch.setattr(W, "_gemini_grounded_call", _gem)
monkeypatch.setattr(W, "_openai_websearch", _oai)
res = await search(SearchBody(query="x", primary="openai"))
# openai must be tried before gemini when it's the primary
assert res["backend"] == "openai_native"
assert "oai.example" in res["results"]
@pytest.mark.asyncio
async def test_everything_fails_is_honest_not_empty(monkeypatch):
_ddg_throttled(monkeypatch) # no keys, no subs (from fixture)
res = await search(SearchBody(query="obscure thing"))
assert res["backend"] == "none"
assert "obscure thing" in res["results"]
# points the user at how to get reliable search
assert "Settings" in res["results"] or "API key" in res["results"]
# --------------------------------------------------------------------------
# /fetch mirrors /search: local httpx + trafilatura is the fast path, grounded
# fetchers are the fallback for JS/paywalled pages, every attempt is bounded.
# --------------------------------------------------------------------------
from backend.apps.web.web import fetch, FetchBody
from backend.apps.agents.tools.web import WebFetchTool
import backend.apps.agents.tools.ssrf_guard as _ssrf
@pytest.fixture(autouse=True)
def _allow_urls(monkeypatch):
async def _ok(url):
return None
monkeypatch.setattr(_ssrf, "assert_safe_url", _ok)
def _local_returns(monkeypatch, text):
async def _exec(self, input_data, context):
return [{"type": "text", "text": text}]
monkeypatch.setattr(WebFetchTool, "execute", _exec)
@pytest.mark.asyncio
async def test_fetch_local_first_wins_and_is_fast(monkeypatch):
big = "Contents of https://x.example:\n\n" + ("real article body " * 50)
_local_returns(monkeypatch, big)
async def _boom(*a, **k):
raise AssertionError("grounded fetch should not run when local has content")
monkeypatch.setattr(W, "_gemini_grounded_call", _boom)
t = time.monotonic()
res = await fetch(FetchBody(url="https://x.example"))
assert res["backend"] == "local"
assert "real article body" in res["content"]
assert time.monotonic() - t < 1.0
@pytest.mark.asyncio
async def test_fetch_thin_local_falls_to_grounded(monkeypatch):
_local_returns(monkeypatch, "Contents of https://spa.example:\n\n") # JS wall, empty body
monkeypatch.setattr(W, "_resolve_gemini_api_key", lambda: "gkey")
async def _gem(api_key, prompt, *, use_url_context):
return {"text": "rendered page text from grounding", "chunks": []}
monkeypatch.setattr(W, "_gemini_grounded_call", _gem)
res = await fetch(FetchBody(url="https://spa.example"))
assert res["backend"] == "gemini_native"
assert "rendered page text" in res["content"]
@pytest.mark.asyncio
async def test_fetch_local_error_returned_as_last_resort(monkeypatch):
_local_returns(monkeypatch, "HTTP error 403 fetching https://blocked.example")
# no grounded keys/subs (autouse fixtures) -> all grounded skip/fail
res = await fetch(FetchBody(url="https://blocked.example"))
assert res["backend"] == "local"
assert "HTTP error 403" in res["content"]