• Parallel Execution
• Reducer Functions
• State Schema and Channels
• The Messages Channel
• State Types: Pydantic, Dataclasses, TypedDict

So, what exactly is State? LangGraph defines it as:

“A shared data structure that represents the current snapshot of your application. It can be any Python type, but is typically a TypedDict or Pydantic BaseModel.”

To expand on this: State is the mechanism you use as a developer to track what your agents are doing. More specifically, it’s how you observe and control the evolution of data throughout your graph execution.

Now, lets cover a few basics.

Parallel Execution

We don’t need to address it in too much detail here, but it is very easy to create a graph where two nodes are executed in parallel, for instance the following code snippet has two nodes, a and b, which are both executed at the same time.

When nodes are executed in parallel, issues arise when both nodes try to update the State at the same time.

from typing_extensions import TypedDict, Annotated
from langgraph.graph import START, StateGraph

class State(TypedDict):
    node_calls: list = []

builder = StateGraph(State)

def node_a(state: State):
    return {"node_calls": ['node_a']}

def node_b(state: State):
    return {"node_calls": ['node_b']}

builder.add_node("a", node_a)
builder.add_node("b", node_b)

builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()

graph_response = graph.invoke({"input":"10"})

If you run the above snippet you will get the following error:

InvalidUpdateError: At key 'node_calls': Can receive only one value per step. Use an Annotated key to handle multiple values.
For troubleshooting, visit: https://python.langchain.com/docs/troubleshooting/errors/INVALID_CONCURRENT_GRAPH_UPDATE_called': Can receive only one value per step. Use an Annotated key to handle multiple values.

This is because, for the first step, which executes both nodes in parallel we are trying to write to the same property at the same time. Now in reality the functions which are nodes a and b actually don’t execute at the exact same time, but we consider them the same step in the graph execution, so we need to treat them as writing at the same time. This depicts the issue:

Reducer Functions

Reducer functions to the rescue! They are used to solve this problem, and they are used to tell LangGraph how you would like information added to the state in case of parallel execution. For example, you can do this:

def reducer(input_list:list, list_to_add:list) -> list:
    return input_list + list_to_add

class State(TypedDict):
    node_calls: Annotated[list, reducer]

builder = StateGraph(State)

def node_a(state: State):
    return {"node_calls": ['node_a']}

def node_b(state: State):
    return {"node_calls": ['node_b']}

builder.add_node("a", node_a)
builder.add_node("b", node_b)

builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()
graph_response = graph.invoke({"input":"10"})
graph_response

Which will print out this:

{'node_calls': ['node_a', 'node_b']}

To make this easier, you can use the add operator like so.

class State(TypedDict):
    node_calls: Annotated[list, operator.add]

State Schema & Channels

The state schema defines the structure of the state object that your LangGraph workflow will use. Each schema consists of what LangGraph calls channels.

Channels are the individual fields (or keys) in the state schema that can hold data. These channels can be read from or written to by nodes in your graph, depending on how you configure each node’s inputs and outputs.

This is simply the terminology LangGraph uses to describe common Python constructs. The state schema itself is typically defined using a TypedDict, dataclass, or Pydantic model.

In practical terms, channels are just the keys or attributes in those Python objects.

# this is the state schema
class State(TypedDict):
    # this is a channel called node_calls
    node_calls: Annotated[list, operator.add]

Messages Channel

A common channel or key that is used in LangGraph State is the Messages channel. If you are unfamiliar with messages within LangChain/LangGraph, I suggest you take a look at this:

https://python.langchain.com/docs/concepts/messages/

This message will be used to store the messages from models output like so:

class State(TypedDict):
    node_calls: Annotated[list, operator.add]
    messages: Annotated[list, operator.add]

builder = StateGraph(State)

def node_a(state: State):
    model_response = model.invoke("Is a channel in LangGraph State really just a key?")
    return {"node_calls": ['node_a'], "messages": [model_response]}

def node_b(state: State):
    model_response = model.invoke("Are reducer functions in LangGraph required because of the parallel execution in a super step?")
    return {"node_calls": ['node_b'], "messages": [model_response]}

builder.add_node("a", node_a)
builder.add_node("b", node_b)

builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()
graph_response = graph.invoke({"input":"10"})
graph_response

