json-memory 0.2.1

Hierarchical associative memory for AI agents — compress, structure, and navigate agent memory like a human brain

json-memory

Structured memory for AI agents — organize, access, and navigate agent memory like a human brain.

pip install json-memory

Performance (991-char memory, commodity hardware):

• Parse: 0.014ms (72K parses/sec)
• Access: 0.60μs per dotted-path lookup
• Zero dependencies, pure Python

The Problem

AI agents have limited memory windows. Storing facts as verbose prose wastes tokens and makes retrieval slow:

"User: Alice (@alice on Telegram). Prefers to be called Alice.
 Uses they/them pronouns. Timezone is UTC. Platform is Telegram. Prefers
 technical precision, especially in coding contexts. Wants a direct, warm..."

~300 chars for basic user info. No structured access — you scan the entire text every time.

The Solution

Store memory as nested JSON with short keys — like synapses in a brain:

{"u":{"n":"Alice","c":"@alice","p":"Alice","g":"they/them","tz":"UTC","plat":"Telegram"}}

~95 chars for the same data. But the real win isn't size — it's O(1) access via dotted paths: memory.u.n → "Alice". No scanning. No parsing prose. Just keys.

Why Structured Memory?

	Prose	JSON Memory
Access pattern	Scan entire text	memory.u.n → instant
Nested hierarchy	❌ Flat	✅ Unlimited depth
Schema validation	❌ No	✅ Yes
Merge/upsert	❌ Rewrite everything	✅ Per-key updates
Human readable	✅ Yes	❌ Compact (but AI reads it)

The trade-off: JSON is less human-readable but machine-optimized. For LLM agents with token budgets, that's the right call.

Key Features

• 🧠 Hierarchical nesting — organize memory like a semantic tree
• 🗜️ Key abbreviation — ~25% size reduction on JSON keys
• 📦 JSON minification — ~30% savings removing whitespace
• ⚡ Sub-millisecond parsing — 0.05ms for 2KB of memory
• 🔗 Synapse-like linking — concepts connect to related concepts with weighted traversal
• 🐕 WeightGate middleware — passive learning from conversation flow
• 📐 Schema validation — define your memory structure once
• 🐍 Zero dependencies — pure Python, stdlib only

Installation

git clone https://github.com/dioncx/json-memory.git
cd json-memory
pip install -e .

Quick Start

from json_memory import Memory

# Create a memory instance
mem = Memory(max_chars=2000)

# Set nested values
mem.set("u.name", "Alice")
mem.set("u.tz", "UTC")
mem.set("bot.binance.restart", "kill && nohup ./bot > log 2>&1")
mem.set("bot.binance.watchlist", ["BNB", "KITE", "AGLD"])

# Get by dotted path
print(mem.get("u.name"))           # "Alice"
print(mem.get("bot.binance.restart"))  # "kill && nohup ./bot > log 2>&1"

# Export/import
json_str = mem.export()            # minified JSON string
mem2 = Memory.from_json(json_str)  # reconstruct

# Stats
print(mem.stats())
# {"entries": 4, "chars_used": 146, "chars_max": 2000, "chars_free": 1854, "utilization": "7.3%"}

Agent Memory — Drop-In for Any LLM

A ready-to-use memory layer for AI agents. Copy examples/agent_memory.py into your project, or use the pattern below.

10-Line Agent Memory

from json_memory import Memory

mem = Memory(auto_flush_path="agent_memory.json", max_chars=5000)

# Store facts
mem.set("user.name", "Alice")
mem.set("user.tz", "UTC+7")
mem.set("session.task", "reviewing PR #42")

# Recall
print(mem.get("user.name"))  # → "Alice"

# Inject into LLM prompt
context = mem.export()  # → compact JSON string, ~100 chars

Full Agent Memory Class

The examples/agent_memory.py file provides a complete AgentMemory class with:

• Persistent storage — survives restarts via auto-flush
• Dotted-path access — remember("user.name", "Alice"), recall("user.name")
• Search — search("user.*") returns all user-related facts
• Associative recall — link concepts like a brain: link("debugging", ["check_logs", "reproduce"])
• Snapshots — save/rollback state before risky operations
• TTL — ephemeral data that auto-expires: remember("token", "xyz", ttl=300)
• Prompt injection — context() returns compact JSON for system prompts

