Parallel retrieval

Ask several retrievers at once, merge what they return, and stop early when one answer is enough. Every call still goes through the executor's guards, caps, timeouts and ledger.

flowchart LR
  Q["query"] --> K["keyword_search"]
  Q --> V["vector_search"]
  Q --> A["archive_search (slow)"]
  K --> M["merge<br/>dedupe · rank"]
  V --> M
  A -. "cancelled when you stop early" .-> M
  M --> ANS["answer"]

Pick a method

Need	Method	Returns
Fan out, wait for all, merge	run_many(calls, ctx, concurrency=, on_error=)	ToolResults in input order
Take results as they finish, stop early	as_completed(calls, ctx, concurrency=)	async iterator in completion order
Steps that depend on other steps	run_plan(steps, ctx)	{step.id: ToolResult}
Non-tool work (LLM, embeddings)	parallel(*work, timeout=, retries=)	values in input order

The retrievers

@registry.tool(kind="search", parallel_safe=True, timeout=5, retries=1)
async def keyword_search(query: str) -> list[str]:
    """Keyword search over the help-center articles."""
    await asyncio.sleep(0.01)
    return INDEX.get(query, [])


@registry.tool(kind="search", parallel_safe=True, timeout=5)
async def vector_search(query: str) -> list[str]:
    """Semantic search over the help-center articles."""
    await asyncio.sleep(0.02)
    return [f"{query} policy", f"{query} similar"]


@registry.tool(parallel_safe=True)
def rerank(hits: list[str], top_k: int = 2) -> list[str]:
    """Order hits by relevance and keep the best."""
    return sorted(set(hits), key=len)[:top_k]


@registry.tool(kind="search", parallel_safe=True)
async def archive_search(query: str) -> list[str]:
    """Slow search over the archive."""
    await asyncio.sleep(30)  # cancelled when the caller stops early
    return []

Mark retrievers parallel_safe=True. A tool without it takes the run's exclusive lock, so the batch runs it alone.

# At most two concurrent "search" calls per run, whatever the batch size.
executor = Executor(registry, limits=RunLimits(per_kind={"search": 2}))

RunLimits(per_kind={"search": 2}) caps concurrent search calls per run, whatever concurrency a batch asks for. Use it to protect a rate-limited backend. For a rate (calls per second) add a RateLimitHook (Hooks).

Fan out, fan in

calls = [
    ToolCall(name="keyword_search", arguments={"query": query}),
    ToolCall(name="vector_search", arguments={"query": query}),
    ToolCall(name="keyword_search", arguments={"query": 7}),  # invalid: not a string
]
results = await executor.run_many(calls, concurrency=8, on_error="collect")
print([r.status for r in results])  # input order; one failure does not stop the rest

merged: dict[str, None] = {}  # fan in: dedupe, keep first-seen order
for result in results:
    if result.ok:
        merged.update(dict.fromkeys(result.value))
return list(merged)

on_error	A failing call
"collect" (default)	stays in its ToolResult (status="error"); the rest finish
"fail_fast"	cancels the siblings; the first error is raised as itself

No batch raises an ExceptionGroup. A batch may not repeat a non-empty call id (ToolUsageError).

Stop at the first good answer

calls = [
    ToolCall(name="archive_search", arguments={"query": query}),
    ToolCall(name="keyword_search", arguments={"query": query}),
]
stream = executor.as_completed(calls)
async with contextlib.aclosing(stream):
    async for result in stream:
        if result.ok and result.value:
            return f"{result.name}: {result.value[0]}"
return "nothing found"

Wrap the stream in contextlib.aclosing. Leaving the block cancels the calls still running: here the 30-second archive_search stops at once. Without aclosing, a break or return leaves them running until the iterator is closed.

Retrieve, then rank

plan = [
    Step(id="kw", tool="keyword_search", args={"query": query}),
    Step(id="vec", tool="vector_search", args={"query": query}),
    Step(
        id="best",
        tool="rerank",  # second wave: runs once both retrievers finished
        args=lambda deps: {"hits": deps["kw"].value + deps["vec"].value},
        depends_on=["kw", "vec"],
    ),
]
by_step = await executor.run_plan(plan)
return list(by_step["best"].value)

The plan runs in waves: kw and vec together, then best. A callable args gets the ToolResults of its own dependencies. A step whose dependency failed does not run: its error kind is dependency-failed. A bad plan (duplicate ids, unknown dependencies, a cycle) raises one PlanInvalid before anything runs.

Non-tool work

async def embed(text: str) -> int:  # stands in for an embedding or LLM call
    await asyncio.sleep(0.01)
    return len(text)

return await executor.parallel(*(lambda t=t: embed(t) for t in texts), timeout=10)

Pass coroutine functions (zero-argument callables) so retries= can start them again; a bare awaitable cannot be retried. Bind loop variables (lambda t=t: ...). A timeout is ToolTimeout(tool="parallel[i]").

Output

['ok', 'ok', 'error']
['returns policy', 'returns label', 'returns similar']
keyword_search: returns policy
['returns label', 'returns policy']
[14, 13]

Tips

Tip	Why
Pass one ctx to related batches	guards (max_calls, once_per_run) and caps count per run; with no ctx each batch is a fresh run
Give each retriever a timeout	one slow backend cannot hold the merge forever
Cache retriever results	cache_ttl= plus an executor cache (Caching)
Cap only leaf kinds in per_kind	nesting capped kinds both ways can deadlock (Executor policies)

Files

File	Does
examples/howto_parallel.py	the retrievers and the four patterns
tests/docs/test_howto_examples.py	runs it offline and checks the output