Python SDK
Use neuromem directly in your Python applications.
Installation
Install the neuromem package from PyPI. Requires Python 3.11 or later.
pip install neuromem
This installs the core library with async support. Provider dependencies (e.g. openai) are included.
Quick Start
A complete example showing the three core operations: ingest (store), recall (search), and digest (analyze).
import asyncio
from neuromem import NeuroMemory
from neuromem.providers import OpenAIEmbedding, OpenAILLM
async def main():
nm = NeuroMemory(
database_url="postgresql+asyncpg://user:pass@localhost/neuromem",
embedding=OpenAIEmbedding(api_key="sk-..."),
llm=OpenAILLM(api_key="sk-..."),
)
await nm.init()
# Store a memory
msg = await nm.ingest(
user_id="user-123",
role="user",
content="I prefer TypeScript and use VS Code daily"
)
# Search memories
results = await nm.recall(
user_id="user-123",
query="programming preferences"
)
for m in results["merged"]:
print(f"{m['memory_type']}: {m['content']} (score: {m['score']:.2f})")
# Discover behavioral patterns
digest_result = await nm.digest(user_id="user-123")
print(f"Traits generated: {digest_result['traits_generated']}")
await nm.close()
asyncio.run(main())Constructor
Creates a new NeuroMemory instance. You must call await nm.init() before using any other methods.
| Name | Type | Default | Description |
|---|---|---|---|
| database_url | str | required | PostgreSQL connection string. Must use the postgresql+asyncpg:// scheme. |
| embedding | EmbeddingProvider | required | Embedding provider instance (e.g. OpenAIEmbedding). |
| llm | LLMProvider | required | LLM provider instance for extraction and reflection (e.g. OpenAILLM). |
| storage | ObjectStorage | None | None | Optional S3-compatible storage for file attachments. |
| auto_extract | bool | True | Automatically extract facts, episodes, and triples on ingest. |
| graph_enabled | bool | False | Enable knowledge graph for entity-relationship queries. |
| pool_size | int | 10 | Database connection pool size. |
| reflection_interval | int | 20 | Auto-digest every N messages. Set to 0 to disable. |
Example
from neuromem import NeuroMemory
from neuromem.providers import OpenAIEmbedding, OpenAILLM
nm = NeuroMemory(
database_url="postgresql+asyncpg://user:pass@localhost/neuromem",
embedding=OpenAIEmbedding(api_key="sk-..."),
llm=OpenAILLM(api_key="sk-..."),
graph_enabled=True,
reflection_interval=50,
pool_size=20,
)
await nm.init()Core Methods
nm.ingest(user_id, role, content, ...)
Stores a conversation message to long-term memory. When auto_extract is enabled, facts, episodes, and knowledge graph triples are extracted in the background.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| user_id | str | required | User identifier for memory isolation. |
| role | str | required | Message role: "user", "assistant", or "system". |
| content | str | required | The message content to store. |
| session_id | str | None | None | Optional session/conversation ID for grouping messages. |
| metadata | dict | None | None | Arbitrary metadata to attach to the message. |
Returns
Message -- The stored message object with generated ID and timestamp.
Example
msg = await nm.ingest(
user_id="user-123",
role="user",
content="I just switched from PyCharm to VS Code",
session_id="session-456",
metadata={"source": "chat"}
)
print(f"Stored: {msg.id}")nm.recall(user_id, query, ...)
Hybrid memory search combining vector similarity, BM25 full-text search, and knowledge graph traversal. Returns merged results ranked by combined score.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| user_id | str | required | User identifier to search within. |
| query | str | required | Natural language search query. |
| limit | int | 20 | Maximum number of results to return. |
| memory_type | str | None | None | Filter by type: "fact", "episode", "conversation". |
| created_after | datetime | None | None | Only return memories created after this timestamp. |
| created_before | datetime | None | None | Only return memories created before this timestamp. |
| include_conversations | bool | False | Include raw conversation messages in results. |
| as_of | datetime | None | None | Time-travel: query memory state as of a past timestamp. |
Returns
dict -- {"vector_results": [...], "graph_results": [...], "merged": [...], "inferred_context": str, "context_confidence": float}
Example
results = await nm.recall(
user_id="user-123",
query="programming preferences",
limit=10,
memory_type="fact"
)
for m in results["merged"]:
print(f"[{m['memory_type']}] {m['content']} (score: {m['score']:.2f})")
print(f"Context: {results['inferred_context']}")
print(f"Confidence: {results['context_confidence']:.2f}")nm.digest(user_id, ...)
Analyzes un-reflected memories and synthesizes behavioral traits. Discovers patterns, preferences, and tendencies from accumulated memory data.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| user_id | str | required | User identifier to analyze. |
| batch_size | int | 50 | Number of memories to process per batch. |
| background | bool | False | Run asynchronously in the background. Returns None immediately. |
Returns
dict | None -- {"memories_analyzed": int, "traits_generated": int, "traits": [...]} or None if background=True.
Example
result = await nm.digest(user_id="user-123", batch_size=100)
print(f"Analyzed: {result['memories_analyzed']} memories")
print(f"Generated: {result['traits_generated']} traits")
for trait in result["traits"]:
print(f" - {trait['content']} (stage: {trait['stage']})")nm.list_memories(user_id, ...)
List memories with pagination. Useful for building memory browsers or audit views.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| user_id | str | required | User identifier. |
| memory_type | str | None | None | Filter by memory type. |
| limit | int | 50 | Maximum items per page. |
| offset | int | 0 | Number of items to skip. |
Returns
tuple[int, list] -- (total_count, list_of_memories)
Example
total, memories = await nm.list_memories(
user_id="user-123",
memory_type="fact",
limit=20,
offset=0
)
print(f"Showing {len(memories)} of {total} facts")nm.update_memory(memory_id, user_id, ...)
Update an existing memory's content or type. If content is changed, the embedding vector is automatically regenerated.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| memory_id | str | required | ID of the memory to update. |
| user_id | str | required | User identifier (for ownership verification). |
| content | str | None | None | New content (triggers re-embedding). |
| memory_type | str | None | None | New memory type. |
Returns
Memory | None -- The updated memory object, or None if not found.
Example
updated = await nm.update_memory(
memory_id="mem-abc123",
user_id="user-123",
content="Prefers VS Code with Vim keybindings"
)
if updated:
print(f"Updated: {updated.content}")nm.delete_memory(memory_id, user_id)
Permanently delete a memory by ID. Returns True if the memory was found and deleted.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| memory_id | str | required | ID of the memory to delete. |
| user_id | str | required | User identifier (for ownership verification). |
Returns
bool -- True if deleted, False if not found.
Example
deleted = await nm.delete_memory(
memory_id="mem-abc123",
user_id="user-123"
)
print("Deleted" if deleted else "Not found")nm.get_user_traits(user_id, ...)
Retrieve active behavioral traits for a user. Traits evolve through stages: trend, candidate, emerging, established, core. Dissolved traits are excluded by default.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| user_id | str | required | User identifier. |
| min_stage | str | "emerging" | Minimum trait stage to include. One of: "trend", "candidate", "emerging", "established", "core". |
| subtype | str | None | None | Filter by trait subtype (e.g. "preference", "habit", "expertise"). |
| context | str | None | None | Filter by context tag. |
Returns
list[dict] -- List of trait objects with content, stage, confidence, and metadata.
Example
traits = await nm.get_user_traits(
user_id="user-123",
min_stage="established",
subtype="preference"
)
for trait in traits:
print(f"[{trait['stage']}] {trait['content']}")nm.profile_view(user_id)
Get an assembled user profile combining extracted facts, behavioral traits, and recent mood indicators. Useful for providing context to AI assistants.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| user_id | str | required | User identifier. |
Returns
dict -- {"facts": [...], "traits": [...], "recent_mood": str | None}
Example
profile = await nm.profile_view(user_id="user-123")
print("Facts:")
for fact in profile["facts"]:
print(f" - {fact}")
print("Traits:")
for trait in profile["traits"]:
print(f" - {trait['content']} ({trait['stage']})")
print(f"Mood: {profile['recent_mood']}")nm.reflect(user_id, force=False)
Manually trigger the reflection cycle. Analyzes recent memories and creates, updates, or dissolves traits based on observed patterns.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| user_id | str | required | User identifier. |
| force | bool | False | Force reflection even if interval threshold has not been reached. |
Returns
dict -- {"triggered": bool, "traits_created": int, "traits_updated": int, "traits_dissolved": int}
Example
result = await nm.reflect(user_id="user-123", force=True)
if result["triggered"]:
print(f"Created: {result['traits_created']}")
print(f"Updated: {result['traits_updated']}")
print(f"Dissolved: {result['traits_dissolved']}")nm.init(schema=None)
Initialize database tables and extensions (pgvector, pg_search). Must be called once before using any other methods. Optionally specify a custom schema name.
Parameters
| Name | Type | Default | Description |
|---|---|---|---|
| schema | str | None | None | Custom PostgreSQL schema name. Defaults to public. |
Returns
None
Example
nm = NeuroMemory(...) await nm.init() # Creates tables if they don't exist
nm.close()
Close the database connection pool and release all resources. Should be called when the NeuroMemory instance is no longer needed.
Returns
None
Example
await nm.close() # Always clean up when done
Providers
NeuroMemory uses a pluggable provider system for embeddings and LLM inference. Each provider accepts an api_key and an optional model override.
Available Providers
from neuromem.providers import (
OpenAIEmbedding, # text-embedding-3-small (default)
OpenAILLM, # gpt-4o-mini (default)
SiliconFlowEmbedding, # BAAI/bge-m3 (default)
SiliconFlowLLM, # Qwen series (default)
)OpenAIEmbedding
Default model: text-embedding-3-small
OpenAI embedding API. 1536 dimensions.
OpenAILLM
Default model: gpt-4o-mini
OpenAI chat completion for extraction and reflection.
SiliconFlowEmbedding
Default model: BAAI/bge-m3
SiliconFlow embedding API. Multilingual support.
SiliconFlowLLM
Default model: Qwen series
SiliconFlow LLM for cost-effective inference.
Custom Model Override
embedding = OpenAIEmbedding(
api_key="sk-...",
model="text-embedding-3-large" # Override default model
)
llm = OpenAILLM(
api_key="sk-...",
model="gpt-4o" # Use a more capable model
)Key Features
Single PostgreSQL
pgvector + BM25 + knowledge graph in one database. No external dependencies.
Async-First
Built on SQLAlchemy 2.0 + asyncpg for high-throughput async workloads.
Pluggable Providers
Swap embedding and LLM providers without changing application code.
Multi-User Isolation
user_id enforced at the query layer. Each user's memories are strictly isolated.
Trait Lifecycle
Traits evolve through stages: trend, candidate, emerging, established, core, dissolved.
Knowledge Graph
Optional entity-relationship graph stored in PostgreSQL for structured queries.
Time-Travel Queries
Use the as_of parameter to query memory state at any point in history.
Source Code
View the full source, report issues, and contribute on GitHub.