LangGraph Architecture Decisions

When to Use LangGraph

Use LangGraph When You Need:

• Stateful conversations - Multi-turn interactions with memory

• Human-in-the-loop - Approval gates, corrections, interventions

• Complex control flow - Loops, branches, conditional routing

• Multi-agent coordination - Multiple LLMs working together

• Persistence - Resume from checkpoints, time travel debugging

• Streaming - Real-time token streaming, progress updates

• Reliability - Retries, error recovery, durability guarantees

Consider Alternatives When:

Scenario Alternative Why

Single LLM call Direct API call Overhead not justified

Linear pipeline LangChain LCEL Simpler abstraction

Stateless tool use Function calling No persistence needed

Simple RAG LangChain retrievers Built-in patterns

Batch processing Async tasks Different execution model

State Schema Decisions

TypedDict vs Pydantic

TypedDict Pydantic

Lightweight, faster Runtime validation

Dict-like access Attribute access

No validation overhead Type coercion

Simpler serialization Complex nested models

Recommendation: Use TypedDict for most cases. Use Pydantic when you need validation or complex nested structures.

Reducer Selection

Use Case Reducer Example

Chat messages add_messages

Handles IDs, RemoveMessage

Simple append operator.add

Annotated[list, operator.add]

Keep latest None (LastValue) field: str

Custom merge Lambda Annotated[list, lambda a, b: ...]

Overwrite list Overwrite

Bypass reducer

State Size Considerations

SMALL STATE (< 1MB) - Put in state

class State(TypedDict): messages: Annotated[list, add_messages] context: str

LARGE DATA - Use Store

class State(TypedDict): messages: Annotated[list, add_messages] document_ref: str # Reference to store

def node(state, *, store: BaseStore): doc = store.get(namespace, state["document_ref"]) # Process without bloating checkpoints

Graph Structure Decisions

Single Graph vs Subgraphs

Single Graph when:

• All nodes share the same state schema

• Simple linear or branching flow

• < 10 nodes

Subgraphs when:

• Different state schemas needed

• Reusable components across graphs

• Team separation of concerns

• Complex hierarchical workflows

Conditional Edges vs Command

Conditional Edges Command

Routing based on state Routing + state update

Separate router function Decision in node

Clearer visualization More flexible

Standard patterns Dynamic destinations

Conditional Edge - when routing is the focus

def router(state) -> Literal["a", "b"]: return "a" if condition else "b" builder.add_conditional_edges("node", router)

Command - when combining routing with updates

def node(state) -> Command: return Command(goto="next", update={"step": state["step"] + 1})

Static vs Dynamic Routing

Static Edges (add_edge ):

• Fixed flow known at build time

• Clearer graph visualization

• Easier to reason about

Dynamic Routing (add_conditional_edges , Command , Send ):

• Runtime decisions based on state

• Agent-driven navigation

• Fan-out patterns

Persistence Strategy

Checkpointer Selection

Checkpointer Use Case Characteristics

InMemorySaver

Testing only Lost on restart

SqliteSaver

Development Single file, local

PostgresSaver

Production Scalable, concurrent

Custom Special needs Implement BaseCheckpointSaver

Checkpointing Scope

Full persistence (default)

graph = builder.compile(checkpointer=checkpointer)

Subgraph options

subgraph = sub_builder.compile( checkpointer=None, # Inherit from parent checkpointer=True, # Independent checkpointing checkpointer=False, # No checkpointing (runs atomically) )

When to Disable Checkpointing

• Short-lived subgraphs that should be atomic

• Subgraphs with incompatible state schemas

• Performance-critical paths without need for resume

Multi-Agent Architecture

Supervisor Pattern

Best for:

• Clear hierarchy

• Centralized decision making

• Different agent specializations

      ┌─────────────┐
      │  Supervisor │
      └──────┬──────┘
  

  ┌────────┬───┴───┬────────┐ ▼ ▼ ▼ ▼ ┌──────┐ ┌──────┐ ┌──────┐ ┌──────┐ │Agent1│ │Agent2│ │Agent3│ │Agent4│ └──────┘ └──────┘ └──────┘ └──────┘

Peer-to-Peer Pattern

Best for:

• Collaborative agents

• No clear hierarchy

• Flexible communication

┌──────┐ ┌──────┐ │Agent1│◄───►│Agent2│ └──┬───┘ └───┬──┘ │ │ ▼ ▼ ┌──────┐ ┌──────┐ │Agent3│◄───►│Agent4│ └──────┘ └──────┘

Handoff Pattern

Best for:

• Sequential specialization

• Clear stage transitions

• Different capabilities per stage

┌────────┐ ┌────────┐ ┌────────┐ │Research│───►│Planning│───►│Execute │ └────────┘ └────────┘ └────────┘

Streaming Strategy

Stream Mode Selection

Mode Use Case Data

updates

UI updates Node outputs only

values

State inspection Full state each step

messages

Chat UX LLM tokens

custom

Progress/logs Your data via StreamWriter

debug

Debugging Tasks + checkpoints

Subgraph Streaming

Stream from subgraphs

async for chunk in graph.astream( input, stream_mode="updates", subgraphs=True # Include subgraph events ): namespace, data = chunk # namespace indicates depth

Human-in-the-Loop Design

Interrupt Placement

Strategy Use Case

interrupt_before

Approval before action

interrupt_after

Review after completion

interrupt() in node Dynamic, contextual pauses

Resume Patterns

Simple resume (same thread)

graph.invoke(None, config)

Resume with value

graph.invoke(Command(resume="approved"), config)

Resume specific interrupt

graph.invoke(Command(resume={interrupt_id: value}), config)

Modify state and resume

graph.update_state(config, {"field": "new_value"}) graph.invoke(None, config)

Error Handling Strategy

Retry Configuration

Per-node retry

RetryPolicy( initial_interval=0.5, backoff_factor=2.0, max_interval=60.0, max_attempts=3, retry_on=lambda e: isinstance(e, (APIError, TimeoutError)) )

Multiple policies (first match wins)

builder.add_node("node", fn, retry_policy=[ RetryPolicy(retry_on=RateLimitError, max_attempts=5), RetryPolicy(retry_on=Exception, max_attempts=2), ])

Fallback Patterns

def node_with_fallback(state): try: return primary_operation(state) except PrimaryError: return fallback_operation(state)

Or use conditional edges for complex fallback routing

def route_on_error(state) -> Literal["retry", "fallback", "end"]: if state.get("error") and state["attempts"] < 3: return "retry" elif state.get("error"): return "fallback" return END

Scaling Considerations

Horizontal Scaling

• Use PostgresSaver for shared state

• Consider LangGraph Platform for managed infrastructure

• Use stores for large data outside checkpoints

Performance Optimization

• Minimize state size - Use references for large data

• Parallel nodes - Fan out when possible

• Cache expensive operations - Use CachePolicy

• Async everywhere - Use ainvoke, astream

Resource Limits

Set recursion limit

config = {"recursion_limit": 50} graph.invoke(input, config)

Track remaining steps in state

class State(TypedDict): remaining_steps: RemainingSteps

def check_budget(state): if state["remaining_steps"] < 5: return "wrap_up" return "continue"

Decision Checklist

Before implementing:

• Is LangGraph the right tool? (vs simpler alternatives)

• State schema defined with appropriate reducers?

• Persistence strategy chosen? (dev vs prod checkpointer)

• Streaming needs identified?

• Human-in-the-loop points defined?

• Error handling and retry strategy?

• Multi-agent coordination pattern? (if applicable)

• Resource limits configured?