Typed State with TypedDict and Annotated

Capsule overview

State design is THE most important decision you make when building a graph. Well-designed state makes everything else easy — the nodes are simple, the edges are clear, and debugging is straightforward. Badly designed state produces silent bugs, data that quietly disappears, and nodes that can't communicate properly.

In capsule 02 you saw TypedDict for the first time when you created your first StateGraph. You defined a basic state with messages and used it to pass information between nodes. Now you're going much deeper: you'll understand exactly how Annotated works with reducers, why operator.add is critical for message lists, what happens when you DON'T use it (spoiler: you lose data), and how to design state for different kinds of applications.

Think of state as your graph's architectural blueprint. Just as an architect doesn't start building without plans, you shouldn't write nodes or edges without designing your state first. This capsule teaches you to think like an architect of AI workflows.

TypedDict: the foundation of state

TypedDict is the standard way to define a graph's state in LangGraph. It defines a dictionary with specific keys and types:

from typing import TypedDict

class MyState(TypedDict):
    messages: list[str]
    current_step: str
    is_done: bool

Why TypedDict and not a dataclass? Because LangGraph works internally with dictionaries — every node receives a dict and returns a dict. TypedDict gives you type hints and IDE autocomplete without changing that nature:

state: MyState = {"messages": ["hello"], "step_count": 0}
print(type(state))       # <class 'dict'>
print(state["messages"])  # ['hello']

The problem: silent overwrites

Before understanding reducers, you need to see the bug they solve. This is the most common mistake in LangGraph, and the hardest one to diagnose if you don't know what to look for.

Without reducers — data that disappears

Imagine a graph with two nodes. Both add a message to the state:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class BuggyState(TypedDict):
    messages: list[str]
    final_answer: str

def node_a(state: BuggyState) -> dict:
    return {"messages": ["Hello from node A"]}

def node_b(state: BuggyState) -> dict:
    return {"messages": ["Hello from node B"]}

graph_builder = StateGraph(BuggyState)
graph_builder.add_node("a", node_a)
graph_builder.add_node("b", node_b)
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "final_answer": ""})
print(result["messages"])
# Output: ['Hello from node B']
# !! "Hello from node A" was LOST !!

Node A wrote ["Hello from node A"] into messages. Then node B wrote ["Hello from node B"]. Without a reducer, LangGraph simply replaces the value — the last node wins, and node A's message disappears.

In a chat application, this would mean that every time a node processes a message, all the previous messages get erased. Your chatbot would have total amnesia.

Annotated and reducers: the solution

What a reducer is

A reducer is a function that tells LangGraph how to combine the value a node returns with the value that already exists in the state. Instead of replacing, the reducer defines the merge logic.

The syntax uses Python's Annotated:

from typing import TypedDict, Annotated
import operator

class MyState(TypedDict):
    messages: Annotated[list[str], operator.add]  # Reducer: accumulates
    current_step: str                              # No reducer: replaces

Annotated[list[str], operator.add] says: "this field is a list of strings, and when a node returns a new value, concatenate the new list with the existing one instead of replacing it."

operator.add for lists: ACCUMULATES

Now let's fix the bug from the previous example:

from dotenv import load_dotenv
load_dotenv()

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

class FixedState(TypedDict):
    messages: Annotated[list[str], operator.add]  # ← Reducer
    final_answer: str

def node_a(state: FixedState) -> dict:
    return {"messages": ["Hello from node A"]}

def node_b(state: FixedState) -> dict:
    return {"messages": ["Hello from node B"]}

graph_builder = StateGraph(FixedState)
graph_builder.add_node("a", node_a)
graph_builder.add_node("b", node_b)
graph_builder.add_edge(START, "a")
graph_builder.add_edge("a", "b")
graph_builder.add_edge("b", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "final_answer": ""})
print(result["messages"])
# Output: ['Hello from node A', 'Hello from node B']
# Both messages preserved!

The only difference is Annotated[list[str], operator.add]. Now:

1. The initial state has messages: []
2. Node A returns {"messages": ["Hello from node A"]} → LangGraph runs [] + ["Hello from node A"] → ["Hello from node A"]
3. Node B returns {"messages": ["Hello from node B"]} → LangGraph runs ["Hello from node A"] + ["Hello from node B"] → ["Hello from node A", "Hello from node B"]

