Documentation Tree

Overview — Intelligent Recall & Diversity Control

Ensuring "Parliamentary Breadth" in your AI's focus.

Introduction

One of the biggest challenges in AI memory is Context Drowning. This happens when a high volume of similar, high-pressure memories (like daily chat logs, repetitive status updates, or automated sensor data) completely fills the context window. This "noise" can bury fewer but more critical facts from other domains (like patents, core goals, or safety rules).

PDM includes the Diversity Bias mechanism to ensure your AI always maintains a broad and balanced perspective, even under high interaction pressure.

The Concept of Semantic Quotas

Instead of a simple "top-k" retrieval that only cares about the highest raw score, PDM can now enforce Parliamentary Breadth. This means the engine actively balances the results across different drawers (domains) to prevent a single topic from monopolizing the response.

Why this is critical

Imagine your PDM contains:

100 memories about "Daily coffee orders" (High Magnitude/Recency).
1 memory about "The secret encryption key for Project Orion" (High Magnitude).

Without Diversity Bias, a query like "What is important right now?" might return 10 coffee notes and zero project notes. With Diversity Bias, the system ensures that "coffee" cannot fill more than a specific percentage of the results, allowing the "Orion key" to surface.

Implementing Diversity Control

You can control this behavior via the diversity_bias parameter in the recall() method.

How it works:

None: Pure score-based ranking. The strongest signals win, regardless of their domain.
0.4 (Default): Caps any single drawer at approximately 40% of the total results (k), provided there are candidates from other drawers available in the candidate pool.
Overflow Logic: If there aren't enough candidates from other drawers to fill the remaining 60%, the system will backfill with the best available hits to ensure you still get the requested k number of results.

Example Code:

python

1from pdm_memory import Memory
2
3with Memory(store="./local.db") as mem:
4    # We want 10 results, but we want them balanced across categories
5    hits = mem.recall(
6        "Give me a project overview",
7        k=10,
8        diversity_bias=0.4  # Intelligent balancing active
9    )
10    
11    print(f"Surfaced {len(hits)} memories across diverse domains:")
12    for h in hits:
13        print(f"[{h.drawer}] {h.text[:50]}...")

Anti-Dilution in GAA

This version also upgrades the Goal-Anchor Alignment (GAA) logic used in verify_alignment(). We've implemented Anti-Dilution Weighting.

When checking if an action is safe, the system ensures that a strong contradiction with one major "Stewardship Goal" isn't "watered down" by many other irrelevant or neutral goals. This ensures that the TORSION signal remains loud and clear when it matters most, maintaining the integrity of the system's safety gates.

Best Practices

1. Use Drawers Strategically: Organize your data into logical drawers (e.g., engineering, finance, legal). Diversity bias relies on these boundaries to determine what counts as "different" information.

2. Contextual Tuning:

For a specialized research bot (one domain only), keep diversity_bias=None.
For a general-purpose OS assistant (like Azus), use 0.4 or 0.5 to maintain broad situational awareness.

3. Monitor with Explorer: Use the PDM Explorer UI to see which drawers are currently dominating your memory field and adjust your bias accordingly.

Why use Middleware?

Standard vector databases are "black boxes"—you put data in, you get data out. PDM's Middleware transforms the memory into an active participant in your application's logic.

Security: Mask PII (Personal Identifiable Information) or sensitive credentials before they ever hit the persistent store.
Validation: Prevent low-fidelity facts from being saved based on custom business rules.
Observability: Stream recall results to external dashboards or logging systems (OpenTelemetry, Datadog, etc.).
Auto-Tagging: Automatically enrich signatures with context-aware tags based on the factual content.

The Three Pillars of Injection

Logic injection happens at three critical "seams" in the PDM lifecycle:

1. The Entry Gate (`pre_save`)

Intercept data before it is written. This is your primary defense line. You can modify the SignatureRecord or veto the save operation entirely.

2. The Execution Signal (`post_save`)

Trigger actions after data is successfully locked. Use this for non-blocking side effects like sending notifications or triggering external syncs.

3. The Retrieval Context (`post_recall`)

Analyze the outcome of a search. Access the original query and the final ranked list of hits to measure system performance or log potential contradictions.

Quick Start: Adding a Hook

You register hooks directly on your Memory instance. They are executed synchronously to ensure that your logic is respected before the next step in the pipeline.

python

1from pdm_memory import Memory
2
3def my_logic_injector(sig):
4    # Add a custom tag to every saved memory
5    sig.intent_tags.append("processed_by_middleware")
6    return sig
7
8with Memory(store="./local.db") as mem:
9    mem.add_hook("pre_save", my_logic_injector)
10    
11    # This fact will now carry the "processed_by_middleware" tag in the DB
12    mem.save("The meeting is at 5 PM", tags=["schedule"])

1. Core Concepts (The Westfield Lore)

PDM (Persistent-Driven Memory)

PDM is a physics-inspired memory system. Every fact you save is a Signature carrying pressure (importance), persistence (how long it stays relevant before it starts fading), tags (intent), and a domain (which decay clock it runs on). Retrieval isn't "nearest embedding in vector space." It's Threshold-Adjustment Search (TAS): lower a pressure gate based on how hard you want to search (search_cost), then rank whatever survives by coupling — how well its tags, domain, regime, and pressure resonate with the query.

PDM vs. vector databases:

Vector DB	PDM
Similarity = cosine/dot on embeddings	Similarity = structured coupling + live `P_effective`
Static chunks	Pressure rises on reinforcement, fades on half-life decay
No built-in contradiction model	Torsion detects logical conflict
No action gate	GAA (`verify_alignment`) blocks unsafe agent actions
Opaque ranking	`explain()` decomposes every pressure component

PDM can coexist with embeddings (a Companion CDG layer, if you have one) — but the SDK core itself is tags, TAS, and pressure math, not cosine similarity.

`p_magnitude` vs. `P_effective`

This is the distinction that trips up most new integrations, so it's worth being precise about.

Field	Role
`p_magnitude`	Stored raw importance (0–100). Written on `save()`, updated on `reinforce()` / `update()`. Never changes just because you read it.
`P_effective`	Live retrieval pressure, computed fresh at query time. It is never persisted as the source of truth.

Canonical formula:

typescript

1P_effective = p_magnitude × V × (1 - decay_factor) × intent_weight × quality × comparator

V — Validation coefficient (Laplace-smoothed accuracy history).
decay_factor — Exponential fade since the memory was last touched, keyed to its domain's half-life.
intent_weight — A boost (0.8–1.0) when the query's tags overlap with the memory's.
quality — Fixed at 0.80 in the SDK's recall paths.

Why it matters: a memory can show p_magnitude = 85 in the database and still rank low in a search, because it decayed or failed validation. Conversely, recall() reads live P_effective and never rewrites p_magnitude in the process — there's no double-decay bug where reading a memory ages it twice.

Torsion / Reverse Resonance

Two memories resonate when they share a topic (high tag/token overlap) but contradict each other in fact, polarity, deadline, or pressure vector. That combination — similarity plus opposition — is Reverse Resonance, and PDM scores it as torsion_score.

torsion_score ∈ [0, 1] ≈ topic_similarity × contradiction_strength.

Conflict kinds (conflict_kind):

Kind	Trigger
`deadline`	Different `t_deadline` values, at least a day apart
`factual`	Same topic, different standalone numbers
`polarity`	Negation/antonym cues (`love` vs. `don't love`)
`pressure`	Opposing pressure vectors on a shared topic
`semantic`	General textual disagreement

Detection happens inside drawer/domain clusters (or an explicit metadata.cluster_id) — never a blind, global N² comparison across your whole store. If the built-in rules miss something more subtle (like a paraphrase), an optional TorsionJudge hook lets you extend detection with your own logic.

Decay Law

PDM uses exponential half-life decay — not a legacy p × decay_rate^days multiplier.

typescript

1λ = ln(2) / half_life_days
2decay_factor = 1 - exp(-λ × days_since_last_touch)

Grace window: while days_since_created ≤ t_persistence, decay_factor = 0 — the memory is still "young" and hasn't started fading yet.

Domain half-lives (DOMAIN_HALF_LIVES):

Domain	Half-life (days)
`market_signal`	1
`warning`	3
`reminder`	7
`pattern`	14
`insight`	30 (default)
`structural`	90
`core_fact`	365

