Memory Architecture¶

Overview¶

Aegis implements a multi-tiered memory system combining relational storage (PostgreSQL), graph databases (FalkorDB), vector search (pgvector), and file-based memory for persistent learning and context retention.

Memory Hierarchy¶

graph TB
    subgraph "Memory Tiers"
        Episodic[Episodic Memory<br/>Events & Experiences]
        Semantic[Semantic Memory<br/>Facts & Knowledge]
        Procedural[Procedural Memory<br/>Skills & Workflows]
        Cache[Semantic Cache<br/>Response Memoization]
    end

    subgraph "Storage Backends"
        Postgres[(PostgreSQL<br/>Relational + Vector)]
        FalkorDB[(FalkorDB<br/>Knowledge Graph)]
        FileSystem[File System<br/>~/memory/]
    end

    subgraph "Interfaces"
        Python[Python API]
        MCP[MCP Tools]
    end

    Episodic --> Postgres
    Episodic --> FalkorDB
    Semantic --> Postgres
    Semantic --> FalkorDB
    Procedural --> FileSystem
    Cache --> Postgres

    Python --> Episodic
    Python --> Semantic
    Python --> Procedural
    Python --> Cache
    MCP --> FalkorDB

    style Episodic fill:#85c1e9
    style Semantic fill:#85c1e9
    style Procedural fill:#85c1e9
    style Cache fill:#f9e79f
    style Postgres fill:#76d7c4
    style FalkorDB fill:#76d7c4

Episodic Memory¶

Purpose: Store events, interactions, and experiences with temporal ordering

Storage: PostgreSQL episodic_memory table Capabilities: Vector similarity search, time-series queries, decision tracking

Architecture¶

Implementation: /home/agent/projects/aegis-core/aegis/memory/episodic.py

Core Methods:

from aegis.memory.episodic import EpisodicMemory

# Record new event
memory_id = EpisodicMemory.record(
    event_type="task_complete",
    summary="Deployed dashboard service",
    details={"service": "dashboard", "duration": 120},
    importance="medium",
    tags=["deployment", "docker"]
)

# Recall recent events
recent = EpisodicMemory.recall_recent(limit=10, event_type="task_complete")

# Vector similarity search
similar = await EpisodicMemory.find_similar_async(
    query_text="deployment failures",
    limit=5
)

Event Types¶

Type	Purpose	Typical Count/Day
task_complete	Task completion	20-50
decision	Strategic decisions	5-10
error	Error occurrences	10-20
learning	Lessons learned	3-5
milestone	Achievements	1-3
research	Research findings	5-15
general	Miscellaneous	10-30

Decision Tracking¶

Purpose: Track decisions with alternatives and outcomes for learning

Workflow:

# 1. Before making a decision
decision_id, outcome_id = EpisodicMemory.record_decision(
    task_context="Choose deployment strategy",
    alternatives=["Docker Compose", "Kubernetes", "Bare metal"],
    chosen_option="Docker Compose",
    reasoning="Simpler for current scale, faster iteration"
)

# 2. After seeing the outcome
EpisodicMemory.update_outcome(
    decision_id=outcome_id,
    status="success",
    lessons_learned="Docker Compose was right choice for MVP. Consider K8s at 10K users."
)

# 3. Find similar past decisions
similar_decisions = EpisodicMemory.find_similar_decisions(
    task_context="deployment",
    limit=5
)

Decision Outcomes Table: - Links to episodic_memory via decision_id - Tracks alternatives, reasoning, outcome status - Stores lessons learned for future reference

Vector Embeddings¶

Model: Ollama nomic-embed-text (768 dimensions) Index: HNSW (Hierarchical Navigable Small World) Similarity Metric: Cosine similarity

Embedding Generation:

from aegis.llm.ollama import get_embedding

# Generate embedding for text
embedding = await get_embedding("Deployed dashboard service successfully")

# Store in episodic memory
EpisodicMemory.update_embedding(memory_id, embedding)