Without Annotated for scalars: REPLACES

Fields without Annotated use the default behavior: replacement. That's exactly what you want for values representing "the current state" of something:

from typing import TypedDict, Annotated
import operator

class ProcessState(TypedDict):
    messages: Annotated[list[str], operator.add]  # Accumulates
    current_topic: str                             # Replaces
    confidence: float                              # Replaces
    is_complete: bool                              # Replaces

Here current_topic, confidence, and is_complete get replaced every time a node updates them — which is the right behavior. If the analysis node decides the current topic is "machine learning" with confidence 0.85, you want that value, not a concatenation with previous ones.

The golden rule

• ✅ Lists that accumulate data (messages, sources, results) → Annotated[list[...], operator.add]
• ✅ Scalars that represent the current state (topic, confidence, step) → No Annotated (replaces)
• ❌ Lists without a reducer → Silent bug, data gets lost
• ❌ Scalars with operator.add → Type errors (you can't add strings with + this way)

Custom reducers

operator.add covers most cases, but sometimes you need custom merge logic. A custom reducer is any function that takes two arguments (the current value and the new one) and returns the combined value.

Example: deduplicating sources

from typing import TypedDict, Annotated
import operator

def deduplicate_sources(current: list[str], new: list[str]) -> list[str]:
    seen = set(current)
    result = list(current)
    for source in new:
        if source not in seen:
            result.append(source)
            seen.add(source)
    return result

class ResearchState(TypedDict):
    messages: Annotated[list, operator.add]
    sources: Annotated[list[str], deduplicate_sources]  # No duplicates
    current_topic: str

If node A returns {"sources": ["arxiv.org/123", "wiki/AI"]} and node B returns {"sources": ["wiki/AI", "docs.python.org"]}, the result will be ["arxiv.org/123", "wiki/AI", "docs.python.org"].

Example: keeping the highest value

def keep_highest(current: float, new: float) -> float:
    return max(current, new)

class AnalysisState(TypedDict):
    best_confidence: Annotated[float, keep_highest]
    current_step: str

When to write a custom reducer

• ✅ Deduplicating items in a list
• ✅ Applying a maximum limit (e.g., keeping only the last N messages)
• ✅ Merging with business logic (e.g., prioritizing certain values over others)
• ❌ Accumulating simple lists → use operator.add
• ❌ Replacing values → don't use a reducer

MessagesState: the shortcut for chat

LangGraph includes a prebuilt state for chat applications: MessagesState. It already has messages configured with operator.add and uses LangChain's message types:

from langgraph.graph import MessagesState

# MessagesState is equivalent to:
# class MessagesState(TypedDict):
#     messages: Annotated[list[AnyMessage], operator.add]

Using MessagesState directly

from dotenv import load_dotenv
load_dotenv()

from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END

model = init_chat_model("openai:gpt-4.1-mini")

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

graph_builder = StateGraph(MessagesState)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [HumanMessage(content="What is Python?")]})
print(result["messages"][-1].content)
# Output: Python is a high-level, interpreted programming language...

Extending MessagesState with extra fields

For most real applications, you need more fields beyond messages. You can extend MessagesState:

from typing import Annotated
import operator
from langgraph.graph import MessagesState

class ChatbotState(MessagesState):
    user_name: str
    conversation_topic: str
    sources: Annotated[list[str], operator.add]

This inherits messages with its reducer already configured and adds your custom fields.

When to use MessagesState vs custom state?

• ✅ Use MessagesState when your graph is primarily a chatbot or a conversational system
• ✅ Extend MessagesState when you need extra fields but the core is still conversation
• ✅ Use custom state when your graph processes non-conversational data (e.g., a data pipeline, ETL)
• ❌ Don't force MessagesState onto graphs that aren't chat — design your own state

Designing state for different use cases

State varies by application type. These three patterns cover most scenarios:

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

# Chat: messages + user context
class ChatState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]  # Accumulates
    user_id: str                                          # Replaces
    conversation_summary: str                             # Replaces

