agentsop-map-reduce-fanout

# Map-Reduce / Dynamic Fan-Out · SOP > Pattern: `results = reduce(combine, parallel_map(f, L))` where `f` is one > or more LM calls. The *only* reason to fan out is that latency or > throughput matters more than the cost of doing it. The *only* reason to > fan in is that the consumer wants one answer, not N. > Source posture: claims grounded in primary docs and 2026 production > write-ups, cited inline with short tags resolved in the citation index. --- ## 1. 何时激活 (Activation Rules) Activate this skill when **any** of these is true: - The task description contains "for each X, do Y" where Y involves an LM call, a retriever hit, or any I/O-bound step costing >100ms. - The coder is about to write a `for item in items: result = llm(item)` loop and the items are independent (no item depends on the previous result). - The codebase already has `asyncio.gather(...)`, `ThreadPoolExecutor(...)`, `Send(...)`, `Process.hierarchical` parallel branches, or `crew.kickoff_for_each(...)` and the question is *how* to use them safely. - The user mentions any of: "summarize N docs", "rank top-K candidates", "vote across M models", "ensemble", "parallel agents", "multi-query retrieval", "scatter-gather", "fan out". - A LangGraph graph is throwing `InvalidUpdateError` on a key two parallel branches write to (see OP-3 cross-link in skill `O5 state-reducer`). - A LangGraph `Send`-based fan-out is hitting `GRAPH_RECURSION_LIMIT` or rate-limit 429s because all N workers fired at once `[aipractitioner/scaling]`. Do **not** activate when: - N is statically 1 or 2 (just write the calls inline; setup cost ≥ win). - Items depend on each other (sequential reasoning, chain-of-thought across docs) — fan-out destroys the dependency. - The downstream code only needs the *first* successful answer — use `asyncio.wait(..., return_when=FIRST_COMPLETED)`, not gather. - The "fan-out" is into a single batched API call (e.g., embedding 100 strings in one OpenAI request). That's a batched single call, not map-reduce. Use it; it's cheaper. --- ## 2. 核心心智模型 (Core Mental Model) **Fan out for latency, fan in for coherence.** The whole protocol is two questions: *what runs concurrently?* and *how do the answers merge?* Everything else — `Send`, `gather`, `Semaphore`, reducers — is mechanics. Three load-bearing concepts: 1. **The unit of fan-out is the item, not the call.** `f(item)` may itself be a multi-step LM workflow (retrieve → rerank → synthesize) — that is fine. What you parallelise is the *per-item function*. Don't confuse "parallel LM calls" with "parallel workflow instances". The latter is what `Send` and `gather(f(x) for x in L)` actually do `[deepwiki/mapreduce]`. 2. **Concurrency is bounded, never infinite.** Every external dependency (OpenAI/Anthropic API, your vector DB, your KV cache) has at least one of: RPM limit, TPM limit, connection-pool limit, GPU KV-cache budget. `asyncio.gather(*[call(x) for x in 100_items])` will *not* call 100 things — it will fire 100, the API will 429 most, and you'll spend the next 20 minutes in retry-storm hell `[newline/asyncio-llm]` `[tianpan/structured-concurrency]`. The first thing you choose, before any code, is **N_concurrent**. 3. **The reducer determines the shape of the answer.** `concatenate` keeps all evidence (large output, no judgment); `summarize` collapses (information loss, smaller output); `vote / majority` picks one (lossy but decisive); `rank-top-K` selects the best few. Pick the reducer *before* you write the map — because the map's `expected_output` shape is dictated by the reduce. The Pregel-lineage version of the same point: **a fan-out / fan-in is one superstep**. Either it all succeeds and the reduce node runs, or one branch fails and (in LangGraph) the whole superstep is discarded `[aipractitioner/scaling]`. Your code must decide *before* fan-out which semantics you want: atomic-or-nothing, or best-effort-with-holes. --- ## 3. SOP 工作流 (Standard Operating Protocol) Walk this top-down. Each step has a decision gate. ### Step 1 · Confirm items are independent Gate: can `f(item_i)` run without seeing `f(item_j)` for any `j ≠ i`? If **no**, stop. You have a sequential or recursive problem masquerading as map-reduce. Use a chain, or model the dependency explicitly (DAG, beam search, etc.). Fan-out will silently drop the cross-talk. ### Step 2 · Estimate cost honestly Before any code, multiply: ``` cost = N · cost_per_item (in $, tokens, AND seconds_wall) peak_rps = N_concurrent / mean_latency_per_item ``` Three numbers must fit inside three budgets: - `cost_$` ≤ task budget - `peak_rps` ≤ min(API RPM/60, vector-DB QPS, GPU concurrency) - `peak_tps` ≤ API TPM / 60 — TPM is the silent killer: 50 parallel calls each with a 4k-token prompt instantly exceeds most providers' TPM even if RPM is fine `[newline/asyncio-llm]`. If any budget is tight, **fan-out is the wrong tool**. Options: batch calls into single API request (embeddings, reranking), reduce N (pre-filter items), or accept sequential. ### Step 3 · Pick N_concurrent (the most important number in the file) Default rubric: | Constraint | Pick | |---|---| | API-bound (OpenAI/Anthropic) | `min(10, RPM/60 · target_latency_s)` — keep at most one "request-second" of headroom | | Self-hosted vLLM/SGLang | Start at `num_kv_blocks / mean_prompt_blocks`; for most setups 8–32 | | Vector DB retrieval | Provider QPS limit / 2 (leave headroom for other paths) | | Mixed (LM + tool calls) | Constrain by the slowest dependency | | No instrumentation yet | **Start at 5**. Always. Then measure. | `N_concurrent` is enforced with `asyncio.Semaphore(N)` for asyncio, `ThreadPoolExecutor(max_workers=N)` for sync, `max_concurrency` config in LangGraph, `max_rpm` in CrewAI. Never rely on the framework's defaults. ### Step 4 · Decide the failure policy *before* writing the fan-out Four canonical policies — pick one explicitly: | Policy | When to use | How | |---|---|---| | **Abort-all** | One missing item invalidates the answer (legal review, regulated workflows) | `asyncio.gather(*, return_exceptions=False)` — first exception cancels siblings. In LangGraph this is the *default* superstep semantic `[aipractitioner/scaling]`. | | **Best-effort** | Concatenate / summarize use cases — partial is OK | `asyncio.gather(*, return_exceptions=True)` then filter; or per-task `try/except` returning a sentinel | | **Retry-then-skip** | API flakiness is the main failure | Wrap `f` with tenacity / backoff: 3 attempts × exponential, then sentinel | | **Quorum** | Vote / ensemble — need at least K of N | `asyncio.as_completed`, collect K, cancel the rest | Anti-pattern: making this decision implicitly. The single most common production bug in this pattern is "I assumed `gather` would skip failures" or "I assumed one failed branch wouldn't kill the rest" — both are wrong defaults `[aipractitioner/scaling]` `[newline/asyncio-llm]`. ### Step 5 · Pick the reducer | Reducer | Shape change | Cost | When to use | |---|---|---|---| | `concatenate` | `[A, B, C]` → `"A\nB\nC"` | None | Downstream LM can handle big context; you want full evidence | | `summarize` (LM call) | `[A, B, C]` → `"abc"` | +1 LM call | Output must fit in a final prompt; lossy by design | | `vote / majority` | `[A, B, A]` → `A` | None | Self-consistency / ensemble agreement | | `rank-top-K` | `[(a,0.9), (b,0.4), (c,0.7)]` → `[a, c]` | None | Multi-query retrieval, reranking | | `merge-dedupe` | overlapping lists → set | Cheap | Multi-query retrieval, multi-source enrichment | | `tree-reduce` | binary `combine(x, y)` recursively | log₂(N) LM calls | When pairwise merging is meaningful (summary-of-summaries) | Choose by asking: *what does the next node consume?* The reducer is just "adapt the fan-out shape to the consumer's input shape." ### Step 6 · Set per-call timeout *and* total wall-clock budget Two independent timeouts: - **Per-call**: `asyncio.wait_for(call, timeout=30)` — protects against one stuck call holding a semaphore slot forever (the canonical "100 items, 99 done in 5s, the whole job blocked 5 min on item 73"). Default 30s. - **Total wall**: `asyncio.wait_for(gather(...), timeout=N · mean + 3σ)` — protects against pathological N. Default 2× the optimistic wall estimate. In LangGraph, `recursion_limit` is *not* a timeout — set both, plus node-level `RetryPolicy` `[lc-docs/errors]`. In CrewAI, `max_rpm` rate- limits but does not timeout — wrap `kickoff()` in `asyncio.wait_for`. ### Step 7 · Reduce, then verify the reduction is sane After the reducer runs, do one cheap LLM-free check: - `concatenate` → assert `len(merged)` is within expected bounds; raise if one branch returned a 50KB blob. - `vote` → log the vote distribution; if it's near-uniform, the items weren't actually deciding the same question — go back to Step 1. - `rank-top-K` → assert scores are not all identical (failure mode of a broken reranker). These cheap checks catch the "fan-out succeeded but the answer is garbage" bug class that no exception will surface. --- ## 4. 操作模型 (Operation Models) Format: **Trigger → Action → Output → Evidence**. ### OP-1 · `asyncio.gather` with bounded concurrency (Python baseline) - **Trigger**: Pure Python, no graph framework, async LM client (Anthropic / OpenAI), N items, you want a list back. - **Action**: ```python sem = asyncio.Semaphore(N_CONCURRENT) async def bounded(item): async with sem: return await asyncio.wait_for(f(item), timeout=PER_CALL_S) results = await asyncio.gather( *[bounded(i) for i in items], return_exceptions=True, # explicit best-effort ) ok = [r for r in results if not isinstance(r, Exception)] ``` - **Output**: A list (or list-with-holes) ready to reduce. Semaphore is the rate limiter; `return_exceptions=True` is the failure policy; `wait_for` is the per-call timeout. - **Evidence**: `[newline/asyncio-llm]`, `[soumendrak/semaphore]`, `[superfastpython/gather]`. ### OP-2 · `asyncio.as_completed` for quorum / first-K - **Trigger**: You only need K-of-N successful results (vote with early-exit, "first 3 of 10 retrieve hits"). - **Action**: ```python tasks = [asyncio.create_task(f(i)) for i in items] done = [] for fut in asyncio.as_completed(tasks): try: done.append(await fut) except Exception: pass if len(done) >= K: for t in tasks: t.cancel() break ``` - **Output**: First K successful results; remaining cancelled to save cost. - **Evidence**: `[instructor/learn-async]`. ### OP-3 · LangGraph `Send` for dynamic fan-out - **Trigger**: Inside a LangGraph state graph, the number of parallel branches is decided at runtime from state. - **Action**: ```python from langgraph.types import Send def route_fanout(state): return [Send("worker", {"item": x}) for x in state["items"]] graph.add_conditional_edges("planner", route_fanout, ["worker"]) ``` The worker writes to a list-typed state key with a reducer (see OP-4). Cap concurrency at `.compile()` time via `compile(...).with_config({"max_concurrency": N})` or `graph.invoke(..., {"max_concurrency": N})`. - **Output**: True per-invocation map-reduce. Each `Send` is its own trace segment in LangSmith. - **Evidence**: `[deepwiki/mapreduce]`, `[mlplus/langgraph-mr]`, `[medium/send-api]`. Skill `langgraph-sop` OP-4 is the canonical entry. ### OP-4 · Add a reducer for parallel writes (cross-link `O5`) - **Trigger**: Two or more `Send`-spawned workers (or any parallel branches) write to the same state key. Without a reducer, LangGraph raises `InvalidUpdateError`. - **Action**: `summaries: Annotated[list[str], operator.add]` for concatenate, `add_messages` for chat, or a custom reducer for dedupe/top-K. - **Output**: Parallel writes merge per the reducer's algebra. - **Evidence**: `[cheatsheet/gotchas]` "Reducers are mandatory, not optional, for parallel execution"; `[lc-docs/persistence]`. ### OP-5 · CrewAI `kickoff_for_each` (parallel crew invocations) - **Trigger**: Same crew, N independent inputs (e.g., one report per topic). - **Action**: ```python results = crew.kickoff_for_each(inputs=[{"topic": t} for t in topics]) # async variant: kickoff_for_each_async ``` Rate-limit at crew creation: `Crew(..., max_rpm=30)`. Wrap calls in `asyncio.wait_for` for total wall budget. - **Output**: Parallel crew runs, one result per input. - **Evidence**: `[crewai-docs/kickoff]`. Note: CrewAI's hierarchical process is *not* a fan-out primitive — see DC-3. ### OP-6 · `ThreadPoolExecutor` for sync LM clients - **Trigger**: The LM client is synchronous (legacy SDK, no async variant), but the workload is I/O-bound. - **Action**: ```python from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers=N_CONCURRENT) as ex: results = list(ex.map(f, items, timeout=PER_CALL_S)) ``` `as_completed(futures, timeout=...)` for failure isolation. Avoid for CPU-bound `f` — use `ProcessPoolExecutor` or a job queue. - **Output**: Bounded-concurrency parallel sync calls. - **Evidence**: Python stdlib `concurrent.futures` docs. ### OP-7 · LlamaIndex multi-query / batch retrieval - **Trigger**: One user question → N rephrased queries → retrieve over each → reduce. - **Action**: `MultiQueryRetriever` or `QueryFusionRetriever` with