3. Typed state design for agents

Overview

In the previous capsule you designed an agent's flow with StateGraph: nodes as functions, edges as transitions, the reason→act→observe cycle. But there's a question we left open: what does the agent know at each step? The answer lives in the state — and designing it well is probably the most important architectural decision of this evolving project.

The state is not an implementation detail. It's the contract between all the nodes of your agent. When a planning node writes a plan and a research node reads it, both need to agree on the shape of that data. When a conditional edge checks iteration_count to decide whether to keep going or stop, that field has to exist and have a predictable type. Designing it badly creates coupling, subtle bugs, and pain when you scale.

The critical part: the state you design here isn't just for Module 4. In M5 you'll add planning fields, in M6 memory, in M8 multi-agent. If your state isn't designed for extensibility, every future module requires rewrites. Let's do it right from the start.

Beyond messages

The problem with a minimal state

When you use create_react_agent or MessagesState, the state boils down to:

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

For a simple chat agent, that's enough. Everything lives inside the message list.

But think about your Research Agent. It needs to:

• Know what it already researched — so it doesn't repeat searches.
• Follow a plan — with sub-questions it can mark as completed.
• Count iterations — so it doesn't get stuck in an infinite loop.
• Accumulate data — tool results that pile up.
• Evaluate quality — a score that determines whether the result is good enough.
• Respect limits — token budget, maximum time, maximum iterations.

Can you cram all that into messages? Technically yes — as system messages, metadata, or plain text. But it's a terrible idea:

# Anti-pattern: everything inside messages
messages = [
    SystemMessage(content="Plan: 1. Search X, 2. Analyze Y. Iteration: 3. Score: 7.2"),
    HumanMessage(content="Research transformers"),
    AIMessage(content="..."),
]

The problems:

1. No type safety. A node can write "Iteration: three" instead of 3 and nothing catches it.
2. Fragile parsing. Extracting data from strings is error-prone and slow.
3. Coupling. Every node needs to know the exact string format used by other nodes.
4. Not scalable. Adding a new field means changing parsers everywhere.

What a real agent needs

A sophisticated agent needs structured state — typed fields with clear semantics:

Nodes read and write specific fields. Conditional edges check typed fields instead of parsing strings. Adding a new field doesn't break anything that already exists.

TypedDict as the state contract

Basic definition

In Python, TypedDict lets you define dictionaries with typed fields:

from typing import TypedDict

class AgentState(TypedDict):
    messages: list
    plan: str
    iteration_count: int

This defines a contract: any node that receives AgentState knows exactly what fields exist and what type they are. If you try to access state["quality_score"] and it isn't defined, your editor warns you.

Why TypedDict and not a dataclass

LangGraph uses TypedDict for specific reasons:

1. Partial updates. Nodes return a dict with only the fields that change — they don't need to include all of them.
2. Native merge. LangGraph merges the returned dict with the existing state automatically.
3. Reducers with Annotated. You can control how each individual field is updated.

# A node only updates what it needs
def increment_iteration(state: AgentState) -> dict:
    return {"iteration_count": state["iteration_count"] + 1}

def update_score(state: AgentState) -> dict:
    return {"quality_score": 8.5}

The types that matter for agents

from typing import TypedDict, Optional, Annotated
from langchain_core.messages import BaseMessage

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]  # With a special reducer
    plan: Optional[str]          # Optional — may not exist at the start
    iteration_count: int         # Numeric — flow control
    max_iterations: int          # Numeric — configurable limit
    research_data: list[str]     # Cumulative — grows with each iteration
    quality_score: float         # Score — for quality gates
    metadata: dict               # Extensible — arbitrary metadata

Practical rule: if a conditional edge needs to read a value to make a decision, that value must be a typed field in the state. Don't stuff it into messages.

Reducers: how state gets updated

The update problem

When a node returns {"messages": [new_message]}, what happens to the previous messages? If LangGraph simply overwrote them, you'd lose the whole history. You need messages to accumulate. But for quality_score, you want it overwritten — the new score replaces the old one.