# Research: multiple sources that accumulate
class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]   # Accumulates
    sources: Annotated[list[str], operator.add]           # Accumulates
    findings: Annotated[list[str], operator.add]          # Accumulates
    current_topic: str                                     # Replaces
    confidence: float                                      # Replaces
    is_complete: bool                                      # Replaces

# Multi-step pipeline: data that gets transformed + errors that accumulate
class PipelineState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]   # Accumulates
    raw_input: str                                         # Replaces
    analysis_results: Annotated[list[dict], operator.add] # Accumulates
    errors: Annotated[list[str], operator.add]            # Accumulates
    current_step: str                                      # Replaces

The pattern is consistent: lists that grow use operator.add, scalars representing "the current thing" get replaced.

Best practices for designing state

What to put in the state

• ✅ Data that needs to flow between nodes — if one node produces something another node consumes, it goes in the state
• ✅ Conversation history — almost always messages with operator.add
• ✅ Flow-control flags — like is_complete, should_continue, current_step
• ✅ Intermediate results — if you need to accumulate data from multiple nodes

What NOT to put in the state

• ❌ Static configuration — API keys, model names, constants. Use module variables or config
• ❌ Non-serializable objects — database connections, file handles, HTTP clients
• ❌ Huge data — if a field is going to hold megabytes of data, rethink your design
• ❌ Duplicates — if you can derive a value from other state fields, don't duplicate it

State as the single source of truth

A node should never depend on global variables, temp files, or side effects. Everything a node needs to do its job must be in the state:

# BAD — depends on a global variable
results_cache = []

def search_node(state):
    results = do_search(state["current_topic"])
    results_cache.extend(results)  # Side effect!
    return {"current_step": "analyze"}

# GOOD — everything in the state
def search_node(state):
    results = do_search(state["current_topic"])
    return {
        "findings": results,
        "current_step": "analyze"
    }

Partial updates

Nodes don't need to return the whole state. They return only the fields they want to update:

class MyState(TypedDict):
    messages: Annotated[list, operator.add]
    current_step: str
    confidence: float
    is_complete: bool

def analyze_node(state: MyState) -> dict:
    # Updates only 2 of 4 fields
    return {
        "confidence": 0.92,
        "current_step": "summarize"
    }
    # messages and is_complete are left alone

LangGraph applies the update only to the returned fields. Fields not mentioned keep their current value.

Complete example: an analysis pipeline

Let's put it all together in a graph with two nodes that accumulate findings:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

class AnalysisState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    findings: Annotated[list[str], operator.add]
    current_step: str
    confidence: float

model = init_chat_model("openai:gpt-4.1-mini")

def extract_topics(state: AnalysisState) -> dict:
    last_message = state["messages"][-1].content
    response = model.invoke(
        f"Extract the 3 main topics, comma-separated:\n{last_message}"
    )
    topics = [t.strip() for t in response.content.split(",")]
    return {"findings": [f"Topics: {', '.join(topics)}"], "current_step": "sentiment"}

def analyze_sentiment(state: AnalysisState) -> dict:
    last_message = state["messages"][-1].content
    response = model.invoke(
        f"Overall sentiment (positive/negative/neutral):\n{last_message}"
    )
    return {
        "findings": [f"Sentiment: {response.content.strip()}"],
        "confidence": 0.85, "current_step": "done"
    }

graph_builder = StateGraph(AnalysisState)
graph_builder.add_node("extract", extract_topics)
graph_builder.add_node("sentiment", analyze_sentiment)
graph_builder.add_edge(START, "extract")
graph_builder.add_edge("extract", "sentiment")
graph_builder.add_edge("sentiment", END)
graph = graph_builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content=(
        "Python still dominates data science and AI. "
        "Its library ecosystem makes it indispensable."
    ))],
    "findings": [], "current_step": "extract", "confidence": 0.0
})

print("Accumulated findings:")
for finding in result["findings"]:
    print(f"  - {finding}")
print(f"Confidence: {result['confidence']}")
# Output:
# Accumulated findings:
#   - Topics: Python, data science, AI
#   - Sentiment: positive
# Confidence: 0.85