from agent_memory import AgentMemory

memory = AgentMemory("my_agent.json", max_chars=5000)

# Store
memory.remember("user.name", "Alice")
memory.remember("user.preferences.style", "direct")
memory.remember("session.temp_token", "xyz_123", ttl=300)  # expires in 5 min

# Associative memory
memory.link("debugging", ["check_logs", "reproduce", "git_bisect"])
memory.associate("debugging")  # → ["check_logs", "reproduce", "git_bisect"]

# Search
memory.search("user.*")  # → {"user.name": "Alice", "user.preferences.style": "direct"}

# Inject into system prompt
system_prompt = f"You are an assistant.\n\n## Memory\n{memory.context()}"

Works With Any Agent Framework

Framework	Integration
Claude (API)	Inject memory.context() into system prompt
OpenAI GPT	Inject into system message, or use Schema.to_openai_tools()
OpenClaw	Load as a tool — see examples/agent_memory.py
LangChain	Custom memory class wrapping AgentMemory
Custom loop	Call remember()/recall() in your agent loop

No lock-in. No dependencies. Just Python.

SmartMemory — Intelligent Retrieval (v0.2.0)

The problem with most agent memory: you inject everything into every prompt. 5000 chars of memory, but the user only asked about their timezone. Wasted tokens.

SmartMemory fixes this with weighted retrieval — only the relevant facts make it into the prompt.

from json_memory import SmartMemory

mem = SmartMemory("agent.json", max_chars=5000)

# Store facts
mem.remember("user.name", "Alice")
mem.remember("user.timezone", "GMT+7")
mem.remember("bot.restart_cmd", "kill && nohup ./bot > log")
mem.remember("server.ip", "10.0.0.1")
mem.remember("bot.symbol", "BNBUSDT")

# Smart recall — only relevant facts per query
mem.recall_relevant("What's my timezone?")
# → {"user.timezone": "GMT+7"}  (not the entire memory)

mem.recall_relevant("How do I restart?")
# → {"bot.restart_cmd": "kill && nohup ./bot > log"}

# Lean prompt injection — only what matters
context = mem.prompt_context("What's my timezone?")
# → "## Memory\n- user.timezone: GMT+7"  (32 chars, not 5000)

How Retrieval Works

Each fact is scored on three dimensions:

Factor	Weight (with query)	What it measures
Keyword relevance	85%	Token overlap between query and fact
Recency	10%	Exponential decay (half-life: 1 hour)
Frequency	5%	How often the fact was accessed

When no query is provided, scoring uses recency (60%) + frequency (40%).

Smart filtering: If the query has strong keyword matches, noise is automatically suppressed. "What's my timezone?" returns 1 result, not 8.

Auto-Extraction

Passively detects facts from conversation — no explicit remember() needed:

extracted = mem.process_conversation("My name is Bob and I live in Tokyo")
# → [{"path": "user.name", "value": "Bob", "confidence": 0.8},
#    {"path": "user.location", "value": "Tokyo", "confidence": 0.85}]

extracted = mem.process_conversation("Remember that the API key is sk-abc123")
# → [{"path": "user.notes", "value": "the API key is sk-abc123", "confidence": 0.95}]

Patterns detected: names, locations, timezones, preferences, platforms, project names, explicit "remember" requests.

Synonym Expansion

Queries are expanded with synonyms for better recall:

• "Who am I?" → searches for name, user, identity
• "What trades the bot?" → searches for exchange, strategy, bot
• "What's the time?" → searches for timezone, gmt, utc

Semantic Search (Optional)

Upgrade to embedding-based retrieval:

pip install json-memory[semantic]

from json_memory import SmartMemory
from json_memory.semantic import enhance_smart_memory

mem = SmartMemory("agent.json")
enhance_smart_memory(mem)  # adds FAISS + sentence-transformers

