Build a Self-Healing AI Assistant

Build an AI agent with typed memory systems and self-healing error recovery.

Introduction

30 Second Summary

Every AI chatbot you have used forgets you the moment you close the tab. It has no memory of past conversations and no ability to learn from its own mistakes.

In this project, you will build a command-line AI assistant called Atlas that remembers you across sessions, follows a structured behavioral contract, and heals itself when things go wrong.

What You'll Build

Atlas is a Python-based AI agent that uses typed memory systems, a behavioral specification engine, and a self-healing loop to demonstrate production-grade agent architecture patterns.

By the end of this project, you'll have:

  • A behavioral specification engine that controls Atlas's personality and decision-making through structured data instead of hardcoded prompts.
  • Three typed memory modules (episodic, semantic, procedural) that persist across sessions via JSON files.
  • A self-healing agent loop that catches errors, reflects on root causes, and stores recovery strategies for future use.
  • Secret Mission: Memory consolidation that compresses old episode histories into distilled semantic facts.

Are there any prerequisites?

You need Python installed on your machine, VS Code as your code editor, and an OpenRouter account with an API key (about $2 in credits is enough for this project).

You do not need any prior experience with AI agents or memory systems. This project teaches those patterns from scratch.

Before We Start

You are building an AI assistant that solves three specific problems most chatbots suffer from. Inconsistent behavior (fixed by a behavioral specification), amnesia between sessions (fixed by typed memory), and brittleness when things go wrong (fixed by a self-healing loop).

In this step, you will create the project folder, install two Python packages, and scaffold all the files your agent needs. By the end, you will have a clean workspace ready for building.

In this step, get ready to:
  • Install the OpenRouter SDK and python-dotenv package.
  • Create the project folder structure with all necessary files.
  • Configure your API key in a .env file and validate the SDK.
Install dependencies and create the project structure

Your agent needs two packages: the OpenRouter SDK for calling LLMs, and python-dotenv for loading environment variables from a file.

  • Press the Windows key to open your search bar.
  • Type PowerShell and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir self-healing-agent
cd self-healing-agent
  • Install the two required packages by running:
pip install openrouter==0.10.8 python-dotenv==1.2.2

What do these packages do?

The openrouter SDK gives you a type-safe Python client for calling any LLM model through the OpenRouter API. It handles authentication, request formatting, and response parsing.

The python-dotenv package reads key-value pairs from a .env file and loads them as environment variables. This keeps your API key out of your code.

  • You should see output ending with Successfully installed openrouter-0.10.8 python-dotenv-1.2.2 (or a notice that the packages are already installed).

Seeing a 'pip not found' error?

If PowerShell does not recognize pip, try pip3 or python -m pip install openrouter==0.10.8 python-dotenv==1.2.2 instead. This happens when Python's Scripts folder is not on your system PATH.

help me fix pip not being recognized in PowerShell

Now you need to create the folder structure for your agent's memory system and main files.

  • Create the memory subdirectory and all project files by running these commands:
New-Item -ItemType Directory -Path memory
New-Item -ItemType File -Path .env, config.py, agent.py, main.py
New-Item -ItemType File -Path memory/__init__.py, memory/episodic.py, memory/semantic.py, memory/procedural.py

What is this folder structure for?

The root files handle the agent's brain: config.py defines its personality, agent.py runs the self-healing loop, and main.py provides the command-line interface.

The memory/ folder holds three typed memory modules: episodic (what happened), semantic (what is known), and procedural (how to handle problems). Each module gets its own file so the architecture stays clean and testable.

  • Confirm the files exist by running:
ls
  • You should see .env, agent.py, config.py, main.py, and a memory folder in the output.
  • Check the memory folder contents by running:
ls memory
  • You should see __init__.py, episodic.py, procedural.py, and semantic.py listed.

Missing files or folder?

Make sure you ran the commands from inside the self-healing-agent folder. If you see an error about the path, double-check that you typed memory/__init__.py with a forward slash.

help me create the project file structure for my self-healing agent

Configure your environment variables

Your agent needs an OpenRouter API key to call LLM models. You will store it in a .env file so it stays out of your code.

  • Open the self-healing-agent folder in VS Code by running:
code .
  • In the Explorer sidebar on the left, click .env to open it.
  • Paste the following into .env:
OPENROUTER_API_KEY=your-api-key-here
  • Replace your-api-key-here with your actual OpenRouter API key. Your key starts with sk-or-v1-.

Where do I find my API key?

Go to openrouter.ai/settings/keys in your browser. You should see any keys you have already created. If you need a new one, click the create button on that page and copy the key immediately (it is only shown once).

  • Save .env by pressing Ctrl+S.

✔️ Awesome, I've got everything!

Great. Double check you have saved your .env file with your real API key.

ⓧ I'd like to double check the full code

Your .env file should look like this (with your own key replacing the placeholder):

OPENROUTER_API_KEY=your-api-key-here

Now validate that the OpenRouter SDK installed correctly and can be imported.

  • Switch back to your PowerShell terminal and verify the SDK by running:
python -c "from openrouter import OpenRouter; print('SDK ready')"
  • You should see SDK ready printed in your terminal.

Seeing an import error?

If you see ModuleNotFoundError: No module named 'openrouter', the package did not install into the Python version you are running. Try running python3 -c "from openrouter import OpenRouter; print('SDK ready')" to check if a different Python binary picks it up.

help me fix the openrouter import error in Python on Windows

Your workspace is ready. Next up, you will build the behavioral specification engine that gives your agent a consistent personality and decision-making framework.

Build the Behavioral Specification Engine

Your project structure is in place and the SDK is ready to go. Now it's time to define how your agent actually behaves.

Most chatbots rely on a single system prompt string to control personality. The problem is that this approach makes behavior hard to test, hard to update, and impossible to reason about programmatically. Instead, you will build a structured data object that defines identity, personality, boundaries, and decision rules as separate fields. The agent assembles this data into a prompt dynamically based on context.

In this step, get ready to:
  • Define the agent's identity and personality as structured data.
  • Set explicit boundaries and conditional decision rules.
  • Add memory directives that bridge behavior and memory systems.
Design the identity and personality structure

A behavioral specification separates what the agent is from how the prompt gets built. You define personality traits as data fields, and the agent code reads those fields to construct the system prompt at runtime.

  • Open config.py in VS Code.
  • Paste the beginning of the behavioral specification by copying this code:
BEHAVIORAL_SPEC = {
    "identity": {
        "name": "Atlas",
        "role": "Personal AI Assistant with Cognitive Memory",
        "version": "1.0"
    },
    "personality": {
        "tone": "warm but direct",
        "verbosity": "concise unless asked to elaborate",
        "formality": "casual professional"
    },

What does this code do?

The identity block gives your agent a name (Atlas), a role description, and a version number. These values get injected into the system prompt so the LLM knows who it is.

The personality block controls how Atlas communicates. Instead of writing "be concise and warm" in a prompt string, you store each trait as a named field. This makes personality testable and swappable without rewriting the whole prompt.

You won't see a change yet. The next two chunks complete the dictionary.

Define boundaries and conditional decision rules

Boundaries tell the agent what it must never do and what it must always do. Decision rules go further by changing the agent's behavior based on detected context.

  • In config.py, directly below the personality section (after the trailing comma), add the boundaries and decision rules:
    "boundaries": {
        "never": [
            "Pretend to be human",
            "Provide medical or legal advice",
            "Share information from other users"
        ],
        "always": [
            "Acknowledge uncertainty explicitly",
            "Reference relevant memories when available",
            "Ask for clarification rather than guessing"
        ]
    },
    "decision_rules": [
        {
            "condition": "user expresses frustration",
            "action": "Switch to empathetic mode, acknowledge the frustration, offer concrete help"
        },
        {
            "condition": "user asks about previous conversation",
            "action": "Search episodic memory first, then semantic memory"
        },
        {
            "condition": "task fails",
            "action": "Engage self-healing: reflect, diagnose, store lesson, retry with new strategy"
        }
    ],

What does this code do?

The boundaries block defines hard constraints. The never list prevents dangerous behaviors. The always list enforces habits like acknowledging uncertainty.

The decision_rules list is where behavior becomes dynamic. Each rule has a condition (what the agent detects) and an action (how it should respond). The third rule is what triggers the self-healing loop you will build in Step 4.

Why structured data instead of a system prompt string?

A plain-text system prompt is a black box. You cannot programmatically inspect or modify individual rules without parsing prose.

With a structured spec, your code can read BEHAVIORAL_SPEC["boundaries"]["never"] directly, add new rules at runtime, or swap personality profiles without touching the prompt-building logic.

You still won't see output yet. The final chunk closes out the dictionary and adds model configuration.

Define memory directives and model configuration

Memory directives create the bridge between behavior and memory. They tell the agent when to store information in each memory type. The memory system (which you will build next) handles how.

  • In config.py, directly below the decision_rules section (after the trailing comma), add the memory directives and close the dictionary:
    "memory_directives": {
        "store_facts": "When user shares personal info or preferences, store in semantic memory",
        "store_episodes": "After each conversation turn, note key topics discussed",
        "store_procedures": "When recovering from an error, store the recovery pattern in procedural memory"
    }
}

MODEL = "anthropic/claude-haiku-4.5"
MAX_RETRIES = 3
WORKING_MEMORY_LIMIT = 10

What does this code do?

The memory_directives section tells the agent when to use each memory type. store_facts triggers semantic memory for personal info. store_episodes triggers episodic memory after each turn. store_procedures triggers procedural memory during error recovery.

The three constants below the dictionary configure the agent's runtime. MODEL points to anthropic/claude-haiku-4.5 via OpenRouter. MAX_RETRIES controls how many self-healing attempts the agent gets before giving up. WORKING_MEMORY_LIMIT caps the number of recent messages sent to the LLM.

Why Claude Haiku 4.5?

Claude Haiku 4.5 costs $1 per million input tokens and $5 per million output tokens on OpenRouter. Claude Sonnet 4 costs $3/$15. For a learning project with a $2 budget, Haiku 4.5 gives you hundreds of interactions without risk of running out.

  • Save config.py.
  • Verify the file is valid Python by running this command in your terminal:
python -c "from config import BEHAVIORAL_SPEC; print(BEHAVIORAL_SPEC['identity']['name'])"

You should see Atlas printed in the terminal.

Seeing a SyntaxError?

Check that every opening brace { has a matching closing brace }, and every opening bracket [ has a matching ]. A missing comma between sections is the most common cause.

Make sure the BEHAVIORAL_SPEC dictionary closes with } before the MODEL line.

help me fix a SyntaxError in my config.py behavioral specification dictionary

✔️ Awesome, I've got everything!

Great. Double check you have saved config.py before moving on.

ⓧ I'd like to double check the full code

BEHAVIORAL_SPEC = {
    "identity": {
        "name": "Atlas",
        "role": "Personal AI Assistant with Cognitive Memory",
        "version": "1.0"
    },
    "personality": {
        "tone": "warm but direct",
        "verbosity": "concise unless asked to elaborate",
        "formality": "casual professional"
    },
    "boundaries": {
        "never": [
            "Pretend to be human",
            "Provide medical or legal advice",
            "Share information from other users"
        ],
        "always": [
            "Acknowledge uncertainty explicitly",
            "Reference relevant memories when available",
            "Ask for clarification rather than guessing"
        ]
    },
    "decision_rules": [
        {
            "condition": "user expresses frustration",
            "action": "Switch to empathetic mode, acknowledge the frustration, offer concrete help"
        },
        {
            "condition": "user asks about previous conversation",
            "action": "Search episodic memory first, then semantic memory"
        },
        {
            "condition": "task fails",
            "action": "Engage self-healing: reflect, diagnose, store lesson, retry with new strategy"
        }
    ],
    "memory_directives": {
        "store_facts": "When user shares personal info or preferences, store in semantic memory",
        "store_episodes": "After each conversation turn, note key topics discussed",
        "store_procedures": "When recovering from an error, store the recovery pattern in procedural memory"
    }
}

MODEL = "anthropic/claude-haiku-4.5"
MAX_RETRIES = 3
WORKING_MEMORY_LIMIT = 10

Your agent now has a complete behavioral contract. Next up, you will build the three typed memory modules that give Atlas the ability to remember conversations, store facts, and learn from its mistakes.

Implement the Typed Memory System

Your behavioral spec tells Atlas what to do and how to behave. But without memory, Atlas forgets everything the moment a session ends. It cannot recognize returning users, recall past conversations, or learn from its own mistakes.

In this step, you will build three typed memory modules that give Atlas persistent knowledge across sessions. Each module stores a different kind of information: what happened, what is known, and how to handle problems.

In this step, get ready to:
  • Build episodic memory for storing timestamped conversation summaries.
  • Build semantic memory for storing factual knowledge as structured triples.
  • Build procedural memory for storing error recovery strategies with reliability tracking.
Build episodic memory

Episodic memory records what happened. It stores timestamped conversation summaries so Atlas can recall previous interactions. The retrieval system scores results by keyword relevance and boosts recent episodes so newer memories surface first.

  • Open memory/episodic.py in VS Code and paste the following code:
import json
from datetime import datetime
from pathlib import Path


class EpisodicMemory:
    def __init__(self, storage_path="memory_data/episodic.json"):
        self.storage_path = Path(storage_path)
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        self.episodes = self._load()

    def _load(self):
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                return json.load(f)
        return []

    def _save(self):
        with open(self.storage_path, "w") as f:
            json.dump(self.episodes, f, indent=2)

    def store(self, summary, context=None):
        episode = {
            "id": len(self.episodes) + 1,
            "timestamp": datetime.now().isoformat(),
            "summary": summary,
            "context": context or {}
        }
        self.episodes.append(episode)
        self._save()
        return episode

    def recall(self, query, limit=5):
        query_lower = query.lower()
        scored = []
        for i, episode in enumerate(self.episodes):
            score = 0
            text = episode["summary"].lower()
            for word in query_lower.split():
                if word in text:
                    score += 1
            recency = (i + 1) / len(self.episodes)
            score += recency * 0.5
            if score > 0:
                scored.append((score, episode))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [ep for _, ep in scored[:limit]]

    def get_recent(self, limit=3):
        return self.episodes[-limit:] if self.episodes else []

What does this code do?

  • The __init__ method creates the memory_data/ directory automatically if it does not exist, then loads any previously saved episodes from disk.
  • store() creates a new episode with a timestamp and immediately persists it to the JSON file.
  • recall() searches episodes by keyword overlap, then adds a recency bonus so newer memories score higher. This prevents old, irrelevant episodes from always dominating results.
  • get_recent() returns the last N episodes without scoring. This is useful for building the system prompt context.
  • Save memory/episodic.py.

You will not see a change yet. The next chunk builds semantic memory, and you will test all three modules together at the end.

✔️ Awesome, I've got everything!

