Checkpointing: Save, Resume, Time-Travel

Checkpointing saves a GraphMoment after every node transition. Resume interrupted conversations, time-travel to any past state, or fork conversations.

Concepts

Checkpointing relies on a few key identifiers to track and manage conversation state across sessions. Understanding these concepts is essential for enabling resume, replay, and time-travel capabilities.

• graph_id : A unique version identifier for your graph topology, set in GraphConfig.
• thread_id : A conversation session, automatically resolved from (graph_id, user_id) when you call compile().
• user_id : Passed via user_config during compile(). Links a user to their conversation thread so they can resume/replay where they left off.
• moment_id : It is the id of moment saved during checkpointing.

How It Works

The graph engine saves a snapshot (called a GraphMoment) after every node transition. Each moment records the current node, the full state, and a pointer to its parent moment : forming a linked timeline of the entire conversation.

Enabling Checkpointing

To enable checkpointing, pass a checkpointer backend to GraphConfig and provide a user_id in user_config when compiling the graph. The engine uses user_id and graph_id to resolve a unique thread for each user's conversation.

from videosdk.conversational_graph import GraphConfig, MongoDBSaver

config = GraphConfig(
    name="my_assistant",
    checkpointer=MongoDBSaver(uri="mongodb://localhost:27017", db_name="conversations"),
)
# user_id required for thread resolution
await graph.compile(user_config={"user_id": "user_123"})

Built-in Backends

Conversational Graph ships with several checkpointer backends out of the box, ranging from in-memory storage for quick prototyping to MongoDB and VideoSDK's managed API for production use.

Backend	Best For	Persistence
InMemorySaver()	Tests, prototyping	Lost on restart
FileSaver(directory="./checkpoints")	Local dev, debugging	JSON files on disk
MongoDBSaver(uri=..., db_name=...)	Production	MongoDB (requires motor)
VideoSDKCheckpointer()	Managed production	VideoSDK REST API (requires httpx)

Replay, Resume, Inspect

Once checkpointing is enabled, you can navigate the conversation timeline : jump back to a past state, continue from where you left off, or inspect the full history of a thread.

• replay(moment_id) : Time-travel to a past checkpoint and fork the conversation from that point.
• resume(moment_id) : Continue the conversation from the latest checkpoint.
• inspect : Inspect by fetching state and state history.

await graph.replay(moment_id="abc-123")                                   # time-travel
await graph.replay(moment_id="abc-123", update={"credit_score": 750})     # fork
await graph.resume(moment_id="abc-123")                                   # continue from latest
moment = await graph.get_state()                                           # latest moment
history = await graph.get_state_history(thread_id="t-456")                 # full timeline

replay and resume do not work without passing user_config in compile()

await graph.compile(user_config={"user_id":"testuser"})

Writing a Custom Checkpointer

If the built-in backends don't fit your infrastructure, you can implement your own by extending BaseCheckpointer. Just override the required async methods for storing, retrieving, and managing moments.

from videosdk.conversational_graph import BaseCheckpointer, GraphMoment

class RedisCheckpointer(BaseCheckpointer):
    async def put(self, moment: GraphMoment) -> None: ...
    async def get(self, thread_id, moment_id=None, *, before=None, after=None) -> Optional[GraphMoment]: ...
    async def get_history(self, thread_id) -> List[GraphMoment]: ...
    async def delete(self, thread_id) -> None: ...
    async def get_or_create_thread(self, user_id, graph_id) -> str: ...