LangGraph Chapter 3 — AgentState & Reducers

Senior Architect Interview Series — LangGraph & Agentic AI

Navigation

← Chapter 2 — StateGraph | Chapter 4 — Tool Calling →

3.0 What This Chapter Covers

State is the backbone of every LangGraph application. Everything flows through it — messages, metadata, tool results, routing decisions. This chapter goes deep on:

How TypedDict defines the state schema
What reducers are and why they exist
How add_messages works under the hood
Common state design patterns for production agents
What happens when you get the state wrong (and how to debug it)

3.1 What Is AgentState?

AgentState is a TypedDict — a Python class that defines the schema of the data flowing through your graph. It tells LangGraph:

What fields exist in the state
What type each field has
How to update each field when a node returns a partial update

# From agent/agent.py
from typing import TypedDict, Annotated
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

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

Three things matter here:

TypedDict — makes AgentState behave like a typed dictionary
Annotated[list[BaseMessage], add_messages] — the type is list[BaseMessage], with a reducer of add_messages
add_messages — the reducer: tells LangGraph how to merge updates to this field

3.2 TypedDict — The Foundation

TypedDict is a standard Python typing construct that lets you create dictionary types with named, typed fields.

from typing import TypedDict

class SimpleState(TypedDict):
    query: str
    count: int
    result: str | None

LangGraph uses TypedDict because:

Runtime validation: LangGraph can validate state updates at runtime
IDE support: Full autocomplete and type checking
Documentation: The state schema documents itself
Serialization: Easy to serialize/deserialize for checkpointing

How LangGraph Uses It

When StateGraph(AgentState) is created, LangGraph inspects the AgentState type hints to build an internal state model. When a node returns {"messages": [new_message]}, LangGraph uses this model to:

Find the messages field
Get its reducer (add_messages)
Apply the reducer: new_messages = add_messages(existing_messages, [new_message])

3.3 Reducers — The Core Concept

A reducer is a function that defines how to merge a new value into an existing field.

reducer(existing_value, new_value) → merged_value

Without a Reducer (Default Behavior)

If a field has no reducer, LangGraph uses replace semantics: the node's return value completely replaces the current value.

class BadState(TypedDict):
    messages: list[BaseMessage]   # NO reducer annotation

# Node returns 1 new message
def call_llm(state: BadState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}   # THIS REPLACES the entire messages list!

# After node: state["messages"] = [response_only]
# User's original question is GONE

This is a critical bug: without a reducer, every node call wipes out the previous messages, making the agent amnesiac.

With a Reducer (Correct Behavior)

class GoodState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # WITH reducer

def call_llm(state: GoodState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}   # appended, NOT replaced

# After node: state["messages"] = [original_question, ..., response]
# Full history preserved

3.4 The `add_messages` Reducer — How It Works

add_messages is LangGraph's built-in reducer for message lists. It does more than just append.

Basic Append Behavior

from langgraph.graph.message import add_messages

existing = [HumanMessage("hello"), AIMessage("hi there")]
new_msg  = [AIMessage("how can I help?")]

result = add_messages(existing, new_msg)
# result = [HumanMessage("hello"), AIMessage("hi there"), AIMessage("how can I help?")]

Deduplication by ID

add_messages also deduplicates by message id. If a new message has the same id as an existing message, it replaces the existing one instead of appending.

msg_with_id = AIMessage(content="original", id="msg-123")
existing = [HumanMessage("hello"), msg_with_id]

updated = AIMessage(content="updated", id="msg-123")  # same id
result = add_messages(existing, [updated])
# result = [HumanMessage("hello"), AIMessage("updated", id="msg-123")]
# The old message was REPLACED, not duplicated

This is used for streaming — you can emit partial messages with the same ID and they'll be updated in-place rather than duplicated.

Message Type Hierarchy

All messages work with add_messages:

BaseMessage
    ├── HumanMessage    — user's input
    ├── AIMessage       — LLM response (may have tool_calls)
    ├── SystemMessage   — system prompt
    ├── ToolMessage     — result of a tool call
    └── FunctionMessage — older OpenAI function calling format

3.5 State in Your Project — Traced End to End

Let's trace the exact state transformations in /Users/c58706/python/agent/agent.py:

Initial State

