myelinmod 0.1.1

Universal Cognitive Memory Adapter — persistent memory, adaptive continuity, and lifelong learning for autonomous AI systems

Myelin

Universal Cognitive Memory Adapter for Autonomous AI Systems

---

What is Myelin?

Every AI model today — GPT-4, Claude, Llama, anything — forgets everything the moment a conversation ends. Myelin fixes that.

Myelin is a drop-in cognitive memory layer that wraps any LLM and gives it:

• Persistent long-term memory across sessions, users, and time
• 9 distinct memory types modelled on human cognition
• Adaptive learning — the system gets smarter from every interaction
• Intelligent forgetting — Ebbinghaus decay, not deletion
• Cross-agent shared cognition — multiple agents sharing the same memory fabric
• Full cognitive pipeline — reflection, dreaming, evolution, compression — running automatically in the background

In the same way RAG became the retrieval layer for modern AI, Myelin is the persistence and continuity layer for next-generation autonomous AI.

---

Install

pip install myelinmod

Requires Python 3.11+ and a running instance of the backend stack (see Infrastructure).

---

Prerequisites

Before running any example, start the backend services and set your API keys:

# 1. Start backend (Qdrant, Neo4j, PostgreSQL, Redis)
docker compose up -d

# 2. Run migrations
alembic upgrade head

import os

# Set your API keys (or put these in a .env file — never commit secrets)
os.environ["OPENAI_API_KEY"] = "sk-..."        # for OpenAI examples
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..." # for Anthropic examples

---

60-Second Quick Start

Jupyter / IPython: use await main() directly — an event loop is already running. Python script: use asyncio.run(main()) instead.

import os
import asyncio
from myelin import CognitiveAdapter, MyelinConfig

os.environ["OPENAI_API_KEY"] = "sk-..."  # or load from .env

async def main():
    config = MyelinConfig(agent_id="my-agent")

    async with CognitiveAdapter.session(config) as agent:
        client = agent.openai_client  # drop-in for openai.AsyncOpenAI

        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "What did we work on last week?"}],
        )
        print(response.choices[0].message.content)
        # The model now has memory of every past interaction.

# In a Jupyter notebook:
await main()

# In a Python script, replace the line above with:
# asyncio.run(main())

That's it. Zero code changes to your existing LLM calls. Myelin intercepts, enriches with memory, and learns from every response automatically.

---

Drop-in Adapters

Step 1 — Connect the fabric (run this cell once; reuse fabric in every example below)

import os
from myelin import HybridMemoryFabric, MyelinConfig

os.environ["OPENAI_API_KEY"] = "sk-..."
os.environ["ANTHROPIC_API_KEY"] = "sk-ant-..."

config = MyelinConfig(agent_id="agent-1")
fabric = HybridMemoryFabric.from_config(config)
await fabric.connect()
print("fabric connected")

OpenAI

# Before (plain openai — no memory):
# from openai import AsyncOpenAI
# client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])

# After — identical API, full cognitive memory:
from myelin import MyelinOpenAI

client = MyelinOpenAI(fabric=fabric, agent_id="agent-1")

response = await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)
print(response.choices[0].message.content)

Anthropic

# Requires: fabric connected (Step 1 above)
from myelin import MyelinAnthropic

client = MyelinAnthropic(fabric=fabric, agent_id="agent-1")

response = await client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello"}],
)
print(response.content[0].text)

Streaming

# Requires: client created (OpenAI or Anthropic step above)
# Memory is logged automatically after the stream completes
async for text_chunk in await client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Tell me a story"}],
    stream=True,
):
    print(text_chunk, end="", flush=True)

LangChain

# Requires: fabric connected (Step 1 above) and llm defined
from myelin.integrations.langchain import MyelinChatMemory
from langchain.chains import ConversationChain

memory = MyelinChatMemory(agent_id="agent-1", session_id="session-abc")
memory.bind_fabric(fabric)

chain = ConversationChain(llm=llm, memory=memory)

Local / Open-Source Models (Ollama, vLLM, LM Studio)

# Any OpenAI-compatible endpoint — requires: fabric connected (Step 1 above)
from myelin import MyelinOpenAI

client = MyelinOpenAI(
    fabric=fabric,
    agent_id="agent-1",
    openai_base_url="http://localhost:11434/v1",  # Ollama
    openai_api_key="ollama",
)

---

The 9 Memory Types

Type	Backend	What it stores
Episodic	Qdrant (vector)	Every interaction — time, context, outcome
Semantic	Neo4j (graph)	Concepts, facts, relationships
Procedural	PostgreSQL	Skills, how-to sequences, success rates
Causal	Neo4j (DAG)	Cause-effect chains, counterfactuals
Reflective	PostgreSQL	Self-model, capabilities, known biases
Predictive	PostgreSQL	Forecasts, simulations, expected outcomes
Short-Term	Redis (TTL)	Active session context, expires automatically
Preference	PostgreSQL	User preferences, emotional weights, affinities
Meta	PostgreSQL	Confidence in its own knowledge, known gaps

---

