Checkpoints¶

CheckpointStore protocol¶

CheckpointStore ¶

Bases: Protocol

Checkpoint¶

Checkpoint `dataclass` ¶

Snapshot of agent state at a point in time.

make_checkpoint¶

make_checkpoint ¶

make_checkpoint(state: dict[str, Any], trajectory_steps: list[Step], id: str | None = None, run_id: str | None = None) -> Checkpoint

Convenience constructor. Generates a UUID id if not supplied.

InMemoryCheckpointStore¶

InMemoryCheckpointStore ¶

Default in-memory implementation.

Guards its dict with an anyio.Lock so concurrent save/load/ latest calls — e.g. from multiple concurrent Agent.run() calls sharing this store — don't race on the underlying dict. This only serializes access to this store's dict; it does not make checkpoint semantics (like "rollback to the latest checkpoint") safe under concurrent writers racing to decide what "latest" means for their own recovery.

SQLiteCheckpointStore¶

Requires pip install triage-agent[sqlite].

SQLiteCheckpointStore ¶

Persistent CheckpointStore backed by SQLite.

Pass a file path for durable storage, or use a shared-memory URI for testing::

store = SQLiteCheckpointStore("runs/checkpoints.db")

Not safe for concurrent writes from multiple processes.

RedisCheckpointStore¶

Requires pip install triage-agent[redis].

RedisCheckpointStore ¶

Distributed CheckpointStore backed by Redis.

Pass a pre-configured redis.asyncio.Redis client::

import redis.asyncio as aioredis
client = aioredis.Redis.from_url("redis://localhost:6379")
store = RedisCheckpointStore(client)

The client is the caller's responsibility to close. save is atomic: checkpoint data and the timestamp index are written in a single pipeline transaction.

Conceptual notes¶

Auto-checkpointing¶

Pass auto_checkpoint=True to save a checkpoint after every record_step() call. The ROLLBACK action then restores from the most recent checkpoint in the current run:

agent = triage.Agent(
    my_agent,
    policy=policy,
    checkpoint_store=SQLiteCheckpointStore("prod.db"),
    auto_checkpoint=True,
)

Manual checkpointing¶

Call update_state to accumulate state and let auto-checkpoint persist it, or save explicitly at key points:

async def my_agent(task: str, *, record_step, update_state, **kwargs) -> str:
    data = await fetch(task)
    record_step(Step(index=0, action="fetch", tool_output=data))
    update_state({"data": data, "step": 0})  # saved into the next checkpoint
    return process(data)

Custom store¶

Implement the CheckpointStore protocol to use any backend:

class MyStore:
    async def save(self, checkpoint: Checkpoint) -> None: ...
    async def load(self, checkpoint_id: str) -> Checkpoint: ...
    async def latest(self, run_id: str | None = None) -> Checkpoint | None: ...