# In run_agent():
history = load_history(session_id, db)  # [HumanMsg, AIMsg, HumanMsg, AIMsg, ...]
history.append(HumanMessage(content=question))  # add new user question

initial_state = {"messages": history}
# e.g., state["messages"] = [
#     HumanMessage("what is Agent Factory?"),   ← from history
#     AIMessage("Agent Factory is..."),          ← from history
#     HumanMessage("how does it work?")          ← NEW question
# ]

After `call_llm` (first pass — LLM decides to call tools)

def call_llm(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}
# response is an AIMessage with tool_calls:
#   AIMessage(content="", tool_calls=[{"name": "rag_search", "args": {"query": "how does Agent Factory work"}}])

# After add_messages reducer:
# state["messages"] = [
#     HumanMessage("what is Agent Factory?"),
#     AIMessage("Agent Factory is..."),
#     HumanMessage("how does it work?"),
#     AIMessage(tool_calls=[{"name": "rag_search", ...}])  ← APPENDED
# ]

After `call_tools`

def call_tools(state: AgentState) -> AgentState:
    last_message = state["messages"][-1]
    tool_results = []
    for tool_call in last_message.tool_calls:
        output = tool_map[tool_call["name"]].invoke(tool_call["args"])
        tool_results.append(ToolMessage(content=str(output), tool_call_id=tool_call["id"]))
    return {"messages": tool_results}

# After add_messages reducer:
# state["messages"] = [
#     HumanMessage("what is Agent Factory?"),
#     AIMessage("Agent Factory is..."),
#     HumanMessage("how does it work?"),
#     AIMessage(tool_calls=[{"name": "rag_search", ...}]),
#     ToolMessage("Agent Factory is PepsiCo's...")   ← APPENDED
# ]

After `call_llm` (second pass — LLM has tool result, writes final answer)

# LLM sees full context including ToolMessage
response = llm_with_tools.invoke(state["messages"])
# response = AIMessage("Based on the search results, Agent Factory is...")

# After add_messages reducer:
# state["messages"] = [
#     HumanMessage(...), AIMessage(...), HumanMessage(...),
#     AIMessage(tool_calls=[...]),
#     ToolMessage("Agent Factory is PepsiCo's..."),
#     AIMessage("Based on the search results, Agent Factory is...")   ← FINAL ANSWER
# ]

Final Extraction

# In run_agent():
final_state = agent.invoke({"messages": history})
answer = final_state["messages"][-1].content  # "Based on the search results..."

3.6 Writing Custom Reducers

You can write any reducer function — it just needs to accept (existing, update) and return the merged result.

Simple Append Reducer

from typing import Annotated

def append_list(existing: list, update: list) -> list:
    return existing + update

class MyState(TypedDict):
    logs: Annotated[list[str], append_list]

Max Reducer

def take_max(existing: int, update: int) -> int:
    return max(existing, update)

class MyState(TypedDict):
    max_score: Annotated[int, take_max]

Set Reducer

def merge_set(existing: set, update: set) -> set:
    return existing | update

class MyState(TypedDict):
    visited_nodes: Annotated[set[str], merge_set]

Operator Reducer (Using `operator.add`)

import operator

class MyState(TypedDict):
    scores: Annotated[list[float], operator.add]  # same as append
    total_calls: Annotated[int, operator.add]     # same as +

3.7 Multi-Field State — Production Patterns

Real production agents need more state than just messages. Here's how to extend AgentState:

Pattern 1 — Adding Metadata Fields

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    session_id: str          # no reducer = replace-on-write (set once)
    user_id: str
    route: str               # set by supervisor, read by router
    attempt_count: Annotated[int, operator.add]  # increment on each retry

Pattern 2 — Tracking Intermediate Results

class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    search_results: Annotated[list[str], operator.add]  # accumulate results
    sources: Annotated[list[str], operator.add]
    final_answer: str | None  # replace semantics (set once at end)

Pattern 3 — Error Tracking

class RobustAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    errors: Annotated[list[str], operator.add]    # accumulate errors
    retry_count: Annotated[int, operator.add]     # count retries
    last_error: str | None                         # most recent error

3.8 Common Mistakes and Debugging

Mistake 1 — Forgot the Reducer

Symptom: Agent gives answers but has no memory of the conversation; every response ignores context.

# WRONG
class BadState(TypedDict):
    messages: list[BaseMessage]   # no reducer!

