The Architecture of Agent Memory: How LangGraph Really Works

A deep dive into state, reducers, and why your AI agent keeps forgetting things

To understand LangGraph's power, one must first understand its memory system. When developers first explore LangGraph, they naturally focus on visible components the Nodes where reasoning happens, the Edges that define flow, and the Tools an agent uses to act. But beneath these lies the execution memory, simply called state, and this is what truly determines how an agent thinks, evolves, persists, and ultimately behaves across time.

State is not just a bucket for data. It is the living record of an agent's reasoning every input it receives, every intermediate thought it produces, every tool output it collects, and every decision it makes as the graph is traversed. This chapter explains why state is designed the way it is in LangGraph, how updates are applied, what reducers are, and why traditional Python models like Pydantic or Dataclass often fail for state in agent workflows.

What State Is And What It Isn't

In LangGraph, the state represents the shared memory of the agent as it executes through nodes. Each node receives the current state as input, performs its logic, and returns only the part of the state it wants to update. LangGraph then merges that update into the existing state according to rules defined by the schema you provided. This isn't a simple assignment; it is a controlled, deterministic update that can append, overwrite, or combine data depending on how the state is defined.

The simplest form of state definition uses a Python TypedDict:

from typing_extensions import TypedDict

class State(TypedDict):
    messages: list
    extra_field: int

This defines the shape of the memory the agent will carry through its execution.

State can also store more than just conversation history. It can accumulate tool outputs, metadata, task status, counters, retrieved documents, and more all in the same object that flows through the graph. In more complex applications, managing this state effectively becomes the agent's core cognitive function, similar to how a human remembers what happened earlier in a conversation.

Why We Don't Rely on Pydantic or Dataclass for State

LangGraph documentation notes that state can in principle be defined using a Pydantic model or a Dataclass in addition to TypedDict. But this flexibility comes with trade offs that are important in real use cases.

The Problem of Efficiency and Partial Updates

Pydantic models and Dataclasses are designed for validated object representation, where each field is expected to be complete and consistent at instantiation. This means that to update even one field, you typically end up reconstructing the entire model or mutating it in place.

In agent workflows, state updates are frequent and often tiny. For example, a node might only want to add a few messages to the history or increment a counter. With Pydantic or Dataclass, you either mutate the whole object, breaking immutability guarantees that LangGraph relies on for deterministic merging, or reconstruct the object with an updated field, which is expensive and often redundant. This is particularly bad when state includes long histories or large collections.

TypedDict, in contrast, allows partial updates easily nodes just return a dict with the keys they want to update. LangGraph then knows how to merge these changes back into the full state object.

Potential Caching Considerations with Pydantic

Some developers have reported that using Pydantic models as state can occasionally lead to unexpected behavior with caching. This may happen because Pydantic internal metadata can vary between instantiations, potentially causing LangGraph to treat logically identical states differently. While this is not universal, it is worth being aware of when choosing your state representation approach.

Reducer Semantics Are Harder to Attach to Rich Models

LangGraph's state merging is driven by reducers functions that define how to combine existing state with new updates. Attaching reducer semantics to Dataclass or Pydantic fields is possible via annotations, but it is much clearer and more natural in a TypedDict declared state where each field's type and reducer annotation are explicit.

For these reasons, most production LangGraph graphs use TypedDict for state schemas, and rely on reducers to control how fields evolve.

What Are Reducers and Why They Matter

Reducers are the core mechanism for merging updates into state. When nodes produce updates, LangGraph looks at each key and either overwrites the value or calls a reducer function to combine the existing value with the new one.

Each reducer function has this signature:

def reducer(current_value, new_value) -> merged_value

By default, if no reducer is specified for a state key, any update to that key will overwrite the existing value. This is fine for simple scalar fields that should be replaced, like status flags or task names.

However, for many common use cases—such as appending messages, accumulating lists of results, or concatenating arrays—overwrite makes no sense. You don't want the history wiped every time a node runs; you want the new data merged into the existing memory.

