Nodes: Functions That Transform State

Capsule overview

In the previous capsule you built your first graph with StateGraph: you defined typed state with TypedDict and Annotated, connected START and END, compiled and ran it. Now you're going to deeply understand the most important component of any graph: nodes.

A node is a regular Python function. It receives the graph's full state, does work (calls a model, runs a tool, classifies text, transforms data), and returns a dictionary with only the fields that changed. That's the entire contract. No special classes, no inheritance, no mandatory decorators.

This concept is radically different from create_agent, where the framework decides internally which functions to run and in what order. With nodes in StateGraph, you define every step of the workflow. Each node is a visible, testable, controllable piece of your flowchart.

What is a node?

A node is a function that follows a simple contract:

1. Receives the graph's full state (a TypedDict)
2. Does work (any Python logic)
3. Returns a dictionary with only the fields it wants to update

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

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

def greet(state: State) -> dict:
    last_msg = state["messages"][-1].content
    return {"messages": [AIMessage(content=f"Hi! I got your message: '{last_msg}'")]}

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

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [HumanMessage(content="Good morning")]})
print(result["messages"][-1].content)
# Expected output: Hi! I got your message: 'Good morning'

The greet function receives all the state, but returns only what changed: a new message that gets appended to the list thanks to the operator.add reducer.

The contract: full state in, partial update out

Let's look at the contract with a state that has multiple fields:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    user_name: str
    message_count: int

def extract_name(state: State) -> dict:
    first_msg = state["messages"][0].content
    name = first_msg.split("I'm ")[-1].strip(".!") if "i'm " in first_msg.lower() else "User"
    return {"user_name": name}

def count_messages(state: State) -> dict:
    return {"message_count": len(state["messages"])}

def respond(state: State) -> dict:
    name = state.get("user_name", "User")
    count = state.get("message_count", 0)
    return {"messages": [AIMessage(content=f"Hi {name}! You're at {count} message(s).")]}

graph_builder = StateGraph(State)
graph_builder.add_node("extract_name", extract_name)
graph_builder.add_node("count_messages", count_messages)
graph_builder.add_node("respond", respond)

graph_builder.add_edge(START, "extract_name")
graph_builder.add_edge("extract_name", "count_messages")
graph_builder.add_edge("count_messages", "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({
    "messages": [HumanMessage(content="Hi, I'm Ana")],
    "user_name": "",
    "message_count": 0,
})
print(result["messages"][-1].content)
# Expected output: Hi Ana! You're at 1 message(s).

Each node updates only the fields it's responsible for:

• extract_name returns {"user_name": name} — it doesn't touch messages or message_count
• count_messages returns {"message_count": ...} — it doesn't touch the other fields
• respond returns {"messages": [...]} — it doesn't touch user_name or message_count

add_node() and naming best practices

add_node() registers a function as a node with a unique name:

graph_builder.add_node("node_name", my_function)

Nodes that call the model

The most common use case: a node that sends the messages to the LLM and appends the answer:

from dotenv import load_dotenv
load_dotenv()

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

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

def chatbot(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = model.invoke(state["messages"])
    return {"messages": [response]}

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

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [HumanMessage(content="What is LangGraph?")]})
print(result["messages"][-1].content)
# Expected output: LangGraph is an orchestration framework for building
# applications with LLMs using state graphs...

Thanks to operator.add, the answer is appended to the list instead of overwriting it.

Nodes that run tools

A node can run tools based on what the model asked for:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage
from langchain_core.tools import tool
from IPython.display import Image, display

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is sunny, 22°C"

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

def call_model(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    model_with_tools = model.bind_tools([get_weather])
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def run_tools(state: State) -> dict:
    last_message = state["messages"][-1]
    tool_map = {"get_weather": get_weather}
    results = []
    for tc in last_message.tool_calls:
        result = tool_map[tc["name"]].invoke(tc["args"])
        results.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))
    return {"messages": results}

graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("run_tools", run_tools)
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "run_tools")
graph_builder.add_edge("run_tools", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [HumanMessage(content="What's the weather in Madrid?")]})
for msg in result["messages"]:
    print(f"{type(msg).__name__}: {msg.content[:80] if msg.content else msg.tool_calls}")
# Expected output:
# HumanMessage: What's the weather in Madrid?
# AIMessage: [{'name': 'get_weather', 'args': {'city': 'Madrid'}, ...}]
# ToolMessage: The weather in Madrid is sunny, 22°C

