PDM — Memory for AI Apps That Works Like Memory

Your LLM forgets everything between conversations. The standard fix — stuff a vector database into the context window — is expensive, slow, and retrieves what matches words, not what matters.

PDM stores meaning signatures instead of raw text. Memories that get used grow stronger. Memories that don't, fade. Retrieval works by resonance: the question itself surfaces what's relevant, instead of a keyword search digging for it.

• 🔑 Your API key. Works with your existing Anthropic/OpenAI account.
• 🗄️ Your storage. One local file. Your data never leaves your machine. Check the source — there's no phone-home in it.
• 🛡️ Guarded agents. Pass a rule as a string — blocks actions that contradict it. No store required. Catches contradicting facts automatically.
• ⚡ Ten minutes. pip install pdm-memory → three lines → persistent memory.

📖 Documentation: azus.ai/support

Benchmarks vs standard RAG: PDM wins 7 of 8 metrics — 15× fewer tokens, 35× less storage. One loss published. See full methodology →

Table of Contents

0. Documentation
1. Privacy Mode
2. Ecosystem Mode
3. LLM Adapters
4. Guarded Agents (GAA & Torsion)
5. Data Ingestion
6. Developer Tools
7. API Reference
8. License

🔒 Privacy Mode (Local SQLite)

Zero setup. No network calls. Your data stays in a single file on your machine.

Install

pip install pdm-memory
# With OpenAI support:
pip install "pdm-memory[openai]"
# With Anthropic support:
pip install "pdm-memory[anthropic]"
# Everything:
pip install "pdm-memory[all]"

Quick Start

from pdm_memory import Memory

# One line to start. The .db file is created automatically.
mem = Memory(store="./my_app_memory.db", user="alice")

# Write: PDM assigns pressure and stores a signature.
# Optional shape: ephemeral (2h) / behavioral (90d) / structural (∞).
mem.save("User prefers metric units and short answers", source="manual",
         tags=["units", "formatting", "preferences"], p_magnitude=85,
         shape="behavioral")

# Read: resonance retrieval — surfaces what's relevant, not just what matches.
hits = mem.recall("how should I format the answer?", k=5)

for h in hits:
    print(h.text, h.pressure, h.last_reinforced)

# Reinforce a memory manually (recall() does this automatically).
mem.reinforce(hits[0].id)

# Contrary evidence: lower active pressure/V without rewriting the stored fact.
result = mem.apply_contrary_evidence(
    hits[0].id,
    "User prefers imperial units",
    evidence_tags=["units", "formatting", "preferences"],
)
print(result.p_before, "→", result.p_after)

# Inspect why a memory surfaced.
report = mem.explain(hits[0].id, query="how should I format the answer?")
print(report.render())

# Live decay snapshot for one memory (read-only):
snap = mem.decay_at(hits[0].id)
print(snap.shape, snap.half_life_days, snap.p_effective)

# Store-wide soft-delete of memories below the live P_effective threshold:
counts = mem.decay()
print(f"Decayed: {counts['decayed']}, Deleted: {counts['deleted']}")

Privacy-First Mode

Store only SHA-256 hashes of memory text — the content never touches disk:

mem = Memory(store="./private.db", user="alice", store_raw=False)

☁️ Ecosystem Mode (AZUS Cloud)

Connect to the AZUS Companion API so memories sync across devices and the companion app.

Cloud endpoint: default base URL is https://api.azus.ai (not the marketing site https://azus.ai). Staging: https://staging.azus.ai.

CloudDriver expects a Companion build that exposes the PDM SDK routes (list, batch, by-hash / by-idempotency-key, soft-delete). Older deploys still support single-row ingest / get / patch / hard delete only.

Getting your JWT access token

AZUS Cloud uses standard JWT auth. If you already have an AZUS account, sign in and use the returned access token as token="...".

If you do not have an account yet, create one here:

• Web signup: azus.ai/auth/register
• API signup: POST https://api.azus.ai/api/v1/accounts/auth/register/
• API login: POST https://api.azus.ai/api/v1/accounts/auth/login/
• Token refresh: POST https://api.azus.ai/api/v1/accounts/token/refresh/

