langgraph_agent_toolkit.core.memory.concurrency module

Bound concurrent conversation operations and release locks after failures.

exception langgraph_agent_toolkit.core.memory.concurrency.ConversationBusyError

Bases: Exception

The conversation queue is full or its wait time has expired.

exception langgraph_agent_toolkit.core.memory.concurrency.NestedConversationError

Bases: ConversationBusyError

An active operation tried to acquire another conversation lock.

exception langgraph_agent_toolkit.core.memory.concurrency.ConversationLockLostError

Bases: Exception

The database session that owns the conversation lock was lost.

class langgraph_agent_toolkit.core.memory.concurrency.PostgresLockSession(connection, scope=None, lock=<factory>, active=True)

Bases: object

Bind checkpoint I/O to the session that owns its advisory lock.

Parameters:

• connection (object)

• scope (str | None)

• lock (Lock)

• active (bool)

connection: object

scope: str | None = None

lock: Lock

active: bool = True

__init__(connection, scope=None, lock=<factory>, active=True)

Parameters:

• connection (object)

• scope (str | None)

• lock (Lock)

• active (bool)

Return type:

None

class langgraph_agent_toolkit.core.memory.concurrency.ConversationCoordinator(timeout=None, max_waiters=None)

Bases: object

Serialize operations on one thread in this process.

Parameters:

• timeout (float | None)

• max_waiters (int | None)

__init__(timeout=None, max_waiters=None)

Parameters:

• timeout (float | None)

• max_waiters (int | None)

lock(key)

Parameters:

key (str)

class langgraph_agent_toolkit.core.memory.concurrency.SQLiteConversationCoordinator(database_path, **kwargs)

Bases: ConversationCoordinator

Use OS file locks to coordinate workers that share one SQLite file.

Parameters:

database_path (str)

__init__(database_path, **kwargs)

Parameters:

database_path (str)

class langgraph_agent_toolkit.core.memory.concurrency.PostgresConversationCoordinator(pool, **kwargs)

Bases: ConversationCoordinator

Hold a session advisory lock in a separate, bounded connection pool.

__init__(pool, **kwargs)

langgraph_agent_toolkit.core.memory.concurrency.serialize_execution(function)

Hold a conversation lock through execution and stream cleanup.