How To Checkpointing

How-To: Checkpointing & State Persistence

AgentMesh integrates with memory and database checkpointers for persistence and session management.

---

In-Memory Checkpointing

from langgraph.checkpoint.memory import MemorySaver
import agentmesh_adapter

saver = MemorySaver()
app = agentmesh_adapter.compile(builder, checkpointer=saver)

config = {"configurable": {"thread_id": "thread-42"}}

# First Turn
app.invoke({"messages": ["Hello"]}, config=config)

# Second Turn (loads previous state automatically)
app.invoke({"messages": ["What did I say?"]}, config=config)

# Inspect Checkpoint
current_state = app.get_state(config)
print("Persisted State:", current_state)

---

PostgreSQL Crash Recovery Engine

For multi-node production persistence, AgentMesh C++ engine includes PostgresStateRepository backed by libpq:

• Transactional write-ahead snapshots after every wave.
• Crash recovery resumes orphaned workflows in <4.8 ms.
• Zero in-flight data loss on SIGKILL.\n