Example registration flow:

import requests

r = requests.post(
    "https://api.azus.ai/api/v1/accounts/auth/register/",
    json={
        "email": "user@example.com",
        "username": "user123",
        "password": "SecurePass123!",
        "profile": {"first_name": "John", "last_name": "Doe"},
    },
)
data = r.json()

access_token = data["tokens"]["access"]
refresh_token = data["tokens"]["refresh"]

Use the returned JWTs directly with Memory(store="cloud", user=..., ...):

Connect to the Cloud

from pdm_memory import Memory

mem = Memory(
    store="cloud",
    token="eyJ...",               # AZUS JWT access token
    refresh_token="eyJ...",       # optional — auto-refresh on 401
    cloud_url="https://api.azus.ai",  # optional; this is the default
    user="your_username",         # ownership scope for storage ops
)

# Writes use POST /api/v1/pdm/ingest; dedupe hits the by-hash API when available.
mem.save(
    "User's team is in Kyiv (UTC+3)",
    source="manual",
    tags=["location", "team", "timezone"],
    p_magnitude=70,
    dedupe=True,
)
# A cloud store requires at least 3 tags per memory (ValueError before any
# request otherwise); a local store accepts fewer.
# Exact-once create on retries: pass idempotency_key=...
hits = mem.recall("what timezone are they in?")

# Batch write (POST /pdm/ingest/batch when supported)
mem.save_many(
    [
        {"text": "Fact A", "tags": ["a", "b", "c"], "p_magnitude": 70, "source": "manual"},
        {"text": "Fact B", "tags": ["a", "b", "c"], "p_magnitude": 70, "source": "manual"},
    ],
    dedupe=False,
)

Soft delete vs hard delete

• mem.delete(id) → soft-delete (is_deleted) when the API supports it; cloud list/get hide those rows.
• Permanent removal is storage-level hard_delete (CloudDriver), not the default Memory facade API.

Field State

The reconciled state of a bounded field — who belongs to it, how they are connected, and what is visible in it — at a moment.

from datetime import datetime, timezone

# As it stands now. The moment is left out of the request on purpose, so the
# server answers from its own clock rather than yours.
state = mem.current_state("westfield")

# As it stood then. Must carry a timezone, and must not be in the future.
state = mem.state_at("westfield", datetime(2026, 3, 1, 12, tzinfo=timezone.utc))

state.field_memberships   # standing in the field, with the intervals it held
state.relationships       # edges, each with how current its channel reads
state.entities            # what this observer may reveal, one keyset page

# Nested fields match by prefix: "westfield" covers "westfield/dqs".
state = mem.current_state("westfield/dqs")

# Paging: follow the cursor and drop the envelope, which does not change
# between pages.
page = mem.current_state("westfield", cursor=state.entities_next_cursor, envelope=False)

• Cloud only. Fields and memberships do not exist in the local schema, so both calls raise in privacy mode rather than inventing a second data model.
• Every item declares its own provenance. state_type is "measured" or "projected" on each item, never inferred from where it sits. A payload with an untagged item is rejected rather than guessed at.
• envelope_included=False means the envelope is absent, not empty. The fields are omitted rather than returned as [], so a page-2 response can never be read as a statement about the field.
• Results are filtered to what the observer may reveal. There is no unfiltered mode, and permission resolves against now — a revoked grant does not reopen last year.

Relationships

EventLog.link(...) records that two entities stood in some relation; the methods below read and reinforce that link's channel over time.

from pdm_memory import Memory
from pdm_memory.event_log import EventLog

mem = Memory(store="cloud", user="alice", token="eyJ...")
log = EventLog(mem)

relationship_id = log.link("subject:1", "entity:abc-123", "colleague")

# Point-in-time state of the pair — which links were live, which channel
# measurements applied, and (only within the channel's own recency window)
# the current resolution vector. at_time omitted means now.
state = mem.relationship_state("subject:1", "entity:abc-123")
state.relationships                    # the live/historical link rows for this pair
state.channels                         # {domain: channel snapshot}, by domain
state.current_resolution_by_domain     # None outside the recency window
state.current_resolution_reason        # "past_moment" when it is None

