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:

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:

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

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:

  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:

  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:

  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:

  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:

  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 num_queries=N; let LlamaIndex parallelise internally. Combine with RRF (reciprocal rank fusion) as the reducer, not naive concat.
• Output: Single ranked node list; recall improves vs. one query.
• Evidence: LlamaIndex retriever modules; skill llamaindex-sop.

OP-8 · Tree-reduce for big-N summarisation

• Trigger: N items too many to summarise in one LM context, but pairwise summarize(a, b) → c is meaningful.
• Action: Recursive pairwise fan-out: layer 1 reduces N → N/2, layer 2 N/2 → N/4, until 1. Each layer is its own bounded fan-out. In LangGraph, model as repeated Send rounds; in plain Python, recursive gather.
• Output: Single summary; total LM cost O(N), latency O(log N).
• Evidence: classical MapReduce; LangChain's "map-reduce chain" predecessor used the same shape.

OP-9 · Retry wrapper per item (decouple from gather)

• Trigger: API flakiness causes intermittent 5xx / 429.
• Action: Wrap f not the gather:

  @tenacity.retry(
      stop=stop_after_attempt(3),
      wait=wait_exponential(multiplier=1, max=10),
      retry=retry_if_exception_type((RateLimitError, APITimeoutError)),
  )
  async def f_retry(item): return await f(item)

  Retry inside the semaphore (slot held during backoff is intentional — it spaces out load).
• Output: Transient failures absorbed; permanent failures still bubble up to the failure-policy layer.
• Evidence: provider rate-limit guides; tenacity docs.

OP-10 · Observability: per-item trace + reduce-step trace

• Trigger: Production fan-out — must be debuggable when one item produces a wrong answer.
• Action: Tag each fan-out call with item_id in the LangSmith / OTel / Logfire trace. Log len(items), len(ok), len(failed), wall_time_s on the reduce step. For LangGraph, each Send is already a distinct trace segment — name the worker node descriptively (summarize_doc not worker).
• Output: Per-item drill-down; reduce-step aggregate metrics.
• Evidence: [swarnendu/best]; LangSmith Send-tracing docs.

5. 困境决策案例 (Dilemma Cases)

DC-1 · "100 items × 5s each, fan out gets 429-storm"

• 困境: Coder wrote await asyncio.gather(*[summarize(d) for d in 100_docs]) against the Anthropic API. First 50 fire instantly, the rest queue inside the client; provider returns 429 for half; tenacity retries pile on; nothing finishes. Wall time worse than sequential.
• 约束:
  • Provider RPM = 50; TPM = 200k; each prompt ≈ 3k tokens (300k TPM if all fire) → TPM is the binding constraint, not RPM.
  • Cannot pre-batch (per-doc tools differ).
  • SLA = 90s.
• 决策步骤:
  1. Compute the real ceiling: N_concurrent = TPM / (60 · tokens_per_call) = 200_000 / (60 · 3000) ≈ 1.1 → round to 2 in-flight, not 10. RPM said 50 — TPM said 2. Bind to the tighter one.
  2. Add a semaphore at 2. Now expected wall is 100 · 5s / 2 = 250s — exceeds SLA.
  3. Reduce N before fan-out: pre-filter docs with a cheap embedding similarity to the query → drop to 20 relevant docs.
  4. Re-compute: 20 · 5s / 2 = 50s ✓ under SLA.
  5. Add return_exceptions=True and a 30s per-call wait_for. If one doc hangs, the others still finish.
• 结果: Throughput drops nominally (2 in-flight vs. 50 attempted), but actual completed-per-second goes up because no retry-storm. SLA met.
• 可提取的操作: The semaphore size is set by the tighter of RPM and TPM, not the looser. When the math says SLA can't be met at the safe concurrency, reduce N before fan-out — never raise concurrency past the real ceiling. [newline/asyncio-llm]