A provider-neutral web-search component for Haystack with bounded retries

[auxiliar-ag]

A provider-neutral web-search component for Haystack with bounded retries

A Haystack component's run() method hands its return value straight to whatever is connected downstream in a Pipeline. If a web-search step treats any 200 response as a finished job, a ChatPromptBuilder or a retriever further down the graph ends up working from an empty or malformed result set as if it had real search results to reason over. The failure stays invisible until someone inspects what actually flowed through the connection.

A component with a retry budget, not a retry loop

Below is BoundedRetryWebSearch, a class decorated with @component that takes an ordered list of injected provider clients and a per-provider attempt limit. It never opens a live HTTP connection itself. Each provider client is a small object with a search(query) method that either returns a ProviderResponse(status_code, payload) or raises TransportError, so the retry and fallback logic runs the same way under test as it would in production, against a FakeProviderClient that plays back a scripted list of responses instead of a real request. A retry is spent only when the call itself did not complete: a caught TransportError, or a non-2xx status. A 2xx response that turns out to carry an empty results list, or a body that is not shaped like the expected contract, stops the retry loop for that provider immediately, because asking the same provider the same query again will not turn an unusable result into a useful one. That case, and only that case, moves straight to the next provider in the list. This is the exact source under test, not a shortened version written for this post:

from collections.abc import Sequence
from dataclasses import asdict, dataclass

from haystack import component

from web_search.contract import Verdict, evaluate_payload
from web_search.transport import SearchProviderClient, TransportError


@dataclass(frozen=True)
class SearchReceipt:
    """
    A bounded, telemetry-safe summary of one search call.

    Deliberately excludes the query text, credentials, raw response bodies, and raw
    result URLs, so it is safe to log or export even when the search itself touched
    sensitive input. ``provider_used`` is a positional label (``provider_0``,
    ``provider_1``, ...), not the provider's name or domain.
    """

    outcome: str
    task_success: bool
    transport_success: bool
    provider_used: str | None
    fallback_used: bool
    attempts_total: int
    result_count: int

    def to_dict(self) -> dict[str, object]:
        return asdict(self)


def _attempt_provider(
    provider: SearchProviderClient, query: str, max_attempts: int
) -> tuple[bool, Verdict | None, int]:
    """
    Call one provider up to ``max_attempts`` times, retrying only on transport
    failure. A transport-successful response (2xx) always stops the retry loop for
    this provider, whether or not the content inside it turns out to be useful,
    since retrying the same provider with the same query will not turn an empty or
    malformed result into a useful one.
    """
    for attempt in range(1, max_attempts + 1):
        try:
            response = provider.search(query)
        except TransportError:
            continue
        if not (200 <= response.status_code < 300):
            continue
        return True, evaluate_payload(response.payload), attempt
    return False, None, max_attempts


@component
class BoundedRetryWebSearch:
    """
    A provider-neutral web-search component with bounded retries and fallback.

    Takes an ordered list of injected provider clients (each implementing
    ``SearchProviderClient.search(query) -> ProviderResponse``, raising
    ``TransportError`` on transport failure). For each provider, retries up to
    ``max_attempts_per_provider`` times on transport failure only; a
    transport-successful call that turns out to be task-invalid (empty results or a
    malformed contract) moves straight to the next provider instead of retrying.
    Returns the validated results plus a bounded receipt safe for telemetry.
    """

    def __init__(self, providers: Sequence[SearchProviderClient], max_attempts_per_provider: int = 2) -> None:
        if max_attempts_per_provider < 1:
            raise ValueError("max_attempts_per_provider must be at least 1")
        self._providers = list(providers)
        self._max_attempts_per_provider = max_attempts_per_provider

    @component.output_types(results=list, receipt=dict)
    def run(self, query: str) -> dict[str, object]:
        if not self._providers:
            receipt = SearchReceipt(
                outcome="no_provider",
                task_success=False,
                transport_success=False,
                provider_used=None,
                fallback_used=False,
                attempts_total=0,
                result_count=0,
            )
            return {"results": [], "receipt": receipt.to_dict()}

        attempts_total = 0
        last_transport_success = False
        last_outcome = "transport_error"

        for provider_index, provider in enumerate(self._providers):
            transport_success, verdict, attempts_used = _attempt_provider(
                provider, query, self._max_attempts_per_provider
            )
            attempts_total += attempts_used
            last_transport_success = transport_success
            fallback_used = provider_index > 0

            if transport_success and verdict is not None:
                last_outcome = verdict.outcome
                if verdict.task_success:
                    receipt = SearchReceipt(
                        outcome="ok",
                        task_success=True,
                        transport_success=True,
                        provider_used=f"provider_{provider_index}",
                        fallback_used=fallback_used,
                        attempts_total=attempts_total,
                        result_count=len(verdict.results),
                    )
                    results = [{"title": r.title, "url": r.url} for r in verdict.results]
                    return {"results": results, "receipt": receipt.to_dict()}
            else:
                last_outcome = "transport_error"

        receipt = SearchReceipt(
            outcome=last_outcome,
            task_success=False,
            transport_success=last_transport_success,
            provider_used=f"provider_{len(self._providers) - 1}",
            fallback_used=len(self._providers) > 1,
            attempts_total=attempts_total,
            result_count=0,
        )
        return {"results": [], "receipt": receipt.to_dict()}

