Memory

Saving Memory

put_memory

def put_memory(
    self,
    namespace_for_memory: tuple,
    thread_id: str,
    memory: dict
) -> None

Store or update a memory record.

# Example
memory_store.put_memory(
    namespace_for_memory=('memories', user_id),
    thread_id='conversation-123',
    memory={'messages': [...], 'context': {...}}
)

upsert_memory

def upsert_memory(
    self,
    namespace_for_memory: tuple,
    thread_id: str,
    key: str,
    new_memory: str
) -> None

Append a new memory entry to the specified key in conversation history.

# Example
memory_store.upsert_memory(
    namespace_for_memory=('memories', user_id),
    thread_id='conversation-123',
    key='user_preferences',
    new_memory='User prefers dark mode'
)

set_current_agent_id

def set_current_agent_id(
    self,
    user_id: str,
    thread_id: str,
    agent_id: str
) -> None

Set the current agent handling a conversation thread.

# Example
memory_store.set_current_agent_id(
    user_id='user-456',
    thread_id='conversation-123',
    agent_id='support-agent'
)

Getting Memory

get_memory

def get_memory(
    self,
    namespace_for_memory: tuple,
    thread_id: str
) -> Optional[MemoryWrapper]

Retrieve a memory record. Returns None if not found.

# Example
memory = memory_store.get_memory(
    namespace_for_memory=('memories', user_id),
    thread_id='conversation-123'
)
if memory:
    print(memory.value)  # Access the conversation data

list_memories

def list_memories(
    self,
    namespace_for_memory: tuple
) -> List[MemoryWrapper]

List recent memories for a user (limited to 20).

# Example
memories = memory_store.list_memories(
    namespace_for_memory=('memories', user_id)
)
for memory in memories:
    print(f"Thread: {memory.key}, Data: {memory.value}")

get_current_agent_id

def get_current_agent_id(
    self,
    user_id: str,
    thread_id: str
) -> Optional[str]

Get the current agent handling a conversation thread.

# Example
agent_id = memory_store.get_current_agent_id(
    user_id='user-456',
    thread_id='conversation-123'
)

get_memories_by_date_range

def get_memories_by_date_range(
    self,
    user_ids: list,
    from_date: str,
    to_date: str
) -> list

Get memories for multiple users within a date range. Dates should be ISO format strings.

# Example
memories = memory_store.get_memories_by_date_range(
    user_ids=['user-1', 'user-2'],
    from_date='2024-01-01',
    to_date='2024-01-31'
)

Deleting Memory

delete_memory

def delete_memory(
    self,
    namespace_for_memory: tuple,
    thread_id: str
) -> bool

Delete a specific memory record. Returns True if deleted, False if not found.

# Example
was_deleted = memory_store.delete_memory(
    namespace_for_memory=('memories', user_id),
    thread_id='conversation-123'
)