To do this, LangGraph uses Python's Annotated type with a reducer function:

from typing import Annotated
from operator import add
from typing_extensions import TypedDict

class State(TypedDict):
    history: Annotated[list[str], add]
    count: int

Here, history will use Python's operator.add function to combine old and new lists. Now, if two nodes produce updates to the history key, LangGraph will append new values rather than overwrite.

The Built-in add_messages Reducer

While operator.add works well for simple lists, LangGraph provides a specialized built-in reducer called add_messages specifically designed for handling conversation messages. This reducer is smarter than simple concatenation because it handles message deduplication by ID, which is essential when working with LangChain message objects like HumanMessage, AIMessage, and ToolMessage.

To use it, import add_messages from langgraph.graph:

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import add_messages

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

The add_messages reducer understands LangChain message formats and intelligently merges them. If a new message has the same ID as an existing message, it replaces the old one rather than creating a duplicate. This behavior is particularly important in agentic workflows where tool calls and responses need to be tracked correctly.

For most conversational agents built with LangGraph, add_messages is the recommended reducer for the messages field. It handles edge cases that operator.add would miss, such as message updates during streaming or tool response matching.

Reducers in Action A Detailed Example

Imagine an agent with three nodes. The first logs the start of a workflow, the second logs a step and returns a count increment, and the third logs completion.

Define the state like this:

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class MyState(TypedDict):
    logs: Annotated[list[str], add]
    counter: Annotated[int, add]
Each node returns only the changes it cares about:

def start_node(state: MyState):
    return {"logs": ["Started"], "counter": 1}

def step_node(state: MyState):
    return {"logs": ["Step done"], "counter": 2}

def finish_node(state: MyState):
    return {"logs": ["Finished"], "counter": 3}

As the graph executes, the state evolves. After start_node runs, logs contains just "Started" and counter equals 1. After step_node, logs becomes ["Started", "Step done"] and counter becomes 3. After finish_node completes, logs holds all three entries and counter totals 6.

Reducers made this possible. The logs were appended and counter values accumulated. Without reducers, each new update would overwrite the previous, discarding history.

Reducer Functions Beyond Simple Add

LangGraph doesn't restrict reducers to simple operations like add. You can write custom reducers to handle almost any merging logic you need. For example, you might want to remove old conversation context when a list grows too long, merge dictionaries while preserving keys, only accumulate distinct items, or handle complex transformations like parsing and stored summaries.

Here is an example of a custom reducer that removes duplicates:

def unique_merge(old_list, new_list):
    merged = old_list + new_list
    return list(dict.fromkeys(merged))

class State(TypedDict):
    unique_items: Annotated[list[str], unique_merge]

When updates arrive, unique_merge combines them into a deduplicated list. You also don't have to limit reducer functions to built-ins; custom logic lets you handle domain-specific merging strategies.

Short-Term vs Long-Term Memory

Aside from reducer driven merging, agents often need to persist memory across conversations or sessions. Here LangGraph distinguishes two memory scopes.

Short-term memory is the state that flows through your graph during a single invocation or thread. It persists conversation history, tool outputs, and intermediate values. LangGraph supports thread-scoped checkpoints, so even if an agent execution is paused or interrupted, the state can be resumed later.

Long-term memory is memory that persists across sessions or threads, often stored in a database or vector store. Long-term memory lets agents recall user preferences or facts from earlier conversations, enabling personalization. It isn't stored in the core state object itself; rather, it's integrated through stores that the agent can query and update.

Imagine building an assistant that greets a user, adds the greeting to memory, counts each interaction, and summarizes the history at the end.

Define the state:

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

class AssistantState(TypedDict):
    conversation: Annotated[list[str], add]
    count: Annotated[int, add]

Nodes might look like:

def greet(state: AssistantState):
    return {"conversation": ["Hello!"], "count": 1}

def chat_step(state: AssistantState):
    user_msg = state["conversation"][-1] + " user said hi"
    return {"conversation": [user_msg], "count": 1}

