Fanning Out Search Calls Without Melting Your Quota

Sequential search in an agent loop is the main source of latency. Here's how to parallelize it with asyncio and still respect a rate limit.

Profile picture of Serply
Serply
Concurrent requests fanning out through a semaphore

A research agent that issues eight queries sequentially, at roughly a second each, spends eight seconds retrieving before it starts thinking. Run them concurrently and that’s about one second. On a multi-hop agent the difference compounds into the gap between “feels responsive” and “did it crash?”

The trap is that naive concurrency finds your rate limit immediately.

The async client

import asyncio
import os
from urllib.parse import quote_plus
import httpx

API_KEY = os.environ["SERPLY_API_KEY"]
BASE = "https://api.serply.io/v1"


class SearchClient:
    def __init__(self, concurrency: int = 5, timeout: float = 30.0):
        self._sem = asyncio.Semaphore(concurrency)
        self._client = httpx.AsyncClient(
            timeout=timeout,
            headers={"X-Api-Key": API_KEY},
            limits=httpx.Limits(max_connections=concurrency * 2),
        )
        self.remaining: int | None = None

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        await self._client.aclose()

    def _note(self, resp: httpx.Response) -> None:
        rem = resp.headers.get("x-ratelimit-requests-remaining")
        if rem is not None:
            self.remaining = int(rem)

    async def search(self, query: str, num: int = 10, region: str = "US") -> dict:
        async with self._sem:
            resp = await self._client.get(
                f"{BASE}/search/q={quote_plus(query)}&num={num}",
                headers={"X-Proxy-Location": region},
            )
            self._note(resp)
            if resp.status_code == 429:
                return {"results": [], "error": "rate_limited", "query": query}
            resp.raise_for_status()
            return resp.json()

The semaphore is the whole design. asyncio.gather over forty queries without one opens forty simultaneous connections, and you’ll see 429 on most of them. Five concurrent is a sane starting point.

Returning a dict on 429 rather than raising keeps gather from cancelling siblings — one rate-limited call shouldn’t discard the nine that succeeded.

Fan-out with partial failure

    async def search_many(self, queries: list[str], num: int = 10) -> list[dict]:
        tasks = [self.search(q, num=num) for q in queries]
        results = await asyncio.gather(*tasks, return_exceptions=True)

        out = []
        for query, result in zip(queries, results):
            if isinstance(result, Exception):
                out.append({"query": query, "results": [], "error": str(result)})
            else:
                out.append(result)
        return out

return_exceptions=True is essential. Without it, one timeout out of twenty discards nineteen good responses — and in an agent loop the model then re-issues all twenty.

Merging and deduplicating

Fan-out produces heavy overlap. Fuse by rank rather than trying to reconcile positions across queries:

from collections import defaultdict


def fuse(responses: list[dict], k: int = 60) -> list[dict]:
    """Reciprocal rank fusion across multiple result sets."""
    scores: dict[str, float] = defaultdict(float)
    seen: dict[str, dict] = {}

    for resp in responses:
        for r in resp.get("results", []):
            link = r.get("link")
            if not link:
                continue
            rank = r.get("position") or 1
            scores[link] += 1.0 / (k + rank)
            seen.setdefault(link, r)

    ranked = sorted(scores.items(), key=lambda kv: kv[1], reverse=True)
    return [seen[link] for link, _ in ranked]

Reciprocal rank fusion rewards results that show up across several queries, which is a decent proxy for relevance when the queries are variations on one question. A URL ranked fifth in three different searches outranks one ranked first in a single search — usually correct.

Reading pages concurrently too

The scraper calls are slower than the searches, so they benefit more:

    async def read(self, url: str) -> str | None:
        async with self._sem:
            try:
                resp = await self._client.post(
                    f"{BASE}/request",
                    json={"url": url, "response_type": "markdown"},
                    headers={"Content-Type": "application/json"},
                    timeout=60.0,
                )
                self._note(resp)
                if resp.status_code != 200:
                    return None
                return resp.text
            except httpx.RequestError:
                return None

    async def read_many(self, urls: list[str]) -> dict[str, str]:
        texts = await asyncio.gather(*(self.read(u) for u in urls))
        return {u: t for u, t in zip(urls, texts) if t}

Markdown mode returns the converted text as the body, so resp.text is right — resp.json() would raise. Dropping failures from the dict rather than storing None means downstream code doesn’t need a guard on every access.

Putting it together

async def research(question: str, variations: list[str]) -> dict:
    async with SearchClient(concurrency=5) as client:
        responses = await client.search_many(variations, num=10)
        ranked = fuse(responses)
        top_urls = [r["link"] for r in ranked[:5] if r.get("link")]
        pages = await client.read_many(top_urls)

        return {
            "question": question,
            "n_results": len(ranked),
            "pages": pages,
            "remaining_quota": client.remaining,
        }


asyncio.run(research(
    "How do SERP APIs handle geo-targeting?",
    [
        "SERP API geo targeting proxy location",
        "search API country specific results",
        "google search api regional results header",
    ],
))

Two round trips total — one for all searches, one for all reads — instead of eight sequential calls.

Adaptive throttling

If you run this continuously, let the remaining-quota signal feed back into concurrency:

    async def maybe_slow_down(self) -> None:
        if self.remaining is None:
            return
        if self.remaining < 50:
            await asyncio.sleep(1.0)
        elif self.remaining < 200:
            await asyncio.sleep(0.2)

Call it at the top of search. Crude, but it converts a hard wall into a gradual slowdown, which is almost always the behavior you want from a background job. Foreground, user-facing requests should skip the sleep and just fail fast with a message the model can act on.