decay() maintenance: this computes live P_effective for every signature and hard-deletes (purges) any that fall below 30 (DECAY_DELETE_THRESHOLD). It never mutates p_magnitude in place — decay is a read-time and maintenance-time concern, not a storage rewrite.

Resonance / Coupling

Resonance is query–memory alignment, scored by TAS's Phase 2 — Impedance Matching:

typescript

1coupling_score = 0.50×tag_overlap + 0.20×domain_match + 0.15×regime_match + 0.15×pressure_proximity

A memory couples with a query only when coupling_score ≥ coupling_min (default 0.3) and it has already passed the Phase 1 threshold on its live P_effective.

Phase 1 — Threshold Lowering:

typescript

1θ_eff = max(θ_floor, θ_base × (1 - α × search_cost))

Defaults: θ_base = 30, α = 0.7, θ_floor = 5. A higher search_cost lowers the bar and lets more candidates through — useful when you're deliberately digging into low-pressure memory.

Semantic gate: high-pressure memories still have to pass _semantic_query_overlap(), or they get damped. This is what stops a very important but unrelated memory (imagine something at P90) from surfacing just because its pressure happens to be high.

Identity Anchor Weight (IAW)

IAW measures how "foundational" a memory is for Goal-Anchor Alignment (GAA). High-IAW signatures — typically living in stewardship, foundational, or goals drawers, or carrying anchor tags — act as integrity parliament anchors that agent actions get checked against.

Computation (compute_iaw()):

1. Explicit: use metadata["iaw"] or metadata["identity_anchor_weight"] if you've set one (0–1).

2. Proxy: otherwise, 0.50×(P/100) + 0.20×(phase_privilege/2) + 0.30×membership, where membership is boosted by the stewardship drawer, the structural/core_fact domains, and anchor tags like goal, safety, integrity.

Contradicting a high-IAW anchor produces a more heavily weighted torsion score inside verify_alignment().

SIG (Semantic Information Gain)

SIG is Westfield ecosystem terminology, not a standalone function in the SDK — it's the concept that explains why dedupe and idempotency exist at all.

Plain English: SIG is the marginal new meaning a write adds to the store. Saving "User prefers metric units" twice adds roughly zero SIG — the second call should just return the existing ID (dedupe=True) or collide on an idempotency_key. Reinforcing a memory after a useful recall increases its future retrieval weight — that's positive SIG realized at read time, not a duplicate row.

SDK mechanisms tied to SIG:

save(..., dedupe=True) — hash-based duplicate suppression
save(..., idempotency_key=...) — action-flow idempotency
reinforce() / recall(reinforce=True) — strengthen useful memories without new rows

`t_persistence`

t_persistence is a per-signature grace period, in days, before domain half-life decay kicks in. While days_since_created ≤ t_persistence, decay_factor stays at 0, no matter how long it's been since the memory was last retrieved.

It also feeds effective_spike:

typescript

1effective_spike = min(100, p_magnitude × (t_persistence / 30) × phase_privilege)

Longer persistence produces a higher spike — a stronger initial "signal energy" in the pressure model.

2. API Reference — `Memory`

Constructor

python

1def __init__(
2    self,
3    store: str = "./pdm_memory.db",
4    user: str = "default",
5    token: Optional[str] = None,
6    refresh_token: Optional[str] = None,
7    cloud_url: str = "https://api.azus.ai",
8    store_raw: bool = True,
9    engine: Optional[RetrievalEngine] = None,
10    storage: Optional[BaseStorage] = None,
11    torsion_judge: Optional[TorsionJudge] = None,
12) -> None

Parameter	Description
`store`	SQLite path, `sqlite:///…`, PostgreSQL DSN, `"cloud"`, or a custom registered URL
`user`	Tenant scope for all operations
`token` / `refresh_token`	JWT for cloud mode
`cloud_url`	Companion API base URL
`store_raw`	If `False`, persists only the SHA-256 hash of the fact text
`engine`	Inject a custom `RetrievalEngine` (testing/tuning)
`storage`	Pre-built `BaseStorage` (bypasses URL resolution)
`torsion_judge`	Optional callback for conflict pairs the rules miss

Raises: TypeError if storage is not a BaseStorage instance.

`Memory.from_env()`

python

1@classmethod
2def from_env(cls, *, prefix: str = "PDM", **kwargs: Any) -> "Memory"

Reads configuration from environment variables, which is the natural fit for 12-factor deployments.

Env var	Required	Default
`{prefix}_STORE`	Yes	—
`{prefix}_USER`	No	`"default"`
`{prefix}_TOKEN`	Cloud only	—
`{prefix}_REFRESH_TOKEN`	No	—
`{prefix}_CLOUD_URL`	No	`https://api.azus.ai`

Raises: ValueError if {prefix}_STORE is missing.

`save()`

python

1def save(
2    self,
3    text: str,
4    source: str = "chat",
5    tags: Optional[List[str]] = None,
6    p_magnitude: float = 50.0,
7    t_persistence: float = 30.0,
8    drawer: str = "general",
9    regime: str = "neutral",
10    phase_privilege: float = 1.0,
11    deadline: Optional[datetime] = None,
12    metadata: Optional[Dict[str, Any]] = None,
13    *,
14    dedupe: bool = True,
15    dedupe_reinforce: bool = False,
16    idempotency_key: Optional[str] = None,
17) -> str

Parameter	Description
`text`	Fact content; stripped, truncated to 500 chars
`source`	Provenance label (`chat`, `manual`, `csv`, …)
`tags`	Intent tags; 3+ recommended for retrieval quality
`p_magnitude`	Initial stored pressure, 0–100
`t_persistence`	Grace/decay persistence, in days
`drawer`	Category namespace (`drawer_domain`)
`regime`	Context regime (`neutral`, `trading`, `engineering`, …)
`phase_privilege`	Nesting multiplier (usually 1.0)
`deadline`	PDM-T temporal deadline (`t_deadline`)
`metadata`	Arbitrary JSON-serializable bag (IAW, cluster_id, …)
`dedupe`	If `True`, returns the existing ID when the SHA-256 of the text matches
`dedupe_reinforce`	On a dedupe hit, calls `reinforce()` on the existing memory
`idempotency_key`	If set, a repeated key returns the same ID without a new row

Returns: str — UUID of the new or existing signature.

Raises: ValueError if text is empty.

Side effects: inserts a row, upserts the drawer record, and infers domain from tags.

`save_many()`

python

1def save_many(
2    self,
3    items: List[Dict[str, Any]],
4    *,
5    dedupe: bool = True,
6    dedupe_reinforce: bool = False,
7) -> Dict[str, int]

Each dict accepts: text / compressed_fact, tags / intent_tags, p_magnitude, t_persistence, drawer / drawer_domain, source, regime / question_regime, metadata, deadline.

Returns: {"saved": int, "skipped": int, "errors": int}

Behavior: wraps the loop in storage.transaction() when the driver supports it. Per-item failures increment errors without aborting the whole batch. Internal save() calls use dedupe=False after a batch-level dedupe check has already run.

Raises: nothing directly — failures are captured in the errors count instead.

`recall()`

python

1def recall(
2    self,
3    query: str,
4    k: int = 5,
5    min_pressure: float = 0.0,
6    search_cost: float = 0.5,
7    drawer: Optional[str] = None,
8    reinforce: bool = True,
9    *,
10    candidate_limit: int = 10_000,
11    page_size: int = 500,
12    on_recall: Optional[RecallHook] = None,
13) -> List[MemoryHit]

Parameter	Description
`query`	Natural-language recall context
`k`	Max hits returned
`min_pressure`	Storage pre-filter on raw `p_magnitude`
`search_cost`	TAS looseness: 0.0 (strict) → 1.0 (loose). Lowers θ_eff. Use 0.85+ for low-P memories
`drawer`	Restrict to one drawer
`reinforce`	If `True`, bumps P and `retrieval_count` for returned hits
`candidate_limit`	Max signatures loaded for ranking
`page_size`	Keyset page size when loading candidates
`on_recall`	`Callable[[MemoryHit], None]` invoked per hit before reinforce writes

Returns: List[MemoryHit] sorted by coupling_score × p_effective, length ≤ k.

Raises: nothing — an empty list means no candidates existed or none coupled.

