TopolearnMany workflows decide their next step inside a node, or need a variable number of parallel workers.
Conditional edges route after a node. Two tools give more flexibility.
`Command`. A node can return Command(update={...}, goto="node_name"): update the state and choose the next node in one go. Annotate the return type, such as -> Command[Literal["approve", "reject"]], so LangGraph knows the possible targets for drawing and validation. Use it when the routing decision is made by the node's own logic (for example from a model's structured output) and you don't want to store a temporary flag in state just for an edge to read. Command is also how agents hand control to each other, and Command(resume=...) resumes interrupts (later lessons).
`Send`. Sometimes the number of parallel tasks is only known at run time: summarize each of N documents, check each of N claims. Make a conditional edge's routing function return a list of Send("node", payload) objects. LangGraph runs that node once per Send, in parallel, each with its own payload as input (which can differ from the graph state). Results flow back through a reducer, such as operator.add, so they accumulate. This is the map-reduce pattern: map work out, reduce results back.
Tips: keep payloads small; cap parallelism with the max_concurrency config key; give each worker's output a reducer or parallel writes will clash.
Example
import operator
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, Send
class State(TypedDict):
docs: list[str]
summaries: Annotated[list[str], operator.add]
def plan(state: State) -> Command[Literal["review"]]:
return Command(update={"docs": [d.strip() for d in state["docs"]]}, goto="review")
def fan_out(state: State) -> list[Send]:
return [Send("summarize", {"doc": d}) for d in state["docs"]] # one task per doc
def summarize(payload: dict) -> dict:
return {"summaries": [payload["doc"][:10]]}
b = StateGraph(State)
b.add_node(plan)
b.add_node("review", lambda s: {})
b.add_node(summarize)
b.add_edge(START, "plan")
b.add_conditional_edges("review", fan_out, ["summarize"])
b.add_edge("summarize", END)
print(b.compile().invoke({"docs": [" alpha report ", "beta notes"], "summaries": []}))
# {'docs': ['alpha report', 'beta notes'], 'summaries': ['alpha repo', 'beta notes']}Remember
- Command(update=..., goto=...) updates state and routes in one return
- Type the return as Command[Literal[...]] to declare targets
- Returning [Send(node, payload), ...] runs a node once per item in parallel
- Collect Send results with a reducer (map-reduce)