# Search by similarity
similar = await EpisodicMemory.find_similar_async(
    query_text="deployment issues",
    limit=5
)

Batch Embedding Generation:

# Generate embeddings for records without them
count = await EpisodicMemory.generate_missing_embeddings_async(batch_size=50)

Daily Cron Job: Generates embeddings at 03:30 UTC via scheduler

Episode Revisiting¶

Purpose: Surface relevant past experiences for current situations

Use Case: Before making a decision, recall similar past episodes and their outcomes

# Revisit similar episodes
context = await EpisodicMemory.revisit_similar_episodes(
    situation="Need to scale database to handle 10x traffic",
    limit=5,
    include_decisions=True,
    min_similarity=0.5
)

# Response includes:
# - similar_episodes: Past events with high similarity
# - related_decisions: Decisions with similar context
# - lessons: Extracted lessons learned
# - suggested_approaches: What worked before
# - warnings: What failed before

Example Output:

{
  "situation": "Need to scale database to handle 10x traffic",
  "similar_episodes": [
    {
      "id": 1234,
      "event_type": "milestone",
      "summary": "Scaled PostgreSQL with read replicas",
      "similarity": 0.87,
      "timestamp": "2024-01-15"
    }
  ],
  "related_decisions": [
    {
      "task_context": "Database scaling strategy",
      "chosen_option": "Read replicas + connection pooling",
      "outcome_status": "success"
    }
  ],
  "lessons": [
    {
      "source_decision": 42,
      "lesson": "Read replicas worked well for read-heavy workloads"
    }
  ],
  "suggested_approaches": [
    {
      "approach": "Add read replicas",
      "reasoning": "Previously successful for 5x scale",
      "confidence": "high"
    }
  ]
}

Semantic Memory¶

Purpose: Store factual knowledge, concepts, and learnings

Storage: PostgreSQL semantic_memory table + FalkorDB knowledge graph Capabilities: Concept lookup, category browsing, confidence tracking

Architecture¶

Implementation: /home/agent/projects/aegis-core/aegis/memory/semantic.py

Core Methods:

from aegis.memory.semantic import SemanticMemory

# Store new knowledge
memory_id = SemanticMemory.store(
    concept="Docker Compose",
    content="Container orchestration tool for defining multi-container apps",
    category="infrastructure",
    source="documentation",
    confidence=1.0,
    tags=["docker", "devops"]
)

# Lookup by concept
knowledge = SemanticMemory.lookup("Docker Compose")

# Search by content
results = SemanticMemory.search("container orchestration", limit=10)

# Browse by category
infra_knowledge = SemanticMemory.by_category("infrastructure", limit=50)

Knowledge Categories¶

Category	Purpose	Example Concepts
infrastructure	System architecture	Docker, Kubernetes, PostgreSQL
ai_models	LLM information	GPT-4, Claude, Gemini
programming	Code patterns	FastAPI, async/await, decorators
security	Security practices	HTTPS, API keys, encryption
business	Business logic	Pricing tiers, customer lifecycle

Confidence Tracking¶

Purpose: Track reliability of knowledge sources

Confidence Levels: - 1.0: Verified from authoritative source - 0.8: Derived from reliable source - 0.6: Inferred from context - 0.4: Uncertain, needs verification - 0.2: Speculative

Update Confidence:

# After verifying knowledge
SemanticMemory.update_confidence(memory_id, confidence=1.0)

Procedural Memory¶

Purpose: Store how-to knowledge, workflows, templates

Storage: File system (~/memory/procedural/) Format: Markdown files with YAML frontmatter

Architecture¶

Implementation: /home/agent/projects/aegis-core/aegis/memory/procedural.py

File Structure:

~/memory/procedural/
├── deployment/
│   ├── docker-compose-deploy.md
│   ├── kubernetes-deploy.md
│   └── rollback-procedure.md
├── monitoring/
│   ├── health-check-setup.md
│   └── alert-configuration.md
├── workflows/
│   ├── morning-routine.md
│   └── evening-summary.md
└── templates/
    ├── api-endpoint-template.md
    └── test-template.md