# CORRECT
class GoodState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

Mistake 2 — Returning Full State Instead of Partial

Symptom: State fields reset to their initial values mid-conversation.

# WRONG — returns full state, LangGraph tries to reduce it and gets confused
def call_llm(state: AgentState) -> AgentState:
    response = llm.invoke(state["messages"])
    return state   # DON'T return the full state

# CORRECT — return only the fields you're updating
def call_llm(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # only the update

Mistake 3 — Wrong Field Name in Return

Symptom: KeyError or field never gets updated.

# WRONG
return {"message": [response]}   # typo — "message" not "messages"

# CORRECT
return {"messages": [response]}

Debugging State

# Print state at each step using stream()
for step, state in agent.stream(initial_state):
    print(f"\n=== Step: {step} ===")
    for key, value in state.items():
        if key == "messages":
            print(f"  messages: {len(value)} messages")
            print(f"  last: {value[-1].__class__.__name__}: {str(value[-1].content)[:100]}")
        else:
            print(f"  {key}: {value}")

3.9 State Initialization

When you call agent.invoke(initial_state), you must provide at least the required fields:

# Minimal initialization
result = agent.invoke({"messages": [HumanMessage("hello")]})

# With additional fields
result = agent.invoke({
    "messages": [HumanMessage("hello")],
    "session_id": "user-123",
    "attempt_count": 0
})

Fields not provided in the initial state will be None (or cause a KeyError when accessed) — always initialize all required fields.

3.10 Interview Q&A

Q: What is a reducer in LangGraph and why is it necessary?

A reducer is a function that defines how to merge a node's partial state update into the current state. Without a reducer, LangGraph uses replace semantics — a node's return value completely overwrites the existing field. For message history, this is catastrophic: each node call would destroy all previous messages. The add_messages reducer implements append semantics with deduplication, so every new message is added to the history rather than replacing it. Reducers are declared with Annotated[type, reducer_fn] in the TypedDict state class.

Q: What does Annotated[list[BaseMessage], add_messages] mean technically?

Annotated is a Python typing construct from the typing module that attaches arbitrary metadata to a type hint. Annotated[list[BaseMessage], add_messages] means: "the type is list[BaseMessage], and the metadata is add_messages". LangGraph reads this metadata at graph construction time — when it pulls the reducer from the annotation — and uses it to process state updates. This is a clean pattern for embedding behavioral metadata in type hints without changing the type itself.

Q: Explain how the full state evolves during one ReAct cycle in your agent.

The state starts with the conversation history plus the new user question. call_llm runs and appends an AIMessage with tool_calls (the LLM decides to search). should_call_tools reads this and routes to call_tools. call_tools executes the rag_search tool, gets the RAG result, and appends a ToolMessage to state. call_llm runs again — now seeing the full history including the ToolMessage — and appends a final AIMessage with the actual answer. should_call_tools sees no tool_calls in the last message and returns END. The caller reads state["messages"][-1].content as the answer.

Q: How would you add an iteration limit to prevent infinite loops?

Add a counter field to the state with an operator.add reducer. In the routing function, check it against the maximum: if state["iteration_count"] >= 10: return END. Each call to call_llm returns {"iteration_count": 1} as part of its update — the reducer accumulates it. This is a zero-cost guard against runaway agent loops, and it degrades gracefully (the LLM can still use its partial information to answer).

Q: When would you use operator.add as a reducer vs. add_messages?

Use add_messages exclusively for message lists — it handles BaseMessage objects and provides ID-based deduplication which is essential for streaming. Use operator.add for numeric counters, simple string concatenation, or basic list accumulation where you don't need deduplication. For primitives (strings, single values that should be overwritten), use no reducer at all — LangGraph's default replace semantics is correct for fields like route, user_id, final_answer.

3.11 Key One-Liners to Memorize

"State is a TypedDict — nodes return partial updates, reducers merge them."

"No reducer = replace. add_messages = append with deduplication."

"Annotated[list[BaseMessage], add_messages] is the canonical LangGraph pattern."

"add_messages deduplicates by message ID — same ID = replace, not append."

"Returning the full state from a node is a bug — return only the changed fields."

"Every field you read in a node, check it's initialized in the starting state."

Next: Chapter 4 — Tool Calling End-to-End

LangGraph & Agentic AIComplete Interview Prep Guide