findings accumulates from both nodes (thanks to operator.add). current_step and confidence get replaced with the latest value. Each node only returns the fields it updates — everything else stays intact.

Troubleshooting

Problem 1: The message list only shows the last value

Symptom: Your messages field only contains what the last node returned, losing everything before it. Cause: You forgot to add Annotated[..., operator.add] to the field. Fix:

# BAD
class MyState(TypedDict):
    messages: list[str]  # No reducer → replaces

# GOOD
class MyState(TypedDict):
    messages: Annotated[list[str], operator.add]  # With reducer → accumulates

Problem 2: TypeError when using operator.add with strings

Symptom: TypeError: can only concatenate str (not "str") to str or results like "hello" + "world" = "helloworld". Cause: You used operator.add on a str field instead of a list[str]. The + operator on strings concatenates the characters. Fix: operator.add is for lists. For scalars you want to replace, don't use a reducer:

# BAD
class MyState(TypedDict):
    current_topic: Annotated[str, operator.add]  # "topic1" + "topic2" = "topic1topic2"

# GOOD
class MyState(TypedDict):
    current_topic: str  # Replaces: "topic1" → "topic2"

Problem 3: The node can't read state fields

Symptom: KeyError when accessing a state field inside a node. Cause: The field wasn't included in the initial state when the graph was invoked. Fix: Include all the fields in the initial input with default values:

# BAD — 'findings' is missing from the input
result = graph.invoke({"messages": [HumanMessage(content="hello")]})

# GOOD — every field is present
result = graph.invoke({
    "messages": [HumanMessage(content="hello")],
    "findings": [],
    "current_step": "start",
    "confidence": 0.0
})

Problem 4: The custom reducer never runs

Symptom: Your custom reducer function is never called; the field behaves as if it had no reducer. Cause: The function isn't correctly referenced in Annotated, or the node doesn't return that field. Fix: Check that the function is callable and that the node includes the field in its return:

def my_reducer(current: list, new: list) -> list:
    print(f"Reducer called: {current} + {new}")  # Debug
    return current + new

class MyState(TypedDict):
    items: Annotated[list[str], my_reducer]

def my_node(state: MyState) -> dict:
    return {"items": ["new_item"]}  # Must include 'items' to trigger the reducer

Exercises

Exercise 1: Diagnose the overwrite bug (Easy)

The following graph has a bug: the first node's messages get lost. Find the problem and fix it.

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class ChatState(TypedDict):
    messages: list[str]
    user_name: str

def greet(state):
    return {"messages": [f"Hello, {state['user_name']}!"]}

def ask_question(state):
    return {"messages": ["How can I help you?"]}

graph_builder = StateGraph(ChatState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask_question)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", END)
graph = graph_builder.compile()

result = graph.invoke({"messages": [], "user_name": "Carlos"})
print(result["messages"])
# Actual: ['How can I help you?'] — the greeting is missing!
# Expected: ['Hello, Carlos!', 'How can I help you?']

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

class ChatState(TypedDict):
    messages: Annotated[list[str], operator.add]  # ← Add the reducer
    user_name: str

def greet(state):
    return {"messages": [f"Hello, {state['user_name']}!"]}

def ask_question(state):
    return {"messages": ["How can I help you?"]}

graph_builder = StateGraph(ChatState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("ask", ask_question)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "ask")
graph_builder.add_edge("ask", END)
graph = graph_builder.compile()

result = graph.invoke({"messages": [], "user_name": "Carlos"})
print(result["messages"])
# Output: ['Hello, Carlos!', 'How can I help you?']

Exercise 2: State for a review system (Easy)

Design a TypedDict called ReviewState for a graph that analyzes product reviews. The graph has three nodes: one that extracts keywords, one that determines sentiment, and one that generates a summary. Decide which fields need a reducer and which don't. You don't need to implement the graph — just the state.

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class ReviewState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    keywords: Annotated[list[str], operator.add]
    sentiment: str
    sentiment_score: float
    summary: str
    review_text: str

Exercise 3: Custom reducer to cap messages (Medium)

Write a custom reducer that keeps only the last 5 messages in the list. Use it in a state and prove it works by creating a graph where 7 nodes each add a message.

