pipeline: Granite web-search gate, racing the hedge

A tiny model (ibm-granite/granite-4.1-8b) answers "does this need a web
search?" at t=0, OR'd into the existing regex and bidder signals.

Motivation is recall, not speed. The regex scores 100% on its own test
file but 62.5% on held-out phrasing -- it misses ordinary questions like
"did the fed cut rates" and "what did apple announce". The gate model
scored 100% on the same held-out set (evals/bench_web_gate.py).

It runs in PARALLEL with the hedge, not before it. The gate's TTFT
(~450-500ms measured) is about the same as the hedge model's, so putting
it on the critical path would roughly double time-to-first-token on the
majority of queries, which need no search at all. Racing instead, a
"yes" cancels the hedge ~500ms in rather than at auction time (2-4s),
so the user sees far less doomed text before the reset.

Fails open: an outage or unparseable reply returns None, which callers
treat as "no opinion", so the regex and bidder flags still decide. Cost
is ~$0.000007/query, tracked as a new "gate" usage stage.

Verified live: all three regex misses now search, non-web queries are
untouched. A/B on time-to-first-token for non-web queries was
inconclusive -- variance (469-4040ms) dwarfs any gate effect at n=6, so
"no measurable regression" is as strong a claim as the data supports.

Revert with: git checkout main

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/config.py

@@ -101,6 +101,17 @@ VERIFIER_MODEL = ModelSpec(
     cost_per_mtok_out=0.30,
 )
 
+def _web_gate_model() -> "ModelSpec":
+    """Built after Settings so the model id stays overridable by env."""
+    return ModelSpec(
+        key="webgate",
+        openrouter_id=settings.web_gate_model_id,
+        display_name="Web gate",
+        cost_per_mtok_in=0.05,
+        cost_per_mtok_out=0.10,
+    )
+
+
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_file=".env", extra="ignore")
 
@@ -120,6 +131,17 @@ class Settings(BaseSettings):
     # models' knowledge cutoff. Bump this as the underlying models advance.
     model_knowledge_cutoff_year: int = 2024
 
+    # t=0 web-search gate: a tiny model answering one yes/no question, run in
+    # parallel with the hedge (see webgate.py for why parallel, not first).
+    # Granite 4.1 8B won on both axes in evals/bench_web_gate.py: 100% on
+    # held-out queries at ~450-500ms TTFT. Smaller models were not faster —
+    # provider routing dominates at this size — and 1B scored 40%.
+    web_gate_enabled: bool = True
+    web_gate_model_id: str = "ibm-granite/granite-4.1-8b"
+    # Generous vs the ~500ms typical: the gate races the bids either way, so a
+    # slow tail costs nothing, while a tight timeout would silently drop it.
+    web_gate_timeout_s: float = 3.0
+
     # Image search (Tavily). The OpenRouter web plugin returns no images, so
     # image-worthy web queries get one extra Tavily call of our own. An empty
     # key disables the feature silently — nothing else changes.
@@ -235,6 +257,8 @@ class Settings(BaseSettings):
 
 settings = Settings()
 
+WEB_GATE_MODEL = _web_gate_model()
+
 
 # --- Tier 2: frontier escalation target -------------------------------------
 # (display name, $/Mtok in, $/Mtok out) per known frontier choice; unknown

backend/app/pipeline.py

