---
title: "Part 3: Scaling LangGraph: State Persistence, Checkpointing, and Parallelism"
description: "Long-running multi-agent workflows require durable state persistence and concurrency management. Here is how to configure SQLite and Redis checkpointers alongside Pregel super-step parallel branching in LangGraph."
date: "May 03, 2024"
url: "https://himanshuat.com/blogs/scaling-langgraph-persistence-and-parallelism"
---
# Part 3: Scaling LangGraph: State Persistence, Checkpointing, and Parallelism

An agent process throws midway through execution. Every message it accumulated and every intermediate reasoning step it produced lived in memory, so all of it goes with the process.

> A graph without a checkpointer is not resumable. One exception discards every super-step that already succeeded, and a retry pays for all of them again.

A stateless workflow that restarts looks like it recovered. It started over. Anything you run in production needs checkpoint persistence, and once you have that, you can also run branches concurrently.

### Chapter 0: Checkpointing mechanisms in graph runtimes

A **checkpointer** serializes the entire state of a graph to external storage (such as SQLite, Redis, or PostgreSQL) after each super-step. Stored data includes accumulated message lists, custom state variables, node execution markers, and routing metadata.

To resume execution after an interruption or process restart, the graph loads the state snapshot associated with a `thread_id` and continues execution from the last completed node.

---

### Step 1: Local persistence with SqliteSaver

For single-instance deployments or local testing, `SqliteSaver` serializes state transitions directly into SQLite tables.

The `thread_id` is the primary lookup key for state snapshots:

```python
import os
from typing import TypedDict, List
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.sqlite import SqliteSaver

class GraphState(TypedDict):
    messages: List[str]
    turn: int

def chat_node(state: GraphState) -> dict:
    current_turn = state.get("turn", 0)
    new_message = f"Agent turn {current_turn}: Step complete."
    return {
        "messages": state.get("messages", []) + [new_message], 
        "turn": current_turn + 1
    }

memory = SqliteSaver.from_conn_string(":memory:")

workflow = StateGraph(GraphState)
workflow.add_node("chat", chat_node)
workflow.set_entry_point("chat")
workflow.add_edge("chat", "chat")

app = workflow.compile(checkpointer=memory)

thread_id = "session_001"
initial_state = {"messages": ["User: Run initial batch."], "turn": 0}

print(f"Executing thread '{thread_id}':")
for i in range(2):
    output = app.invoke(
        input=initial_state if i == 0 else {},
        config={"configurable": {"thread_id": thread_id}}
    )
    print(output['messages'][-1])

print("\nResuming from saved checkpoint:")
output_resumed = app.invoke(
    input={},
    config={"configurable": {"thread_id": thread_id}}
)
print("Resumed output:", output_resumed['messages'][-1])
print("Total history length:", len(output_resumed['messages']))
```

Invoking the compiled graph with an empty input dictionary and an existing `thread_id` automatically retrieves the latest checkpoint from SQLite and executes the next step.

---

### Step 2: Distributed persistence with RedisSaver

In distributed environments where multiple worker nodes process tasks concurrently, single-file SQLite databases create file-lock contention.

`RedisSaver` moves state snapshots into a networked in-memory key-value store, so any available worker can pick up a suspended thread:

```python
import os
import redis
from typing import TypedDict, List
from langgraph.graph import StateGraph
from langgraph.checkpoint.redis import RedisSaver

class GraphState(TypedDict):
    messages: List[str]
    turn: int

def chat_node(state: GraphState) -> dict:
    current_turn = state.get("turn", 0)
    new_message = f"Turn {current_turn} processed via Redis."
    return {
        "messages": state.get("messages", []) + [new_message], 
        "turn": current_turn + 1
    }

# Connect to Redis
redis_url = os.getenv("REDIS_URL", "redis://localhost:6379/0")
memory = RedisSaver.from_conn_info(host="localhost", port=6379, db=0)

workflow = StateGraph(GraphState)
workflow.add_node("chat", chat_node)
workflow.set_entry_point("chat")
workflow.add_edge("chat", "chat")

app_redis = workflow.compile(checkpointer=memory)

thread_id_redis = "distributed_thread_42"
initial_state = {"messages": ["Init distributed task."], "turn": 0}

output = app_redis.invoke(
    input=initial_state,
    config={"configurable": {"thread_id": thread_id_redis}}
)
print("First turn output:", output['messages'][-1])
```

| Checkpointer | Storage engine | Concurrency model | Best fit |
|---|---|---|---|
| `SqliteSaver` | Local SQLite file / memory | Single-process only | Local development and tests |
| `RedisSaver` | Redis cluster | Multi-process networked workers | Low-latency distributed agents |
| `PostgresSaver` | PostgreSQL tables | Transactional ACID storage | High-durability enterprise workflows |

---

### Step 3: Branch parallelism via Pregel super-steps

LangGraph executes graphs using the Pregel computational model. Computations progress through synchronized super-steps:

1. In each super-step, all nodes whose inbound edge dependencies are met execute concurrently.
2. Nodes emit state updates and pass messages to downstream targets.
3. The super-step completes when all active nodes finish their execution cycle.

To execute tasks in parallel, define multiple outgoing edges from a single parent node without adding edges between the sibling workers:

```mermaid
flowchart LR
  P[plan_topics] --> A[research_ethics]
  P --> B[research_quantum]
  P --> C[research_neuroscience]
  A --> W[compile_report]
  B --> W
  C --> W
  W --> E[END]
```

```python
from typing import List, Literal, TypedDict
from langgraph.graph import StateGraph, END
import time

class MultiAgentState(TypedDict):
    research_topics: List[str]
    report_sections: List[str]
    final_report: str
    current_step: Literal["plan", "research", "write", "finish"]

def plan_topics(state: MultiAgentState) -> dict:
    topics = ["AI Ethics", "Quantum Computing", "Neuroscience in AI"]
    print("Planner assigned topics:", topics)
    return {"research_topics": topics, "current_step": "research"}

def research_ai_ethics(state: MultiAgentState) -> dict:
    time.sleep(0.1)
    section = "AI Ethics: Evaluated algorithmic transparency and alignment."
    return {"report_sections": state.get("report_sections", []) + [section]}

def research_quantum_computing(state: MultiAgentState) -> dict:
    time.sleep(0.08)
    section = "Quantum Computing: Analyzed qubit coherence algorithms."
    return {"report_sections": state.get("report_sections", []) + [section]}

def research_neuroscience_ai(state: MultiAgentState) -> dict:
    time.sleep(0.12)
    section = "Neuroscience in AI: Modeled sparse activation graphs."
    return {"report_sections": state.get("report_sections", []) + [section]}

def compile_report(state: MultiAgentState) -> dict:
    sections = "\n".join(state.get("report_sections", []))
    report = f"Consolidated Findings:\n{sections}"
    return {"final_report": report, "current_step": "finish"}

workflow_parallel = StateGraph(MultiAgentState)

workflow_parallel.add_node("plan", plan_topics)
workflow_parallel.add_node("research_ethics", research_ai_ethics)
workflow_parallel.add_node("research_quantum", research_quantum_computing)
workflow_parallel.add_node("research_neuroscience", research_neuroscience_ai)
workflow_parallel.add_node("compile_report", compile_report)

workflow_parallel.set_entry_point("plan")

# Fan-out to independent research nodes
workflow_parallel.add_edge("plan", "research_ethics")
workflow_parallel.add_edge("plan", "research_quantum")
workflow_parallel.add_edge("plan", "research_neuroscience")

# Fan-in synchronization barrier: compile_report waits for all three nodes
workflow_parallel.add_edge("research_ethics", "compile_report")
workflow_parallel.add_edge("research_quantum", "compile_report")
workflow_parallel.add_edge("research_neuroscience", "compile_report")

workflow_parallel.add_edge("compile_report", END)

app_parallel = workflow_parallel.compile()
```

The runtime executes `research_ethics`, `research_quantum`, and `research_neuroscience` in parallel within the same super-step, waiting for all three to complete before triggering `compile_report`.

---

### Step 4: Node-level async concurrency

Pregel coordinates high-level graph topology. However, I/O-bound tasks within individual nodes (such as querying multiple APIs or embedding text batches) should use native asynchronous Python concurrency:

```python
import asyncio
import httpx

async def fetch_api_data(endpoints: List[str]) -> List[dict]:
    """Fetches multiple external endpoints concurrently using httpx."""
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url, timeout=10.0) for url in endpoints]
        responses = await asyncio.gather(*tasks, return_exceptions=True)
        
        valid_results = []
        for r in responses:
            if isinstance(r, httpx.Response) and r.status_code == 200:
                valid_results.append(r.json())
        return valid_results
```

Using `asyncio.gather` inside node definitions prevents network round trips from blocking the overall Pregel event loop.

---

## When this is the wrong choice

- **The whole run is one request and cheap to repeat.** A checkpointer writes the full graph state to storage after every super-step. If re-running the workflow costs one model call, those writes buy you nothing and add a storage dependency to a process that had none.
- **The concurrency you need lives inside a single node.** Splitting three API calls into three graph nodes gets you the same wall-clock time as `asyncio.gather` in one node, plus three checkpoint writes and a fan-in barrier. Fan out in the graph only when the branches are separately resumable.
- **The branches depend on each other.** A super-step runs nodes whose inbound edges are satisfied. If the quantum research needs the ethics section first, the barrier serializes them anyway, and you have written a parallel-looking graph that executes sequentially.
- **You are on one process and staying there.** `SqliteSaver` covers this. Redis and PostgreSQL exist for multiple workers contending on the same threads; adopting them earlier adds an operational service to keep alive for no concurrency you actually have.

---

### Architecture rules for scaled graphs

1. **State payload optimization:** avoid storing raw binary files or extensive document corpora in checkpoint stores. Store pointers (such as S3 URIs or database primary keys) to prevent storage bloat.
2. **Deterministic reducer policies:** when multiple parallel nodes write to the same list field within a super-step, use explicit reducer functions to prevent race conditions.
3. **Appropriate store selection:** use SQLite for local unit testing, Redis for low-latency session caching, and PostgreSQL for long-term auditable checkpoints.

---

Source: https://himanshuat.com/blogs/scaling-langgraph-persistence-and-parallelism