Engramia

Reusable execution memory and evaluation infrastructure for AI agent frameworks.

Status: v0.6.5 — 1200+ tests / 80%+ coverage. Core library · REST API · 9 framework integrations · multi-tenancy · RBAC · async jobs · observability · data governance · ROI analytics. See GitHub Issues for planned work and discussion.

What it is

Engramia solves the problem every agent framework has: agents don't learn from previous runs.

LangChain, CrewAI, OpenAI Agents SDK, Pydantic AI, AutoGen, and similar frameworks are stateless — every run starts from scratch. Engramia is a memory layer you add beneath any framework, and it:

• Remembers what worked (success patterns with time-decay)
• Finds relevant agents for a new task (semantic search + eval weighting)
• Evaluates code quality (multi-evaluator with variance detection)
• Improves automatically (feedback injection, pattern aging)
• Composes multi-agent pipelines from proven components (Experimental)

Extracted from Agent Factory V2 — a system that reached a 93% task success rate over 254 runs using Engramia as its memory layer.

Who this is for

Primary users:

• AI platform teams building multi-agent pipelines
• Agent builders using LangChain, CrewAI, OpenAI Agents SDK, Pydantic AI, AutoGen, or custom frameworks
• Automation studios running repeated agentic workflows

Not designed for:

• End users without agent systems
• Pure ML/training workflows
• Single-run, one-shot LLM tasks

Feature maturity

Installation

# Base (JSON storage, no LLM/embeddings provider)
pip install engramia

# With OpenAI provider (recommended to start)
pip install "engramia[openai]"

# REST API + PostgreSQL
pip install "engramia[openai,api,postgres]"

Optional extras

Quick start

from engramia import Memory
from engramia.providers import OpenAIProvider, OpenAIEmbeddings, JSONStorage

mem = Memory(
    llm=OpenAIProvider(model="gpt-4.1"),
    embeddings=OpenAIEmbeddings(),
    storage=JSONStorage(path="./engramia_data"),
)

Python API Reference

mem.learn(task, code, eval_score, output=None) → LearnResult

Records the result of a run. Stores the success pattern and updates metrics.

result = mem.learn(
    task="Parse CSV file and compute statistics",
    code="import csv\nimport statistics\n...",
    eval_score=8.5,
    output="mean=42.3, std=7.1",  # optional: agent stdout
)
print(result.stored)        # True
print(result.pattern_count) # total number of patterns

• eval_score — number 0–10, how well the agent completed the task
• Pattern is automatically deduplicated against existing similar patterns (Jaccard > 0.7)

mem.recall(task, limit=5, deduplicate=True, eval_weighted=True) → list[Match]

Finds relevant success patterns for a new task using semantic search.

matches = mem.recall(task="Read CSV and calculate averages", limit=5)

for m in matches:
    print(f"{m.similarity:.2f} | score={m.pattern.success_score:.1f} | {m.pattern.task}")
    print(f"  key: {m.pattern_key}")   # use for delete_pattern()

Each Match contains:

• similarity — cosine similarity of embeddings (0.0–1.0)
• reuse_tier — "duplicate" / "adapt" / "fresh" based on similarity thresholds
• pattern_key — storage key for delete_pattern()
• pattern — Pattern object with task, design, success_score, reuse_count

Parameters:

• deduplicate=True — groups patterns of the same task (Jaccard > 0.7), returns only top-scoring per group
• eval_weighted=True — similarity is multiplied by a multiplier [0.5, 1.0] based on eval score; unrated patterns receive 0.75

mem.evaluate(task, code, output=None, num_evals=3) → EvalResult

Runs N independent LLM evaluations and aggregates the results. Requires an llm provider.

result = mem.evaluate(
    task="Parse CSV file",
    code="import csv\n...",
    output="done",    # optional
    num_evals=3,      # number of parallel LLM evaluations (min 1)
)

print(result.median_score)       # aggregated score (0–10)
print(result.variance)           # score variance across runs
print(result.high_variance)      # True if variance > 1.5
print(result.feedback)           # recommendation from the worst run
print(result.adversarial_detected)  # True if code contains hardcoded output

• Evaluations run in parallel (ThreadPoolExecutor)
• Feedback comes from the worst run (most relevant for improvement)
• Adversarial code detection (hardcoded output instead of computation)

mem.compose(task) → Pipeline (Experimental)

Decomposes a task into a staged pipeline from existing success patterns. Requires an llm provider.

Experimental: This feature works best as an assistive tool. Pipeline validity depends on pattern coverage and LLM output quality. Do not treat composed pipelines as guaranteed production-ready outputs.