Two specialized nodes: call_model handles the LLM, run_tools handles running the tools. Each with a clear responsibility.

Pure-logic nodes

Not every node needs a model. You can create classification, validation, or transformation nodes with pure Python:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    intent: str
    language: str

def classify_intent(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    if any(w in text for w in ["code", "program", "function", "script"]):
        return {"intent": "code"}
    elif any(w in text for w in ["story", "tale", "creative"]):
        return {"intent": "creative"}
    return {"intent": "qa"}

def detect_language(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    spanish_words = ["hola", "qué", "cómo", "por", "para", "el", "la"]
    is_spanish = sum(1 for w in text.split() if w in spanish_words) >= 2
    return {"language": "es" if is_spanish else "en"}

def respond(state: State) -> dict:
    return {"messages": [AIMessage(
        content=f"Intent: {state.get('intent', 'qa')} | Language: {state.get('language', 'es')}"
    )]}

graph_builder = StateGraph(State)
graph_builder.add_node("classify_intent", classify_intent)
graph_builder.add_node("detect_language", detect_language)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "classify_intent")
graph_builder.add_edge("classify_intent", "detect_language")
graph_builder.add_edge("detect_language", "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({
    "messages": [HumanMessage(content="How do I write a function in Python?")],
    "intent": "", "language": "",
})
print(result["messages"][-1].content)
# Expected output: Intent: code | Language: en

Pure-logic nodes are fast (they don't call any API), deterministic, and easy to test.

The edges define the execution order

A common mistake: thinking nodes run in the order you register them with add_node(). The order is defined by the edges:

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

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

def step_a(state: State) -> dict:
    return {"messages": [AIMessage(content="Step A executed")]}

def step_b(state: State) -> dict:
    return {"messages": [AIMessage(content="Step B executed")]}

def step_c(state: State) -> dict:
    return {"messages": [AIMessage(content="Step C executed")]}

graph_builder = StateGraph(State)
graph_builder.add_node("step_c", step_c)  # Registered first...
graph_builder.add_node("step_a", step_a)
graph_builder.add_node("step_b", step_b)

graph_builder.add_edge(START, "step_a")    # ...but the edges drive execution
graph_builder.add_edge("step_a", "step_b")
graph_builder.add_edge("step_b", "step_c")
graph_builder.add_edge("step_c", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [HumanMessage(content="start")]})
for msg in result["messages"][1:]:
    print(msg.content)
# Expected output:
# Step A executed
# Step B executed
# Step C executed

Even though step_c was registered first, it runs last because the edges say: A → B → C.

Async nodes

For I/O operations (API calls, databases), use async def:

from dotenv import load_dotenv
load_dotenv()

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

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

async def async_chatbot(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    response = await model.ainvoke(state["messages"])
    return {"messages": [response]}

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

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = await graph.ainvoke({"messages": [HumanMessage(content="What is async?")]})
print(result["messages"][-1].content)
# Expected output: Async in Python lets you run I/O operations
# concurrently using async/await...

Comparison: nodes in StateGraph vs functions in create_agent

from dotenv import load_dotenv
load_dotenv()

from langchain.agents import create_agent
from langchain_core.tools import tool
from IPython.display import Image, display

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"Sunny, 22°C in {city}"

agent = create_agent("openai:gpt-4.1-mini", tools=[get_weather])
display(Image(agent.get_graph().draw_mermaid_png()))

create_agent generates "agent" and "tools" nodes wired together automatically. You didn't pick those names or define those functions. Convenient for 80% of cases, but when you need custom nodes, you need StateGraph.

Troubleshooting

Problem 1: "The node doesn't update the state"

Symptom: A state field doesn't change after running. Cause: The node returns an empty dictionary, or one without the expected field. Fix: Check that the returned dict includes the right keys.

Problem 2: "Messages get overwritten instead of accumulating"

Symptom: After each node, only one message remains. Cause: The operator.add reducer is missing from the state. Fix:

# ❌ Without a reducer — overwrites
class State(TypedDict):
    messages: list[AnyMessage]

# ✅ With a reducer — accumulates
class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]

Problem 3: "InvalidUpdateError: Expected dict, got NoneType"

Symptom: An error because a node returns None. Cause: The function has no explicit return. Fix: Always return a dictionary, even an empty one: return {}

Problem 4: "TypeError when accessing state['messages'][-1]"

