How To Send Map Reduce

How-To: Send() Dynamic Map-Reduce

Send(node, arg) allows dynamically spawning $M$ parallel tasks at runtime where $M$ is not known at graph compile time.

---

Dynamic Map-Reduce Pattern

from typing import TypedDict, Annotated, List
import operator
from langgraph.types import Send
from langgraph.graph import StateGraph
import agentmesh_adapter

class DocumentState(TypedDict):
    documents: List[str]
    summaries: Annotated[List[str], operator.add]

def splitter(state: DocumentState):
    # Dynamically generate a Send command for each document
    return [Send("summarizer", {"doc": doc}) for doc in state["documents"]]

def summarizer(item: dict):
    # Receives isolated item argument
    doc = item["doc"]
    return {"summaries": [f"Summary of {doc}"]}

def reducer(state: DocumentState):
    return {"summaries": ["Combined: " + " & ".join(state["summaries"])]}

builder = StateGraph(DocumentState)
builder.add_node("splitter", splitter)
builder.add_node("summarizer", summarizer)
builder.add_node("reducer", reducer)

builder.add_edge("__start__", "splitter")
builder.add_conditional_edges("splitter", splitter)
builder.add_edge("summarizer", "reducer")
builder.add_edge("reducer", "__end__")

app = agentmesh_adapter.compile(builder)
res = app.invoke({"documents": ["Doc A", "Doc B", "Doc C"], "summaries": []})
print(res["summaries"])

---

C++ Engine Isolation Guarantees

Each Send() item is packaged into a distinct WaveItem with a dedicated argument pointer (customArg). The worker executes with isolated input data without mutating the global state dictionary before reduction.\n