pipeline = mem.compose(task="Fetch stock data, compute moving average, write report")

print(f"valid={pipeline.valid}, errors={pipeline.contract_errors}")
for stage in pipeline.stages:
    print(f"[{stage.task}]  reads={stage.reads}  writes={stage.writes}")

• LLM decomposes the task into 2–4 stages
• Each stage is matched against success patterns via semantic search
• Contract validation verifies data flow consistency (reads/writes chain) including cycle detection
• Falls back to a single-stage pipeline if the LLM fails

mem.get_feedback(task_type=None, limit=5) → list[str]

Returns recurring feedback patterns for injection into prompts.

feedback = mem.get_feedback(limit=4)
# ["Add error handling for missing input files.",
#  "Validate CSV headers before processing.", ...]

• Returns only feedback with count >= 2 (recurring issues)
• Sorted by frequency and recency (score × count)
• Suitable for automatic injection into the coder's system prompt

mem.delete_pattern(pattern_key) → bool

Permanently deletes a stored pattern. Returns True if the pattern existed.

matches = mem.recall(task="Parse CSV")
deleted = mem.delete_pattern(matches[0].pattern_key)
print(deleted)  # True

mem.run_aging() → int

Applies time-decay to all success patterns. Returns the number of removed patterns.

pruned = mem.run_aging()
print(f"Removed {pruned} outdated patterns")

• Decay: 2% per week (success_score *= 0.98^weeks)
• Pattern is removed if success_score < 0.1
• Recommended to run periodically (e.g., once a week)

mem.metrics → Metrics

Current metrics of the memory instance.

m = mem.metrics

print(m.runs)            # total number of recorded runs
print(m.success_rate)    # fraction of successful runs
print(m.avg_eval_score)  # average eval score (None if no evals)
print(m.pattern_count)   # current number of success patterns
print(m.pipeline_reuse)  # number of runs where an existing pattern was used

mem.evolve_prompt(role, current_prompt) → EvolutionResult (Experimental)

Generates an improved prompt based on recurring quality issues.

Experimental: Returns a candidate for manual review and A/B testing — not for direct automatic deployment.

result = mem.evolve_prompt(role="coder", current_prompt="You are a coder...")
if result.accepted:
    print(result.improved_prompt)
    print(f"Changes: {result.changes}")

• Analyzes top feedback patterns from eval history
• LLM generates an improved version of the prompt
• Returns a candidate for manual/automated A/B testing

mem.analyze_failures(min_count=1) → list[FailureCluster] (Experimental)

Groups recurring errors into clusters for identifying systemic issues.

Experimental: Cluster quality depends on pattern volume and LLM classification accuracy.

clusters = mem.analyze_failures(min_count=2)
for c in clusters:
    print(f"{c.representative} (count={c.total_count}, members={len(c.members)})")

mem.register_skills(pattern_key, skills) / mem.find_by_skills(required)

Skill registry for capability-based pattern search.

# Register
matches = mem.recall(task="Parse CSV")
mem.register_skills(matches[0].pattern_key, ["csv_parsing", "statistics"])

# Find
results = mem.find_by_skills(["csv_parsing"], match_all=True)

mem.export() → list[dict] / mem.import_data(records, overwrite=False) → int

Backup and migration of patterns (JSON storage → PostgreSQL or vice versa).

# Export all patterns to a JSONL file
import json

records = mem.export()
with open("backup.jsonl", "w") as f:
    for r in records:
        f.write(json.dumps(r) + "\n")

# Import from backup into a new instance
with open("backup.jsonl") as f:
    records = [json.loads(line) for line in f]

new_mem = Memory(embeddings=embeddings, storage=postgres_storage)
imported = new_mem.import_data(records)
print(f"Imported {imported} patterns")

Exceptions

Engramia uses a custom exception hierarchy for precise error handling:

from engramia import (
    EngramiaError, ProviderError, ValidationError,
    StorageError, QuotaExceededError, AuthorizationError,
)

try:
    result = mem.evaluate(task, code)
except ProviderError:
    # LLM provider is not configured
    pass
except ValidationError:
    # Invalid input (empty task, code too long, ...)
    pass
except QuotaExceededError:
    # Billing quota exceeded
    pass
except EngramiaError:
    # Any Engramia exception
    pass

Framework integrations

Engramia integrates with every major agent framework. Each integration automatically recalls relevant patterns before a run and learns from the result after.

OpenAI Agents SDK

pip install "engramia[openai-agents]"

