Nodes, Edges & Control Flow

Overview

Nodes are the computational units in LangGraph, while edges define the execution order. Together they form a directed graph that can contain cycles for iterative workflows.

Nodes

A node is any callable that accepts state and returns a partial state update.

Function Nodes

The simplest node is a Python function:

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

class State(TypedDict):
    count: int
    message: str

def increment(state: State) -> dict:
    """A basic node that increments count."""
    return {"count": state["count"] + 1}

def format_message(state: State) -> dict:
    """Format a message based on count."""
    return {"message": f"Count is {state['count']}"}

builder = StateGraph(State)
builder.add_node("increment", increment)
builder.add_node("format", format_message)

Runnable Nodes

Any LangChain Runnable can be a node:

from langchain_core.runnables import RunnableLambda
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI

# Runnable chain as a node
chain = (
    ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
    ])
    | ChatOpenAI()
)

builder.add_node("llm", chain)

Class-Based Nodes

Callable classes work as nodes:

class ProcessorNode:
    def __init__(self, config: dict):
        self.config = config
    
    def __call__(self, state: State) -> dict:
        # Use self.config in processing
        return {"count": state["count"] * 2}

processor = ProcessorNode(config={"multiplier": 2})
builder.add_node("processor", processor)

Node Signatures

Nodes can access runtime context:

from langchain_core.runnables import RunnableConfig
from langgraph.runtime import Runtime

class Context(TypedDict):
    user_id: str

# With RunnableConfig
def node_with_config(state: State, config: RunnableConfig) -> dict:
    thread_id = config["configurable"]["thread_id"]
    return {"message": f"Thread: {thread_id}"}

# With Runtime context
def node_with_context(state: State, runtime: Runtime[Context]) -> dict:
    user_id = runtime.context["user_id"]
    return {"message": f"User: {user_id}"}

builder.add_node("node1", node_with_config)
builder.add_node("node2", node_with_context)

Async Nodes

Nodes can be async for concurrent I/O:

import asyncio

async def fetch_data(state: State) -> dict:
    # Simulate API call
    await asyncio.sleep(1)
    data = await fetch_from_api()
    return {"data": data}

builder.add_node("fetch", fetch_data)

# Use ainvoke or astream
result = await graph.ainvoke({"count": 0})

Node Metadata

Attach metadata for observability:

builder.add_node(
    "expensive_operation",
    my_node,
    metadata={
        "cost": "high",
        "timeout": 30,
        "version": "2.0"
    }
)

Edges

Edges define the execution order between nodes.

Unconditional Edges

Fixed transitions from one node to another:

builder = StateGraph(State)
builder.add_node("step1", node1)
builder.add_node("step2", node2)
builder.add_node("step3", node3)

# Linear flow
builder.add_edge(START, "step1")
builder.add_edge("step1", "step2")
builder.add_edge("step2", "step3")
builder.add_edge("step3", END)

# Or use helper methods
builder.set_entry_point("step1")  # START -> step1
builder.set_finish_point("step3")  # step3 -> END

Parallel Edges

Multiple nodes execute in parallel:

builder.add_node("fetch_user", fetch_user)
builder.add_node("fetch_orders", fetch_orders)
builder.add_node("combine", combine_results)

# Both fetch nodes run in parallel
builder.add_edge(START, "fetch_user")
builder.add_edge(START, "fetch_orders")

# Wait for both before combining
builder.add_edge(["fetch_user", "fetch_orders"], "combine")
builder.add_edge("combine", END)

Conditional Edges

Dynamic routing based on state:

def route_after_llm(state: State) -> str:
    """Route based on LLM response."""
    if "tool_call" in state["last_message"]:
        return "tools"
    return "end"

builder.add_node("llm", llm_node)
builder.add_node("tools", tool_node)

builder.add_conditional_edges(
    "llm",
    route_after_llm,
    {
        "tools": "tools",
        "end": END
    }
)

# Tool output goes back to LLM (creating a loop)
builder.add_edge("tools", "llm")

Multi-Destination Routing

Return multiple destinations to run in parallel:

from typing import Sequence

def fan_out(state: State) -> Sequence[str]:
    """Route to multiple nodes based on conditions."""
    destinations = []
    if state["needs_validation"]:
        destinations.append("validate")
    if state["needs_logging"]:
        destinations.append("log")
    return destinations

builder.add_conditional_edges(
    "prepare",
    fan_out,
    # No path_map needed - function returns node names directly
)

Type-Safe Routing with Literals

from typing import Literal

def strict_router(state: State) -> Literal["success", "failure"]:
    """Type-checked routing."""
    if state["count"] > 0:
        return "success"
    return "failure"

builder.add_conditional_edges(
    "process",
    strict_router,
    {
        "success": "handle_success",
        "failure": "handle_failure"
    }
)

The Send API

Dynamically invoke nodes with custom input (map-reduce pattern):

from langgraph.types import Send
from typing import Annotated
import operator

class OverallState(TypedDict):
    subjects: list[str]
    results: Annotated[list[str], operator.add]

class SubjectState(TypedDict):
    subject: str

def continue_to_process(state: OverallState) -> list[Send]:
    """Fan out to process each subject independently."""
    return [
        Send("process_subject", {"subject": s})
        for s in state["subjects"]
    ]

def process_subject(state: SubjectState) -> dict:
    """Process a single subject."""
    result = f"Processed {state['subject']}"
    return {"results": [result]}

builder = StateGraph(OverallState)
builder.add_node("process_subject", process_subject)
builder.add_conditional_edges(START, continue_to_process)
builder.add_edge("process_subject", END)

graph = builder.compile()

result = graph.invoke({"subjects": ["math", "science", "history"]})
# {'subjects': [...], 'results': ['Processed math', 'Processed science', 'Processed history']}

Send allows nodes to execute with different input than the main graph state, perfect for map-reduce workflows.

Command API

Return Command objects from nodes for advanced control flow:

from langgraph.types import Command

def smart_node(state: State) -> Command:
    """Node that controls its own routing."""
    # Update state and navigate
    return Command(
        update={"count": state["count"] + 1},
        goto="next_step" if state["count"] < 5 else END
    )

def multi_send_node(state: State) -> Command:
    """Send to multiple nodes with custom input."""
    return Command(
        update={"status": "processing"},
        goto=[
            Send("worker", {"task_id": 1}),
            Send("worker", {"task_id": 2}),
        ]
    )

builder.add_node("smart", smart_node)
builder.add_edge(START, "smart")
# No need to define edges from 'smart' - Command handles routing

Command Fields

Field	Type	Description
`update`	`dict \| Any`	State updates to apply
`goto`	`str \| list[str] \| Send \| list[Send]`	Next node(s) to execute
`graph`	`None \| "__parent__"`	Target graph for command
`resume`	`dict \| Any`	Resume value for interrupts

Sequences

Quickly add a chain of nodes:

def step1(state: State) -> dict:
    return {"count": 1}

def step2(state: State) -> dict:
    return {"count": state["count"] + 1}

def step3(state: State) -> dict:
    return {"count": state["count"] * 2}

builder = StateGraph(State)

# Automatically creates edges: START -> step1 -> step2 -> step3 -> END
builder.add_sequence([step1, step2, step3])
builder.set_entry_point(step1.__name__)
builder.set_finish_point(step3.__name__)

Cycles and Loops

LangGraph supports cycles for iterative workflows:

def check_completion(state: State) -> str:
    if state["count"] >= 10:
        return "done"
    return "continue"

builder = StateGraph(State)
builder.add_node("work", do_work)
builder.add_edge(START, "work")

# Create a cycle
builder.add_conditional_edges(
    "work",
    check_completion,
    {
        "continue": "work",  # Loop back
        "done": END
    }
)

Cycles can run indefinitely. Set a recursion_limit to prevent infinite loops:

graph = builder.compile()
result = graph.invoke(
    {"count": 0},
    config={"recursion_limit": 25}
)

Retry Policies

Automatically retry failed nodes:

from langgraph.types import RetryPolicy

