Advanced Memory System

Multi-tiered memory with quality scoring and graph support

Advanced Memory System

The advanced memory system provides sophisticated memory management with short-term, long-term, entity, and user-specific storage, enhanced by quality scoring and optional graph database support.

Diagram

Key Features

Card

Separate short-term and long-term memory systems

Card

4-metric quality assessment for stored memories

Card

User, agent, and run-specific memory scoping

Card

Automatic entity extraction and storage

Card

Optional Neo4j/Memgraph for relationships

Card

Quality-based filtering and relevance ranking

Quick Start

Agent with Memory

from praisonaiagents import Agent, AgentTeam

# Single agent with memory (simplest)
agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant with memory.",
    memory=True  # Enable with defaults
)

# Agent with memory config
agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant with memory.",
    memory={
        "backend": "rag",
        "user_id": "user123",
        "use_embedding": True
    }
)

# Multi-agent with shared memory
agents = AgentTeam(
    agents=[agent1, agent2],
    memory={
        "backend": "rag",
        "use_embedding": True
    }
)

Direct Memory Use

from praisonaiagents import Memory

# Initialise memory
memory = Memory(
    config={
        "provider": "rag",
        "use_embedding": True,
        "short_db": ".praison/short_term.db",
        "long_db": ".praison/long_term.db"
    }
)

# Store memories with quality
memory.store_long_term(
    text="The user prefers technical explanations",
    memory={"user_id": "user123"},
    metadata={"type": "preference"},
    quality=0.9
)

# Search memories — scope with the first-class user_id kwarg
results = memory.search_long_term(
    query="user preferences",
    user_id="user123",
    min_quality=0.7
)

Quality Metrics

# Store with individual quality metrics
memory.store_long_term(
    text="Quantum computing uses qubits for computation",
    completeness=0.95,  # How complete is the information
    relevance=0.90,     # How relevant to the context
    clarity=0.88,       # How clear is the explanation
    accuracy=0.92,      # How accurate is the content
    memory={"user_id": "user123"}
)

# Quality score is automatically calculated
# Overall quality = weighted average of metrics

Memory Tiers

Short-term Memory (STM)

Short-term memory is cleared between sessions and used for immediate context.

python

# Store temporary context and capture the returned id
mid = memory.store_short_term(
    text="User is asking about Python programming",
    metadata={"user_id": "user123", "context": "current_conversation"}
)

# Search recent context
context = memory.search_short_term(
    query="Python",
    limit=5
)

# Targeted delete uses the id (returns True on success)
memory.delete_short_term(mid)

# Or wipe everything
memory.reset_short_term()

memory.reset_short_term() / .reset_long_term() now work on the default rag / chroma provider, not just sqlite / in_memory and dakera. On sqlite / in_memory providers, reset_ / delete_ delegate to the active adapter (mirroring search_short_term), so they operate on the adapter-created short_term_memory / long_term_memory tables.

Long-term Memory (LTM)

Long-term memory persists across sessions and stores important information.

python

# Store persistent knowledge
memory.store_long_term(
    text="User works as a data scientist at TechCorp",
    memory={"user_id": "user123"},
    quality=0.95,
    metadata={"type": "user_info", "category": "professional"}
)

# Retrieve with quality filter — user_id is a first-class kwarg
memories = memory.search_long_term(
    query="user profession",
    user_id="user123",
    min_quality=0.8
)

Entity Memory

Entity memory stores information about specific people, places, or things.

python

# Store entity information
memory.store_entity(
    name="TechCorp",
    text="TechCorp is a leading AI company founded in 2020",
    entity_type="organization",
    metadata={"industry": "technology", "size": "large"}
)

# Search entity information
entity_info = memory.search_entity(
    name="TechCorp",
    query="company details"
)

User Memory

User memory stores personalised information and preferences.

python

# Store user preferences
memory.store_user_memory(
    memory={"user_id": "user123"},
    text="Prefers concise explanations with code examples",
    extra={"preference_type": "communication_style"}
)

# Retrieve user context
user_context = memory.search_user_memory(
    memory={"user_id": "user123"},
    query="preferences",
    limit=3
)

Configuration Options

RAG Configuration (Default)

python

memory_config = {
    "provider": "rag",
    "use_embedding": True,
    "short_db": ".praison/short_term.db",
    "long_db": ".praison/long_term.db",
    "entity_db": ".praison/entity.db",
    "rag_db_path": ".praison/chroma_db",
    "embedding_model": "text-embedding-3-small"
}

Mem0 Configuration

python

memory_config = {
    "provider": "mem0",
    "config": {
        "api_key": "your-mem0-api-key",
        "org_id": "your-org-id",
        "project_id": "your-project-id"
    }
}

Dakera Configuration

Requires: pip install "praisonaiagents[dakera]"

python