Wiring it into a Pipeline

run() returns two keys. results is a list of plain {"title", "url"} dicts for whatever component consumes them next. receipt is a small dict meant for a log line or a metrics sink, not for the pipeline graph itself. The two are separate outputs on purpose: results is operational data a downstream component needs to do its job, so it legitimately carries the URL forward, while receipt never does. receipt has a fixed set of seven keys (outcome, task_success, transport_success, provider_used, fallback_used, attempts_total, result_count), and provider_used is a positional label like provider_1 rather than a provider's own name, so nothing in it can leak the query text, a credential, a raw response body, or a raw URL even if a caller logs the whole dict. The test below builds a two-component Pipeline, connects search.results to a small counting component's results input, and runs it end to end with two FakeProviderClient instances standing in for a primary and a fallback provider:

from haystack import Pipeline, component

from web_search.component import BoundedRetryWebSearch
from web_search.transport import FakeProviderClient, ProviderResponse


@component
class ResultCounter:
    """Minimal downstream component: counts how many results it received."""

    @component.output_types(count=int)
    def run(self, results: list) -> dict:
        return {"count": len(results)}


def test_bounded_retry_web_search_runs_inside_a_real_haystack_pipeline():
    primary = FakeProviderClient([ProviderResponse(200, {"results": []})])
    fallback = FakeProviderClient(
        [
            ProviderResponse(
                200,
                {
                    "results": [
                        {"title": "A", "url": "https://example.com/a"},
                        {"title": "B", "url": "https://example.com/b"},
                    ]
                },
            )
        ]
    )

    pipeline = Pipeline()
    pipeline.add_component("search", BoundedRetryWebSearch(providers=[primary, fallback]))
    pipeline.add_component("counter", ResultCounter())
    pipeline.connect("search.results", "counter.results")

    output = pipeline.run({"search": {"query": "haystack pipeline integration"}})

    assert output["counter"]["count"] == 2
    # the receipt is not connected downstream, so it surfaces directly from the search component
    assert output["search"]["receipt"]["fallback_used"] is True

Why the retry budget only covers transport failures

Spending every retry on transport errors and none on a content failure is a deliberate choice, not an oversight: a TransportError or a 503 says nothing about the query, so trying again can plausibly help, while an empty results array from a 200 response is the provider's actual answer, and repeating it changes nothing. I drew that same split, a provider's error rate scored apart from whether its response was actually useful, from NativePort's public page on how it measures providers, which records a non-2xx response and an unsatisfying 2xx as two different numbers instead of folding both into one pass or fail count.

This is a sample component written against the current haystack-ai API, not a packaged NativePort-Haystack integration. NativePort has no involvement in the retry, fallback, or validation logic above; the link is to the page that shaped how I think about separating those two kinds of failure, nothing more.

Has anyone settled on a fixed attempt limit per provider before falling back, or does the right number vary enough across providers that you end up tuning it per deployment?

Disclosure: NativePort is where I work. This post was drafted with AI assistance; the component and the Pipeline test above ran exactly as shown before publication.