04. State Management

LangGraph's state model is a single shared dict accessed by every node. This is both its strength and a source of contention if multiple nodes write the same keys simultaneously in a concurrent setting. For most single-threaded agent pipelines, this is not a problem—LangGraph executes one node at a time. But in multi-agent setups with subgraphs running in parallel, coordinate writes through dedicated state keys per agent rather than a shared mutable field.

The merge strategy defaults to shallow overwrite for scalar fields and Annotated-operator append for list fields. To customize merge behavior for a field, wrap it with Annotated[field_type, operator_func]. LangGraph also supports Annotated[field_type, reducer_func] where reducer_func(state_value, update_value) -> merged_value. This lets you implement custom logic like "only keep last 5 messages" or "merge dict fields recursively".

from typing import TypedDict, Annotated
from operator import add

def last_n reducer(current, update, *, n=5):
    """Keep only the last n items in a list."""
    combined = current + update
    return combined[-n:]

class SharedState(TypedDict):
    conversation: Annotated[list[str], last_n]
    worker_log: dict[str, str]

When a node raises an exception, LangGraph by default bubbles up the error and terminates the graph run. To implement recovery, catch the exception inside the node and return an error marker in state that a conditional edge then routes to a recovery node:

def fragile_node(state: AgentState) -> AgentState:
    try:
        result = risky_external_call()
        return {"result": result}
    except Exception as e:
        return {"error": str(e), "result": None}

Local verification checkpoint

Run the smallest example from this chapter in a local workspace and record the package version, runtime, data path, and observed output. If the result depends on model size, vector count, CPU/GPU backend, or available memory, note that constraint beside the exercise so the lesson remains reproducible.