Memory SDK

Agent
  ├── MemoryMixin (hooks into prompt, tool exec, post-query)
  │     ├── Embedding pipeline (Lemonade → user.embeddinggemma-300m-GGUF, 768-dim)
  │     ├── FAISS index (IndexFlatIP, cosine similarity)
  │     ├── Cross-encoder reranker (ms-marco-MiniLM-L-6-v2)
  │     └── MemoryStore (SQLite: ~/.gaia/memory.db)
  │           ├── conversations table (+ consolidated_at column)
  │           ├── knowledge table (FTS5 + embedding BLOB + superseded_by)
  │           └── tool_history table
  └── SystemDiscovery (file system, git, apps, browser scans)

from gaia.agents.base.agent import Agent
from gaia.agents.base.memory import MemoryMixin

# Correct -- MemoryMixin before Agent
class MyAgent(MemoryMixin, Agent):
    def __init__(self, **kwargs):
        self.init_memory()              # Before super().__init__()
        super().__init__(**kwargs)

    def _register_tools(self):
        super()._register_tools()
        self.register_memory_tools()    # Exposes 5 tools to the LLM

def init_memory(
    self,
    db_path: Optional[Path] = None,
    context: str = "global"
) -> None:

def get_memory_system_prompt(self) -> str

=== MEMORY (Persistent Second Brain) ===
You have persistent memory across sessions. USE IT PROACTIVELY:
- When the user states a fact, preference, or commitment → call `remember` immediately
- When the user asks what you know, what was discussed, or about a person/project → call `recall`
- When information changes or is corrected → call `recall` to find the old item, then `update_memory`
- When the user mentions a deadline or reminder → call `remember` with due_at (ISO 8601)
- When the user wants to forget something → call `recall` to find it, then `forget`
- BIAS TOWARD REMEMBERING: if in doubt, store it. It's better to remember too much than too little.
- Every fact, preference, name, project detail, deadline, or observation is worth storing.

Preferences:
  - tone: professional but friendly
  - code_style: black formatter, 88 char lines

Known facts:
  - Project uses React 19 with app router (confidence: 0.82)
  - User's name is Alex, role is tech lead (confidence: 0.95)

Skills:
  - Deploy workflow: test → build → push → verify (confidence: 0.88)
  - Docker compose: always use --build flag on first run (confidence: 0.72)
  - Git bisect: use binary search for regression hunting (confidence: 0.65)

Known errors to avoid:
  - execute_code: "import torch" fails -- torch not installed on this machine
  - pip install: always use --index-url for PyTorch packages

=== MEMORY (Persistent Second Brain) ===
You have persistent memory across sessions. USE IT PROACTIVELY:
[... same instructions ...]

No memories stored yet. Start building your knowledge base by remembering what the user tells you.

def get_memory_dynamic_context(self) -> str

[GAIA Memory Context]
Current time: 2026-03-25T10:30:00-0700 (Tuesday)

Upcoming/overdue:
  - [DUE Mar 27] Online course starts next week
  - [OVERDUE Mar 24] Follow up on deployment review
After mentioning a time-sensitive item, call update_memory to set reminded_at so you don't repeat yourself.

def register_memory_tools(self) -> None

def set_memory_context(self, context: str) -> None

agent.set_memory_context("work")      # Switch to work context
agent.set_memory_context("personal")  # Switch to personal context

def reset_memory_session(self) -> None

def _get_embedder(self) -> Any

def _embed_text(self, text: str) -> np.ndarray

def _backfill_embeddings(self, limit: int = 100) -> int

def _hybrid_search(
    self,
    query: str,
    category: str = None,
    context: str = None,
    entity: str = None,
    include_sensitive: bool = False,
    top_k: int = 5,
    time_from: str = None,
    time_to: str = None,
) -> List[Dict]:

def _classify_query_complexity(self, query: str) -> int

def _extract_via_llm(
    self,
    user_input: str,
    assistant_response: str,
    existing_items: List[Dict],
) -> List[Dict]:

def consolidate_old_sessions(self, max_sessions: int = 5) -> Dict:

def reconcile_memory(self, max_pairs: int = 20) -> Dict:

# Simplified flow inside MemoryMixin.process_query():
def process_query(self, user_input, **kwargs):
    self._original_user_input = user_input           # Save clean version
    dynamic = self.get_memory_dynamic_context()       # Time + upcoming
    augmented = f"{dynamic}\n\n{user_input}" if dynamic else user_input
    return super().process_query(augmented, **kwargs)  # System prompt stays frozen

class MemoryStore:
    def __init__(self, db_path: Path = None):
        """Open or create the database.
        Default path: ~/.gaia/memory.db
        Uses WAL mode. Thread-safe."""

ALTER TABLE knowledge ADD COLUMN embedding BLOB;
ALTER TABLE knowledge ADD COLUMN superseded_by TEXT;
ALTER TABLE conversations ADD COLUMN consolidated_at TEXT;

