Skip to main content
Memory lifecycle hooks let your memory backend react when the agent compresses context, switches sessions, writes to memory, or delegates to a subagent.

​
Quick Start

1

Simplest

Override just one hook to get started:
from praisonaiagents import Agent
from praisonaiagents.memory import Memory

class MyMemory(Memory):
    def on_pre_compress(self, messages):
        print(f"About to discard {len(messages)} messages")
        return "User prefers concise answers"

agent = Agent(
    name="Assistant",
    instructions="Help the user",
    memory=MyMemory(),
)

agent.start("Tell me about Mars")
2

All Four Hooks

Implement all lifecycle hooks for full control:
from praisonaiagents import Agent
from praisonaiagents.memory import Memory

class FullHooksMemory(Memory):
    def on_pre_compress(self, messages):
        # Extract facts before compression
        facts = [msg['content'] for msg in messages if 'fact:' in msg.get('content', '')]
        for fact in facts:
            self.store_long_term(fact)
        return f"Preserved {len(facts)} facts"

    def on_session_switch(self, new_session_id, *, parent_session_id="", reset=False):
        if reset:
            print(f"Starting fresh conversation: {new_session_id}")
        else:
            print(f"Continuing from {parent_session_id} to {new_session_id}")

    def on_memory_write(self, action, target, content, metadata=None):
        print(f"Memory {action}: {content[:50]}... -> {target}")

    def on_delegation(self, task, result, *, agent_name="", metadata=None):
        # Store subagent results in knowledge base
        self.store_long_term(f"Task: {task} | Result: {result} | Agent: {agent_name}")

agent = Agent(name="Assistant", memory=FullHooksMemory())

​
How It Works

EventCall SiteWhen It Fires
on_pre_compresschat_mixin.py:_apply_context_managementContext utilization ≥ compact_threshold
on_memory_writeagent.py:store_memoryAfter successful memory storage
on_delegationhandoff.py (3 paths)After subagent task completion
on_session_switchNot yet wiredReserved for future session rotation

​
The Four Hooks

​
on_pre_compress

Called before context compression discards messages. 
def on_pre_compress(self, messages: list[dict]) -> str:
    """Extract facts before messages are discarded."""
    pass

# Async version
async def aon_pre_compress(self, messages: list[dict]) -> str:
    """Async version for async agent contexts."""
    pass
When does this fire? When context utilization exceeds compact_threshold and the agent needs to discard older messages. Example: 
def on_pre_compress(self, messages):
    # Extract structured facts before compression
    facts = []
    for msg in messages:
        if msg.get('role') == 'user' and 'preference:' in msg.get('content', ''):
            facts.append(msg['content'])
    
    # Store in external graph database
    for fact in facts:
        self.graph_db.store(fact)
    
    return f"Extracted {len(facts)} preferences"

​
on_session_switch

Called when the active session ID changes. 
def on_session_switch(self, new_session_id: str, *, parent_session_id: str = "", reset: bool = False) -> None:
    """Update routing for new session."""
    pass

# Async version  
async def aon_session_switch(self, new_session_id: str, *, parent_session_id: str = "", reset: bool = False) -> None:
    """Async version for async agent contexts."""
    pass
When does this fire? Currently defined but not yet called - reserved for future session rotation features. Example: 
def on_session_switch(self, new_session_id, *, parent_session_id="", reset=False):
    if reset:
        # Fresh conversation - clear session-specific cache
        self.session_cache.clear()
    else:
        # Session continuation - migrate cache
        self.session_cache[new_session_id] = self.session_cache.pop(parent_session_id, {})
    
    self.current_session = new_session_id

​
on_memory_write

Called after successful memory storage operations. 
def on_memory_write(self, action: str, target: str, content: str, metadata: dict = None) -> None:
    """Mirror memory writes to external store."""
    pass

# Async version
async def aon_memory_write(self, action: str, target: str, content: str, metadata: dict = None) -> None:
    """Async version for async agent contexts."""
    pass