from dotenv import load_dotenv
load_dotenv()

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

def keep_last_five(current: list[str], new: list[str]) -> list[str]:
    """Accumulate messages but keep only the last 5."""
    combined = current + new
    return combined[-5:]

class LimitedState(TypedDict):
    messages: Annotated[list[str], keep_last_five]
    step: int

def make_node(node_id: int):
    def node_fn(state: LimitedState) -> dict:
        return {
            "messages": [f"Message from node {node_id}"],
            "step": node_id
        }
    return node_fn

graph_builder = StateGraph(LimitedState)

for i in range(1, 8):
    graph_builder.add_node(f"node_{i}", make_node(i))

graph_builder.add_edge(START, "node_1")
for i in range(1, 7):
    graph_builder.add_edge(f"node_{i}", f"node_{i+1}")
graph_builder.add_edge("node_7", END)

graph = graph_builder.compile()
result = graph.invoke({"messages": [], "step": 0})

print(f"Total messages: {len(result['messages'])}")
for msg in result["messages"]:
    print(f"  - {msg}")
# Output:
# Total messages: 5
#   - Message from node 3
#   - Message from node 4
#   - Message from node 5
#   - Message from node 6
#   - Message from node 7

Exercise 4: Migrate to MessagesState (Medium)

You have the following custom state. Refactor it so that it extends MessagesState instead of defining messages manually. Verify the behavior is identical.

from typing import TypedDict, Annotated
import operator
from langchain_core.messages import AnyMessage

class MyState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    user_name: str
    topic: str
    sources: Annotated[list[str], operator.add]

from dotenv import load_dotenv
load_dotenv()

from typing import Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, MessagesState, START, END

class MyState(MessagesState):
    user_name: str
    topic: str
    sources: Annotated[list[str], operator.add]

model = init_chat_model("openai:gpt-4.1-mini")

def greet(state: MyState) -> dict:
    return {
        "messages": [HumanMessage(content=f"Hi, I'm {state['user_name']}")],
        "sources": ["user_input"]
    }

def respond(state: MyState) -> dict:
    response = model.invoke(state["messages"])
    return {
        "messages": [response],
        "topic": "greeting",
        "sources": ["model_response"]
    }