from agents import Agent, Runner
from engramia.sdk.openai_agents import EngramiaRunHooks, engramia_instructions

agent = Agent(
    name="coder",
    instructions=engramia_instructions(mem, base="You are a senior developer."),
)
result = await Runner.run(agent, "Build a CSV parser", hooks=EngramiaRunHooks(mem))

Anthropic Agent SDK

pip install "engramia[anthropic-agents]"

from engramia.sdk.anthropic_agents import engramia_query

async for message in engramia_query(mem, prompt="Build a CSV parser"):
    print(message)
# Automatically recalls context → injects into system_prompt → learns from result.

Pydantic AI

pip install "engramia[pydantic-ai]"

from pydantic_ai import Agent
from engramia.sdk.pydantic_ai import EngramiaCapability

agent = Agent('openai:gpt-4o', capabilities=[EngramiaCapability(mem)])
result = agent.run_sync("Build a CSV parser")

AutoGen

pip install "engramia[autogen]"

from autogen_agentchat.agents import AssistantAgent
from engramia.sdk.autogen import EngramiaMemory, learn_from_result

agent = AssistantAgent(name="coder", model_client=client, memory=[EngramiaMemory(mem)])
result = await agent.run(task="Build a CSV parser")
learn_from_result(mem, task="Build a CSV parser", result=result)

LangChain

pip install "engramia[langchain]"

from engramia.sdk.langchain import EngramiaCallback

callback = EngramiaCallback(mem, auto_learn=True, auto_recall=True)
chain.invoke(input, config={"callbacks": [callback]})

CrewAI

pip install "engramia[crewai]"

from engramia.sdk.crewai import EngramiaCrewCallback

callback = EngramiaCrewCallback(mem, auto_learn=True, auto_recall=True)
result = callback.kickoff(crew)

REST API client (any language)

from engramia.sdk.webhook import EngramiaWebhook

client = EngramiaWebhook(url="http://localhost:8000", api_key="sk-...")
client.learn(task="Parse CSV", code=code, eval_score=8.5)
matches = client.recall(task="Read CSV and compute averages")

Generic framework (EngramiaBridge)

from engramia.sdk.bridge import EngramiaBridge

bridge = EngramiaBridge(data_path="./engramia_data")
context = bridge.before_run("Build a CSV parser")
result = my_agent(task, system_context=context)
bridge.after_run("Build a CSV parser", code=result, eval_score=8.0)

REST API

Starting up

# JSON storage (dev, no DB)
docker compose up

# PostgreSQL storage (prod)
ENGRAMIA_STORAGE=postgres \
ENGRAMIA_DATABASE_URL=postgresql://user:pass@localhost:5432/engramia \
OPENAI_API_KEY=sk-... \
docker compose up

After startup: Swagger UI at http://localhost:8000/docs

Configuration (env vars)

Endpoints

All endpoints are available under the /v1/ prefix:

Examples

# Learn
curl -X POST http://localhost:8000/v1/learn \
  -H "Content-Type: application/json" \
  -d '{"task": "Parse CSV", "code": "import csv", "eval_score": 8.5}'

# Recall
curl -X POST http://localhost:8000/v1/recall \
  -H "Content-Type: application/json" \
  -d '{"task": "Read CSV and compute averages", "limit": 3}'

# Metrics
curl http://localhost:8000/v1/metrics

# Health
curl http://localhost:8000/v1/health

Authentication

# Set keys
ENGRAMIA_API_KEYS=my-secret-key docker compose up

# Use key
curl -H "Authorization: Bearer my-secret-key" http://localhost:8000/v1/metrics

Security configuration

Observability configuration

PostgreSQL storage

Starting with pgvector backend:

# 1. Uncomment pgvector service in docker-compose.yml

# 2. Start
ENGRAMIA_STORAGE=postgres \
ENGRAMIA_DATABASE_URL=postgresql://engramia:engramia@pgvector:5432/engramia \
docker compose up

# 3. Apply migrations (first run)
docker compose exec engramia-api alembic upgrade head

Or without Docker:

pip install "engramia[openai,postgres]"

from engramia.providers.postgres import PostgresStorage
storage = PostgresStorage(database_url="postgresql://...")
mem = Memory(embeddings=OpenAIEmbeddings(), storage=storage, llm=OpenAIProvider())

Provider configuration

OpenAI (recommended)

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

from engramia.providers import OpenAIProvider, OpenAIEmbeddings, JSONStorage