@@ -24,6 +24,7 @@ from .config import (BASELINE_MODEL, SPECULATIVE_HINT_MODELS, TIER1_MODELS,
 from .llm import LLMError, chat, extract_json
 from .schemas import Bid, RunResult, SourceImage, Usage, Verification
 from .store import get_store
+from . import webgate
 from .websearch import enabled as image_search_enabled
 from .websearch import search_images
 
@@ -36,6 +37,7 @@ class RouterState(TypedDict, total=False):
     winner: Optional[str]
     needs_web: bool
     web_via_bidder: bool
+    gate_web: bool
     wants_images: bool
     images: list[SourceImage]
     draft_answer: Optional[str]
@@ -470,12 +472,16 @@ async def auction(state: RouterState) -> RouterState:
     # answer was written from stale training data, so discard it and force
     # a real (web-enabled) draft.
     regex_web = _needs_web_heuristic(state["query"])
-    needs_web = winner.needs_web or regex_web
+    # The gate model has already spoken by now (it resolves ~500ms in, well
+    # before the bids) and has far better recall than the regex on natural
+    # phrasing, so it joins the OR rather than replacing anything.
+    gate_web = bool(state.get("gate_web"))
+    needs_web = winner.needs_web or regex_web or gate_web
     out["needs_web"] = needs_web
-    # True when the upfront regex missed but a bidder flagged the query — the
-    # UI announces "let me search the web" in that case (regex hits go straight
-    # to the search silently, since we knew from the start).
-    out["web_via_bidder"] = needs_web and not regex_web
+    # True when neither the upfront regex nor the gate caught it but a bidder
+    # did — only then does the UI still owe the user a "let me search" line.
+    # Regex hits search silently; the gate already announced its own.
+    out["web_via_bidder"] = needs_web and not regex_web and not gate_web
     # Images ride along with a web search and never on their own: a picture
     # for a query we answered from training data would be unsourced, and
     # gating on needs_web caps image lookups at one per web-searched query.
@@ -853,6 +859,14 @@ async def run_query_stream(query: str, history: list[dict] | None = None,
     if not run_hedge:
         state["needs_web"] = True
 
+    # The gate model races the hedge rather than preceding it: its TTFT is
+    # about the same as the hedge model's, so making the hedge wait would
+    # double time-to-first-token on every query that needs no search at all.
+    # Skipped when the regex already decided — that answer costs nothing and
+    # is never wrong in the "web needed" direction.
+    gate_task = (asyncio.ensure_future(webgate.decide(query))
+                 if run_hedge and webgate.enabled() else None)
+
     hedge_q: asyncio.Queue = asyncio.Queue()
 
     async def _hedge_stream():
@@ -876,7 +890,26 @@ async def run_query_stream(query: str, history: list[dict] | None = None,
     # Forward hedge tokens the moment they exist, while bids come in
     hedge_final: Optional[object] = None
     hedge_done = not run_hedge
+    gate_says_web = False
     while not bid_task.done():
+        # Gate verdict usually lands ~500ms in, long before the bids. A "yes"
+        # kills the hedge here rather than at auction time, so the user sees a
+        # few hundred ms of doomed text instead of several seconds of it.
+        if gate_task is not None and gate_task.done():
+            verdict, gate_usage = gate_task.result()
+            gate_task = None
+            if gate_usage is not None:
+                state["usages"] = state["usages"] + [gate_usage]
+            if verdict is True:
+                gate_says_web = True
+                state["needs_web"] = True
+                if hedge_task is not None:
+                    hedge_task.cancel()
+                    hedge_task = None
+                run_hedge, hedge_done = False, True
+                yield {"type": "reset"}
+                yield {"type": "token",
+                       "text": "🔎 Let me search the web for you…\n\n"}
         if not run_hedge:
             await asyncio.sleep(0.05)
             continue
@@ -888,6 +921,18 @@ async def run_query_stream(query: str, history: list[dict] | None = None,
             yield {"type": "token", "text": val}
         else:
             hedge_final, hedge_done = val, True
+    # Bids won the race; collect the gate's verdict if it still has one
+    if gate_task is not None:
+        try:
+            verdict, gate_usage = await asyncio.wait_for(gate_task, timeout=0.5)
+            if gate_usage is not None:
+                state["usages"] = state["usages"] + [gate_usage]
+            if verdict is True:
+                gate_says_web = True
+                state["needs_web"] = True
+        except Exception:
+            pass  # no opinion; the regex and bidder flags still decide
+    state["gate_web"] = gate_says_web
     state.update(bid_task.result())
     state.update(await auction(state))
     # Fired here so Tavily runs concurrently with the draft — by the time the

backend/app/schemas.py

@@ -50,7 +50,7 @@ class Verification(BaseModel):
 class Usage(BaseModel):
     model_key: str
     model_name: str
-    stage: Literal["bid", "draft", "verify", "escalate"]
+    stage: Literal["bid", "draft", "verify", "escalate", "gate"]
     tokens_in: int = 0
     tokens_out: int = 0
     cost_usd: float = 0.0

backend/app/webgate.py

@@ -0,0 +1,75 @@
+"""A tiny model that answers one question: does this query need a web search?
+
+Runs at t=0 alongside the hedge and the bids, not before them. That ordering is
+load-bearing: the gate's TTFT (~450-500ms measured) is roughly the same as the
+hedge model's, so putting it on the critical path would double time-to-first-
+token on the majority of queries, which need no web search at all. In parallel
+it costs nothing and simply cancels the hedge far earlier than the bidders
+would have (~500ms instead of 2-4s).
+
+It exists because the regex it supplements has poor recall on natural
+phrasing — 62.5% on held-out queries against this model's 100%. See
+evals/bench_web_gate.py for the comparison and how the model was chosen.
+"""
+
+import asyncio
+import logging
+
+from .config import WEB_GATE_MODEL, settings
+from .llm import LLMError, chat
+from .schemas import Usage
+
+log = logging.getLogger(__name__)
+
+# Every token here is on a latency-critical path, so the prompt stays short.
+# Identical to the one the benchmark scored, so measured accuracy carries over.
+GATE_SYSTEM = (
+    "You decide if a user query needs a live web search to answer correctly.\n"
+    "Answer YES if it needs current, real-time, or post-training information: "
+    "news, prices, scores, weather, releases, results, who currently holds a "
+    "role, or a specific named item you may not know.\n"
+    "Answer NO for timeless questions: math, code, definitions, writing, "
+    "advice, explanations, general knowledge.\n"
+    "Reply with exactly one word: YES or NO."
+)
+
+
+def enabled() -> bool:
+    return bool(settings.web_gate_enabled and settings.openrouter_api_key)
+
+
+async def decide(query: str) -> tuple[bool | None, Usage | None]:
+    """Return (needs_web, usage). None means "no opinion" — never a guess.
+
+    Callers must treat None as "fall back to the other signals" rather than
+    as False, so a gate outage can only lose recall, never invent it.
+    """
+    if not enabled() or not query.strip():
+        return None, None
+    try:
+        resp = await asyncio.wait_for(
+            chat(WEB_GATE_MODEL, GATE_SYSTEM, query,
+                 max_tokens=8, prefer_paid=True),
+            timeout=settings.web_gate_timeout_s,
+        )
+    except (LLMError, asyncio.TimeoutError, Exception) as e:
+        log.warning("web gate failed: %s: %s", type(e).__name__, str(e)[:150])
+        return None, None
+
+    usage = Usage(
+        model_key=WEB_GATE_MODEL.key,
+        model_name=WEB_GATE_MODEL.display_name,
+        stage="gate",
+        tokens_in=resp.tokens_in,
+        tokens_out=resp.tokens_out,
+        cost_usd=WEB_GATE_MODEL.estimate_cost(resp.tokens_in, resp.tokens_out,
+                                              resp.served_model),
+        latency_ms=resp.latency_ms,
+    )
+    answer = resp.content.strip().upper()
+    if answer.startswith("YES"):
+        return True, usage
+    if answer.startswith("NO"):
+        return False, usage
+    log.warning("web gate returned unparseable %r", resp.content[:60])
+    return None, usage