/langgraph-architecture

name: langgraph-architecture
description: Guides architectural decisions for LangGraph applications. Use when deciding between LangGraph vs alternatives, choosing state management strategies, designing multi-agent systems, or selecting persistence and streaming approaches.

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,
    strea