This is what reducers solve: functions that define how the existing value combines with the new value.

add_messages: the reducer for messages

add_messages is the most important reducer in LangGraph:

from langgraph.graph.message import add_messages
from typing import Annotated

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

What does add_messages do?

1. Appends new messages to the end of the existing list.
2. Deduplicates by ID. If a new message has the same id as an existing one, it replaces it instead of duplicating it.
3. Handles ToolMessages correctly, associating them with their tool_calls.

from langchain_core.messages import HumanMessage, AIMessage

existing = [HumanMessage(content="Hello", id="msg-1")]
new = [AIMessage(content="How can I help you?", id="msg-2")]

result = add_messages(existing, new)
# → [HumanMessage("Hello"), AIMessage("How can I help you?")]

# Deduplication: same ID → replaces
updated = [HumanMessage(content="Hello, edited", id="msg-1")]
result = add_messages(existing, updated)
# → [HumanMessage("Hello, edited")]  — replaced, didn't duplicate

Last-write-wins: the default reducer

When you don't use Annotated with a reducer, LangGraph applies last-write-wins: the new value simply replaces the previous one.

class AgentState(TypedDict):
    quality_score: float  # No Annotated → last-write-wins

Perfect for: quality_score (always the most recent), plan (the updated one replaces it), max_iterations (configuration that doesn't change).

operator.add: accumulating lists

For fields where you want to accumulate values, use operator.add:

import operator

class AgentState(TypedDict):
    research_data: Annotated[list[str], operator.add]

# Current state: research_data = ["data 1", "data 2"]
# Node returns: {"research_data": ["data 3"]}
# Result: research_data = ["data 1", "data 2", "data 3"]

Difference from add_messages: operator.add concatenates lists with no deduplication. add_messages handles IDs and deduplication. For simple lists of strings, use operator.add. For messages, always add_messages.

Custom reducers

For more complex logic, you define your own reducer function:

def increment_reducer(current: int, update: int) -> int:
    """Adds the update to the current value."""
    return current + update

class AgentState(TypedDict):
    iteration_count: Annotated[int, increment_reducer]

# Current state: iteration_count = 3
# Node returns: {"iteration_count": 1}
# Result: iteration_count = 4 (3 + 1)

Another example — a reducer that only updates if the new value is greater:

def max_reducer(current: float, update: float) -> float:
    return max(current, update)

class AgentState(TypedDict):
    best_quality_score: Annotated[float, max_reducer]

Reducer summary table

Designing the Research Agent state

With TypedDict, Annotated, and clear reducers, let's design the real state of the Research Agent:

from typing import TypedDict, Annotated, Optional
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
import operator


class ResearchPlan(TypedDict):
    """Research plan with sub-questions."""
    query: str
    sub_questions: list[str]
    status: str  # "pending" | "in_progress" | "complete"


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

    # --- Planning ---
    plan: Optional[ResearchPlan]

    # --- Flow control ---
    iteration_count: int
    max_iterations: int

    # --- Accumulated data ---
    research_data: Annotated[list[str], operator.add]

    # --- Evaluation ---
    quality_score: float

    # --- Extensibility ---
    metadata: dict

Field by field

messages — the main channel between the LLM and the tools. The add_messages reducer guarantees accumulation without duplicates. What NOT to put here: the plan, the scores, the counters — those go in dedicated fields.

plan — Optional because it doesn't exist at the start. Last-write-wins because the plan is replaced whole when the agent re-plans:

def planning_node(state: AgentState) -> dict:
    query = state["messages"][-1].content
    response = model.invoke(
        f"Break this research question down into 3-5 sub-questions:\n{query}"
    )
    return {
        "plan": {
            "query": query,
            "sub_questions": parse_questions(response.content),
            "status": "in_progress"
        }
    }

iteration_count and max_iterations — last-write-wins. They're used together for stop conditions:

def research_node(state: AgentState) -> dict:
    return {
        "messages": [AIMessage(content=result)],
        "iteration_count": state["iteration_count"] + 1
    }

def should_continue(state: AgentState) -> str:
    if state["iteration_count"] >= state["max_iterations"]:
        return "synthesize"
    if state["quality_score"] >= 8.0:
        return "synthesize"
    return "research"

research_data — with operator.add, each node appends data to the existing list:

def research_node(state: AgentState) -> dict:
    results = search_tool.invoke(state["plan"]["sub_questions"][0])
    return {
        "research_data": [results],
        "iteration_count": state["iteration_count"] + 1
    }

def synthesis_node(state: AgentState) -> dict:
    all_data = "\n".join(state["research_data"])
    response = model.invoke(f"Synthesize this data:\n{all_data}")
    return {"messages": [AIMessage(content=response.content)]}

quality_score — last-write-wins. Conditional edges use it as a quality gate:

def evaluation_node(state: AgentState) -> dict:
    data = "\n".join(state["research_data"])
    response = model.invoke(f"Evaluate this research (0.0 to 10.0):\n{data}")
    return {"quality_score": float(response.content.strip())}

metadata — an open dict for info that doesn't deserve its own field yet (timestamps, source tracking, config). When to promote it to its own field: when a conditional edge needs to read it.

Initializing the state

When you invoke the graph, you initialize every field:

initial_state = {
    "messages": [HumanMessage(content="Research transformers in NLP")],
    "plan": None,
    "iteration_count": 0,
    "max_iterations": 5,
    "research_data": [],
    "quality_score": 0.0,
    "metadata": {}
}

result = graph.invoke(initial_state)

State extensibility

The principle: add without breaking

The golden rule: adding new fields should never break existing nodes. This works because:

1. Nodes only read the fields they need.
2. Nodes only return the fields they update.
3. LangGraph merges the partial return with the complete state.

If you add memory_context: list[str] in Module 6, no Module 4 node needs to change.

The planned evolution of the state

Your AgentState is going to grow module by module:

# --- Module 4: Base ---
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    plan: Optional[ResearchPlan]
    iteration_count: int
    max_iterations: int
    research_data: Annotated[list[str], operator.add]
    quality_score: float
    metadata: dict

# --- Module 5 adds: ---
    current_step: int
    reflection_notes: Annotated[list[str], operator.add]
    retry_count: int

# --- Module 6 adds: ---
    conversation_id: str
    long_term_facts: list[str]

# --- Module 8 adds: ---
    active_agent: str
    agent_results: dict[str, str]
    handoff_context: Optional[str]

Pattern: TypedDict inheritance

In practice, you organize the extension with inheritance:

class BaseAgentState(TypedDict):
    """Base state — Module 4."""
    messages: Annotated[list[BaseMessage], add_messages]
    plan: Optional[ResearchPlan]
    iteration_count: int
    max_iterations: int
    research_data: Annotated[list[str], operator.add]
    quality_score: float
    metadata: dict

class PlanningAgentState(BaseAgentState):
    """Extends the base with planning — Module 5."""
    current_step: int
    reflection_notes: Annotated[list[str], operator.add]
    retry_count: int

class MemoryAgentState(PlanningAgentState):
    """Extends planning with memory — Module 6."""
    conversation_id: str
    long_term_facts: list[str]

The M4 nodes keep working with PlanningAgentState because they only access fields from BaseAgentState.

What you should NOT do

# BAD: A generic state with no types — typos = silent bugs
class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    data: dict  # Everything here. state["data"]["plaan"] → no error, silent bug.

# BAD: "Just in case" fields that never get used
class OverEngineeredState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    plan: Optional[str]
    backup_plan: Optional[str]       # Never used
    emergency_plan: Optional[str]    # Never used

Rule: only add a field when a node needs it now. TypedDict + inheritance lets you add without breaking anything.

Comparison: simple state vs rich state

Use MessagesState when your agent is a simple chatbot or a quick prototype. Use a custom state when your agent has phases, stop conditions, or is going to grow. For the Research Agent: custom state, no doubt.

Connection with the project

In this capsule you define the AgentState you'll use from M4 through M10. In capsule 08 (the project), you integrate it with the Research Agent's complete StateGraph. The state feeds every node: planning_node reads messages and writes plan, research_node reads plan and writes research_data, evaluation_node reads research_data and writes quality_score, synthesis_node reads everything and writes messages.

The extension chain:

M4: AgentState (base)
 └─ M5: + planning fields (current_step, reflection_notes, retry_count)
     └─ M6: + memory fields (conversation_id, long_term_facts)
         └─ M7: + MCP fields (available_tools, tool_configs)
             └─ M8: + multi-agent fields (active_agent, agent_results)

If you design well now, every extension is adding fields. If you design badly, every extension is rewriting nodes.

Troubleshooting

Problem 1: "The field doesn't exist in the state"

Symptom: KeyError: 'quality_score'. Cause: the field wasn't included in the initial state.

Solution: Always include every field when you invoke:

result = graph.invoke({
    "messages": [HumanMessage(content="query")],
    "plan": None,
    "iteration_count": 0,
    "max_iterations": 5,
    "research_data": [],
    "quality_score": 0.0,
    "metadata": {}
})

Problem 2: "research_data gets overwritten instead of accumulating"

Cause: You forgot the Annotated with operator.add.

# BAD — last-write-wins
research_data: list[str]

# GOOD — accumulates
research_data: Annotated[list[str], operator.add]

Problem 3: "add_messages duplicates messages"

Cause: You're modifying state["messages"] directly in addition to returning messages.

# BAD — modifies the state AND returns (duplicate)
def bad_node(state: AgentState) -> dict:
    msg = AIMessage(content="answer")
    state["messages"].append(msg)
    return {"messages": [msg]}

# GOOD — only returns
def good_node(state: AgentState) -> dict:
    msg = AIMessage(content="answer")
    return {"messages": [msg]}

Problem 4: "The type checker doesn't catch errors"

Cause: TypedDict is a static hint, not runtime enforcement. Solution: use mypy for static validation. For runtime, add asserts in critical nodes: assert isinstance(score, float) and 0.0 <= score <= 10.0.

Problem 5: "I don't know whether to use a reducer or last-write-wins"

If you're not sure, start with last-write-wins. You can switch to a reducer later without breaking nodes.

Exercises

Exercise 1: State for a support agent

Design a TypedDict for a customer support agent that needs: message history, the current ticket (id, status, priority), number of interactions, whether the problem was resolved, and internal notes. Define the appropriate reducers.

from typing import TypedDict, Annotated, Optional
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages
import operator

class Ticket(TypedDict):
    id: str
    status: str      # "open" | "in_progress" | "resolved" | "escalated"
    priority: str    # "low" | "medium" | "high" | "critical"
    category: str

class SupportAgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    ticket: Optional[Ticket]                              # last-write-wins
    interaction_count: int                                 # last-write-wins
    resolved: bool                                         # last-write-wins
    internal_notes: Annotated[list[str], operator.add]    # accumulates notes

Exercise 2: Custom reducer for quality tracking

Create a reducer that keeps only the last 5 quality scores. Use it in an AgentState.

def last_n_scores(current: list[float], new: list[float], n: int = 5) -> list[float]:
    combined = current + new
    return combined[-n:]

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

def evaluation_node(state: AgentState) -> dict:
    score = evaluate_research(state["research_data"])
    return {"quality_history": [score]}

Exercise 3: Spot the badly initialized state

This code has a bug in the initialization. Find it and fix it:

class AgentState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    research_data: Annotated[list[str], operator.add]
    iteration_count: int
    quality_score: float

result = graph.invoke({
    "messages": [HumanMessage(content="Research AI agents")],
    "iteration_count": 0,
})

result = graph.invoke({
    "messages": [HumanMessage(content="Research AI agents")],
    "research_data": [],
    "iteration_count": 0,
    "quality_score": 0.0,
})

Exercise 4: Refactor from MessagesState to a custom state

Refactor this agent to support iteration counting and quality evaluation:

from langgraph.graph import StateGraph, MessagesState, START, END

def research(state: MessagesState) -> dict:
    response = model.invoke(state["messages"])
    return {"messages": [response]}

def should_continue(state: MessagesState) -> str:
    last = state["messages"][-1]
    if last.tool_calls:
        return "research"
    return END

graph = StateGraph(MessagesState)
graph.add_node("research", research)
graph.add_conditional_edges(START, lambda _: "research")
graph.add_conditional_edges("research", should_continue)
app = graph.compile()

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


class ResearchState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]
    iteration_count: int
    max_iterations: int
    quality_score: float