Symptom: An error because the list is empty. Fix: Check before accessing:

def safe_node(state: State) -> dict:
    if not state["messages"]:
        return {"messages": [AIMessage(content="No previous messages")]}
    last = state["messages"][-1]
    return {"messages": [AIMessage(content=f"Processed: {last.content}")]}

Exercises

Exercise 1: Transformation node (Easy)

Create a graph with a single node "uppercase" that takes the user's last message and converts it to uppercase. Visualize it with draw_mermaid_png().

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

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

def uppercase(state: State) -> dict:
    original = state["messages"][-1].content
    return {"messages": [AIMessage(content=original.upper())]}

graph_builder = StateGraph(State)
graph_builder.add_node("uppercase", uppercase)
graph_builder.add_edge(START, "uppercase")
graph_builder.add_edge("uppercase", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({"messages": [HumanMessage(content="hello world")]})
print(result["messages"][-1].content)
# Expected output: HELLO WORLD

Exercise 2: Two-node pipeline with validation (Easy)

Create a graph with "validate" (checks the message has at least 5 characters, stores is_valid in state) and "respond" (generates a response based on validity). Visualize the graph.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    is_valid: bool

def validate(state: State) -> dict:
    return {"is_valid": len(state["messages"][-1].content) >= 5}

def respond(state: State) -> dict:
    if state["is_valid"]:
        return {"messages": [AIMessage(content="Valid message. Processing...")]}
    return {"messages": [AIMessage(content="Message too short. Minimum 5 characters.")]}

graph_builder = StateGraph(State)
graph_builder.add_node("validate", validate)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "validate")
graph_builder.add_edge("validate", "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

for msg in ["Hi", "Hello, I need help with my project"]:
    result = graph.invoke({"messages": [HumanMessage(content=msg)], "is_valid": False})
    print(f"'{msg}' → {result['messages'][-1].content}")
# Expected output:
# 'Hi' → Message too short. Minimum 5 characters.
# 'Hello, I need help with my project' → Valid message. Processing...

Exercise 3: Node with a model and a dynamic system prompt (Medium)

Create a graph with an "expert" node that uses init_chat_model with a system prompt customized by a topic field in the state. Visualize the graph.

from dotenv import load_dotenv
load_dotenv()

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

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

def expert(state: State) -> dict:
    topic = state.get("topic", "programming")
    system = SystemMessage(content=f"You are an expert in {topic}. Answer concisely in English.")
    model = init_chat_model("openai:gpt-4.1-mini")
    return {"messages": [model.invoke([system] + state["messages"])]}

graph_builder = StateGraph(State)
graph_builder.add_node("expert", expert)
graph_builder.add_edge(START, "expert")
graph_builder.add_edge("expert", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

result = graph.invoke({
    "messages": [HumanMessage(content="What is a decorator?")],
    "topic": "Python",
})
print(result["messages"][-1].content)
# Expected output: A decorator in Python is a function that takes another function
# and extends its behavior without modifying it directly...

Exercise 4: classify → enrich → respond pipeline (Medium)

Create a 3-node graph: "classify" (detects whether it's a question, a greeting, or a command), "enrich" (adds routing metadata to the state), and "respond" (generates a response using the metadata). Visualize the graph.

from dotenv import load_dotenv
load_dotenv()

from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain_core.messages import AnyMessage, HumanMessage, AIMessage
from IPython.display import Image, display

class State(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]
    msg_type: str
    route_info: str

def classify(state: State) -> dict:
    text = state["messages"][-1].content.lower()
    if text.startswith(("what", "how", "when", "where", "why")):
        return {"msg_type": "question"}
    elif any(w in text for w in ["hello", "good morning", "hey"]):
        return {"msg_type": "greeting"}
    return {"msg_type": "command"}

def enrich(state: State) -> dict:
    info = {"question": "Q&A module", "greeting": "welcome module", "command": "execution module"}
    return {"route_info": info.get(state["msg_type"], "unknown")}

def respond(state: State) -> dict:
    responses = {
        "question": "Looking into your question...",
        "greeting": "Hello! How can I help you?",
        "command": "Processing your command...",
    }
    text = responses.get(state["msg_type"], "I didn't get that.")
    return {"messages": [AIMessage(content=f"{text} [Routing: {state['route_info']}]")]}

graph_builder = StateGraph(State)
graph_builder.add_node("classify", classify)
graph_builder.add_node("enrich", enrich)
graph_builder.add_node("respond", respond)
graph_builder.add_edge(START, "classify")
graph_builder.add_edge("classify", "enrich")
graph_builder.add_edge("enrich", "respond")
graph_builder.add_edge("respond", END)

graph = graph_builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

for msg in ["How does Python work?", "Hello world", "Run the script"]:
    result = graph.invoke({"messages": [HumanMessage(content=msg)], "msg_type": "", "route_info": ""})
    print(f"'{msg}' → {result['messages'][-1].content}")
# Expected output:
# 'How does Python work?' → Looking into your question... [Routing: Q&A module]
# 'Hello world' → Hello! How can I help you? [Routing: welcome module]
# 'Run the script' → Processing your command... [Routing: execution module]

Exercise 5: Compare create_agent vs StateGraph (Advanced)

Solve the same problem with both approaches: an assistant with a get_time tool. First with create_agent, then with StateGraph. Visualize both graphs and compare the structure.

from dotenv import load_dotenv
load_dotenv()

from langchain_core.tools import tool
from IPython.display import Image, display

@tool
def get_time() -> str:
    """Get the current time."""
    from datetime import datetime
    return datetime.now().strftime("%H:%M:%S")

# --- Approach 1: create_agent ---
from langchain.agents import create_agent

agent = create_agent("openai:gpt-4.1-mini", tools=[get_time])
result_agent = agent.invoke({"messages": [("user", "What time is it?")]})
print("=== create_agent ===")
print(result_agent["messages"][-1].content)
display(Image(agent.get_graph().draw_mermaid_png()))

# --- Approach 2: StateGraph ---
from typing import TypedDict, Annotated
import operator
from langgraph.graph import StateGraph, START, END
from langchain.chat_models import init_chat_model
from langchain_core.messages import AnyMessage, HumanMessage, ToolMessage

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

def call_model(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    return {"messages": [model.bind_tools([get_time]).invoke(state["messages"])]}

def run_tools(state: State) -> dict:
    results = []
    for tc in state["messages"][-1].tool_calls:
        output = get_time.invoke(tc["args"])
        results.append(ToolMessage(content=str(output), tool_call_id=tc["id"]))
    return {"messages": results}

def final_response(state: State) -> dict:
    model = init_chat_model("openai:gpt-4.1-mini")
    return {"messages": [model.invoke(state["messages"])]}

graph_builder = StateGraph(State)
graph_builder.add_node("call_model", call_model)
graph_builder.add_node("run_tools", run_tools)
graph_builder.add_node("final_response", final_response)
graph_builder.add_edge(START, "call_model")
graph_builder.add_edge("call_model", "run_tools")
graph_builder.add_edge("run_tools", "final_response")
graph_builder.add_edge("final_response", END)

graph = graph_builder.compile()
print("\n=== StateGraph ===")
result_graph = graph.invoke({"messages": [HumanMessage(content="What time is it?")]})
print(result_graph["messages"][-1].content)
display(Image(graph.get_graph().draw_mermaid_png()))
# Expected output: Both answer with the current time.
# create_agent has a loop (conditional edge).
# StateGraph is linear (no conditional edges yet).

Summary

In this capsule you learned:

• A node is a Python function that receives the full state and returns a dictionary with only the fields that changed (a partial update)
• graph_builder.add_node("name", function) registers a function as a node of the graph
• Nodes can do any work: call the model, run tools, classify, validate, transform
• The execution order is defined by the edges, not by the order of add_node()
• Async nodes (async def) are useful for I/O operations and web apps
• In create_agent, the nodes are opaque and predefined; in StateGraph, you define and control them completely
• draw_mermaid_png() is your main tool for checking the graph's structure

Next capsule: Edges and Conditional Edges — you'll learn to connect nodes with fixed and conditional edges to create workflows that branch based on dynamic decisions.

Additional resources

1. LangGraph Nodes — Conceptual Guide — Official documentation on nodes in LangGraph
2. How to create a StateGraph — Step-by-step tutorial with StateGraph
3. LangGraph State Management — How typed state and reducers work
4. Visualization with draw_mermaid_png — Graph visualization guide
5. create_agent API Reference — Reference for comparing against the high-level API
6. LangGraph Async Support — Async support in LangGraph
7. TypedDict Documentation — Python — TypedDict reference for designing state

Module 5 — LangChain & LangGraph: From Chains to Agents