{ 'messages': [ AIMessage(content='Yes, in LangGraph, a channel is essentially just a key in the state dictionary. When you define a channel with `add_node` or through other methods, you\'re specifying a key that will be used to store and retrieve data in the graph\'s state.\n\nFor example, when you define something like:\n\n```python\n@graph.node\ndef process_data(state):\n    # Do something with state["input_data"]\n    return {"processed_result": result}\n```\n\nThe "processed_result" is a channel (key) in the state dictionary. When you connect nodes together, you\'re establishing a flow of data where the output channel from one node becomes the input channel for another node.\n\nThe channel concept helps with organizing the flow of data through the graph, but fundamentally it\'s implemented as a key in the state dictionary that gets passed between nodes in the execution flow.', additional_kwargs={}, response_metadata={'id': 'msg_01XX8c1ybrB3xxy69ipaGP3k', 'model': 'claude-3-7-sonnet-20250219', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'input_tokens': 20, 'output_tokens': 193, 'server_tool_use': None, 'service_tier': 'standard'}, 'model_name': 'claude-3-7-sonnet-20250219'}, id='run--f46501a6-ff96-4a5b-88d7-b02d47871175-0', usage_metadata={'input_tokens': 20, 'output_tokens': 193, 'total_tokens': 213, 'input_token_details': {'cache_read': 0, 'cache_creation': 0}}),
                AIMessage(content='# Reducer Functions in LangGraph and Parallel Execution\n\nYes, reducer functions in LangGraph are specifically designed to handle the challenge of aggregating results from parallel operations during a super step in the model\'s Pregel-inspired execution pattern.\n\n## Why Reducers Are Needed\n\nWhen a node in a LangGraph workflow fans out to process multiple branches in parallel (such as processing multiple agents or documents simultaneously), you need a way to combine these parallel results back into a coherent state. This is precisely what reducer functions do.\n\nThe key reasons reducers are required:\n\n1. **Parallel Execution Aggregation**: They combine the outputs from parallel branches that execute during the same super step\n\n2. **State Management**: They ensure the graph\'s state remains consistent after parallel operations\n\n3. **Data Transformation**: They transform multiple outputs into a format suitable for downstream processing\n\n## Example Use Case\n\nConsider a multi-agent system where you send the same query to several specialist agents in parallel:\n\n```python\n@graph(name="multi_agent_system")\ndef multi_agent_workflow():\n    query = MessageState()\n    \n    # Fan out to multiple agents in parallel\n    agent_responses = query.map(process_with_agents)\n    \n    # Need a reducer to combine all agent responses\n    final_answer = agent_responses.reduce(combine_agent_responses)\n    \n    return final_answer\n```\n\nWithout a reducer, there would be no defined way to reconcile the multiple parallel outputs back into a single coherent state for the workflow to continue.\n\n## Reducer Implementation\n\nA reducer typically takes a list of outputs from parallel branches and returns a single combined result:\n\n```python\ndef combine_agent_responses(responses: List[str]) -> str:\n    """Combine multiple agent responses into a single coherent answer."""\n    # Logic to synthesize, prioritize, or otherwise combine responses\n    return synthesized_answer\n```\n\nThis pattern is fundamental to the Pregel computational model that LangGraph builds upon for distributed processing of graph-structured workflows.', additional_kwargs={}, response_metadata={'id': 'msg_012FKSFKLU4KQjk6XUtSiCVL', 'model': 'claude-3-7-sonnet-20250219', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'input_tokens': 25, 'output_tokens': 453, 'server_tool_use': None, 'service_tier': 'standard'}, 'model_name': 'claude-3-7-sonnet-20250219'}, id='run--3d261f69-b45c-42c5-99b6-b3a184eed049-0', usage_metadata={'input_tokens': 25, 'output_tokens': 453, 'total_tokens': 478, 'input_token_details': {'cache_read': 0, 'cache_creation': 0}})],
  'node_calls': ['node_a', 'node_b']}

LangGraph has provided an easier way to do this than manually adding a channel and a reducer function. You can do this instead:

from langgraph.graph import MessagesState

class State(TypedDict, MessagesState):
    node_calls: Annotated[list, operator.add]

builder = StateGraph(State)

def node_a(state: State):
    model_response = model.invoke("Is a channel in LangGraph State really just a key?")
    return {"node_calls": ['node_a'], "messages": [model_response]}

def node_b(state: State):
    model_response = model.invoke("Are reducer functions in LangGraph required because of the parallel execution in a super step?")
    return {"node_calls": ['node_b'], "messages": [model_response]}

builder.add_node("a", node_a)
builder.add_node("b", node_b)

builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()
graph_response = graph.invoke({"input":"10"})

State Types: Pydantic, Dataclasses, TypedDict

You can inherit from Pydantic models to form your State class. This is helpful if you want to use features built in to Pydantic like data validation. For instance:

class State(BaseModel):
    input: str
    node_calls: Annotated[list, operator.add]

builder = StateGraph(State)

def node_a(state: State):
    model_response = model.invoke("Is a channel in LangGraph State really just a key?")
    return {"node_calls": ['node_a'], "messages": [model_response]}

def node_b(state: State):
    model_response = model.invoke("Are reducer functions in LangGraph required because of the parallel execution in a super step?")
    return {"node_calls": ['node_b'], "messages": [model_response]}

builder.add_node("a", node_a)
builder.add_node("b", node_b)

builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()
graph_response = graph.invoke({"input":"10","node_calls":[]})
pprint.pprint(graph_response)

{'input': '10', 'node_calls': ['node_a', 'node_b']}

But, if I pass 10 as a int — I’ll get a error:

graph_response = graph.invoke({"input":10,"node_calls":[]})

ValidationError: 1 validation error for State
input
  Input should be a valid string [type=string_type, input_value=10, input_type=int]
    For further information visit https://errors.pydantic.dev/2.11/v/string_type