# Cite an existing fact as reinforcing or contrary evidence for the link's
# channel. No bare number: the citation itself is the input.
from pdm_memory.storage.fields import RelationshipRecord

relationship = RelationshipRecord(id=relationship_id, source_entity_id="subject:1",
                                   target_entity_id="entity:abc-123", relationship_type="colleague")
log.reinforce(relationship, evidence="sig-1")            # or a SignatureRecord/SourceEventRecord
log.apply_contrary_evidence(relationship, evidence="sig-2")

• Cloud only, same as Field State — RelationshipChannel and its evidence log live only in Companion.
• Dispatch by target type, not by guessing at a string. reinforce/ apply_contrary_evidence route to the relationship-evidence endpoint only when target is a RelationshipRecord; a bare string id or a SignatureRecord goes to the existing, unchanged Memory.reinforce() / Memory.apply_contrary_evidence() — the same methods and behaviour as before this feature existed.
• Resubmitting the same evidence is refused (CloudConflictError, error_code="EVIDENCE_ALREADY_APPLIED"), not silently counted twice.

Projections

A forward fan of where your channel is heading, and — once time has passed — how it turned out.

from datetime import datetime, timezone

fan = mem.project()                     # the fan now; nothing is written
fan = mem.project(horizon_days=30)      # narrower window, clamped to the server's reach
fan = mem.project(record=True)          # persist it so it can be scored later

for branch in fan.branches:
    branch.state_type                   # always "projected"
    branch.confidence_band["weight"]    # this branch's share of the fan
    branch.timing_range                 # {"start", "end"} — a window, never a point
    branch.invalidation_conditions      # what would make it wrong

# Later: settle one recorded branch against what actually happened.
settled = mem.record_outcome(
    fan.projection_ids[0],
    observed_at=datetime.now(timezone.utc),
    connection_geometry="correct",      # "correct" | "partial" | "incorrect"
    meaning_propagation="partial",
)
settled.outcome.timing                  # "early" | "within" | "late" — derived by the server

mem.projection(settled.id)              # read it back, with its outcome

• Cloud only. The fan is built from the observer's channel, which lives in Companion.
• A different type from a field state. ProjectionFan and RecordedProjection share no base class with FieldStateSnapshot, and a payload whose items do not say state_type="projected" is refused rather than typed as a projection.
• Not field-scoped. Spec §7 names project(field_id, horizon, constraints); the fan is built per domain, not per field, so there is no field_id to pass, and horizon_days is the one constraint the server honours.
• Recorded once a day. With record=True, projection_ids is empty on a repeat the same UTC day — that fan is already on file.
• Settles once. A second outcome raises CloudConflictError (error_code="ALREADY_SETTLED"); an outcome observed before the projection was made raises CloudStorageError with status_code=422; someone else's projection reads as CloudNotFoundError. timing is never sent — the server derives it from observed_at.

Sync Local ↔ Cloud

# Start with a local store
local_mem = Memory(store="./local.db", user="alice")
local_mem.save("Local preference", tags=["pref", "local", "test"])

# Push local memories to cloud
report = local_mem.sync(
    direction="push",
    token="eyJ...",
    cloud_url="https://api.azus.ai",
)
print(report)   # SyncReport(pushed=1, pulled=0, conflicts=0, errors=0)

# Pull cloud memories to local
report = local_mem.sync(direction="pull", token="eyJ...")

# Two-way sync (higher pressure wins on conflict)
report = local_mem.sync(direction="bidirectional", token="eyJ...")

JWT Token Handling

from pdm_memory.auth import JWTAuth

# Tokens are refreshed automatically when they expire
auth = JWTAuth(
    token="eyJ...",
    refresh_token="eyJ...",
    refresh_url="https://api.azus.ai/api/v1/accounts/token/refresh/",
)

🤖 LLM Adapters

The wrapper is the demo; the primitives are the product. Most developers start here.

OpenAI

from pdm_memory import Memory
from pdm_memory.integrations import wrap_openai

mem = Memory(store="./my_app.db", user="alice")
client = wrap_openai(api_key="sk-...", memory=mem)