mem = Memory(
    llm=OpenAIProvider(model="gpt-4.1"),
    embeddings=OpenAIEmbeddings(model="text-embedding-3-small"),
    storage=JSONStorage(path="./engramia_data"),
)

Embeddings only, without LLM

mem = Memory(
    embeddings=OpenAIEmbeddings(),
    storage=JSONStorage(path="./engramia_data"),
    llm=None,  # default
)

mem.learn(...)    # ✅ works
mem.recall(...)   # ✅ works
mem.evaluate(...) # ❌ ProviderError: evaluate() requires llm=...

CLI

# Installation
pip install "engramia[cli]"

# Initialize
engramia init --path ./engramia_data

# Start REST API server
engramia serve --host 0.0.0.0 --port 8000

# Metrics and statistics
engramia status --path ./engramia_data

# Semantic search
engramia recall "Parse CSV and compute statistics" --limit 5

# Pattern aging (decay + prune)
engramia aging --path ./engramia_data

MCP Server

Engramia can be run as an MCP server (Model Context Protocol) and connected directly to Claude Desktop, Cursor, Windsurf, or VS Code Copilot.

Installation

pip install "engramia[openai,mcp]"

Starting up

engramia-mcp

The server runs via stdio transport — the MCP client starts it as a subprocess automatically.

Client configuration

Claude Desktop (~/.config/claude/claude_desktop_config.json on Linux/macOS, %APPDATA%\Claude\claude_desktop_config.json on Windows):

{
  "mcpServers": {
    "engramia": {
      "command": "engramia-mcp",
      "env": {
        "ENGRAMIA_DATA_PATH": "/path/to/engramia_data",
        "OPENAI_API_KEY": "sk-..."
      }
    }
  }
}

Cursor / Windsurf — same JSON format in the MCP servers settings of the respective IDE.

Available MCP tools

Configuration (env vars)

The MCP server uses the same env vars as the REST API:

Architecture

engramia/
├── memory.py            # Memory facade — thin delegator (~165 LOC)
├── types.py             # Pydantic models (Scope, AuthContext, Pattern, ...)
├── exceptions.py        # EngramiaError hierarchy (ProviderError, StorageError, ...)
├── _context.py          # Scope contextvar: get_scope / set_scope / reset_scope
├── _util.py             # Helpers (extract_json_from_llm, jaccard, reuse_tier)
├── _factory.py          # Provider factory from env vars
│
├── core/                # Pattern storage + evaluation
│   ├── services/             # Business logic (extracted from Memory god object)
│   │   ├── learning.py       # LearningService — store patterns, embeddings, ROI
│   │   ├── recall.py         # RecallService — semantic search, dedup, eval-weighted
│   │   ├── evaluation.py     # EvaluationService — multi-evaluator scoring + feedback
│   │   └── composition.py    # CompositionService — LLM pipeline decomposition
│   ├── success_patterns.py   # Aging, reuse boost
│   ├── eval_store.py         # Eval history + quality multiplier
│   ├── eval_feedback.py      # Feedback clustering + decay
│   ├── metrics.py            # Run statistics
│   └── skill_registry.py     # Capability-based pattern tagging
│
├── reuse/               # Reuse engine
│   ├── matcher.py            # Semantic search + eval weighting
│   ├── composer.py           # LLM pipeline decomposition (Experimental)
│   └── contracts.py          # Data-flow validation + cycle detection
│
├── eval/
│   └── evaluator.py          # MultiEvaluator (concurrent, median, variance)
│
├── providers/
│   ├── base.py               # ABC: LLMProvider, EmbeddingProvider, StorageBackend
│   ├── openai.py             # OpenAI LLM + embeddings
│   ├── anthropic.py          # Anthropic/Claude LLM provider
│   ├── local_embeddings.py   # sentence-transformers (no API key)
│   ├── json_storage.py       # JSON storage (thread-safe, atomic writes)
│   └── postgres.py           # PostgreSQL + pgvector (scope-aware queries)
│
├── api/                 # REST API
│   ├── app.py                # App factory, lifespan, startup/shutdown
│   ├── routes.py             # Core endpoints (learn, recall, evaluate, ...)
│   ├── auth.py               # Multi-mode auth (auto/env/db/dev)
│   ├── keys.py               # API key management router (/v1/keys)
│   ├── permissions.py        # RBAC: 4 roles, require_permission() factory
│   ├── deps.py               # Dependency injection (Memory singleton, AuthContext)
│   ├── schemas.py            # Request/response Pydantic models
│   ├── audit.py              # Structured audit logging
│   ├── middleware.py         # SecurityHeaders, RateLimit, BodySize, RequestID, Timing
│   └── prom_metrics.py       # Prometheus counter/histogram definitions
│
├── jobs/                # Async job queue
│   ├── service.py            # JobService — submit, poll, cancel, retry/backoff
│   ├── worker.py             # JobWorker — daemon thread, ThreadPoolExecutor
│   └── dispatch.py           # Operation → Memory method dispatcher
│
├── analytics/           # ROI analytics
│   ├── collector.py          # ROICollector — fire-and-ignore event recorder
│   ├── aggregator.py         # ROIAggregator — hourly/daily/weekly rollups
│   └── models.py             # ROIEvent, ROIRollup, RecallOutcome, LearnSummary
│
├── governance/          # Data governance (GDPR)
│   ├── redaction.py          # PII/secrets redaction pipeline
│   ├── retention.py          # RetentionManager — per-tenant/project TTL policies
│   ├── deletion.py           # ScopedDeletion — GDPR Art. 17 right to erasure
│   ├── export.py             # DataExporter — NDJSON export (GDPR Art. 20)
│   └── lifecycle.py          # Async lifecycle jobs (retention_cleanup, compact_audit)
│
├── telemetry/           # Observability (opt-in, zero overhead when disabled)
│   ├── tracing.py            # OTel init + @traced decorator
│   └── metrics.py            # Prometheus histogram/counter definitions
│
├── db/                  # Database
│   ├── models.py             # SQLAlchemy 2.x models (13 migrations applied)
│   └── migrations/           # Alembic (001_initial → 013_cloud_users)
│
├── evolution/           # Prompt evolution + failure clustering (Experimental)
│   ├── prompt_evolver.py
│   └── failure_cluster.py
│
├── sdk/                 # Framework integrations (9 adapters)
│   ├── openai_agents.py      # OpenAI Agents SDK — RunHooks + dynamic instructions
│   ├── anthropic_agents.py   # Anthropic Agent SDK — query wrapper + hooks
│   ├── pydantic_ai.py        # Pydantic AI — Capability (before/after run)
│   ├── autogen.py            # AutoGen — Memory ABC for AssistantAgent
│   ├── langchain.py          # LangChain — EngramiaCallback
│   ├── crewai.py             # CrewAI — EngramiaCrewCallback
│   ├── bridge.py             # EngramiaBridge — drop-in for any agent factory
│   └── webhook.py            # Lightweight HTTP SDK client (stdlib only)
│
├── cli/                 # CLI tool (Typer + Rich)
│
└── mcp/                 # MCP server (stdio transport)
    └── server.py             # 7 MCP tools (learn, recall, evaluate, ...)

