Parallel retrieval with a deadline

Ask three retrievers at once. One backend hangs. Answer with what arrived in time.

sequenceDiagram
  participant App
  participant Ex as Executor
  App->>Ex: as_completed(archive, forum, articles)
  Ex-->>App: articles (10 ms)
  Ex-->>App: forum (20 ms)
  Note over App: the 150 ms deadline passes
  App->>Ex: close the stream
  Ex->>Ex: cancel archive (still waiting)

The retrievers

@registry.tool(kind="search", parallel_safe=True, timeout=0.5)
async def articles(query: str) -> list[str]:
    """Search the help-center articles."""
    await asyncio.sleep(0.01)
    return [f"{query}: article"]


@registry.tool(kind="search", parallel_safe=True, timeout=0.5)
async def forum(query: str) -> list[str]:
    """Search community forum posts."""
    await asyncio.sleep(0.02)
    return [f"{query}: post"]


@registry.tool(kind="search", parallel_safe=True, timeout=0.3)  # this backend hangs
async def archive(query: str) -> list[str]:
    """Search the old archive."""
    await asyncio.sleep(30)
    return []

timeout= bounds each call. parallel_safe=True lets the calls run at the same time.

Wait for all, each with its own timeout

results = await executor.run_many(calls(query, "articles", "forum", "archive"))
return results  # input order; the hung archive is a tool-timeout, the rest are ok

[('articles', 'ok', ''), ('forum', 'ok', ''), ('archive', 'error', 'tool-timeout')]

The batch takes as long as the slowest timeout (here 0.3 s). No call raises.

Stop at a deadline for the whole batch

hits: list[str] = []
stream = executor.as_completed(calls(query, "archive", "forum", "articles"))
with contextlib.suppress(TimeoutError):  # keep what arrived in time
    async with asyncio.timeout(seconds), contextlib.aclosing(stream):
        async for result in stream:  # completion order
            if result.ok:
                hits += result.value
return hits  # leaving the block cancelled whatever was still running

['returns: article', 'returns: post']

asyncio.timeout ends the loop; aclosing then cancels the calls still running.

Need	Use
every result, each call bounded	run_many and a tool timeout
an answer by a time, partial is fine	as_completed inside asyncio.timeout
the first good answer	as_completed, then return (Parallel retrieval)
non-tool work (embeddings, LLM calls)	executor.parallel(*work, timeout=)

Pitfalls

Pitfall	Do instead
as_completed without aclosing	a break leaves calls running until the iterator is closed
A deadline shorter than every source	you get nothing: keep one fast source
A rate-limited backend	cap it with RunLimits(per_kind={"search": 2})

Files

File	Does
examples/cookbook_parallel_timeout.py	three retrievers, wait-for-all, a batch deadline
tests/docs/test_cookbook.py	runs it offline and checks the output