myelinmod 0.1.0

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

---

60-Second Quick Start

import asyncio
from myelin import CognitiveAdapter, MyelinConfig

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.

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

OpenAI

# Before
from openai import AsyncOpenAI
client = AsyncOpenAI()

# After — identical API, full cognitive memory
from myelin import MyelinOpenAI, HybridMemoryFabric, MyelinConfig
fabric = HybridMemoryFabric.from_config(MyelinConfig(agent_id="agent-1"))
await fabric.connect()
client = MyelinOpenAI(fabric=fabric, agent_id="agent-1")

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

Anthropic

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"}],
)

Streaming

# Both adapters support streaming — memory is logged after 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

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

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

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.