Great. Make sure you have saved memory/episodic.py before moving on.

ⓧ I'd like to double check the full code

import json
from datetime import datetime
from pathlib import Path


class EpisodicMemory:
    def __init__(self, storage_path="memory_data/episodic.json"):
        self.storage_path = Path(storage_path)
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        self.episodes = self._load()

    def _load(self):
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                return json.load(f)
        return []

    def _save(self):
        with open(self.storage_path, "w") as f:
            json.dump(self.episodes, f, indent=2)

    def store(self, summary, context=None):
        episode = {
            "id": len(self.episodes) + 1,
            "timestamp": datetime.now().isoformat(),
            "summary": summary,
            "context": context or {}
        }
        self.episodes.append(episode)
        self._save()
        return episode

    def recall(self, query, limit=5):
        query_lower = query.lower()
        scored = []
        for i, episode in enumerate(self.episodes):
            score = 0
            text = episode["summary"].lower()
            for word in query_lower.split():
                if word in text:
                    score += 1
            recency = (i + 1) / len(self.episodes)
            score += recency * 0.5
            if score > 0:
                scored.append((score, episode))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [ep for _, ep in scored[:limit]]

    def get_recent(self, limit=3):
        return self.episodes[-limit:] if self.episodes else []
Build semantic memory

Semantic memory stores what is known. It holds facts as subject-relation-value triples (for example: "user / lives in / Tokyo"). It uses upsert semantics, meaning if you store a fact that already exists, it updates the value instead of creating a duplicate.

  • Open memory/semantic.py and paste the following code:
import json
from datetime import datetime
from pathlib import Path


class SemanticMemory:
    def __init__(self, storage_path="memory_data/semantic.json"):
        self.storage_path = Path(storage_path)
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        self.facts = self._load()

    def _load(self):
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                return json.load(f)
        return []

    def _save(self):
        with open(self.storage_path, "w") as f:
            json.dump(self.facts, f, indent=2)

    def store(self, subject, relation, value, confidence=1.0):
        for existing in self.facts:
            if existing["subject"] == subject and existing["relation"] == relation:
                existing["value"] = value
                existing["confidence"] = confidence
                existing["updated"] = datetime.now().isoformat()
                self._save()
                return existing
        fact = {
            "id": len(self.facts) + 1,
            "subject": subject,
            "relation": relation,
            "value": value,
            "confidence": confidence,
            "created": datetime.now().isoformat(),
            "access_count": 0
        }
        self.facts.append(fact)
        self._save()
        return fact

    def recall(self, query, limit=5):
        query_lower = query.lower()
        scored = []
        for fact in self.facts:
            score = 0
            searchable = f"{fact['subject']} {fact['relation']} {fact['value']}".lower()
            for word in query_lower.split():
                if word in searchable:
                    score += 1
            if score > 0:
                fact["access_count"] += 1
                scored.append((score * fact["confidence"], fact))
        if scored:
            self._save()
        scored.sort(key=lambda x: x[0], reverse=True)
        return [f for _, f in scored[:limit]]

    def get_all(self):
        return self.facts

What does this code do?

  • store() loops through existing facts looking for a matching subject + relation pair. If one exists, it updates the value in place. Otherwise it creates a new fact entry. This prevents duplicate facts from piling up.
  • recall() scores facts by keyword match and multiplies by the fact's confidence value. Higher-confidence facts rank higher in results. It also increments access_count each time a fact is recalled, letting you track which knowledge gets used most.
  • Save memory/semantic.py.

You will not see a change yet. The next chunk builds procedural memory, then you will test everything together.

✔️ Awesome, I've got everything!

Great. Make sure you have saved memory/semantic.py before moving on.

ⓧ I'd like to double check the full code

import json
from datetime import datetime
from pathlib import Path


class SemanticMemory:
    def __init__(self, storage_path="memory_data/semantic.json"):
        self.storage_path = Path(storage_path)
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        self.facts = self._load()

    def _load(self):
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                return json.load(f)
        return []

    def _save(self):
        with open(self.storage_path, "w") as f:
            json.dump(self.facts, f, indent=2)

    def store(self, subject, relation, value, confidence=1.0):
        for existing in self.facts:
            if existing["subject"] == subject and existing["relation"] == relation:
                existing["value"] = value
                existing["confidence"] = confidence
                existing["updated"] = datetime.now().isoformat()
                self._save()
                return existing
        fact = {
            "id": len(self.facts) + 1,
            "subject": subject,
            "relation": relation,
            "value": value,
            "confidence": confidence,
            "created": datetime.now().isoformat(),
            "access_count": 0
        }
        self.facts.append(fact)
        self._save()
        return fact

    def recall(self, query, limit=5):
        query_lower = query.lower()
        scored = []
        for fact in self.facts:
            score = 0
            searchable = f"{fact['subject']} {fact['relation']} {fact['value']}".lower()
            for word in query_lower.split():
                if word in searchable:
                    score += 1
            if score > 0:
                fact["access_count"] += 1
                scored.append((score * fact["confidence"], fact))
        if scored:
            self._save()
        scored.sort(key=lambda x: x[0], reverse=True)
        return [f for _, f in scored[:limit]]

    def get_all(self):
        return self.facts
Build procedural memory and wire the exports

Procedural memory stores how to handle problems. It pairs error triggers with recovery strategies and tracks how often each strategy succeeds or fails. When the agent encounters an error it has seen before, it looks up the most reliable strategy instead of guessing from scratch.

  • Open memory/procedural.py and paste the following code:
import json
from datetime import datetime
from pathlib import Path


class ProceduralMemory:
    def __init__(self, storage_path="memory_data/procedural.json"):
        self.storage_path = Path(storage_path)
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        self.procedures = self._load()

    def _load(self):
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                return json.load(f)
        return []

    def _save(self):
        with open(self.storage_path, "w") as f:
            json.dump(self.procedures, f, indent=2)

    def store(self, trigger, strategy, outcome, category="general"):
        for existing in self.procedures:
            if existing["trigger"].lower() == trigger.lower():
                if outcome == "success":
                    existing["success_count"] += 1
                    existing["strategy"] = strategy
                else:
                    existing["failure_count"] += 1
                existing["last_used"] = datetime.now().isoformat()
                self._save()
                return existing
        procedure = {
            "id": len(self.procedures) + 1,
            "trigger": trigger,
            "strategy": strategy,
            "outcome": outcome,
            "category": category,
            "success_count": 1 if outcome == "success" else 0,
            "failure_count": 0 if outcome == "success" else 1,
            "created": datetime.now().isoformat(),
            "last_used": datetime.now().isoformat()
        }
        self.procedures.append(procedure)
        self._save()
        return procedure

    def find_strategy(self, error_description):
        error_lower = error_description.lower()
        matches = []
        for proc in self.procedures:
            trigger_lower = proc["trigger"].lower()
            score = sum(1 for word in error_lower.split() if word in trigger_lower)
            if score > 0:
                total = proc["success_count"] + proc["failure_count"]
                reliability = proc["success_count"] / max(total, 1)
                matches.append((score * reliability, proc))
        matches.sort(key=lambda x: x[0], reverse=True)
        return matches[0][1] if matches else None

    def get_all(self):
        return self.procedures

What does this code do?

  • store() checks if a matching trigger already exists. If it does, it increments the success or failure count rather than creating a duplicate. On success, it also updates the strategy to the latest working approach.
  • find_strategy() scores each stored procedure by keyword overlap with the error description, then multiplies by reliability (success rate). The most relevant AND most reliable strategy wins.
  • Save memory/procedural.py.

