LangGraph Day 2: Watching State Changes Like a TV Series with State + Reducer

State‑update modes

LangGraph provides two ways to update a node's state:

Overwrite – the value returned by the node replaces the existing field.

Merge – the returned field is combined with the existing field according to a reducer.

# Overwrite mode (default)
return {"counter": 10}  # counter becomes 10, other fields unchanged

# Merge mode (requires a reducer)
return {"messages": ["new"]}  # messages are appended, not replaced

Annotated + reducer for custom merge logic

Using Annotated together with a reducer makes the merge rule explicit. Example:

from typing import Annotated, TypedDict
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]  # append mode
    counter: int  # overwrite (default)

def node_a(state: AgentState) -> dict:
    return {
        "messages": ["new message A"],
        "counter": state["counter"] + 1,
    }

Common reducers: operator.add: list concatenation ( left + right) – typical for message history. operator.and_: set intersection – useful for deduplication scenarios.

Custom function – define any merge logic required for special cases.

Message‑specific reducer add_messages

add_messages

is a built‑in reducer that preserves message order and prevents duplicate entries.

from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages, AnyMessage

class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

def chat_node(state: ChatState) -> dict:
    return {"messages": [{"role": "assistant", "content": "reply"}]}

When reducers run

Reducers execute after a node returns but before the state is written:

Node returns {"messages": ["new"]}
   ↓
Reducer runs: existing_messages + ["new"]
   ↓
State update: messages become the merged list

Full example: conversation with message history

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class ConversationState(TypedDict):
    messages: Annotated[list, add_messages]

def chat_node(state: ConversationState) -> dict:
    last_msg = state["messages"][-1]["content"]
    return {"messages": [{"role": "assistant", "content": f"Received: {last_msg}"}]}

builder = StateGraph(ConversationState)
builder.add_node("chat", chat_node)
builder.add_edge(START, "chat")
builder.add_edge("chat", END)

graph = builder.compile()

# First turn
result1 = graph.invoke({"messages": [{"role": "user", "content": "Hello"}]})
print(result1["messages"])
# [{'role': 'user', 'content': 'Hello'}, {'role': 'assistant', 'content': 'Received: Hello'}]

Each graph.invoke() creates a fresh state, so messages do not accumulate across calls. To retain conversation history, compile the graph with a checkpointer and pass a consistent thread_id:

from langgraph.checkpoint.memory import InMemorySaver

graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "demo-1"}}

graph.invoke({"messages": [{"role": "user", "content": "Hello"}]}, config)
# Subsequent calls with the same thread_id automatically append new messages

Custom reducer function

If built‑in reducers are insufficient, a user‑defined merge function can be supplied:

def merge_dicts(left: dict, right: dict) -> dict:
    """Deep merge two dictionaries"""
    result = left.copy()
    for key, value in right.items():
        if (
            key in result
            and isinstance(result[key], dict)
            and isinstance(value, dict)
        ):
            result[key] = merge_dicts(result[key], value)
        else:
            result[key] = value
    return result

class CustomState(TypedDict):
    config: Annotated[dict, merge_dicts]

Day 2 recap

Annotated – marks a field’s type and assigns a reducer.

operator.add – list‑concatenation reducer.

add_messages – message‑append reducer that avoids duplicates.

State merge – returned fields are merged into the existing state according to the reducer rule.

Related links

Official documentation: https://langchain-ai.github.io/langgraph/concepts/low_level/#state

Reducer guide: https://langchain-ai.github.io/langgraph/concepts/low_level/#annotated-state-types