Checkpointing and Threads

By default every app.invoke call is stateless — each call starts with an empty AgentState. A checkpointer adds persistence: state is saved after each node run and restored at the start of the next call for the same thread.

The config dict

Every checkpointer method accepts a config dict. At minimum it must contain a thread_id:

config = {
    "thread_id": "conv-abc123",   # required — identifies the conversation
    "user_id": "user-42",         # optional — used by PgCheckpointer for row scoping
}

thread_id is the primary key for all state, message, and thread-info operations. user_id scopes data within a multi-tenant deployment and is used by PgCheckpointer when you query threads for a specific user.

Thread lifecycle

flowchart LR
  NewCall[invoke with thread_id] --> Check{Thread exists?}
  Check -->|no| CreateState[Create fresh AgentState]
  Check -->|yes| LoadState[Load saved AgentState]
  CreateState --> RunGraph[Run graph nodes]
  LoadState --> RunGraph
  RunGraph --> SaveState[Save updated AgentState]
  SaveState --> Return[Return result to caller]

A thread is created on the first call and persists until you explicitly delete it. For InMemoryCheckpointer data is lost on process restart; PgCheckpointer survives restarts.

Attaching a checkpointer

Pass the checkpointer to graph.compile():

from tenxgraph.storage.checkpointer import InMemoryCheckpointer
from tenxgraph.core.state import AgentState, Message

checkpointer = InMemoryCheckpointer()
app = graph.compile(checkpointer=checkpointer)

Provide thread_id in every call. The first argument is always a dict with a "messages" key — not an AgentState object:

config = {"thread_id": "session-1", "recursion_limit": 10}

# Turn 1
res = app.invoke(
    {"messages": [Message.text_message("What is 2 + 2?")]},
    config=config,
)

# Turn 2 — same thread_id, resumes from saved state
res = app.invoke(
    {"messages": [Message.text_message("Now multiply that by 3.")]},
    config=config,
)

InMemoryCheckpointer

InMemoryCheckpointer stores everything in Python dicts guarded by asyncio.Lock objects. It is async-first — all operations are non-blocking coroutines. Sync wrappers (put_state, get_state, …) run the coroutines via run_coroutine().

from tenxgraph.storage.checkpointer import InMemoryCheckpointer

checkpointer = InMemoryCheckpointer()

Internal storage:

Thread key is simply str(thread_id) — there is no user scoping in the in-memory implementation.

Use for: development, unit tests, and short-lived single-process deployments.
Do not use for: anything that requires state to survive a restart or be shared across multiple workers.

PgCheckpointer

PgCheckpointer uses PostgreSQL for durable storage and Redis for a low-latency read cache. It is designed for production multi-worker deployments.

Constructor signature

from tenxgraph.storage.checkpointer import PgCheckpointer

checkpointer = PgCheckpointer(
    # --- PostgreSQL connection (pick one) ---
    postgres_dsn="postgresql://user:pass@localhost/mydb",  # DSN string
    # OR
    pg_pool=existing_asyncpg_pool,                         # pass an existing pool
    # OR
    pool_config={"min_size": 2, "max_size": 10},           # kwargs forwarded to asyncpg.create_pool

    # --- Redis connection (pick one, optional) ---
    redis_url="redis://localhost:6379/0",
    # OR
    redis=existing_redis_instance,
    # OR
    redis_pool=existing_connection_pool,
    # OR
    redis_pool_config={"max_connections": 20},

    # --- Optional tuning ---
    schema="public",          # PostgreSQL schema (default: "public")
    cache_ttl=86400,          # Redis TTL in seconds (default: 24 h)
    user_id_type="string",    # Column type for user_id: "string" | "int" | "bigint"
    release_resources=True,   # Close pools on cleanup
)

Schema migration

On the first call to await checkpointer.asetup() (or checkpointer.setup()) the checkpointer:

1. Checks if the tables already exist.
2. Runs any pending migration steps to bring the schema to the current version.
3. Creates tables if they do not exist.

You must call setup() once before using the checkpointer:

await checkpointer.asetup()
app = graph.compile(checkpointer=checkpointer)

Installing the extra

pip install "10xgraph[pg_checkpoint]"

This installs asyncpg and redis[asyncio].

