flowgentic.langGraph.memory

memory

Memory management system for LangGraph agents.

This module provides memory management capabilities for LangGraph integrations: - Short-term memory: Thread-scoped conversation history management - Memory configuration: Configurable memory strategies and limits - Integration hooks: Memory-aware graph construction and execution - Facade pattern: Simple interface following the established component pattern

Initial implementation focuses on short-term memory management. Long-term memory features will be added in future iterations.

Classes:

Name	Description
MemoryConfig	Configuration for memory management strategies.
ShortTermMemoryItem	Represents a memory item in short-term storage.
ShortTermMemoryManager	Manages short-term memory within conversation threads.
MemoryManager	Main memory management interface for LangGraph integration.
MemoryEnabledState	State schema for graphs with memory support.
LangraphMemoryManager	Facade for memory management in LangGraph integration.

MemoryConfig

MemoryConfig(max_short_term_messages: int = 50, short_term_strategy: str = 'trim_last', context_window_buffer: int = 10, memory_update_threshold: int = 5, summarization_batch_size: int = 10, enable_summarization: bool = False)

Configuration for memory management strategies.

ShortTermMemoryItem

Bases: BaseModel

Represents a memory item in short-term storage.

Methods:

Name	Description
from_message	Create a ShortTermMemoryItem from a BaseMessage.

from_message classmethod

from_message(message: BaseMessage) -> ShortTermMemoryItem

Create a ShortTermMemoryItem from a BaseMessage.

ShortTermMemoryManager

ShortTermMemoryManager(config: MemoryConfig, llm: Optional[BaseChatModel] = None)

Manages short-term memory within conversation threads.

Methods:

Name	Description
add_messages	Add messages to short-term memory and apply trimming strategy.
get_recent_messages	Get the most recent messages.
get_memory_stats	Get statistics about current memory state.
clear	Clear all short-term memory.
consolidate_memory	Consolidate and optimize memory storage.

add_messages

add_messages(messages: List[BaseMessage]) -> List[BaseMessage]

Add messages to short-term memory and apply trimming strategy.

get_recent_messages

get_recent_messages(count: Optional[int] = None) -> List[BaseMessage]

Get the most recent messages.

get_memory_stats

get_memory_stats() -> Dict[str, Any]

Get statistics about current memory state.

clear

clear()

Clear all short-term memory.

consolidate_memory

consolidate_memory() -> Dict[str, Any]

Consolidate and optimize memory storage.

MemoryManager

MemoryManager(config: MemoryConfig, llm: Optional[BaseChatModel] = None)

Main memory management interface for LangGraph integration.

Methods:

Name	Description
add_interaction	Add an interaction to memory and track statistics.
get_relevant_context	Get relevant context from memory with smart retrieval.
get_short_term_messages	Get current short-term messages.
clear_short_term_memory	Clear short-term memory.
consolidate_memory	Consolidate and optimize all memory systems.
get_memory_health	Get overall memory system health and statistics.

add_interaction async

add_interaction(user_id: str, messages: List[BaseMessage], metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Add an interaction to memory and track statistics.

get_relevant_context async

get_relevant_context(user_id: str, query: Optional[str] = None) -> Dict[str, Any]

Get relevant context from memory with smart retrieval.

get_short_term_messages

get_short_term_messages() -> List[BaseMessage]

Get current short-term messages.

clear_short_term_memory

clear_short_term_memory()

Clear short-term memory.

consolidate_memory async

consolidate_memory() -> Dict[str, Any]

Consolidate and optimize all memory systems.

get_memory_health

get_memory_health() -> Dict[str, Any]

Get overall memory system health and statistics.

MemoryEnabledState

Bases: BaseModel

State schema for graphs with memory support.

LangraphMemoryManager

LangraphMemoryManager(config: Optional[MemoryConfig] = None, llm: Optional[BaseChatModel] = None)

Facade for memory management in LangGraph integration.

Follows the same pattern as LangraphToolFaultTolerance and AgentLogger, providing a simple interface for memory operations within the LangraphIntegration.

Parameters:

Name	Type	Description	Default
config	Optional[MemoryConfig]	Memory configuration. If None, uses default configuration.	None
llm	Optional[BaseChatModel]	Language model for summarization features. Optional.	None

Methods:

Name	Description
add_interaction	Add an interaction to memory and track statistics.
get_relevant_context	Get relevant context from memory with smart retrieval.
get_short_term_messages	Get current short-term messages.
clear_short_term_memory	Clear short-term memory.
consolidate_memory	Consolidate and optimize all memory systems.
get_memory_health	Get overall memory system health and statistics.
update_config	Update memory configuration.

add_interaction async

add_interaction(user_id: str, messages: List[BaseMessage], metadata: Optional[Dict[str, Any]] = None) -> Dict[str, Any]

Add an interaction to memory and track statistics.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier for the user/conversation thread	required
messages	List[BaseMessage]	List of messages to add to memory	required
metadata	Optional[Dict[str, Any]]	Optional metadata for the interaction	None

Returns:

Type	Description
Dict[str, Any]	Dictionary with memory statistics and operation results

get_relevant_context async

get_relevant_context(user_id: str, query: Optional[str] = None) -> Dict[str, Any]

Get relevant context from memory with smart retrieval.

Parameters:

Name	Type	Description	Default
user_id	str	Identifier for the user/conversation thread	required
query	Optional[str]	Optional query string for semantic search	None

Returns:

Type	Description
Dict[str, Any]	Dictionary containing relevant context and memory statistics

get_short_term_messages

get_short_term_messages() -> List[BaseMessage]

Get current short-term messages.

Returns:

Type	Description
List[BaseMessage]	List of current messages in short-term memory

clear_short_term_memory

clear_short_term_memory() -> None

Clear short-term memory.

consolidate_memory async

consolidate_memory() -> Dict[str, Any]

Consolidate and optimize all memory systems.

Returns:

Type	Description
Dict[str, Any]	Dictionary with consolidation results and statistics

get_memory_health

get_memory_health() -> Dict[str, Any]

Get overall memory system health and statistics.

Returns:

Type	Description
Dict[str, Any]	Dictionary with memory health metrics and configuration

update_config

update_config(config: MemoryConfig) -> None

Update memory configuration.

Parameters:

Name	Type	Description	Default
config	MemoryConfig	New memory configuration to apply	required