graph_builder = StateGraph(MyState)
graph_builder.add_node("greet", greet)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "greet")
graph_builder.add_edge("greet", "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
result = graph.invoke({
    "messages": [],
    "user_name": "Ana",
    "topic": "",
    "sources": []
})

print(f"Messages: {len(result['messages'])}")
print(f"Sources: {result['sources']}")
print(f"Topic: {result['topic']}")
# Output:
# Messages: 2
# Sources: ['user_input', 'model_response']
# Topic: greeting

Exercise 5: Complex state with multiple reducers (Advanced)

Design and implement a 3-node graph (search, analyze, summarize) for a research pipeline. The state must use operator.add for sources and findings, a custom reducer for confidence (keeping the highest), and normal replacement for summary.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langchain.chat_models import init_chat_model
from langchain_core.messages import HumanMessage, AnyMessage
from langgraph.graph import StateGraph, START, END

def keep_highest(current: float, new: float) -> float:
    return max(current, new)

class ResearchState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    sources: Annotated[list[str], operator.add]
    findings: Annotated[list[str], operator.add]
    confidence: Annotated[float, keep_highest]
    summary: str
    current_step: str

model = init_chat_model("openai:gpt-4.1-mini")

def search_node(state: ResearchState) -> dict:
    topic = state["messages"][-1].content
    response = model.invoke(f"Find 2 facts about: {topic}. Separate them with '|'.")
    findings = [f.strip() for f in response.content.split("|")]
    return {
        "sources": ["web_search", "knowledge_base"],
        "findings": findings,
        "confidence": 0.6,
        "current_step": "analyze"
    }

def analyze_node(state: ResearchState) -> dict:
    findings_text = "\n".join(state["findings"])
    response = model.invoke(f"Analyze this and add one concise observation:\n{findings_text}")
    return {
        "findings": [f"Analysis: {response.content.strip()}"],
        "confidence": 0.82,
        "current_step": "summarize"
    }

def summarize_node(state: ResearchState) -> dict:
    findings_text = "\n".join(state["findings"])
    response = model.invoke(f"Summarize in one sentence:\n{findings_text}")
    return {"summary": response.content.strip(), "current_step": "done"}

graph_builder = StateGraph(ResearchState)
graph_builder.add_node("search", search_node)
graph_builder.add_node("analyze", analyze_node)
graph_builder.add_node("summarize", summarize_node)
graph_builder.add_edge(START, "search")
graph_builder.add_edge("search", "analyze")
graph_builder.add_edge("analyze", "summarize")
graph_builder.add_edge("summarize", END)
graph = graph_builder.compile()

result = graph.invoke({
    "messages": [HumanMessage(content="What is RAG?")],
    "sources": [], "findings": [], "confidence": 0.0,
    "summary": "", "current_step": "search"
})

print(f"Sources: {result['sources']}")
print(f"Findings: {len(result['findings'])} items")
print(f"Confidence (highest): {result['confidence']}")
print(f"Summary: {result['summary'][:100]}...")
# Output (example):
# Sources: ['web_search', 'knowledge_base']
# Findings: 3 items
# Confidence (highest): 0.82
# Summary: RAG combines document retrieval with text generation to...

Exercise 6: Reducer with validation (Advanced)

Create a reducer that accumulates errors but raises an exception if it goes past 3 errors (a circuit breaker). Demonstrate it with a graph that reports errors progressively.

from dotenv import load_dotenv
load_dotenv()

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

MAX_ERRORS = 3

def accumulate_with_limit(current: list[str], new: list[str]) -> list[str]:
    combined = current + new
    if len(combined) > MAX_ERRORS:
        raise ValueError(f"Too many errors ({len(combined)}). Maximum: {MAX_ERRORS}")
    return combined

class RobustState(TypedDict):
    results: Annotated[list[str], operator.add]
    errors: Annotated[list[str], accumulate_with_limit]

def step_with_error(state: RobustState) -> dict:
    return {"errors": ["Error: API timeout"]}

def step_with_two_errors(state: RobustState) -> dict:
    return {"errors": ["Error: invalid data", "Error: wrong format"]}

graph_builder = StateGraph(RobustState)
graph_builder.add_node("error1", step_with_error)
graph_builder.add_node("error2", step_with_two_errors)
graph_builder.add_node("error3", step_with_error)
graph_builder.add_edge(START, "error1")
graph_builder.add_edge("error1", "error2")
graph_builder.add_edge("error2", "error3")
graph_builder.add_edge("error3", END)
graph = graph_builder.compile()

try:
    result = graph.invoke({"results": [], "errors": []})
except ValueError as e:
    print(f"Graph halted: {e}")
# Output: Graph halted: Too many errors (4). Maximum: 3

Summary

In this capsule you learned:

• TypedDict is the foundation of state in LangGraph — it defines which fields your graph has, with explicit types
• Without Annotated, fields get replaced (the last node to write wins)
• With Annotated[list, operator.add], fields accumulate (each node appends items)
• Custom reducers give you full control over how values get combined (deduplicate, cap, validate)
• MessagesState is the prebuilt shortcut for chat applications (messages with operator.add already configured)
• State is the single source of truth — nodes should not use global variables or side effects
• Nodes return partial updates — only the fields they want to modify
• Designing state well is designing your graph well — invest time here before writing nodes

Next capsule: Compilation and Execution — how to compile your graph, run it with invoke and stream, and visualize it with draw_mermaid_png().

Additional resources

1. State Management — LangGraph Docs — Official reference on how state works in LangGraph
2. Reducers — LangGraph Docs — Detailed documentation on reducers and Annotated
3. MessagesState — LangGraph API — Reference for the prebuilt chat state
4. TypedDict — Python Docs — Official TypedDict documentation
5. Annotated — Python Docs — Official Annotated documentation
6. operator — Python Docs — Reference for the operator module, including operator.add
7. How to define graph state — Step-by-step tutorial for defining state
8. LangGraph Quick Start — Official introductory tutorial with state examples

Module 5 — LangChain & LangGraph: From Chains to Agents