retry_policy = RetryPolicy(
    initial_interval=0.5,    # Start with 0.5s
    backoff_factor=2.0,      # Double each retry
    max_interval=10.0,       # Cap at 10s
    max_attempts=3,          # Try 3 times
    jitter=True,             # Add randomness
    retry_on=Exception       # Retry on any exception
)

builder.add_node(
    "api_call",
    make_api_request,
    retry_policy=retry_policy
)

Multiple Retry Policies

from langgraph.types import RetryPolicy

# Different policies for different errors
retry_policies = [
    RetryPolicy(
        max_attempts=5,
        retry_on=RateLimitError,
        initial_interval=1.0
    ),
    RetryPolicy(
        max_attempts=2,
        retry_on=TimeoutError,
        initial_interval=0.1
    ),
]

builder.add_node(
    "flaky_node",
    my_node,
    retry_policy=retry_policies  # First matching policy applies
)

Caching

Cache expensive node results:

from langgraph.types import CachePolicy

cache_policy = CachePolicy(
    ttl=3600,  # 1 hour
)

builder.add_node(
    "expensive_llm_call",
    llm_node,
    cache_policy=cache_policy
)

# Must provide cache at compile time
from langgraph.cache.memory import InMemoryCache

graph = builder.compile(cache=InMemoryCache())

Cached nodes:

Skip execution on cache hit
Use input state as cache key (customizable)
Respect TTL for expiration
Store results across invocations

Deferred Nodes

Delay node execution until graph end:

builder.add_node(
    "cleanup",
    cleanup_resources,
    defer=True  # Runs after all other nodes complete
)

Use cases:

Cleanup operations
Final logging/metrics
Post-processing steps

Best Practices

Node Design

Keep nodes pure and focused on single tasks
Use type hints for state and return values
Handle errors gracefully within nodes
Return partial state updates, not full state
Make nodes testable in isolation

Control Flow

Prefer explicit edges for simple flows
Use conditional edges for dynamic routing
Use Command for complex multi-step routing
Document routing logic clearly
Test all routing paths

Performance

Parallelize independent nodes
Use async nodes for I/O operations
Cache expensive operations
Set reasonable retry limits
Monitor node execution times

Next Steps

Checkpointing

Persist state between node executions

Streaming

Stream node outputs in real-time

Human-in-the-Loop

Add human review points with interrupts

Get Started

Core Concepts

Guides

Tutorials

CLI

Nodes, Edges & Control Flow

Overview

Nodes

Function Nodes

Runnable Nodes

Class-Based Nodes

Node Signatures

Async Nodes

Node Metadata

Edges

Unconditional Edges

Parallel Edges

Conditional Edges

Multi-Destination Routing

Type-Safe Routing with Literals

The Send API

Command API

Command Fields

Sequences

Cycles and Loops

Retry Policies

Multiple Retry Policies

Caching

Deferred Nodes

Best Practices

Next Steps

Checkpointing

Streaming

Human-in-the-Loop

Get Started

Core Concepts

Guides

Tutorials

CLI

​Overview

​Nodes

​Function Nodes

​Runnable Nodes

​Class-Based Nodes

​Node Signatures

​Async Nodes

​Node Metadata

​Edges

​Unconditional Edges

​Parallel Edges

​Conditional Edges

​Multi-Destination Routing

​Type-Safe Routing with Literals

​The Send API

​Command API

​Command Fields

​Sequences

​Cycles and Loops

​Retry Policies

​Multiple Retry Policies

​Caching

​Deferred Nodes

​Best Practices

​Next Steps

Checkpointing

Streaming

Human-in-the-Loop

Overview

Nodes

Function Nodes

Runnable Nodes

Class-Based Nodes

Node Signatures

Async Nodes

Node Metadata

Edges

Unconditional Edges

Parallel Edges

Conditional Edges

Multi-Destination Routing

Type-Safe Routing with Literals

The Send API

Command API

Command Fields

Sequences

Cycles and Loops

Retry Policies

Multiple Retry Policies

Caching

Deferred Nodes

Best Practices

Next Steps