01Real-world example
Parallel branches and reducers
Independent nodes run at the same time. A reducer defines how their results merge into one key.
02The graph
03How the code does it
- Three edges from
STARTfan out:api,uiandperfrun in the same step. - Three edges into
summarizefan in: it waits for all three before it runs. Annotated[list[str], operator.add]tells LangGraph to concatenate parallel writes. Without a reducer two writers in one step is an error.summarizehas no type hint on purpose: LangGraph reads a node's type hint as its input schema.
Key points
- Many edges from START run in parallel
- Many writers, one key? Add a reducer
- No reducer means InvalidUpdateError
04Run it
terminal
cd chapter_18_LangGraph/src/chapters
python 005_Parallel_Reducer.pyoutput
api: 42/42 passed
perf: p95 380ms (budget 500ms)
ui: 118/120 passed
3 suites reported
Same graph WITHOUT the reducer:
InvalidUpdateError: At key 'results': Can receive only one value per step. Use an Annotated key to handle multiple values.DDrills for lesson 5
Work in chapter_18_LangGraph/src/chapters. Change one thing, run it, and compare with the expected result.
- Add a security suite
Add a fourth suite that returns
{"results": ["security: 0 high findings"]}.Expected result
4 suites reported, and the new line appears inresults. - Break it on purpose
Remove the reducer from
Stateand run.Expected result
InvalidUpdateError: At key 'results': Can receive only one value per step. Use an Annotated key to handle multiple values. - Check the order
The nodes were added as api, ui, perf. In what order do the results come back?
Expected result
api, perf, ui. Updates from one step are applied in a fixed order, not the order you added the nodes or the order they finish, so never depend on order inside a step.
SThe full script
The exact file from the course repo, plus the output it printed when this page was written.
005_Parallel_Reducer.py
"""005 - Parallel branches and reducers.
Three check suites fan out from START and run in the same step. They all write
to `results`. Without a reducer, LangGraph refuses: two writers, one key, which
one wins? Annotated[list, operator.add] says "append, don't overwrite".
"""
import operator
from typing import Annotated, TypedDict
from langgraph.errors import InvalidUpdateError
from langgraph.graph import END, START, StateGraph
class State(TypedDict):
results: Annotated[list[str], operator.add] # reducer: concatenate
summary: str
def api_checks(state): return {"results": ["api: 42/42 passed"]}
def ui_checks(state): return {"results": ["ui: 118/120 passed"]}
def perf_checks(state): return {"results": ["perf: p95 380ms (budget 500ms)"]}
def summarize(state) -> dict: # untyped: LangGraph reads node hints as input schemas
return {"summary": f"{len(state['results'])} suites reported"}
def build(state_cls):
b = StateGraph(state_cls)
for name, fn in [("api", api_checks), ("ui", ui_checks), ("perf", perf_checks)]:
b.add_node(name, fn)
b.add_edge(START, name) # fan-out: three edges from START
b.add_edge(name, "summarize") # fan-in: summarize waits for all three
b.add_node("summarize", summarize)
b.add_edge("summarize", END)
return b.compile()
class StateNoReducer(TypedDict):
results: list[str] # no reducer on purpose
summary: str
if __name__ == "__main__":
out = build(State).invoke({"results": [], "summary": ""})
for line in out["results"]:
print(" ", line)
print(" ", out["summary"])
print("\nSame graph WITHOUT the reducer:")
try:
build(StateNoReducer).invoke({"results": [], "summary": ""})
except InvalidUpdateError as e:
print(" InvalidUpdateError:", str(e).splitlines()[0])Expected output
output
api: 42/42 passed
perf: p95 380ms (budget 500ms)
ui: 118/120 passed
3 suites reported
Same graph WITHOUT the reducer:
InvalidUpdateError: At key 'results': Can receive only one value per step. Use an Annotated key to handle multiple values.