Now you need to wire up the exports so the rest of the project can import all three memory classes from the memory package.

  • Open memory/__init__.py and paste the following code:
from .episodic import EpisodicMemory
from .semantic import SemanticMemory
from .procedural import ProceduralMemory

What does this code do?

This __init__.py file turns the memory/ folder into a Python package. The imports let other files write from memory import EpisodicMemory instead of the longer from memory.episodic import EpisodicMemory.

  • Save memory/__init__.py.
  • Verify all three memory modules load correctly by running this command in your terminal:
python -c "from memory import EpisodicMemory, SemanticMemory, ProceduralMemory; print('All memory modules loaded')"

You should see All memory modules loaded printed in the terminal.

  • Now test that storage and retrieval work end-to-end by running this command:
python -c "from memory import EpisodicMemory, SemanticMemory, ProceduralMemory; e = EpisodicMemory(); e.store('test episode'); s = SemanticMemory(); s.store('user', 'likes', 'Python'); p = ProceduralMemory(); p.store('timeout', 'retry', 'success'); print('Stored and retrieved successfully')"
  • Confirm the JSON files were created by running:
ls memory_data

You should see episodic.json, semantic.json, and procedural.json listed in the output.

Seeing a ModuleNotFoundError?

  • Make sure you are running the command from inside the self-healing-agent folder (the one containing the memory/ directory). Run ls and confirm you see memory/ in the listing.
  • Check that memory/__init__.py exists and is not empty. Without this file, Python does not recognize memory/ as an importable package.
  • Verify the class names match exactly. Python imports are case-sensitive, so EpisodicMemory must be spelled identically in both the class definition and the import statement.

Help me debug my memory module imports

✔️ Awesome, I've got everything!

Great. Make sure all four files are saved: memory/episodic.py, memory/semantic.py, memory/procedural.py, and memory/__init__.py.

ⓧ I'd like to double check the full code

import json
from datetime import datetime
from pathlib import Path


class ProceduralMemory:
    def __init__(self, storage_path="memory_data/procedural.json"):
        self.storage_path = Path(storage_path)
        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
        self.procedures = self._load()

    def _load(self):
        if self.storage_path.exists():
            with open(self.storage_path, "r") as f:
                return json.load(f)
        return []

    def _save(self):
        with open(self.storage_path, "w") as f:
            json.dump(self.procedures, f, indent=2)

    def store(self, trigger, strategy, outcome, category="general"):
        for existing in self.procedures:
            if existing["trigger"].lower() == trigger.lower():
                if outcome == "success":
                    existing["success_count"] += 1
                    existing["strategy"] = strategy
                else:
                    existing["failure_count"] += 1
                existing["last_used"] = datetime.now().isoformat()
                self._save()
                return existing
        procedure = {
            "id": len(self.procedures) + 1,
            "trigger": trigger,
            "strategy": strategy,
            "outcome": outcome,
            "category": category,
            "success_count": 1 if outcome == "success" else 0,
            "failure_count": 0 if outcome == "success" else 1,
            "created": datetime.now().isoformat(),
            "last_used": datetime.now().isoformat()
        }
        self.procedures.append(procedure)
        self._save()
        return procedure

    def find_strategy(self, error_description):
        error_lower = error_description.lower()
        matches = []
        for proc in self.procedures:
            trigger_lower = proc["trigger"].lower()
            score = sum(1 for word in error_lower.split() if word in trigger_lower)
            if score > 0:
                total = proc["success_count"] + proc["failure_count"]
                reliability = proc["success_count"] / max(total, 1)
                matches.append((score * reliability, proc))
        matches.sort(key=lambda x: x[0], reverse=True)
        return matches[0][1] if matches else None

    def get_all(self):
        return self.procedures
from .episodic import EpisodicMemory
from .semantic import SemanticMemory
from .procedural import ProceduralMemory

Why keyword matching instead of embeddings?

Production memory systems often use vector embeddings for retrieval. That approach requires additional dependencies and infrastructure (a vector database, an embedding model, numpy).

Keyword matching with recency and confidence weighting teaches the same architectural pattern without the overhead. Once you understand how typed retrieval works, upgrading to embeddings later is a natural extension.

  • Clean up the test data so you start fresh. Delete the memory_data folder by running:
rm -rf memory_data

Your three memory modules are ready. Atlas can now remember what happened (episodic), what it knows (semantic), and how to fix things (procedural). Next up, you will build the agent loop that ties these memories into every conversation.

Build the Self-Healing Agent Loop

Your three memory modules can store and retrieve knowledge. Now you need an agent that actually uses them.

In this step, you will build the core SelfHealingAgent class in agent.py. This class assembles a dynamic system prompt from your behavioral spec and memory context, calls the LLM through OpenRouter, and wraps every call in a self-healing retry loop that learns from its own failures.

In this step, get ready to:
  • Assemble a system prompt dynamically from the behavioral spec and live memory context.
  • Build in-band memory extraction that stores facts and episodes after every response.
  • Create a self-healing loop that catches errors, reflects, and stores recovery strategies.
Implement system prompt assembly

A behavioral spec is a structured data object, not a hardcoded string. The agent assembles its system prompt at runtime by combining spec fields with live memory context. This means the same agent can behave differently depending on what it remembers.

  • Open agent.py in VS Code. It should be empty.
  • Add the imports, class definition, and __init__ method by copying this code:
import json
import os

from openrouter import OpenRouter
from memory import EpisodicMemory, SemanticMemory, ProceduralMemory
from config import BEHAVIORAL_SPEC, MODEL, MAX_RETRIES, WORKING_MEMORY_LIMIT


class SelfHealingAgent:
    def __init__(self):
        # Load API key from environment
        self.api_key = os.getenv("OPENROUTER_API_KEY")
        if not self.api_key:
            raise ValueError("OPENROUTER_API_KEY environment variable not set")

        # Set model and initialize all three memory systems
        self.model = MODEL
        self.episodic = EpisodicMemory()
        self.semantic = SemanticMemory()
        self.procedural = ProceduralMemory()
        self.conversation_history = []

What does this code do?

The imports pull in the OpenRouter SDK, your three memory classes, and all the config values from config.py.

The __init__ method reads the API key from your .env file (loaded via os.getenv), creates one instance of each memory type, and starts with an empty conversation history.

  • Save agent.py.
  • Test that the class instantiates by running this in your terminal:
python -c "from dotenv import load_dotenv; load_dotenv(); from agent import SelfHealingAgent; a = SelfHealingAgent(); print('Agent initialized')"

You should see Agent initialized printed to the terminal.

Seeing a ValueError or ImportError?

  • If you see OPENROUTER_API_KEY environment variable not set, make sure your .env file exists in the self-healing-agent folder and contains your key.
  • If you see ModuleNotFoundError for openrouter, confirm you installed it with pip install openrouter==0.10.8 in Step 1.
  • help me debug an ImportError or ValueError when instantiating SelfHealingAgent

