Memory & Conversations¶
FlowgentraAI provides several memory mechanisms for tracking conversations across turns.
ConversationMemory¶
Low-level message storage with optional sliding window:
from flowgentra_ai import ConversationMemory, Message
# Create with optional max messages per thread
mem = ConversationMemory(max_messages=100)
# Add messages to a thread
mem.add_message("thread-1", Message.user("Hello"))
mem.add_message("thread-1", Message.assistant("Hi! How can I help?"))
mem.add_message("thread-1", Message.user("What is Rust?"))
# Retrieve messages
messages = mem.messages("thread-1") # all messages
messages = mem.messages("thread-1", limit=5) # last 5 messages
# Clear a thread
mem.clear("thread-1")
MemoryAwareAgent¶
High-level agent with automatic memory management per thread:
from flowgentra_ai import MemoryAwareAgent
agent = MemoryAwareAgent.from_config("config.yaml")
agent.set_thread_id("user_123")
# Each turn automatically remembers previous context
answer1 = agent.run_turn("What is Rust?")
answer2 = agent.run_turn("What are its main features?") # knows we're talking about Rust
# Check memory usage
stats = agent.memory_stats()
print(stats.message_count)
print(stats.user_messages)
print(stats.assistant_messages)
print(stats.approximate_tokens)
# Switch threads
agent.set_thread_id("user_456")
answer = agent.run_turn("Hello!") # fresh conversation
# Clear memory for current thread
agent.clear_memory()
TokenBufferMemory¶
Keeps conversation history within a token budget:
SummaryMemory¶
Automatically summarizes older messages to save tokens:
Checkpointing¶
For graph-level state persistence (not conversation memory), use the checkpointer:
from flowgentra_ai import FileCheckpointer
# Set up on a graph builder
builder.set_checkpointer("./checkpoints")
graph = builder.compile()
# Invoke with thread ID -- state is persisted
result = graph.invoke_with_thread("thread-1", initial_state)
# Resume later
result = graph.resume("thread-1")
See Human-in-the-Loop for interrupt/resume patterns.