def store(
    self,
    category: str,              # 'fact' | 'preference' | 'error' | 'skill' | 'note' | 'reminder'
    content: str,               # Human-readable description
    domain: str = None,         # Optional grouping (e.g., 'python', 'deployment')
    metadata: dict = None,      # JSON blob for structured data
    confidence: float = 0.5,    # 0.0 to 1.0
    due_at: str = None,         # ISO 8601 for time-sensitive items
    source: str = "tool",       # 'tool' | 'llm_extract' | 'error_auto' | 'user' | 'discovery' | 'consolidation'
    context: str = "global",    # Context scope
    sensitive: bool = False,    # Exclude from system prompt if True
    entity: str = None,         # Entity reference (e.g., 'person:sarah_chen')
) -> str:                       # Returns knowledge_id (UUID)

def search(
    self,
    query: str,                     # FTS5 search query
    category: str = None,           # Filter by category
    context: str = None,            # Filter by context
    entity: str = None,             # Filter by entity
    include_sensitive: bool = False, # Include sensitive items
    top_k: int = 5,                 # Max results
    time_from: str = None,          # ISO 8601 lower bound on created_at
    time_to: str = None,            # ISO 8601 upper bound on created_at
) -> List[Dict]:

def get_by_category(
    self,
    category: str,
    context: str = None,
    limit: int = 10,
) -> List[Dict]:

def get_by_entity(
    self,
    entity: str,          # e.g., 'person:sarah_chen'
    limit: int = 20,
) -> List[Dict]:

def get_upcoming(
    self,
    within_days: int = 7,
    include_overdue: bool = True,
    context: str = None,
    limit: int = 10,
) -> List[Dict]:

def update(
    self,
    knowledge_id: str,
    content: str = None,
    category: str = None,
    domain: str = None,
    metadata: dict = None,
    context: str = None,
    sensitive: bool = None,
    entity: str = None,
    due_at: str = None,
    reminded_at: str = None,
    superseded_by: str = None,  # ID of newer item that replaces this one
) -> bool:                  # False if ID not found

def delete(self, knowledge_id: str) -> bool

def apply_confidence_decay(
    self,
    days_threshold: int = 30,
    decay_factor: float = 0.9,
) -> int:                   # Returns number of entries decayed

def update_confidence(self, knowledge_id: str, delta: float) -> None

def delete_by_source(self, source: str) -> int

def store_turn(self, session_id: str, role: str, content: str,
               context: str = "global") -> None

def get_history(self, session_id: str = None, context: str = None,
                limit: int = 20) -> List[Dict]

def search_conversations(self, query: str, context: str = None,
                         limit: int = 10) -> List[Dict]

def get_recent_conversations(self, days: int = 7, context: str = None,
                             limit: int = 50) -> List[Dict]

def log_tool_call(self, session_id: str, tool_name: str,
                  args: dict, result_summary: str,
                  success: bool, error: str = None,
                  duration_ms: int = None) -> None

def get_tool_errors(self, tool_name: str = None,
                    limit: int = 10) -> List[Dict]

def get_tool_stats(self, tool_name: str) -> Dict
    # Returns: {total_calls, success_rate, avg_duration_ms, last_error}

def get_stats(self) -> Dict
    # Returns counts by category, context, entity, conversations, tools, temporal

def get_all_knowledge(self, category=None, context=None, entity=None,
                      sensitive=None, search=None, sort_by="updated_at",
                      order="desc", offset=0, limit=50,
                      include_superseded=False) -> Dict
    # Returns: {"items": [...], "total": 142, "offset": 0, "limit": 50}
    # When include_superseded=False (default), filters on superseded_by IS NULL

def get_tool_summary(self) -> List[Dict]
    # Per-tool stats: total_calls, success_rate, avg_duration_ms, last_error

def get_activity_timeline(self, days: int = 30) -> List[Dict]
    # Daily activity counts for the last N days

def get_recent_errors(self, limit: int = 20) -> List[Dict]

def prune(self, days: int = 90) -> Dict
    # Delete tool_history and conversations older than N days.
    # Also prunes low-confidence knowledge (confidence < 0.1) last used > N days ago.
    # Returns: {"tool_history_deleted": N, "conversations_deleted": N, "knowledge_deleted": N}
    # Called automatically on agent startup via init_memory().

def rebuild_fts(self) -> None
    # Rebuild all FTS5 indexes from source tables.
    # Use if search results seem wrong or incomplete.
    # Also available via POST /api/memory/rebuild-fts

def store_embedding(self, knowledge_id: str, embedding: bytes) -> bool
    # Store a float32 embedding BLOB for a knowledge item.
    # Called by MemoryMixin after store() to persist the vector.
    # Returns False if knowledge_id not found.

def get_items_with_embeddings(
    self,
    category: str = None,
    context: str = None,
    entity: str = None,
    include_sensitive: bool = False,
    top_k: int = 100,
    time_from: str = None,          # ISO 8601 lower bound on created_at
    time_to: str = None,            # ISO 8601 upper bound on created_at
) -> List[Dict]
    # Returns knowledge items that have embeddings (embedding IS NOT NULL,
    # superseded_by IS NULL). Includes the embedding BLOB in each dict.
    # Used to build/rebuild the FAISS index and for filtered vector retrieval.