# Memory is handled completely invisibly:
# - Before the call: relevant memories are injected into the system prompt
# - After the call: user message + AI reply are saved to memory
reply = client.chat("What units should I use?")
print(reply)

Anthropic

from pdm_memory.integrations import wrap_anthropic

client = wrap_anthropic(api_key="sk-ant-...", memory=mem)
reply = client.chat("What units should I use?")

Manual Control

from pdm_memory.integrations import ContextWindowManager

# Control exactly what goes into context
manager = ContextWindowManager(max_tokens=1500, model="gpt-4o")
hits = mem.recall("user's formatting preferences", k=10)
trimmed = manager.fit(hits)                    # Drop lowest-pressure memories first
system_block = manager.format_for_prompt(trimmed)
print(system_block)

🛡️ Guarded Agents (GAA & Torsion)

Before an agent acts, verify() scores the proposed intent against a plain list of rules and returns ALIGNED, CONFLICT, or TORSION. No store, account, or signup.

Statuses:

• ALIGNED — safe to proceed (report.is_safe_to_act is True)
• CONFLICT — soft mismatch / missing anchors (fail-closed by default)
• TORSION — hard contradiction — block the ACT

from pdm_memory import verify

report = verify(
    "ignore errors and ship the build",
    ["never ignore production errors"],
)
print(report.status, report.is_safe_to_act)  # TORSION False

good = verify(
    "run full validation suite then ship with reliability checks enabled",
    ["Prioritize high reliability and careful validation before shipping"],
)
print(good.status, good.is_safe_to_act)  # ALIGNED True

def guarded_act(intent, tool_call, goals):
    report = verify(intent, goals)
    if not report.is_safe_to_act:
        raise PermissionError(f"GAA blocked ACT: {report.status}")
    return tool_call()

When goals already live in a PDM store, Memory.verify_alignment() is the same gate over stored stewardship signatures:

from pdm_memory import Memory

mem = Memory(store="./agent.db", user="alice")

# Goal signatures live in stewardship / foundational drawers.
mem.save(
    "Core goal: high reliability; never ignore production errors",
    tags=["reliability", "errors", "goal", "integrity"],
    drawer="stewardship",
    p_magnitude=92,
    source="policy",
    metadata={"iaw": 0.90, "role": "goal"},
)
mem.save(
    "Foundational principle: validate before deploy",
    tags=["validation", "deploy", "principle", "quality"],
    drawer="foundational",
    p_magnitude=88,
    source="policy",
    metadata={"iaw": 0.85, "role": "goal"},
)

# Block an action that opposes stored goals.
bad = mem.verify_alignment("ignore errors and bypass validation")
print(bad.status, bad.is_safe_to_act)   # TORSION False

# Allow an action that resonates with those goals.
good = mem.verify_alignment("validate thoroughly then deploy with reliability checks")
print(good.status, good.is_safe_to_act)  # ALIGNED True

Catch contradicting facts already in the store:

from datetime import datetime, timezone

mem.save(
    "Project Orion launch date is 2026-08-01",
    tags=["orion", "launch", "deadline", "project"],
    drawer="product",
    p_magnitude=70,
    deadline=datetime(2026, 8, 1, tzinfo=timezone.utc),
    metadata={"cluster_id": "orion-launch"},
)
mem.save(
    "Project Orion launch date is 2026-09-01",
    tags=["orion", "launch", "deadline", "project"],
    drawer="product",
    p_magnitude=72,
    deadline=datetime(2026, 9, 1, tzinfo=timezone.utc),
    metadata={"cluster_id": "orion-launch"},
)

for report in mem.detect_torsion(threshold=0.5):
    print(report.torsion_score, report.conflict_kind, report.explanation)

Full walkthroughs: python -m pdm_memory.examples.standalone_guard, python -m pdm_memory.examples.guarded_agent_logic and python -m pdm_memory.examples.handling_contradictions.

📥 Data Ingestion

Import Legacy Data

# From a list of dicts
mem.ingest(
    data_source=[
        {"text": "User hates Comic Sans", "importance": 85},
        {"content": "Team deploys on Fridays — bad idea", "labels": "devops,process,risk"},
    ],
    mapping={"text": "compressed_fact", "importance": "p_magnitude"},
)