def summarize(state: AssistantState):
    summary = f"Total interactions: {state['count']}"
    return {"conversation": [summary]}

As the graph runs, each node produces only the updates it cares about. The reducer accumulates conversation history and interaction counts, producing a coherent memory at the end.

Why Reducers Matter in Real Workflows

Reducers solve a core semantic problem: how to combine updates from different parts of an agent's reasoning process so nothing is lost. Without reducers, only the last update would be retained for each state field, making conversation history, tool outputs, and incremental accumulations impossible to track reliably.

Reducers also support more complex workflows like parallel node execution, where multiple branches update the same key in the same cycle. Without a clear merge strategy, this would lead to conflicts. Reducers ensure updates are merged according to policy, avoiding silent overwrites.

Summary

State in LangGraph isn't just a static data object. It is the execution memory of the agent, persisted as it moves through the graph and updated in a controlled fashion using reducers. Compared to Pydantic and Dataclasses, TypedDict with annotated reducers provides lightweight, partial updates without unnecessary validation overhead, deterministic merging semantics for concurrent and incremental updates, and support for both short-term and long-termmemory patterns.

Use Case	Recommended Type	Why
Execution state of agent	TypedDict	Lightweight, supports partial updates, compatible with reducers
Input validation	Pydantic	Ensures correct structure before execution
Output validation	Pydantic	Ensures final results adhere to schema
Node-internal structured objects	Dataclass	Provides type safety and immutability within nodes
Merging incremental updates	TypedDict + Reducers	Ensures deterministic accumulation without overwriting

Reducers are the mechanism that turns isolated node outputs into a coherent story—a central function in LangGraph's memory architecture and the foundation for building robust, stateful agents.

Example 1: Basic Reducer Demonstration

"""
Example 1: Basic Reducer Demonstration
======================================
Shows how reducers accumulate values instead of overwriting.
"""

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

print("=" * 60)
print("BASIC REDUCER DEMONSTRATION")
print("=" * 60)

# Define State with Reducers
class State(TypedDict):
    logs: Annotated[list[str], add]      # Will APPEND
    counter: Annotated[int, add]          # Will SUM
    status: str                           # Will OVERWRITE (no reducer)

# Simulate node outputs
def start_node(state: State) -> dict:
    return {
        "logs": ["🚀 Started"],
        "counter": 1,
        "status": "running"
    }

def process_node(state: State) -> dict:
    return {
        "logs": ["⚙️ Processing"],
        "counter": 1,
        "status": "processing"
    }

def finish_node(state: State) -> dict:
    return {
        "logs": ["✅ Finished"],
        "counter": 1,
        "status": "done"
    }

# Simulate LangGraph's reducer behavior
def apply_reducer(current_state: dict, update: dict, state_class) -> dict:
    """Simulates how LangGraph applies reducers to merge updates."""
    new_state = current_state.copy()

    for key, new_value in update.items():
        if key in current_state:
            # Check if field has a reducer annotation
            annotations = state_class.__annotations__.get(key)
            if hasattr(annotations, '__metadata__'):
                # Has reducer - apply it
                reducer_fn = annotations.__metadata__[0]
                new_state[key] = reducer_fn(current_state[key], new_value)
            else:
                # No reducer - overwrite
                new_state[key] = new_value
        else:
            new_state[key] = new_value

    return new_state

# Run simulation
print("\n📍 Initial State:")
state = {"logs": [], "counter": 0, "status": "init"}
print(f"   {state}")

print("\n📍 After start_node:")
update = start_node(state)
print(f"   Node returned: {update}")
state = apply_reducer(state, update, State)
print(f"   State now: {state}")

print("\n📍 After process_node:")
update = process_node(state)
print(f"   Node returned: {update}")
state = apply_reducer(state, update, State)
print(f"   State now: {state}")

print("\n📍 After finish_node:")
update = finish_node(state)
print(f"   Node returned: {update}")
state = apply_reducer(state, update, State)
print(f"   State now: {state}")