Implementation status

Development and testing

# Install for development
pip install -e ".[dev,openai]"

# Run tests
pytest

# With coverage report
pytest --cov=engramia --cov-report=term-missing

Tests do not require API keys — they use FakeEmbeddings (deterministic vectors from MD5 hash) and a mocked LLM. FastAPI tests use TestClient from httpx.

Origin

Extracted from Agent Factory V2 — a self-improving AI agent factory. Agent Factory V2 remains as an open-source reference implementation that demonstrates Engramia working in a production-grade multi-agent system.

License

Engramia is licensed under Business Source License 1.1 (BUSL-1.1).

• ✅ Free for: personal use, evaluation, research
• ❌ Not allowed without a commercial license:
  • production use in commercial environments
  • SaaS / hosted services
  • integration into paid products
  • building competing products

On the Change Date specified in LICENSE.txt, the license automatically converts to Apache 2.0.

See LICENSE.txt for full terms.

For commercial licenses: support@engramia.dev

License FAQ

Why does PyPI show "UNKNOWN" or no license classifier? BUSL-1.1 is not OSI-approved, so PyPI lacks a dedicated classifier. Engramia ships the full BUSL-1.1 text in LICENSE.txt and declares it via license = {file = "LICENSE.txt"} in pyproject.toml. Package scanners that flag "Proprietary" for non-OSI licenses are technically correct — BUSL-1.1 is source-available, not open source or proprietary in the classical sense. On the Change Date, the license converts to Apache 2.0 (OSI-approved).

Can I use Engramia in my company's internal tools? Yes — internal, non-competing production use is permitted under the Additional Use Grant in LICENSE.txt. Re-hosting Engramia as a paid service, or embedding it in a competing commercial offering, requires a commercial license.

Can I fork and modify the code? Yes. Modifications and redistribution under BUSL-1.1 terms are allowed. Any derivative work carries the same license until the Change Date.