def research(state: ResearchState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration_count": state["iteration_count"] + 1
    }


def evaluate(state: ResearchState) -> dict:
    score = run_evaluation(state["messages"])
    return {"quality_score": score}


def should_continue(state: ResearchState) -> str:
    if state["iteration_count"] >= state["max_iterations"]:
        return END
    if state["quality_score"] >= 8.0:
        return END
    last = state["messages"][-1]
    if last.tool_calls:
        return "research"
    return END


graph = StateGraph(ResearchState)
graph.add_node("research", research)
graph.add_node("evaluate", evaluate)
graph.add_edge(START, "research")
graph.add_edge("research", "evaluate")
graph.add_conditional_edges("evaluate", should_continue)
app = graph.compile()

result = app.invoke({
    "messages": [HumanMessage(content="Research query")],
    "iteration_count": 0,
    "max_iterations": 5,
    "quality_score": 0.0
})

Exercise 5: Plan the state extension

Given the Research Agent's AgentState, design the additional fields for Module 5 (planning + reflection). The agent needs to: know which sub-question it's working on, save reflection notes, and count correction retries. Use TypedDict inheritance.

class AgentState(TypedDict):
    """Base state — Module 4."""
    messages: Annotated[list[BaseMessage], add_messages]
    plan: Optional[ResearchPlan]
    iteration_count: int
    max_iterations: int
    research_data: Annotated[list[str], operator.add]
    quality_score: float
    metadata: dict