print("\n" + "=" * 60)
print("RESULTS:")
print("=" * 60)
print(f"✅ logs APPENDED: {state['logs']}")
print(f"✅ counter SUMMED: {state['counter']} (1+1+1=3)")
print(f"✅ status OVERWROTE: '{state['status']}' (only last value)")
print("=" * 60)

Example 2: Custom Reducers

"""
Example 2: Custom Reducers
==========================
Shows how to create custom reducer functions for specialized merge logic.
"""

from typing import Annotated
from typing_extensions import TypedDict

print("=" * 60)
print("CUSTOM REDUCERS DEMONSTRATION")
print("=" * 60)

# Custom Reducer 1: Keep only unique items
def unique_merge(old_list: list, new_list: list) -> list:
    """Merge lists keeping only unique items (preserves order)."""
    combined = old_list + new_list
    return list(dict.fromkeys(combined))

# Custom Reducer 2: Keep last N items
def keep_last_5(old_list: list, new_list: list) -> list:
    """Keep only the last 5 items."""
    combined = old_list + new_list
    return combined[-5:]

# Custom Reducer 3: Merge dictionaries
def merge_dicts(old_dict: dict, new_dict: dict) -> dict:
    """Deep merge dictionaries."""
    result = old_dict.copy()
    result.update(new_dict)
    return result

# Custom Reducer 4: Max value
def keep_max(old_val: int, new_val: int) -> int:
    """Keep the maximum value."""
    return max(old_val, new_val)

# Define State with custom reducers
class State(TypedDict):
    visited_urls: Annotated[list[str], unique_merge]
    recent_messages: Annotated[list[str], keep_last_5]
    metadata: Annotated[dict, merge_dicts]
    high_score: Annotated[int, keep_max]

# Simulate updates
def simulate_reducer(current, new_value, reducer_fn, name):
    result = reducer_fn(current, new_value)
    print(f"\n📍 {name}:")
    print(f"   Current: {current}")
    print(f"   New:     {new_value}")
    print(f"   Result:  {result}")
    return result

print("\n" + "-" * 60)
print("1️⃣  UNIQUE MERGE (removes duplicates)")
print("-" * 60)
urls = ["google.com", "github.com"]
new_urls = ["github.com", "stackoverflow.com", "google.com"]
urls = simulate_reducer(urls, new_urls, unique_merge, "visited_urls")

print("\n" + "-" * 60)
print("2️⃣  KEEP LAST 5 (sliding window)")
print("-" * 60)
messages = ["msg1", "msg2", "msg3"]
new_messages = ["msg4", "msg5", "msg6", "msg7"]
messages = simulate_reducer(messages, new_messages, keep_last_5, "recent_messages")

print("\n" + "-" * 60)
print("3️⃣  MERGE DICTS (combine dictionaries)")
print("-" * 60)
meta = {"user": "john", "role": "admin"}
new_meta = {"role": "superadmin", "team": "engineering"}
meta = simulate_reducer(meta, new_meta, merge_dicts, "metadata")

print("\n" + "-" * 60)
print("4️⃣  KEEP MAX (maximum value)")
print("-" * 60)
score = 85
new_score = 72
score = simulate_reducer(score, new_score, keep_max, "high_score (72 < 85)")

score = 85
new_score = 95
score = simulate_reducer(score, new_score, keep_max, "high_score (95 > 85)")

print("\n" + "=" * 60)
print("SUMMARY: Custom reducers give you full control over merging!")
print("=" * 60)

Example 3: Full LangGraph Workflow with Reducers

"""
Example 3: Full LangGraph Workflow with Reducers
================================================
A complete working LangGraph example demonstrating state and reducers.
"""

from operator import add
from typing import Annotated
from typing_extensions import TypedDict

try:
    from langgraph.graph import StateGraph, END
    LANGGRAPH_AVAILABLE = True
except ImportError:
    LANGGRAPH_AVAILABLE = False
    print("⚠️  LangGraph not installed. Running simulation mode.")
    print("   Install with: pip install langgraph")
    print()