Now add the two methods that dynamically assemble the system prompt. The first method gathers context from all three memory systems. The second stitches that context together with the behavioral spec.

  • In agent.py, add the _build_system_prompt and _gather_memory_context methods inside the SelfHealingAgent class, directly below __init__:
    def _build_system_prompt(self):
        spec = BEHAVIORAL_SPEC
        memories_context = self._gather_memory_context()

        # Assemble personality, boundaries, and rules from the spec
        system_prompt = f"""You are {spec['identity']['name']}, a {spec['identity']['role']}.

PERSONALITY:
- Tone: {spec['personality']['tone']}
- Verbosity: {spec['personality']['verbosity']}
- Formality: {spec['personality']['formality']}

BOUNDARIES:
Never: {', '.join(spec['boundaries']['never'])}
Always: {', '.join(spec['boundaries']['always'])}

DECISION RULES:
"""
        for rule in spec["decision_rules"]:
            system_prompt += f"- If {rule['condition']}: {rule['action']}\n"

        # Append memory context and the MEMORY_OPS response format instruction
        system_prompt += f"""
MEMORY CONTEXT:
{memories_context}

RESPONSE FORMAT:
After responding to the user, output a JSON block on a new line starting with MEMORY_OPS: followed by a JSON object.
Format: MEMORY_OPS: {{"facts": [{{"subject": "...", "relation": "...", "value": "..."}}], "episode_note": "brief note about this interaction"}}
If nothing to remember: MEMORY_OPS: {{"facts": [], "episode_note": ""}}
"""
        return system_prompt

    def _gather_memory_context(self):
        parts = []

        # Pull recent episodes (what happened)
        recent_episodes = self.episodic.get_recent(3)
        if recent_episodes:
            parts.append("Recent conversations:")
            for ep in recent_episodes:
                parts.append(f"  - [{ep['timestamp'][:10]}] {ep['summary']}")

        # Pull known facts (what is true)
        all_facts = self.semantic.get_all()
        if all_facts:
            parts.append("\nKnown facts about the user:")
            for fact in all_facts[-10:]:
                parts.append(f"  - {fact['subject']} {fact['relation']} {fact['value']}")

        # Pull learned procedures (how to handle problems)
        procedures = self.procedural.get_all()
        if procedures:
            parts.append("\nLearned procedures:")
            for proc in procedures[-5:]:
                total = proc["success_count"] + proc["failure_count"]
                reliability = proc["success_count"] / max(total, 1)
                parts.append(f"  - When '{proc['trigger']}': {proc['strategy']} (reliability: {reliability:.0%})")

        return "\n".join(parts) if parts else "No memories yet. This is a new user."

What does this code do?

  • _build_system_prompt reads the BEHAVIORAL_SPEC dictionary and formats it into natural language sections (personality, boundaries, decision rules). It also injects a MEMORY_OPS response format instruction that tells the LLM to include structured memory operations in every reply.
  • _gather_memory_context pulls the latest data from all three memory systems and formats it as a text block. This is what gives the agent "awareness" of past conversations, known facts, and learned recovery strategies.
  • The MEMORY_OPS instruction keeps memory extraction in-band. Instead of making a separate LLM call to decide what to remember, the agent outputs memory operations alongside its response. This halves your token cost.

Why in-band memory extraction instead of a separate call?

Making a second LLM call to extract memory would double the token cost per interaction. The in-band approach asks the model to include structured memory operations alongside its response. This is a common production pattern where the model does reasoning and bookkeeping in a single pass.

  • Save agent.py.
  • Verify the prompt assembly works by running this in your terminal:
python -c "from dotenv import load_dotenv; load_dotenv(); from agent import SelfHealingAgent; a = SelfHealingAgent(); print(a._build_system_prompt()[:200])"

You should see the first 200 characters of the assembled prompt, starting with You are Atlas, a Personal AI Assistant.

Seeing a syntax error or AttributeError?

  • Check that both methods are indented inside the SelfHealingAgent class (4 spaces of indentation for the def lines).
  • Make sure the triple-quoted f-strings use {{ and }} for literal curly braces in the JSON format instructions.
  • help me fix a syntax error in my _build_system_prompt method
Implement the chat method with memory operations

The agent needs three utility methods to handle the LLM call and process memory operations from the response. These form the message pipeline that every conversation turn flows through.

  • Add the following three methods below _gather_memory_context in the SelfHealingAgent class:
    def _call_llm(self, messages):
        # Send messages to OpenRouter and return the response content
        with OpenRouter(api_key=self.api_key) as client:
            response = client.chat.send(
                model=self.model,
                messages=messages
            )
        return response.choices[0].message.content

    def _parse_memory_ops(self, response_text):
        # Split the MEMORY_OPS JSON from the user-facing response
        if "MEMORY_OPS:" not in response_text:
            return response_text, None

        parts = response_text.split("MEMORY_OPS:")
        user_response = parts[0].strip()
        try:
            memory_ops = json.loads(parts[1].strip())
            return user_response, memory_ops
        except (json.JSONDecodeError, IndexError):
            return user_response, None

    def _process_memory_ops(self, memory_ops):
        # Store extracted facts in semantic memory, episodes in episodic memory
        if not memory_ops:
            return
        for fact in memory_ops.get("facts", []):
            if all(k in fact for k in ("subject", "relation", "value")):
                self.semantic.store(fact["subject"], fact["relation"], fact["value"])
        episode_note = memory_ops.get("episode_note", "")
        if episode_note:
            self.episodic.store(episode_note)

What does this code do?

  • _call_llm opens a connection to OpenRouter using the SDK's context manager, sends your messages with client.chat.send(), and returns the text content from the first choice.
  • _parse_memory_ops splits the raw LLM response at the MEMORY_OPS: marker. Everything before it is what the user sees. Everything after it is a JSON block the agent uses internally.
  • _process_memory_ops takes the parsed JSON and routes facts to semantic memory and episode notes to episodic memory. It validates that each fact has all three required keys before storing.
  • Save agent.py. You will not see a change yet. The next substep's chat method uses these three utilities together.
Implement the self-healing loop

The self-healing pattern is what separates demo agents from production agents. When a call fails, the agent checks procedural memory for a known recovery strategy. If none exists, it asks the LLM to reflect on the failure and invent one. Either way, the outcome gets stored so the agent handles the same problem faster next time.

  • Add the _self_heal method below _process_memory_ops:
    def _self_heal(self, error, user_message, attempt):
        # Check if we already know how to handle this error
        existing_strategy = self.procedural.find_strategy(str(error))

        if existing_strategy:
            # Use the learned strategy
            healing_prompt = (
                f"A previous error occurred: {error}\n"
                f"I found a learned strategy for this: '{existing_strategy['strategy']}'\n"
                f"Apply this strategy to recover and respond to: {user_message}"
            )
        else:
            # Ask the LLM to reflect and diagnose the failure
            healing_prompt = (
                f"An error occurred while processing the user's message.\n"
                f"Error: {error}\n"
                f"User's original message: {user_message}\n"
                f"Attempt: {attempt} of {MAX_RETRIES}\n\n"
                f"Reflect on what went wrong and provide a response to the user. "
                f"Also suggest what strategy should be used if this happens again."
            )

        messages = [
            {"role": "system", "content": "You are a self-healing AI assistant. Diagnose the error and recover gracefully."},
            {"role": "user", "content": healing_prompt}
        ]

        try:
            recovery_response = self._call_llm(messages)
            # Store successful recovery in procedural memory
            self.procedural.store(
                trigger=str(error)[:100],
                strategy=f"Recovery: adjusted approach on attempt {attempt}",
                outcome="success",
                category="error_recovery"
            )
            return recovery_response
        except Exception:
            # Store failed recovery attempt
            self.procedural.store(
                trigger=str(error)[:100],
                strategy=f"Recovery attempt {attempt} also failed",
                outcome="failure",
                category="error_recovery"
            )
            return None

