Advanced Memory System

from praisonaiagents import Agent, MemoryConfig

agent = Agent(
    name="memory-agent",
    instructions="Remember context across turns.",
    memory=MemoryConfig(use_long_term=True),
)
agent.start("What did we decide about the launch date?")

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
)

from praisonaiagents import Memory

memory = Memory(config={"provider": "rag"})

memory.store_long_term(
    text="Quantum computing uses qubits for computation",
    completeness=0.95,
    relevance=0.90,
    clarity=0.88,
    accuracy=0.92,
    memory={"user_id": "user123"}
)

# 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()

# 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
)

from praisonaiagents import Memory

memory = Memory(config={"provider": "rag"})

memory.store_long_term(
    text="Project atlas kicked off",
    metadata={
        "project": "atlas",              # stored as "atlas"
        "priority": 3,                    # stored as 3
        "shipped": False,                 # stored as False
        "owner": None,                    # DROPPED — not stored
        "tags": ["backend", "urgent"],    # stored as the string "['backend', 'urgent']"
        "config": {"env": "prod"},        # stored as the string "{'env': 'prod'}"
    },
)

# 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"
)

# 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
)

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"
}

# Shared memory across Agent instances using db_path
from praisonaiagents import Agent
from praisonaiagents.config.feature_configs import MemoryConfig

shared = "/tmp/team-memory.db"

writer = Agent(
    name="Writer",
    instructions="Remember what the user tells you.",
    memory=MemoryConfig(backend="sqlite", config={"db_path": shared}),
)
writer.start("Remember codename ORANGE-PANDA for the Q4 launch.")

reader = Agent(
    name="Reader",
    instructions="Answer questions using memory.",
    memory=MemoryConfig(backend="sqlite", config={"db_path": shared}),
)
reader.start("What is the codename for the Q4 launch?")

Explicit short_db / long_db  >  db_path  >  Per-user default paths

# Explicit short_db / long_db still win over db_path
from praisonaiagents import Agent
from praisonaiagents.config.feature_configs import MemoryConfig

agent = Agent(
    name="Writer",
    instructions="Keep short- and long-term memory in separate files.",
    memory=MemoryConfig(
        backend="sqlite",
        config={
            "db_path":  "/tmp/ignored.db",   # ignored because the explicit paths win
            "short_db": "/tmp/short.db",
            "long_db":  "/tmp/long.db",
        },
    ),
)

> pip install "praisonaiagents[memory]"
> ```
> If the package is missing, configuring `"provider": "mem0"` raises:
> ```
> ImportError: mem0ai is not installed. Run: pip install 'praisonaiagents[memory]'
> ```

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

> pip install "praisonaiagents[dakera]"
> ```
> If the package is missing, configuring `"provider": "dakera"` raises:
> ```
> ImportError: dakera is required for the dakera adapter. Install with: pip install 'praisonaiagents[dakera]' (or: pip install 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,
    }
}

> pip install "praisonaiagents[memory]"
> ```

```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"
    }
}

# 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"]
)

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,
)

from praisonaiagents import Memory

memory = Memory(config={"provider": "rag"})

context = memory.build_context_for_task(
    task_descr="Research the latest AI developments",
    user_id="user_123",
    max_items=3,
    include_in_output=True  # Force include for prompt assembly
)

# Construct cache-friendly prompt structure
system_prompt = f"System instructions and tools\n\n{context}\n\nUser message: {{user_input}}"

# 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

from praisonaiagents 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"}
)

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

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
    )

# 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
    }
)

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())

# 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

from praisonaiagents import Agent, MemoryConfig

# Enable auto-memory extraction
agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    memory=MemoryConfig(auto_memory=True, user_id="user123")
)

# Patterns are extracted automatically from conversation
agent.start("I prefer Python for data science and I'm based in London")
# AutoMemory stores: preference="Python for data science", location="London"

from praisonaiagents import AutoMemory

# Create with custom patterns
auto_mem = AutoMemory(
    user_id="user123",
    patterns={
        "preference": r"(?:I prefer|I like|I love)\s+(.+)",
        "location": r"(?:I'm based in|I live in|I'm from)\s+(.+)",
        "name": r"(?:My name is|I'm called|call me)\s+(\w+)",
        "fact": r"(?:I work at|I study at|my job is)\s+(.+)",
    }
)

# Search extracted memories
results = auto_mem.search("user preferences")

from praisonaiagents import Agent, MemoryConfig

agent = Agent(
    instructions="You are a helpful assistant.",
    memory=MemoryConfig(
        backend="file",
        auto_memory=True,   # Enable pattern-based extraction
        user_id="user123",
        session_id="session456",
    )
)

from praisonaiagents.memory import FileMemory, AutoMemory

base = FileMemory(user_id="user123")
auto = AutoMemory(base, enabled=True)

# Preview — extraction only, no state change
auto.process_interaction("I prefer dark mode.", store=False)

# Real store — writes because the preview did not consume dedup
auto.process_interaction("I prefer dark mode.", store=True)

# Repeated real store — deduplicated (returns [])
auto.process_interaction("I prefer dark mode.", store=True)

# A failed store leaves dedup un-consumed — retry after recovery stores
original_add_long_term = base.add_long_term
base.add_long_term = lambda *a, **kw: (_ for _ in ()).throw(OSError("disk full"))

try:
    auto.process_interaction("I like concise answers.", store=True)
except OSError:
    pass

base.add_long_term = original_add_long_term  # backend recovers

retried = auto.process_interaction("I like concise answers.", store=True)
assert retried != []   # stores on retry