Cognitive Pipeline

Every call triggers this pipeline automatically:

User message
    ↓
[Myelin intercepts]
    ↓
Search all 9 memory stores simultaneously
    ↓
Score + rank via Memory Router (8-component relevance function)
    ↓
Inject top memories into system prompt
    ↓
Call LLM → return response to user
    ↓  (background — never blocks the user)
Write episodic memory
    ↓
Reflection (failure analysis, self-model update)
    ↓
Evolution (router weight update from retrieval quality)
    ↓
Dreaming (nightly: replay, compress, extract patterns)

---

Writing and Querying Memory Directly

# Requires: fabric connected (Step 1 in Drop-in Adapters above)
from myelin import EpisodicMemory, CQLQuery, EpisodeOutcome, OutcomeLabel

# Write any memory type directly
episode = EpisodicMemory(
    agent_id="agent-1",
    content="Deployed the new auth service. All tests passed.",
    outcome=EpisodeOutcome(label=OutcomeLabel.SUCCESS),
    tags=["deployment", "auth"],
)
memory_id = await fabric.write(episode)

# Query across all memory types
results = await fabric.query(CQLQuery(
    query_text="auth deployment",
    agent_id="agent-1",
    limit=10,
))

for r in results:
    print(f"[{r.memory.memory_type}] score={r.score.total_score:.2f}: {r.memory}")

# Convenience methods
episodes = await fabric.query_episodic("agent-1", "deployment", limit=5)
concepts = await fabric.query_semantic("agent-1", "authentication")
skill    = await fabric.find_skill("deploy_service", "agent-1")
prefs    = await fabric.get_preferences("agent-1", subject="code style")
gaps     = await fabric.get_known_gaps("agent-1")

---

Cross-Agent Memory Sharing

from myelin import MemorySyncLayer, SyncScope

# Agent A publishes a memory
sync_a = MemorySyncLayer("agent-a", namespace="my-team", scope=SyncScope.TEAM)
await sync_a.connect("redis://localhost:6379")
await sync_a.publish(episode)

# Agent B receives it automatically
sync_b = MemorySyncLayer("agent-b", namespace="my-team")
await sync_b.connect("redis://localhost:6379")
sync_b.on_receive(lambda event: print(f"Received from {event.source_agent_id}"))

---

Identity Core

from myelin import IdentityCore

# Create a cryptographically signed agent identity
identity, private_key_pem = IdentityCore.create(
    agent_id="agent-1",
    name="Research Assistant",
    core_values=["accuracy", "helpfulness", "honesty"],
    behavioral_constraints=["never fabricate citations"],
    primary_goal="Answer research questions accurately",
)
identity.save("agent_identity.json", private_key_pem)

# Load and verify on restart — raises ValueError if tampered
identity = IdentityCore.load("agent_identity.json")

---

Infrastructure

Myelin requires these backend services. Start them with the included Docker Compose:

docker compose up -d

Service	Purpose	Default port
Qdrant	Episodic + predictive vector search	6333
Neo4j	Semantic + causal graph	7687
TimescaleDB (PostgreSQL)	Procedural, reflective, preference, meta	5432
Redis	Short-term memory + cross-agent sync	6379

Run database migrations:

alembic upgrade head

---

Configuration

All configuration via environment variables or .env file:

MYELIN_AGENT_ID=my-agent
MYELIN_QDRANT_URL=http://localhost:6333
MYELIN_NEO4J_URI=bolt://localhost:7687
MYELIN_NEO4J_USER=neo4j
MYELIN_NEO4J_PASSWORD=password
MYELIN_POSTGRES_DSN=postgresql+asyncpg://myelin:myelin@localhost:5432/myelin
MYELIN_REDIS_URL=redis://localhost:6379
OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...

---

OpenTelemetry

from myelin.telemetry.tracing import configure_tracing

configure_tracing(
    service_name="my-ai-service",
    service_version="1.0.0",
    # auto-reads OTEL_EXPORTER_OTLP_ENDPOINT from env
)

Spans are emitted for: fabric.write, fabric.query, adapter.openai.create, adapter.anthropic.create, module.dreaming, module.reflection.

---

Cognitive Health

import os
from myelin import CognitiveAdapter, MyelinConfig

os.environ["OPENAI_API_KEY"] = "sk-..."

config = MyelinConfig(agent_id="my-agent")

async with CognitiveAdapter.session(config) as agent:
    health = await agent.cognitive_health()
    # {
    #   "stores": {"episodic": true, "semantic": true, ...},
    #   "budget_utilization": {"memory": 0.23, "context_tokens": 0.41},
    #   "warnings": [],
    # }

    # Manual triggers
    await agent.trigger_dreaming()
    await agent.trigger_compression()

---

Supported Python Versions

Python	Support
3.11	✅ Fully supported
3.12	✅ Fully supported
3.10	❌ Not supported (StrEnum, match)

---

Contributing

See CONTRIBUTING.md. Issues and PRs welcome.

Security

See SECURITY.md for reporting vulnerabilities.

License

MIT — free for commercial and open-source use.