memory_config = {
    "provider": "dakera",
    "config": {
        "url": "http://localhost:3000",      # or DAKERA_URL / DAKERA_API_URL env var
        "api_key": "dk-...",                 # or DAKERA_API_KEY env var
        "agent_id": "my-agent",             # or DAKERA_AGENT_ID env var
        "short_term_type": "working",
        "long_term_type": "episodic",
        "default_importance": 0.5,
    }
}

See Dakera Memory for the full guide.

Graph Memory Configuration

python

memory_config = {
    "provider": "mem0",
    "config": {
        "graph_store": {
            "provider": "neo4j",
            "config": {
                "url": "bolt://localhost:7687",
                "username": "neo4j",
                "password": "password"
            }
        },
        "vector_store": {
            "provider": "qdrant",
            "config": {
                "host": "localhost",
                "port": 6333
            }
        },
        "version": "v1.1"
    }
}

Quality Scoring System

Quality Metrics

Completeness

Measures how complete and comprehensive the information is

Relevance

Measures how relevant the information is to the context

Clarity

Measures how clear and understandable the information is

Accuracy

Measures the factual accuracy of the information

Quality Calculation

python

# Default weights
weights = {
    "completeness": 0.25,
    "relevance": 0.35,
    "clarity": 0.20,
    "accuracy": 0.20
}

# Overall quality = weighted average
quality = (
    completeness * weights["completeness"] +
    relevance * weights["relevance"] +
    clarity * weights["clarity"] +
    accuracy * weights["accuracy"]
)

Local LLM for quality scoring

Memory quality scoring uses litellm when available and falls back to the OpenAI client. Point it at any OpenAI-compatible local server by either:

Via memory config (recommended when the rest of your memory layer is config-driven):

python

from praisonaiagents import Memory

memory = Memory(config={
    "llm": {
        "config": {
            "base_url": "http://localhost:11434/v1",
            "model": "ollama/qwen3:0.6b",
        }
    }
})

Via environment (works with any call site, not just memory):

bash

export OPENAI_BASE_URL=http://localhost:11434/v1
export PRAISONAI_AUXILIARY_MODEL=ollama/qwen3:0.6b

Advanced Features

Scoped search with `user_id` and `metadata_filter`

search_short_term and search_long_term accept user_id and metadata_filter as first-class kwargs to scope results per tenant.

python

from praisonaiagents import Agent, Memory

memory = Memory(config={"provider": "rag", "use_embedding": True})
agent = Agent(
    name="TenantAssistant",
    instructions="Answer scoped to the current tenant.",
    memory=True,
)

# Multi-tenant memory scoping
results = memory.search_long_term(
    query="project details",
    user_id="tenant_a",             # tenant isolation — always wins
    metadata_filter={"project": "atlas"},  # extra scoping
    limit=10,
)

user_id is enforced inside Memory.search_short_term / Memory.search_long_term as of PraisonAI PR #4819. The method auto-derives a user_id metadata filter, so adapters that accept user_id but ignore it in their own query (Chroma/RAG, SQLite, in-memory) still isolate per-tenant results. If a caller passes both an explicit user_id and a conflicting metadata_filter["user_id"], the explicit user_id always wins — a caller cannot widen scope to another tenant’s data.

Context Building

Build comprehensive context for tasks:

python

# Automatically merge relevant memories
context = memory.build_context_for_task(
    task_descr="Write a Python tutorial",
    memory={"user_id": "user123"},
    additional="Focus on data science applications",
    max_items=5
)

# Context includes:
# - Relevant long-term memories
# - User preferences
# - Recent short-term context

Task Output Finalisation

Store task results with quality assessment:

python

from praisonaiagents.agent.task import TaskOutput

# Create task output
output = TaskOutput(
    raw="Tutorial content...",
    agent_name="Writer",
    task_description="Write Python tutorial"
)

# Finalise with quality scoring
memory.finalize_task_output(
    task_output=output,
    memory={"user_id": "user123"}
)

Memory Citations

Automatically cite memory sources:

python

# Memories include source information
result = memory.search_long_term("Python", user_id="user123")
# Returns: {
#     'text': 'Python is...',
#     'metadata': {'source': 'conversation_2024_01_15', ...}
# }

Memory Reranking

Enhance search results with intelligent reranking based on relevance scores:

python

from praisonaiagents import Agent, AgentTeam

# Create agent
agent = Agent(
    name="Researcher",
    role="Research assistant"
)

# Initialize with memory enabled
agents = AgentTeam(
    agents=[agent],
    tasks=[],
    memory=True
)

# Search with reranking for better relevance
if agents.shared_memory:
    results = agents.shared_memory.search_long_term(
        "quantum computing applications",
        limit=10,
        rerank=True,  # Enable reranking
        relevance_cutoff=0.7  # Minimum relevance score
    )

Reranking Features

Semantic Reranking: Re-scores results based on semantic similarity
Context-Aware Ranking: Considers current context when ranking
Quality-Weighted Ranking: Combines relevance with quality scores