Example Workflow (deployment/docker-compose-deploy.md):

---
title: Docker Compose Deployment
tags: [docker, deployment, devops]
difficulty: intermediate
last_updated: 2024-01-25
---

# Docker Compose Deployment

## Prerequisites
- Docker installed
- docker-compose.yml configured
- Environment variables set

## Steps

1. **Build images**
   ```bash
   docker compose build
   ```

2. **Start services**
   ```bash
   docker compose up -d
   ```

3. **Verify health**
   ```bash
   docker ps
   curl http://localhost:8080/health
   ```

## Rollback
```bash
docker compose down
docker compose up -d --force-recreate

**Core Methods:**
```python
from aegis.memory.procedural import ProceduralMemory

# Store new workflow
ProceduralMemory.store(
    title="Deploy with Docker Compose",
    content=workflow_markdown,
    category="deployment",
    tags=["docker", "devops"]
)

# Retrieve workflow
workflow = ProceduralMemory.get("deployment/docker-compose-deploy")

# List all workflows in category
deployment_workflows = ProceduralMemory.list_category("deployment")

Knowledge Graph (FalkorDB + Graphiti)¶

Purpose: Extract entities and relationships from episodes for semantic querying

Storage: FalkorDB (Redis-compatible graph database) Framework: Graphiti (entity extraction, relationship modeling) Port: 6379 (Redis protocol), 3001 (Browser UI)

Architecture¶

Implementation: /home/agent/projects/aegis-core/aegis/memory/graphiti_client.py

Components: - LLM: GLM-4.7 via Z.ai (entity extraction) - Embedder: Ollama nomic-embed-text (semantic search) - Reranker: BGE-reranker-v2-m3 (cross-encoder ranking) - Graph DB: FalkorDB (storage)

Usage¶

Initialize Client:

from aegis.memory.graphiti_client import GraphitiClient

client = GraphitiClient()
await client.initialize()

Add Episode:

await client.add_episode(
    name="deployment-event-001",
    content="Deployed nginx service to production cluster using Docker Compose",
    source_description="journal",
    reference_time=datetime.now()
)

What Happens: 1. LLM extracts entities: nginx, production cluster, Docker Compose 2. LLM identifies relationships: nginx DEPLOYED_TO production cluster, nginx USES Docker Compose 3. Entities and relationships stored in graph 4. Embeddings generated for semantic search

Search Entities:

# Find entities related to "nginx deployment"
entities = await client.search_nodes("nginx deployment", limit=5)

# Returns:
# [
#   {"name": "nginx", "labels": ["Service"], "summary": "Web server"},
#   {"name": "production cluster", "labels": ["Infrastructure"]},
#   ...
# ]

Search Relationships:

# Find facts about deployment
facts = await client.search_facts("deployed", limit=5)

# Returns:
# [
#   {"fact": "nginx DEPLOYED_TO production cluster", "created_at": "2024-01-25T10:00:00Z"},
#   {"fact": "dashboard DEPLOYED_WITH docker-compose", "created_at": "2024-01-24T15:00:00Z"},
#   ...
# ]

Transcript Digester¶

Purpose: Bulk ingest Claude session history and journals into knowledge graph

Implementation: /home/agent/projects/aegis-core/aegis/memory/transcript_digester.py