def get_items_without_embeddings(self, limit: int = 100) -> List[Dict]
    # Returns knowledge items missing embeddings (embedding IS NULL).
    # Used by _backfill_embeddings() during startup.

def get_unconsolidated_sessions(
    self,
    older_than_days: int = 14,
    min_turns: int = 5,
    limit: int = 5,
) -> List[str]
    # Returns session_ids eligible for consolidation.
    # Criteria: all turns > older_than_days old, >= min_turns, at least one
    # turn with consolidated_at IS NULL.

def mark_turns_consolidated(self, turn_ids: List[int]) -> int
    # Sets consolidated_at = now for the given conversation turn IDs.
    # Returns count of turns marked. Turns remain until 90-day prune;
    # consolidated_at prevents re-processing.

def get_items_for_reconciliation(
    self,
    context: str = None,
    limit: int = 100,
) -> List[Dict]
    # Returns active knowledge items (superseded_by IS NULL) with embeddings,
    # suitable for pairwise similarity comparison during reconciliation.

def get_sessions(self, limit: int = 20) -> List[Dict]
    # List conversation sessions with turn counts and preview text.
    # Used by the Memory Dashboard.

def get_entities(self, limit: int = 100) -> List[Dict]
    # List all unique entities with knowledge counts and last update time.
    # Returns: [{"entity": "person:sarah_chen", "count": 5, "last_updated": "..."}, ...]

def get_contexts(self, limit: int = 100) -> List[Dict]
    # List all contexts with knowledge counts.
    # Returns: [{"context": "work", "count": 42}, ...]

from gaia.agents.base.discovery import SystemDiscovery

discovery = SystemDiscovery()
results = discovery.scan_all()
# results is a Dict[str, List[Dict]] — source name → list of discovered facts
# Example: {"file_system": [...], "git_repos": [...], "installed_apps": [...]}

# To iterate all findings:
findings = []
for source_name, items in results.items():
    findings.extend(items)

# Each item dict:
# {content, category, context, entity, sensitive, confidence, source, approved}

{
    "content": "Project 'gaia' -- Python/TypeScript, github.com/amd/gaia",
    "category": "fact",
    "context": "work",
    "entity": "project:gaia",
    "sensitive": False,
    "confidence": 0.4,       # Lower than user-stated (inferred)
    "source": "discovery",
    "approved": None,        # Set by user review: True/False
}

from gaia.agents.base.agent import Agent
from gaia.agents.base.memory import MemoryMixin


class RememberBot(MemoryMixin, Agent):
    """A simple agent that remembers everything."""

    def __init__(self):
        self.init_memory(context="global")
        super().__init__(
            name="RememberBot",
            system_prompt="You are a helpful assistant with persistent memory.",
        )

    def _register_tools(self):
        super()._register_tools()
        self.register_memory_tools()


# Usage
agent = RememberBot()
result = agent.process_query("My name is Alex and I prefer concise answers")
# Memory auto-extracts: fact("My name is Alex"), preference("prefer concise answers")
# Next session, system prompt includes these automatically

class WorkPersonalAgent(MemoryMixin, Agent):
    def __init__(self, context="work"):
        self.init_memory(context=context)
        super().__init__(name="DualContext")

    def _register_tools(self):
        super()._register_tools()
        self.register_memory_tools()


agent = WorkPersonalAgent(context="work")
agent.process_query("Remember: deploy with kubectl apply -f prod.yaml")
# Stored in 'work' context

agent.set_memory_context("personal")
agent.process_query("Remember: dentist appointment Thursday at 2pm")
# Stored in 'personal' context

agent.set_memory_context("work")
# System prompt now shows work items only (plus global)
# Personal items are invisible until context switches back

from gaia.agents.base.memory_store import MemoryStore

store = MemoryStore()  # Uses ~/.gaia/memory.db

# Store a fact
kid = store.store(
    category="fact",
    content="Project uses Python 3.12 with uv",
    context="work",
    entity="project:gaia",
)

# Search
results = store.search("Python version", context="work")

# Get everything about an entity
gaia_facts = store.get_by_entity("project:gaia")

# Get upcoming items
upcoming = store.get_upcoming(within_days=7)

# Dashboard stats
stats = store.get_stats()
print(f"Total memories: {stats['knowledge']['total']}")
print(f"Tool success rate: {stats['tools']['overall_success_rate']:.0%}")

# Cleanup
store.close()

from pathlib import Path

class TestAgent(MemoryMixin, Agent):
    def __init__(self):
        self.init_memory(
            db_path=Path("./test_memory.db"),
            context="testing",
        )
        super().__init__(name="TestAgent")

    def _register_tools(self):
        super()._register_tools()
        self.register_memory_tools()