What does this code do?

  • First, it searches procedural memory using find_strategy() to see if the agent has encountered this error before.
  • If a known strategy exists, it builds a prompt that references that strategy. If not, it builds a reflection prompt asking the LLM to diagnose the root cause.
  • Either way, the outcome (success or failure) gets stored back into procedural memory. Over time, the agent accumulates a library of recovery patterns ranked by reliability.
  • Save agent.py. You will not see a change yet. The chat method below connects everything together.

Now add the chat and end_session methods. The chat method wraps every LLM call in a retry loop that invokes self-healing on failure.

  • Add the chat and end_session methods below _self_heal:
    def chat(self, user_message):
        # Add the user's message to conversation history
        self.conversation_history.append({"role": "user", "content": user_message})

        # Build the full message list with system prompt and recent history
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            *self.conversation_history[-WORKING_MEMORY_LIMIT:]
        ]

        # Retry loop with self-healing on each failure
        for attempt in range(1, MAX_RETRIES + 1):
            try:
                raw_response = self._call_llm(messages)
                user_response, memory_ops = self._parse_memory_ops(raw_response)
                self._process_memory_ops(memory_ops)
                self.conversation_history.append({"role": "assistant", "content": user_response})
                return user_response

            except Exception as e:
                print(f"\n[Self-Healing] Error on attempt {attempt}: {e}")
                recovery = self._self_heal(e, user_message, attempt)
                if recovery:
                    self.conversation_history.append({"role": "assistant", "content": recovery})
                    return f"[Recovered] {recovery}"

        return "I encountered an issue I could not recover from. Please try again."

    def end_session(self):
        # Store a session summary in episodic memory when quitting
        if len(self.conversation_history) > 0:
            message_count = len(self.conversation_history)
            summary = f"Session with {message_count} messages"
            self.episodic.store(summary, context={"message_count": message_count})

What does this code do?

  • chat appends the user's message to history, assembles the full prompt (system + last 10 messages), and attempts the LLM call up to 3 times.
  • On success, it parses out the MEMORY_OPS JSON, processes memory operations, stores the assistant's response in history, and returns the user-facing text.
  • On failure, it prints a [Self-Healing] log, calls _self_heal, and returns the recovery response if healing succeeds.
  • end_session logs a summary episode to episodic memory when the user quits, so the agent remembers the session happened.
  • Save agent.py.
  • Validate the self-healing loop by temporarily breaking the model name. Open config.py and find this line:
MODEL = "anthropic/claude-haiku-4.5"
  • Change it temporarily to an invalid slug:
MODEL = "anthropic/broken-model-test"
  • Save config.py, then run the following test in your terminal:
python -c "from dotenv import load_dotenv; load_dotenv(); from agent import SelfHealingAgent; a = SelfHealingAgent(); print(a.chat('Hello'))"

You should see [Self-Healing] Error on attempt 1: followed by error details. The agent attempts recovery up to 3 times. After all attempts, it either returns a [Recovered] response or the final fallback message.

  • Now restore the correct model in config.py:
MODEL = "anthropic/claude-haiku-4.5"
  • Save config.py.
  • Check that a procedure was stored by running:
python -c "from memory import ProceduralMemory; p = ProceduralMemory(); print(p.get_all())"

You should see at least one stored procedure with a trigger related to the model error and a recovery strategy.

Not seeing self-healing output?

  • If the script exits immediately with no output, check that you have a valid API key in .env. The self-healing loop itself calls the LLM, so it needs a working key even when testing with a broken model slug.
  • If you see NameError for MAX_RETRIES, confirm the import line at the top of agent.py includes MAX_RETRIES and WORKING_MEMORY_LIMIT.
  • help me debug why my self-healing loop is not activating

✔️ Awesome, I've got everything!

Great. Double check you have saved agent.py and restored the correct model in config.py.

ⓧ I'd like to double check the full code

Here is the complete agent.py file for comparison:

import json
import os

from openrouter import OpenRouter
from memory import EpisodicMemory, SemanticMemory, ProceduralMemory
from config import BEHAVIORAL_SPEC, MODEL, MAX_RETRIES, WORKING_MEMORY_LIMIT


class SelfHealingAgent:
    def __init__(self):
        self.api_key = os.getenv("OPENROUTER_API_KEY")
        if not self.api_key:
            raise ValueError("OPENROUTER_API_KEY environment variable not set")

        self.model = MODEL
        self.episodic = EpisodicMemory()
        self.semantic = SemanticMemory()
        self.procedural = ProceduralMemory()
        self.conversation_history = []

    def _build_system_prompt(self):
        spec = BEHAVIORAL_SPEC
        memories_context = self._gather_memory_context()

        system_prompt = f"""You are {spec['identity']['name']}, a {spec['identity']['role']}.

PERSONALITY:
- Tone: {spec['personality']['tone']}
- Verbosity: {spec['personality']['verbosity']}
- Formality: {spec['personality']['formality']}

BOUNDARIES:
Never: {', '.join(spec['boundaries']['never'])}
Always: {', '.join(spec['boundaries']['always'])}

DECISION RULES:
"""
        for rule in spec["decision_rules"]:
            system_prompt += f"- If {rule['condition']}: {rule['action']}\n"

        system_prompt += f"""
MEMORY CONTEXT:
{memories_context}

RESPONSE FORMAT:
After responding to the user, output a JSON block on a new line starting with MEMORY_OPS: followed by a JSON object.
Format: MEMORY_OPS: {{"facts": [{{"subject": "...", "relation": "...", "value": "..."}}], "episode_note": "brief note about this interaction"}}
If nothing to remember: MEMORY_OPS: {{"facts": [], "episode_note": ""}}
"""
        return system_prompt

    def _gather_memory_context(self):
        parts = []

        recent_episodes = self.episodic.get_recent(3)
        if recent_episodes:
            parts.append("Recent conversations:")
            for ep in recent_episodes:
                parts.append(f"  - [{ep['timestamp'][:10]}] {ep['summary']}")

        all_facts = self.semantic.get_all()
        if all_facts:
            parts.append("\nKnown facts about the user:")
            for fact in all_facts[-10:]:
                parts.append(f"  - {fact['subject']} {fact['relation']} {fact['value']}")

        procedures = self.procedural.get_all()
        if procedures:
            parts.append("\nLearned procedures:")
            for proc in procedures[-5:]:
                total = proc["success_count"] + proc["failure_count"]
                reliability = proc["success_count"] / max(total, 1)
                parts.append(f"  - When '{proc['trigger']}': {proc['strategy']} (reliability: {reliability:.0%})")

        return "\n".join(parts) if parts else "No memories yet. This is a new user."

    def _call_llm(self, messages):
        with OpenRouter(api_key=self.api_key) as client:
            response = client.chat.send(
                model=self.model,
                messages=messages
            )
        return response.choices[0].message.content

    def _parse_memory_ops(self, response_text):
        if "MEMORY_OPS:" not in response_text:
            return response_text, None

        parts = response_text.split("MEMORY_OPS:")
        user_response = parts[0].strip()
        try:
            memory_ops = json.loads(parts[1].strip())
            return user_response, memory_ops
        except (json.JSONDecodeError, IndexError):
            return user_response, None

    def _process_memory_ops(self, memory_ops):
        if not memory_ops:
            return
        for fact in memory_ops.get("facts", []):
            if all(k in fact for k in ("subject", "relation", "value")):
                self.semantic.store(fact["subject"], fact["relation"], fact["value"])
        episode_note = memory_ops.get("episode_note", "")
        if episode_note:
            self.episodic.store(episode_note)

    def _self_heal(self, error, user_message, attempt):
        existing_strategy = self.procedural.find_strategy(str(error))

        if existing_strategy:
            healing_prompt = (
                f"A previous error occurred: {error}\n"
                f"I found a learned strategy for this: '{existing_strategy['strategy']}'\n"
                f"Apply this strategy to recover and respond to: {user_message}"
            )
        else:
            healing_prompt = (
                f"An error occurred while processing the user's message.\n"
                f"Error: {error}\n"
                f"User's original message: {user_message}\n"
                f"Attempt: {attempt} of {MAX_RETRIES}\n\n"
                f"Reflect on what went wrong and provide a response to the user. "
                f"Also suggest what strategy should be used if this happens again."
            )

        messages = [
            {"role": "system", "content": "You are a self-healing AI assistant. Diagnose the error and recover gracefully."},
            {"role": "user", "content": healing_prompt}
        ]

        try:
            recovery_response = self._call_llm(messages)
            self.procedural.store(
                trigger=str(error)[:100],
                strategy=f"Recovery: adjusted approach on attempt {attempt}",
                outcome="success",
                category="error_recovery"
            )
            return recovery_response
        except Exception:
            self.procedural.store(
                trigger=str(error)[:100],
                strategy=f"Recovery attempt {attempt} also failed",
                outcome="failure",
                category="error_recovery"
            )
            return None

    def chat(self, user_message):
        self.conversation_history.append({"role": "user", "content": user_message})

        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            *self.conversation_history[-WORKING_MEMORY_LIMIT:]
        ]

        for attempt in range(1, MAX_RETRIES + 1):
            try:
                raw_response = self._call_llm(messages)
                user_response, memory_ops = self._parse_memory_ops(raw_response)
                self._process_memory_ops(memory_ops)
                self.conversation_history.append({"role": "assistant", "content": user_response})
                return user_response

            except Exception as e:
                print(f"\n[Self-Healing] Error on attempt {attempt}: {e}")
                recovery = self._self_heal(e, user_message, attempt)
                if recovery:
                    self.conversation_history.append({"role": "assistant", "content": recovery})
                    return f"[Recovered] {recovery}"

        return "I encountered an issue I could not recover from. Please try again."

    def end_session(self):
        if len(self.conversation_history) > 0:
            message_count = len(self.conversation_history)
            summary = f"Session with {message_count} messages"
            self.episodic.store(summary, context={"message_count": message_count})