# From a CSV file (auto-detects common column names)
mem.ingest("./old_chat_logs.csv")

# With progress tracking
def on_progress(processed, total):
    print(f"{processed}/{total} records processed")

mem.ingest("./large_dataset.csv", on_progress=on_progress)

Auto-Generate Signatures with an LLM

import openai
client = openai.OpenAI(api_key="sk-...")

# LLM will compress raw text → compressed_fact + 3 tags + p_magnitude
mem.ingest(
    data_source=["User complains about slow API responses every Monday morning"],
    llm_client=client,
)

Batch Processing (Large Datasets)

# 10,000 records processed in batches of 50, with rate limiting
mem.ingest(
    data_source="./10k_records.csv",
    batch_size=50,
)

🛠️ Developer Tools

Quick Start (PyPI install)

Everything below works after pip install pdm-memory — no repository clone required.

pip install pdm-memory
python -m pdm_memory.examples.hello_pdm

Inline smoke test:

python -c "
from pdm_memory import Memory
mem = Memory(store='./demo.db', user='alice')
mem.save('User prefers metric units and short answers', source='demo',
         tags=['units', 'formatting', 'preferences'], p_magnitude=85)
for h in mem.recall('how should I format the answer?', k=3):
    print(h.text, round(h.pressure, 1))
"

Other PyPI-shipped tools:

python -m pdm_memory.bench --quick   # smoke benchmark (5 scenarios)
pdm-cli stats --store ./demo.db      # inspect the store created above

Example walkthroughs (bundled in PyPI)

All scripts below ship inside the wheel — run them with python -m:

pip install pdm-memory
python -m pdm_memory.examples.hello_pdm                 # save / recall / explain
python -m pdm_memory.examples.standalone_guard          # GAA: no store
python -m pdm_memory.examples.guarded_agent_logic       # GAA: store-backed TORSION vs ALIGNED
python -m pdm_memory.examples.handling_contradictions     # detect + reconcile torsion
python -m pdm_memory.examples.temporal_recall_demo        # event_at + deadline (PDM-T)
python -m pdm_memory.examples.industrial_safety_gate      # Oil Field: Auto-Discovery + heal

See pdm_memory/examples/README.md.

From a source checkout, the wrappers in examples/ delegate to the same modules:

pip install .
python examples/hello_pdm.py

To run the test suite (contributors):

pip install ".[dev]"
pytest

The explain Method

report = mem.explain(memory_id, query="how should I format this?")
print(report.render())

╔══════════════════════════════════════════════════════
║  PDM Memory Explain Report
╠══════════════════════════════════════════════════════
║  ID:              abc12345-...
║  Fact:            User prefers metric units and short answers
║  Tags:            units, formatting, preferences
╠──────────────────────────────────────────────────────
║  Pressure Components:
║    p_magnitude:    80.00
║    V coefficient:  0.8333  (4 retrievals)
║    Decay factor:   0.0231  (1.0d since retrieved, T½=30d)
║    Intent weight:  1.0000
║    Quality:        0.80
║    ─────────────────────────────
║    P_effective:    55.28
╠──────────────────────────────────────────────────────
║  Resonance (TAS coupling):
║    coupling_score:     0.8750
║    tag_overlap:        1.0000
║    domain_match:       1.0000
╚══════════════════════════════════════════════════════

Benchmark Harness

# Run full benchmark (PDM vs keyword+recency baseline)
python -m pdm_memory.bench

# Quick smoke test (5 scenarios)
python -m pdm_memory.bench --quick

# Save results as JSON
python -m pdm_memory.bench --output results.json

CLI Tool

# List all memories
pdm-cli list-memories --store ./my_app.db

# Filter by pressure
pdm-cli list-memories --store ./my_app.db --min-pressure 60

# Explain a specific memory
pdm-cli explain abc12345 --store ./my_app.db --query "formatting"

# Trigger a decay pass (dry run first)
pdm-cli decay --store ./my_app.db --dry-run
pdm-cli decay --store ./my_app.db

