Quick Start

Checkpointing

pydantic-deep supports conversation checkpointing — saving snapshots of conversation state at key points so the agent (or user) can rewind to any checkpoint or fork from a checkpoint into a new session.

Quick Start

from pydantic_deep import create_deep_agent, InMemoryCheckpointStore

store = InMemoryCheckpointStore()
agent = create_deep_agent(
    include_checkpoints=True,
    checkpoint_store=store,
    checkpoint_frequency="every_tool",  # auto-save after each tool call
)

The agent now has three tools:

Configuration

create_deep_agent() Parameters

Checkpoint Frequencies

=== "every_tool" Saves a checkpoint after each tool call. Provides the finest granularity for rewinding.

```python
agent = create_deep_agent(
    include_checkpoints=True,
    checkpoint_frequency="every_tool",
)
```

=== "every_turn" Saves a checkpoint before each model request (i.e. at the start of each turn).

```python
agent = create_deep_agent(
    include_checkpoints=True,
    checkpoint_frequency="every_turn",
)
```

=== "manual_only" No auto-checkpoints. The agent must explicitly call save_checkpoint.

```python
agent = create_deep_agent(
    include_checkpoints=True,
    checkpoint_frequency="manual_only",
)
```

Checkpoint Stores

InMemoryCheckpointStore

Default store. Checkpoints are kept in a Python dict — fast, but lost when the process exits.

from pydantic_deep import InMemoryCheckpointStore

store = InMemoryCheckpointStore()

FileCheckpointStore

Persists checkpoints as JSON files on disk. Each checkpoint is a separate file in the given directory.

from pydantic_deep import FileCheckpointStore

store = FileCheckpointStore("/path/to/checkpoints")

Messages are serialized using pydantic-ai's ModelMessagesTypeAdapter, so all message types (text, tool calls, tool results) are preserved.

Custom Store

Implement the [CheckpointStore][pydantic_deep.toolsets.checkpointing.CheckpointStore] protocol:

from pydantic_deep.toolsets.checkpointing import Checkpoint, CheckpointStore

class RedisCheckpointStore:
    async def save(self, checkpoint: Checkpoint) -> None: ...
    async def get(self, checkpoint_id: str) -> Checkpoint | None: ...
    async def get_by_label(self, label: str) -> Checkpoint | None: ...
    async def list_all(self) -> list[Checkpoint]: ...
    async def remove(self, checkpoint_id: str) -> bool: ...
    async def remove_oldest(self) -> bool: ...
    async def count(self) -> int: ...
    async def clear(self) -> None: ...

Rewinding

When the agent calls rewind_to, a [RewindRequested][pydantic_deep.toolsets.checkpointing.RewindRequested] exception propagates out of agent.run(). Your application loop catches it and restores the conversation:

from pydantic_deep import create_deep_agent, DeepAgentDeps, RewindRequested
from pydantic_ai_backends import StateBackend

agent = create_deep_agent(include_checkpoints=True)
deps = DeepAgentDeps(backend=StateBackend())
message_history = []

while True:
    user_input = input("> ")
    try:
        result = await agent.run(
            user_input,
            deps=deps,
            message_history=message_history,
        )
        message_history = result.all_messages()
        print(result.output)
    except RewindRequested as e:
        print(f"Rewinding to '{e.label}'...")
        message_history = e.messages

Session Forking

Fork creates a new session from a checkpoint — the original session stays unchanged.

from pydantic_deep import fork_from_checkpoint, InMemoryCheckpointStore

store = InMemoryCheckpointStore()
# ... after some conversation with checkpoints saved ...

# Fork from a specific checkpoint
messages = await fork_from_checkpoint(store, checkpoint_id="abc123")

# Start a new session with the forked history
new_result = await agent.run(
    "Try a different approach",
    deps=new_deps,
    message_history=messages,
)

Per-Session Stores

For multi-user applications, set checkpoint_store on deps at runtime:

from pydantic_deep import DeepAgentDeps, FileCheckpointStore

# Each user gets their own checkpoint store
deps = DeepAgentDeps(
    backend=backend,
    checkpoint_store=FileCheckpointStore(f"/checkpoints/{user_id}"),
)

The middleware resolves the store from deps.checkpoint_store first, falling back to the store passed at agent creation time.

!!! warning "Multi-User Applications" Always create a fresh CheckpointStore per user session. A shared InMemoryCheckpointStore means all users see each other's checkpoints. See Multi-User Guide.

Components

Next Steps

• Teams — Multi-agent collaboration with shared state
• Hooks — Claude Code-style lifecycle hooks
• Human-in-the-Loop — Approval workflows