Your agent can now assemble dynamic prompts, call the LLM, extract memory from responses, and recover from its own failures. Next up, you will build the CLI interface and test cross-session memory persistence.

Create the CLI and Test Cross-Session Memory

Your agent has a behavioral spec, typed memory, and a self-healing loop. Right now there is no way to actually talk to it.

In this step, you will build a command-line interface that lets you chat with Atlas, inspect its memory state, and test that memories persist across sessions.

In this step, get ready to:
  • Build an interactive CLI with memory inspection commands.
  • Test memory persistence by running the agent across multiple sessions.
  • Trigger the self-healing loop and verify that recovery strategies are stored.
Build the command-line interface

The CLI is what turns your agent from a collection of modules into something you can interact with. It handles user input, routes slash commands to the right memory module, and passes everything else to the chat() method.

  • Open main.py in VS Code and add the imports, environment loading, and agent setup by pasting this code:
import os
from dotenv import load_dotenv
from agent import SelfHealingAgent

# Load environment variables from .env file
load_dotenv()


def main():
    # Print a startup banner
    print("=" * 50)
    print("  Self-Healing AI Assistant with Cognitive Memory")
    print("=" * 50)
    print()

    # Initialize the agent, catching missing API key errors
    try:
        agent = SelfHealingAgent()
    except ValueError as e:
        print(f"Error: {e}")
        print("Create a .env file with: OPENROUTER_API_KEY=your-key-here")
        return

    # Display agent identity and available commands
    spec = agent._build_system_prompt().split("\n")[0]
    print(f"Agent ready: {spec}")
    print("Commands: /memories, /facts, /procedures, /quit")
    print("-" * 50)

What does this code do?

load_dotenv() reads your .env file and loads the API key into the environment so the agent can access it.

The try/except block catches the ValueError that SelfHealingAgent.__init__ raises when the key is missing, so the learner gets a helpful message instead of a traceback.

  • Now add the interactive input loop with command handling below the setup code you just pasted:
    while True:
        try:
            user_input = input("\nYou: ").strip()
        except (KeyboardInterrupt, EOFError):
            agent.end_session()
            print("\nSession saved. Goodbye!")
            break

        if not user_input:
            continue

        # Route slash commands to the appropriate memory module
        if user_input == "/quit":
            agent.end_session()
            print("\nSession saved. Goodbye!")
            break
        elif user_input == "/memories":
            episodes = agent.episodic.get_recent(5)
            if episodes:
                print("\n[Recent Episodes]")
                for ep in episodes:
                    print(f"  [{ep['timestamp'][:10]}] {ep['summary']}")
            else:
                print("\n[No episodes stored yet]")
            continue
        elif user_input == "/facts":
            facts = agent.semantic.get_all()
            if facts:
                print("\n[Known Facts]")
                for fact in facts:
                    print(f"  {fact['subject']} {fact['relation']} {fact['value']}")
            else:
                print("\n[No facts stored yet]")
            continue
        elif user_input == "/procedures":
            procs = agent.procedural.get_all()
            if procs:
                print("\n[Learned Procedures]")
                for proc in procs:
                    total = proc["success_count"] + proc["failure_count"]
                    reliability = proc["success_count"] / max(total, 1)
                    print(f"  Trigger: {proc['trigger']}")
                    print(f"  Strategy: {proc['strategy']}")
                    print(f"  Reliability: {reliability:.0%}\n")
            else:
                print("\n[No procedures learned yet]")
            continue

        # Pass non-command input to the agent's chat method
        response = agent.chat(user_input)
        print(f"\nAtlas: {response}")


if __name__ == "__main__":
    main()

What does this code do?

  • /memories pulls recent episodes from episodic memory, showing timestamped summaries of past interactions.
  • /facts displays all semantic facts as subject-relation-value triples (e.g., "user lives_in Tokyo").
  • /procedures shows learned recovery strategies with their reliability percentages.
  • /quit calls end_session() to log the session as an episode before exiting. Pressing Ctrl+C does the same thing.
  • Save main.py.
  • Test that the agent starts up by running this command in your terminal:
python main.py
  • You should see the banner, an "Agent ready:" line showing Atlas's identity, and a list of available commands.
  • Type /quit to exit for now.

Seeing an import error or missing key message?

  • If you see "OPENROUTER_API_KEY environment variable not set", check that your .env file exists in the same folder as main.py and contains your key.
  • If you see a ModuleNotFoundError, make sure you ran pip install openrouter==0.10.8 python-dotenv==1.2.2 in the same terminal session.

Why won't my agent start up?

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py before moving on.

ⓧ I'd like to double check the full code

import os
from dotenv import load_dotenv
from agent import SelfHealingAgent

load_dotenv()