Before task with name 'a' and path '('__pregel_pull', 'a')'

Now, dealing with messages is a little bit different with Pydantic, you can’t use the MessageState class shown above, it is not compatible at least with Pydantic’s BaseModel. There are a few different ways to deal with this, but one way is to just create an annotated list and use AnyMessage, with the out of box reducer for messages LangGraph gives you, add_mesages.

from langchain_core.messages import AnyMessage
from langgraph.graph.message import add_messages

class State(BaseModel):
    input: str
    messages: Annotated[list[AnyMessage], add_messages]
    node_calls: Annotated[list, operator.add]

builder = StateGraph(State)

def node_a(state: State):
    model_response = model.invoke("Is a channel in LangGraph State really just a key?")
    return {"node_calls": ['node_a'], "messages": [model_response]}

def node_b(state: State):
    model_response = model.invoke("Are reducer functions in LangGraph required because of the parallel execution in a super step?")
    return {"node_calls": ['node_b'], "messages": [model_response]}

builder.add_node("a", node_a)
builder.add_node("b", node_b)

builder.add_edge(START, "a")
builder.add_edge(START, "b")

graph = builder.compile()
graph_response = graph.invoke({"input":"10","node_calls":[]})
pprint.pprint(graph_response)

{'input': '10',
 'messages': [AIMessage(content='Yes, in LangGraph, a channel in the State object is essentially just a key in a dictionary-like structure.\n\nWhen you define a State class with channels, like:\n\n```python\nclass MyState(State):\n    my_channel: list[str] = []\n    another_channel: str = ""\n```\n\nYou\'re basically creating a structured dictionary where "my_channel" and "another_channel" are keys that can be accessed to retrieve or update values.\n\nThe State object provides a convenient API to:\n1. Access these channels (via dot notation or dict-like syntax)\n2. Update them in an immutable manner with methods like `set`\n3. Keep track of changes in a way that works with LangGraph\'s state management\n\nSo while channels provide structure and type hints for your workflow state, at their core, they are indeed just keys in a state dictionary that LangGraph manages for you.', additional_kwargs={}, response_metadata={'id': 'msg_019hKgST9BoEUB5ELq5KY4BT', 'model': 'claude-3-7-sonnet-20250219', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'input_tokens': 20, 'output_tokens': 201, 'server_tool_use': None, 'service_tier': 'standard'}, 'model_name': 'claude-3-7-sonnet-20250219'}, id='run--21b1f29b-d0bb-4145-b171-f0b5363e199a-0', usage_metadata={'input_tokens': 20, 'output_tokens': 201, 'total_tokens': 221, 'input_token_details': {'cache_read': 0, 'cache_creation': 0}}),
              AIMessage(content="## Are Reducer Functions in LangGraph Required Due to Parallel Execution?\n\nYes, that's exactly right. Reducer functions in LangGraph are primarily needed because of the parallel execution model in supersteps.\n\nIn LangGraph's implementation of the Bulk Synchronous Parallel (BSP) model:\n\n1. During a superstep, multiple operations can execute in parallel\n2. These parallel operations all work on their own separate copies of state\n3. After the parallel execution phase, the results need to be combined back together\n\nThe reducer function serves this critical role of merging the results from parallel executions into a consistent state. Without reducers, there would be no defined way to reconcile potentially conflicting state changes from parallel operations.\n\nThis is particularly important for:\n- Graph operations where multiple nodes might update simultaneously\n- Maintaining consistency when multiple agents operate independently\n- Ensuring deterministic behavior despite parallel execution\n\nIf you were using LangGraph without parallel execution, the reducer functions would be much less important since there wouldn't be multiple state versions to combine.", additional_kwargs={}, response_metadata={'id': 'msg_015kGCaGmJNjCnGA3uisw9qU', 'model': 'claude-3-7-sonnet-20250219', 'stop_reason': 'end_turn', 'stop_sequence': None, 'usage': {'cache_creation_input_tokens': 0, 'cache_read_input_tokens': 0, 'input_tokens': 25, 'output_tokens': 230, 'server_tool_use': None, 'service_tier': 'standard'}, 'model_name': 'claude-3-7-sonnet-20250219'}, id='run--5833a444-80f2-4f2c-b57a-7a802a01ab5d-0', usage_metadata={'input_tokens': 25, 'output_tokens': 230, 'total_tokens': 255, 'input_token_details': {'cache_read': 0, 'cache_creation': 0}})],
 'node_calls': ['node_a', 'node_b']}

Dataclasses

You can also use Python built in dataclasses. They are faster than Pydantic, but won’t perform the same type of data validation. For instance:

import dataclasses

@dataclasses.dataclass
class State():
    input: str
    messages: Annotated[list[AnyMessage], add_messages]
    node_calls: Annotated[list, operator.add]

"""
Graph building code
"""

graph_response = graph.invoke({"input":10,"node_calls":[]})
pprint.pprint(graph_response)

This works, it will return the same result as above, but you can see that input is passed as an int, when we’ve used the str type hint above.

Well, that covers the basics of State. I’ll post a few more articles which will delve deeper into state.