python

# Advanced reranking with multiple criteria
results = agents.shared_memory.search_long_term(
    query="machine learning frameworks",
    limit=20,  # Retrieve more initially
    rerank=True,  # Enable reranking
    relevance_cutoff=0.6,  # Minimum relevance
    rerank_config={
        "method": "semantic",  # or "hybrid", "quality-weighted"
        "top_k": 5,  # Return top 5 after reranking
        "consider_recency": True,  # Factor in timestamp
        "boost_user_memories": True  # Prioritize user-specific memories
    }
)

Custom Reranking Logic

Implement custom reranking strategies:

python

from datetime import datetime
from praisonaiagents import RerankStrategy

class DomainSpecificReranker(RerankStrategy):
    def rerank(self, results, query, context=None):
        """Custom reranking based on domain knowledge"""
        scored_results = []

        for result in results:
            score = result['relevance']

            # Boost technical content
            if 'technical' in result.get('metadata', {}).get('tags', []):
                score *= 1.2

            # Boost recent memories
            if result.get('timestamp'):
                days_old = (datetime.now() - result['timestamp']).days
                recency_boost = 1.0 + (0.1 * max(0, 30 - days_old) / 30)
                score *= recency_boost

            scored_results.append((score, result))

        # Sort by score and return top results
        scored_results.sort(key=lambda x: x[0], reverse=True)
        return [result for score, result in scored_results]

# Use custom reranker
# Note: Assuming 'agents' is defined from previous examples
if agents.shared_memory:
    agents.shared_memory.set_rerank_strategy(DomainSpecificReranker())

Reranking Performance

Optimize reranking for large result sets:

python

# Efficient two-stage retrieval and reranking
# Stage 1: Fast retrieval of candidates
candidates = agents.shared_memory.search_long_term(
    query="data analysis techniques",
    limit=50,  # Get more candidates
    rerank=False  # Skip reranking in first stage
)

# Stage 2: Precise reranking of top candidates
if len(candidates) > 10:
    reranked = agents.shared_memory.rerank_results(
        results=candidates[:20],  # Rerank top 20
        query="data analysis techniques",
        method="hybrid",
        relevance_cutoff=0.8
    )
else:
    reranked = candidates

Performance Optimisation

Batch Operations

# Batch store memories
memories = [
    ("Fact 1", 0.9),
    ("Fact 2", 0.85),
    ("Fact 3", 0.95)
]

for text, quality in memories:
    memory.store_long_term(
        text=text,
        quality=quality,
        memory={"user_id": "user123"}
    )

Quality Filtering

# Pre-filter by quality to reduce results
high_quality = memory.search_long_term(
    query="important facts",
    min_quality=0.8,  # Only high-quality memories
    limit=10
)

Scoped Search

# Scope search to improve performance
results = memory.search_long_term(
    query="project details",
    user_id="user123",
    agent_id="assistant",
    run_id="session_456"
)

Complete Example

Personal Assistant with Memory

from praisonaiagents import Agent, Task, AgentTeam
from praisonaiagents import Memory

# Configure memory with quality scoring
memory_config = {
    "provider": "rag",
    "use_embedding": True,
    "embedding_model": "text-embedding-3-small"
}

# Create memory instance
memory = Memory(memory_config)

# Create assistant agent
assistant = Agent(
    name="Personal Assistant",
    instructions="""You are a personal assistant with perfect memory.
    Remember user preferences, important information, and context.
    Always use your memory to provide personalised responses.""",
    memory={"user_id": "john_doe"}
)

# Create task with memory integration
task = Task(
    description="Help me plan my day based on my preferences",
    agent=assistant,
    expected_output="Personalised daily schedule"
)

# Run the system
agents = AgentTeam(
    agents=[assistant],
    tasks=[task],
    memory=memory
)

# Memory automatically:
# 1. Retrieves user preferences
# 2. Searches relevant past interactions
# 3. Builds context for the task
# 4. Stores the interaction with quality score
result = agents.start()

# Store user feedback as high-quality memory
memory.store_long_term(
    text="User preferred morning schedule with exercise first",
    memory={"user_id": "john_doe"},
    completeness=0.9,
    relevance=1.0,
    clarity=0.95,
    accuracy=1.0
)

Best Practices

Card

Set minimum quality for critical information (0.8+)
Use lower thresholds for exploratory searches
Regular quality audits

Card

Periodically review and clean old memories
Update quality scores as information ages
Deduplicate similar memories

Card

Use user_id for personalisation
Use agent_id for agent-specific knowledge
Use run_id for session isolation

Card

Enable embeddings for semantic search
Use quality filters to reduce search space
Implement caching for frequent queries

Next Steps

Knowledge

Integrate with document knowledge bases

Sessions

Learn about stateful sessions

Advanced Memory System