Metadata-Version: 2.4
Name: human-memory-sdk
Version: 0.1.1
Summary: Python SDK for the Human-Like Memory System — ACT-R based memory for AI agents
License: MIT
Keywords: act-r,agents,ai,cognitive,memory
Requires-Python: >=3.12
Requires-Dist: httpx>=0.28.0
Requires-Dist: pydantic>=2.10.0
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.25; extra == 'dev'
Requires-Dist: pytest-httpx>=0.35; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.9; extra == 'dev'
Description-Content-Type: text/markdown

# human-memory

[![Python 3.12+](https://img.shields.io/badge/python-3.12+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](https://opensource.org/licenses/MIT)

Python SDK for the **Human-Like Memory System** — give your AI agents human-like memory based on the [ACT-R cognitive architecture](https://en.wikipedia.org/wiki/ACT-R).

Unlike simple vector stores, this system mimics how the human brain actually works:
- **Frequently used memories get stronger** (activation from access history)
- **Unused memories naturally fade** (power-law decay)
- **Emotional events persist longer** (salience-modified decay)
- **Context boosts related memories** (spreading activation via knowledge graph)
- **Repeated patterns become skills** (consolidation extracts knowledge)

## Installation

```bash
pip install human-memory
```

**Requirements:** Python 3.12+ | Only 2 dependencies: `httpx`, `pydantic`

## Prerequisites

The SDK connects to a running [human-memory server](https://github.com/AbdulrahmanMasoud/human-memory). Start it with:

```bash
git clone https://github.com/AbdulrahmanMasoud/human-memory.git
cd human-memory
docker compose up -d
```

The server runs at `http://localhost:8000` by default.

---

## Quick Start

```python
from human_memory import Memory

# Connect to the memory server
memory = Memory("http://localhost:8000")

# Store a memory
result = memory.store("User prefers dark mode in all applications")
print(result.memory_id)  # "a1b2c3d4-..."
print(result.activation)  # 1.0 (initial activation)

# Search memories (ranked by ACT-R activation, not just similarity)
results = memory.search("what are the user's preferences?")
for r in results:
    print(f"  {r.content}")
    print(f"  activation: {r.activation:.2f} | similarity: {r.similarity:.2f}")

# Close when done
memory.close()
```

---

## Core Concepts

### How Activation Works

Every memory has an **activation level** that determines how easily it can be retrieved:

```
A = B + S + P + noise
```

- **B (Base-level):** Memories accessed recently and frequently have higher B. Follows a power-law decay — rapid drop at first, then slower.
- **S (Spreading):** If related concepts are active in the current context, connected memories get a boost.
- **P (Partial matching):** Similar but imperfect matches get adjusted.
- **noise:** Random variation, like real human recall.

**The key insight:** A memory accessed 10 times this week has much higher activation than one accessed once a year ago. This is computed mathematically, not heuristically.

### Memory Lifecycle

```
Store --> Active (activation = 1.0)
  |
Access --> Activation increases (access recorded)
  |
Time passes --> Decay reduces activation (every 30 min)
  |
Below threshold --> Decayed (no longer retrievable)
  |
Consolidation --> Patterns extracted as semantic knowledge
```

---

## API Reference

### Initialize

```python
from human_memory import Memory, AsyncMemory

# Sync client
memory = Memory(
    base_url="http://localhost:8000",  # Server URL
    timeout=30.0,                      # Request timeout in seconds
)

# Async client (same API, all methods are async)
memory = AsyncMemory("http://localhost:8000")
```

### Context Manager

```python
# Sync -- auto-closes when done
with Memory("http://localhost:8000") as memory:
    memory.store("auto-managed connection")

# Async
async with AsyncMemory("http://localhost:8000") as memory:
    await memory.store("async auto-managed")
```

---

### Storing Memories

```python
result = memory.store(
    content="User asked about Python performance optimization",
    memory_type="episodic",   # Optional: "episodic" (default), "semantic", etc.
    context="support_chat",   # Optional: context tag for the access record
)
```

**Returns:** `MemoryCreated`

| Field | Type | Description |
|-------|------|-------------|
| `memory_id` | `str` | UUID of the stored memory |
| `activation` | `float` | Initial activation (1.0) |
| `created_at` | `str` | ISO timestamp |

---

### Searching Memories

```python
results = memory.search(
    query="Python performance",  # Natural language query
    top_k=5,                     # Max results (default: 7)
    min_activation=-0.5,         # Optional: override retrieval threshold
)

for r in results:
    print(f"{r.content}")
    print(f"  activation: {r.activation:.3f}")  # ACT-R activation score
    print(f"  similarity: {r.similarity:.3f}")   # Cosine similarity to query
    print(f"  accesses: {r.access_count}")
```

**Returns:** `list[SearchResult]`

| Field | Type | Description |
|-------|------|-------------|
| `memory_id` | `str` | UUID |
| `content` | `str` | Memory text |
| `activation` | `float` | ACT-R activation (ranking key) |
| `similarity` | `float` | Cosine similarity to query |
| `last_accessed` | `str` | When last retrieved |
| `access_count` | `int` | Total access count |

**Important:** Results are ranked by **activation** (recency + frequency + context), not just similarity. This is what makes the system cognitively plausible.

---

### Recalling by ID

```python
info = memory.recall("a1b2c3d4-...")
print(info.content)       # "User prefers dark mode..."
print(info.access_count)  # 6 (incremented by this recall)
```

**Note:** Every `recall()` records an access event, which **strengthens** the memory. This is the ACT-R reinforcement loop -- useful memories get stronger through use.

**Returns:** `MemoryInfo`

| Field | Type | Description |
|-------|------|-------------|
| `memory_id` | `str` | UUID |
| `content` | `str` | Full text |
| `activation` | `float` | Current activation |
| `memory_type` | `str` | "episodic", "semantic", etc. |
| `created_at` | `str` | Creation timestamp |
| `last_accessed` | `str` | Last access timestamp |
| `access_count` | `int` | Total accesses (including this one) |

---

### Inspecting Full Metadata

```python
detail = memory.inspect("a1b2c3d4-...")

print(f"Activation: {detail.activation:.4f}")
print(f"Salience: {detail.salience:.4f}")
print(f"Emotion: valence={detail.emotion_valence:.2f}, arousal={detail.emotion_arousal:.2f}")
print(f"Decay rate: {detail.decay_rate}")
print(f"Status: {detail.status}")  # "active", "decayed", or "deleted"
print(f"Access history: {len(detail.access_history)} entries")

for access in detail.access_history[:5]:
    print(f"  {access.accessed_at} -- {access.context}")
```

**Returns:** `MemoryDetail`

| Field | Type | Description |
|-------|------|-------------|
| `activation` | `float` | Current ACT-R activation |
| `salience` | `float` | Emotional salience (0-1) |
| `emotion_valence` | `float` | Positive/negative (-1 to 1) |
| `emotion_arousal` | `float` | Calm/excited (0 to 1) |
| `decay_rate` | `float` | ACT-R d parameter (default 0.5) |
| `status` | `str` | "active", "decayed", "deleted" |
| `access_history` | `list[AccessRecord]` | Timestamped access log |

---

### Forgetting

```python
# Soft-delete a specific memory
deleted = memory.forget("a1b2c3d4-...")  # Returns True

# The memory is now inaccessible
memory.recall("a1b2c3d4-...")  # Raises MemoryNotFound
```

---

### Listing All Memories

```python
# Paginated list (includes active, decayed, and deleted)
items = memory.list(offset=0, limit=50)

for item in items:
    print(f"[{item.status}] {item.content[:60]}... (activation: {item.activation:.2f})")
```

**Returns:** `list[MemoryItem]`

---

## Knowledge Graph

The knowledge graph stores **semantic memory** -- general facts and relationships between concepts. It powers the **spreading activation** component of ACT-R.

### Building the Graph

```python
# Create concepts
memory.add_concept("Python", type="language", activation=0.9)
memory.add_concept("AI", type="field")
memory.add_concept("FastAPI", type="framework")
memory.add_concept("Web Development", type="field")

# Create typed relationships with weights
memory.add_relation("Python", "AI", "USED_FOR", weight=0.9)
memory.add_relation("FastAPI", "Python", "WORKS_WITH", weight=0.95)
memory.add_relation("FastAPI", "Web Development", "USED_FOR", weight=0.85)
```

### Querying Concepts

```python
concept = memory.get_concept("Python")
print(f"{concept.name} ({concept.type})")
print(f"Activation: {concept.activation}")

for rel in concept.relationships:
    print(f"  --{rel.relation_type}--> {rel.target} (weight: {rel.weight})")
```

### Spreading Activation

Find related concepts by spreading activation through the graph:

```python
# What's related to Python?
related = memory.spread(
    concepts=["Python"],  # Active concepts in current context
    depth=2,              # How many hops to traverse (1-4)
    limit=10,             # Max results
)

for r in related:
    print(f"  {r.name}: activation={r.activation:.2f}, spread={r.path_weight:.3f}")

# Output:
#   AI: activation=0.80, spread=0.900
#   FastAPI: activation=0.80, spread=0.950
#   Web Development: activation=0.80, spread=0.808
```

**How it works:** When "Python" is the active context, activation spreads through the graph edges. AI gets a boost because Python->AI has weight 0.9. Web Development gets a smaller boost because it's 2 hops away (Python->FastAPI->Web Dev).

**Returns:** `list[SpreadResult]`

| Field | Type | Description |
|-------|------|-------------|
| `name` | `str` | Concept name |
| `activation` | `float` | Concept's base activation |
| `path_weight` | `float` | Accumulated spread weight |

---

## Operations

### Temporal Decay

Recalculates activation for all active memories using the ACT-R equation. Memories below the retrieval threshold are marked as decayed.

```python
report = memory.decay()
print(f"Processed: {report.memories_processed}")
print(f"Decayed: {report.memories_decayed}")
```

**Note:** Runs automatically every 30 minutes. Use this for manual triggering.

### Consolidation

Runs the 4-phase consolidation cycle (inspired by sleep):

1. **Replay** -- boost high-salience recent memories
2. **Extract** -- cluster similar episodes, LLM extracts semantic facts
3. **Prune** -- global activation downscaling, archive weak memories
4. **Compile** -- convert repeated patterns to procedural skills

```python
report = memory.consolidate()
print(f"Replayed: {report.episodes_replayed}")
print(f"Facts extracted: {report.facts_extracted}")
print(f"Pruned: {report.memories_pruned}")
print(f"Downscaled: {report.memories_downscaled}")
```

**Note:** Runs automatically every 6 hours. Use this for manual triggering.

### Strategic Forgetting

```python
# Weaken memories not relevant to current goals
report = memory.forget_strategy(
    "strategic_prune",
    goals=["Python", "machine learning", "optimization"]
)
print(f"Memories affected: {report.memories_affected}")

# Archive weakest memories when store is full
report = memory.forget_strategy("capacity_overflow")
```

| Strategy | Description |
|----------|-------------|
| `strategic_prune` | Weaken memories unrelated to specified `goals` |
| `capacity_overflow` | Archive weakest memories when capacity exceeded |

---

## System Status

```python
# Statistics
stats = memory.stats()
print(f"Total: {stats.total}")
print(f"Active: {stats.active}")
print(f"Decayed: {stats.decayed}")
print(f"Deleted: {stats.deleted}")
print(f"Avg activation: {stats.avg_activation:.3f}")

# Backend readiness
checks = memory.ready()
# {'postgres': 'ok', 'qdrant': 'ok', 'neo4j': 'ok'}
```

---

## Async Client

`AsyncMemory` has the exact same API -- all methods are `async`:

```python
import asyncio
from human_memory import AsyncMemory

async def main():
    async with AsyncMemory("http://localhost:8000") as memory:
        result = await memory.store("Async memory example")
        results = await memory.search("async example")
        await memory.add_concept("AsyncIO", type="concept")
        stats = await memory.stats()
        print(f"Total memories: {stats.total}")

asyncio.run(main())
```

### Use with FastAPI

```python
from fastapi import FastAPI
from human_memory import AsyncMemory

app = FastAPI()
memory = AsyncMemory("http://memory-server:8000")

@app.post("/chat")
async def chat(message: str):
    # Retrieve relevant context
    context = await memory.search(message, top_k=5)

    # ... call LLM with context ...

    # Store the interaction
    await memory.store(f"User: {message}\nAssistant: {response}")
    return {"response": response}
```

---

## Error Handling

```python
from human_memory import Memory
from human_memory.exceptions import (
    MemoryNotFound,
    ServiceUnavailable,
    ValidationError,
    HumanMemoryError,
)

memory = Memory("http://localhost:8000")

try:
    memory.recall("nonexistent-id")
except MemoryNotFound:
    print("Memory not found")  # 404

try:
    memory.store("")  # Empty content
except ValidationError:
    print("Invalid input")  # 422

try:
    memory.spread(["Python"])  # Neo4j down
except ServiceUnavailable:
    print("Knowledge graph not available")  # 503
```

**Exception hierarchy:**

```
HumanMemoryError          # Base (any HTTP error)
  MemoryNotFound          # 404
  ServiceUnavailable      # 503
  ValidationError         # 422
```

---

## Complete Example: AI Agent with Memory

```python
from human_memory import Memory

def create_agent():
    memory = Memory("http://localhost:8000")

    # Build the agent's knowledge base
    memory.add_concept("Python", type="language")
    memory.add_concept("JavaScript", type="language")
    memory.add_concept("Web", type="field")
    memory.add_relation("JavaScript", "Web", "USED_FOR", weight=0.95)
    memory.add_relation("Python", "Web", "USED_FOR", weight=0.6)

    return memory

def agent_respond(memory: Memory, user_message: str) -> str:
    # 1. Retrieve relevant memories
    memories = memory.search(user_message, top_k=5)
    context = "\n".join(
        f"- {m.content} (relevance: {m.activation:.2f})"
        for m in memories
    )

    # 2. Check knowledge graph for related concepts
    try:
        related = memory.spread(["Python"], limit=3)
        for r in related:
            context += f"\n- Related: {r.name} (weight: {r.path_weight:.2f})"
    except Exception:
        pass

    # 3. Build prompt for LLM
    prompt = f"""You are a helpful assistant with memory.

Relevant context from past interactions:
{context if context else "No relevant memories yet."}

User: {user_message}
Assistant:"""

    # 4. Call your LLM here
    response = call_your_llm(prompt)

    # 5. Store this interaction
    memory.store(f"User asked: {user_message}")

    return response

# Usage
memory = create_agent()
print(agent_respond(memory, "How do I optimize Python code?"))
print(agent_respond(memory, "What about JavaScript for web apps?"))

# Maintenance (also runs automatically in background)
memory.decay()
memory.consolidate()

stats = memory.stats()
print(f"Agent has {stats.active} active memories")
memory.close()
```

---

## All Models

| Model | Returned by | Key fields |
|-------|-------------|------------|
| `MemoryCreated` | `store()` | `memory_id`, `activation`, `created_at` |
| `SearchResult` | `search()` | `memory_id`, `content`, `activation`, `similarity`, `access_count` |
| `MemoryInfo` | `recall()` | `memory_id`, `content`, `activation`, `memory_type`, `access_count` |
| `MemoryDetail` | `inspect()` | All of MemoryInfo + `salience`, `emotion_valence`, `emotion_arousal`, `decay_rate`, `status`, `access_history` |
| `MemoryItem` | `list()` | `memory_id`, `content`, `activation`, `salience`, `status`, `access_count` |
| `Stats` | `stats()` | `total`, `active`, `decayed`, `deleted`, `avg_activation` |
| `Concept` | `add_concept()`, `get_concept()` | `name`, `type`, `activation`, `relationships` |
| `Relation` | `add_relation()` | `source`, `target`, `relation_type`, `weight` |
| `SpreadResult` | `spread()` | `name`, `activation`, `path_weight` |
| `DecayReport` | `decay()` | `memories_processed`, `memories_decayed` |
| `ConsolidationReport` | `consolidate()` | `episodes_replayed`, `facts_extracted`, `memories_pruned`, `memories_downscaled` |
| `ForgetReport` | `forget_strategy()` | `strategy`, `memories_affected` |
| `AccessRecord` | Inside `MemoryDetail` | `accessed_at`, `context` |

---

## License

MIT
