haive.core.persistence.memory

In-memory persistence implementation for the Haive framework.

This module provides an in-memory checkpointer implementation that stores state data temporarily in memory. It’s primarily intended for development, testing, and demonstration purposes where persistence across application restarts is not required.

The memory checkpointer is the simplest implementation, requiring no external dependencies or infrastructure. It supports both synchronous and asynchronous operation modes, making it suitable for a wide range of use cases where temporary state management is sufficient.

Classes

MemoryCheckpointerConfig

Configuration for in-memory checkpoint persistence.

Module Contents

class haive.core.persistence.memory.MemoryCheckpointerConfig

Bases: haive.core.persistence.base.CheckpointerConfig[dict[str, Any]]

Configuration for in-memory checkpoint persistence.

This implementation provides a simple non-persistent memory-based checkpointer suitable for development, testing, and demonstration purposes. It stores all state data in memory, which means that state is lost when the application is restarted.

The memory checkpointer is the simplest and most lightweight option, requiring no external dependencies, connection setup, or infrastructure. It’s ideal for:

• Development environments

• Testing scenarios

• Quick prototyping

• Demonstration purposes

• Stateless applications that don’t need persistence

It supports both synchronous and asynchronous operation modes, with the same underlying implementation since the memory checkpointer is inherently thread-safe.

Examples

from haive.core.persistence import MemoryCheckpointerConfig

# Create a basic memory checkpointer config = MemoryCheckpointerConfig() checkpointer = config.create_checkpointer()

# Use with a graph from langgraph.graph import Graph graph = Graph(checkpointer=checkpointer)

async create_async_checkpointer()

Create an asynchronous memory checkpointer.

For memory checkpointers, this method returns the same synchronous MemorySaver instance since it’s inherently thread-safe and can be used in asynchronous contexts without modification. This simplifies the implementation while still providing the expected interface.

The method sets the mode to ASYNC for consistency, but the underlying implementation remains the same as the synchronous version.

Returns:

A LangGraph MemorySaver instance ready for async use

Return type:

Any

Examples

config = MemoryCheckpointerConfig(mode=CheckpointerMode.ASYNC) async_checkpointer = await config.create_async_checkpointer() # Use with an async graph graph = AsyncGraph(checkpointer=async_checkpointer)

create_checkpointer()

Create a synchronous memory checkpointer.

This method creates a LangGraph MemorySaver instance that stores all state data in memory. The memory checkpointer is thread-safe and can be shared between multiple graph instances if needed.

The implementation is very lightweight, simply instantiating a MemorySaver object with no additional configuration needed.

Returns:

A LangGraph MemorySaver instance ready for use

Return type:

Any

Raises:

RuntimeError – If the MemorySaver class cannot be imported or instantiated, typically due to a missing dependency

Examples

config = MemoryCheckpointerConfig() checkpointer = config.create_checkpointer() # Use with a synchronous graph graph = Graph(checkpointer=checkpointer)

async initialize_async_checkpointer()

Initialize an async checkpointer with resource management.

For memory checkpointers, this method simply returns the checkpointer directly since there are no external resources (like database connections) to manage. Unlike database-backed implementations, the memory checkpointer doesn’t need an async context manager for resource lifecycle management.

This method exists primarily for interface compatibility with other checkpointer implementations that do require resource management.

Returns:

A LangGraph MemorySaver instance ready for async use

Return type:

Any

Examples

config = MemoryCheckpointerConfig(mode=CheckpointerMode.ASYNC) # Simple usage without context management checkpointer = await config.initialize_async_checkpointer() # Use the checkpointer…

is_async_mode()

Check if this configuration is set to operate in asynchronous mode.

This method determines whether the memory checkpointer should use asynchronous operations based on the configured mode. For memory checkpointers, this is primarily a compatibility feature since the underlying implementation is the same for both modes.

Returns:

True if configured for async operations, False for synchronous

Return type:

bool