# Now semantic search works — meaning, not just keywords
mem.recall_relevant("When do I wake up?")
# → {"user.timezone": "GMT+7"}  (understands "wake up" relates to timezone)

Without the optional deps, falls back gracefully to keyword scoring. Zero lock-in.

Score Debugging

Understand why a fact was or wasn't returned:

mem.explain_score("user.timezone", "What's my timezone?")
# → {"path": "user.timezone", "recency": 0.95, "frequency": 0.8,
#    "keyword_relevance": 0.43, "final_score": 0.45, "access_count": 12}

API Reference

Memory

Method	Description
mem.set(path, value, ttl)	Set value with optional Time-To-Live (sec)
mem.get(path, default)	Get value (auto-purges if expired)
mem.purge_expired()	Manually clear all stale data
mem.batch_get(paths)	Get multiple values in one call
mem.watch(path, cb)	React to state changes
mem.increment(path, delta)	Atomically increment a numeric value
mem.touch(path, ts)	Set current timestamp at path
mem.delete(path, prune)	Delete path, optionally prune empty parents
mem.clear(path)	Clear subtree or the entire memory
mem.has(path)	Check if path exists
mem.paths(prefix)	List all leaf paths
mem.merge(data, prefix)	Bulk merge a dict into memory
mem.stats()	Get size and utilization metrics

Synapse

Method	Description
brain.link(c, assocs)	Create bidirectional weighted links
brain.activate(c, depth)	Recall associated concepts
brain.strengthen(c, a)	Increase association weight
brain.weaken(c, a)	Decrease association weight (decay)
brain.find_path(s, e)	Find shortest path between concepts
brain.hubs()	Find the most connected concepts
brain.merge(other)	Combine two independent graphs
brain.rename_concept(old, new)	Rename a concept globally.
subgraph(concepts)	Extract a new Synapse with only related nodes.
find_strongest_path(a, b)	Find the highest-weight path using Dijkstra.

Schema Validation

Method	Description
validate(data, strict)	Validate a dict against the schema.
validate_memory(mem)	Validate a Memory instance.
defaults()	Generate a skeleton dict with default types/lists.

WeightGate (Context Aware)

Method	Description
process_input(text)	Strengthen concepts mentioned in prose.
process_output(text)	Strengthen concepts produced by agent.
ngram_size (Config)	Detect multi-word concepts (e.g. "machine learning").

Synapse Mode (Associative Memory)

Like how thinking of "coffee" activates "morning", "energy", "routine":

from json_memory import Synapse

brain = Synapse()

# Define associations
brain.link("trading", ["binance", "strategy", "risk", "signals"])
brain.link("binance", ["api", "demo", "watchlist", "orders"])
brain.link("strategy", ["entry", "exit", "stoploss", "take_profit"])

# Traverse like a brain
results = brain.activate("trading")
# → ["binance", "strategy", "risk", "signals"]

results = brain.activate("trading", depth=2)
# → ["binance", "api", "demo", "watchlist", "orders", "strategy", "entry", "exit", ...]

# Find connections
brain.connections("binance")
# → {"parent": "trading", "children": ["api", "demo", "watchlist", "orders"]}

Personalized Weights

Everyone's brain works differently. Set weights to customize recall order:

# Person A: loves cappuccino
person_a = Synapse()
person_a.link("coffee", ["cappuccino", "americano", "espresso"],
              weights={"cappuccino": 0.95, "americano": 0.2, "espresso": 0.5})

# Person B: loves americano
person_b = Synapse()
person_b.link("coffee", ["cappuccino", "americano", "espresso"],
              weights={"cappuccino": 0.2, "americano": 0.9, "espresso": 0.4})

person_a.activate("coffee")  # → ["cappuccino", "espresso", "americano"]
person_b.activate("coffee")  # → ["americano", "espresso", "cappuccino"]

Learning & Decay

Mimic how human memory strengthens with use and decays without:

brain = Synapse()
brain.link("coffee", ["cappuccino", "americano"],
           weights={"cappuccino": 0.5, "americano": 0.5})

# User always picks cappuccino → connection strengthens
for _ in range(10):
    brain.strengthen("coffee", "cappuccino", boost=0.05)

