In LangGraph, a reducer attached via Annotated is the mechanism for a state key that several parallel nodes write in the same super-step, and returning “patch commands” (like your (i, j, val) tuples) instead of a full matrix is exactly the recommended pattern for partial updates of a nested structure. But there are three things worth knowing - one of them will bite the moment you initialize the matrix.
1. Why a reducer is required here (not just nice-to-have)
Without a reducer, a state key is backed by a LastValue channel, which by design accepts one write per super-step. Two parallel nodes writing matrix in the same step raise:
InvalidUpdateError: At key 'matrix': Can receive only one value per step.
Use an Annotated key to handle multiple values.
That error text comes straight from the source: langgraph/channels/last_value.py (update() raises InvalidUpdateError with error code INVALID_CONCURRENT_GRAPH_UPDATE when it receives more than one value). With Annotated[..., reducer], the key becomes a BinaryOperatorAggregate channel instead, which folds every write of the step into the current value one by one (langgraph/channels/binop.py, update()).
Docs: Custom reducers.
Also note the order is deterministic: writes within a super-step are applied sorted by task path, not in whatever order threads finish (apply_writes in langgraph/pregel/_algo.py: “sort tasks on path, to ensure deterministic order for update application”). For your use case that doesn’t matter as long as two nodes never patch the same cell in the same step - if they do, decide explicitly who wins, because “last node in sort order” is not something you want to rely on.
2. A gotcha you’ll hit when you seed the matrix
Your snippet doesn’t show how the matrix gets initialized, but your reducer’s current is None guard suggests you’re thinking about it - and this is the part most people miss. Graph input (and update_state values) are applied to channels the same way node writes are: through the reducer. And BinaryOperatorAggregate initializes a list-typed channel to [] (it calls typ() in its __init__), so the current is None guard never fires. The natural way to seed:
graph.invoke({"matrix": [[0, 0], [0, 0]]})
calls apply_matrix_updates([], [[0, 0], [0, 0]]) - the reducer tries to unpack each row as an (i, j, val) triple and crashes:
ValueError: not enough values to unpack (expected 3, got 2)
(Verified on langgraph 1.1.3. Worse: with exactly 3 columns it wouldn’t crash on unpacking - it would silently misinterpret rows as patches.)
Fix - seed via Overwrite, which bypasses the reducer (available since langgraph 1.0.2). The documented pattern is returning it from a node (Bypass reducers with Overwrite), so the robust version is a dedicated init node:
from langgraph.types import Overwrite
def init_matrix(state: State):
return {"matrix": Overwrite([[0, 0], [0, 0]])}
# START -> init_matrix -> (fan out to a, b, ...)
Only one Overwrite per key per super-step is allowed, otherwise InvalidUpdateError (test_pregel.py::test_overwrite_parallel_error) - so init before the fan-out, and don’t combine it with parallel patch writes in the same step.
Passing Overwrite directly as graph input - graph.invoke({"matrix": Overwrite([[0, 0], [0, 0]])}) - also works (verified on 1.1.3: input writes go through the same BinaryOperatorAggregate.update() path, which recognizes Overwrite regardless of who wrote it), but the docs and the test suite only cover the node-return form, so treat input-side Overwrite as an implementation detail that could change; prefer the init node.
If you’re on an older langgraph without Overwrite, or want the seed to stay in invoke(), make the update type self-describing so the reducer can distinguish “seed” from “patch”, e.g. a small tagged union:
MatrixUpdate = Union[
dict, # {"set": [[...], [...]]} - replace
List[Tuple[int, int, int]], # patches
]
def apply_matrix_updates(current, update):
if isinstance(update, dict) and "set" in update:
return deepcopy(update["set"])
result = deepcopy(current)
for i, j, val in update:
result[i][j] = val
return result
3. Reducer signature nuance: it’s called once per write, pairwise
Your annotation Annotated[List[List[int]], apply_matrix_updates] types the second argument as List[Tuple[int, int, int]] - that’s right: each node’s return value is one “update”, and the reducer is invoked pairwise, reducer(current, one_update), once per writing node (see the loop in BinaryOperatorAggregate.update()). It does not receive all of the step’s updates concatenated in a single call. Your implementation already handles this correctly since each node returns a list of patches; just don’t be surprised that with nodes A and B in parallel the reducer runs twice.
Two smaller notes:
- Pydantic works too. Your first snippet used
BaseModel - reducers attach the same way there: data: Annotated[List[List[int]], apply_matrix_updates] as a model field. TypedDict vs BaseModel doesn’t change channel behavior, only input validation.
- Keep the reducer pure - no mutation of
current (your deepcopy is correct), no side effects. It also runs when you call update_state() on a thread, so it’s not exclusively a “node write” hook (time-travel docs).
Working end-to-end example (tested on langgraph 1.1.3)
from typing import Annotated
from typing_extensions import TypedDict
from copy import deepcopy
from langgraph.graph import StateGraph, START, END
from langgraph.types import Overwrite
def apply_matrix_updates(current, updates):
result = deepcopy(current)
for i, j, val in updates:
result[i][j] = val
return result
class State(TypedDict):
matrix: Annotated[list[list[int]], apply_matrix_updates]
def node_a(state: State):
return {"matrix": [(1, 0, 10)]}
def node_b(state: State):
return {"matrix": [(1, 1, 20), (0, 1, 30)]}
def init_matrix(state: State):
return {"matrix": Overwrite([[0, 0], [0, 0]])}
builder = StateGraph(State)
builder.add_node("init", init_matrix)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "init")
builder.add_edge("init", "a")
builder.add_edge("init", "b") # a and b run in the same super-step
builder.add_edge("a", END)
builder.add_edge("b", END)
graph = builder.compile()
result = graph.invoke({})
# {'matrix': [[0, 30], [10, 20]]}
Long story short
- Yes,
Annotated[..., reducer] + patch-command returns is the standard, intended pattern for concurrent partial updates - without it parallel writes raise InvalidUpdateError.
- One latent gotcha awaits at initialization: seeding via
invoke({"matrix": ...}) feeds the full matrix through the reducer against an empty-list default, so it crashes (or silently corrupts for 3-column matrices) - and the current is None guard never fires. Seed from an init node returning Overwrite(initial_matrix) - the documented pattern - or make the reducer accept a “replace” update.
- The reducer is called pairwise, once per writing node; write order within a step is deterministic (task-path sort), but avoid two nodes patching the same cell in one step.
Sources
- Custom reducers: Graph API overview - Docs by LangChain
Overwrite: Use the graph API - Docs by LangChainLastValue.update() single-write guard: libs/langgraph/langgraph/channels/last_value.pyBinaryOperatorAggregate (pairwise fold, typ() default, Overwrite handling): libs/langgraph/langgraph/channels/binop.py- Deterministic write ordering:
apply_writes() in libs/langgraph/langgraph/pregel/_algo.py
- Empirical repro on langgraph 1.1.3 (naive seed →
ValueError; Overwrite seed → correct merge; no reducer → InvalidUpdateError)