SqliteCheckpointer

SqliteCheckpointer collapses the two-layer model into a single local SQLite file. Durable state, the realtime state cache, messages, threads, and the generic cache all live in one .db file — there is no Redis and no Postgres. I/O is fully async via aiosqlite; a single persistent connection runs in WAL mode with writes serialized behind an asyncio.Lock.

from tenxgraph.storage.checkpointer import SqliteCheckpointer

# Defaults to ~/.10xgraph/checkpointer.db; ":memory:" gives an ephemeral DB.
checkpointer = SqliteCheckpointer("agent_state.db")
app = graph.compile(checkpointer=checkpointer)

Like PgCheckpointer, it reads the state cache table first and falls back to durable state, and it reconstructs custom AgentState subclasses on read via an embedded class path. Data is keyed purely by thread_id; user_id is not required.

Use it for client-side / single-user agents — a desktop app shipping a Python sidecar (Tauri, Electron, PyInstaller), a local CLI agent, or any deployment where each user has their own process and their own database file (one writer per file). Avoid it for shared multi-user servers: SQLite serializes writers and does not scale horizontally — use PgCheckpointer instead.

Install the extra:

pip install "10xgraph[sqlite_checkpoint]"

Call await checkpointer.arelease() (or checkpointer.release()) at shutdown to close the connection.

Checkpointer API reference

All methods exist in both async (preferred) and sync (convenience wrapper) flavours. The sync wrappers call run_coroutine() internally.

State methods

The state cache is a thin Redis-backed read layer in PgCheckpointer (TTL-controlled). It avoids a Postgres round-trip on repeated reads for the same thread. For InMemoryCheckpointer both _states and _state_cache live in memory.

Message methods

Checkpointers also store the full message history for a thread independently of the compressed state:

Thread methods

Thread metadata (ThreadInfo) is stored separately from state:

Generic cache methods

An optional key-value cache layer for arbitrary JSON values (e.g., pre-computed embeddings or rendered prompts):

# Store a value with a 10-minute TTL
await checkpointer.aput_cache_value("llm-cache", "prompt-hash-abc", {"response": "..."}, ttl_seconds=600)

# Retrieve it
value = await checkpointer.aget_cache_value("llm-cache", "prompt-hash-abc")

# Delete it
await checkpointer.aclear_cache_value("llm-cache", "prompt-hash-abc")

# List all keys in a namespace
keys = await checkpointer.alist_cache_keys("llm-cache", prefix="prompt-")

InMemoryCheckpointer supports TTL on the generic cache via expiry timestamps. PgCheckpointer uses Redis SETEX / GET / DEL / SCAN for these operations.

Direct checkpointer usage (outside the graph)

You can interact with a checkpointer directly — useful for admin scripts, migrations, or custom REST endpoints:

from tenxgraph.storage.checkpointer import InMemoryCheckpointer
from tenxgraph.core.state import AgentState, Message

cp = InMemoryCheckpointer()
config = {"thread_id": "t1", "user_id": "u1"}

# Save state manually
state = AgentState(context=[Message.text_message("Hello", role="user")])
await cp.aput_state(config, state)

# Load it back
loaded = await cp.aget_state(config)

# List all messages
messages = await cp.alist_messages(config, limit=50)

# Clean the thread
await cp.aclean_thread(config)

Reading thread state via REST

When running behind the API server the same data is exposed over HTTP:

# Read current state
GET /v1/threads/{thread_id}/state

# Overwrite state
PUT /v1/threads/{thread_id}/state

# Read message history  
GET /v1/threads/{thread_id}/messages

# List threads for a user
GET /v1/threads

See REST API: Threads for request/response schemas.

Using a checkpointer with the API server

The API server uses the checkpointer the compiled graph carries, so pass it to compile() in the module that agent points at:

# graph/react.py
from graph.dependencies import my_checkpointer

app = state_graph.compile(checkpointer=my_checkpointer)

{
  "agent": "graph.react:app"
}

10xgraph.json also recognises a checkpointer key, but the server does not apply it yet, so setting it has no effect. See Configure 10xgraph.json.

Choosing a checkpointer

Related concepts

• Memory and store
• State and messages
• REST API: Threads