How To Parallel Fanout

How-To: Parallel Fan-Out & Barrier Joins

AgentMesh executes parallel branches simultaneously without GIL contention during I/O operations.

---

Diamond DAG Pattern (1 -> N -> 1)

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

class SwarmState(TypedDict):
    directives: List[str]
    reports: Annotated[List[str], operator.add]
    final_memo: str

def strategist(s):
    return {"reports": ["Directive Issued"]}

def specialist_1(s): return {"reports": ["Valuation: Undervalued"]}
def specialist_2(s): return {"reports": ["Technical: Bullish Divergence"]}
def specialist_3(s): return {"reports": ["Sentiment: Strong Inflows"]}

def aggregator(s):
    return {"final_memo": "Approved: " + " | ".join(s["reports"])}

builder = StateGraph(SwarmState)
builder.add_node("strategist", strategist)
builder.add_node("spec1", specialist_1)
builder.add_node("spec2", specialist_2)
builder.add_node("spec3", specialist_3)
builder.add_node("aggregator", aggregator)

builder.add_edge("__start__", "strategist")

# Fan-out to 3 parallel specialists
builder.add_edge("strategist", "spec1")
builder.add_edge("strategist", "spec2")
builder.add_edge("strategist", "spec3")

# Barrier join at aggregator
builder.add_edge("spec1", "aggregator")
builder.add_edge("spec2", "aggregator")
builder.add_edge("spec3", "aggregator")
builder.add_edge("aggregator", "__end__")

app = agentmesh_adapter.compile(builder)
result = app.invoke({"directives": ["Analyze AAPL"], "reports": [], "final_memo": ""})
print(result["final_memo"])

---

Multi-Target Conditional Routers

A conditional router can return a List[str] to fan out dynamically based on runtime state:

def route_tasks(state: SwarmState) -> List[str]:
    targets = []
    if "macro" in state["directives"]:
        targets.append("spec1")
    if "technical" in state["directives"]:
        targets.append("spec2")
    return targets if targets else ["aggregator"]

builder.add_conditional_edges("strategist", route_tasks)
```\n