When does this fire? After Agent.store_memory() successfully writes to built-in memory. Example: 
def on_memory_write(self, action, target, content, metadata=None):
    # Mirror all writes to external vector store
    if action == "add":
        self.vector_store.add(content, tags=[target])
    elif action == "replace":
        self.vector_store.update(content, tags=[target])
    elif action == "remove":
        self.vector_store.delete(content, tags=[target])

​
on_delegation

Called after a subagent completes a delegated task. 
def on_delegation(self, task: str, result: str, *, agent_name: str = "", metadata: dict = None) -> None:
    """Store delegation results."""
    pass

# Async version
async def aon_delegation(self, task: str, result: str, *, agent_name: str = "", metadata: dict = None) -> None:
    """Async version for async agent contexts."""
    pass
When does this fire? After subagent execution completes in handoff operations. Example: 
def on_delegation(self, task, result, *, agent_name="", metadata=None):
    # Build knowledge graph from delegation results
    entry = {
        "task": task,
        "result": result, 
        "agent": agent_name,
        "timestamp": time.time()
    }
    self.knowledge_graph.add_delegation(entry)

​
Sync vs Async Selection

The agent automatically detects the execution context. If you’re in an async context and provide both sync and async versions, the async version runs as a fire-and-forget task for better performance.

​
The New Action Parameter

Agent.store_memory() now accepts an action parameter: 
# Add new memory (default)
agent.store_memory("User likes coffee", action="add")

# Replace existing memory
agent.store_memory("User prefers tea", action="replace") 

# Remove memory
agent.store_memory("User likes coffee", action="remove")
ActionRequirementsDescription
"add"Default behaviorStores new content
"replace"Provider must have replace_<type> or update_<type> methodUpdates existing content
"remove"Provider must have remove_<type> or delete_<type> methodDeletes content
The memory provider must support the requested action or a ValueError is raised.

​
Configuration Options

All four hooks are optional with no-op defaults. The protocol uses runtime_checkable so your memory class only needs to implement the hooks you need.
HookArgumentsReturn TypeWhen It Fires
on_pre_compressmessages: list[dict]strContext utilization ≥ threshold
on_session_switchnew_session_id: str, parent_session_id: str, reset: boolNoneSession ID changes (not yet called)
on_memory_writeaction: str, target: str, content: str, metadata: dictNoneAfter store_memory() success
on_delegationtask: str, result: str, agent_name: str, metadata: dictNoneAfter subagent completion

​
Common Patterns

Mirror writes to external vector store: 
def on_memory_write(self, action, target, content, metadata=None):
    if action == "add":
        self.pinecone.upsert([{
            "id": str(uuid.uuid4()),
            "values": self.embed(content),
            "metadata": {"type": target, **metadata}
        }])
Persist subagent results in knowledge graph: 
def on_delegation(self, task, result, *, agent_name="", metadata=None):
    self.neo4j.run("""
        CREATE (d:Delegation {
            task: $task, 
            result: $result, 
            agent: $agent_name,
            timestamp: timestamp()
        })
    """, task=task, result=result, agent_name=agent_name)
Summarize discarded turns before compression: 
def on_pre_compress(self, messages):
    turns = [msg for msg in messages if msg.get('role') in ['user', 'assistant']]
    summary = self.summarizer.summarize(turns)
    self.store_long_term(f"Conversation summary: {summary}")
    return "Key points preserved"

​
Best Practices

Hooks run on the agent thread and can block execution. Keep operations lightweight or delegate to background workers for heavy processing like vector embeddings or API calls.
Hook failures are caught and logged at warning level. They should never break the agent loop. Always wrap your hook logic in try-catch and log errors appropriately.
If your memory provider does network I/O (database calls, API requests), implement the aon_* async variants. The agent will automatically schedule them as fire-and-forget tasks in async contexts.
All hooks are optional. Only implement the lifecycle events that matter for your use case. Empty implementations have zero performance impact.

Memory

Core memory concepts and storage types

Advanced Memory

Custom backends and advanced patterns

Context Compression

How context compression triggers hooks

Handoffs

Agent delegation and subagent patterns