/agentsop-map-reduce-fanout

name: agentsop-map-reduce-fanout
version: 0.1.0
description: |
  Decision protocol for the map-reduce / dynamic fan-out pattern in LM
  pipelines — "given list L, run f(item) for each item in parallel, then
  combine". Activates when the coder agent is about to process N items with N
  LM calls (per-doc summarize, per-query retrieve, per-candidate rank,
  parallel tool fan-out). Encodes the *when*, *how many at once*, *what to
  do when one fails*, and *how to reduce* — not the API of any single
  framework. Cross-framework: LangGraph `Send`, CrewAI parallel tasks /
  Flow, `asyncio.gather`, `ThreadPoolExecutor`, LlamaIndex batch retrieval.

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