State Updates and Reducers

How State Updates Work

When a node returns a dictionary, LangGraph merges it into the current state. By default, returned keys replace existing values.

Python

from typing import TypedDict

class State(TypedDict):
    count: int
    messages: list[str]
    status: str

def node(state: State) -> dict:
    # Only the keys you return get updated
    # Keys you don't return are left unchanged
    return {"count": state["count"] + 1}
    # messages and status are unchanged

Replacement (Default)

Python

from typing import TypedDict

class State(TypedDict):
    name: str
    value: int

def update_name(state: State) -> dict:
    return {"name": "new_name"}  # Replaces previous name

def update_value(state: State) -> dict:
    return {"value": 42}  # Replaces previous value

Replacement is correct for scalar fields (status, current step, confidence score) where you want the latest value only.

Accumulation with operator.add

Use Annotated[list, operator.add] to append to lists rather than replace:

Python

from typing import TypedDict, Annotated
import operator

class PipelineState(TypedDict):
    # Accumulated — each node's return is appended
    messages: Annotated[list[str], operator.add]
    drug_findings: Annotated[list[dict], operator.add]
    errors: Annotated[list[str], operator.add]

    # Replaced — only latest value kept
    current_step: str
    is_complete: bool

def research_node(state: PipelineState) -> dict:
    return {
        "messages": ["Research complete"],
        "drug_findings": [{"drug": "warfarin", "mechanism": "VKOR inhibition"}],
        "current_step": "research_done",
    }

def analysis_node(state: PipelineState) -> dict:
    return {
        "messages": ["Analysis complete"],  # Appended, not replaced
        "drug_findings": [{"drug": "aspirin", "mechanism": "COX-1 inhibition"}],
        "current_step": "analysis_done",
    }

After both nodes run:

messages = ["Research complete", "Analysis complete"]
drug_findings = both drug dicts in a list
current_step = "analysis_done" (latest value)

Custom Reducer Functions

For complex merge logic, provide your own reducer:

Python

from typing import TypedDict, Annotated

def merge_drug_data(existing: dict, update: dict) -> dict:
    """
    Merge drug information dictionaries.
    Later values take precedence for duplicate keys.
    """
    if not existing:
        return update
    return {**existing, **update}

def deduplicate_list(existing: list, update: list) -> list:
    """Merge two lists, removing duplicates (by string value)."""
    seen = set(existing)
    merged = list(existing)
    for item in update:
        if item not in seen:
            seen.add(item)
            merged.append(item)
    return merged

class DrugState(TypedDict):
    # Custom reducers
    drug_profile: Annotated[dict, merge_drug_data]
    unique_interactions: Annotated[list[str], deduplicate_list]

    # Standard accumulation
    messages: Annotated[list[str], operator.add]

def node_a(state: DrugState) -> dict:
    return {
        "drug_profile": {"name": "warfarin", "class": "anticoagulant"},
        "unique_interactions": ["aspirin", "ibuprofen"],
    }

def node_b(state: DrugState) -> dict:
    return {
        "drug_profile": {"mechanism": "VKOR inhibition", "monitoring": "INR"},
        "unique_interactions": ["ibuprofen", "naproxen"],  # ibuprofen is duplicate
    }

# After both nodes:
# drug_profile = {"name": "warfarin", "class": "anticoagulant", "mechanism": "VKOR...", "monitoring": "INR"}
# unique_interactions = ["aspirin", "ibuprofen", "naproxen"]  (no duplicates)

Messages Pattern (LangChain Integration)

When using LangChain message objects, use the built-in add_messages reducer:

Python

from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage
from typing import TypedDict, Annotated

class ChatState(TypedDict):
    messages: Annotated[list, add_messages]  # Built-in reducer for LangChain messages

def llm_node(state: ChatState) -> dict:
    from langchain_openai import ChatOpenAI
    llm = ChatOpenAI(model="gpt-4o-mini")

    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # Appended via add_messages reducer

graph = StateGraph(ChatState)
graph.add_node("llm", llm_node)
graph.set_entry_point("llm")
graph.add_edge("llm", END)

app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="What is warfarin?")]
})

for msg in result["messages"]:
    print(f"{msg.__class__.__name__}: {msg.content[:80]}")

add_messages handles message deduplication (same message ID is replaced, not duplicated) — important when checkpointing and resuming.

Partial State Updates

Nodes only need to return the keys they're updating. Unreturn keys are preserved:

Python

class State(TypedDict):
    a: str
    b: str
    c: str

def node_updates_only_a(state: State) -> dict:
    return {"a": "new_a"}  # b and c unchanged

def node_updates_only_b(state: State) -> dict:
    return {"b": "new_b"}  # a and c unchanged

This is important for efficiency in large state objects — don't return unchanged fields.

Debugging State Updates

Print state after each node to trace updates:

Python

for event in app.stream(initial_state):
    for node_name, node_state in event.items():
        print(f"\nAfter node '{node_name}':")
        for key, value in node_state.items():
            print(f"  {key}: {repr(value)[:100]}")

app.stream() emits a dict per node execution, where the key is the node name and the value is the state change returned by that node. This gives you full visibility into what each node is doing.

State Updates and Reducers

How State Updates Work

Replacement (Default)

Accumulation with operator.add

Custom Reducer Functions

Messages Pattern (LangChain Integration)

Partial State Updates

Debugging State Updates

Enjoyed this article?

Leave a comment