# User never picks americano → connection decays
for _ in range(10):
    brain.weaken("coffee", "americano", decay=0.03)

brain.top_associations("coffee")
# → [("cappuccino", 1.0), ("americano", 0.2)]

brain.get_frequency("coffee", "cappuccino")  # → 10 (activation count)

WeightGate — Passive Learning Middleware

Update weights automatically as messages flow through. No tool calls needed.

from json_memory import WeightGate

# Create a gate (disabled by default — opt-in)
gate = WeightGate("synapse.json", enabled=True)

# Set up your concepts
gate.add_concept("coffee", {"cappuccino": 0.9, "americano": 0.3})
gate.add_concept("debug", {"check_logs": 0.9, "ask_user": 0.2})

# Process messages — weights update automatically
gate.process_input("How do I restart the bot?")
# → bot.restart strengthened, unused associations decay

gate.process_output("Run: kill && nohup ./bot > log")
# → Agent's response also updates weights

# After 20 interactions:
gate.top_associations("debug")
# → [("check_logs", 0.95), ("ask_user", 0.18)]  ← learned your pattern

Enable/Disable

# Disabled by default (opt-in)
gate = WeightGate("synapse.json")          # OFF
gate.enable()                               # ON
gate.disable()                              # OFF
gate.toggle()                               # Toggle

# Context manager (auto-enable, auto-save)
with WeightGate("synapse.json") as gate:
    gate.process_conversation(user_msg, agent_response)
# Gate disabled and saved on exit

How It Works

User msg ──→ process_input() ──→ detect concepts ──→ weights ↑/↓
                                               ↓
                                    Agent processes
                                               ↓
Agent msg ──→ process_output() ──→ detect usage ──→ weights ↑
                                               ↓
                                    Response to user

• Mentioned concepts → associations strengthen (+0.05)
• Unused associations → decay (-0.01)
• Agent's response → further strengthens used concepts (+0.025)
• Disabled gate → returns empty dict, no side effects

Compression Reality

The compress() module abbreviates JSON keys (e.g., email → em). Here's what it actually saves:

Technique	Savings	What it does
Key abbreviation	~25%	email → em, configuration → cfg
JSON minification	~30%	Removes whitespace from pretty-printed JSON
Combined	~45-50%	Abbreviation + minification applied together

What it does NOT do: compress values, deduplicate data, or apply general-purpose compression (gzip, zstd, etc.).

from json_memory import compress, minify, savings_report

data = {"user": {"email": "[e-mail]", "timezone": "UTC+1"}}
compressed = compress(data)  # {"u": {"em": "[e-mail]", "tz": "UTC+1"}}

# Measure real savings (JSON vs JSON, not prose vs JSON)
report = savings_report(
    json.dumps(data),
    json.dumps(compressed)
)
# {"savings_pct": 8.3, "ratio": 0.917}  ← honest numbers

Parse speed: 0.05ms for 2KB (tested on commodity hardware)

Comparison

Feature	Prose Memory	JSON Memory
Human readable	✅ Yes	❌ Compact (but AI reads it)
Structured access	❌ Scan entire text	✅ Dotted path lookup
Nested hierarchy	❌ Flat	✅ Unlimited depth
Merge/upsert	❌ Rewrite everything	✅ Per-key updates
Parse speed	N/A	✅ 0.05ms
Schema validation	❌ No	✅ Yes

Why Not Just Use [MemGPT/Letta]?

Those are full agent memory frameworks. This is a building block — a lightweight, zero-dependency library for structuring agent memory as JSON. Use it inside your agent, your RAG pipeline, your CLI tool, or your trading bot.

Use Cases

• 🤖 AI Agent memory — compress context windows for LLMs
• 📊 Trading bot state — structured config and position tracking
• 🔧 CLI tools — compact persistent state
• 🎮 Game state — nested world/player/inventory data
• 📱 IoT/Edge — memory-constrained devices

Contributing

PRs welcome! See CONTRIBUTING.md.

License

MIT — see LICENSE.

Ephemeral Memory (TTL)

