Skip to main content

checkpoints.store

Checkpoint store protocol and implementations.

Provides in-memory and file-based stores for conversation checkpoints.

RewindRequested Objects​

class RewindRequested(Exception)

Raised when the agent requests a rewind to a checkpoint.

The controlling application code should catch this, restore the message history from checkpoint, and restart the agent loop.

ConversationCheckpoint Objects​

@dataclass
class ConversationCheckpoint()

Snapshot of conversation state at a point in time.

created_at​

ISO format

CheckpointStore Objects​

class CheckpointStore(ABC)

Abstract checkpoint storage backend.

save​

@abstractmethod
async def save(checkpoint: ConversationCheckpoint) -> None

Save a checkpoint.

get​

@abstractmethod
async def get(checkpoint_id: str) -> ConversationCheckpoint | None

Retrieve a checkpoint by ID.

list_all​

@abstractmethod
async def list_all() -> list[ConversationCheckpoint]

List all checkpoints, newest first.

delete​

@abstractmethod
async def delete(checkpoint_id: str) -> None

Delete a checkpoint.

prune​

@abstractmethod
async def prune(max_count: int) -> int

Remove oldest checkpoints to keep at most max_count. Returns number removed.

create_checkpoint​

async def create_checkpoint(
        label: str,
        turn: int,
        messages: list[dict[str, Any]],
        max_checkpoints: int = 20,
        metadata: dict[str, Any] | None = None) -> ConversationCheckpoint

Create, save, and prune in one step.

InMemoryCheckpointStore Objects​

class InMemoryCheckpointStore(CheckpointStore)

In-memory checkpoint store. Data lost on process restart.

FileCheckpointStore Objects​

class FileCheckpointStore(CheckpointStore)

File-based checkpoint store. Persists to JSON files.

create_checkpoint_store​

def create_checkpoint_store(store_type: str = "in_memory",
                            **kwargs: Any) -> CheckpointStore

Factory to create a checkpoint store from config.