Sources: - ~/.claude/history.jsonl - Claude Code session commands - ~/memory/journal/*.md - Daily journal entries

Usage:

from aegis.memory.transcript_digester import TranscriptDigester

digester = TranscriptDigester()
await digester.initialize()

# Ingest all sources
results = await digester.ingest_all()

# Selective ingestion
await digester.ingest_claude_history(after_date="2024-01-01")
await digester.ingest_journals(after_date="2024-01-01")

CLI Script:

cd /home/agent/projects/aegis-core
python scripts/ingest_transcripts.py --dry-run      # Preview
python scripts/ingest_transcripts.py --stats        # Statistics
python scripts/ingest_transcripts.py --after 2024-01-01  # Filter by date

Daily Cron: Runs at 03:00 UTC via scheduler

Episode Types Extracted: - decision: Strategic choices - implementation: Code changes - error: Failures and debugging - learning: Insights gained - milestone: Achievements - research: Information gathering - general: Other events

Graph Visualization¶

FalkorDB Browser: http://localhost:3001

Example Cypher Query:

// Find all entities related to "deployment"
MATCH (n)
WHERE n.name CONTAINS "deploy"
RETURN n
LIMIT 10

// Find deployment relationships
MATCH (a)-[r:DEPLOYED_TO]->(b)
RETURN a, r, b
LIMIT 20

// Find entity by UUID
MATCH (n {uuid: "abc123"})
RETURN n

Semantic Cache¶

Purpose: Memoize LLM responses to reduce API costs and latency

Storage: PostgreSQL llm_cache table TTL: Configurable (default: 24 hours)

Architecture¶

Implementation: /home/agent/projects/aegis-core/aegis/memory/cache.py

Core Methods:

from aegis.memory.cache import cache_llm_response, get_cached_llm_response

# Cache response
cache_llm_response(
    prompt="What is Docker Compose?",
    response="Docker Compose is a tool for defining...",
    model="glm-4.7",
    ttl=86400  # 24 hours
)

# Retrieve cached response
cached = get_cached_llm_response(
    prompt="What is Docker Compose?",
    model="glm-4.7"
)

Cache Key Generation:

import hashlib

def generate_cache_key(prompt: str, model: str, system: str = None, temperature: float = 0.7) -> str:
    key_parts = [prompt, model, system or "", str(temperature)]
    key_string = "|".join(key_parts)
    return hashlib.sha256(key_string.encode()).hexdigest()

Cache Hit Rate: ~40-60% (varies by workload)

Decorator Usage:

from aegis.memory.cache import cached_query

@cached_query(ttl=3600)
async def expensive_llm_call(prompt: str) -> str:
    return await query(prompt)

File-Based Memory¶

Location: ~/memory/

Directory Structure:

~/memory/
├── episodic/                # Event logs (deprecated, migrated to PostgreSQL)
├── semantic/                # Knowledge articles (Markdown)
│   ├── ai-agent-primitives.md
│   ├── aegis-architecture-patterns.md
│   └── ...
├── procedural/              # Workflows and templates
│   ├── deployment/
│   ├── monitoring/
│   └── workflows/
├── journal/                 # Daily journals
│   ├── 2024-01-25.md
│   └── ...
├── research/                # Research notes
├── intel/                   # Geopolitical intelligence
└── snapshots/               # Cognitive snapshots

Journal Files¶

Format: Markdown with date-based naming (YYYY-MM-DD.md)

Example (~/memory/journal/2024-01-25.md):

# 2024-01-25

## Tasks Completed
- Deployed dashboard v2.3.0
- Fixed WhatsApp webhook validation
- Updated architecture documentation

## Decisions Made
- **Decision**: Use FalkorDB over Neo4j for knowledge graph
  - **Reasoning**: Redis-compatible, easier to deploy in LXC
  - **Alternatives**: Neo4j, ArangoDB
  - **Outcome**: Pending

## Errors Encountered
- PostgreSQL connection pool exhausted (max_connections=100)
  - **Fix**: Increased to 200, added connection pooling

## Lessons Learned
- Vector search with HNSW is 10x faster than IVFFlat
- Docker host networking breaks Traefik routing

## Tomorrow's Priorities
- [ ] Complete API documentation
- [ ] Set up monitoring alerts
- [ ] Test WhatsApp command center

Daily Creation: Automated via morning routine (/morning)

Semantic Files¶

Purpose: Long-form knowledge articles

Format: Markdown with YAML frontmatter

Example (~/memory/semantic/docker-compose-patterns.md):

---
title: Docker Compose Best Practices
category: infrastructure
tags: [docker, devops, containers]
created: 2024-01-15
updated: 2024-01-25
confidence: 0.9
---

# Docker Compose Best Practices

## Service Naming
Use descriptive, hyphenated names...

## Health Checks
Always define health checks for services...

## Volume Management
...

Memory Integration Patterns¶

Morning Routine¶

Flow:

# 1. Recall yesterday's tasks
yesterday = EpisodicMemory.recall_recent(
    limit=20,
    event_type="task_complete"
)

# 2. Review pending decisions
pending = EpisodicMemory.get_decision_history(status="pending", limit=10)

# 3. Retrieve today's journal
journal = ProceduralMemory.get(f"journal/{today}")

# 4. Query knowledge graph for relevant context
entities = await graphiti_client.search_nodes("current priorities", limit=5)

Learning from Failures¶

Flow:

# 1. Record error in episodic memory
error_id = EpisodicMemory.record(
    event_type="error",
    summary="PostgreSQL connection failed",
    details={"error": str(e), "query": query},
    importance="high",
    tags=["database", "error"]
)

# 2. Search for similar past errors
similar_errors = await EpisodicMemory.find_similar_async(
    query_text="PostgreSQL connection failed",
    limit=5,
    event_type="error"
)

# 3. Extract lessons from past resolutions
for error in similar_errors:
    if error.get('details', {}).get('resolution'):
        print(f"Past resolution: {error['details']['resolution']}")

# 4. Store lesson in semantic memory
SemanticMemory.store(
    concept="PostgreSQL connection errors",
    content="Common causes: max_connections exhausted, firewall blocking...",
    category="troubleshooting",
    confidence=0.8
)

Context Augmentation¶

Pattern: Enrich LLM prompts with relevant memory

# 1. User asks a question
user_query = "How do I scale the database?"

# 2. Search episodic memory for relevant experiences
experiences = await EpisodicMemory.find_similar_async(
    query_text=user_query,
    limit=3
)

# 3. Search semantic memory for factual knowledge
knowledge = SemanticMemory.search(user_query, limit=3)

# 4. Search knowledge graph for relationships
entities = await graphiti_client.search_nodes(user_query, limit=5)
facts = await graphiti_client.search_facts("scaling", limit=5)

# 5. Construct enriched prompt
context = format_context(experiences, knowledge, entities, facts)
enriched_prompt = f"{context}\n\nUser question: {user_query}"

# 6. Query LLM
response = await query(enriched_prompt)

Memory Statistics¶

Current State¶

Memory Type	Records	Disk Size	Growth Rate
Episodic	18,682	10 MB	~100/day
Semantic	500	240 KB	~5/day
Procedural	150 files	5 MB	~1 file/week
Knowledge Graph	~5,000 nodes	50 MB	~50 nodes/day
Cache Entries	~1,000	2 MB	~100/day (rolling)

Embedding Coverage¶

Table	Total Records	With Embeddings	Coverage
episodic_memory	18,682	15,234	81%
semantic_memory	500	500	100%

Goal: 95% embedding coverage by Q2 2026

Future Memory Enhancements¶

Q1 2026: Performance¶

Implement memory prioritization (forget low-importance events)
Add memory consolidation (merge similar episodes)
Optimize vector index parameters
Add full-text search (PostgreSQL tsvector)

Q2 2026: Advanced Features¶

Multi-agent memory sharing
Memory versioning (track knowledge evolution)
Counterfactual reasoning (what-if scenarios)
Memory decay (reduce confidence over time)

Q3 2026: Scale¶

Distributed vector database (Qdrant/Weaviate)
Memory sharding by time period
Compression for old memories
Archive to cold storage (S3)

Q4 2026: Intelligence¶

Automatic memory summarization
Cross-reference detection (link related memories)
Anomaly detection (unusual patterns)
Predictive memory (anticipate future needs)