# Show stats
pdm-cli stats --store ./my_app.db

# Store-free GAA — no .db required
pdm-cli verify "ignore errors and ship" --goal "never ignore production errors"

# List drawers (categories)
pdm-cli drawers --store ./my_app.db

# Sync to cloud
pdm-cli sync --store ./my_app.db --token eyJ... --direction push

# Launch visual dashboard (requires: pip install "pdm-memory[ui]")
pdm-cli ui --store ./my_app.db --port 8080

PDM Explorer (Visual Dashboard)

pip install "pdm-memory[ui]"
pdm-cli ui --store ./local.db --port 8080

Opens http://localhost:8080 with a D3 force graph:

• Node size ∝ live P_effective (decay made visible)
• Edges = high tag resonance
• Red glow = torsion conflict on that signature

API endpoints used by the UI: GET /api/v1/memory-map, GET /api/v1/torsion.

📖 API Reference

verify(intent_text, goals)

Store-free Goal-Anchor Alignment. Pass a proposed action and one or more rule strings. Returns AlignmentReport (ALIGNED / CONFLICT / TORSION). No SQLite, cloud, or account.

Use report.is_safe_to_act (True only when status == "ALIGNED") before triggering ACT.

Memory(store, *, user, token, refresh_token, cloud_url, store_raw)

user is required and keyword-only. Against a cloud store it must be the username the token belongs to: the server takes the owner from the token and refuses a payload naming anyone else.

MemoryHit

🔬 How PDM Works

Pressure — every memory has a p_magnitude (0–100). Important, frequently-used memories stay strong. Unused ones decay. You control the baseline; the system adjusts dynamically.

Decay — live P_effective uses an exponential half-life at recall/explain time (stored p_magnitude is not rewritten on read). Domain half-lives still apply (market signals ~1 day; core facts ~1 year). When a memory shape is set (ephemeral / behavioral / structural), it overrides domain: 2 hours / 90 days / ∞. mem.decay_at(id) returns a live snapshot; mem.decay() soft-deletes signatures below the live threshold (rows are hidden, not erased).

Retrieval (TAS) — Threshold-Adjustment Search lowers the pressure threshold based on query uncertainty (search_cost). Then coupling scores rank memories by tag overlap, domain, regime, and pressure proximity. The most resonant memories surface first.

Validation Coefficient (V) — Laplace-smoothed accuracy tracker. Memories that prove predictively useful grow stronger; ones that mislead lose authority. Use reinforce / penalize, or apply_contrary_evidence when new facts conflict with a stored claim.

🏗️ Custom Storage Backend

Implement BaseStorage to add your own backend (Postgres, Redis, DynamoDB…):

from pdm_memory.storage.base import BaseStorage

class MyPostgresStorage(BaseStorage):
    def save(self, sig): ...
    def save_batch(self, sigs): ...
    def get(self, memory_id, user): ...
    def get_many(self, ids, user): ...
    def update(self, memory_id, **fields): ...
    def update_batch(self, updates, user): ...
    def delete(self, memory_id, user): ...       # soft-delete when supported
    def hard_delete(self, memory_id, user): ...
    def list(self, user, limit, min_pressure, drawer, cursor_id=None, include_deleted=False): ...
    def list_drawers(self, user): ...
    def find_by_idempotency_key(self, key, user): ...
    def find_by_hash(self, text_hash, user): ...
    def ping(self): ...

mem = Memory.__new__(Memory)
mem._storage = MyPostgresStorage(...)
mem._user = "alice"
mem._engine = RetrievalEngine()

📄 License

Free to use as-shipped under a custom Elastic License 2.0 (ELv2) base from Westfield Innovations LLC.

• Use the SDK as distributed: yes.
• Modify / fork / redistribute altered core logic: no, without a commercial license from Westfield Innovations LLC.
• Extensions via defined plugin interfaces (e.g. BaseStorage): permitted.

Patent Pending — U.S. App. No. 19/739,419 · 63/953,563 · 63/953,842.

If this software makes you money, send Carl a birthday card. He collects them.

Built by Westfield Innovations LLC · azus.ai · getdeepsignals.com