def main():
    print("=" * 50)
    print("  Self-Healing AI Assistant with Cognitive Memory")
    print("=" * 50)
    print()

    try:
        agent = SelfHealingAgent()
    except ValueError as e:
        print(f"Error: {e}")
        print("Create a .env file with: OPENROUTER_API_KEY=your-key-here")
        return

    spec = agent._build_system_prompt().split("\n")[0]
    print(f"Agent ready: {spec}")
    print("Commands: /memories, /facts, /procedures, /quit")
    print("-" * 50)

    while True:
        try:
            user_input = input("\nYou: ").strip()
        except (KeyboardInterrupt, EOFError):
            agent.end_session()
            print("\nSession saved. Goodbye!")
            break

        if not user_input:
            continue

        if user_input == "/quit":
            agent.end_session()
            print("\nSession saved. Goodbye!")
            break
        elif user_input == "/memories":
            episodes = agent.episodic.get_recent(5)
            if episodes:
                print("\n[Recent Episodes]")
                for ep in episodes:
                    print(f"  [{ep['timestamp'][:10]}] {ep['summary']}")
            else:
                print("\n[No episodes stored yet]")
            continue
        elif user_input == "/facts":
            facts = agent.semantic.get_all()
            if facts:
                print("\n[Known Facts]")
                for fact in facts:
                    print(f"  {fact['subject']} {fact['relation']} {fact['value']}")
            else:
                print("\n[No facts stored yet]")
            continue
        elif user_input == "/procedures":
            procs = agent.procedural.get_all()
            if procs:
                print("\n[Learned Procedures]")
                for proc in procs:
                    total = proc["success_count"] + proc["failure_count"]
                    reliability = proc["success_count"] / max(total, 1)
                    print(f"  Trigger: {proc['trigger']}")
                    print(f"  Strategy: {proc['strategy']}")
                    print(f"  Reliability: {reliability:.0%}\n")
            else:
                print("\n[No procedures learned yet]")
            continue

        response = agent.chat(user_input)
        print(f"\nAtlas: {response}")


if __name__ == "__main__":
    main()
Test memory persistence across sessions

The real test of a memory system is whether the agent remembers you after you leave. You will run two separate sessions and verify that facts from the first session show up in the second.

  • Start your first session by running:
python main.py
  • Share some personal information with Atlas. Type something like:
I live in Tokyo
  • Wait for Atlas to respond, then share another fact:
I prefer dark mode
  • After Atlas responds, type /facts to inspect what was stored in semantic memory.
  • You should see entries like "user lives_in Tokyo" and "user prefers dark mode" (the exact wording depends on how the LLM chose to structure the facts).
  • Type /quit to end the first session.

What happened when you quit?

The end_session() method stored a summary episode in memory_data/episodic.json. The semantic facts from your conversation were already written to memory_data/semantic.json in real time as Atlas processed each response.

  • Start a second session by running:
python main.py
  • Ask Atlas what it remembers about you:
What do you know about me?
  • Atlas should reference the facts from your previous session (living in Tokyo, preferring dark mode) in its response.
  • Type /memories to see the episodic log from your first session.
  • Type /quit to end this session.

Atlas doesn't remember anything?

  • Open memory_data/semantic.json directly. If it is empty or contains [], the LLM did not output a valid MEMORY_OPS: line. Try being more explicit in your first session (e.g., "Remember this: I live in Tokyo").
  • Make sure you are running python main.py from inside the self-healing-agent folder so the relative path to memory_data/ resolves correctly.

Why isn't my agent remembering facts between sessions?

Test self-healing in action

The self-healing loop activates when an API call fails. You can trigger it by temporarily pointing the agent at a model slug that does not exist.

  • In config.py, find this line near the bottom of the file:
MODEL = "anthropic/claude-haiku-4.5"
  • Replace it with an invalid model slug:
MODEL = "anthropic/fake-model-that-does-not-exist"
  • Save config.py and run the agent:
python main.py
  • Type any message (e.g., "Hello") and watch the terminal output.
  • You should see [Self-Healing] Error on attempt 1: messages as the agent detects the failure and invokes the healing loop.
  • The self-healing method will also fail (since it uses the same broken model), so the agent will exhaust all 3 retries.
  • Type /procedures to verify that recovery attempts were stored in procedural memory.
  • Type /quit to exit.

Why did procedures get stored even though recovery failed?

The _self_heal() method stores outcomes regardless of success or failure. Failed strategies get a failure_count increment, which lowers their reliability score over time. This teaches the agent which approaches do not work.

  • Now restore the correct model. In config.py, change the MODEL line back to:
MODEL = "anthropic/claude-haiku-4.5"
  • Save config.py.
  • Run the agent one final time and confirm it responds normally:
python main.py
  • Type a message and verify that Atlas responds without any self-healing output.
  • Type /procedures one more time. You should still see the procedure entry from your broken-model test. The agent has learned from its failure.
  • Type /quit to exit.

Agent still failing after restoring the model?

  • Double-check that the MODEL value in config.py is exactly anthropic/claude-haiku-4.5 with no typos or extra spaces.
  • Verify that your OpenRouter account still has available credits at openrouter.ai/settings/keys.

Why is my agent still triggering self-healing after I fixed the model?

  • Confirm that your memory_data/ folder contains populated JSON files by running:
ls memory_data/
  • You should see episodic.json, semantic.json, and procedural.json listed. All three should contain data from your test sessions.

Your agent is fully operational. It remembers you, follows its behavioral spec, and learns from its own failures. Next up, the Secret Mission challenges you to add memory consolidation that compresses raw episodes into distilled facts.

Secret mission

Add Memory Consolidation

Your agent remembers individual episodes, but it never distills patterns from them. What if it could compress weeks of conversations into a handful of core facts? Add memory consolidation to make Atlas smarter over time.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs. You only spend OpenRouter credits when the agent is actively running.

Resources you used:

  • Local project directory self-healing-agent/ on your Desktop with all Python files.
  • Memory data directory memory_data/ containing JSON files (episodic, semantic, procedural).
  • Installed Python packages: openrouter and python-dotenv.
  • OpenRouter API credits consumed (approximately $0.10-0.30 depending on conversation length).

Keep everything running

No action needed. Choose this if you want to keep building on your agent or accumulate more memory over time.

  • Your memory_data/ directory will continue to grow as you have more conversations with Atlas.
  • The agent only costs money when you actively run it and chat. No background charges.
  • Your OpenRouter API key remains stored in the .env file inside the project directory.

Pause - I'll come back to this later

There are no running processes or paid services to pause. The agent only uses API credits while it is actively running in your terminal.

  • If you want to free up disk space without losing your code, delete only the memory data by running this from inside your self-healing-agent/ folder:
rm -rf memory_data/

Your code stays intact. Next time you run the agent, it will start with a fresh memory.

Delete - I don't want to use this again

Remove all project files, installed packages, and start fresh.

  • Uninstall the Python packages by running:
pip uninstall openrouter python-dotenv -y
  • Delete the entire project directory. On macOS/Linux, run:
cd ~/Desktop
rm -rf self-healing-agent/
  • On Windows, run:
cd ~\Desktop
Remove-Item -Recurse -Force self-healing-agent

This removes all code, memory data, and your .env file containing the API key reference.

  • Your OpenRouter API key still exists on your account at openrouter.ai/settings/keys. If you want to revoke it, visit that page and delete the key.

Nice Work!

Nice Work!

Nice work! You just built a production-grade AI assistant from scratch that remembers conversations, follows a structured behavioral contract, and heals itself when things go wrong.

You've learned how to:

  • Build a behavioral specification engine that controls agent personality through structured data instead of hardcoded prompts.
  • Implement three typed memory systems (episodic, semantic, procedural) with JSON persistence and keyword-based retrieval.
  • Create a self-healing agent loop that catches errors, searches for known recovery strategies, reflects on failures, and stores lessons in procedural memory.
  • Secret Mission: Added memory consolidation that compresses episode histories into distilled semantic facts.

Ready to quiz yourself?