Utilities

Supporting classes for checkpoint message export and key management.

MessageExporter

class MessageExporter(redis_saver, recipe=None)

Bases: object

Export messages from Redis checkpoints.

Parameters:

• redis_saver (Any)

• recipe (MessageRecipe | None)

export(thread_id, checkpoint_id=None)

Export messages from checkpoint data.

Parameters:

• thread_id (str) – The conversation thread ID

• checkpoint_id (str | None) – Specific checkpoint ID (latest if None)

Returns:

List of extracted message dictionaries

Return type:

List[Dict[str, Any]]

export_thread(thread_id)

Export all messages from all checkpoints in a thread.

Parameters:

thread_id (str) – The conversation thread ID

Returns:

Dict with thread_id, messages, and export timestamp

Return type:

Dict[str, Any]

class LangChainRecipe

Bases: object

Default recipe for extracting LangChain messages.

extract(message)

Extract data from LangChain message objects.

Parameters:

message (Any)

Return type:

Dict[str, Any] | None

class MessageRecipe(*args, **kwargs)

Bases: Protocol

Protocol for message extraction recipes.

Implement this interface to support custom message formats.

extract(message)

Extract structured data from a message.

Parameters:

message (Any) – The message to extract data from.

Returns:

Dict with at least ‘role’ and ‘content’ keys, or None if message cannot be extracted.

Return type:

Dict[str, Any] | None

CheckpointKeyRegistry

class CheckpointKeyRegistry

Bases: object

Base class for checkpoint-based key registry using sorted sets.

static make_write_keys_zset_key(thread_id, checkpoint_ns, checkpoint_id)

Create the key for the write keys sorted set for a specific checkpoint.

Parameters:

• thread_id (str) – Thread identifier

• checkpoint_ns (str) – Checkpoint namespace (will be converted to storage-safe format)

• checkpoint_id (str) – Checkpoint identifier (will be converted to storage-safe format)

Returns:

The Redis key for the write keys sorted set

Return type:

str