`get()`

python

1def get(self, memory_id: str) -> Optional[MemoryHit]

Returns: a MemoryHit with live P_effective, or None if the memory doesn't exist or was soft-deleted.

Raises: nothing.

`update()`

python

1def update(
2    self,
3    memory_id: str,
4    *,
5    text: Optional[str] = None,
6    tags: Optional[List[str]] = None,
7    p_magnitude: Optional[float] = None,
8    t_persistence: Optional[float] = None,
9    drawer: Optional[str] = None,
10    regime: Optional[str] = None,
11    source: Optional[str] = None,
12    metadata: Optional[Dict[str, Any]] = None,
13) -> MemoryHit

Returns: the updated MemoryHit.

Raises:

ValueError — memory not found, empty text, p_magnitude outside 0–100, or no fields provided at all
The storage layer may also raise ValueError on non-whitelisted column names

`delete()`

python

1def delete(self, memory_id: str) -> bool

Behavior: soft delete (is_deleted=1) on SQLite/Postgres; a metadata flag on cloud.

Returns: True if deleted, False if the memory wasn't found.

`reinforce()`

python

1def reinforce(self, memory_id: str, coupling_score: float = 0.5) -> None

Formula: Δp = REINF_BASE × (1 + ln(1 + retrieval_count)) × coupling_score, capped at P=100.

Raises: ValueError if the memory isn't found.

`decay()`

python

1def decay(self, dry_run: bool = False) -> Dict[str, int]

Parameter	Description
`dry_run`	If `True`, counts purges without writing anything

Returns: {"decayed": 0, "deleted": int, "skipped": int} — decayed is always 0, kept only for API compatibility.

Behavior: hard-deletes signatures whose live P_effective is below 30.

`explain()`

python

1def explain(self, memory_id: str, query: Optional[str] = None) -> ExplainReport

Returns: an ExplainReport — call .render() on it for an ASCII breakdown of every pressure component.

Raises: KeyError if the memory doesn't exist.

`sync()`

python

1def sync(
2    self,
3    direction: str = "push",
4    cloud_url: Optional[str] = None,
5    token: Optional[str] = None,
6) -> SyncReport

`direction`	Behavior
`push`	Local → cloud
`pull`	Cloud → local
`bidirectional`	Both; higher `p_magnitude` wins conflicts

Returns: SyncReport(pushed, pulled, conflicts_resolved, errors).

Raises: RuntimeError if cloud isn't configured, or if storage is cloud-only.

`surface()`

python

1def surface(
2    self,
3    query: str,
4    k: int = 5,
5    *,
6    search_cost: float = 0.65,
7    torsion_threshold: float = 0.70,
8    min_goal_pressure: float = 60.0,
9    reinforce: bool = False,
10) -> SurfaceReport

Returns: a combined snapshot — recall, a full-store torsion scan, and a GAA check — all run against query in one call.

`detect_torsion()`

python

1def detect_torsion(
2    self,
3    drawer: Optional[str] = None,
4    threshold: float = 0.7,
5    *,
6    apply_v_penalty: bool = False,
7    limit: int = 10_000,
8) -> List[TorsionReport]

Parameter	Description
`drawer`	Scope to one drawer
`threshold`	Minimum `torsion_score` to report
`apply_v_penalty`	Increments `validation_prediction_total` on conflicting IDs (lowers future V)
`limit`	Max signatures loaded

Returns: List[TorsionReport], descending by score.

Raises: ValueError if threshold is outside [0, 1].

`verify_alignment()`

python

1def verify_alignment(
2    self,
3    intent_text: str,
4    *,
5    min_pressure: float = 60.0,
6    k_goals: int = 8,
7    torsion_threshold: float = 0.70,
8) -> AlignmentReport

Returns: an AlignmentReport with status ∈ {ALIGNED, CONFLICT, TORSION}.

Raises: ValueError if intent_text is empty (raised inside the alignment module).

Gate: check report.is_safe_to_act or status == "ALIGNED" before letting an agent act.

Additional public methods

Method	Signature	Returns
`reconcile_torsion`	`(signature_a_id, signature_b_id, reconciled_text) -> str`	New signature ID; soft-deletes the conflicting pair
`export_json`	`(path, *, limit=100_000) -> int`	Count exported
`import_json`	`(path, *, skip_duplicates=True) -> dict`	`{saved, skipped, errors}`
`list`	`(limit=50, min_pressure=0, drawer=None, cursor_id=None) -> MemoryListPage`	Keyset page of `MemoryHit`
`list_drawers`	`() -> List[DrawerInfo]`	Drawer aggregates
`count`	`() -> int`	Active signature count
`ingest`	`(data_source, mapping=None, …) -> dict`	Bulk ingest counts
`close`	`() -> None`	Release connections

3. Storage Drivers & Configuration

SQLite (default)

python

1from pdm_memory import Memory
2
3mem = Memory(store="./app.db")                          # path
4mem = Memory(store="sqlite:///./data/pdm.db")           # URL

Zero extra dependencies (stdlib sqlite3)
WAL mode, thread-local connections
transaction() uses BEGIN IMMEDIATE
Schema and migrations apply automatically on init

PostgreSQL

bash

1pip install "pdm-memory[postgres]"

python

1mem = Memory(store="postgresql://user:pass@localhost:5432/pdm")
2# or
3from pdm_memory.storage.factory import create_storage
4driver = create_storage("postgresql://admin@localhost:5432/pdm_smoke")

Requires psycopg >= 3.1. Schema semantics are identical to SQLite.

Cloud (Westfield Companion API)

python

1mem = Memory(
2    store="cloud",
3    token="eyJ...",
4    refresh_token="...",          # optional auto-refresh
5    cloud_url="https://api.azus.ai",
6    user="alice",
7)

HTTP via httpx; fails fast on auth errors or 5xx
Soft delete via metadata._pdm_is_deleted
Idempotency via metadata._idempotency_key
ping() issues a GET retrieve with limit=1

`create_storage()` factory

python

1from pdm_memory.storage.factory import create_storage, register_storage
2
3driver = create_storage("./local.db", store_raw=True)
4driver = create_storage("postgresql://...", store_raw=False)
5
6# Custom backend
7register_storage("redis", lambda url, **kw: RedisDriver.from_url(url, **kw))
8mem = Memory(store="redis://localhost:6379/0")

