Recap — Making Agents Resumable and Stateful

# Making Agents Resumable and Stateful LangGraph saves state at every node transition through **checkpointers**. With checkpointing, agents become: - Resumable (continue from where you stopped). - Multi-turn (state persists across user inputs). - Human-in-the-loop ready (pause, get approval, resume). - Debuggable (inspect state at any point). - Replayable (re-run from any checkpoint). ## Without checkpointing ```python graph = builder.compile() graph.invoke({"messages": [user_msg]}) # State exists only during invoke(). Returns final state. State lost. ``` Stateless. Each invocation is fresh. Fine for single-turn workflows. ## With checkpointing ```python from langgraph.checkpoint.memory import MemorySaver graph = builder.compile(checkpointer=MemorySaver()) config = {"configurable": {"thread_id": "user-123"}} graph.invoke({"messages": [user_msg]}, config=config) # State saved under thread_id "user-123" # Later (different request, same user): graph.invoke({"messages": [next_user_msg]}, config=config) # Resumes from saved state — accumulated messages, all prior decisions ``` The `thread_id` is the key. Each thread has its own state. ## Checkpointer choices ### MemorySaver In-process memory. Lost on restart. ```python from langgraph.checkpoint.memory import MemorySaver checkpointer = MemorySaver() ``` Good for: development, testing, single-process apps. ### SQLite checkpointer Persistent on disk; single-process. ```python from langgraph.checkpoint.sqlite import SqliteSaver checkpointer = SqliteSaver.from_conn_string("checkpoints.db") ``` Good for: simple production deployments, single-instance services. ### Postgres checkpointer Persistent, multi-process, production-grade. ```python from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver.from_conn_string("postgresql://...") ``` Good for: multi-instance production services, sharing state across replicas, leveraging existing Postgres. ### Redis checkpointer In-memory speed, multi-process. ```python from langgraph.checkpoint.redis import RedisSaver checkpointer = RedisSaver.from_conn_string("redis://localhost:6379") ``` Good for: high-throughput agents, short-lived state (with TTL). ### Custom checkpointer Implement the `BaseCheckpointSaver` interface for custom storage (S3, DynamoDB, etc.). ## What gets checkpointed Every state transition (node execution) creates a checkpoint with: - Full state at that point. - Which node ran. - Timestamp. - Parent checkpoint reference (for the history tree). Storage cost is real. A complex agent with 20 node transitions creates 20 checkpoints. Most checkpointers offer cleanup policies. ## Reading state from a thread ```python # Get current state of thread state = graph.get_state(config) print(state.values) # the actual state print(state.next) # next node(s) to execute (if interrupted) print(state.config) # the config including checkpoint_id # Get state history for state in graph.get_state_history(config): print(state.values["iteration"], state.next) ``` Useful for: - Building admin UIs that show conversation state. - Debugging: "what was state at step 5?" - Auditing. ## Updating state directly You can mutate state from outside the graph: ```python graph.update_state( config, {"user_preference": "concise"}, ) ``` Useful for: - Injecting external context. - Recovery from human-in-the-loop decisions. - Admin overrides. ## Replaying from a checkpoint ```python # Get a specific checkpoint all_states = list(graph.get_state_history(config)) target_checkpoint = all_states[3] # 4th state in history # Replay from there new_config = target_checkpoint.config graph.invoke(None, config=new_config) # Re-runs from checkpoint 4 onward ``` Useful for: - Debugging: re-run from where things went wrong. - A/B testing: re-run with different prompts from a specific state. - Recovery: resume after a crash. ## Threading concerns The `thread_id` is YOUR responsibility to manage. Common patterns: - **Per-user**: `thread_id=user_id`. One conversation per user. - **Per-conversation**: `thread_id=conversation_id`. User can have multiple parallel conversations. - **Per-session**: `thread_id=session_id`. Resets per session. - **Per-task**: `thread_id=task_id`. For long-running tasks. Pick based on your data model. Many apps use `thread_id=f"{user_id}:{conversation_id}"`. ## Cleaning up old checkpoints Production agents accumulate state. Add cleanup: ```python # Manually delete a thread graph.delete_thread(config) # Or use TTL-based checkpointer (Redis with expiry) checkpointer = RedisSaver.from_conn_string("redis://...", ttl=86400) # 24h TTL ``` For Postgres-backed: schedule a periodic DELETE on old checkpoints based on your retention policy. ## What checkpointing enables 1. **Multi-turn conversations.** Same thread_id across user turns. 2. **Long-running agents.** Pause overnight, resume next day. 3. **Human-in-the-loop.** Interrupt, persist, wa