Prevent long-term "context bloat" by setting automated expiration on keys. Perfect for short-term session data or scratchpads.

# Set a temporary secret that expires in 5 minutes
mem.set("session.token", "xyz_123", ttl=300)

# 6 minutes later...
mem.get("session.token") # → None (auto-purged)

Recursion Note: If a parent key expires, all its children are implicitly expired and cleared upon access.

---

Technical Reference

Persistence

The Memory class distinguishes between data (raw JSON) and state (data + metadata like TTLs).

• mem.to_dict(): Returns raw JSON data (standard use case).
• mem.get_state(): Returns data + TTL metadata for full backup/restore.
• mem.set_state(dict): Restores a full state.
• Memory.from_json(str): Intelligent enough to detect and load both raw JSON or a Metadata state dict.

N-gram Detection

WeightGate supports multi-word concepts via ngram_size:

gate = WeightGate(synapse=s, enabled=True, ngram_size=2)
# "machine learning" will be detected as a single concept

Advanced Search & Transactions

Wildcard Search

Use mem.find(pattern) to query paths using glob-like wildcards.

# Find all character healths across different teams
healths = mem.find("teams.*.members.*.health")
# healths = {"teams.red.members.alice.health": 100, ...}

# Find all 'status' keys at any depth
all_statuses = mem.find("**.status")

State Snapshots (Transactions)

Safety first. Take snapshots before risky operations and rollback if needed.

mem.snapshot("before_task")

# Attempt complex logic
mem.set("agent.working", True)
# ... something goes wrong ...

mem.rollback("before_task") # Entire state (including TTLs) is restored

Autonomous Resilience (Ph. 6)

Automatic Memory Fading (LRU)

Prevent crashes when agents exceed their character budget. When is hit, json-memory automatically "forgets" the oldest, least-accessed data.

# Initialize with LRU eviction policy
mem = Memory(max_chars=2000, eviction_policy="lru")

# If you set a large object that pushes memory over 2000 chars,
# the least-recently used keys are automatically deleted to make room.

Auto-Flush Persistence

Never lose a thought. Automatically sync state to a physical JSON file on every mutation.

# Initialize with a flush path
mem = Memory(auto_flush_path="agent_brain.json")

# Any set(), delete(), or merge() will instantly sync to the file
mem.set("tasks.current", "Analyzing logs...")
# 'agent_brain.json' is updated immediately.

Enterprise Features (Ph. 7)

Thread-Safety

Running a multi-threaded swarm? json-memory uses threading.RLock to ensure that concurrent reads/writes never corrupt your state.

# Shared memory across 50 threads is 100% safe
mem = Memory()

Mutation History (Audit Trail)

Track exactly how your agent's memory changed over time. Perfect for debugging hallucinations or tracing decision-making logic.

mem = Memory(track_history=True)
mem.set("plan", "Step 1: Get coffee")

# View the full log of changes
for event in mem.history():
    print(f"[{event['time']}] {event['action']} {event['path']} -> {event['value']}")

OpenAI Tool Integration

Instantly turn your schema into a tool definition for gpt-4o, gpt-4-turbo, or claude-3.5-sonnet.

schema = Schema({
    "!bio": "str",
    "preferences": ["str"]
})

# Get the JSON array for OpenAI API 'tools' parameter
tools = schema.to_openai_tools("update_memory", "Update the agent's internal profile")

Framework Maturity (Ph. 8)

Persistence Adapters

Move beyond raw JSON files with professional storage adapters.

from json_memory.adapters import SQLiteAdapter

# Use atomic, corruption-proof SQLite persistence
adapter = SQLiteAdapter("agent_brain.db")
mem = Memory(storage_adapter=adapter)

Data Redaction (PII Security)

Ensure sensitive data like API keys and passwords never leak into audit logs or UI exports.

# Keys to protect
mem = Memory(redact_keys=["api_key", "password"], track_history=True)

mem.set("config.api_key", "sk-12345")

# Audit logs are automatically masked
print(mem.history()[0]["value"]) # "***REDACTED***"

# Safe export for dashboards
clean_dict = mem.to_dict(redact=True)