Raises: ValueError (unknown scheme, missing cloud token), ImportError (the postgres extra isn't installed).

`Memory.from_env()` — 12-factor

bash

1export PDM_STORE=postgresql://localhost/pdm
2export PDM_USER=alice
3export PDM_TOKEN=eyJ...   # if cloud

python

1mem = Memory.from_env()

Automatic migrations

Migrations run automatically on driver init — you never write manual SQL to upgrade an existing database.

SQLite — apply_sqlite_migrations(conn):

ALTER TABLE … ADD COLUMN is_deleted if missing
ALTER TABLE … ADD COLUMN idempotency_key if missing
Creates idx_pdm_user_active_pressure, idx_pdm_user_idempotency

PostgreSQL — apply_postgres_migrations(conn):

ADD COLUMN IF NOT EXISTS for the same columns and indexes

Health check

python

1driver = create_storage("./app.db")
2assert driver.ping() is True   # SELECT 1 on SQL drivers

Explorer equivalent: GET /api/v1/health → { "status": "ok", "storage_ok": true, … }

4. Models & Data Structures

`SignatureRecord` (dataclass)

This is the raw row shape stored by every backend.

python

1@dataclass
2class SignatureRecord:
3    id: str                          # UUID, default factory
4    user: str = "default"
5    compressed_fact: str = ""
6    source: str = "chat"
7    p_magnitude: float = 50.0
8    t_persistence: float = 30.0
9    phase_privilege: float = 1.0
10    effective_spike: Optional[float] = None
11    intent_tags: List[str] = field(default_factory=list)
12    question_regime: str = "neutral"
13    domain: str = "insight"
14    retrieval_count: int = 0
15    last_retrieved: Optional[datetime] = None
16    created_at: Optional[datetime] = None
17    validation_prediction_total: int = 0
18    validation_prediction_correct: int = 0
19    decay_rate: float = 0.9          # legacy schema field; not used in P_eff
20    t_deadline: Optional[datetime] = None
21    urgency_rate: float = 2.0
22    drawer_domain: str = "general"
23    metadata: Dict[str, Any] = field(default_factory=dict)
24    is_deleted: bool = False
25    idempotency_key: Optional[str] = None

`MemoryHit` (dataclass)

This is what recall(), get(), and list() actually hand back to you — a signature plus everything computed at read time.

python

1@dataclass
2class MemoryHit:
3    id: str
4    text: str
5    source: str
6    drawer: str
7    pressure: float              # live P_effective
8    p_raw: float                 # stored p_magnitude
9    p_effective: float
10    decay_factor: float
11    intent_weight: float
12    v_coefficient: float
13    quality: float
14    last_reinforced: Optional[datetime]
15    retrieval_count: int
16    intent_tags: List[str]
17    domain: str
18    coupling_score: float = 0.0
19    tag_overlap: float = 0.0
20    domain_match: float = 0.0
21    regime_match: float = 0.0
22    pressure_proximity: float = 0.0
23    e_temporal: Optional[float] = None
24    is_urgent: bool = False

`TorsionReport` (dataclass, slots)

python

1@dataclass(slots=True)
2class TorsionReport:
3    signature_a_id: str
4    signature_b_id: str
5    signature_a_text: str
6    signature_b_text: str
7    drawer: str
8    domain: str
9    torsion_score: float
10    topic_similarity: float
11    contradiction_strength: float
12    explanation: str
13    conflict_kind: str           # deadline | factual | polarity | pressure | semantic
14    cluster_key: Optional[str] = None

`AlignmentReport` (dataclass, slots)

python

1@dataclass(slots=True)
2class AlignmentReport:
3    status: str                  # ALIGNED | CONFLICT | TORSION
4    score: float
5    conflicting_goals: List[str] = field(default_factory=list)
6    explanation: str = ""
7    resonance: float = 0.0
8    torsion: float = 0.0
9    anchor_count: int = 0
10
11    @property
12    def is_safe_to_act(self) -> bool: ...
13    def as_dict(self) -> dict[str, Any]: ...
14    def render(self) -> str: ...

`SurfaceReport` (dataclass, slots)

python

1@dataclass(slots=True)
2class SurfaceReport:
3    hits: List[Any]              # List[MemoryHit]
4    torsion_count: int
5    alignment: str
6    alignment_score: float = 0.0
7    torsion_reports: List[TorsionReport] = field(default_factory=list)
8
9    def as_dict(self) -> dict[str, Any]: ...

`MemoryListPage` (keyset pagination)

python

1@dataclass(slots=True)
2class MemoryListPage:
3    items: List[Any]             # List[MemoryHit]
4    next_cursor_id: Optional[str] = None

`ExplainReport` (dataclass)

Full pressure decomposition plus optional TAS coupling fields. See pdm_memory/core/signature.py for the exact field list. Method: .render() -> str.

`SyncReport` (dataclass)

python

1@dataclass
2class SyncReport:
3    direction: str
4    pushed: int = 0
5    pulled: int = 0
6    conflicts_resolved: int = 0
7    errors: int = 0

5. CLI Command Manual (`pdm-cli`)

Global options, available on every command:

Flag	Default	Description
`--store`	`./pdm_memory.db`	Path or URL (`sqlite:///`, `postgresql://`)
`--user`	`default`	User scope

`list-memories`

bash

1pdm-cli list-memories --store ./app.db --user alice \
2  --min-pressure 50 --limit 50 --drawer prefs

Flag	Default	Description
`--min-pressure`	`0.0`	Filter on raw P
`--limit`	`50`	Max rows
`--drawer`	none	Drawer filter

`search`

bash

1pdm-cli search "metric units" --store ./app.db --limit 10 \
2  --search-cost 0.85 --min-pressure 0

Flag	Default	Description
`query`	(positional)	Recall query
`--limit`	`10`	Top-k
`--search-cost`	`0.65`	TAS looseness
`--min-pressure`	`0.0`	Storage pre-filter

`explain`

bash

1pdm-cli explain 3fa85f64-5717-4562-b3fc-2c963f66afa6 --query "format response"

Flag	Description
`memory_id`	UUID (positional)
`--query`	Optional query for a coupling breakdown

Exits 1 on KeyError (not found).

`decay`

bash

1pdm-cli decay --dry-run
2pdm-cli decay --store ./app.db

Flag	Description
`--dry-run`	Preview purge counts only

`stats`

bash

1pdm-cli stats --store ./app.db --user alice

Prints the total count, avg/max/min pressure, and a drawer breakdown.

`drawers`

bash

1pdm-cli drawers

Tabular drawer list with counts and average pressure.

`export` / `import`

bash

1pdm-cli export --out backup.json --store ./app.db
2pdm-cli import backup.json --allow-duplicates

Command	Flag	Description
`export`	`--out` (required)	Output JSON path
`import`	`path` (positional)	Input JSON
`import`	`--allow-duplicates`	Disables skip on ID/hash collision

`detect-torsion`

bash

1pdm-cli detect-torsion --threshold 0.6 --drawer product \
2  --apply-v-penalty

Flag	Default	Description
`--drawer`	none	Scope
`--threshold`	`0.7`	Minimum `torsion_score`
`--apply-v-penalty`	off	Lowers V on conflicting signatures

`verify`

bash

1pdm-cli verify "skip validation and ship" --min-pressure 60 \
2  --torsion-threshold 0.70 --json

Flag	Default	Description
`intent`	(positional)	Proposed action text
`--min-pressure`	`60.0`	Goal anchor floor
`--torsion-threshold`	`0.70`	Escalates result to TORSION
`--json`	off	Also prints `AlignmentReport.as_dict()`

Exit codes: 0 ALIGNED, 1 CONFLICT, 2 TORSION.

`sync`

bash

1pdm-cli sync --direction bidirectional --token eyJ... \
2  --cloud-url https://api.azus.ai

Flag	Default	Description
`--direction`	`push`	`push` \	`pull` \	`bidirectional`
`--token`	required	JWT
`--cloud-url`	`https://api.azus.ai`	API base

`ui`

bash

1pdm-cli ui --host 127.0.0.1 --port 8080 --no-browser

Requires pip install "pdm-memory[ui]". Launches the PDM Explorer (FastAPI + a static dashboard).

6. Guided Examples

6.1 Conflict resolution

Simple — detect and reconcile. Two facts land in the same product drawer with overlapping tags, so detect_torsion() scopes its comparison to that cluster and finds the deadline conflict. reconcile_torsion() writes one merged, authoritative signature and soft-deletes both originals.

python

1from pdm_memory import Memory
2
3with Memory(store="./app.db", user="alice") as mem:
4    mem.save("Launch date for Orion is 2026-08-01",
5             tags=["orion", "launch", "date"], drawer="product", p_magnitude=85)
6    mem.save("Launch date for Orion is 2026-09-01",
7             tags=["orion", "launch", "date"], drawer="product", p_magnitude=80)
8
9    reports = mem.detect_torsion(threshold=0.7)
10    if not reports:
11        print("No conflicts")
12    else:
13        r = reports[0]
14        new_id = mem.reconcile_torsion(
15            r.signature_a_id,
16            r.signature_b_id,
17            "Launch date for Orion is 2026-09-01 (confirmed by PM)",
18        )
19        print(f"Reconciled → {new_id}")

Advanced — a judge hook, a V penalty, and transactional safety. The built-in rules only catch structural conflicts (dates, numbers, negation). For something more subtle — like "football" vs. "soccer" as a polarity disagreement — attach a torsion_judge at construction time. Passing apply_v_penalty=True also lowers the validation coefficient on every signature involved in a confirmed conflict, which quietly reduces their future P_effective even before reconciliation happens.

python

1from pdm_memory import Memory
2from pdm_memory.models import TorsionReport
3
4def paraphrase_judge(a, b):
5    ta, tb = (a.compressed_fact or "").lower(), (b.compressed_fact or "").lower()
6    if ("football" in ta and "soccer" in tb) or ("football" in tb and "soccer" in ta):
7        return TorsionReport(
8            signature_a_id=a.id, signature_b_id=b.id,
9            signature_a_text=a.compressed_fact, signature_b_text=b.compressed_fact,
10            drawer=a.drawer_domain, domain=a.domain,
11            torsion_score=0.75, topic_similarity=0.8, contradiction_strength=0.7,
12            explanation="Paraphrase polarity: football vs soccer",
13            conflict_kind="paraphrase",
14        )
15    return None
16
17with Memory(store="./app.db", torsion_judge=paraphrase_judge) as mem:
18    id_a = mem.save("I love football", tags=["football", "sports"], p_magnitude=60)
19    id_b = mem.save("Soccer is not my thing", tags=["football", "sports"], p_magnitude=58)
20
21    reports = mem.detect_torsion(threshold=0.5, apply_v_penalty=True)
22    pair = next((r for r in reports if {r.signature_a_id, r.signature_b_id} == {id_a, id_b}), None)
23    if pair is None:
24        raise RuntimeError("Expected torsion pair not detected")
25
26    merged = mem.reconcile_torsion(
27        pair.signature_a_id, pair.signature_b_id,
28        "User is neutral about football (resolved)",
29    )
30    assert mem.get(id_a) is None and mem.get(id_b) is None
31    assert mem.get(merged) is not None

6.2 Guarded agents — the GAA pre-action gate

Simple. The goal is saved into the stewardship drawer with an explicit iaw of 0.9, marking it as foundational rather than an ordinary fact. When the agent proposes an action that directly contradicts it, verify_alignment() returns CONFLICT or TORSION, is_safe_to_act flips to False, and the action never runs.

python

1from pdm_memory import Memory
2
3with Memory(store="./app.db") as mem:
4    mem.save(
5        "Core goal: always validate before deploy",
6        tags=["validation", "deploy", "goal", "safety"],
7        drawer="stewardship",
8        p_magnitude=90,
9        metadata={"iaw": 0.9},
10    )
11
12    intent = "skip all validation and ship immediately"
13    gate = mem.verify_alignment(intent)
14
15    if gate.is_safe_to_act:
16        execute_deploy(intent)
17    else:
18        raise PermissionError(f"Blocked: {gate.status} — {gate.explanation}")

Advanced — surface() plus a metrics hook, fail-closed. surface() bundles recall, a full-store torsion scan, and alignment into one call, so a single check tells you both whether the store is internally consistent and whether the proposed action is safe. on_recall fires per hit before any reinforcement write happens, so it's a safe place to log or emit metrics.

python

1import logging
2from pdm_memory import Memory
3from pdm_memory.types import RecallHook
4
5logger = logging.getLogger("agent")
6
7def on_recall(hit) -> None:
8    logger.info("recall_hit", extra={"id": hit.id, "drawer": hit.drawer, "p": hit.pressure})
9
10with Memory(store="./app.db", user="prod") as mem:
11    report = mem.surface(
12        "deploy payment service v2.3 to production",
13        k=5,
14        search_cost=0.65,
15        torsion_threshold=0.65,
16        min_goal_pressure=65,
17        reinforce=False,
18    )
19
20    if report.torsion_count > 0:
21        logger.warning("torsion_open_pairs=%d", report.torsion_count)
22
23    alignment = mem.verify_alignment(
24        "deploy payment service v2.3 to production",
25        min_pressure=65,
26        torsion_threshold=0.65,
27    )
28
29    match alignment.status:
30        case "ALIGNED":
31            hits = mem.recall(
32                "deploy payment service",
33                k=3,
34                on_recall=on_recall,
35                reinforce=False,
36            )
37            proceed(hits)
38        case "CONFLICT":
39            raise PermissionError(alignment.explanation)
40        case "TORSION":
41            raise PermissionError(f"TORSION vs goals: {alignment.conflicting_goals}")
42        case _:
43            raise RuntimeError(f"Unknown alignment status: {alignment.status}")

6.3 Data mobility — JSON export/import

Simple. export_json() serializes every active signature for the user; import_json() replays them, and with skip_duplicates=True it skips anything that collides on ID or content hash rather than duplicating it.

python

1from pdm_memory import Memory
2
3with Memory(store="./prod.db", user="alice") as src:
4    n = src.export_json("./backup-alice.json")
5    print(f"Exported {n} signatures")
6
7with Memory(store="./staging.db", user="alice") as dst:
8    stats = dst.import_json("./backup-alice.json", skip_duplicates=True)
9    print(stats)  # {'saved': N, 'skipped': M, 'errors': 0}

Advanced — cross-backend migration with verification. The same export/import pair doubles as a migration path between backends. Always check export_json()'s return value against count() before trusting a migration, and check import_json()'s errors key afterward — an import never raises on a bad row, it just tallies it.

python

1from pdm_memory import Memory
2from pdm_memory.storage.factory import create_storage
3
4SQLITE = "./local.db"
5PG = "postgresql://admin@localhost:5432/pdm_smoke"
6BACKUP = "./migration.json"
7
8# Export from SQLite
9with Memory(store=SQLITE, user="team") as mem:
10    count = mem.export_json(BACKUP, limit=100_000)
11    assert count == mem.count(), "Export count mismatch"
12
13# Import into Postgres
14with Memory(store=PG, user="team") as mem:
15    result = mem.import_json(BACKUP, skip_duplicates=True)
16    if result["errors"]:
17        raise RuntimeError(f"Import errors: {result}")
18
19    assert mem._storage.ping(), "Postgres unreachable"
20    page = mem.list(limit=50)
21    while page.next_cursor_id:
22        page = mem.list(limit=50, cursor_id=page.next_cursor_id)
23    print(f"Postgres active count: {mem.count()}")

6.4 Custom backends — implementing `BaseStorage`

Storage is an explicit interface. If you need a backend beyond SQLite, Postgres, or Cloud — Redis, DynamoDB, whatever fits your infrastructure — implement BaseStorage and register it. Once registered, the scheme in your store URL routes straight to your driver, and the rest of the SDK (recall, save, decay, surface) works unmodified on top of it.

python

1from __future__ import annotations
2
3import json
4from typing import List, Optional
5
6from pdm_memory.core.signature import DrawerInfo, SignatureRecord
7from pdm_memory.storage.base import BaseStorage
8from pdm_memory.storage.factory import register_storage
9
10
11class RedisStorage(BaseStorage):
12    """Minimal in-memory Redis-backed sketch — production needs indexing."""
13
14    def __init__(self, url: str) -> None:
15        import redis
16        self._r = redis.from_url(url)
17        self._prefix = "pdm:sig:"
18
19    def save(self, sig: SignatureRecord) -> str:
20        key = f"{self._prefix}{sig.user}:{sig.id}"
21        self._r.set(key, json.dumps(sig.__dict__, default=str))
22        return sig.id
23
24    def get(self, memory_id: str, user: str = "default") -> Optional[SignatureRecord]:
25        raw = self._r.get(f"{self._prefix}{user}:{memory_id}")
26        if not raw:
27            return None
28        data = json.loads(raw)
29        if data.get("is_deleted"):
30            return None
31        return SignatureRecord(**data)
32
33    def update(self, memory_id: str, user: str = "default", **fields) -> None:
34        rec = self.get(memory_id, user)
35        if rec is None:
36            return
37        for k, v in fields.items():
38            setattr(rec, k, v)
39        self.save(rec)
40
41    def delete(self, memory_id: str, user: str = "default") -> None:
42        self.update(memory_id, user=user, is_deleted=True)
43
44    def list(
45        self,
46        user: str = "default",
47        limit: int = 100,
48        min_pressure: float = 0.0,
49        drawer: Optional[str] = None,
50        cursor_id: Optional[str] = None,
51        include_deleted: bool = False,
52    ) -> List[SignatureRecord]:
53        # Production: use Redis sorted sets by (p_magnitude, id)
54        keys = self._r.keys(f"{self._prefix}{user}:*")
55        out: List[SignatureRecord] = []
56        for k in keys[:limit]:
57            rec = self.get(k.decode().split(":")[-1], user=user)
58            if rec and rec.p_magnitude >= min_pressure:
59                if drawer and rec.drawer_domain != drawer:
60                    continue
61                out.append(rec)
62        out.sort(key=lambda r: (-r.p_magnitude, r.id))
63        return out[:limit]
64
65    def list_drawers(self, user: str = "default") -> List[DrawerInfo]:
66        return []
67
68    def ping(self) -> bool:
69        return self._r.ping()
70
71
72def _build_redis(url: str, **_: object) -> BaseStorage:
73    return RedisStorage(url)
74
75
76register_storage("redis", _build_redis)
77
78# Usage
79from pdm_memory import Memory
80
81mem = Memory(store="redis://localhost:6379/0", user="alice")
82mem.save("Redis-backed fact", tags=["redis", "custom", "backend"])

Production requirements for custom drivers:

Keyset pagination (cursor_id) on an indexed (p_magnitude DESC, id DESC)
Indexed find_by_hash / find_by_idempotency_key
A real transaction() for atomic batches
hard_delete() kept distinct from the soft delete()
A working ping() for health probes

7. RetrievalEngine & Math Constants

Reach for RetrievalEngine directly when you already have SignatureRecord lists in memory — batch analytics, unit tests, custom pipelines — and don't need the full Memory facade.

python

1from pdm_memory.core.retrieval import RetrievalEngine, ALPHA_DEFAULT, THETA_BASE_DEFAULT
2from pdm_memory.core.math import P_MAX, DECAY_DELETE_THRESHOLD, calculate_temporal_geometry
3
4engine = RetrievalEngine(alpha=0.7, coupling_min=0.3)
5hits = engine.recall(
6    records=candidates,
7    query="format numbers for user",
8    k=5,
9    search_cost=0.65,
10    base_threshold=30.0,
11)

TAS constants (`pdm_memory.core.retrieval`)

Constant	Default	Meaning
`ALPHA_DEFAULT`	`0.7`	Threshold-lowering aggressiveness: `θ_eff = θ_base × (1 - α × search_cost)`
`THETA_FLOOR`	`5.0`	Absolute minimum effective threshold
`THETA_BASE_DEFAULT`	`30.0`	Starting pressure gate before search-cost adjustment
`COUPLING_MIN_DEFAULT`	`0.3`	Minimum coupling to count as "coupled"
`W_TAGS` / `W_DOMAIN` / `W_REGIME` / `W_PRESSURE`	`0.50` / `0.20` / `0.15` / `0.15`	Impedance-matching weights
`AUTO_FIRE_THRESHOLD`	`85.0`	P_eff above this → auto-fire eligible
`REINF_BASE`	`2.0`	Reinforcement delta base: `Δp = REINF_BASE × log(1 + retrieval_count) × coupling`
`_TOPIC_GATE`	`0.35`	Minimum topic similarity before a torsion pair is even evaluated

Pressure math constants (`pdm_memory.core.math`)

Constant	Value	Meaning
`P_MAX`	`100.0`	Pressure ceiling
`DECAY_DELETE_THRESHOLD`	`30.0`	`decay()` hard-deletes below this live P_eff
`DEFAULT_HALF_LIFE`	`30.0`	Fallback half-life (days)
`DOMAIN_HALF_LIVES`	see table below	Per-domain decay clock

Domain half-lives (days):

Domain	Half-life
`market_signal`	1
`warning`	3
`reminder`	7
`pattern`	14
`insight`	30
`structural`	90
`core_fact`	365

Core formulas

typescript

1effective_spike = min(100, P_magnitude × (t_persistence/30) × phase_privilege)
2decay_factor    = 1 - exp(-λ × t)     where λ = ln2 / half_life
3V               = (correct + 1) / (total + 2)   [Laplace smoothing]
4P_effective     = P × V × (1 - decay_factor) × intent_weight × quality

Grace period: if days_since_created ≤ t_persistence → decay_factor = 0.

PDM-T temporal geometry

For deadline-driven memories (save(..., deadline=datetime)), PDM computes an urgency profile alongside the standard pressure math:

python

1from pdm_memory.core.math import calculate_temporal_geometry
2
3geom = calculate_temporal_geometry(
4    t_remaining_days=5.0,
5    persist_days=30.0,
6    c_base=1.0,
7    s_base=1.0,
8    p_base=1.0,
9    urgency_rate=2.0,
10    decay_rate=0.9,
11    temporal_weight=1.0,
12)
13# geom["is_urgent"], geom["e_temporal"], geom["status"]  # "URGENT" | "ACTIVE" | "EXPIRED"

RetrievalEngine methods beyond `recall`

Method	Purpose
`detect_torsion(records, threshold=0.7, drawer=None, judge=None)`	Find contradictory pairs
`verify_alignment(records, goal_text, iaw_threshold=0.5)`	GAA gate before agent actions

Passing torsion_judge=fn to Memory(...) simply forwards it to detect_torsion under the hood.

8. BaseStorage Contract & Database Schema

Full interface (`pdm_memory.storage.base.BaseStorage`)

Every backend — SQLite, Postgres, Cloud, or a custom driver you register — implements this same contract, which is what lets Memory stay backend-agnostic.

Method	Signature	Notes
`save`	`(sig: SignatureRecord) -> str`	INSERT only; duplicates raise or are deduped at the `Memory` layer
`get`	`(memory_id, user="default") -> SignatureRecord \	None`	Returns None for soft-deleted rows
`update`	`(memory_id, user="default", **fields) -> None`	Whitelist enforced — see `UPDATABLE_COLUMNS`
`update_batch`	`(updates: list[tuple[str, dict]], user) -> None`	Default implementation just loops `update()`
`delete`	`(memory_id, user) -> None`	Soft delete (`is_deleted=True`)
`hard_delete`	`(memory_id, user) -> None`	Permanent removal; used by `decay()`
`list`	`(user, limit=100, min_pressure=0, drawer=None, cursor_id=None, include_deleted=False)`	Keyset order: `p_magnitude DESC, id DESC`
`list_drawers`	`(user) -> list[DrawerInfo]`	Aggregates per drawer
`count`	`(user) -> int`	Default implementation scans up to 10k rows
`find_by_hash`	`(text_hash, user) -> SignatureRecord \	None`	Dedupe lookup
`find_by_idempotency_key`	`(key, user) -> SignatureRecord \	None`	Idempotency lookup
`ping`	`() -> bool`	Connectivity probe
`transaction`	`() -> contextmanager`	SQLite/Postgres use real transactions
`close`	`() -> None`	Release connections

`UPDATABLE_COLUMNS` whitelist

Only the following fields may be passed to storage.update() / Memory.update():

typescript

1compressed_fact, compressed_fact_hash, source, p_magnitude, t_persistence,
2phase_privilege, effective_spike, intent_tags, question_regime, domain,
3drawer_domain, retrieval_count, last_retrieved, created_at,
4validation_prediction_total, validation_prediction_correct, decay_rate,
5t_deadline, urgency_rate, metadata, is_deleted, idempotency_key

Passing an unknown key raises ValueError on the SQLite and Postgres drivers.

SQLite schema (`pdm_signatures`)

Key columns: id, user, compressed_fact, compressed_fact_hash, p_magnitude, intent_tags (JSON), drawer_domain, is_deleted (INTEGER 0/1), idempotency_key.

Indexes:

(user, p_magnitude DESC) — recall listing
(user, drawer_domain) — drawer filter
(user, compressed_fact_hash) — dedupe
(user, p_magnitude DESC, id DESC) WHERE is_deleted = 0 — keyset pagination
(user, idempotency_key) WHERE idempotency_key IS NOT NULL — idempotency unique

Migrations run automatically on driver init via schema.py — you don't manage them by hand.

Privacy mode (`store_raw=False`)

When you construct Memory(store="./app.db", store_raw=False):

compressed_fact is stored as [HASH:<sha256>] — raw text is never persisted locally.
recall() still works via tag/semantic coupling, but recall quality on hashes is limited — use cloud storage or store_raw=True if you need full-text recall locally.

Thread safety

SQLiteDriver uses thread-local connections — use one Memory instance per thread, or switch to Postgres if you need shared, concurrent writers.

9. Metadata Conventions & Domain Inference

Recommended `metadata` keys

Key	Type	Purpose
`iaw`	`float` 0–1	Intent Alignment Weight for GAA (`verify_alignment`)
`cluster_id`	`str`	Groups related memories for torsion clustering
`is_anchor`	`bool`	Goal anchor — higher alignment weight in GAA
`role`	`str`	e.g. `"user"`, `"assistant"`, `"system"` for ingest
`_idempotency_key`	`str`	Cloud fallback when the column is absent
`_pdm_is_deleted`	`bool`	Cloud soft-delete flag

Domain inference (`infer_domain(tags)`)

If you don't set drawer explicitly, PDM infers a domain from your tags, checking these keyword groups in priority order (first match wins):

Tag keywords	Domain
market, signal, price, trade, stock	`market_signal`
pattern, historical, analogue	`pattern`
structure, structural, model	`structural`
remind, deadline, due, by	`reminder`
warning, risk, danger	`warning`
fact, law, rule, principle	`core_fact`
(default)	`insight`

Regime inference (`infer_regime(tags)`)

Tag keywords	Regime
trade, stock, market, price	`trading`
code, engineer, deploy, bug, api	`engineering`
personal, health, family	`personal`
patent, ip, monetize, license	`ip_monetize`
(default)	`neutral`

If you omit both drawer and regime on save(), the domain is inferred from tags and the regime falls back to "neutral" unless you pass it explicitly.

10. LLM Integrations

Install the extra you need: pip install "pdm-memory[openai]" (or [anthropic], [gemini], [ollama], [groq], or [all] for everything).

Every wrapper follows the same three-step pattern:

recall(query) → inject the top-k memories into the system/context.
Call the provider's API.
Optionally save() the assistant's reply back into memory.

OpenAI

python

1from pdm_memory import Memory
2from pdm_memory.integrations import wrap_openai
3
4mem = Memory(store="./app.db", user="alice")
5client = wrap_openai(
6    api_key="sk-...",
7    memory=mem,
8    model="gpt-4o-mini",
9    max_memory_tokens=1500,
10    recall_k=5,
11    system_prompt="You are a helpful assistant.",
12)
13response = client.chat.completions.create(
14    messages=[{"role": "user", "content": "How should I format numbers?"}],
15)

Anthropic, Gemini, Ollama, Groq

The other adapters take the same shape of arguments — provider credentials plus the shared memory, model, max_memory_tokens, recall_k kwargs.

python

1from pdm_memory.integrations import (
2    wrap_anthropic, wrap_gemini, wrap_ollama, wrap_groq,
3)
4
5client = wrap_anthropic(api_key="...", memory=mem)
6client = wrap_gemini(api_key="...", memory=mem)
7client = wrap_ollama(base_url="http://localhost:11434", memory=mem)
8client = wrap_groq(api_key="...", memory=mem)

`ContextWindowManager` — manual injection

If you'd rather not use a wrapper and want to build the prompt yourself, use the token-budget manager directly:

python

1from pdm_memory.integrations import ContextWindowManager
2
3mgr = ContextWindowManager(max_tokens=1500, model="gpt-4o-mini")
4hits = mem.recall("user preferences", k=10)
5trimmed = mgr.fit(hits)
6block = mgr.format_for_prompt(trimmed)  # inject into your own prompt

Trimming drops the lowest P_effective memories first, until everything fits inside the token budget.

11. Data Ingestion Pipeline

Via `Memory.ingest()`

ingest() is the bulk-loading entry point — point it at rows of data and it maps them onto signatures automatically.

python

1stats = mem.ingest(
2    data_source=[
3        {"text": "User prefers metric", "importance": 70, "tags": "units,prefs"},
4        {"message": "Deadline Q3 2026", "priority": 80},
5    ],
6    mapping={"text": "compressed_fact", "importance": "p_magnitude"},  # optional
7    llm_client=None,       # optional OpenAI/Anthropic client for auto-tags
8    batch_size=50,
9    on_progress=lambda done, total: print(f"{done}/{total}"),
10)
11# stats: {"saved": N, "skipped": M, "errors": E}

Supported data_source types:

list[dict] — row-oriented data
str — a CSV file path or a raw CSV string
list[str] — one fact per string (routed through BatchProcessor)

Auto field aliases (`DataIngester`)

You don't have to normalize your column names before ingesting — the ingester recognizes common aliases automatically:

Source alias	Maps to
text, content, message, body, fact, memory, summary	`compressed_fact`
importance, priority, pressure, score, weight	`p_magnitude`
tags, labels, categories, keywords	`intent_tags`
category, drawer, topic, domain	`drawer_domain`
origin, channel	`source`
context, regime	`question_regime`

Rows without extractable text are skipped (counted in skipped). If tags are missing, they're auto-generated from keywords — but 3 or more explicit tags are still recommended for the best recall quality.

Standalone ingester

python

1from pdm_memory.ingest import DataIngester
2
3ingester = DataIngester(storage=mem._storage, user="alice")
4stats = ingester.ingest_csv("/path/to/export.csv")

12. PDM Explorer HTTP API

The Explorer is a small FastAPI app that gives you a visual dashboard and a REST surface over your store.

Launch it either through the CLI or directly with uvicorn:

bash

1pdm-cli ui --store ./local.db --user alice --port 8080
2# or
3uvicorn pdm_memory.tools.server:create_app --factory --host 127.0.0.1 --port 8080

Factory kwargs: create_app(store="./pdm_memory.db", user="default").

Routes

Method	Path	Params / Body	Response
`GET`	`/`	—	Static dashboard HTML
`GET`	`/api/v1/health`	—	`{status, store, user, storage_ok}`
`GET`	`/api/v1/memory-map`	`limit`, `torsion_threshold`, `link_threshold`, `projected_days`	`{nodes[], links[], count, torsion_count}`
`GET`	`/api/v1/torsion`	`threshold`, `drawer?`	`{count, latest, reports[]}`
`GET`	`/api/v1/search`	`q` (required), `k`, `search_cost`	`{query, hits[{id,text,p_effective,coupling_score,drawer}]}` — does not reinforce
`POST`	`/api/v1/memories/{id}/reinforce`	`{coupling_score: 0.65}`	`{ok, p_magnitude_before, node}`
`DELETE`	`/api/v1/memories/{id}`	—	Soft-delete via `Memory.delete()`
`POST`	`/api/v1/torsion/resolve`	`{signature_a_id, signature_b_id, use_ai?, reconciled_text?}`	`{ok, method, reconciled_text, new_memory_id, deleted_ids[], node}`

Note: the Explorer's DELETE route performs a soft delete, not a hard one. Use decay() or call storage.hard_delete() directly when you need a permanent purge.

13. Sync, Auth & Errors

Environment-based construction

python

1import os
2os.environ["PDM_STORE"] = "postgresql://localhost/pdm"
3os.environ["PDM_USER"] = "alice"
4# os.environ["PDM_TOKEN"] = "eyJ..."       # required for cloud
5# os.environ["PDM_REFRESH_TOKEN"] = "..."  # optional
6# os.environ["PDM_CLOUD_URL"] = "https://api.azus.ai"
7
8mem = Memory.from_env()  # fails fast if PDM_STORE is missing

Cloud storage

python

1mem = Memory(
2    store="cloud",
3    user="alice",
4    token="eyJ...",
5    refresh_token="eyJ...",           # optional auto-refresh
6    cloud_url="https://api.azus.ai",
7)

CloudDriver maps onto the Companion API:

GET /api/v1/pdm/retrieve — list
GET/PATCH/DELETE /api/v1/pdm/signatures/{id}/ — CRUD
Soft delete via metadata._pdm_is_deleted=True (there's no native column for this yet)

`JWTAuth`

python

1from pdm_memory.auth import JWTAuth
2
3auth = JWTAuth(
4    token="eyJ...",
5    refresh_token="eyJ...",
6    refresh_url="https://api.azus.ai/api/v1/accounts/auth/refresh/",
7)
8headers = auth.headers()          # {"Authorization": "Bearer ..."}
9auth.ensure_fresh()               # refresh if near expiry

`MemorySync`

python

1from pdm_memory.sync import MemorySync
2
3sync = MemorySync(local_storage=local._storage, cloud_storage=cloud._storage)
4report = sync.sync(user="alice", direction="bidirectional", batch_size=50)
5# report.pushed, report.pulled, report.conflicts_resolved, report.errors

Conflict resolution rule: higher p_magnitude wins on push and pull.

Or, more simply, go through the facade: mem.sync(direction="push").

Errors (`pdm_memory.storage.errors`)

Exception	When	Handling
`CloudStorageError`	Network/HTTP 5xx, auth failure	Propagate; don't treat this as an empty store
`CloudNotFoundError`	HTTP 404	`CloudDriver.get()` returns `None`

Never catch CloudStorageError and quietly return [] — doing so hides real outages from your application.

14. JSON Export/Import Format

File envelope (version 1)

json

1{
2  "version": "1",
3  "exported_at": "2026-07-17T12:00:00+00:00",
4  "user": "alice",
5  "count": 42,
6  "signatures": [ { "...": "..." } ]
7}

Per-signature fields (export)

id, user, compressed_fact, source, p_magnitude, t_persistence, phase_privilege, effective_spike, intent_tags, question_regime, domain, drawer_domain, retrieval_count, last_retrieved, created_at, validation_prediction_*, decay_rate, t_deadline, urgency_rate, metadata.

Known gap : the export omits is_deleted and idempotency_key. Import accepts the aliases text/tags/drawer. Re-importing with skip_duplicates=True skips rows by ID or fact hash.

API

python

1n = mem.export_json("./backup.json", limit=100_000)
2stats = mem.import_json("./backup.json", skip_duplicates=True)

Low-level functions, if you need them directly: pdm_memory.io.json_transfer.export_signatures_json / import_signatures_json.

15. Developer Footguns & Best Practices

Pitfall	Fix
Low-P memories never get recalled	Raise `search_cost` (0.7–0.85) or increase `p_magnitude` / add more tags
`save()` without tags	Add 3+ intent tags — domain/regime inference depends on them
N+1 patterns in custom code	Batch with `save_many()`, and page with `list()` rather than looping `count()`
Confusing `delete()` with `decay()`	`delete()` is a soft hide; `decay()` is a hard purge below the P_eff threshold
Explorer `DELETE` isn't a hard delete	Use `storage.hard_delete()` for GDPR-style erasure
Cloud soft delete invisible to old clients	Check `metadata._pdm_is_deleted`
Idempotency across restarts	Pass a stable `idempotency_key=` on `save()`
`store_raw=False`	Recall quality drops — the text is hashed only, not stored
`recall(reinforce=True)` in read-only paths	Pass `reinforce=False` for search/explorer-style queries
Sync lists cap at 10k	Paginate manually for larger corpora
SQLite concurrent writes	Use one writer thread, or switch to Postgres
Passing ORM objects to Celery	Pass memory IDs only
Custom `update()` fields	Must appear in `UPDATABLE_COLUMNS`

User scoping

Every operation is scoped to Memory(..., user="..."). Cross-user access is prevented at the storage layer — don't reuse IDs across users expecting isolation, unless you're also using separate database files per user.

Validation workflow (pre-release)

bash

1cd pdm-memory && .venv/bin/pytest tests/ -q
2cd ../test-pdm-sdk && python feature_smoke.py && python postgres_smoke.py

16. Benchmark Harness

The bundled benchmark compares PDM recall against a keyword-plus-recency baseline on an embedded synthetic dataset.

bash

1python -m pdm_memory.bench              # full suite (10 queries)
2python -m pdm_memory.bench --quick        # 5 scenarios smoke
3python -m pdm_memory.bench --output results.json

It returns a BenchmarkReport covering accuracy, latency, token usage, and storage bytes. Use a fixed seed (--seed 42) for reproducible runs.

17. Appendix — Architecture Map

typescript

1Memory (facade)
2 ├── RetrievalEngine (TAS, torsion, GAA)
3 │    └── alignment.py (IAW, goal anchors)
4 ├── BaseStorage (interface)
5 │    ├── SQLiteDriver
6 │    ├── PostgresDriver
7 │    └── CloudDriver
8 ├── sync.MemorySync (local ↔ cloud)
9 └── io/json_transfer (export/import)

Authorized extension points:

BaseStorage implementations via register_storage()
TorsionJudge callback on Memory(..., torsion_judge=fn)
RecallHook on recall(..., on_recall=fn)
A custom RetrievalEngine via Memory(..., engine=...)

Factual Time vs. Entry Time

Imagine you are talking to Azus today about a meeting that happened last month.

If you use only created_at, Azus will think this meeting is a "new" and "fresh" event.
If you use event_at, Azus anchors the memory in its correct historical context.

Why this matters

Without temporal anchoring, your AI suffers from Recency Bias—new, trivial notes about breakfast can bury older, critical facts about a patent filing simply because the breakfast notes are "newer" in the database.

Using the Temporal Anchor

1. Saving Historical Facts

When ingesting archives or old logs, always provide the event_at parameter.

python

1from datetime import datetime
2from pdm_memory import Memory
3
4with Memory(store="./app.db") as mem:
5    # Anchor this memory to December 2025
6    mem.save(
7        "Project Orion initial architecture was locked",
8        tags=["orion", "history"],
9        event_at=datetime(2025, 12, 15)
10    )

Overview — Intelligent Recall & Diversity Control

Introduction

The Concept of Semantic Quotas

Why this is critical

Implementing Diversity Control

How it works:

Example Code:

Anti-Dilution in GAA

Best Practices

Why use Middleware?

The Three Pillars of Injection

1. The Entry Gate (pre_save)

2. The Execution Signal (post_save)

3. The Retrieval Context (post_recall)

Quick Start: Adding a Hook

1. Core Concepts (The Westfield Lore)

PDM (Persistent-Driven Memory)

p_magnitude vs. P_effective

Torsion / Reverse Resonance

Decay Law

Resonance / Coupling

Identity Anchor Weight (IAW)

SIG (Semantic Information Gain)

t_persistence

2. API Reference — `Memory`

Constructor

Memory.from_env()

save()

save_many()

recall()

get()

update()

delete()

reinforce()

decay()

explain()

sync()

surface()

detect_torsion()

verify_alignment()

Additional public methods

3. Storage Drivers & Configuration

SQLite (default)

PostgreSQL

Cloud (Westfield Companion API)

create_storage() factory

Memory.from_env() — 12-factor

Automatic migrations

Health check

4. Models & Data Structures

SignatureRecord (dataclass)

MemoryHit (dataclass)

TorsionReport (dataclass, slots)

AlignmentReport (dataclass, slots)

SurfaceReport (dataclass, slots)

MemoryListPage (keyset pagination)

ExplainReport (dataclass)

SyncReport (dataclass)

5. CLI Command Manual (`pdm-cli`)

list-memories

search

explain

decay

stats

drawers

export / import

detect-torsion

verify

sync

ui

6. Guided Examples

6.1 Conflict resolution

6.2 Guarded agents — the GAA pre-action gate

6.3 Data mobility — JSON export/import

6.4 Custom backends — implementing BaseStorage

7. RetrievalEngine & Math Constants

TAS constants (pdm_memory.core.retrieval)

Pressure math constants (pdm_memory.core.math)

Core formulas

PDM-T temporal geometry

1. The Entry Gate (`pre_save`)

2. The Execution Signal (`post_save`)

3. The Retrieval Context (`post_recall`)

`p_magnitude` vs. `P_effective`

`t_persistence`

`Memory.from_env()`

`save()`

`save_many()`

`recall()`

`get()`

`update()`

`delete()`

`reinforce()`

`decay()`

`explain()`

`sync()`

`surface()`

`detect_torsion()`

`verify_alignment()`

`create_storage()` factory

`Memory.from_env()` — 12-factor

`SignatureRecord` (dataclass)

`MemoryHit` (dataclass)

`TorsionReport` (dataclass, slots)

`AlignmentReport` (dataclass, slots)

`SurfaceReport` (dataclass, slots)

`MemoryListPage` (keyset pagination)

`ExplainReport` (dataclass)

`SyncReport` (dataclass)

`list-memories`

`search`

`explain`

`decay`

`stats`

`drawers`

`export` / `import`

`detect-torsion`

`verify`

`sync`

`ui`

6.4 Custom backends — implementing `BaseStorage`

TAS constants (`pdm_memory.core.retrieval`)

Pressure math constants (`pdm_memory.core.math`)

RetrievalEngine methods beyond `recall`

Full interface (`pdm_memory.storage.base.BaseStorage`)

`UPDATABLE_COLUMNS` whitelist

SQLite schema (`pdm_signatures`)

Privacy mode (`store_raw=False`)

Recommended `metadata` keys

Domain inference (`infer_domain(tags)`)

Regime inference (`infer_regime(tags)`)

`ContextWindowManager` — manual injection

Via `Memory.ingest()`

Auto field aliases (`DataIngester`)

`JWTAuth`

`MemorySync`

Errors (`pdm_memory.storage.errors`)