print("=" * 60)
print("FULL LANGGRAPH WORKFLOW WITH REDUCERS")
print("=" * 60)

# 1. Define State with Reducers
class AgentState(TypedDict):
    messages: Annotated[list[str], add]      # Conversation history
    tool_outputs: Annotated[list[str], add]  # Tool results
    iteration: Annotated[int, add]           # Step counter
    status: str                              # Current status (overwrites)

# 2. Define Node Functions
def think_node(state: AgentState) -> dict:
    """Reasoning node - analyzes the situation."""
    iteration = state.get("iteration", 0) + 1
    thought = f"🧠 Thinking... (iteration {iteration})"
    print(f"   [think_node] {thought}")
    return {
        "messages": [thought],
        "iteration": 1,
        "status": "thinking"
    }

def act_node(state: AgentState) -> dict:
    """Action node - executes a tool."""
    action = "🔧 Executed tool: search_database"
    print(f"   [act_node] {action}")
    return {
        "tool_outputs": [action],
        "status": "acting"
    }

def respond_node(state: AgentState) -> dict:
    """Response node - generates final output."""
    iterations = state.get("iteration", 0)
    tools_used = len(state.get("tool_outputs", []))
    response = f"✅ Completed: {iterations} iterations, {tools_used} tools used"
    print(f"   [respond_node] {response}")
    return {
        "messages": [response],
        "status": "complete"
    }

if LANGGRAPH_AVAILABLE:
    # 3. Build the Graph
    print("\n📍 Building LangGraph...")

    graph = StateGraph(AgentState)

    # Add nodes
    graph.add_node("think", think_node)
    graph.add_node("act", act_node)
    graph.add_node("respond", respond_node)

    # Define edges
    graph.set_entry_point("think")
    graph.add_edge("think", "act")
    graph.add_edge("act", "respond")
    graph.add_edge("respond", END)

    # Compile
    agent = graph.compile()
    print("   ✅ Graph compiled successfully!")

    # 4. Run the Agent
    print("\n📍 Running Agent...")
    print("-" * 60)

    initial_state = {
        "messages": ["👤 User: What is the weather?"],
        "tool_outputs": [],
        "iteration": 0,
        "status": "starting"
    }

    result = agent.invoke(initial_state)

    print("-" * 60)
    print("\n📍 Final State:")
    print(f"   messages: {result['messages']}")
    print(f"   tool_outputs: {result['tool_outputs']}")
    print(f"   iteration: {result['iteration']}")
    print(f"   status: {result['status']}")

else:
    # Simulation mode without LangGraph
    print("\n📍 Simulating workflow (LangGraph not installed)...")
    print("-" * 60)

    state = {
        "messages": ["👤 User: What is the weather?"],
        "tool_outputs": [],
        "iteration": 0,
        "status": "starting"
    }

    # Simulate node execution with reducers
    def apply_update(state, update):
        new_state = state.copy()
        for key, value in update.items():
            if key in ["messages", "tool_outputs", "iteration"]:
                # These have add reducer
                new_state[key] = state[key] + value if isinstance(value, list) else state[key] + value
            else:
                # No reducer - overwrite
                new_state[key] = value
        return new_state

    # Run nodes
    update = think_node(state)
    state = apply_update(state, update)

    update = act_node(state)
    state = apply_update(state, update)

    update = respond_node(state)
    state = apply_update(state, update)

    print("-" * 60)
    print("\n📍 Final State:")
    print(f"   messages: {state['messages']}")
    print(f"   tool_outputs: {state['tool_outputs']}")
    print(f"   iteration: {state['iteration']}")
    print(f"   status: {state['status']}")

print("\n" + "=" * 60)
print("KEY OBSERVATIONS:")
print("=" * 60)
print("✅ messages: APPENDED (user msg + thought + response)")
print("✅ tool_outputs: APPENDED (all tool results collected)")
print("✅ iteration: SUMMED (0 + 1 = 1)")
print("✅ status: OVERWROTE (only 'complete' remains)")
print("=" * 60)

Thanks
Sreeni Ramadorai