class PlanningAgentState(AgentState):
    """Extension for Module 5."""
    current_step: int
    reflection_notes: Annotated[list[str], operator.add]
    retry_count: int
    max_retries: int

# The M4 nodes keep working:
def research_node(state: AgentState) -> dict:
    return {"research_data": ["data_point"], "iteration_count": state["iteration_count"] + 1}

# New M5 nodes:
def reflection_node(state: PlanningAgentState) -> dict:
    critique = evaluate_output(state["research_data"])
    return {"reflection_notes": [critique], "retry_count": state["retry_count"] + 1}

def should_retry(state: PlanningAgentState) -> str:
    if state["retry_count"] >= state["max_retries"]:
        return "synthesize"
    if state["quality_score"] >= 8.0:
        return "synthesize"
    return "research"

Summary

• The agent's state is an architectural decision, not an implementation detail. It defines what the agent knows and how information flows between nodes.
• TypedDict is the state contract: typed fields with partial updates — each node returns only the fields it modifies.
• Reducers control how each field is updated: add_messages for messages, operator.add for cumulative lists, last-write-wins for values that get replaced, custom reducers for specific logic.
• The Research Agent State has 7 core fields: messages, plan, iteration_count, max_iterations, research_data, quality_score, metadata.
• Extensibility through inheritance: every future module adds fields to the state without breaking existing nodes. Only add fields when a node needs them — don't pre-design.
• Practical rule: if it grows → operator.add. If it gets replaced → no Annotated. If it needs logic → a custom reducer.

Additional resources

1. LangGraph State — official documentation — TypedDict, Annotated, reducers.
2. LangGraph add_messages — How add_messages works internally.
3. Python TypedDict (PEP 589) — The TypedDict specification.
4. Python Annotated (PEP 593) — How Annotated works for type metadata.
5. LangGraph How-to: Define graph state — Practical guide to designing state.